# 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, timedelta 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", "undelivered": "顺心-未到数据.xlsx", }, "中通": { "expected": "中通-应到货物数据.xlsx", "actual": "中通-实到货物数据.xlsx", "undelivered": "中通-未到数据.xlsx", }, "韵达": { "expected": "韵达-应到货物数据.xlsx", "actual": "韵达-实到货物数据.xlsx", "undelivered": "韵达-未到数据.xlsx", }, "安能": { "expected": "安能-应到货物数据.xlsx", "actual": "安能-实到货物数据.xlsx", "undelivered": "安能-未到数据.xlsx", }, "百世": {"expected": "", "actual": "", "undelivered": "百世-应到未到货物数据.xlsx"}, } # ============================ 安能启动(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 应用(自动选取空闲端口),返回子进程对象。""" # 【环境兼容】WorkBuddy 等 shell 会注入 NODE_OPTIONS(含 --use-system-ca), # Electron 内置 Node 拒绝该 flag 导致安能启动即退出(rc=9)。 # 拉起前从子进程环境里摘掉 NODE_OPTIONS。 anneng_env = os.environ.copy() anneng_env.pop("NODE_OPTIONS", None) port = _find_free_port() print(f">> 以调试模式启动【安能】应用(端口 {port}):{app_path}") proc = subprocess.Popen([app_path, f"--remote-debugging-port={port}"], env=anneng_env) 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 seed_legacy_config(): """一次性:把 config.yaml 里的站点配置(百世密码 / 韵达账密 / 安能 exe 路径) 灌入 state.db。已存在的值不覆盖(前端改过的不动)。""" if not os.path.exists(CONFIG_PATH): return try: with open(CONFIG_PATH, "r", encoding="utf-8") as f: cfg = yaml.safe_load(f) or {} except Exception as e: print(f"⚠️ 读取 config.yaml 做 seed 失败: {e}") return items = [ ("百世", "password", (cfg.get("baishi", {}) or {}).get("password", "")), ("韵达", "username", (cfg.get("yunda", {}) or {}).get("username", "")), ("韵达", "password", (cfg.get("yunda", {}) or {}).get("password", "")), ("安能", "app_path", (cfg.get("anneng", {}) or {}).get("app_path", "")), ] seeded = [] for site, key, val in items: if val and not state_store.get_setting(site, key): state_store.set_setting(site, key, str(val)) seeded.append(f"{site}.{key}") if seeded: print(f">> [seed] 从 config.yaml 灌入站点配置: {', '.join(seeded)}") def launch_and_prepare(debug_mode=False, debug_target=""): """启动 Playwright + 各站 page + 就绪轮询 + 弹窗清理 + 心跳初值,返回 RuntimeContext。 必须在"持有 Playwright 的线程"调用(交互模式主线程 / 服务模式 worker 线程)。 阻塞至所有站点登录就绪才返回。 """ # 0. 状态库建表/迁移 + 从 config.yaml 灌入站点配置(须在 reset_login_states 等之前) state_store.init_db() seed_legacy_config() # 1. 读 debug 配置(config.yaml;服务模式也生效)+ 安能路径(state.db,seed 已灌入) if os.path.exists(CONFIG_PATH): try: with open(CONFIG_PATH, "r", encoding="utf-8") as f: _cfg = yaml.safe_load(f) or {} _dbg = _cfg.get("debug", {}) or {} if _dbg.get("enabled"): debug_mode = True debug_target = str(_dbg.get("target_site", "") or "") except Exception as e: print(f"⚠️ 读取 config.yaml(debug) 失败: {e}") anneng_app_path = state_store.get_setting("安能", "app_path") # 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() # 调试模式临时开启 CDP 端口,便于外部 Playwright(如 Playwright CLI 技能) # 通过 connectOverCDP 挂载到已登录的页面读取内容。仅调试模式生效,不影响服务模式。 _launch_args = ["--remote-debugging-port=9223"] if debug_mode else [] browser = pw.chromium.launch(headless=False, args=_launch_args) 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. 心跳初值(就绪轮询刚通过 → 各站视为已登录;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(">> 正在处理【百世】初始弹窗(阅读消息 / 配置检查 / 优惠券广告)...") # 委托给 site_baishi 的专用清理:优惠券广告是全屏居中 modal, # 关闭键为 .ant-modal-close(纯图标无文字),必须点它才能真正关掉。 site_baishi.dismiss_baishi_popups(bs_page) print(" ✅ 【百世】初始弹窗处理完成。") except Exception: pass if "韵达" in pages_map: try: yd_page = pages_map["韵达"] yd_page.bring_to_front() print(">> 正在检查【韵达】音频设备授权提示...") site_yunda.dismiss_audio_prompt(yd_page) 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 def _site_undelivered_handler(site): """4 站未到:下应到+实到 → 比对写 downloads/<站>-未到数据.xlsx。 任一下载失败 → 清掉旧未到文件、返回 False(前端不展示陈旧未到)。""" def handler(ctx): # 各站下载入口约定返回 True/False;顺心历史返回 None(视为成功,与 dispatch 一致) exp_ok = TASK_HANDLERS[(site, "expected")](ctx) is not False act_ok = ( (TASK_HANDLERS[(site, "actual")](ctx) is not False) if exp_ok else False ) if exp_ok and act_ok: return expected_undelivered.write_site_file(site) stale = os.path.join( DOWNLOAD_DIR, expected_undelivered.SITE_UNDELIVERED_FILE.format(name=site) ) if os.path.exists(stale): os.remove(stale) return False return handler # 「跑比对」= 纯离线比对(用 downloads/ 现有文件生成全站汇总;下载交由各站定时/手动)。 TASK_HANDLERS = { ("顺心", "expected"): _web_handler("顺心", site_shunxin.shunxin_expected_download), ("顺心", "actual"): _web_handler("顺心", site_shunxin.shunxin_actual_download), ("顺心", "undelivered"): _site_undelivered_handler("顺心"), ("百世", "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), ("中通", "undelivered"): _site_undelivered_handler("中通"), ("韵达", "expected"): _web_handler("韵达", site_yunda.yunda_expected_download), ("韵达", "actual"): _web_handler("韵达", site_yunda.yunda_actual_download), ("韵达", "undelivered"): _site_undelivered_handler("韵达"), ("安能", "expected"): lambda ctx: site_anneng.anneng_expected_download(), ("安能", "actual"): lambda ctx: site_anneng.anneng_actual_download(), ("安能", "undelivered"): _site_undelivered_handler("安能"), ("__compare__", "compare"): lambda ctx: (expected_undelivered.main() or True), } def _record_business_date(site, kind): """下载成功后,把本次数据的业务日期快照写进状态库(供前端/报告显示「是哪天的数据」)。 业务日期 = 下载当天 − 该数据对应的日期偏移。__compare__ 无数据概念,跳过。 kind → 写入: expected/actual:各写自己一列(偏移各取其列)。 undelivered:百世直供(恒 0)写 undelivered;4 站未到由 _site_undelivered_handler 内部连带下了 expected+actual(不经 dispatch,无业务日期写入),故此处一并补写 expected/actual/undelivered 三列——actual 用 actual 偏移、未到跟随 expected 偏移。 顺带置 ready=True,让前端不必等心跳即可反映下载成功;写入失败仅告警、不影响任务判定。""" if site == "__compare__": return today = datetime.now().date() now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") def _write(k, off): biz = (today - timedelta(days=off)).strftime("%Y-%m-%d") try: state_store.set_data_state( site, k, ready=True, generated_at=now, business_date=biz ) except Exception as e: print(f">> [状态] 写业务日期失败 {site}/{k}: {e}") if kind == "expected": _write("expected", state_store.get_offset(site, "expected")) elif kind == "actual": _write("actual", state_store.get_offset(site, "actual")) elif site == "百世": _write("undelivered", 0) else: # 4 站 undelivered:连带补写 expected/actual/undelivered 三列 _write("expected", state_store.get_offset(site, "expected")) _write("actual", state_store.get_offset(site, "actual")) _write("undelivered", state_store.get_offset(site, "expected")) 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, "任务执行失败(重试耗尽)") _record_business_date(site, kind) 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", "undelivered"): ready, gen_at = probe_data_file(site_name, kind) state_store.set_data_state(site_name, kind, ready, gen_at)