InboundVerify/inbound_verify/cli/router.py

# inbound_verify/cli/router.py
#
# 交互菜单模式入口（调试 / 人工操作）。核心 Playwright 管理、任务派发、心跳
# 已抽到 runtime.py 共享；本文件只保留交互菜单与自动化测试。
# 服务模式（常驻 + FastAPI 接收指令）见 cli/server.py。

import os
import queue
import threading
import time

import yaml

from inbound_verify.paths import CONFIG_PATH
from inbound_verify.runtime import (
    APP_SITES,
    HEARTBEAT_INTERVAL,
    dispatch_task,
    launch_and_prepare,
    run_heartbeat,
)

from inbound_verify import state_store

# 各站点模块（自动化测试 + 比对用；任务派发在 runtime）
from inbound_verify.sites import shunxin, baishi, zto, yunda, anneng
from inbound_verify 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": ("应到", shunxin.shunxin_expected_download_impl),
            "actual": ("实到", shunxin.shunxin_actual_download_impl),
        },
        "中通": {
            "expected": ("应到", zto.zto_expected_download_impl),
            "actual": ("实到", zto.zto_actual_download_impl),
        },
        "韵达": {
            "expected": ("应到", yunda.yunda_expected_download_impl),
            "actual": ("实到", yunda.yunda_actual_download_impl),
        },
        "安能": {
            "expected": ("应到", anneng.anneng_expected_download_impl),
            "actual": ("实到", 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 = 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(
            (
                "百世",
                "应到未到",
                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("程序已退出。")


def main():
    """交互菜单模式入口。"""
    run_multi_site_daemon()


if __name__ == "__main__":
    main()