修复:- 上次选中的节点在本机已不可用(如 IPv6 专用节点但本机无 IPv6 出口)时启动不再停留其上:新增 _verify_selected_node/_first_reachable_index,启动后台校验并自动回退到第一个可达节点、写入设置、刷新标题栏(原来只在点开启代理时才报错)
This commit is contained in:
@@ -263,6 +263,34 @@ def _reload_nodes():
|
||||
_IPV6_ONLY.clear()
|
||||
return _load_nodes()
|
||||
|
||||
def _first_reachable_index(nodes):
|
||||
"""第一个本机可达的节点下标;都不可达返回 0。供选中不可达节点时回退。"""
|
||||
for i, n in enumerate(nodes):
|
||||
if _node_reachable(n):
|
||||
return i
|
||||
return 0
|
||||
|
||||
def _verify_selected_node():
|
||||
"""启动时校验上次选中的节点在本机是否还可用。
|
||||
|
||||
典型场景:上次选了 IPv6 专用节点,这次本机没有 IPv6 出口(或该节点已不可达),
|
||||
继续选中它只会在开启代理时才报错。这里自动回退到第一个可达节点并记录到设置。
|
||||
返回 True 表示发生了切换(调用方据此刷新标题栏)。"""
|
||||
global _SELECTED_NODE
|
||||
nodes = _load_nodes()
|
||||
if not nodes or not (0 <= _SELECTED_NODE < len(nodes)):
|
||||
return False
|
||||
if _node_reachable(nodes[_SELECTED_NODE]):
|
||||
return False
|
||||
idx = _first_reachable_index(nodes)
|
||||
old = str(nodes[_SELECTED_NODE].get("domain") or "")
|
||||
new = str(nodes[idx].get("domain") or "")
|
||||
_SELECTED_NODE = idx
|
||||
_save_current_node(new)
|
||||
if new and new != old:
|
||||
log_red(f"节点 {old} 在本机不可用,已自动切换到 {new}")
|
||||
return True
|
||||
|
||||
def _node_val(key, default=""):
|
||||
node = _current_node()
|
||||
if not node:
|
||||
@@ -2130,6 +2158,7 @@ class MainWindow(QWidget):
|
||||
_update_download_done = pyqtSignal(bool, str)
|
||||
|
||||
_status_ready = pyqtSignal(object)
|
||||
_node_switched = pyqtSignal() # 启动校验时代替用户切换了节点(刷新标题栏)
|
||||
|
||||
def __init__(self):
|
||||
super().__init__()
|
||||
@@ -2143,6 +2172,7 @@ class MainWindow(QWidget):
|
||||
self._update_done.connect(self._on_update_done)
|
||||
self._update_download_done.connect(self._on_update_download_done)
|
||||
self._status_ready.connect(self._apply_status)
|
||||
self._node_switched.connect(self._on_node_switched)
|
||||
|
||||
_user_init_local() # 首启初始化用户数据目录(节点模板),须在 init_ui 读节点之前
|
||||
|
||||
@@ -2160,9 +2190,27 @@ class MainWindow(QWidget):
|
||||
threading.Thread(target=_user_init_remote, daemon=True).start()
|
||||
# 后台用已保存凭据同步最新节点配置(服务端改动下次启动生效)
|
||||
threading.Thread(target=_nodes_sync_remote, daemon=True).start()
|
||||
# 后台校验上次选中的节点在本机是否还可用(IPv6 专用节点本机无 IPv6 时回退)
|
||||
threading.Thread(target=self._verify_node_thread, daemon=True).start()
|
||||
# 节点未配置时延迟弹出登录框(等窗口显示后)
|
||||
QTimer.singleShot(600, self._maybe_show_login)
|
||||
|
||||
def _verify_node_thread(self):
|
||||
"""等节点配置在线同步完成后,校验当前选中节点是否可用,必要时自动回退。
|
||||
|
||||
探测要解析 DNS 并尝试 TCP 连接,放在后台线程,避免卡住启动。"""
|
||||
import time
|
||||
time.sleep(1.5)
|
||||
try:
|
||||
if _verify_selected_node():
|
||||
self._node_switched.emit()
|
||||
except Exception as e:
|
||||
log(f"节点可用性校验失败: {e}")
|
||||
|
||||
def _on_node_switched(self):
|
||||
"""节点被自动切换后刷新标题栏(在 UI 线程执行)。"""
|
||||
self.subtitle.setText(f"{_endpoint_domain()} ({_endpoint_country()})")
|
||||
|
||||
def init_ui(self):
|
||||
"""初始化主界面布局与样式。"""
|
||||
self.setWindowTitle("R.R.P.P.")
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""离屏验证节点菜单:IPv6-only 节点的标注与置灰(真实 PyQt5,不出窗口)。
|
||||
|
||||
需要 Windows venv 的真实 PyQt5:
|
||||
..\\.venv\\Scripts\\python.exe test\\check_node_menu.py
|
||||
读取 test/nodes.json 作为节点数据,只构建菜单不进事件循环。
|
||||
"""
|
||||
import os, sys
|
||||
os.environ["QT_QPA_PLATFORM"] = "offscreen"
|
||||
from pathlib import Path
|
||||
|
||||
LITE = Path(__file__).resolve().parent.parent
|
||||
# 脚本所在目录(test/)会被解释器自动放到 sys.path 首位,那里有 PyQt5 桩,
|
||||
# 会顶掉真实 PyQt5 —— 必须移除,改用本程序自己的 venv
|
||||
sys.path = [str(LITE)] + [p for p in sys.path if Path(p).resolve() != Path(__file__).resolve().parent]
|
||||
|
||||
sys.stdout.reconfigure(encoding="utf-8")
|
||||
|
||||
import proxy_lite as pp
|
||||
from PyQt5.QtWidgets import QApplication
|
||||
|
||||
# 构造窗口时不写用户数据目录(避免污染),构造后再指向真实 nodes.json
|
||||
pp._user_init_local = lambda: None
|
||||
|
||||
app = QApplication(sys.argv)
|
||||
w = pp.MainWindow()
|
||||
w.timer.stop() # 停掉状态轮询,纯静态检查
|
||||
|
||||
pp._USER_DIR = LITE / "test" # 真实 nodes.json 所在目录
|
||||
nodes = pp._reload_nodes()
|
||||
print(f"本机有全局 IPv6: {pp._local_has_ipv6()}")
|
||||
print(f"节点数: {len(nodes)}")
|
||||
|
||||
menu = w._build_node_menu(nodes)
|
||||
print("\n菜单项(enabled / 文案 / tooltip):")
|
||||
for act in menu.actions():
|
||||
print(f" {str(act.isEnabled()):5} | {act.text()}")
|
||||
if act.toolTip():
|
||||
print(f" └ {act.toolTip()}")
|
||||
|
||||
# 点击被禁用的项不应切换节点
|
||||
before = pp._selected_index()
|
||||
for act in menu.actions():
|
||||
if not act.isEnabled():
|
||||
act.trigger()
|
||||
print(f"\n点击禁用项后节点下标: {before} -> {pp._selected_index()} (应不变)")
|
||||
|
||||
# ── 上次选中的节点已不可用:启动时自动回退到第一个可达节点 ──────────
|
||||
print("\n启动校验(_verify_selected_node):")
|
||||
bad = [i for i, n in enumerate(nodes) if not pp._node_reachable(n)]
|
||||
if not bad:
|
||||
print(" 本机所有节点均可达,无法验证回退(本机有 IPv6?)")
|
||||
else:
|
||||
import json
|
||||
_bak = (pp._USER_DIR / "settings.json").read_bytes() if (pp._USER_DIR / "settings.json").is_file() else None
|
||||
bad_idx = bad[0]
|
||||
pp._SELECTED_NODE = bad_idx # 模拟上次选了不可用节点
|
||||
switched = pp._verify_selected_node()
|
||||
expect = pp._first_reachable_index(nodes)
|
||||
ok = switched and pp._selected_index() == expect
|
||||
print(f" 上次选中 {nodes[bad_idx].get('domain')}(下标 {bad_idx}) 不可用 -> "
|
||||
f"回退到下标 {pp._selected_index()} ({nodes[pp._selected_index()].get('domain')}) "
|
||||
f"{'PASS' if ok else 'FAIL'}")
|
||||
saved = json.loads((pp._USER_DIR / "settings.json").read_text("utf-8")).get("current_node") if (pp._USER_DIR / "settings.json").is_file() else ""
|
||||
ok2 = saved == str(nodes[expect].get("domain") or "")
|
||||
print(f" 回退结果已写入设置 current_node={saved!r} {'PASS' if ok2 else 'FAIL'}")
|
||||
# 再校验一次:已经是可达节点,不应再切换
|
||||
ok3 = pp._verify_selected_node() is False
|
||||
print(f" 再次校验不重复切换 {'PASS' if ok3 else 'FAIL'}")
|
||||
if _bak is None:
|
||||
(pp._USER_DIR / "settings.json").unlink(missing_ok=True)
|
||||
else:
|
||||
(pp._USER_DIR / "settings.json").write_bytes(_bak)
|
||||
|
||||
print("OK")
|
||||
Reference in New Issue
Block a user