diff --git a/main_router.py b/main_router.py index 449840a..281dab8 100644 --- a/main_router.py +++ b/main_router.py @@ -16,7 +16,17 @@ SITES_CONFIG = { "顺心": "https://sxne.sxjdfreight.com", "百世": "https://v5.800best.com", "中通": "https://ws.zto56.com/", - "韵达": "https://ky-sso.yunda56.com", # 预设韵达快运大运系统入口 + "韵达": "https://ky-sso.yunda56.com", +} + +# ==================================================================== +# 🛡️ 网点特征注册表：定义每个网点登录成功、成功进入工作台的标志性控件 +# ==================================================================== +READY_SELECTORS = { + "顺心": 'h1:has-text("盟商门户网")', + "百世": 'h1[title="百世快运"]', + "中通": '.logo:has-text("网点版")', + "韵达": '.el-menu-item:has-text("首页")', } @@ -70,7 +80,7 @@ def task_process_undelivered_data(site_name="顺心"): def run_multi_site_daemon(): - """多网点自动化主控引擎""" + """多网点自动化主控引擎 (状态机卫语句驱动版)""" # 1. 读取配置文件 debug_mode = False @@ -99,25 +109,56 @@ def run_multi_site_daemon(): pages_map = {} print("\n====================================================") - print("【初始化阶段】正在构建多网点运行环境...") + print("【唤醒阶段】正在构建多网点并行运行环境...") print("====================================================") for site_name, url in active_sites.items(): - print(f">> 正在打开【{site_name}】页面: {url}") + print(f">> 正在启动【{site_name}】页面: {url}") page = context.new_page() page.goto(url) pages_map[site_name] = page - page.wait_for_timeout(1000) print("\n====================================================") - print("⚠️ 【等待人工介入】") - print("请在弹出的浏览器中，人工完成已开启网点的登录！") + print("【状态自检】启动前置智能接管机制...") print("====================================================") - input(">> 登录全部完成后，请在此处按下【回车键】正式接管中控台...") + # 针对支持纯代码自动登录的站点，在此处前置注入登录事件 + 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("【接管阶段】所有活跃网点均已就绪，正在执行环境净化...") print("====================================================") # 顺心环境净化 @@ -125,78 +166,53 @@ def run_multi_site_daemon(): try: sx_page = pages_map["顺心"] sx_page.bring_to_front() - print(">> 正在处理【顺心】网点初始状态...") - sx_page.wait_for_selector('h1:has-text("盟商门户网")', timeout=30000) - print(" 🎉 登录成功！系统已接管顺心浏览器。") - sx_page.wait_for_timeout(1000) - sx_page.locator("a").nth(4).click() - sx_page.wait_for_timeout(1000) - sx_page.get_by_role("button", name="Close").click() - sx_page.wait_for_timeout(1000) - sx_page.get_by_role("button", name="不再询问").click() + 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 as e: - print(f" ⚠️ 【顺心】初始化波动: {e}") + except Exception: + pass # 环境可能很干净没有弹窗，无视报错 # 百世环境打地鼠 if "百世" in pages_map: try: bs_page = pages_map["百世"] bs_page.bring_to_front() - print("\n>> 正在处理【百世】网点初始状态...") - bs_page.wait_for_selector('h1[title="百世快运"]', timeout=30000) - print(" 🎉 登录成功！系统已接管百世浏览器。") - bs_page.wait_for_timeout(2000) - - for round_idx in range(5): + 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(): + 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(): + 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(1000) + bs_page.wait_for_timeout(800) print(" ✅ 【百世】界面净化完毕！") - except Exception as e: - print(f" ⚠️ 【百世】初始化异常: {e}") + except Exception: + pass - # 中通环境检测 if "中通" in pages_map: - try: - zto_page = pages_map["中通"] - zto_page.bring_to_front() - print("\n>> 正在处理【中通】网点初始状态...") - zto_page.wait_for_selector('.logo:has-text("网点版")', timeout=30000) - print(" 🎉 登录成功！系统已接管中通浏览器。") - print(" ✅ 【中通】状态就绪！") - except Exception as e: - print(f" ⚠️ 【中通】初始化异常: {e}") + print(" ✅ 【中通】状态就绪！") - # 韵达环境检测 if "韵达" in pages_map: - try: - yd_page = pages_map["韵达"] - yd_page.bring_to_front() - print("\n>> 正在处理【韵达】网点初始状态...") - yd_page.wait_for_selector( - '.el-menu-item:has-text("首页")', timeout=30000 - ) - print(" 🎉 登录成功！系统已接管韵达快运浏览器。") - print(" ✅ 【韵达】工作区状态就绪！") - except Exception as e: - print(f" ⚠️ 【韵达】初始化异常: {e}") + print(" ✅ 【韵达】工作区状态就绪！") def is_site_ready(site_name): if site_name not in pages_map: diff --git a/site_yunda.py b/site_yunda.py index 26c4fdd..aa05752 100644 --- a/site_yunda.py +++ b/site_yunda.py @@ -7,48 +7,46 @@ from datetime import datetime, timedelta import pandas as pd -def _wait_and_get_frame(page, text_indicator, timeout_ms=20000): - """【文本雷达探测器】全域跨框架检索包含特定文本的活动上下文""" - start_time = datetime.now() - while (datetime.now() - start_time).total_seconds() * 1000 < timeout_ms: - try: - if page.get_by_text(text_indicator).count() > 0: - return page - except: - pass +def yunda_login(page): + """ + 韵达大运系统智能自动登录流 + 支持‘未登录自动填充提交’与‘已登录静默过客’双工模式 + """ + print(">> 正在探测韵达当前的登录会话状态...") + try: + # 定位“账号密码登录”切换按钮 + switch_btn = page.locator("span", has_text="账号密码登录") - for frame in page.frames: - try: - if frame.get_by_text(text_indicator).count() > 0: - return frame - except: - pass + # 设定 5 秒探针延迟。如果 5 秒内按钮可见，说明处于未登录的初始状态 + if switch_btn.is_visible(timeout=5000): + print(" -> 捕获到标准的未登录界面，正在强切至【账号密码登录】模式...") + switch_btn.click() + page.wait_for_timeout(500) - page.wait_for_timeout(300) - raise TimeoutError( - f"爆栈超时：未能在活动框架中死守到包含 [{text_indicator}] 的视窗。" - ) + # 从配置读取凭证（默认空串；真实凭据仅存于被忽略的 config.yaml） + username = "" + password = "" + if os.path.exists("config.yaml"): + with open("config.yaml", "r", encoding="utf-8") as f: + config = yaml.safe_load(f) or {} + yd_cfg = config.get("yunda", {}) + username = str(yd_cfg.get("username", "")) + password = str(yd_cfg.get("password", "")) + print(f" -> 正在程序化充填表单凭证 (账号: {username})...") + page.locator("#username").fill(username) + page.locator("#password").fill(password) + page.wait_for_timeout(300) -def _wait_and_get_frame_by_selector(page, selector, timeout_ms=20000): - """【控件雷达探测器】无视 IFrame 的 src 或层级，直接扫描谁包含指定 CSS 控件""" - start_time = datetime.now() - while (datetime.now() - start_time).total_seconds() * 1000 < timeout_ms: - try: - if page.locator(selector).count() > 0: - return page - except: - pass - - for frame in page.frames: - try: - if frame.locator(selector).count() > 0: - return frame - except: - pass - - page.wait_for_timeout(300) - raise TimeoutError(f"爆栈超时：全域未探测到包含控件 [{selector}] 的业务框架。") + print(" -> 正在点击【登录】按钮并提交表单...") + page.locator('button[type="submit"]', has_text="登录").click() + page.wait_for_timeout(1000) + else: + print( + " -> 探针未发现登录按钮，判定当前会话已处于持久化登录状态，静默穿透。" + ) + except Exception as e: + print(f" ⚠️ 自动登录状态机探测发生波动（可能已处于工作台内部）: {e}") def yunda_smart_menu_click(page, menu_path): @@ -108,8 +106,9 @@ def yunda_expected_download(page): yunda_smart_menu_click(page, ["运营管理", "进站管理", "进站交接单查询"]) - print(">> [雷达扫描] 正在跨域动态追踪【进站交接单查询】业务窗体...") - ws_frame = _wait_and_get_frame_by_selector(page, "#startTime") + print(">> 正在跨域动态追踪【进站交接单查询】业务窗体...") + # 彻底移除会引起竞速超时坑的 custom 探测器，使用原生懒加载 frame_locator + ws_frame = page.frame_locator("section iframe") ws_frame.locator("#startTime").wait_for(state="attached", timeout=15000) print("✅ 进站交接单查询工作区就绪") @@ -212,7 +211,17 @@ def yunda_expected_download(page): current_row = ws_frame.locator("#exampleTable1 tbody tr[data-index]").nth(i) raw_no = current_row.locator("td").nth(1).inner_text().strip() - print(f" -> 锁定交接单号: {raw_no}") + # ==================================================================== + # 🛡️ 状态机拦截卫语句：识别绑定状态，过滤已绑定交接单 + # ==================================================================== + bind_status = current_row.locator("td").nth(2).inner_text().strip() + print(f" -> 锁定提取单号: {raw_no} [绑定状态: {bind_status}]") + + if bind_status == "已绑定": + print(" ⏭️ 状态拦截：该交接单处于【已绑定】状态，安全跳过。") + continue + + # ==================================================================== current_row.dblclick() @@ -264,20 +273,23 @@ def yunda_expected_download(page): ws_frame.locator(".layui-layer-close1").click() page.wait_for_timeout(500) - ws_frame.locator("#myTab a", has_text="交接单信息").click() page.wait_for_timeout(800) - export_times.append(datetime.now()) - # 6. 一阶段全量数据提交闭环，销毁当前业务 Tab print(">> 📤 任务提交流闭环，正在执行【进站交接单查询】工作台销毁...") page.locator(".tags-view-item", has_text="进站交接单查询").locator( ".el-icon-close" ).click() page.wait_for_timeout(500) - # 7. 进入【导出服务】队列收割 + # 防线：如果全部记录都被跳过了，export_times 为空，直接结束 + if not export_times: + print( + ">> ⚠️ 本次查询未产生任何有效的离线下载任务（全部空单或已被跳过），中止后端收割流。" + ) + return + _yunda_poll_and_download_tasks( page, export_times, @@ -290,137 +302,6 @@ def yunda_expected_download(page): print(f"\n❌ 任务执行过程中发生异常: {e}") -def _yunda_poll_and_download_tasks( - page, export_times, target_task_title, download_dir, final_filename -): - """韵达专属离线任务轮询下载引擎""" - print("\n>> 正在前往【导出服务】中心...") - yunda_smart_menu_click(page, ["基础数据", "导出服务"]) - - export_ws_frame = page.frame_locator("section iframe") - - export_ws_frame.get_by_role("cell", name="模块名称", exact=True).wait_for( - state="visible", timeout=15000 - ) - page.wait_for_timeout(1000) - - print(">> 离线文件队列已对接，启动【1分钟高频精确校对+缺单局部重载刷新】断言...") - total_expected = len(export_times) - - while True: - task_rows = export_ws_frame.locator( - ".datagrid-view2 .datagrid-btable tbody tr.datagrid-row" - ) - row_count = task_rows.count() - - ready_indices = [] - processing_indices = [] - - for idx in range(row_count): - row = task_rows.nth(idx) - module_name = row.locator("td[field='modueName']").inner_text().strip() - status_name = row.locator("td[field='fileStatus']").inner_text().strip() - create_time_str = ( - row.locator("td[field='createdTime']").inner_text().strip() - ) - - if module_name == target_task_title: - try: - row_time = datetime.strptime(create_time_str, "%Y-%m-%d %H:%M:%S") - matched = any( - abs((row_time - et).total_seconds()) <= 60 - for et in export_times - ) - - if matched: - if status_name == "导出完成": - ready_indices.append(idx) - else: - processing_indices.append(idx) - except Exception: - pass - - total_found = len(ready_indices) + len(processing_indices) - print( - f" 📊 状态研判：自建期望数 [{total_expected}]，实际入表 [{total_found}] (完成 [{len(ready_indices)}]，生成中 [{len(processing_indices)}])" - ) - - if total_found < total_expected or len(processing_indices) > 0: - print(" ⏳ 队列未齐，执行【点击查询按钮】触发局部无痕刷新...") - export_ws_frame.locator( - "#ydkyimport_basic_export_searchData1_ky_export_common" - ).click() - page.wait_for_timeout(3000) - else: - print(">> ✅ 所有目标离线任务全量就绪！开始依序接入文件流...") - break - - # 下载成果流 - downloaded_files = [] - for row_idx in ready_indices: - try: - target_row = export_ws_frame.locator( - ".datagrid-view2 .datagrid-btable tbody tr.datagrid-row" - ).nth(row_idx) - time_flag = ( - target_row.locator("td[field='createdTime']").inner_text().strip() - ) - print(f" 🎯 触发下载 -> 离线任务时间节点: [{time_flag}] ...") - - with page.expect_download() as download_info: - target_row.locator("td[field='extreFile'] a").get_by_text( - "下载" - ).first.click() - - download = download_info.value - - safe_timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f") - custom_filename = f"韵达_temp_{safe_timestamp}.xlsx" - save_path = os.path.join(download_dir, custom_filename) - - download.save_as(save_path) - downloaded_files.append(save_path) - print(f" ⬇️ 文件已用安全序列号落盘: downloads/{custom_filename}") - page.wait_for_timeout(500) - except Exception as e: - print(f" ❌ 文件流接收失败: {e}") - - # 环境清理：彻底关闭“导出服务”Tab - print(">> 📥 【导出服务】数据提取链闭环，正在执行当前 Tab 窗口销毁...") - try: - page.locator(".tags-view-item", has_text="导出服务").locator( - ".el-icon-close" - ).click() - print(" ✅ 【导出服务】工作区已安全关闭。") - except: - pass - - # 9. 合并扁平数据集 - if downloaded_files: - print("\n>> 🧪 正在启动离线数据清洗与高能扁平合并流...") - all_dfs = [] - for file_path in downloaded_files: - try: - df = pd.read_excel(file_path, dtype=str) - if not df.empty: - all_dfs.append(df) - except: - pass - - if all_dfs: - combined_df = pd.concat(all_dfs, ignore_index=True) - final_output = os.path.join(download_dir, final_filename) - combined_df.to_excel(final_output, index=False) - print(f"====================================================") - print(f" 🎉 恭喜！韵达网点数据拉取合流大功告成！") - print(f" 📁 最终存储位置: {final_output}") - print(f"====================================================") - - for file_path in downloaded_files: - os.remove(file_path) - print(" ✅ 临时缓存阵列已无缝净化。") - - def yunda_actual_download(page): """韵达：实到货物数据下载""" print("\n▶ 开始执行【韵达 - 实到货物数据下载】任务...") @@ -440,8 +321,8 @@ def yunda_actual_download(page): yunda_smart_menu_click(page, ["报表管理", "扫描记录查询"]) - print(">> [雷达扫描] 正在跨域动态追踪【扫描记录查询】业务窗体...") - ws_frame = _wait_and_get_frame_by_selector(page, "#startDate") + print(">> 正在跨域动态追踪【扫描记录查询】业务窗体...") + ws_frame = page.frame_locator("section iframe") ws_frame.locator( ".no-records-found", has_text="没有找到匹配的记录" @@ -489,7 +370,6 @@ def yunda_actual_download(page): page.wait_for_timeout(800) - # 同样在这里加上防 strict mode 穿透的 first 隔离 loading_mask = ws_frame.locator( ".fixed-table-loading", has_text="正在努力地加载数据中" ).first @@ -580,6 +460,11 @@ def yunda_actual_download(page): ).click() page.wait_for_timeout(500) + # 防线：双重保护 + if not export_times: + print(">> ⚠️ 本次查询未产生有效的离线下载任务，中止后端收割流。") + return + # 6. 收割下载 _yunda_poll_and_download_tasks( page, @@ -591,3 +476,131 @@ def yunda_actual_download(page): except Exception as e: print(f"\n❌ 任务执行过程中发生异常: {e}") + + +def _yunda_poll_and_download_tasks( + page, export_times, target_task_title, download_dir, final_filename +): + """韵达专属离线任务轮询下载引擎""" + print("\n>> 正在前往【导出服务】中心...") + yunda_smart_menu_click(page, ["基础数据", "导出服务"]) + + export_ws_frame = page.frame_locator("section iframe") + + export_ws_frame.get_by_role("cell", name="模块名称", exact=True).wait_for( + state="visible", timeout=15000 + ) + page.wait_for_timeout(1000) + + print(">> 离线文件队列已对接，启动【1分钟高频精确校对+缺单局部重载刷新】断言...") + total_expected = len(export_times) + + while True: + task_rows = export_ws_frame.locator( + ".datagrid-view2 .datagrid-btable tbody tr.datagrid-row" + ) + row_count = task_rows.count() + + ready_indices = [] + processing_indices = [] + + for idx in range(row_count): + row = task_rows.nth(idx) + module_name = row.locator("td[field='modueName']").inner_text().strip() + status_name = row.locator("td[field='fileStatus']").inner_text().strip() + create_time_str = ( + row.locator("td[field='createdTime']").inner_text().strip() + ) + + if module_name == target_task_title: + try: + row_time = datetime.strptime(create_time_str, "%Y-%m-%d %H:%M:%S") + matched = any( + abs((row_time - et).total_seconds()) <= 60 + for et in export_times + ) + + if matched: + if status_name == "导出完成": + ready_indices.append(idx) + else: + processing_indices.append(idx) + except Exception: + pass + + total_found = len(ready_indices) + len(processing_indices) + print( + f" 📊 状态研判：自建期望数 [{total_expected}]，实际入表 [{total_found}] (完成 [{len(ready_indices)}]，生成中 [{len(processing_indices)}])" + ) + + if total_found < total_expected or len(processing_indices) > 0: + print(" ⏳ 队列未齐，执行【点击查询按钮】触发局部无痕刷新...") + export_ws_frame.locator( + "#ydkyimport_basic_export_searchData1_ky_export_common" + ).click() + page.wait_for_timeout(3000) + else: + print(">> ✅ 所有目标离线任务全量就绪！开始依序接入文件流...") + break + + downloaded_files = [] + for row_idx in ready_indices: + try: + target_row = export_ws_frame.locator( + ".datagrid-view2 .datagrid-btable tbody tr.datagrid-row" + ).nth(row_idx) + time_flag = ( + target_row.locator("td[field='createdTime']").inner_text().strip() + ) + print(f" 🎯 触发下载 -> 离线任务时间节点: [{time_flag}] ...") + + with page.expect_download() as download_info: + target_row.locator("td[field='extreFile'] a").get_by_text( + "下载" + ).first.click() + + download = download_info.value + + safe_timestamp = datetime.now().strftime("%Y%m%d_%H%M%S_%f") + custom_filename = f"韵达_temp_{safe_timestamp}.xlsx" + save_path = os.path.join(download_dir, custom_filename) + + download.save_as(save_path) + downloaded_files.append(save_path) + print(f" ⬇️ 文件已用安全序列号落盘: downloads/{custom_filename}") + page.wait_for_timeout(500) + except Exception as e: + print(f" ❌ 文件流接收失败: {e}") + + print(">> 📥 【导出服务】数据提取链闭环，正在执行当前 Tab 窗口销毁...") + try: + page.locator(".tags-view-item", has_text="导出服务").locator( + ".el-icon-close" + ).click() + print(" ✅ 【导出服务】工作区已安全关闭。") + except: + pass + + if downloaded_files: + print("\n>> 🧪 正在启动离线数据清洗与高能扁平合并流...") + all_dfs = [] + for file_path in downloaded_files: + try: + df = pd.read_excel(file_path, dtype=str) + if not df.empty: + all_dfs.append(df) + except: + pass + + if all_dfs: + combined_df = pd.concat(all_dfs, ignore_index=True) + final_output = os.path.join(download_dir, final_filename) + combined_df.to_excel(final_output, index=False) + print(f"====================================================") + print(f" 🎉 恭喜！韵达网点数据拉取合流大功告成！") + print(f" 📁 最终存储位置: {final_output}") + print(f"====================================================") + + for file_path in downloaded_files: + os.remove(file_path) + print(" ✅ 临时缓存阵列已无缝净化。")