# main_router.py import os import yaml import pandas as pd from playwright.sync_api import sync_playwright # 导入抽离出去的各个网点模块 import site_shunxin import site_baishi import site_zto import site_yunda # 定义网点及对应的初始登录 URL SITES_CONFIG = { "顺心": "https://sxne.sxjdfreight.com", "百世": "https://v5.800best.com", "中通": "https://ws.zto56.com/", "韵达": "https://ky-sso.yunda56.com", } # ==================================================================== # 🛡️ 网点特征注册表：定义每个网点登录成功、成功进入工作台的标志性控件 # ==================================================================== READY_SELECTORS = { "顺心": 'h1:has-text("盟商门户网")', "百世": 'h1[title="百世快运"]', "中通": '.logo:has-text("网点版")', "韵达": '.el-menu-item:has-text("首页")', } def task_process_undelivered_data(site_name="顺心"): """全局模块：应到未到异常件比对引擎 (支持动态网点前缀)""" print(f"\n▶ 开始执行【{site_name} - 应到未到数据处理】任务...") download_dir = os.path.join(os.getcwd(), "downloads") expected_path = os.path.join(download_dir, f"{site_name}-应到货物数据.xlsx") actual_path = os.path.join(download_dir, f"{site_name}-实到货物数据.xlsx") output_path = os.path.join(download_dir, f"{site_name}-应到未到货物数据.xlsx") if not os.path.exists(expected_path): print(f"❌ 错误：找不到【{site_name}-应到货物数据】主文档：{expected_path}") return if not os.path.exists(actual_path): print(f"❌ 错误：找不到【{site_name}-实到货物数据】主文档：{actual_path}") return try: print(">> 正在载入本地 Excel 文档...") df_expected = pd.read_excel(expected_path) df_actual = pd.read_excel(actual_path) if "运单号" not in df_expected.columns or "运单号" not in df_actual.columns: print("❌ 核心资产校验失败：数据源中缺失【运单号】字段，请检查导出配置。") return print(">> 正在启动多维数据集比对引擎...") # Left Anti-Join：在应到中找出不存在于实到里的运单号 df_undelivered = df_expected[~df_expected["运单号"].isin(df_actual["运单号"])] target_columns = ["班次号", "交接单号", "运单号"] available_columns = [ col for col in target_columns if col in df_undelivered.columns ] df_output = df_undelivered[available_columns] print(f">> 筛选完毕！共捕获到异常【应到未到】货物数据: {len(df_output)} 条。") df_output.to_excel(output_path, index=False) print(f"====================================================") print(f" 🎉 异常比对流完成！独立数据已安全输出。") print(f" 📁 成果归档路径: {output_path}") print(f"====================================================") except Exception as e: print(f"❌ 数据处理引擎在执行连接和输出时发生致命异常: {e}") def run_multi_site_daemon(): """多网点自动化主控引擎 (状态机卫语句驱动版)""" # 1. 读取配置文件 debug_mode = False debug_target = "" try: if os.path.exists("config.yaml"): with open("config.yaml", "r", encoding="utf-8") as f: config = yaml.safe_load(f) debug_mode = config.get("debug", {}).get("enabled", False) debug_target = config.get("debug", {}).get("target_site", "") except Exception as e: print(f"⚠️ 读取 config.yaml 异常，将使用全量模式启动: {e}") # 动态确定需要挂载启动的网页 active_sites = {} if debug_mode and debug_target in SITES_CONFIG: print(f"\n🛠️ 【调试模式激活】当前中控台仅挂载并开启目标站点: [{debug_target}]") active_sites = {debug_target: SITES_CONFIG[debug_target]} else: active_sites = SITES_CONFIG 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(): print(f">> 正在启动【{site_name}】页面: {url}") page = context.new_page() page.goto(url) pages_map[site_name] = page 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()} print("\n>> 正在静默轮询全网点就绪状态 (自动免密/手工登录均可激活)...") while not all(ready_status.values()): for site_name, page in pages_map.items(): if not ready_status[site_name]: try: # 0.5秒轻量级探针，避免阻塞主循环 if page.locator(READY_SELECTORS[site_name]).is_visible( timeout=500 ): 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)}] ... (请在浏览器中操作)" ) page.wait_for_timeout(3000) # 挂起 3 秒后执行下一轮盘点 print("\n====================================================") print("【接管阶段】所有活跃网点均已就绪，正在执行环境净化...") print("====================================================") # 顺心环境净化 if "顺心" in pages_map: try: sx_page = pages_map["顺心"] sx_page.bring_to_front() print(">> 正在处理【顺心】弹窗与遮罩...") 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(" ✅ 【顺心】环境准备就绪！") 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: pass try: if bs_page.locator("button:has-text('关 闭')").is_visible( timeout=500 ): bs_page.locator("button:has-text('关 闭')").click() handled_any = True except: 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(" ✅ 【韵达】工作区状态就绪！") def is_site_ready(site_name): if site_name not in pages_map: print(f"\n🚫 【安全拦截】当前处于局部调试，[{site_name}] 未挂载加载！") return False return True 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(" 全局离线数据引擎") print(" [9] 执行 - 异常数据清洗比对 (Left Anti-Join)") print("-" * 52) print(" [0] 退出系统") print("====================================================") choice = input("请输入任务编号并回车: ") try: if choice == "1" and is_site_ready("顺心"): pages_map["顺心"].bring_to_front() site_shunxin.shunxin_expected_download(pages_map["顺心"]) elif choice == "2" and is_site_ready("顺心"): pages_map["顺心"].bring_to_front() site_shunxin.shunxin_actual_download(pages_map["顺心"]) elif choice == "3" and is_site_ready("百世"): pages_map["百世"].bring_to_front() site_baishi.baishi_download_undelivered_data(pages_map["百世"]) elif choice == "4" and is_site_ready("中通"): pages_map["中通"].bring_to_front() site_zto.zto_expected_download(pages_map["中通"]) elif choice == "5" and is_site_ready("中通"): pages_map["中通"].bring_to_front() site_zto.zto_actual_download(pages_map["中通"]) elif choice == "6" and is_site_ready("韵达"): pages_map["韵达"].bring_to_front() site_yunda.yunda_expected_download(pages_map["韵达"]) elif choice == "7" and is_site_ready("韵达"): pages_map["韵达"].bring_to_front() site_yunda.yunda_actual_download(pages_map["韵达"]) elif choice == "9": site_name = input( "请输入要比对的网点名称 (如 顺心/中通/韵达): " ).strip() if not site_name: site_name = "顺心" task_process_undelivered_data(site_name) elif choice == "0": print("\n正在释放浏览器并安全登出系统...") break else: if choice not in ["1", "2", "3", "4", "5", "6", "7", "9", "0"]: print("\n⚠️ 无效输入，请查证后回车。") except Exception as e: print(f"❌ 调度枢纽异常: {e}") browser.close() print("中控守护进程安全退出。") if __name__ == "__main__": run_multi_site_daemon()