diff --git a/CLAUDE.md b/CLAUDE.md index 6aa0f79..8566885 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -7,6 +7,11 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co 自动登录 5 家物流承运商工作台，下载"应到 / 实到"货物数据，离线比对出**应到未到** 异常运单并汇总成 Excel。4 个网页站点 + 1 个 Electron 应用（安能）。 +**两种运行模式**（阶段1 起）： +- **交互模式** `main_router.py`：人工调试 / 操作，交互菜单（登录、触发下载、[12] 状态盘）。 +- **服务模式** `server.py`：常驻 + FastAPI，客户端经 HTTP 触发任务、查状态、下载数据（API 文档 `/docs`）。 +- 两者共享 `runtime.py`（启动 / 就绪 / 任务派发 / 心跳）与 `state_store.py`（SQLite 状态持久化）。 + ## 常用命令 所有 Python 一律在项目虚拟环境 `.venv` 中运行（Windows 下可直接用 @@ -20,6 +25,9 @@ playwright install chromium # 运行主程序（交互式菜单，详见 main_router 的 run_multi_site_daemon） .venv/Scripts/python.exe main_router.py +# 服务模式（常驻 + FastAPI，客户端经 HTTP 触发；默认 :8000，API 文档见 /docs） +.venv/Scripts/python.exe server.py + # 单站点联调：在 config.yaml 设 debug.enabled=true + debug.target_site=顺心|百世|中通|韵达|安能 # 网页站：只挂载该站；安能：只启动 Electron 应用。 @@ -38,6 +46,14 @@ playwright install chromium ## 架构（big picture） +### 运行模式与共享核心（阶段0/1 重构） +- **`runtime.py`**：两种模式共享的核心——`launch_and_prepare`（启动 Playwright + 各站就绪 + 弹窗 + 心跳初值，阻塞至就绪）、`dispatch_task(ctx, {site,kind})`（派发任务，掉登录直接判 failed）、`run_heartbeat`、`probe_site_login/probe_data_file`、`RuntimeContext.stop()`。常量 `SITES_CONFIG/READY_SELECTORS/APP_SITES/HEARTBEAT_INTERVAL/DATA_FILENAMES` 在此。 +- **`state_store.py`**：SQLite 状态持久化（`state/state.db`）。`site_status`（登录态 + 数据态 + 时间戳，心跳刷新）、`task_history`（任务记录）。重启不丢。 +- **`main_router.py`**：交互模式，菜单循环（input 后台线程 + `_await_command` + `dispatch_task` + 心跳）。 +- **`server.py`**：服务模式。**FastAPI（主线程）+ Playwright worker（独立线程）**——主线程处理 HTTP（绝不碰 Playwright），worker 独占 page 操作，经 `task_queue` + `state_store` 通信。API：`POST/GET /tasks`、`GET /status`、`GET /data/{file}`。 +- **关键线程约束**：Playwright sync 对象绑定创建它的线程；`launch_and_prepare`（含 `sync_playwright().start()`）必须在持有 Playwright 的线程调用（交互=主线程，服务=worker 线程）。FastAPI 路由绝不访问 page。 +- 站点模块 `site_*.py` 的 `with_retry` 返回 `True/False`（成功 / 放弃），供 `dispatch_task` 判成败。 + ### 两套驱动模态 —— 这是理解全局的关键 - **网页 4 站**（顺心/百世/中通/韵达）：`main_router` 用 Playwright 开 chromium， 每站一个 `page`，流程函数签名为 `xxx_download_impl(page)`。**例外：顺心是双账号** diff --git a/main_router.py b/main_router.py index 8cb88db..475f229 100644 --- a/main_router.py +++ b/main_router.py @@ -1,142 +1,35 @@ # main_router.py +# +# 交互菜单模式入口（调试 / 人工操作）。核心 Playwright 管理、任务派发、心跳 +# 已抽到 runtime.py 共享；本文件只保留交互菜单与自动化测试。 +# 服务模式（常驻 + FastAPI 接收指令）见 server.py。 import os import queue -import socket -import subprocess import threading import time -import urllib.request -from datetime import datetime import yaml -from playwright.sync_api import sync_playwright -from paths import DOWNLOAD_DIR, CONFIG_PATH +from paths import CONFIG_PATH +from runtime import ( + APP_SITES, + HEARTBEAT_INTERVAL, + dispatch_task, + launch_and_prepare, + run_heartbeat, +) -# 阶段0：状态持久化（站点登录态 / 数据态，心跳刷新；重启不丢） import state_store -# 导入抽离出去的各个网点模块 +# 各站点模块（自动化测试 + 比对用；任务派发在 runtime） import site_shunxin import site_baishi import site_zto import site_yunda import site_anneng - -# 应到未到比对（全站点，离线处理 downloads/ 下的应到/实到数据） import expected_undelivered -# 各网页站点首页 URL 统一取自对应站点模块的 HOME_URL（单一来源，不在两处维护）。 -SITES_CONFIG = { - "顺心": site_shunxin.HOME_URL, - "百世": site_baishi.HOME_URL, - "中通": site_zto.HOME_URL, - "韵达": site_yunda.HOME_URL, -} - -# ==================================================================== -# 站点就绪特征：每个站点登录成功进入工作台后的标志性控件 -# ==================================================================== -READY_SELECTORS = { - "顺心": 'h1:has-text("盟商门户网")', - "百世": 'h1[title="百世快运"]', - "中通": '.logo:has-text("网点版")', - "韵达": '.el-menu-item:has-text("首页")', -} - -# 安能是 Electron 桌面应用（不是 Playwright 打开的网页），需要单独启动。 -APP_SITES = {"安能"} - -# 阶段0：心跳间隔（秒）——各站就绪后，每隔此时长探测一次登录态/数据态 -HEARTBEAT_INTERVAL = 30 - -# 各站最终数据文件名（用于探测"数据是否已跑出来"）；百世为单流程，仅应到未到 -DATA_FILENAMES = { - "顺心": {"expected": "顺心-应到货物数据.xlsx", "actual": "顺心-实到货物数据.xlsx"}, - "中通": {"expected": "中通-应到货物数据.xlsx", "actual": "中通-实到货物数据.xlsx"}, - "韵达": {"expected": "韵达-应到货物数据.xlsx", "actual": "韵达-实到货物数据.xlsx"}, - "安能": {"expected": "安能-应到货物数据.xlsx", "actual": "安能-实到货物数据.xlsx"}, - "百世": {"expected": "百世-应到未到货物数据.xlsx", "actual": ""}, -} - - -def probe_site_login(site_name, pages_map): - """探测单站是否登录（复用就绪轮询判据）。任何异常一律返回 False。 - - 只在主线程调用（Playwright sync 对象绑定主线程）。 - """ - try: - if site_name not in pages_map: - return False - if site_name == "安能": - return site_anneng.anneng_ready() - if site_name == "顺心": - return all( - pg.locator(READY_SELECTORS["顺心"]).is_visible(timeout=500) - for pg in pages_map["顺心"] - ) - return ( - pages_map[site_name] - .locator(READY_SELECTORS[site_name]) - .is_visible(timeout=500) - ) - except Exception: - return False - - -def probe_data_file(site_name, kind): - """探测单站应到/实到数据文件是否存在且为今天。返回 (is_today: bool, mtime_str)。""" - fname = DATA_FILENAMES.get(site_name, {}).get(kind, "") - if not fname: - return (False, "") - path = os.path.join(DOWNLOAD_DIR, fname) - if not os.path.exists(path): - return (False, "") - dt = datetime.fromtimestamp(os.path.getmtime(path)) - is_today = dt.date() == datetime.now().date() - return (is_today, dt.strftime("%Y-%m-%d %H:%M:%S")) - - -def _find_free_port(): - """让操作系统分配一个空闲端口，避免固定端口冲突。""" - with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: - s.bind(("127.0.0.1", 0)) - return s.getsockname()[1] - - -def _wait_cdp_up(port, timeout=60.0): - """轮询直到 CDP 调试端口就绪（应用启动需要时间）。""" - deadline = time.monotonic() + timeout - while time.monotonic() < deadline: - try: - with urllib.request.urlopen( - f"http://localhost:{port}/json/version", timeout=2 - ) as resp: - if resp.status == 200: - return True - except Exception: - pass - time.sleep(1) - return False - - -def launch_anneng(app_path): - """以调试模式启动安能 Electron 应用（自动选取空闲端口），返回子进程对象。 - - 启动后请在应用内手动登录；就绪状态由 run_multi_site_daemon 的就绪轮询判断。 - """ - port = _find_free_port() - print(f">> 以调试模式启动【安能】应用（端口 {port}）：{app_path}") - proc = subprocess.Popen([app_path, f"--remote-debugging-port={port}"]) - site_anneng.set_cdp_port(port) - if not _wait_cdp_up(port): - raise RuntimeError( - f"安能应用调试端口 {port} 未就绪——可能应用已在运行（单实例），" - "请先关闭已有的安能窗口再试" - ) - return proc - def run_undelivered_compare(): """应到未到比对（全站点）：调用 expected_undelivered，读 downloads/ 下的应到/实到 @@ -279,399 +172,174 @@ def _print_test_report(results): print("====================================================") -def run_multi_site_daemon(): - """多站点自动化主控流程""" +# ==================================================================== +# 交互菜单模式 +# ==================================================================== - # 1. 读取配置文件 + +def _read_debug_config(): + """读 config.yaml 的 debug 段，返回 (debug_mode, debug_target)。""" debug_mode = False debug_target = "" - anneng_app_path = "" try: if os.path.exists(CONFIG_PATH): with open(CONFIG_PATH, "r", encoding="utf-8") as f: config = yaml.safe_load(f) or {} debug_mode = (config.get("debug", {}) or {}).get("enabled", False) debug_target = (config.get("debug", {}) or {}).get("target_site", "") - anneng_app_path = (config.get("anneng", {}) or {}).get("app_path", "") except Exception as e: print(f"⚠️ 读取 config.yaml 异常，将使用全量模式启动: {e}") + return debug_mode, debug_target - # 动态确定需要挂载启动的网页站点；安能（Electron 应用）单独标记 - anneng_active = False - if debug_mode: - if debug_target in SITES_CONFIG: - print(f"\n🛠️ 【调试模式】仅加载目标站点: [{debug_target}]") - active_sites = {debug_target: SITES_CONFIG[debug_target]} - elif debug_target == "安能": - print(f"\n🛠️ 【调试模式】仅加载目标站点: [安能]") - active_sites = {} - anneng_active = True - else: - active_sites = dict(SITES_CONFIG) - anneng_active = True - else: - active_sites = dict(SITES_CONFIG) - anneng_active = True - if anneng_active and not anneng_app_path: - print("⚠️ 已启用安能但 config.yaml 未配置 anneng.app_path，将跳过安能。") - anneng_active = False +# 菜单编号 → 任务规格（dispatch_task 消费） +CHOICE_TO_TASK = { + "1": {"site": "顺心", "kind": "expected"}, + "2": {"site": "顺心", "kind": "actual"}, + "3": {"site": "百世", "kind": "undelivered"}, + "4": {"site": "中通", "kind": "expected"}, + "5": {"site": "中通", "kind": "actual"}, + "6": {"site": "韵达", "kind": "expected"}, + "7": {"site": "韵达", "kind": "actual"}, + "10": {"site": "安能", "kind": "expected"}, + "11": {"site": "安能", "kind": "actual"}, + "9": {"site": "__compare__", "kind": "compare"}, +} - with sync_playwright() as p: - browser = p.chromium.launch(headless=False) - context = browser.new_context(viewport={"width": 1920, "height": 1080}) - pages_map = {} +def _interactive_menu_loop(ctx): + """交互菜单循环：input 后台线程 + _await_command + dispatch_task + 心跳 + 状态盘。 - print("\n====================================================") - print("【启动】正在打开各站点页面...") - print("====================================================") + 所有 page 操作经 runtime（主线程），满足 Playwright sync 线程安全。 + """ + pages_map = ctx.pages_map + sites_to_watch = ctx.sites_to_watch - for site_name, url in active_sites.items(): - if site_name == "顺心": - # 顺心：两个归属地账号在同一窗口各开一个标签页（顺心站点支持 - # 同浏览器双账号并存）；pages_map["顺心"] 存为 page 列表。 - sx_pages = [] - for acct in range(1, 3): - print(f">> 正在启动【顺心】账号{acct}标签页: {url}") - sx_page = context.new_page() - sx_page.goto(url) - sx_pages.append(sx_page) - pages_map["顺心"] = sx_pages - else: - print(f">> 正在启动【{site_name}】页面: {url}") - page = context.new_page() - page.goto(url) - pages_map[site_name] = page + def is_site_ready(site_name): + if site_name not in pages_map: + print(f"\n🚫 站点 [{site_name}] 未加载（当前为调试模式），已跳过。") + return False + return True - # 安能：以调试模式启动 Electron 应用（非 Playwright 网页） - anneng_proc = None - if anneng_active: - try: - anneng_proc = launch_anneng(anneng_app_path) - pages_map["安能"] = True # 哨兵：表示已启动（无 Playwright page 对象） - except Exception as e: - print(f"⚠️ 启动安能应用失败，已跳过安能：{e}") - anneng_active = False - - print("\n====================================================") - print("【登录检测】正在准备各站点登录...") - print("====================================================") - - # 针对支持纯代码自动登录的站点，在此处前置注入登录事件 - if "韵达" in pages_map: - try: - pages_map["韵达"].bring_to_front() - site_yunda.yunda_login(pages_map["韵达"]) - except Exception as e: - print(f" ⚠️ 韵达前置自动登录模块发生波动: {e}") - - # ==================================================================== - # 就绪轮询 (Ready Guard Polling) - # 自动识别各站点登录完成状态，无需手动回车 - # ==================================================================== - ready_status = {site: False for site in active_sites.keys()} - if anneng_active: - ready_status["安能"] = False - - print("\n>> 正在轮询各站点就绪状态 (自动登录或手动登录均可)...") - while not all(ready_status.values()): - for site_name in list(ready_status.keys()): - if ready_status[site_name]: - continue - try: - if site_name == "安能": - # 安能走 CDP 判断主页是否就绪（连不上返回 False，不抛异常） - ok = site_anneng.anneng_ready() - elif site_name == "顺心": - # 顺心双账号：两个 page 都进主页才算就绪 - ok = all( - pg.locator(READY_SELECTORS["顺心"]).is_visible(timeout=500) - for pg in pages_map["顺心"] - ) - else: - # 0.5 秒轻量探测，避免阻塞主循环 - ok = ( - pages_map[site_name] - .locator(READY_SELECTORS[site_name]) - .is_visible(timeout=500) - ) - if ok: - ready_status[site_name] = True - print(f" ✅ 【{site_name}】已检测到主页，登录就绪。") - except Exception: - pass - - pending_sites = [s for s, ready in ready_status.items() if not ready] - if pending_sites: - print( - f" ⏳ 等待以下站点完成登录: [{', '.join(pending_sites)}] ... " - "(请在浏览器/应用中操作)" - ) - time.sleep(3) # 等待 3 秒后进行下一轮检查 - - print("\n====================================================") - print("【准备】所有站点已就绪，正在清理初始弹窗...") - print("====================================================") - - # 顺心：处理初始弹窗（双账号两个 page 各处理一遍） - if "顺心" in pages_map: - for acct, sx_page in enumerate(pages_map["顺心"], start=1): - try: - sx_page.bring_to_front() - print(f">> 正在处理【顺心】账号{acct}弹窗与遮罩...") - sx_page.locator("a").nth(4).click(timeout=2000) - sx_page.wait_for_timeout(500) - sx_page.get_by_role("button", name="Close").click(timeout=2000) - sx_page.wait_for_timeout(500) - sx_page.get_by_role("button", name="不再询问").click(timeout=2000) - print(f" ✅ 【顺心】账号{acct}初始弹窗处理完成。") - except Exception: - pass # 环境可能很干净没有弹窗，无视报错 - - # 百世：循环关闭初始弹窗 - if "百世" in pages_map: - try: - bs_page = pages_map["百世"] - bs_page.bring_to_front() - print(">> 正在处理【百世】阅读完毕与关闭按钮...") - for round_idx in range(4): - handled_any = False - try: - read_btns = bs_page.locator("button:has-text('阅读完毕')") - if read_btns.count() > 0: - for i in range(read_btns.count()): - if read_btns.nth(i).is_visible(timeout=500): - read_btns.nth(i).click() - handled_any = True - except Exception: - pass - try: - if bs_page.locator("button:has-text('关 闭')").is_visible( - timeout=500 - ): - bs_page.locator("button:has-text('关 闭')").click() - handled_any = True - except Exception: - pass - if not handled_any: - break - bs_page.wait_for_timeout(800) - print(" ✅ 【百世】初始弹窗处理完成。") - except Exception: - pass - - if "中通" in pages_map: - print(" ✅ 【中通】已就绪。") - - if "韵达" in pages_map: - print(" ✅ 【韵达】已就绪。") - - if "安能" in pages_map: - print(" ✅ 【安能】已就绪。") - - def is_site_ready(site_name): - if site_name not in pages_map: - print(f"\n🚫 站点 [{site_name}] 未加载（当前为调试模式），已跳过。") - return False - return True - - # ================================================================== - # 阶段0：后台心跳 + 状态持久化 - # input 放后台线程（只读 stdin，不碰 Playwright），主线程在 _await_command - # 里轮询命令队列并定期跑心跳；所有 page 操作仍在主线程，满足 Playwright - # sync 的线程安全。 - # ================================================================== - state_store.init_db() - sites_to_watch = list(ready_status.keys()) - # 就绪轮询刚通过 → 各站视为已登录写一次初值，后续心跳校正 - for _site in sites_to_watch: - state_store.set_login_state(_site, True) - - last_heartbeat = 0.0 - - def _run_heartbeat(): - """一轮心跳：探测各站登录态 + 数据文件，写状态库；登录态变化时提示。""" - nonlocal last_heartbeat - prev = state_store.get_all_status() - for site_name in sites_to_watch: - logged_in = probe_site_login(site_name, pages_map) - prev_login = prev.get(site_name, {}).get("login_state") - state_store.set_login_state(site_name, logged_in) - now_login = state_store.LOGIN_IN if logged_in else state_store.LOGIN_OUT - if prev_login and prev_login not in ( - now_login, - state_store.LOGIN_UNKNOWN, - ): - print( - f"\n ⚠️【{site_name}】登录态变化: {prev_login} → {now_login}" - ) - for kind in ("expected", "actual"): - ready, gen_at = probe_data_file(site_name, kind) - state_store.set_data_state(site_name, kind, ready, gen_at) - last_heartbeat = time.monotonic() - - def _print_status_board(): - """打印各站登录态 / 数据就绪状态盘（菜单 [12]）。""" - print("\n====================== 站点状态盘 ======================") - status = state_store.get_all_status() - if not status: - print(" （暂无状态记录）") - print("======================================================") - return - login_text = { - state_store.LOGIN_IN: "✅ 已登录", - state_store.LOGIN_OUT: "❌ 未登录", - state_store.LOGIN_UNKNOWN: "❔ 未知", - } - for site_name in sites_to_watch: - s = status.get(site_name) - if not s: - continue - login_mark = login_text.get(s["login_state"], s["login_state"]) - exp = ( - f"应到{'✅' if s['expected_ready'] else '—'} " - f"{s['expected_generated_at'] or '无'}" - ) - act = ( - f"实到{'✅' if s['actual_ready'] else '—'} " - f"{s['actual_generated_at'] or '无'}" - ) - print( - f" 【{site_name}】 {login_mark} | {exp} | {act} " - f"| 探测于 {s['login_checked_at']}" - ) - print("======================================================") - - command_queue = queue.Queue() - - def _input_loop(): - """后台线程：读用户输入塞进队列。不碰任何 Playwright 对象。""" - while True: - try: - command_queue.put(input()) - except EOFError: - return - - threading.Thread(target=_input_loop, daemon=True).start() - - def _await_command(): - """等待一条命令（期间跑心跳）。命令到来即返回。""" - nonlocal last_heartbeat - while True: - try: - return command_queue.get(timeout=0.5) - except queue.Empty: - if time.monotonic() - last_heartbeat >= HEARTBEAT_INTERVAL: - _run_heartbeat() + last_heartbeat = 0.0 + command_queue = queue.Queue() + def _await_command(): + nonlocal last_heartbeat while True: - print("\n====================================================") - print(" 物流数据下载主菜单 ") - if debug_mode: - print(f" [ 调试模式，仅加载: {debug_target} ]") - print("====================================================") - print(" 模块一：【顺心】数据处理流") - print(" [1] 执行 - 应到货物数据下载") - print(" [2] 执行 - 实到货物数据下载") - print("-" * 52) - print(" 模块二：【百世】数据处理流") - print(" [3] 执行 - 一键提取应到未到异常数据") - print("-" * 52) - print(" 模块三：【中通】数据处理流") - print(" [4] 执行 - 应到货物数据下载") - print(" [5] 执行 - 实到货物数据下载") - print("-" * 52) - print(" 模块四：【韵达】数据处理流") - print(" [6] 执行 - 应到货物数据下载") - print(" [7] 执行 - 实到货物数据下载") - print("-" * 52) - print(" 模块五：【安能】数据处理流（Electron 应用）") - print(" [10] 执行 - 应到货物数据下载（运单信息）") - print(" [11] 执行 - 实到货物数据下载（网点到件扫描）") - print("-" * 52) - print(" 自动化测试") - print(" [8] 执行 - 全站点下载流程自动化测试 (交叉跑通校验)") - print("-" * 52) - print(" 全局离线数据处理") + try: + return command_queue.get(timeout=0.5) + except queue.Empty: + if time.monotonic() - last_heartbeat >= HEARTBEAT_INTERVAL: + run_heartbeat(ctx) + last_heartbeat = time.monotonic() + + def _print_status_board(): + print("\n====================== 站点状态盘 ======================") + status = state_store.get_all_status() + if not status: + print(" （暂无状态记录）") + print("======================================================") + return + login_text = { + state_store.LOGIN_IN: "✅ 已登录", + state_store.LOGIN_OUT: "❌ 未登录", + state_store.LOGIN_UNKNOWN: "❔ 未知", + } + for site_name in sites_to_watch: + s = status.get(site_name) + if not s: + continue + login_mark = login_text.get(s["login_state"], s["login_state"]) + exp = f"应到{'✅' if s['expected_ready'] else '—'} {s['expected_generated_at'] or '无'}" + act = f"实到{'✅' if s['actual_ready'] else '—'} {s['actual_generated_at'] or '无'}" print( - " [9] 执行 - 应到未到比对（全站点汇总，输出 output/应到未到数据.xlsx）" + f" 【{site_name}】 {login_mark} | {exp} | {act} " + f"| 探测于 {s['login_checked_at']}" ) - print("-" * 52) - print(" 站点状态") - print(" [12] 查看 - 各站登录态 / 数据就绪状态") - print("-" * 52) - print(" [0] 退出系统") - print("====================================================") - - print("请输入任务编号并回车: ", end="", flush=True) - choice = _await_command() + print("======================================================") + def _input_loop(): + while True: try: - if choice == "1" and is_site_ready("顺心"): - site_shunxin.shunxin_expected_download(pages_map["顺心"]) - elif choice == "2" and is_site_ready("顺心"): - site_shunxin.shunxin_actual_download(pages_map["顺心"]) - elif choice == "3" and is_site_ready("百世"): - page = pages_map["百世"] - page.bring_to_front() - site_baishi.baishi_download_undelivered_data(page) - elif choice == "4" and is_site_ready("中通"): - page = pages_map["中通"] - page.bring_to_front() - site_zto.zto_expected_download(page) - elif choice == "5" and is_site_ready("中通"): - page = pages_map["中通"] - page.bring_to_front() - site_zto.zto_actual_download(page) - elif choice == "6" and is_site_ready("韵达"): - page = pages_map["韵达"] - page.bring_to_front() - site_yunda.yunda_expected_download(page) - elif choice == "7" and is_site_ready("韵达"): - page = pages_map["韵达"] - page.bring_to_front() - site_yunda.yunda_actual_download(page) - elif choice == "10" and is_site_ready("安能"): - site_anneng.anneng_expected_download() - elif choice == "11" and is_site_ready("安能"): - site_anneng.anneng_actual_download() - elif choice == "8": - run_automation_test(pages_map) - elif choice == "9": - run_undelivered_compare() - elif choice == "12": - _print_status_board() - elif choice == "0": - print("\n正在关闭浏览器并退出...") - break - else: - if choice not in [ - "1", - "2", - "3", - "4", - "5", - "6", - "7", - "8", - "9", - "10", - "11", - "12", - "0", - ]: - print("\n⚠️ 无效输入，请查证后回车。") - except Exception as e: - print(f"❌ 任务调度异常: {e}") + command_queue.put(input()) + except EOFError: + return - browser.close() - if anneng_proc is not None: - try: - anneng_proc.terminate() - print("已关闭安能应用。") - except Exception: - pass + threading.Thread(target=_input_loop, daemon=True).start() + + while True: + print("\n====================================================") + print(" 物流数据下载主菜单 ") + if ctx.debug_mode: + print(f" [ 调试模式，仅加载: {ctx.debug_target} ]") + print("====================================================") + print(" 模块一：【顺心】数据处理流") + print(" [1] 执行 - 应到货物数据下载") + print(" [2] 执行 - 实到货物数据下载") + print("-" * 52) + print(" 模块二：【百世】数据处理流") + print(" [3] 执行 - 一键提取应到未到异常数据") + print("-" * 52) + print(" 模块三：【中通】数据处理流") + print(" [4] 执行 - 应到货物数据下载") + print(" [5] 执行 - 实到货物数据下载") + print("-" * 52) + print(" 模块四：【韵达】数据处理流") + print(" [6] 执行 - 应到货物数据下载") + print(" [7] 执行 - 实到货物数据下载") + print("-" * 52) + print(" 模块五：【安能】数据处理流（Electron 应用）") + print(" [10] 执行 - 应到货物数据下载（运单信息）") + print(" [11] 执行 - 实到货物数据下载（网点到件扫描）") + print("-" * 52) + print(" 自动化测试") + print(" [8] 执行 - 全站点下载流程自动化测试 (交叉跑通校验)") + print("-" * 52) + print(" 全局离线数据处理") + print(" [9] 执行 - 应到未到比对（全站点汇总，输出 output/应到未到数据.xlsx）") + print("-" * 52) + print(" 站点状态") + print(" [12] 查看 - 各站登录态 / 数据就绪状态") + print("-" * 52) + print(" [0] 退出系统") + print("====================================================") + + print("请输入任务编号并回车: ", end="", flush=True) + choice = _await_command() + + try: + if choice in CHOICE_TO_TASK: + task = CHOICE_TO_TASK[choice] + site = task["site"] + if site == "__compare__" or is_site_ready(site): + status, error = dispatch_task(ctx, task) + if status == state_store.TASK_FAILED: + print(f"❌ 任务失败: {error}") + elif choice == "8": + run_automation_test(pages_map) + elif choice == "12": + _print_status_board() + elif choice == "0": + break + elif choice.strip() != "": + print("\n⚠️ 无效输入，请查证后回车。") + except Exception as e: + print(f"❌ 任务调度异常: {e}") + + +def run_multi_site_daemon(): + """多站点自动化主控流程（交互菜单模式）。 + + 启动 → 等待各站登录就绪 → 进入交互菜单；退出时关闭浏览器与安能。 + """ + debug_mode, debug_target = _read_debug_config() + ctx = launch_and_prepare(debug_mode, debug_target) + try: + _interactive_menu_loop(ctx) + finally: + print("\n正在关闭浏览器并退出...") + ctx.stop() print("程序已退出。") diff --git a/requirements.txt b/requirements.txt index 8a493c0..3a67faf 100644 --- a/requirements.txt +++ b/requirements.txt @@ -3,3 +3,5 @@ playwright>=1.40.0 openpyxl>=3.1.0 PyYAML>=6.0 websocket-client>=1.0.0 +fastapi>=0.110.0 +uvicorn>=0.27.0 diff --git a/runtime.py b/runtime.py new file mode 100644 index 0000000..b4d34e3 --- /dev/null +++ b/runtime.py @@ -0,0 +1,447 @@ +# runtime.py +# 阶段1：服务化共享核心。把"启动浏览器 + 各站就绪 + 弹窗 + 心跳初值" +# (launch_and_prepare)、"执行一条任务"(dispatch_task)、"一轮心跳"(run_heartbeat) +# 抽出来，供 +# - main_router.py（交互菜单模式：调试 / 人工操作） +# - server.py（FastAPI 服务模式：常驻 + 接收 API 指令） +# 共同复用，避免两处重复维护。 +# +# 线程模型：launch_and_prepare 内 sync_playwright().start() 必须在"持有 Playwright 的 +# 线程"调用（交互模式=主线程；服务模式=worker 线程）。该线程独占所有 page 操作； +# 其他线程（如 FastAPI 路由）只能经 task_queue + state_store 与之通信，绝不跨线程 +# 访问 page。 + +import os +import socket +import subprocess +import time +import urllib.request +from datetime import datetime + +import yaml +from playwright.sync_api import sync_playwright + +from paths import DOWNLOAD_DIR, CONFIG_PATH + +import state_store +import site_shunxin +import site_baishi +import site_zto +import site_yunda +import site_anneng +import expected_undelivered # dispatch 的 compare 任务用 + +# 各网页站点首页 URL（单一来源，取自各站点模块 HOME_URL） +SITES_CONFIG = { + "顺心": site_shunxin.HOME_URL, + "百世": site_baishi.HOME_URL, + "中通": site_zto.HOME_URL, + "韵达": site_yunda.HOME_URL, +} + +# 站点就绪特征：登录成功进入工作台后的标志性控件 +READY_SELECTORS = { + "顺心": 'h1:has-text("盟商门户网")', + "百世": 'h1[title="百世快运"]', + "中通": '.logo:has-text("网点版")', + "韵达": '.el-menu-item:has-text("首页")', +} + +# 安能是 Electron 桌面应用（不是 Playwright 网页），单独启动 +APP_SITES = {"安能"} + +# 心跳间隔（秒） +HEARTBEAT_INTERVAL = 30 + +# 各站最终数据文件名（探测"数据是否已跑出来"）；百世为单流程 +DATA_FILENAMES = { + "顺心": {"expected": "顺心-应到货物数据.xlsx", "actual": "顺心-实到货物数据.xlsx"}, + "中通": {"expected": "中通-应到货物数据.xlsx", "actual": "中通-实到货物数据.xlsx"}, + "韵达": {"expected": "韵达-应到货物数据.xlsx", "actual": "韵达-实到货物数据.xlsx"}, + "安能": {"expected": "安能-应到货物数据.xlsx", "actual": "安能-实到货物数据.xlsx"}, + "百世": {"expected": "百世-应到未到货物数据.xlsx", "actual": ""}, +} + + +# ============================ 安能启动（CDP）============================ + + +def _find_free_port(): + """让操作系统分配一个空闲端口，避免固定端口冲突。""" + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: + s.bind(("127.0.0.1", 0)) + return s.getsockname()[1] + + +def _wait_cdp_up(port, timeout=60.0): + """轮询直到 CDP 调试端口就绪（应用启动需要时间）。""" + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + try: + with urllib.request.urlopen( + f"http://localhost:{port}/json/version", timeout=2 + ) as resp: + if resp.status == 200: + return True + except Exception: + pass + time.sleep(1) + return False + + +def launch_anneng(app_path): + """以调试模式启动安能 Electron 应用（自动选取空闲端口），返回子进程对象。""" + port = _find_free_port() + print(f">> 以调试模式启动【安能】应用（端口 {port}）：{app_path}") + proc = subprocess.Popen([app_path, f"--remote-debugging-port={port}"]) + site_anneng.set_cdp_port(port) + if not _wait_cdp_up(port): + raise RuntimeError( + f"安能应用调试端口 {port} 未就绪——可能应用已在运行（单实例），" + "请先关闭已有的安能窗口再试" + ) + return proc + + +# ============================ 探测函数 ============================ + + +def probe_site_login(site_name, pages_map): + """探测单站是否登录（复用就绪判据）。任何异常一律返回 False。 + + 只在 Playwright 所属线程调用。 + """ + try: + if site_name not in pages_map: + return False + if site_name == "安能": + return site_anneng.anneng_ready() + if site_name == "顺心": + return all( + pg.locator(READY_SELECTORS["顺心"]).is_visible(timeout=500) + for pg in pages_map["顺心"] + ) + return ( + pages_map[site_name] + .locator(READY_SELECTORS[site_name]) + .is_visible(timeout=500) + ) + except Exception: + return False + + +def probe_data_file(site_name, kind): + """探测单站应到/实到数据文件是否存在且为今天。返回 (is_today, mtime_str)。""" + fname = DATA_FILENAMES.get(site_name, {}).get(kind, "") + if not fname: + return (False, "") + path = os.path.join(DOWNLOAD_DIR, fname) + if not os.path.exists(path): + return (False, "") + dt = datetime.fromtimestamp(os.path.getmtime(path)) + is_today = dt.date() == datetime.now().date() + return (is_today, dt.strftime("%Y-%m-%d %H:%M:%S")) + + +# ============================ 运行上下文 ============================ + + +class RuntimeContext: + """launch_and_prepare 的返回值，持有 Playwright 运行所需对象。""" + + def __init__( + self, + pw, + browser, + pages_map, + ready_status, + anneng_proc, + sites_to_watch, + debug_mode, + debug_target, + ): + self.pw = pw + self.browser = browser + self.pages_map = pages_map + self.ready_status = ready_status + self.anneng_proc = anneng_proc + self.sites_to_watch = sites_to_watch + self.debug_mode = debug_mode + self.debug_target = debug_target + + def stop(self): + """关闭 browser + 安能 + Playwright。退出时调用。""" + try: + self.browser.close() + except Exception: + pass + if self.anneng_proc is not None: + try: + self.anneng_proc.terminate() + print("已关闭安能应用。") + except Exception: + pass + try: + self.pw.stop() + except Exception: + pass + + +# ============================ 启动 + 就绪 + 弹窗 ============================ + + +def launch_and_prepare(debug_mode=False, debug_target=""): + """启动 Playwright + 各站 page + 就绪轮询 + 弹窗清理 + 心跳初值，返回 RuntimeContext。 + + 必须在"持有 Playwright 的线程"调用（交互模式主线程 / 服务模式 worker 线程）。 + 阻塞至所有站点登录就绪才返回。 + """ + # 1. 读 config（anneng.app_path） + anneng_app_path = "" + try: + if os.path.exists(CONFIG_PATH): + with open(CONFIG_PATH, "r", encoding="utf-8") as f: + config = yaml.safe_load(f) or {} + anneng_app_path = (config.get("anneng", {}) or {}).get("app_path", "") + except Exception as e: + print(f"⚠️ 读取 config.yaml 异常: {e}") + + # 2. 确定要挂载的网页站点 + 安能标记 + anneng_active = False + if debug_mode: + if debug_target in SITES_CONFIG: + print(f"\n🛠️ 【调试模式】仅加载目标站点: [{debug_target}]") + active_sites = {debug_target: SITES_CONFIG[debug_target]} + elif debug_target == "安能": + print(f"\n🛠️ 【调试模式】仅加载目标站点: [安能]") + active_sites = {} + anneng_active = True + else: + active_sites = dict(SITES_CONFIG) + anneng_active = True + else: + active_sites = dict(SITES_CONFIG) + anneng_active = True + + if anneng_active and not anneng_app_path: + print("⚠️ 已启用安能但 config.yaml 未配置 anneng.app_path，将跳过安能。") + anneng_active = False + + # 3. 启动 Playwright（不用 with，改 .start()，由 RuntimeContext.stop() 收尾） + pw = sync_playwright().start() + browser = pw.chromium.launch(headless=False) + context = browser.new_context(viewport={"width": 1920, "height": 1080}) + + pages_map = {} + print("\n====================================================") + print("【启动】正在打开各站点页面...") + print("====================================================") + for site_name, url in active_sites.items(): + if site_name == "顺心": + # 顺心：两个归属地账号在同一窗口各开一个标签页 + sx_pages = [] + for acct in range(1, 3): + print(f">> 正在启动【顺心】账号{acct}标签页: {url}") + sx_page = context.new_page() + sx_page.goto(url) + sx_pages.append(sx_page) + pages_map["顺心"] = sx_pages + else: + print(f">> 正在启动【{site_name}】页面: {url}") + page = context.new_page() + page.goto(url) + pages_map[site_name] = page + + # 4. 安能 Electron + anneng_proc = None + if anneng_active: + try: + anneng_proc = launch_anneng(anneng_app_path) + pages_map["安能"] = True # 哨兵：已启动（无 Playwright page） + except Exception as e: + print(f"⚠️ 启动安能应用失败，已跳过安能：{e}") + anneng_active = False + + # 5. 韵达前置自动登录 + print("\n====================================================") + print("【登录检测】正在准备各站点登录...") + print("====================================================") + if "韵达" in pages_map: + try: + pages_map["韵达"].bring_to_front() + site_yunda.yunda_login(pages_map["韵达"]) + except Exception as e: + print(f" ⚠️ 韵达前置自动登录模块发生波动: {e}") + + # 6. 就绪轮询（复用 probe_site_login） + ready_status = {site: False for site in active_sites.keys()} + if anneng_active: + ready_status["安能"] = False + print("\n>> 正在轮询各站点就绪状态 (自动登录或手动登录均可)...") + while not all(ready_status.values()): + for site_name in list(ready_status.keys()): + if ready_status[site_name]: + continue + if probe_site_login(site_name, pages_map): + ready_status[site_name] = True + print(f" ✅ 【{site_name}】已检测到主页，登录就绪。") + pending = [s for s, r in ready_status.items() if not r] + if pending: + print( + f" ⏳ 等待以下站点完成登录: [{', '.join(pending)}] ... " + "(请在浏览器/应用中操作)" + ) + time.sleep(3) + + # 7. 初始弹窗清理 + print("\n====================================================") + print("【准备】所有站点已就绪，正在清理初始弹窗...") + print("====================================================") + _dismiss_initial_popups(pages_map) + for s in ("中通", "韵达", "安能"): + if s in pages_map: + print(f" ✅ 【{s}】已就绪。") + + # 8. 状态库 + 心跳初值（就绪轮询刚通过 → 各站视为已登录） + state_store.init_db() + sites_to_watch = list(ready_status.keys()) + for _site in sites_to_watch: + state_store.set_login_state(_site, True) + + return RuntimeContext( + pw, + browser, + pages_map, + ready_status, + anneng_proc, + sites_to_watch, + debug_mode, + debug_target, + ) + + +def _dismiss_initial_popups(pages_map): + """顺心 / 百世 初始弹窗清理。""" + if "顺心" in pages_map: + for acct, sx_page in enumerate(pages_map["顺心"], start=1): + try: + sx_page.bring_to_front() + print(f">> 正在处理【顺心】账号{acct}弹窗与遮罩...") + sx_page.locator("a").nth(4).click(timeout=2000) + sx_page.wait_for_timeout(500) + sx_page.get_by_role("button", name="Close").click(timeout=2000) + sx_page.wait_for_timeout(500) + sx_page.get_by_role("button", name="不再询问").click(timeout=2000) + print(f" ✅ 【顺心】账号{acct}初始弹窗处理完成。") + except Exception: + pass + if "百世" in pages_map: + try: + bs_page = pages_map["百世"] + bs_page.bring_to_front() + print(">> 正在处理【百世】阅读完毕与关闭按钮...") + for _ in range(4): + handled = False + try: + read_btns = bs_page.locator("button:has-text('阅读完毕')") + if read_btns.count() > 0: + for i in range(read_btns.count()): + if read_btns.nth(i).is_visible(timeout=500): + read_btns.nth(i).click() + handled = True + except Exception: + pass + try: + if bs_page.locator("button:has-text('关 闭')").is_visible( + timeout=500 + ): + bs_page.locator("button:has-text('关 闭')").click() + handled = True + except Exception: + pass + if not handled: + break + bs_page.wait_for_timeout(800) + print(" ✅ 【百世】初始弹窗处理完成。") + except Exception: + pass + + +# ============================ 任务派发 ============================ + + +def _web_handler(site, download_func): + """构造网页站任务 handler：单 page 先 bring_to_front 再 download；顺心(list) 直接传。""" + + def handler(ctx): + pg = ctx.pages_map[site] + if not isinstance(pg, list): + pg.bring_to_front() + return download_func(pg) + + return handler + + +TASK_HANDLERS = { + ("顺心", "expected"): _web_handler("顺心", site_shunxin.shunxin_expected_download), + ("顺心", "actual"): _web_handler("顺心", site_shunxin.shunxin_actual_download), + ("百世", "undelivered"): _web_handler( + "百世", site_baishi.baishi_download_undelivered_data + ), + ("中通", "expected"): _web_handler("中通", site_zto.zto_expected_download), + ("中通", "actual"): _web_handler("中通", site_zto.zto_actual_download), + ("韵达", "expected"): _web_handler("韵达", site_yunda.yunda_expected_download), + ("韵达", "actual"): _web_handler("韵达", site_yunda.yunda_actual_download), + ("安能", "expected"): lambda ctx: site_anneng.anneng_expected_download(), + ("安能", "actual"): lambda ctx: site_anneng.anneng_actual_download(), + ("__compare__", "compare"): lambda ctx: (expected_undelivered.main() or True), +} + + +def dispatch_task(ctx, task_spec): + """执行一条任务。task_spec = {"site", "kind"}。返回 (status, error)。 + + 掉登录的站点直接判 failed（呼应"掉登录则该站任务全停"）。 + 只在 Playwright 所属线程调用。 + """ + site = task_spec.get("site") + kind = task_spec.get("kind") + if site != "__compare__": + if site not in ctx.pages_map: + return (state_store.TASK_FAILED, f"站点 {site} 未加载") + if not probe_site_login(site, ctx.pages_map): + return ( + state_store.TASK_FAILED, + f"站点 {site} 未登录，已跳过（需人工重登）", + ) + handler = TASK_HANDLERS.get((site, kind)) + if handler is None: + return (state_store.TASK_FAILED, f"未知任务: {site}/{kind}") + try: + ret = handler(ctx) + if ret is False: + return (state_store.TASK_FAILED, "任务执行失败（重试耗尽）") + return (state_store.TASK_SUCCESS, None) + except Exception as e: + return (state_store.TASK_FAILED, str(e)) + + +# ============================ 心跳 ============================ + + +def run_heartbeat(ctx): + """一轮心跳：探测各站登录态 + 数据文件，写状态库；登录态变化时提示。 + + 只在 Playwright 所属线程调用。 + """ + prev = state_store.get_all_status() + for site_name in ctx.sites_to_watch: + logged_in = probe_site_login(site_name, ctx.pages_map) + prev_login = prev.get(site_name, {}).get("login_state") + state_store.set_login_state(site_name, logged_in) + now_login = state_store.LOGIN_IN if logged_in else state_store.LOGIN_OUT + if prev_login and prev_login not in (now_login, state_store.LOGIN_UNKNOWN): + print(f"\n ⚠️【{site_name}】登录态变化: {prev_login} → {now_login}") + for kind in ("expected", "actual"): + ready, gen_at = probe_data_file(site_name, kind) + state_store.set_data_state(site_name, kind, ready, gen_at) diff --git a/server.py b/server.py new file mode 100644 index 0000000..4bcc8f9 --- /dev/null +++ b/server.py @@ -0,0 +1,152 @@ +# server.py +# +# 服务模式入口：FastAPI（主线程，uvicorn/asyncio）+ Playwright worker（独立线程）。 +# 客户端经 HTTP 触发任务、查状态、下载数据；worker 串行执行任务并跑心跳。 +# +# 线程模型（关键）： +# - 主线程：uvicorn + FastAPI。路由【绝不】访问 Playwright 对象，只经 +# task_queue（投递任务）+ state_store（查状态/任务）+ 文件系统（下载数据）。 +# - worker 线程：runtime.launch_and_prepare（sync_playwright 在此）+ 任务循环， +# 独占所有 page 操作；与主线程仅经 Queue + SQLite 通信。 +# 违反"路由不碰 Playwright"会崩（sync 对象跨线程访问）。 +# +# 运行：python server.py （默认监听 0.0.0.0:8000） + +import os +import queue +import threading +import time +from contextlib import asynccontextmanager + +import uvicorn +from fastapi import FastAPI, HTTPException +from fastapi.responses import FileResponse +from pydantic import BaseModel + +from paths import DOWNLOAD_DIR +import state_store +from runtime import ( + HEARTBEAT_INTERVAL, + TASK_HANDLERS, + dispatch_task, + launch_and_prepare, + run_heartbeat, +) + +# 任务队列：元素 (task_id, task_spec)。主线程投递，worker 消费。 +task_queue: "queue.Queue" = queue.Queue() +# worker 运行状态（主线程只读，worker 写） +worker_state = { + "ctx": None, + "stop": False, + "thread": None, + "ready": False, # launch_and_prepare 完成（各站就绪，可接任务） + "error": None, # worker 启动失败原因 +} + + +def _worker_loop(): + """worker 线程：启动 Playwright + 等就绪 + 任务循环（执行任务 + 心跳）。""" + try: + ctx = launch_and_prepare() + worker_state["ctx"] = ctx + worker_state["ready"] = True + print(">> [worker] 各站就绪，开始接收任务 ...") + except Exception as e: + worker_state["error"] = str(e) + print(f"❌ [worker] 启动失败: {e}") + return + + last_heartbeat = 0.0 + while not worker_state["stop"]: + try: + task_id, task_spec = task_queue.get(timeout=1) + except queue.Empty: + # 空闲时跑心跳 + if time.monotonic() - last_heartbeat >= HEARTBEAT_INTERVAL: + run_heartbeat(ctx) + last_heartbeat = time.monotonic() + continue + + state_store.update_task(task_id, state_store.TASK_RUNNING) + print(f">> [worker] 执行任务 #{task_id}: {task_spec}") + status, error = dispatch_task(ctx, task_spec) + state_store.update_task(task_id, status, error) + print(f">> [worker] 任务 #{task_id} 完成: {status} {error or ''}") + + try: + ctx.stop() + except Exception: + pass + worker_state["ready"] = False + print(">> [worker] 已退出。") + + +@asynccontextmanager +async def lifespan(_app): + """服务启停：起 worker 线程 / 通知 worker 停。""" + t = threading.Thread(target=_worker_loop, daemon=True) + worker_state["thread"] = t + t.start() + yield + worker_state["stop"] = True + t.join(timeout=10) + + +app = FastAPI(title="InboundVerify 服务端", lifespan=lifespan) + + +class TaskRequest(BaseModel): + site: str + kind: str + + +@app.post("/tasks") +def create_task(req: TaskRequest): + """提交任务 {site, kind} → 入队，返回 task_id。""" + if (req.site, req.kind) not in TASK_HANDLERS: + raise HTTPException(status_code=400, detail=f"无效任务: {req.site}/{req.kind}") + task_id = state_store.create_task(req.site, req.kind) + task_queue.put((task_id, {"site": req.site, "kind": req.kind})) + return {"task_id": task_id} + + +@app.get("/tasks/{task_id}") +def get_task(task_id: int): + t = state_store.get_task(task_id) + if not t: + raise HTTPException(status_code=404, detail="任务不存在") + return t + + +@app.get("/tasks") +def list_tasks(limit: int = 20): + return state_store.list_tasks(limit) + + +@app.get("/status") +def get_status(): + """各站登录态 + 数据态（前端状态盘用），另含 worker 就绪状态。""" + return { + "worker_ready": worker_state["ready"], + "worker_error": worker_state["error"], + "sites": state_store.get_all_status(), + } + + +@app.get("/data/{filename}") +def download_data(filename: str): + """下载 downloads/ 下的数据文件（防路径穿越）。""" + if not filename or "/" in filename or "\\" in filename or ".." in filename: + raise HTTPException(status_code=400, detail="非法文件名") + path = os.path.join(DOWNLOAD_DIR, filename) + # 双重校验：解析后绝对路径仍在 DOWNLOAD_DIR 内 + if not os.path.abspath(path).startswith(os.path.abspath(DOWNLOAD_DIR) + os.sep): + raise HTTPException(status_code=400, detail="非法路径") + if not os.path.isfile(path): + raise HTTPException(status_code=404, detail="文件不存在") + return FileResponse(path, filename=filename) + + +if __name__ == "__main__": + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/site_anneng.py b/site_anneng.py index 1eabfd4..52835d5 100644 --- a/site_anneng.py +++ b/site_anneng.py @@ -47,6 +47,7 @@ def with_retry(site_name, label, flow, reset, max_attempts=3): 每次失败都重置（含最终放弃那一次）：既是重试前的清场，也保证最终放弃时环境干净。 flow 为零参可调用；返回 False 视为失败，其余视为成功。 + 返回 True=最终成功，False=重试耗尽放弃（供调度层判断任务成败）。 """ for attempt in range(1, max_attempts + 1): try: @@ -55,7 +56,7 @@ def with_retry(site_name, label, flow, reset, max_attempts=3): raise RuntimeError("流程返回失败状态") if attempt > 1: print(f">> 【{site_name}-{label}】第 {attempt} 次尝试成功 ✅") - return + return True except Exception as e: print(f"⚠️ 【{site_name}-{label}】第 {attempt}/{max_attempts} 次失败: {e}") print(f" → 重置【{site_name}】到初始态，清理环境 ...") @@ -68,6 +69,7 @@ def with_retry(site_name, label, flow, reset, max_attempts=3): print( f"❌ 【{site_name}-{label}】已达最大尝试次数 {max_attempts}，放弃（环境已清理）。" ) + return False CDP_PORT = 9222 # 默认端口（独立运行 site_anneng.py 时用）；main_router 启动时会用 set_cdp_port 覆盖 diff --git a/site_baishi.py b/site_baishi.py index b444dfc..969cd8b 100644 --- a/site_baishi.py +++ b/site_baishi.py @@ -12,6 +12,7 @@ def with_retry(site_name, label, flow, reset, max_attempts=3): 每次失败都重置（含最终放弃那一次）：既是重试前的清场，也保证最终放弃时环境干净。 flow 为零参可调用；返回 False 视为失败，其余视为成功。 + 返回 True=最终成功，False=重试耗尽放弃（供调度层判断任务成败）。 """ for attempt in range(1, max_attempts + 1): try: @@ -20,7 +21,7 @@ def with_retry(site_name, label, flow, reset, max_attempts=3): raise RuntimeError("流程返回失败状态") if attempt > 1: print(f">> 【{site_name}-{label}】第 {attempt} 次尝试成功 ✅") - return + return True except Exception as e: print(f"⚠️ 【{site_name}-{label}】第 {attempt}/{max_attempts} 次失败: {e}") print(f" → 重置【{site_name}】到初始态，清理环境 ...") @@ -33,6 +34,7 @@ def with_retry(site_name, label, flow, reset, max_attempts=3): print( f"❌ 【{site_name}-{label}】已达最大尝试次数 {max_attempts}，放弃（环境已清理）。" ) + return False # 站点首页 URL：异常兜底重置用，也供 main_router 的 SITES_CONFIG 引用（单一来源） diff --git a/site_shunxin.py b/site_shunxin.py index 5def303..e5f0104 100644 --- a/site_shunxin.py +++ b/site_shunxin.py @@ -15,6 +15,7 @@ def with_retry(site_name, label, flow, reset, max_attempts=3): 每次失败都重置（含最终放弃那一次）：既是重试前的清场，也保证最终放弃时环境干净。 flow 为零参可调用；返回 False 视为失败，其余视为成功。 + 返回 True=最终成功，False=重试耗尽放弃（供调度层判断任务成败）。 """ for attempt in range(1, max_attempts + 1): try: @@ -23,7 +24,7 @@ def with_retry(site_name, label, flow, reset, max_attempts=3): raise RuntimeError("流程返回失败状态") if attempt > 1: print(f">> 【{site_name}-{label}】第 {attempt} 次尝试成功 ✅") - return + return True except Exception as e: print(f"⚠️ 【{site_name}-{label}】第 {attempt}/{max_attempts} 次失败: {e}") print(f" → 重置【{site_name}】到初始态，清理环境 ...") @@ -36,6 +37,7 @@ def with_retry(site_name, label, flow, reset, max_attempts=3): print( f"❌ 【{site_name}-{label}】已达最大尝试次数 {max_attempts}，放弃（环境已清理）。" ) + return False # 站点首页 URL：异常兜底重置用，也供 main_router 的 SITES_CONFIG 引用（单一来源） diff --git a/site_yunda.py b/site_yunda.py index 4e8f7ed..dfaf9f6 100644 --- a/site_yunda.py +++ b/site_yunda.py @@ -15,6 +15,7 @@ def with_retry(site_name, label, flow, reset, max_attempts=3): 每次失败都重置（含最终放弃那一次）：既是重试前的清场，也保证最终放弃时环境干净。 flow 为零参可调用；返回 False 视为失败，其余视为成功。 + 返回 True=最终成功，False=重试耗尽放弃（供调度层判断任务成败）。 """ for attempt in range(1, max_attempts + 1): try: @@ -23,7 +24,7 @@ def with_retry(site_name, label, flow, reset, max_attempts=3): raise RuntimeError("流程返回失败状态") if attempt > 1: print(f">> 【{site_name}-{label}】第 {attempt} 次尝试成功 ✅") - return + return True except Exception as e: print(f"⚠️ 【{site_name}-{label}】第 {attempt}/{max_attempts} 次失败: {e}") print(f" → 重置【{site_name}】到初始态，清理环境 ...") @@ -36,6 +37,7 @@ def with_retry(site_name, label, flow, reset, max_attempts=3): print( f"❌ 【{site_name}-{label}】已达最大尝试次数 {max_attempts}，放弃（环境已清理）。" ) + return False # 站点首页 URL：异常兜底重置用，也供 main_router 的 SITES_CONFIG 引用（单一来源） diff --git a/site_zto.py b/site_zto.py index d823a45..f050f47 100644 --- a/site_zto.py +++ b/site_zto.py @@ -15,6 +15,7 @@ def with_retry(site_name, label, flow, reset, max_attempts=3): 每次失败都重置（含最终放弃那一次）：既是重试前的清场，也保证最终放弃时环境干净。 flow 为零参可调用；返回 False 视为失败，其余视为成功。 + 返回 True=最终成功，False=重试耗尽放弃（供调度层判断任务成败）。 """ for attempt in range(1, max_attempts + 1): try: @@ -23,7 +24,7 @@ def with_retry(site_name, label, flow, reset, max_attempts=3): raise RuntimeError("流程返回失败状态") if attempt > 1: print(f">> 【{site_name}-{label}】第 {attempt} 次尝试成功 ✅") - return + return True except Exception as e: print(f"⚠️ 【{site_name}-{label}】第 {attempt}/{max_attempts} 次失败: {e}") print(f" → 重置【{site_name}】到初始态，清理环境 ...") @@ -36,6 +37,7 @@ def with_retry(site_name, label, flow, reset, max_attempts=3): print( f"❌ 【{site_name}-{label}】已达最大尝试次数 {max_attempts}，放弃（环境已清理）。" ) + return False # 站点首页 URL：异常兜底重置用，也供 main_router 的 SITES_CONFIG 引用（单一来源） diff --git a/state_store.py b/state_store.py index 5148715..8d00cdb 100644 --- a/state_store.py +++ b/state_store.py @@ -16,6 +16,13 @@ LOGIN_UNKNOWN = "unknown" # 尚未探测过 LOGIN_IN = "logged_in" LOGIN_OUT = "logged_out" +# 任务状态枚举（task_history.status） +TASK_PENDING = "pending" # 已入队，待执行 +TASK_RUNNING = "running" # 正在执行 +TASK_SUCCESS = "success" # 成功（有数据） +TASK_NO_DATA = "no_data" # 成功但本站本次无数据 +TASK_FAILED = "failed" # 失败（重试耗尽 / 未登录 / 异常） + def _now(): """本地时间的字符串（到秒），用于时间戳列。""" @@ -38,6 +45,17 @@ def init_db(): updated_at TEXT ) """) + conn.execute(""" + CREATE TABLE IF NOT EXISTS task_history ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + site TEXT, + kind TEXT, + status TEXT, + started_at TEXT, + finished_at TEXT, + error TEXT + ) + """) conn.commit() @@ -134,3 +152,79 @@ def get_all_status(): } for r in rows } + + +# ============================ 任务历史 ============================ + + +def create_task(site, kind): + """新建一条 pending 任务，返回其 id。""" + now = _now() + with sqlite3.connect(STATE_DB_PATH) as conn: + cur = conn.execute( + "INSERT INTO task_history (site, kind, status, started_at, finished_at, error) " + "VALUES (?, ?, ?, ?, '', '')", + (site, kind, TASK_PENDING, now), + ) + conn.commit() + return cur.lastrowid + + +def update_task(task_id, status, error=None): + """更新任务状态。终态(success/no_data/failed)写入 finished_at。""" + finished = _now() if status in (TASK_SUCCESS, TASK_NO_DATA, TASK_FAILED) else "" + with sqlite3.connect(STATE_DB_PATH) as conn: + if finished: + conn.execute( + "UPDATE task_history SET status=?, finished_at=?, error=? WHERE id=?", + (status, finished, error or "", task_id), + ) + else: + conn.execute( + "UPDATE task_history SET status=?, error=? WHERE id=?", + (status, error or "", task_id), + ) + conn.commit() + + +def get_task(task_id): + """返回单条任务 dict，不存在返回 None。""" + with sqlite3.connect(STATE_DB_PATH) as conn: + row = conn.execute( + "SELECT id, site, kind, status, started_at, finished_at, error " + "FROM task_history WHERE id=?", + (task_id,), + ).fetchone() + if not row: + return None + return { + "id": row[0], + "site": row[1], + "kind": row[2], + "status": row[3], + "started_at": row[4], + "finished_at": row[5], + "error": row[6], + } + + +def list_tasks(limit=20): + """返回最近 limit 条任务（按 id 倒序）。""" + with sqlite3.connect(STATE_DB_PATH) as conn: + rows = conn.execute( + "SELECT id, site, kind, status, started_at, finished_at, error " + "FROM task_history ORDER BY id DESC LIMIT ?", + (limit,), + ).fetchall() + return [ + { + "id": r[0], + "site": r[1], + "kind": r[2], + "status": r[3], + "started_at": r[4], + "finished_at": r[5], + "error": r[6], + } + for r in rows + ]