Files
InboundVerify/main_router.py
Misaka a5ceb42ac1 阶段0:登录态心跳 + 状态持久化(重构地基)
- 新增 state_store.py:SQLite 状态存储(site_status 表:登录态/数据态/时间戳),
  UPSERT 保留字段,重启不丢
- paths.py:加 STATE_DB_PATH(state/state.db)
- main_router:提取 probe_site_login / probe_data_file;菜单循环改为
  input 后台线程 + 主线程心跳轮询(满足 Playwright sync 线程安全);
  每 30s 探测各站登录态 + 数据文件写库,登录态变化时提示;新增菜单 [12] 站点状态盘
- .gitignore:忽略 state/

为后续 FastAPI 化与 Web 前端提供状态地基。

Co-Authored-By: Claude <noreply@anthropic.com>
2026-07-16 22:44:15 +08:00

680 lines
29 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# main_router.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
# 阶段0状态持久化站点登录态 / 数据态,心跳刷新;重启不丢)
import state_store
# 导入抽离出去的各个网点模块
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/ 下的应到/实到
数据,生成 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 run_multi_site_daemon():
"""多站点自动化主控流程"""
# 1. 读取配置文件
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}")
# 动态确定需要挂载启动的网页站点安能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
with sync_playwright() as p:
browser = p.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 == "顺心":
# 顺心:两个归属地账号在同一窗口各开一个标签页(顺心站点支持
# 同浏览器双账号并存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(" ✅ 【安能】已就绪。")
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()
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(" 全局离线数据处理")
print(
" [9] 执行 - 应到未到比对(全站点汇总,输出 output/应到未到数据.xlsx"
)
print("-" * 52)
print(" 站点状态")
print(" [12] 查看 - 各站登录态 / 数据就绪状态")
print("-" * 52)
print(" [0] 退出系统")
print("====================================================")
print("请输入任务编号并回车: ", end="", flush=True)
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()
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}")
browser.close()
if anneng_proc is not None:
try:
anneng_proc.terminate()
print("已关闭安能应用。")
except Exception:
pass
print("程序已退出。")
if __name__ == "__main__":
run_multi_site_daemon()