LogisticsHubIPA/InboundVerify
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Integrate Yunda site module and rewire router menu numbering

Browse Source 
- Add site_yunda.py: Yunda expected-arrival download (进站交接单查询)
  with Layui date-range setup, row double-click export, a retry state
  machine for the offline-export dialog, and a shared poll/download/
  merge engine; actual-arrival is a placeholder
- Wire Yunda into main_router: site URL, login detection, menu items,
  and read yunda.query_days from config.yaml
- Rework the hub menu: drop the standalone Shunxin local-compare entry,
  renumber modules (百世 3, 中通 4/5, 韵达 6/7), keep global compare [9]
- Streamline site-init cleanup and log wording

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Misaka
2026-06-20 18:20:46 +08:00
parent eb9bbe9b70
commit 194fbc7cda
2 changed files with 471 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

121
main_router.py
View File

@@ -9,12 +9,14 @@ 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", # 预设韵达快运大运系统入口
}
@@ -82,10 +84,10 @@ def run_multi_site_daemon():
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}]")
print(f"\n🛠️ 【调试模式激活】当前中控台仅挂载并开启目标站点: [{debug_target}]")
active_sites = {debug_target: SITES_CONFIG[debug_target]}
else:
active_sites = SITES_CONFIG
@@ -109,7 +111,7 @@ def run_multi_site_daemon():
print("\n====================================================")
print("⚠️ 【等待人工介入】")
print("请在弹出的浏览器中，人工完成已加载网点的登录！")
print("请在弹出的浏览器中，人工完成已开启网点的登录！")
print("====================================================")
input(">> 登录全部完成后，请在此处按下【回车键】正式接管中控台...")
@@ -124,32 +126,26 @@ def run_multi_site_daemon():
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()
sx_page.wait_for_timeout(1000)
print(" ✅ 【顺心】弹窗清理完毕，状态就绪！")
print(" ✅ 【顺心】环境准备就绪！")
except Exception as e:
print(f" ⚠️ 【顺心】初始化或弹窗清理异常 (如无弹窗可忽略): {e}")
print(f" ⚠️ 【顺心】初始化波动: {e}")
# 百世环境检测与“打地鼠”清理引擎
# 百世环境打地鼠
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(" 🎉 登录成功！系统已接管百世浏览器。")
print(" >> 启动无序弹窗清理机制 (打地鼠模式)...")
bs_page.wait_for_timeout(2000)
for round_idx in range(5):
@@ -160,45 +156,19 @@ def run_multi_site_daemon():
for i in range(read_btns.count()):
if read_btns.nth(i).is_visible():
read_btns.nth(i).click()
bs_page.wait_for_timeout(500)
handled_any = True
except Exception:
except:
pass
try:
close_config = bs_page.locator("button:has-text('关 闭')")
if close_config.is_visible():
close_config.click()
bs_page.wait_for_timeout(500)
if bs_page.locator("button:has-text('关 闭')").is_visible():
bs_page.locator("button:has-text('关 闭')").click()
handled_any = True
except Exception:
except:
pass
try:
notice_close = bs_page.locator(
"a.ant-notification-notice-close"
)
if notice_close.is_visible():
notice_close.click()
bs_page.wait_for_timeout(500)
handled_any = True
except Exception:
pass
try:
ad_close = bs_page.locator("svg[data-icon='close-circle']")
if ad_close.is_visible():
ad_close.click()
bs_page.wait_for_timeout(500)
handled_any = True
except Exception:
pass
if not handled_any:
break
bs_page.wait_for_timeout(1000)
print(" ✅ 【百世】界面净化完毕，状态就绪！")
print(" ✅ 【百世】界面净化完毕！")
except Exception as e:
print(f" ⚠️ 【百世】初始化异常: {e}")
@@ -208,19 +178,29 @@ def run_multi_site_daemon():
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}")
# 路由拦截器
# 韵达环境检测
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}")
def is_site_ready(site_name):
if site_name not in pages_map:
print(
f"\n🚫 【拒绝执行】当前处于单站点调试模式，或【{site_name}】未加载，无法执行此操作！"
)
print(f"\n🚫 【安全拦截】当前处于局部调试，[{site_name}] 未挂载加载！")
return False
return True
@@ -228,22 +208,25 @@ def run_multi_site_daemon():
print("\n====================================================")
print(" 物流数据多端提取总枢纽 ")
if debug_mode:
print(f" [ 🛠️ 调试模式已激活 : {debug_target} ]")
print(f" [ 🛠️ 调试模式独立聚焦 : {debug_target} ]")
print("====================================================")
print(" 模块一：【顺心】数据处理流")
print(" [1] 执行 - 应到货物数据下载")
print(" [2] 执行 - 实到货物数据下载")
print(" [3] 执行 - 应到未到数据清洗比对 (本地运算)")
print("-" * 52)
print(" 模块二：【百世】数据处理流")
print(" [4] 执行 - 一键提取应到未到异常数据")
print(" [3] 执行 - 一键提取应到未到异常数据")
print("-" * 52)
print(" 模块三：【中通】数据处理流")
print(" [5] 执行 - 应到货物数据下载")
print(" [6] 执行 - 实到货物数据下载")
print(" [4] 执行 - 应到货物数据下载")
print(" [5] 执行 - 实到货物数据下载")
print("-" * 52)
print(" 全局数据引擎")
print(" [9] 执行 - 离线异常数据清洗比对 (Left Anti-Join)")
print(" 模块四：【韵达】数据处理流")
print(" [6] 执行 - 应到货物数据下载")
print(" [7] 执行 - 实到货物数据下载 (待开发)")
print("-" * 52)
print(" 全局离线数据引擎")
print(" [9] 执行 - 异常数据清洗比对 (Left Anti-Join)")
print("-" * 52)
print(" [0] 退出系统")
print("====================================================")
@@ -257,33 +240,39 @@ def run_multi_site_daemon():
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("顺心"):
task_process_undelivered_data("顺心")
elif choice == "4" and is_site_ready("百世"):
elif choice == "3" and is_site_ready("百世"):
pages_map["百世"].bring_to_front()
site_baishi.baishi_download_undelivered_data(pages_map["百世"])
elif choice == "5" and is_site_ready("中通"):
elif choice == "4" and is_site_ready("中通"):
pages_map["中通"].bring_to_front()
site_zto.zto_expected_download(pages_map["中通"])
elif choice == "6" and is_site_ready("中通"):
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()
site_name = input(
"请输入要比对的网点名称 (如 顺心/中通/韵达): "
).strip()
if not site_name:
site_name = "顺心"
task_process_undelivered_data(site_name)
elif choice == "0":
print("\n准备退出程序，释放浏览器资源...")
print("\n正在释放浏览器并安全登出系统...")
break
else:
if choice not in ["1", "2", "3", "4", "5", "6", "9", "0"]:
print("\n⚠️ 无效输入，请重新选一下。")
if choice not in ["1", "2", "3", "4", "5", "6", "7", "9", "0"]:
print("\n⚠️ 无效输入，请查证后回车。")
except Exception as e:
print(f"❌ 调度执行异常: {e}")
print(f"❌ 调度枢纽异常: {e}")
browser.close()
print("系统已安全关闭。")
print("中控守护进程安全退出。")
if __name__ == "__main__":

