diff --git a/src/trader/config.py b/src/trader/config.py index 9e7572c..c9bfa7a 100644 --- a/src/trader/config.py +++ b/src/trader/config.py @@ -13,6 +13,7 @@ class TraderConfig: account_name: Optional[str] = None paired_at: Optional[str] = None xiadan_path_manual: Optional[str] = None # 用户手动指定的 xiadan.exe 路径 + tesseract_path_manual: Optional[str] = None # 用户手动指定的 tesseract.exe 路径 ws_endpoint: Optional[str] = None # 自定义中转地址:只填域名或 IP[:端口],协议和路径自动补全 enable_ths_plugin: bool = True # 是否启用同花顺实盘交易插件,默认启用 order_watch_idle_secs: int = 300 # order_watch 空闲轮询周期(秒):默认 5 分钟 @@ -74,6 +75,7 @@ def load() -> TraderConfig: account_name=data.get("account_name"), paired_at=data.get("paired_at"), xiadan_path_manual=data.get("xiadan_path_manual"), + tesseract_path_manual=data.get("tesseract_path_manual"), ws_endpoint=data.get("ws_endpoint"), enable_ths_plugin=data.get("enable_ths_plugin", True), order_watch_idle_secs=data.get("order_watch_idle_secs", 300), @@ -95,6 +97,7 @@ def save(config: TraderConfig) -> None: "account_name": config.account_name, "paired_at": config.paired_at, "xiadan_path_manual": config.xiadan_path_manual, + "tesseract_path_manual": config.tesseract_path_manual, "ws_endpoint": config.ws_endpoint, "enable_ths_plugin": config.enable_ths_plugin, "order_watch_idle_secs": config.order_watch_idle_secs, @@ -106,4 +109,3 @@ def save(config: TraderConfig) -> None: with open(config_path, "w", encoding="utf-8") as f: json.dump(data, f, indent=2, ensure_ascii=False) - diff --git a/src/trader/installer/auto_install.py b/src/trader/installer/auto_install.py index 89fa646..71e6ed7 100644 --- a/src/trader/installer/auto_install.py +++ b/src/trader/installer/auto_install.py @@ -64,8 +64,8 @@ async def ensure_xiadan( 确保 xiadan.exe 可用——纯检测,不自动下载。 返回 None 时,调用方应该让用户: - - 点「下载同花顺」按钮 → 看 README 手动装 - - 或点「指定路径...」→ 选 xiadan.exe 写到 config + - 选择本机已安装的 xiadan.exe 写到 config + - 或从同花顺官方渠道安装后重新检测 历史:早期版本 (v0.2.0-v0.2.6) 这里走 ensure_private_install 自动下载 214MB + silent install。实测 HTTP 403(CDN 屏蔽)+ 同花顺 EULA 不允许第三方分发, @@ -83,7 +83,7 @@ async def ensure_xiadan( )) return detected - logger.info("未检测到 xiadan——等用户在 UI 点「下载同花顺」或「指定路径」") + logger.info("未检测到 xiadan——等待用户选择路径或完成官方安装后重新检测") return None diff --git a/src/trader/installer/tesseract.py b/src/trader/installer/tesseract.py index 95c95ea..f139677 100644 --- a/src/trader/installer/tesseract.py +++ b/src/trader/installer/tesseract.py @@ -37,6 +37,15 @@ def detect_tesseract() -> Optional[str]: - 具体路径 → 在固定安装目录命中 - ``None`` → 未安装 """ + try: + from .. import config + + manual = config.load().tesseract_path_manual + if manual and Path(manual).is_file(): + return manual + except Exception: + logger.debug("读取手动 Tesseract 路径失败", exc_info=True) + if shutil.which("tesseract"): return "" for p in TESSERACT_COMMON_PATHS: @@ -81,11 +90,11 @@ def log(msg: str) -> None: return existing if not shutil.which("winget"): - log("⚠ 未找到 winget,无法自动安装 Tesseract(手动:winget install " - f"{WINGET_PACKAGE_ID})") + log("⚠ 未找到 winget,无法自动安装 Tesseract。请选择本地 tesseract.exe," + f"或手动运行:winget install -e --id {WINGET_PACKAGE_ID}") return None - log("正在通过 winget 安装 Tesseract OCR(首次约 1-2 分钟)...") + log("正在通过 winget 后台尝试安装 Tesseract OCR;不影响连接和窗口检测...") cmd = [ "winget", "install", "-e", "--id", WINGET_PACKAGE_ID, "--silent", @@ -117,9 +126,14 @@ def log(msg: str) -> None: return found await asyncio.sleep(1.5) - # 仍没有:多半是 winget 静默安装需要管理员权限(UB-Mannheim 是 per-machine 装), - # 无提权时静默失败。明确引导手动安装。 - log(f"⚠ Tesseract 自动安装未成功(winget rc={rc})。请用【管理员】PowerShell 运行:" - f"winget install -e --id {WINGET_PACKAGE_ID} 装好后重启 trader。" - "(在此之前下单验证码需手动输入)") + unsigned_rc = (rc or 0) & 0xFFFFFFFF + code = f"0x{unsigned_rc:08X}" + if "InternetOpenUrl() failed" in out or unsigned_rc == 0x80072EE2: + reason = "下载 GitHub 安装包超时,请检查虚拟机网络,或选择本地 tesseract.exe" + elif unsigned_rc == 0xC000013A: + reason = "安装进程已中断" + else: + reason = "winget 未完成安装" + log(f"⚠ Tesseract 自动安装失败({code}):{reason}。" + "OCR 已跳过,不影响连接和交易窗口检测。") return None diff --git a/src/trader/main.py b/src/trader/main.py index e75026d..ead3265 100644 --- a/src/trader/main.py +++ b/src/trader/main.py @@ -285,30 +285,27 @@ async def _async_main( if platform.system() == "Windows" and bootstrap_result.found_xiadan_path: _check_xiadan_conflict(state) - # Step 3.5: 确保 Tesseract OCR 就绪(缺则 winget 静默安装),再喂给 OCR 后端。 - # 旧架构在 server.py 启动时调 thsauto_setup(window_title, tesseract_cmd);PH-061 - # 迁到本仓库时这步丢了 → pytesseract 只认 PATH,验证码识别静默失效。这里恢复该 - # setup 调用,并补上旧版没有的"自动安装"。None=没装→装;""=在 PATH;具体路径=已应用。 + # Step 3.5: 先接通已有 OCR;缺失时的 winget 安装转为后台任务,绝不能阻塞 + # xiadan 窗口检测或网关连接。None=未装;""=PATH;具体路径=已应用。 if platform.system() == "Windows": try: - from .installer.tesseract import ensure_tesseract + from .installer.tesseract import verify_ocr_runnable from .ths.win import setup as ths_setup tesseract_cmd = bootstrap_result.found_tesseract_cmd - if tesseract_cmd is None: - tesseract_cmd = await ensure_tesseract(on_log=state.log) - bootstrap_result.found_tesseract_cmd = tesseract_cmd ths_setup("网上股票交易系统5.0", tesseract_cmd or "", str(trader_config.tmp_dir())) if tesseract_cmd is None: - state.log("⚠ Tesseract 仍不可用,下单验证码无法自动识别") + state.update(ocr_status="installing", ocr_message="正在后台尝试安装") + state.log("OCR 未安装,正在后台尝试安装;不影响连接和窗口检测") else: - # 启动自检:确认 OCR 真能跑,而不只是文件在。 - from .installer.tesseract import verify_ocr_runnable ocr_ok, ocr_info = verify_ocr_runnable() if ocr_ok: + state.update(ocr_status="ready", ocr_message=str(ocr_info)) state.log(f"✓ OCR 就绪(Tesseract {ocr_info})") else: + state.update(ocr_status="unavailable", ocr_message="OCR 自检失败") state.log(f"⚠ OCR 自检失败:{ocr_info}(下单验证码可能无法识别)") except Exception as e: + state.update(ocr_status="unavailable", ocr_message="OCR 初始化失败") logger.warning("Tesseract 准备失败: %s", e) # Step 4: WS 连接 @@ -372,6 +369,37 @@ def on_pair_pending(code, expires_at) -> None: order_event_task = asyncio.create_task(order_watch.order_watch_task(state, client)) from . import watchlist_watch watchlist_task = asyncio.create_task(watchlist_watch.watchlist_watch_task(state, client)) + ocr_task = None + if platform.system() == "Windows" and bootstrap_result.found_tesseract_cmd is None: + async def _install_ocr_in_background() -> None: + try: + from .installer.tesseract import ensure_tesseract, verify_ocr_runnable + from .ths.win import setup as ths_setup + + installed = await ensure_tesseract(on_log=state.log) + # 用户在安装期间已选择有效的离线程序时,保留其选择和当前状态。 + if trader_config.load().tesseract_path_manual: + return + if installed is None: + state.update(ocr_status="unavailable", ocr_message="自动安装失败,可选择本地 tesseract.exe") + return + ths_setup("网上股票交易系统5.0", installed, str(trader_config.tmp_dir())) + ok, info = verify_ocr_runnable() + if ok: + bootstrap_result.found_tesseract_cmd = installed + state.update(ocr_status="ready", ocr_message=str(info)) + state.log(f"✓ OCR 就绪(Tesseract {info})") + else: + state.update(ocr_status="unavailable", ocr_message="安装后自检失败") + state.log(f"⚠ Tesseract 安装后自检失败:{info}") + except asyncio.CancelledError: + raise + except Exception as e: + state.update(ocr_status="unavailable", ocr_message="自动安装异常,可选择本地 tesseract.exe") + state.log(f"⚠ Tesseract 后台安装异常:{e}") + logger.exception("Tesseract 后台安装失败") + + ocr_task = asyncio.create_task(_install_ocr_in_background()) try: await client.run() except asyncio.CancelledError: @@ -385,7 +413,11 @@ def on_pair_pending(code, expires_at) -> None: order_event_task.cancel() watchlist_task.cancel() try: - await asyncio.gather(polling_task, refresh_task, order_event_task, watchlist_task, return_exceptions=True) + tasks = [polling_task, refresh_task, order_event_task, watchlist_task] + if ocr_task is not None: + ocr_task.cancel() + tasks.append(ocr_task) + await asyncio.gather(*tasks, return_exceptions=True) except Exception: pass @@ -597,7 +629,7 @@ def on_open_xiadan() -> None: except Exception as e: state.log(f"⚠ 启动同花顺失败: {e}") else: - state.log("⚠ xiadan 路径未知。先点「指定路径...」选 xiadan.exe,或「下载同花顺」装一份") + state.log("⚠ xiadan 路径未知。请选择 xiadan.exe,或从同花顺官方渠道安装后重新检测。") def on_redetect_xiadan() -> None: """重新检测 xiadan 路径""" @@ -611,7 +643,7 @@ def on_redetect_xiadan() -> None: state.log(f"✓ 重新检测命中:{found}") else: state.update(xiadan_path=None) - state.log("⚠ 重新检测未找到 xiadan。点「下载同花顺」或「指定路径...」") + state.log("⚠ 重新检测未找到 xiadan。请选择 xiadan.exe,或确认已从同花顺官方渠道安装。") except Exception as e: state.log(f"⚠ 检测异常: {e}") @@ -637,6 +669,32 @@ def on_set_xiadan_path(path: str) -> None: except Exception as e: state.log(f"⚠ 设置路径失败: {e}") + def on_set_tesseract_path(path: str) -> None: + """接入用户离线提供的 tesseract.exe,并立即验证。""" + try: + from pathlib import Path as _P + from .installer.tesseract import verify_ocr_runnable + from .ths.win import setup as ths_setup + + p = _P(path) + if not p.is_file(): + state.log(f"⚠ 指定的 Tesseract 路径不是文件:{path}") + return + ths_setup("网上股票交易系统5.0", str(p), str(trader_config.tmp_dir())) + ok, info = verify_ocr_runnable() + if not ok: + state.update(ocr_status="unavailable", ocr_message="所选程序无法运行") + state.log(f"⚠ 所选 Tesseract 无法运行:{info}") + return + result.config.tesseract_path_manual = str(p) + trader_config.save(result.config) + bootstrap_result.found_tesseract_cmd = str(p) + state.update(ocr_status="ready", ocr_message=str(info)) + state.log(f"✓ 已设置本地 Tesseract:{p}") + except Exception as e: + state.update(ocr_status="unavailable", ocr_message="设置本地 OCR 失败") + state.log(f"⚠ 设置 Tesseract 路径失败:{e}") + def on_reset_pair() -> None: try: # 1. 清 config 中的 pairing 字段 @@ -722,6 +780,7 @@ def on_tray_exit() -> None: on_exit=on_main_exit, on_redetect_xiadan=on_redetect_xiadan, on_set_xiadan_path=on_set_xiadan_path, + on_set_tesseract_path=on_set_tesseract_path, on_apply_self_update=on_apply_self_update, minimize_to_tray=(platform.system() == "Windows"), ) diff --git a/src/trader/main_window.py b/src/trader/main_window.py index f63ada3..14a57e7 100644 --- a/src/trader/main_window.py +++ b/src/trader/main_window.py @@ -28,6 +28,20 @@ def _format_pair_code(code: Optional[str]) -> str: return code or "" +def _display_path(path: Optional[str], max_chars: int = 42) -> str: + """在紧凑状态栏展示路径,保留盘符和可执行文件名。""" + if not path or len(path) <= max_chars: + return path or "" + + normalized = path.replace("/", "\\") + filename = normalized.rsplit("\\", 1)[-1] + prefix = normalized[:3] if len(normalized) >= 3 and normalized[1:3] == ":\\" else "" + compact = f"{prefix}...\\{filename}" + if len(compact) <= max_chars: + return compact + return f"...{filename[-(max_chars - 3):]}" + + @dataclass class SharedState: """主线程 + asyncio 线程共享状态。需要 lock 才能安全修改。""" @@ -40,6 +54,8 @@ class SharedState: last_pong_at: Optional[float] = None fatal_reason: Optional[str] = None install_progress: Optional[tuple[int, int]] = None # (done, total) + ocr_status: str = "checking" # checking | installing | ready | unavailable + ocr_message: str = "" self_update_info: Optional[UpdateInfo] = None # 检测到的新版本信息,None=无更新 self_update_progress: Optional[tuple[int, int]] = None # (done, total) self_update_status: str = "idle" # idle | downloading | error @@ -67,6 +83,8 @@ def snapshot(self) -> dict: "last_pong_at": self.last_pong_at, "fatal_reason": self.fatal_reason, "install_progress": self.install_progress, + "ocr_status": self.ocr_status, + "ocr_message": self.ocr_message, "ths_steps_complete": self.ths_steps_complete, "ths_expanded": self.ths_expanded, "ths_refreshing": self.ths_refreshing, @@ -124,6 +142,7 @@ def __init__( on_exit: Optional[Callable[[], None]] = None, on_redetect_xiadan: Optional[Callable[[], None]] = None, on_set_xiadan_path: Optional[Callable[[str], None]] = None, + on_set_tesseract_path: Optional[Callable[[str], None]] = None, on_apply_self_update: Optional[Callable[[], None]] = None, minimize_to_tray: bool = False, ): @@ -133,6 +152,7 @@ def __init__( self.on_exit_cb = on_exit self.on_redetect_xiadan = on_redetect_xiadan self.on_set_xiadan_path = on_set_xiadan_path + self.on_set_tesseract_path = on_set_tesseract_path self.on_apply_self_update = on_apply_self_update self._minimize_to_tray = minimize_to_tray @@ -162,9 +182,9 @@ def __init__( except Exception as e: logger.debug("set window iconbitmap failed: %s", e) - # 宽几何排版设计,锁定最小宽高 - self.root.geometry("740x480+200+200") - self.root.minsize(700, 420) + # 两栏都有常用操作,给右栏保留足够的按钮和状态文字宽度。 + self.root.geometry("900x620+160+120") + self.root.minsize(820, 540) # Windows + tray 可用时:关闭按钮最小化到托盘;否则真退出 self.root.protocol("WM_DELETE_WINDOW", self._on_close) @@ -237,9 +257,9 @@ def _build_ui(self) -> None: self._self_update_rendered = None # 记录上次渲染的 (status, version),避免每帧重排闪烁 # ========================================== - # 【左分栏】MCP 网关连接与监控舱 (固定宽 300px) + # 【左分栏】MCP 网关连接与监控舱 (固定宽 330px) # ========================================== - left_frame = tk.Frame(main_pane, bg="#f6f8fa", width=310) + left_frame = tk.Frame(main_pane, bg="#f6f8fa", width=330) self.left_frame = left_frame # 供自更新横幅 pack(before=...) 稳定定位 left_frame.pack(side="left", fill="both", padx=(12, 6), pady=(12, 4)) left_frame.pack_propagate(False) # 强制宽度锁定 @@ -410,26 +430,28 @@ def _build_ui(self) -> None: for s_idx in range(1, 4): step_frame = tk.Frame(self.ths_wizard_box, bg="#ffffff") step_frame.pack(fill="x", pady=3) + step_title_row = tk.Frame(step_frame, bg="#ffffff") + step_title_row.pack(fill="x") # 圆圈样式标签 circle = tk.Label( - step_frame, text=str(s_idx), bg="#fafbfc", fg="#57606a", + step_title_row, text=str(s_idx), bg="#fafbfc", fg="#57606a", bd=1, relief="solid", font=("Helvetica", 8, "bold"), width=2, height=1 ) circle.pack(side="left", padx=(0, 8)) text_lbl = tk.Label( - step_frame, text=step_names[s_idx - 1], bg="#ffffff", fg="#24292f", + step_title_row, text=step_names[s_idx - 1], bg="#ffffff", fg="#24292f", font=("Helvetica", 9) ) text_lbl.pack(side="left") hint_lbl = tk.Label( step_frame, text="", bg="#ffffff", fg="#e67e22", - font=("Helvetica", 9) + font=("Helvetica", 9), anchor="w", justify="left", wraplength=400, ) - hint_lbl.pack(side="left", padx=8) + hint_lbl.pack(fill="x", padx=(30, 0), pady=(2, 0)) self.ths_steps_display.append({ "step_num": s_idx, @@ -462,25 +484,63 @@ def _build_ui(self) -> None: ) self.change_path_btn.pack(side="right") - # 同花顺卡片操作底部 row + # 同花顺卡片操作区:操作与状态分行,避免窄屏时互相挤压。 self.ths_action_footer_row = tk.Frame(self.ths_body, bg="#ffffff") self.ths_action_footer_row.pack(fill="x", pady=(8, 0)) sep = tk.Frame(self.ths_action_footer_row, height=1, bg="#f0f2f5") sep.pack(fill="x", side="top", pady=(0, 6)) + self.ths_action_buttons_row = tk.Frame(self.ths_action_footer_row, bg="#ffffff") + self.ths_action_buttons_row.pack(fill="x") + self.open_ths_btn = tk.Button( - self.ths_action_footer_row, text="启动并打开同花顺", command=self._open_xiadan, + self.ths_action_buttons_row, text="启动并打开同花顺", command=self._open_xiadan, relief="flat", bg="#b8913b", fg="#ffffff", font=("Helvetica", 9, "bold"), padx=12, pady=4, cursor="hand2", bd=0 ) self.open_ths_btn.pack(side="left") + self.pick_xiadan_btn = tk.Button( + self.ths_action_buttons_row, text="选择 xiadan.exe...", command=self._pick_xiadan_path, + relief="flat", bg="#ffffff", fg="#0969da", font=("Helvetica", 8), + padx=8, pady=3, cursor="hand2", bd=0, + highlightbackground="#d0d7de", highlightthickness=1, + ) + self.pick_xiadan_btn.pack(side="left", padx=(8, 0)) + + self.redetect_xiadan_btn = tk.Button( + self.ths_action_buttons_row, text="重新检测", command=self._redetect_xiadan, + relief="flat", bg="#ffffff", fg="#57606a", font=("Helvetica", 8), + padx=8, pady=3, cursor="hand2", bd=0, + highlightbackground="#d0d7de", highlightthickness=1, + ) + self.redetect_xiadan_btn.pack(side="left", padx=(6, 0)) + self.ths_action_hint = tk.Label( self.ths_action_footer_row, text="未启动,请点击唤起客户端", - bg="#ffffff", fg="#57606a", font=("Helvetica", 9) + bg="#ffffff", fg="#57606a", font=("Helvetica", 9), anchor="w", justify="left", + wraplength=400, + ) + self.ths_action_hint.pack(fill="x", pady=(6, 0)) + + self.ocr_status_row = tk.Frame(self.ths_body, bg="#ffffff") + self.ocr_status_row.pack(fill="x", pady=(8, 0)) + self.ocr_status_label = tk.Label( + self.ocr_status_row, text="OCR: 检查中", bg="#ffffff", fg="#57606a", + font=("Helvetica", 8), anchor="w", justify="left", + ) + self.ocr_status_label.pack(fill="x") + self.ocr_status_row.bind( + "", + lambda event: self.ocr_status_label.config(wraplength=max(260, event.width - 2)), + ) + self.pick_tesseract_btn = tk.Button( + self.ocr_status_row, text="选择本地 Tesseract...", command=self._pick_tesseract_path, + relief="flat", bg="#ffffff", fg="#0969da", font=("Helvetica", 8), + padx=8, pady=3, cursor="hand2", bd=0, + highlightbackground="#d0d7de", highlightthickness=1, ) - self.ths_action_hint.pack(side="right") # 右分栏底部:退出程序排版 right_footer = tk.Frame(right_frame, bg="#f6f8fa") @@ -670,7 +730,7 @@ def _sync_state(self) -> None: self.ths_ready_box.pack(fill="x", pady=4) if snap.get("xiadan_path"): - self.ths_done_path_label.config(text=snap["xiadan_path"]) + self.ths_done_path_label.config(text=_display_path(snap["xiadan_path"])) self.ths_action_hint.config(text="✓ 客户端自检就绪", fg="#00c800") else: self.ths_status_badge.config(text="等待启动 ⏳", bg="#fffbe6", fg="#e67e22") @@ -706,6 +766,25 @@ def _sync_state(self) -> None: circle.config(text=str(s_num), bg="#fafbfc", fg="#999999", highlightbackground="#d0d7de") hint.config(text="") + # 3. OCR 状态与离线恢复入口 + ocr_status = snap.get("ocr_status", "checking") + ocr_message = snap.get("ocr_message", "") + ocr_text = { + "checking": "OCR: 检查中", + "installing": "OCR: 后台安装中(不影响连接)", + "ready": "OCR: 已就绪", + "unavailable": "OCR: 不可用(验证码与自选股识别不可用)", + }.get(ocr_status, "OCR: 状态未知") + self.ocr_status_label.config( + text=f"{ocr_text}{f' - {ocr_message}' if ocr_message else ''}", + fg="#00a35a" if ocr_status == "ready" else "#e67e22" if ocr_status == "installing" else "#cf222e" if ocr_status == "unavailable" else "#57606a", + ) + if ocr_status in ("installing", "unavailable"): + if not self.pick_tesseract_btn.winfo_ismapped(): + self.pick_tesseract_btn.pack(anchor="w", pady=(4, 0)) + elif self.pick_tesseract_btn.winfo_ismapped(): + self.pick_tesseract_btn.pack_forget() + # 3. 自更新提示横幅状态自适应 update_info = snap.get("self_update_info") update_status = snap.get("self_update_status", "idle") @@ -797,6 +876,28 @@ def _pick_xiadan_path(self) -> None: else: self.state.log(f"⚠ 路径设置回调未注册(选了 {path})") + def _redetect_xiadan(self) -> None: + if self.on_redetect_xiadan: + self.on_redetect_xiadan() + else: + self.state.log("⚠ 重新检测:未注册回调") + + def _pick_tesseract_path(self) -> None: + """选择离线安装或解压后的 tesseract.exe。""" + from tkinter import filedialog + + path = filedialog.askopenfilename( + parent=self.root, + title="选择 tesseract.exe", + filetypes=[("tesseract.exe", "tesseract.exe"), ("所有 exe", "*.exe"), ("所有文件", "*.*")], + ) + if not path: + return + if self.on_set_tesseract_path: + self.on_set_tesseract_path(path) + else: + self.state.log(f"⚠ Tesseract 路径设置回调未注册(选了 {path})") + def _on_footer_link_click(self, event=None) -> None: """官网链接点击处理""" import webbrowser diff --git a/src/trader/ws_client.py b/src/trader/ws_client.py index 54bc952..6bd1b8f 100644 --- a/src/trader/ws_client.py +++ b/src/trader/ws_client.py @@ -205,7 +205,11 @@ async def run(self) -> None: logger.error("握手失败:%s", result.error) self._set_state(ConnectionState.UNPAIRED) if result.should_clear_config: - config.save(config.TraderConfig(device_id=cfg.device_id)) + latest_cfg = config.load() + latest_cfg.agent_token = None + latest_cfg.account_name = None + latest_cfg.paired_at = None + config.save(latest_cfg) # 握手失败(如网络、服务不可用)退避 5 秒重试 await asyncio.sleep(5) continue @@ -316,7 +320,10 @@ async def _handle_frame(self, frame: dict[str, Any], origin_ws: Any = None) -> N if reason in ("token_invalid", "account_removed"): cfg = config.load() logger.warning("Token 失效或账户已移除,清空本地 agent_token 配置...") - config.save(config.TraderConfig(device_id=cfg.device_id)) + cfg.agent_token = None + cfg.account_name = None + cfg.paired_at = None + config.save(cfg) raise SessionRejectedException(reason) diff --git a/tests/test_main_window_state.py b/tests/test_main_window_state.py index e6fcce8..c7c53d2 100644 --- a/tests/test_main_window_state.py +++ b/tests/test_main_window_state.py @@ -30,3 +30,30 @@ def test_shared_state_self_update_fields_roundtrip(): assert snap["self_update_info"].latest_version == "0.6.0" assert snap["self_update_progress"] == (10, 100) assert snap["self_update_status"] == "downloading" + + +def test_shared_state_ocr_fields_roundtrip(): + from trader.main_window import SharedState + + state = SharedState() + assert state.snapshot()["ocr_status"] == "checking" + + state.update(ocr_status="unavailable", ocr_message="自动安装失败") + snap = state.snapshot() + + assert snap["ocr_status"] == "unavailable" + assert snap["ocr_message"] == "自动安装失败" + + +def test_display_path_keeps_drive_and_executable_name(): + from trader.main_window import _display_path + + path = r"C:\同花顺软件\同花顺\very-long-folder-name\xiadan.exe" + + assert _display_path(path, max_chars=26) == r"C:\...\xiadan.exe" + + +def test_display_path_leaves_short_paths_unchanged(): + from trader.main_window import _display_path + + assert _display_path(r"C:\ths\xiadan.exe") == r"C:\ths\xiadan.exe" diff --git a/tests/test_tesseract_installer.py b/tests/test_tesseract_installer.py index d8fbea1..1b75f15 100644 --- a/tests/test_tesseract_installer.py +++ b/tests/test_tesseract_installer.py @@ -54,6 +54,21 @@ def test_found_in_common_path_skips_winget(monkeypatch, tmp_path): assert result == str(fake_exe) +def test_manual_path_has_priority_over_path(monkeypatch, tmp_path): + manual_exe = tmp_path / "tesseract.exe" + manual_exe.write_text("x") + from trader import config + + monkeypatch.setattr( + config, + "load", + lambda: config.TraderConfig(device_id="test", tesseract_path_manual=str(manual_exe)), + ) + monkeypatch.setattr(tess.shutil, "which", lambda name: "C:/PATH/tesseract.exe") + + assert tess.detect_tesseract() == str(manual_exe) + + def test_missing_and_no_winget_returns_none(monkeypatch): _force_windows(monkeypatch) # tesseract 未装、winget 也不存在 @@ -106,6 +121,20 @@ async def fake_subprocess(*args, **kwargs): assert result is None +def test_winget_timeout_explains_local_fallback(monkeypatch): + _force_windows(monkeypatch) + monkeypatch.setattr(tess.shutil, "which", lambda name: "C:/winget.exe" if name == "winget" else None) + monkeypatch.setattr(tess, "TESSERACT_COMMON_PATHS", []) + + async def fake_subprocess(*args, **kwargs): + return _FakeProc(out=b"InternetOpenUrl() failed.", rc=-2147012894) + + monkeypatch.setattr(tess.asyncio, "create_subprocess_exec", fake_subprocess) + logs = [] + assert asyncio.run(tess.ensure_tesseract(on_log=logs.append)) is None + assert any("超时" in message and "本地" in message for message in logs) + + def test_non_windows_returns_none(monkeypatch): monkeypatch.setattr(tess.platform, "system", lambda: "Darwin") result = asyncio.run(tess.ensure_tesseract())