阶段1:服务化骨架(FastAPI + Playwright worker + 任务队列)
- 新增 runtime.py:抽离共享核心 launch_and_prepare / dispatch_task / run_heartbeat /
probe_* / launch_anneng / RuntimeContext,常量 SITES_CONFIG 等;main_router 复用
- 新增 server.py:FastAPI(主线程)+ Playwright worker(独立线程)+ 任务队列;
API:POST/GET /tasks、GET /status、GET /data/{file}(防路径穿越)
- state_store:加 task_history 表 + create/update/get/list 接口
- 5 站点 with_retry 改为返回 True/False,供 dispatch_task 判成败
- main_router:重写为复用 runtime 的交互模式(行为不变)
- requirements:加 fastapi、uvicorn
- CLAUDE.md:补充运行模式与共享核心架构说明
线程模型:主线程 FastAPI 不碰 Playwright,worker 线程独占 page,经 Queue + SQLite 通信。
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
16
CLAUDE.md
16
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)`。**例外:顺心是双账号**
|
||||
|
||||
476
main_router.py
476
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,196 +172,48 @@ 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("====================================================")
|
||||
|
||||
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
|
||||
|
||||
# 安能:以调试模式启动 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(" ✅ 【安能】已就绪。")
|
||||
所有 page 操作经 runtime(主线程),满足 Playwright sync 线程安全。
|
||||
"""
|
||||
pages_map = ctx.pages_map
|
||||
sites_to_watch = ctx.sites_to_watch
|
||||
|
||||
def is_site_ready(site_name):
|
||||
if site_name not in pages_map:
|
||||
@@ -476,43 +221,20 @@ def run_multi_site_daemon():
|
||||
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
|
||||
command_queue = queue.Queue()
|
||||
|
||||
def _run_heartbeat():
|
||||
"""一轮心跳:探测各站登录态 + 数据文件,写状态库;登录态变化时提示。"""
|
||||
def _await_command():
|
||||
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)
|
||||
while True:
|
||||
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():
|
||||
"""打印各站登录态 / 数据就绪状态盘(菜单 [12])。"""
|
||||
print("\n====================== 站点状态盘 ======================")
|
||||
status = state_store.get_all_status()
|
||||
if not status:
|
||||
@@ -529,24 +251,15 @@ def run_multi_site_daemon():
|
||||
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 '无'}"
|
||||
)
|
||||
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(
|
||||
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())
|
||||
@@ -555,21 +268,11 @@ def run_multi_site_daemon():
|
||||
|
||||
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()
|
||||
|
||||
while True:
|
||||
print("\n====================================================")
|
||||
print(" 物流数据下载主菜单 ")
|
||||
if debug_mode:
|
||||
print(f" [ 调试模式,仅加载: {debug_target} ]")
|
||||
if ctx.debug_mode:
|
||||
print(f" [ 调试模式,仅加载: {ctx.debug_target} ]")
|
||||
print("====================================================")
|
||||
print(" 模块一:【顺心】数据处理流")
|
||||
print(" [1] 执行 - 应到货物数据下载")
|
||||
@@ -594,9 +297,7 @@ def run_multi_site_daemon():
|
||||
print(" [8] 执行 - 全站点下载流程自动化测试 (交叉跑通校验)")
|
||||
print("-" * 52)
|
||||
print(" 全局离线数据处理")
|
||||
print(
|
||||
" [9] 执行 - 应到未到比对(全站点汇总,输出 output/应到未到数据.xlsx)"
|
||||
)
|
||||
print(" [9] 执行 - 应到未到比对(全站点汇总,输出 output/应到未到数据.xlsx)")
|
||||
print("-" * 52)
|
||||
print(" 站点状态")
|
||||
print(" [12] 查看 - 各站登录态 / 数据就绪状态")
|
||||
@@ -608,70 +309,37 @@ def run_multi_site_daemon():
|
||||
choice = _await_command()
|
||||
|
||||
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()
|
||||
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 == "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",
|
||||
]:
|
||||
elif choice.strip() != "":
|
||||
print("\n⚠️ 无效输入,请查证后回车。")
|
||||
except Exception as e:
|
||||
print(f"❌ 任务调度异常: {e}")
|
||||
|
||||
browser.close()
|
||||
if anneng_proc is not None:
|
||||
|
||||
def run_multi_site_daemon():
|
||||
"""多站点自动化主控流程(交互菜单模式)。
|
||||
|
||||
启动 → 等待各站登录就绪 → 进入交互菜单;退出时关闭浏览器与安能。
|
||||
"""
|
||||
debug_mode, debug_target = _read_debug_config()
|
||||
ctx = launch_and_prepare(debug_mode, debug_target)
|
||||
try:
|
||||
anneng_proc.terminate()
|
||||
print("已关闭安能应用。")
|
||||
except Exception:
|
||||
pass
|
||||
_interactive_menu_loop(ctx)
|
||||
finally:
|
||||
print("\n正在关闭浏览器并退出...")
|
||||
ctx.stop()
|
||||
print("程序已退出。")
|
||||
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
447
runtime.py
Normal file
447
runtime.py
Normal file
@@ -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)
|
||||
152
server.py
Normal file
152
server.py
Normal file
@@ -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)
|
||||
@@ -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 覆盖
|
||||
|
||||
@@ -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 引用(单一来源)
|
||||
|
||||
@@ -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 引用(单一来源)
|
||||
|
||||
@@ -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 引用(单一来源)
|
||||
|
||||
@@ -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 引用(单一来源)
|
||||
|
||||
@@ -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
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user