416
site_yunda.py Normal file
View File

@@ -0,0 +1,416 @@
# site_yunda.py
import os
import re
import yaml
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
for frame in page.frames:
try:
if frame.get_by_text(text_indicator).count() > 0:
return frame
except:
pass
page.wait_for_timeout(300)
raise TimeoutError(
f"爆栈超时：未能在活动框架中死守到包含 [{text_indicator}] 的视窗。"
)
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}] 的业务框架。")
def yunda_smart_menu_click(page, menu_path):
"""韵达排他性二级树形菜单智能展开器"""
print(f">> 正在导航韵达菜单: {' -> '.join(menu_path)}")
for item in menu_path:
locator = page.locator(".el-submenu__title, .el-menu-item", has_text=item).first
locator.click()
page.wait_for_timeout(600)
def yunda_expected_download(page):
"""韵达：应到货物数据下载"""
print("\n▶ 开始执行【韵达 - 应到货物数据下载】任务...")
target_task_title = "进站主单表"
download_dir = os.path.join(os.getcwd(), "downloads")
if not os.path.exists(download_dir):
os.makedirs(download_dir)
export_times = []
try:
# 1. 验证首页并智能路由
page.locator(".el-menu-item", has_text="首页").wait_for(
state="visible", timeout=15000
)
print("✅ 韵达工作台首页成功加载")
yunda_smart_menu_click(page, ["运营管理", "进站管理", "进站交接单查询"])
print(">> [雷达扫描] 正在跨域动态追踪【进站交接单查询】业务窗体...")
ws_frame = _wait_and_get_frame_by_selector(page, "#startTime")
ws_frame.locator("#startTime").wait_for(state="attached", timeout=15000)
print("✅ 进站交接单查询工作区就绪")
# 2. 从配置文件中解析并计算绝对日期跨度
query_days = 1
try:
if os.path.exists("config.yaml"):
with open("config.yaml", "r", encoding="utf-8") as f:
config = yaml.safe_load(f) or {}
query_days = int(config.get("yunda", {}).get("query_days", 1))
except Exception as e:
print(f" ⚠️ 读取 config.yaml 失败，默认查询 1 天: {e}")
today = datetime.now()
start_date = today - timedelta(days=(query_days - 1))
today_ymd = f"{today.year}-{today.month}-{today.day}"
start_date_ymd = f"{start_date.year}-{start_date.month}-{start_date.day}"
print(f">> 正在精准对焦时间区间: [{start_date_ymd}] 至 [{today_ymd}]")
# 设定起始时间
print(" >> 呼出起始时间控件...")
page.wait_for_timeout(1000)
ws_frame.locator("#startTime").click(force=True)
calendar1 = ws_frame.locator(".layui-laydate:visible").first
calendar1.wait_for(state="visible", timeout=5000)
calendar1.locator(f"td[lay-ymd='{start_date_ymd}']").click()
calendar1.locator(".laydate-btns-confirm").click()
page.wait_for_timeout(400)
# 设定截止时间
print(" >> 呼出截止时间控件...")
ws_frame.locator("#endTime").click(force=True)
calendar2 = ws_frame.locator(".layui-laydate:visible").first
calendar2.wait_for(state="visible", timeout=5000)
calendar2.locator(f"td[lay-ymd='{today_ymd}']").click()
calendar2.locator(".laydate-btns-confirm").click()
page.wait_for_timeout(500)
# 3. 多维状态机静默加载校验
print(">> 正在触发现场查询数据流...")
ws_frame.locator("a.btn-success", has_text="查询").click()
page.wait_for_timeout(1000)
sum_panel = ws_frame.locator("#sum")
has_data = False
if sum_panel.is_visible():
sum_text = sum_panel.inner_text()
match_tickets = re.search(r"进站实际票数:(\d+)", sum_text)
if match_tickets and int(match_tickets.group(1)) > 0:
has_data = True
print(
f" ✅ 深度断言：局部统计面板加载完毕，实际票数: [{match_tickets.group(1)}]，执行穿透。"
)
if not has_data:
if ws_frame.locator(
".no-records-found", has_text="没有找到匹配的记录"
).is_visible():
print(
" ⚠️ 确认为空数据环境：系统提示【没有找到匹配的记录】，正在执行复位净化..."
)
page.locator(".tags-view-item", has_text="进站交接单查询").locator(
".el-icon-close"
).click()
return
# 4. 深度等待表格第一行数据行渲染就绪
ws_frame.locator("#exampleTable1 tbody tr[data-index='0']").wait_for(
state="visible", timeout=10000
)
main_rows = ws_frame.locator("#exampleTable1 tbody tr[data-index]")
row_count = main_rows.count()
print(f">> 当前视窗共捕获到活跃交接单记录: {row_count} 条")
# 5. 循环双击穿透提交
for i in range(row_count):
print(f" ⏳ 正在处理第 {i+1}/{row_count} 个交接单模块...")
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}")
current_row.dblclick()
ws_frame.locator("#docSum").wait_for(state="visible", timeout=15000)
page.wait_for_timeout(500)
ws_frame.locator('a.btn-info[onclick*="exportFile"]').click()
ws_frame.locator(".layui-layer-title", has_text="数据导出").wait_for(
state="visible", timeout=15000
)
export_frame = ws_frame.frame_locator('iframe[name="target1"]')
# 首次触发全选所有字段
export_frame.locator(".allRight").click()
page.wait_for_timeout(400)
# ====================================================================
# 🛡️ 智能容错重试状态机：完美兼容【离线建立失败】与【字段全选漏掉报错】
# ====================================================================
print(" >> 正在建立后台离线任务...")
task_success = False
for attempt in range(5):
export_frame.locator("#submitbutton", has_text="导出数据").click()
print(
f" -> 已触发第 {attempt + 1} 次【导出数据】指令，等待校验响应..."
)
# 依靠具防抖特性的 Link 角色定位“确定”按钮
confirm_link = export_frame.get_by_role("link", name="确定")
try:
# 等待提交重载后的新文档渲染出提示框
confirm_link.wait_for(state="visible", timeout=6000)
# 场景 1：建立成功
if export_frame.get_by_text("导出任务建立成功").is_visible():
print(" ✅ 判定通过：成功捕获到【导出任务建立成功】特征！")
confirm_link.click()
task_success = True
break
# 场景 2：核心优化 - 发现字段全选未生效报错
elif export_frame.get_by_text(
"请选择格式相应的导出字段"
).is_visible():
print(
" ⚠️ 警告：检测到【未选择字段】错误提示！正在清除弹窗并强行补点【全选】按钮..."
)
confirm_link.click()
page.wait_for_timeout(500)
# 再次执行强行补点全选
export_frame.locator(".allRight").click()
print(
" -> 【全选】按钮已重新触发补点，准备进入下一轮提交。"
)
page.wait_for_timeout(500)
# 场景 3：其他网络或未知异常提示
else:
print(
" ⚠️ 警告：检测到其他异常提示消息，正在清除弹窗并重试..."
)
confirm_link.click()
page.wait_for_timeout(1000)
except Exception as e:
print(
f" ⚠️ 提示：第 {attempt + 1} 次提交后未能在安全时间内捕获到提示框 ({e})，准备重试..."
)
page.wait_for_timeout(1000)
if not task_success:
raise RuntimeError(
"致命异常：连续 5 次尝试均无法成功建立任务，请检查网点后台事务层。"
)
# ====================================================================
# 关闭高层 Layui 挂载的数据导出窗口
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. 进入【导出服务】队列收割
_yunda_poll_and_download_tasks(
page,
export_times,
target_task_title,
download_dir,
"韵达-应到货物数据.xlsx",
)
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
# 8. 下载成果流
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▶ 开始执行【韵达 - 实到货物数据下载】任务...")
print("🚧 韵达实到数据功能规划中...")
pass