66bd8af421
- 新增 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>
348 lines
14 KiB
Python
348 lines
14 KiB
Python
# main_router.py
|
||
#
|
||
# 交互菜单模式入口（调试 / 人工操作）。核心 Playwright 管理、任务派发、心跳
|
||
# 已抽到 runtime.py 共享；本文件只保留交互菜单与自动化测试。
|
||
# 服务模式（常驻 + FastAPI 接收指令）见 server.py。
|
||
|
||
import os
|
||
import queue
|
||
import threading
|
||
import time
|
||
|
||
import yaml
|
||
|
||
from paths import CONFIG_PATH
|
||
from runtime import (
|
||
APP_SITES,
|
||
HEARTBEAT_INTERVAL,
|
||
dispatch_task,
|
||
launch_and_prepare,
|
||
run_heartbeat,
|
||
)
|
||
|
||
import state_store
|
||
|
||
# 各站点模块（自动化测试 + 比对用；任务派发在 runtime）
|
||
import site_shunxin
|
||
import site_baishi
|
||
import site_zto
|
||
import site_yunda
|
||
import site_anneng
|
||
import expected_undelivered
|
||
|
||
|
||
def run_undelivered_compare():
|
||
"""应到未到比对（全站点）：调用 expected_undelivered，读 downloads/ 下的应到/实到
|
||
数据，生成 output/应到未到数据.xlsx（汇总报表 + 各站明细）。"""
|
||
print("\n▶ 开始执行【应到未到比对（全站点）】任务 ...")
|
||
expected_undelivered.main()
|
||
|
||
|
||
# ====================================================================
|
||
# 自动化测试入口
|
||
# ====================================================================
|
||
# 每个(非百世)站点的交叉测试序列：覆盖两种下载流程之间的全部 4 种相邻转换，
|
||
# 用于验证无论上一个流程把页面留在什么状态，下一个流程都能正常运行：
|
||
# 应到->实到、实到->应到、应到->应到、实到->实到
|
||
CROSS_TEST_SEQUENCE = ["expected", "actual", "expected", "expected", "actual", "actual"]
|
||
|
||
|
||
def run_automation_test(pages_map):
|
||
"""自动化测试入口：按交叉序列逐个跑通各站点的下载流程，结束后打印统计报告。
|
||
|
||
判定规则：流程函数返回 False 或抛出异常记为 FAIL，其余记为 PASS。
|
||
"""
|
||
# 站点 -> {流程键: (中文名, 流程函数)}；百世为单流程，单独处理
|
||
# 自动化测试刻意走各站点的 _impl（单次执行、无兜底重试），以便探测原始失败、
|
||
# 不被模块内部"失败→重置→重试"机制掩盖。
|
||
flow_table = {
|
||
"顺心": {
|
||
"expected": ("应到", site_shunxin.shunxin_expected_download_impl),
|
||
"actual": ("实到", site_shunxin.shunxin_actual_download_impl),
|
||
},
|
||
"中通": {
|
||
"expected": ("应到", site_zto.zto_expected_download_impl),
|
||
"actual": ("实到", site_zto.zto_actual_download_impl),
|
||
},
|
||
"韵达": {
|
||
"expected": ("应到", site_yunda.yunda_expected_download_impl),
|
||
"actual": ("实到", site_yunda.yunda_actual_download_impl),
|
||
},
|
||
"安能": {
|
||
"expected": ("应到", site_anneng.anneng_expected_download_impl),
|
||
"actual": ("实到", site_anneng.anneng_actual_download_impl),
|
||
},
|
||
}
|
||
|
||
# 构建测试计划：[(站点, 流程中文名, 流程函数, 绑定page, 归属标签), ...]
|
||
# 顺心为双账号：两个 page 各自读归属地后跑一遍交叉序列；其余站点单 page。
|
||
plan = []
|
||
for site_name, flows in flow_table.items():
|
||
if site_name == "顺心":
|
||
if "顺心" not in pages_map:
|
||
continue
|
||
for sx_idx, sx_page in enumerate(pages_map["顺心"], start=1):
|
||
try:
|
||
tag = site_shunxin.shunxin_belonging(sx_page)
|
||
except Exception:
|
||
tag = f"账号{sx_idx}" # 读不到归属地时用序号占位，不阻断测试
|
||
for flow_key in CROSS_TEST_SEQUENCE:
|
||
label, func = flows[flow_key]
|
||
plan.append((f"顺心·{tag}", label, func, sx_page, tag))
|
||
continue
|
||
if site_name not in pages_map:
|
||
continue
|
||
bound_page = pages_map[site_name]
|
||
for flow_key in CROSS_TEST_SEQUENCE:
|
||
label, func = flows[flow_key]
|
||
plan.append((site_name, label, func, bound_page, ""))
|
||
# 百世：单流程，跑一次即可
|
||
if "百世" in pages_map:
|
||
plan.append(
|
||
(
|
||
"百世",
|
||
"应到未到",
|
||
site_baishi.baishi_download_undelivered_data_impl,
|
||
pages_map["百世"],
|
||
"",
|
||
)
|
||
)
|
||
|
||
if not plan:
|
||
print("\n⚠️ 当前没有已就绪的站点，无法执行自动化测试。")
|
||
return
|
||
|
||
total = len(plan)
|
||
print("\n====================================================")
|
||
print(f"自动化测试开始，共 {total} 个步骤。")
|
||
print("（双流程站点按应到/实到交叉序列执行，覆盖全部相邻转换）")
|
||
print("====================================================")
|
||
|
||
results = [] # [(站点, 流程, 状态, 耗时秒, 错误信息)]
|
||
for idx, (site_name, label, func, bound_page, out_tag) in enumerate(plan, start=1):
|
||
print("\n----------------------------------------------------")
|
||
print(f"[步骤 {idx}/{total}] 站点【{site_name}】流程【{label}】")
|
||
print("----------------------------------------------------")
|
||
start = time.time()
|
||
status = "PASS"
|
||
err = ""
|
||
try:
|
||
if site_name in APP_SITES:
|
||
# 安能：Electron 应用，无 Playwright page，函数不收 page 参数
|
||
ret = func()
|
||
else:
|
||
bound_page.bring_to_front()
|
||
# 顺心 _impl 带 out_tag（归属地）；其余站点 _impl 仅收 page
|
||
ret = func(bound_page, out_tag=out_tag) if out_tag else func(bound_page)
|
||
if ret is False:
|
||
status = "FAIL"
|
||
err = "流程返回失败状态"
|
||
except Exception as e:
|
||
status = "FAIL"
|
||
err = str(e)
|
||
elapsed = time.time() - start
|
||
results.append((site_name, label, status, elapsed, err))
|
||
print(f">> 步骤结果: {status} (耗时 {elapsed:.1f}s)")
|
||
|
||
_print_test_report(results)
|
||
|
||
|
||
def _print_test_report(results):
|
||
"""打印自动化测试统计报告。"""
|
||
passed = sum(1 for r in results if r[2] == "PASS")
|
||
failed = len(results) - passed
|
||
|
||
print("\n====================================================")
|
||
print("自动化测试统计报告")
|
||
print("====================================================")
|
||
for i, (site_name, label, status, elapsed, err) in enumerate(results, start=1):
|
||
mark = "✅" if status == "PASS" else "❌"
|
||
print(
|
||
f" {i:>2}. {mark} {status} {site_name} - {label} (耗时 {elapsed:.1f}s)"
|
||
)
|
||
if err:
|
||
note = err if len(err) <= 60 else err[:57] + "..."
|
||
print(f" 说明: {note}")
|
||
print("----------------------------------------------------")
|
||
print(f" 合计 {len(results)} 步：通过 {passed}，失败 {failed}")
|
||
if failed == 0:
|
||
print(" ✅ 全部流程跑通。")
|
||
else:
|
||
print(" ❌ 存在失败流程，请结合上方说明与运行日志排查。")
|
||
print("====================================================")
|
||
|
||
|
||
# ====================================================================
|
||
# 交互菜单模式
|
||
# ====================================================================
|
||
|
||
|
||
def _read_debug_config():
|
||
"""读 config.yaml 的 debug 段，返回 (debug_mode, debug_target)。"""
|
||
debug_mode = False
|
||
debug_target = ""
|
||
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", "")
|
||
except Exception as e:
|
||
print(f"⚠️ 读取 config.yaml 异常，将使用全量模式启动: {e}")
|
||
return debug_mode, debug_target
|
||
|
||
|
||
# 菜单编号 → 任务规格（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"},
|
||
}
|
||
|
||
|
||
def _interactive_menu_loop(ctx):
|
||
"""交互菜单循环：input 后台线程 + _await_command + dispatch_task + 心跳 + 状态盘。
|
||
|
||
所有 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:
|
||
print(f"\n🚫 站点 [{site_name}] 未加载（当前为调试模式），已跳过。")
|
||
return False
|
||
return True
|
||
|
||
last_heartbeat = 0.0
|
||
command_queue = queue.Queue()
|
||
|
||
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(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(
|
||
f" 【{site_name}】 {login_mark} | {exp} | {act} "
|
||
f"| 探测于 {s['login_checked_at']}"
|
||
)
|
||
print("======================================================")
|
||
|
||
def _input_loop():
|
||
while True:
|
||
try:
|
||
command_queue.put(input())
|
||
except EOFError:
|
||
return
|
||
|
||
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("程序已退出。")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
run_multi_site_daemon()
|