# 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 # 2.5 重置各站登录态为 unknown:登录态是会话级的,避免显示上一会话残留的陈旧登录态。 # 数据态(文件就绪)会话无关、保留不动;心跳就绪后会重新探测写真实值。 reset_sites = list(active_sites.keys()) if anneng_active: reset_sites.append("安能") state_store.reset_login_states(reset_sites) # 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)