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

Add cross-flow automation test and tone down log/comment wording

Browse Source 
- main_router: add run_automation_test that runs each site's download
  flows in a cross sequence (expected/actual) plus a pass/fail report
  with timings; expose it as menu [8]
- Site download functions now return False on their exception paths so
  the test harness can record failures (baishi also flags config-read
  failure)
- Replace overblown log/comment phrasing across all modules with plain
  statements; trim docstrings

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Misaka
2026-06-20 23:33:22 +08:00
parent ab3f25a10e
commit 229e474b58
5 changed files with 298 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

174
main_router.py
View File

@@ -1,6 +1,7 @@
# main_router.py
import os
import time
import yaml
import pandas as pd
from playwright.sync_api import sync_playwright
@@ -22,7 +23,7 @@ SITES_CONFIG = {
}
# ====================================================================
# 🛡️ 网点特征注册表：定义每个网点登录成功、成功进入工作台的标志性控件
# 站点就绪特征：每个站点登录成功进入工作台后的标志性控件
# ====================================================================
READY_SELECTORS = {
"顺心": 'h1:has-text("盟商门户网")',
@@ -33,7 +34,7 @@ READY_SELECTORS = {
def task_process_undelivered_data(site_name="顺心"):
"""全局模块：应到未到异常件比对引擎 (支持动态网点前缀)"""
"""应到未到比对：找出应到但未实到的运单 (按站点前缀)"""
print(f"\n▶ 开始执行【{site_name} - 应到未到数据处理】任务...")
download_dir = DOWNLOAD_DIR
@@ -56,10 +57,10 @@ def task_process_undelivered_data(site_name="顺心"):
df_actual = pd.read_excel(actual_path, dtype=str, keep_default_na=False)
if "运单号" not in df_expected.columns or "运单号" not in df_actual.columns:
print("❌ 核心资产校验失败：数据源中缺失【运单号】字段，请检查导出配置。")
print("❌ 校验失败：数据源缺少【运单号】字段，请检查导出配置。")
return
print(">> 正在启动多维数据集比对引擎...")
print(">> 正在比对应到与实到数据...")
# 统一运单号为去空白字符串，消除 int/float 与 str 混读导致 isin 永不命中的隐患
df_expected["运单号"] = df_expected["运单号"].astype(str).str.strip()
@@ -78,20 +79,120 @@ def task_process_undelivered_data(site_name="顺心"):
]
df_output = df_undelivered[available_columns]
print(f">> 筛选完毕！共捕获到异常【应到未到】货物数据: {len(df_output)} 条。")
print(f">> 比对完成，共筛选出【应到未到】运单: {len(df_output)} 条。")
df_output.to_excel(output_path, index=False)
print(f"====================================================")
print(f" 🎉 异常比对流完成！独立数据已安全输出。")
print(f" 📁 成果归档路径: {output_path}")
print(f" 比对完成，结果已输出。")
print(f" 📁 输出路径: {output_path}")
print(f"====================================================")
except Exception as e:
print(f"❌ 数据处理引擎在执行连接和输出时发生致命异常: {e}")
print(f"❌ 比对过程中发生异常: {e}")
# ====================================================================
# 自动化测试入口
# ====================================================================
# 每个(非百世)站点的交叉测试序列：覆盖两种下载流程之间的全部 4 种相邻转换，
# 用于验证无论上一个流程把页面留在什么状态，下一个流程都能正常运行：
# 应到->实到、实到->应到、应到->应到、实到->实到
CROSS_TEST_SEQUENCE = ["expected", "actual", "expected", "expected", "actual", "actual"]
def run_automation_test(pages_map):
"""自动化测试入口：按交叉序列逐个跑通各站点的下载流程，结束后打印统计报告。
判定规则：流程函数返回 False 或抛出异常记为 FAIL，其余记为 PASS。
"""
# 站点 -> {流程键: (中文名, 流程函数)}；百世为单流程，单独处理
flow_table = {
"顺心": {
"expected": ("应到", site_shunxin.shunxin_expected_download),
"actual": ("实到", site_shunxin.shunxin_actual_download),
},
"中通": {
"expected": ("应到", site_zto.zto_expected_download),
"actual": ("实到", site_zto.zto_actual_download),
},
"韵达": {
"expected": ("应到", site_yunda.yunda_expected_download),
"actual": ("实到", site_yunda.yunda_actual_download),
},
}
# 构建测试计划：[(站点, 流程中文名, 流程函数), ...]
plan = []
for site_name, flows in flow_table.items():
if site_name not in pages_map:
continue
for flow_key in CROSS_TEST_SEQUENCE:
label, func = flows[flow_key]
plan.append((site_name, label, func))
# 百世：单流程，跑一次即可
if "百世" in pages_map:
plan.append(("百世", "应到未到", site_baishi.baishi_download_undelivered_data))
if not plan:
print("\n⚠️ 当前没有已就绪的站点，无法执行自动化测试。")
return
total = len(plan)
print("\n====================================================")
print(f"自动化测试开始，共 {total} 个步骤。")
print("（双流程站点按应到/实到交叉序列执行，覆盖全部相邻转换）")
print("====================================================")
results = [] # [(站点, 流程, 状态, 耗时秒, 错误信息)]
for idx, (site_name, label, func) in enumerate(plan, start=1):
print("\n----------------------------------------------------")
print(f"[步骤 {idx}/{total}] 站点【{site_name}】流程【{label}】")
print("----------------------------------------------------")
page = pages_map[site_name]
start = time.time()
status = "PASS"
err = ""
try:
page.bring_to_front()
ret = func(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
@@ -108,7 +209,7 @@ def run_multi_site_daemon():
# 动态确定需要挂载启动的网页
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
@@ -120,7 +221,7 @@ def run_multi_site_daemon():
pages_map = {}
print("\n====================================================")
print("【唤醒阶段】正在构建多网点并行运行环境...")
print("【启动】正在打开各站点页面...")
print("====================================================")
for site_name, url in active_sites.items():
@@ -130,7 +231,7 @@ def run_multi_site_daemon():
pages_map[site_name] = page
print("\n====================================================")
print("【状态自检】启动前置智能接管机制...")
print("【登录检测】正在准备各站点登录...")
print("====================================================")
# 针对支持纯代码自动登录的站点，在此处前置注入登录事件
@@ -142,37 +243,37 @@ def run_multi_site_daemon():
print(f" ⚠️ 韵达前置自动登录模块发生波动: {e}")
# ====================================================================
# 🛡️ 就绪卫语句轮询器 (Ready Guard Polling)
# 无需死板的输入回车，全自动识别状态并无缝放行
# 就绪轮询 (Ready Guard Polling)
# 自动识别各站点登录完成状态，无需手动回车
# ====================================================================
ready_status = {site: False for site in active_sites.keys()}
print("\n>> 正在静默轮询全网点就绪状态 (自动免密/手工登录均可激活)...")
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秒轻量级探针，避免阻塞主循环
# 0.5 秒轻量探测，避免阻塞主循环
if page.locator(READY_SELECTORS[site_name]).is_visible(
timeout=500
):
ready_status[site_name] = True
print(f" 🎉 【{site_name}】探测到主页特征，状态已就绪！")
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)}] ... (请在浏览器中操作)"
f" ⏳ 等待以下站点完成登录: [{', '.join(pending_sites)}] ... (请在浏览器中操作)"
)
page.wait_for_timeout(3000) # 挂起 3 秒后执行下一轮盘点
page.wait_for_timeout(3000) # 等待 3 秒后进行下一轮检查
print("\n====================================================")
print("【接管阶段】所有活跃网点均已就绪，正在执行环境净化...")
print("【准备】所有站点已就绪，正在清理初始弹窗...")
print("====================================================")
# 顺心环境净化
# 顺心：处理初始弹窗
if "顺心" in pages_map:
try:
sx_page = pages_map["顺心"]
@@ -183,11 +284,11 @@ def run_multi_site_daemon():
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(" ✅ 【顺心】环境准备就绪！")
print(" ✅ 【顺心】初始弹窗处理完成。")
except Exception:
pass # 环境可能很干净没有弹窗，无视报错
# 百世环境打地鼠
# 百世：循环关闭初始弹窗
if "百世" in pages_map:
try:
bs_page = pages_map["百世"]
@@ -215,27 +316,27 @@ def run_multi_site_daemon():
if not handled_any:
break
bs_page.wait_for_timeout(800)
print(" ✅ 【百世】界面净化完毕！")
print(" ✅ 【百世】初始弹窗处理完成。")
except Exception:
pass
if "中通" in pages_map:
print(" ✅ 【中通】状态就绪！")
print(" ✅ 【中通】已就绪。")
if "韵达" in pages_map:
print(" ✅ 【韵达】工作区状态就绪！")
print(" ✅ 【韵达】已就绪。")
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
while True:
print("\n====================================================")
print(" 物流数据多端提取总枢纽 ")
print(" 物流数据下载主菜单 ")
if debug_mode:
print(f" [ 🛠️ 调试模式独立聚焦 : {debug_target} ]")
print(f" [ 调试模式，仅加载: {debug_target} ]")
print("====================================================")
print(" 模块一：【顺心】数据处理流")
print(" [1] 执行 - 应到货物数据下载")
@@ -252,7 +353,10 @@ def run_multi_site_daemon():
print(" [6] 执行 - 应到货物数据下载")
print(" [7] 执行 - 实到货物数据下载")
print("-" * 52)
print(" 全局离线数据引擎")
print(" 自动化测试")
print(" [8] 执行 - 全站点下载流程自动化测试 (交叉跑通校验)")
print("-" * 52)
print(" 全局离线数据处理")
print(" [9] 执行 - 异常数据清洗比对 (Left Anti-Join)")
print("-" * 52)
print(" [0] 退出系统")
@@ -282,6 +386,8 @@ def run_multi_site_daemon():
elif choice == "7" and is_site_ready("韵达"):
pages_map["韵达"].bring_to_front()
site_yunda.yunda_actual_download(pages_map["韵达"])
elif choice == "8":
run_automation_test(pages_map)
elif choice == "9":
site_name = input(
"请输入要比对的网点名称 (如 顺心/中通/韵达): "
@@ -290,16 +396,16 @@ def run_multi_site_daemon():
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", "7", "9", "0"]:
if choice not in ["1", "2", "3", "4", "5", "6", "7", "8", "9", "0"]:
print("\n⚠️ 无效输入，请查证后回车。")
except Exception as e:
print(f"❌ 调度枢纽异常: {e}")
print(f"❌ 任务调度异常: {e}")
browser.close()
print("中控守护进程安全退出。")
print("程序已退出。")
if __name__ == "__main__":

27
site_baishi.py
View File

@@ -21,19 +21,19 @@ def baishi_download_undelivered_data(page):
# 打开基础服务菜单面板
page.locator("div.nav-level1", has_text="基础服务").click()
# 【重要修复】：通过限制在菜单面板（nav-level2-wrapper）内查找，
# 完美避开页面右侧同名 Tab 标签页的干扰
# 限制在菜单面板（nav-level2-wrapper）内查找，
# 避免命中右侧同名标签页
page.locator(".nav-level2-wrapper").locator(
"a", has_text="扫描综合查询"
).click()
# 验证表格主界面加载完毕
page.get_by_role("tab", name="实时扫描率").wait_for(state="visible")
print("✅ 扫描综合查询界面加载完毕")
print("✅ 扫描综合查询界面已加载")
# 2. 精准定位并点击【到/接件扫描 -> 当日 -> 未扫】的数字控件
print(">> 正在解析表格，提取当日到件未扫明细...")
# 表头结构复杂，精准定位第一行数据的第 13 列（索引 12）
# 定位第一行数据的第 13 列（索引 12）
target_cell = page.locator(".ant-table-tbody > tr").first.locator("td").nth(12)
# 特殊情况处理：检查未扫数量是否为0
@@ -51,15 +51,15 @@ def baishi_download_undelivered_data(page):
page.wait_for_timeout(1000)
# 3. 触发导出设置模态框
print(">> 正在呼出导出配置面板...")
# 锁定在明细区域内的导出按钮，防止误点主表的导出
print(">> 正在打开导出配置面板...")
# 使用明细区域内的导出按钮，避免误点主表导出
detail_section.locator(".export-wrap a[title='导出']").click()
# 验证模态框弹出
page.locator(".ant-modal-title", has_text="导出设置").wait_for(state="visible")
# 4. 加载 YAML 配置并输入密码
print(">> 正在读取本地配置文件并进行授权验证...")
print(">> 正在读取配置并填写密码...")
password = ""
try:
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
@@ -72,14 +72,14 @@ def baishi_download_undelivered_data(page):
)
except Exception as e:
print(f" ⚠️ 读取 config.yaml 失败，请确保文件存在且格式正确: {e}")
return
return False
page.get_by_placeholder("请输入登录密码").fill(password)
# 5. 执行最终下载
print(">> 验证就绪，正在触发下载流...")
print(">> 正在下载...")
with page.expect_download() as download_info:
# 精准锁定模态框底部的“导 出”大按钮
# 点击模态框底部的“导 出”按钮
page.locator(".ant-modal-footer").get_by_role(
"button", name="导 出"
).click()
@@ -93,11 +93,12 @@ def baishi_download_undelivered_data(page):
download.save_as(save_path)
print(f"====================================================")
print(f" 🎉 恭喜！数据提取成功！")
print(f" ⬇️ 文件已落盘: {save_path}")
print(f" 提取成功。")
print(f" 已下载: {save_path}")
print(f"====================================================")
print("\n🎉 【百世 - 应到未到数据提取】全流程测试完毕！")
print("\n【百世 - 应到未到数据提取】流程结束。")
except Exception as e:
print(f"\n❌ 任务执行过程中发生异常: {e}")
return False

36
site_shunxin.py
View File

@@ -39,7 +39,7 @@ def shunxin_expected_download(page):
download_dir = DOWNLOAD_DIR
if not os.path.exists(download_dir):
os.makedirs(download_dir)
print(f">> 已创建专属下载文件夹: {download_dir}")
print(f">> 已创建下载目录: {download_dir}")
export_times = []
target_task_timestamps = []
@@ -124,7 +124,7 @@ def shunxin_expected_download(page):
page.wait_for_timeout(2000)
# 7. 轮询任务状态
print(">> 列表中已渲染，开始匹配并检查后端处理状态...")
print(">> 列表已加载，开始匹配并检查任务状态...")
while True:
rows = page.locator(".ant-table-tbody > tr.ant-table-row")
row_count = rows.count()
@@ -173,7 +173,7 @@ def shunxin_expected_download(page):
else:
target_task_timestamps = current_ready_timestamps
if len(target_task_timestamps) > 0:
print(">> ✅ 所有目标任务已就绪！开始并行下载...")
print(">> 所有目标任务已就绪，开始下载...")
break
# 8. 下载逻辑
@@ -183,7 +183,7 @@ def shunxin_expected_download(page):
target_row = page.locator(".ant-table-tbody > tr.ant-table-row").filter(
has=page.locator(f"td:nth-child(3):has-text('{time_str}')")
)
print(f" 🎯 触发下载 -> 任务 [{time_str}] ...")
print(f" 开始下载任务 [{time_str}] ...")
with page.expect_download() as download_info:
target_row.locator("td").nth(8).locator(
@@ -197,13 +197,13 @@ def shunxin_expected_download(page):
save_path = os.path.join(download_dir, custom_filename)
download.save_as(save_path)
downloaded_files.append(save_path)
print(f" ⬇️ 文件已落盘: downloads/{custom_filename}")
print(f" 已下载: downloads/{custom_filename}")
except Exception as e:
print(f" ❌ 下载任务 [{time_str}] 失败: {e}")
# 9. 合并数据 (修改了文件名前缀)
if downloaded_files:
print("\n>> 🧪 正在开始执行扁平数据高能合并流程...")
print("\n>> 正在合并下载的数据...")
all_data_frames = []
for file_path in downloaded_files:
try:
@@ -218,20 +218,21 @@ def shunxin_expected_download(page):
final_output_path = os.path.join(download_dir, "顺心-应到货物数据.xlsx")
combined_df.to_excel(final_output_path, index=False)
print(f"====================================================")
print(f" 🎉 恭喜！合并成功！最终输出路径: {final_output_path}")
print(f" 合并完成，输出文件: {final_output_path}")
print(f"====================================================")
for file_path in downloaded_files:
os.remove(file_path)
print("✅ 临时数据清理完毕。")
print("✅ 临时文件已清理。")
# 10. 关闭【数据导出】标签页（该页面工作已完成）
_close_tab(page, "数据导出")
print("\n🎉 【顺心 - 应到货物数据下载】全流程测试完毕！")
print("\n【顺心 - 应到货物数据下载】流程结束。")
except Exception as e:
print(f"\n❌ 任务执行过程中发生异常: {e}")
return False
def shunxin_actual_download(page):
@@ -296,7 +297,7 @@ def shunxin_actual_download(page):
page.wait_for_timeout(2000)
# 6. 轮询任务状态
print(">> 列表中已渲染，开始匹配并检查后端处理状态...")
print(">> 列表已加载，开始匹配并检查任务状态...")
while True:
rows = page.locator(".ant-table-tbody > tr.ant-table-row")
row_count = rows.count()
@@ -345,7 +346,7 @@ def shunxin_actual_download(page):
else:
target_task_timestamps = current_ready_timestamps
if len(target_task_timestamps) > 0:
print(">> ✅ 目标任务已就绪！开始下载...")
print(">> 目标任务已就绪，开始下载...")
break
# 7. 下载逻辑
@@ -355,7 +356,7 @@ def shunxin_actual_download(page):
target_row = page.locator(".ant-table-tbody > tr.ant-table-row").filter(
has=page.locator(f"td:nth-child(3):has-text('{time_str}')")
)
print(f" 🎯 触发下载 -> 任务 [{time_str}] ...")
print(f" 开始下载任务 [{time_str}] ...")
with page.expect_download() as download_info:
target_row.locator("td").nth(8).locator(
@@ -369,13 +370,13 @@ def shunxin_actual_download(page):
save_path = os.path.join(download_dir, custom_filename)
download.save_as(save_path)
downloaded_files.append(save_path)
print(f" ⬇️ 文件已落盘: downloads/{custom_filename}")
print(f" 已下载: downloads/{custom_filename}")
except Exception as e:
print(f" ❌ 下载任务 [{time_str}] 失败: {e}")
# 8. 合并数据 (修改了文件名前缀)
if downloaded_files:
print("\n>> 🧪 正在开始执行数据归档整理...")
print("\n>> 正在合并下载的数据...")
all_data_frames = []
for file_path in downloaded_files:
try:
@@ -390,17 +391,18 @@ def shunxin_actual_download(page):
final_output_path = os.path.join(download_dir, "顺心-实到货物数据.xlsx")
combined_df.to_excel(final_output_path, index=False)
print(f"====================================================")
print(f" 🎉 恭喜！处理成功！最终输出路径: {final_output_path}")
print(f" 合并完成，输出文件: {final_output_path}")
print(f"====================================================")
for file_path in downloaded_files:
os.remove(file_path)
print("✅ 临时数据清理完毕。")
print("✅ 临时文件已清理。")
# 9. 关闭【数据导出】标签页（该页面工作已完成）
_close_tab(page, "数据导出")
print("\n🎉 【顺心 - 实到货物数据下载】全流程测试完毕！")
print("\n【顺心 - 实到货物数据下载】流程结束。")
except Exception as e:
print(f"\n❌ 任务执行过程中发生异常: {e}")
return False

146
site_yunda.py
View File

@@ -10,18 +10,15 @@ from paths import DOWNLOAD_DIR, CONFIG_PATH
def yunda_login(page):
"""
韵达大运系统智能自动登录流
支持‘未登录自动填充提交’与‘已登录静默过客’双工模式
"""
print(">> 正在探测韵达当前的登录会话状态...")
"""韵达自动登录：未登录则填充表单并提交，已登录则跳过。"""
print(">> 正在检查韵达登录状态...")
try:
# 定位“账号密码登录”切换按钮
switch_btn = page.locator("span", has_text="账号密码登录")
# 设定 5 秒探针延迟。如果 5 秒内按钮可见，说明处于未登录的初始状态
# 5 秒内若出现该按钮，说明当前未登录
if switch_btn.is_visible(timeout=5000):
print(" -> 捕获到标准的未登录界面，正在强切至【账号密码登录】模式...")
print(" -> 检测到未登录界面，正在切换到【账号密码登录】...")
switch_btn.click()
page.wait_for_timeout(500)
@@ -35,7 +32,7 @@ def yunda_login(page):
username = str(yd_cfg.get("username", ""))
password = str(yd_cfg.get("password", ""))
print(f" -> 正在程序化充填表单凭证 (账号: {username})...")
print(f" -> 正在填充登录表单 (账号: {username})...")
page.locator("#username").fill(username)
page.locator("#password").fill(password)
page.wait_for_timeout(300)
@@ -45,18 +42,15 @@ def yunda_login(page):
page.wait_for_timeout(1000)
else:
print(
" -> 探针未发现登录按钮，判定当前会话已处于持久化登录状态，静默穿透。"
" -> 未发现登录按钮，判定为已登录，跳过。"
)
except Exception as e:
print(f" ⚠️ 自动登录状态机探测发生波动（可能已处于工作台内部）: {e}")
print(f" ⚠️ 登录检测出错（可能已在工作台内）: {e}")
def yunda_smart_menu_click(page, menu_path):
"""
韵达排他性多级树形菜单智能导航器（带防折叠状态断言机制）
采用 XPath 亲子轴绝对隔离技术，彻底斩断嵌套手风琴菜单的向下漏包干扰
"""
print(f">> 正在智能路由韵达菜单: {' -> '.join(menu_path)}")
"""韵达多级菜单导航：展开父级菜单并点击目标项（已展开则跳过，避免误折叠）。"""
print(f">> 导航韵达菜单: {' -> '.join(menu_path)}")
for item in menu_path:
title_locator = page.locator(
@@ -74,16 +68,16 @@ def yunda_smart_menu_click(page, menu_path):
is_opened = "is-opened" in current_class
if not is_opened:
print(f" -> 探测到父菜单 [{item}] 当前处于【收起】状态，执行点击展开")
print(f" -> 父菜单 [{item}] 处于收起状态，点击展开")
title_locator.click()
page.wait_for_timeout(500)
else:
print(
f" -> 探测到父菜单 [{item}] 当前已经处于【展开】状态，安全跳过点击（防折叠保护激活）"
f" -> 父菜单 [{item}] 已展开，跳过点击"
)
elif leaf_locator.is_visible():
print(f" -> 成功对焦目标叶子节点 [{item}]，直接执行跳转点击")
print(f" -> 点击菜单项 [{item}]")
leaf_locator.click()
page.wait_for_timeout(1000)
@@ -100,20 +94,20 @@ def yunda_expected_download(page):
export_times = []
try:
# 1. 验证首页并智能路由
# 1. 验证首页并导航菜单
page.locator(".el-menu-item", has_text="首页").wait_for(
state="visible", timeout=15000
)
print("✅ 韵达工作台首页成功加载")
print("✅ 韵达工作台首页已加载")
yunda_smart_menu_click(page, ["运营管理", "进站管理", "进站交接单查询"])
print(">> 正在跨域动态追踪【进站交接单查询】业务窗体...")
# 彻底移除会引起竞速超时坑的 custom 探测器，使用原生懒加载 frame_locator
print(">> 正在定位【进站交接单查询】iframe...")
# 使用 frame_locator 定位业务 iframe
ws_frame = page.frame_locator("section iframe")
ws_frame.locator("#startTime").wait_for(state="attached", timeout=15000)
print("✅ 进站交接单查询工作区就绪")
print("✅ 进站交接单查询页面就绪")
# 2. 从配置文件中解析并计算绝对日期跨度
query_days = 1
@@ -131,10 +125,10 @@ def yunda_expected_download(page):
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(f">> 设置查询时间范围: [{start_date_ymd}] 至 [{today_ymd}]")
# 设定起始时间
print(" >> 呼出起始时间控件...")
print(" >> 设置起始时间...")
page.wait_for_timeout(1000)
ws_frame.locator("#startTime").click(force=True)
@@ -145,7 +139,7 @@ def yunda_expected_download(page):
page.wait_for_timeout(400)
# 设定截止时间
print(" >> 呼出截止时间控件...")
print(" >> 设置截止时间...")
ws_frame.locator("#endTime").click(force=True)
calendar2 = ws_frame.locator(".layui-laydate:visible").first
@@ -154,21 +148,21 @@ def yunda_expected_download(page):
calendar2.locator(".laydate-btns-confirm").click()
page.wait_for_timeout(500)
# 3. 多维状态机静默加载校验
print(">> 正在触发现场查询数据流...")
# 3. 等待数据加载完成
print(">> 正在执行查询...")
ws_frame.locator("a.btn-success", has_text="查询").click()
page.wait_for_timeout(800)
# ====================================================================
# 🛡️ 核心修复：将 Loading 蒙层和数据断言限制在当前的 #tab-1 结界内部
# 彻底解决多标签页导致 Strict Mode (5 elements found) 的污染问题
# 将 Loading 蒙层与数据判断限制在 #tab-1 内
# 避免多标签页命中多个元素 (Strict Mode)
# ====================================================================
loading_mask = ws_frame.locator(
"#tab-1 .fixed-table-loading", has_text="正在努力地加载数据中"
).first
if loading_mask.is_visible():
print(" ⏳ 检测到专属数据加载罩，正在等待后端重载返回...")
print(" ⏳ 检测到数据加载遮罩，等待加载完成...")
loading_mask.wait_for(state="hidden", timeout=30000)
page.wait_for_timeout(500)
@@ -182,16 +176,16 @@ def yunda_expected_download(page):
if match_tickets and int(match_tickets.group(1)) > 0:
has_data = True
print(
f" ✅ 深度断言：局部统计面板加载完毕，实际票数: [{match_tickets.group(1)}]，执行穿透。"
f" ✅ 统计面板已加载，实际票数: [{match_tickets.group(1)}]"
)
if not has_data:
# 同样将空数据提示隔离在当前 tab 中
# 空数据提示同样限制在当前 tab 内
if ws_frame.locator(
"#tab-1 .no-records-found", has_text="没有找到匹配的记录"
).first.is_visible():
print(
" ⚠️ 确认为空数据环境：系统提示【没有找到匹配的记录】，正在执行复位净化..."
" ⚠️ 确认为空数据，正在关闭当前标签页..."
)
page.locator(".tags-view-item", has_text="进站交接单查询").locator(
".el-icon-close"
@@ -207,20 +201,20 @@ def yunda_expected_download(page):
row_count = main_rows.count()
print(f">> 当前视窗共捕获到活跃交接单记录: {row_count} 条")
# 5. 循环双击穿透提交
# 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()
# ====================================================================
# 🛡️ 状态机拦截卫语句：识别绑定状态，过滤已绑定交接单
# 跳过已绑定的交接单
# ====================================================================
bind_status = current_row.locator("td").nth(2).inner_text().strip()
print(f" -> 锁定提取单号: {raw_no} [绑定状态: {bind_status}]")
print(f" -> 交接单号: {raw_no} [绑定状态: {bind_status}]")
if bind_status == "已绑定":
print(" ⏭️ 状态拦截：该交接单处于【已绑定】状态，安全跳过。")
print(" ⏭️ 该交接单已绑定，跳过。")
continue
# ====================================================================
@@ -248,7 +242,7 @@ def yunda_expected_download(page):
try:
confirm_link.wait_for(state="visible", timeout=6000)
if export_frame.get_by_text("导出任务建立成功").is_visible():
print(" ✅ 判定通过：成功捕获到【导出任务建立成功】特征！")
print(" ✅ 已确认导出任务建立成功。")
confirm_link.click()
task_success = True
break
@@ -256,7 +250,7 @@ def yunda_expected_download(page):
"请选择格式相应的导出字段"
).is_visible():
print(
" ⚠️ 警告：检测到【未选择字段】错误，重新触发补点全选..."
" ⚠️ 检测到未选择字段，重新点击全选..."
)
confirm_link.click()
page.wait_for_timeout(500)
@@ -270,7 +264,7 @@ def yunda_expected_download(page):
if not task_success:
raise RuntimeError(
"致命异常：连续 5 次尝试均无法成功建立应到数据离线任务。"
"连续 5 次尝试均未能建立应到数据离线任务。"
)
ws_frame.locator(".layui-layer-close1").click()
@@ -279,16 +273,16 @@ def yunda_expected_download(page):
page.wait_for_timeout(800)
export_times.append(datetime.now())
print(">> 📤 任务提交流闭环，正在执行【进站交接单查询】工作台销毁...")
print(">> 任务提交完成，正在关闭【进站交接单查询】标签页...")
page.locator(".tags-view-item", has_text="进站交接单查询").locator(
".el-icon-close"
).click()
page.wait_for_timeout(500)
# 防线：如果全部记录都被跳过了，export_times 为空，直接结束
# 若所有记录都被跳过，export_times 为空，直接结束
if not export_times:
print(
">> ⚠️ 本次查询未产生任何有效的离线下载任务（全部空单或已被跳过），中止后端收割流。"
">> ⚠️ 本次未产生任何离线下载任务（无数据或已全部跳过），结束。"
)
return
@@ -302,6 +296,7 @@ def yunda_expected_download(page):
except Exception as e:
print(f"\n❌ 任务执行过程中发生异常: {e}")
return False
def yunda_actual_download(page):
@@ -319,17 +314,17 @@ def yunda_actual_download(page):
page.locator(".el-menu-item", has_text="首页").wait_for(
state="visible", timeout=15000
)
print("✅ 韵达工作台首页成功加载")
print("✅ 韵达工作台首页已加载")
yunda_smart_menu_click(page, ["报表管理", "扫描记录查询"])
print(">> 正在跨域动态追踪【扫描记录查询】业务窗体...")
print(">> 正在定位【扫描记录查询】iframe...")
ws_frame = page.frame_locator("section iframe")
ws_frame.locator(
".no-records-found", has_text="没有找到匹配的记录"
).first.wait_for(state="visible", timeout=15000)
print("✅ 扫描记录查询工作区初始化完毕")
print("✅ 扫描记录查询页面已初始化")
query_days = 1
try:
@@ -343,7 +338,7 @@ def yunda_actual_download(page):
today = datetime.now()
start_date = today - timedelta(days=(query_days - 1))
print(f">> 正在精准对焦实到时间区间: [近 {query_days} 天]")
print(f">> 设置实到查询时间范围: [近 {query_days} 天]")
print(" >> 正在设定起始时间...")
ws_frame.locator("#startDate").click()
@@ -367,7 +362,7 @@ def yunda_actual_download(page):
ws_frame.locator("#scanRecordTyp").select_option(value="03")
page.wait_for_timeout(500)
print(">> 正在触发现场查询数据流...")
print(">> 正在执行查询...")
ws_frame.locator('input[type="button"][value="查询"]').click()
page.wait_for_timeout(800)
@@ -376,7 +371,7 @@ def yunda_actual_download(page):
".fixed-table-loading", has_text="正在努力地加载数据中"
).first
if loading_mask.is_visible():
print(" ⏳ 检测到专属数据加载罩，正在等待后端重载返回...")
print(" ⏳ 检测到数据加载遮罩，等待加载完成...")
loading_mask.wait_for(state="hidden", timeout=30000)
page.wait_for_timeout(500)
@@ -390,7 +385,7 @@ def yunda_actual_download(page):
if match_total and int(match_total.group(1)) > 0:
has_records = True
print(
f" ✅ 深度断言：实到数据渲染完毕，总记录数: [{match_total.group(1)}] 条。"
f" ✅ 实到数据已加载，总记录数: [{match_total.group(1)}] 条。"
)
if not has_records:
@@ -398,21 +393,21 @@ def yunda_actual_download(page):
".no-records-found", has_text="没有找到匹配的记录"
).first.is_visible():
print(
" ⚠️ 当前查询范围内确认为空数据：系统提示【没有找到匹配的记录】，终止并关闭环境。"
" ⚠️ 当前查询范围内为空数据，终止并关闭标签页。"
)
page.locator(".tags-view-item", has_text="扫描记录查询").locator(
".el-icon-close"
).click()
return
else:
print(" ⚠️ 发生渲染异常，未找到数据也未找到空记录提示。安全收尾...")
print(" ⚠️ 未找到数据，也未出现空数据提示，结束。")
page.locator(".tags-view-item", has_text="扫描记录查询").locator(
".el-icon-close"
).click()
return
# 5. 执行导出流
print(">> 正在发起【导出】申请指令...")
print(">> 正在发起导出...")
ws_frame.locator('input[type="button"][id="export"]').click()
ws_frame.locator(".layui-layer-title", has_text="数据导出").wait_for(
@@ -431,12 +426,12 @@ def yunda_actual_download(page):
try:
confirm_link.wait_for(state="visible", timeout=6000)
if export_frame.get_by_text("导出任务建立成功").is_visible():
print(" ✅ 判定通过：成功捕获到【导出任务建立成功】特征！")
print(" ✅ 已确认导出任务建立成功。")
confirm_link.click()
task_success = True
break
elif export_frame.get_by_text("请选择格式相应的导出字段").is_visible():
print(" ⚠️ 警告：检测到字段未全选，执行强补点击...")
print(" ⚠️ 检测到字段未全选，重新点击全选...")
confirm_link.click()
page.wait_for_timeout(500)
export_frame.locator(".allRight").click()
@@ -449,14 +444,14 @@ def yunda_actual_download(page):
if not task_success:
raise RuntimeError(
"致命异常：连续 5 次尝试均无法成功建立实到数据离线任务。"
"连续 5 次尝试均未能建立实到数据离线任务。"
)
ws_frame.locator(".layui-layer-close1").click()
page.wait_for_timeout(500)
export_times.append(datetime.now())
print(">> 📤 任务提交流闭环，正在执行【扫描记录查询】工作台销毁...")
print(">> 任务提交完成，正在关闭【扫描记录查询】标签页...")
page.locator(".tags-view-item", has_text="扫描记录查询").locator(
".el-icon-close"
).click()
@@ -464,10 +459,10 @@ def yunda_actual_download(page):
# 防线：双重保护
if not export_times:
print(">> ⚠️ 本次查询未产生有效的离线下载任务，中止后端收割流。")
print(">> ⚠️ 本次未产生任何离线下载任务，结束。")
return
# 6. 收割下载
# 6. 轮询并下载
_yunda_poll_and_download_tasks(
page,
export_times,
@@ -478,13 +473,14 @@ def yunda_actual_download(page):
except Exception as e:
print(f"\n❌ 任务执行过程中发生异常: {e}")
return False
def _yunda_poll_and_download_tasks(
page, export_times, target_task_title, download_dir, final_filename
):
"""韵达专属离线任务轮询下载引擎"""
print("\n>> 正在前往【导出服务】中心...")
"""韵达离线任务的轮询与下载"""
print("\n>> 正在前往【导出服务】界面...")
yunda_smart_menu_click(page, ["基础数据", "导出服务"])
export_ws_frame = page.frame_locator("section iframe")
@@ -494,7 +490,7 @@ def _yunda_poll_and_download_tasks(
)
page.wait_for_timeout(1000)
print(">> 离线文件队列已对接，启动【1分钟高频精确校对+缺单局部重载刷新】断言...")
print(">> 开始轮询离线文件队列（定时刷新直到任务齐全）...")
total_expected = len(export_times)
while True:
@@ -532,17 +528,17 @@ def _yunda_poll_and_download_tasks(
total_found = len(ready_indices) + len(processing_indices)
print(
f" 📊 状态研判：自建期望数 [{total_expected}]，实际入表 [{total_found}] (完成 [{len(ready_indices)}]，生成中 [{len(processing_indices)}])"
f" 📊 状态统计：期望 [{total_expected}]，已入表 [{total_found}] (完成 [{len(ready_indices)}]，生成中 [{len(processing_indices)}])"
)
if total_found < total_expected or len(processing_indices) > 0:
print(" ⏳ 队列未齐，执行【点击查询按钮】触发局部无痕刷新...")
print(" ⏳ 队列未齐全，点击查询刷新...")
export_ws_frame.locator(
"#ydkyimport_basic_export_searchData1_ky_export_common"
).click()
page.wait_for_timeout(3000)
else:
print(">> ✅ 所有目标离线任务全量就绪！开始依序接入文件流...")
print(">> 所有目标离线任务已就绪，开始依次下载...")
break
downloaded_files = []
@@ -554,7 +550,7 @@ def _yunda_poll_and_download_tasks(
time_flag = (
target_row.locator("td[field='createdTime']").inner_text().strip()
)
print(f" 🎯 触发下载 -> 离线任务时间节点: [{time_flag}] ...")
print(f" 开始下载任务 [{time_flag}] ...")
with page.expect_download() as download_info:
target_row.locator("td[field='extreFile'] a").get_by_text(
@@ -569,22 +565,22 @@ def _yunda_poll_and_download_tasks(
download.save_as(save_path)
downloaded_files.append(save_path)
print(f" ⬇️ 文件已用安全序列号落盘: downloads/{custom_filename}")
print(f" 已下载: downloads/{custom_filename}")
page.wait_for_timeout(500)
except Exception as e:
print(f" ❌ 文件流接收失败: {e}")
print(f" ❌ 下载失败: {e}")
print(">> 📥 【导出服务】数据提取链闭环，正在执行当前 Tab 窗口销毁...")
print(">> 【导出服务】下载完成，正在关闭标签页...")
try:
page.locator(".tags-view-item", has_text="导出服务").locator(
".el-icon-close"
).click()
print(" ✅ 【导出服务】工作区已安全关闭。")
print(" ✅ 【导出服务】标签页已关闭。")
except Exception:
pass
if downloaded_files:
print("\n>> 🧪 正在启动离线数据清洗与高能扁平合并流...")
print("\n>> 正在合并下载的数据...")
all_dfs = []
for file_path in downloaded_files:
try:
@@ -599,10 +595,10 @@ def _yunda_poll_and_download_tasks(
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" 合并完成。")
print(f" 📁 输出路径: {final_output}")
print(f"====================================================")
for file_path in downloaded_files:
os.remove(file_path)
print(" ✅ 临时缓存阵列已无缝净化。")
print(" 临时文件已清理。")

106
site_zto.py
View File

@@ -10,7 +10,7 @@ from paths import DOWNLOAD_DIR, CONFIG_PATH
def _wait_and_get_frame(page, text_indicator, timeout_ms=20000):
"""动态雷达探测器：全域扫描所有视窗"""
"""在主页面与所有 iframe 中查找包含指定文本的窗口"""
start_time = datetime.now()
while (datetime.now() - start_time).total_seconds() * 1000 < timeout_ms:
try:
@@ -27,11 +27,11 @@ def _wait_and_get_frame(page, text_indicator, timeout_ms=20000):
pass
page.wait_for_timeout(300)
raise TimeoutError(f"爆栈超时：全域未死守到包含 [{text_indicator}] 的弹窗视窗。")
raise TimeoutError(f"超时：未找到包含 [{text_indicator}] 的窗口。")
def zto_smart_menu_click(page, menu_path):
"""中通智能菜单导航器"""
"""中通菜单导航"""
print(f">> 正在导航: {' -> '.join(menu_path)}")
for i in range(len(menu_path)):
current_menu = menu_path[i]
@@ -58,12 +58,12 @@ def zto_expected_download(page):
export_times = []
try:
# 1. 智能导航
# 1. 菜单导航
zto_smart_menu_click(page, ["运营管理", "进站管理", "进站交接单查询"])
ewb_frame = page.frame_locator('iframe[src*="inEwbsListNoSearch"]')
print(">> 正在探针检测表格统计面板 (#inEwbCount)...")
print(">> 正在检测表格统计面板 (#inEwbCount)...")
ewb_frame.locator("#inEwbCount").wait_for(state="attached", timeout=15000)
# 读取 YAML 配置设定天数
@@ -105,7 +105,7 @@ def zto_expected_download(page):
page.wait_for_timeout(500)
# 3. 触发查询与深度状态机判定
# 3. 触发查询并判断数据状态
print(">> 正在点击【查询】按钮并等待数据响应...")
old_count = ewb_frame.locator("#inEwbCount").inner_text().strip()
ewb_frame.locator("#searchbtn").click()
@@ -129,7 +129,7 @@ def zto_expected_download(page):
wait_cycles += 1
ticket_count_str = ewb_frame.locator("#inEwbCount").inner_text().strip()
print(f" ✅ 数据面板就绪！当前进站实际票数为: [{ticket_count_str}]")
print(f" ✅ 数据已加载，进站实际票数: [{ticket_count_str}]")
# 数据分流
if (
@@ -137,13 +137,13 @@ def zto_expected_download(page):
or not ticket_count_str.isdigit()
or int(ticket_count_str) == 0
):
print(" >> 票数为 0，正在验证是否为确切的空数据区...")
print(" >> 票数为 0，正在确认是否为空数据...")
empty_flag = ewb_frame.locator("#datagrid1").get_by_text(
"没有搜索到符合条件的数据记录"
)
if empty_flag.is_visible():
print(" ⚠️ 确认为空数据区，正在关闭当前标签页清理环境...")
# 即使为空数据，我们也清理关闭当前已经打开的 Tab
print(" ⚠️ 确认为空数据，正在关闭当前标签页...")
# 即使无数据也关闭已打开的标签页
try:
page.locator(".mini-tab", has_text="进站交接单查询").locator(
".mini-tab-close"
@@ -161,7 +161,7 @@ def zto_expected_download(page):
pass
return
else:
print(" >> 票数验证通过，等待主数据表格渲染实体行...")
print(" >> 票数校验通过，等待主表格渲染数据行...")
ewb_frame.locator(
"#datagrid1 .mini-grid-rows-view .mini-grid-row"
).first.wait_for(state="visible", timeout=15000)
@@ -180,7 +180,7 @@ def zto_expected_download(page):
raw_text = row.locator("td").nth(3).inner_text()
match = re.search(r"\d{18}", raw_text)
handover_no = match.group(0) if match else raw_text.strip()
print(f" -> 锁定提取单号：{handover_no}")
print(f" -> 当前交接单号：{handover_no}")
row.dblclick()
@@ -203,12 +203,12 @@ def zto_expected_download(page):
page.wait_for_timeout(300)
export_frame.locator(".mini-button-text", has_text="确定").click()
print(" >> [雷达扫描] 正在跨域动态追踪【温馨提示】所在的隐身视窗...")
print(" >> 正在查找【温馨提示】弹窗...")
ctx_alert = _wait_and_get_frame(page, "温馨提示")
ctx_alert.locator(
".mini-messagebox-buttons .mini-button-text", has_text="确定"
).click()
print(" ✅ 【温馨提示】已成功通过父级过滤并确认！")
print(" ✅ 已确认【温馨提示】弹窗。")
print(" >> 正在等待服务器建立后台离线任务...")
try:
@@ -218,9 +218,9 @@ def zto_expected_download(page):
ctx_tips.locator(".mini-tips-success").wait_for(
state="hidden", timeout=15000
)
print(" ✅ 成功提示框已平滑隐藏。")
print(" ✅ 成功提示框已消失。")
except Exception:
print(" ✅ 成功提示框及其容器已自动销毁。")
print(" ✅ 成功提示框已关闭。")
export_times.append(datetime.now())
@@ -229,19 +229,19 @@ def zto_expected_download(page):
page.wait_for_timeout(1000)
# ====================================================================
# 🛡️ 优化优化：完成交接单主表的所有穿透提交后，立刻销毁本 Tab 净化内存
# 完成所有交接单导出提交后，关闭当前标签页
# ====================================================================
print(">> 📥 【进站交接单查询】任务已就绪，正在精准执行 Tab 窗口销毁复位...")
print(">> 【进站交接单查询】已完成，正在关闭标签页...")
try:
page.locator(".mini-tab", has_text="进站交接单查询").locator(
".mini-tab-close"
).click()
page.wait_for_timeout(500)
print(" ✅ 【进站交接单查询】Tab 已成功关闭。")
print(" ✅ 【进站交接单查询】标签页已关闭。")
except Exception as e:
print(f" ⚠️ 尝试关闭【进站交接单查询】Tab 时遇到轻微阻碍: {e}")
print(f" ⚠️ 关闭【进站交接单查询】标签页时出错: {e}")
# 核心逻辑交由统一的轮询下载引擎处理
# 交由统一的轮询下载流程处理
_zto_poll_and_download_tasks(
page,
export_times,
@@ -252,6 +252,7 @@ def zto_expected_download(page):
except Exception as e:
print(f"\n❌ 任务执行过程中发生异常: {e}")
return False
def zto_actual_download(page):
@@ -266,12 +267,12 @@ def zto_actual_download(page):
export_times = []
try:
# 1. 智能导航
# 1. 菜单导航
zto_smart_menu_click(page, ["运营管理", "扫描操作与监控", "到件扫描监控"])
arr_frame = page.frame_locator('iframe[src*="ArriveScan"]')
print(">> 正在探针检测主页面 (#daterange)...")
print(">> 正在检测主页面 (#daterange)...")
arr_frame.locator("#daterange").wait_for(state="attached", timeout=15000)
# 2. 读取 YAML 并设定时间范围
@@ -322,7 +323,7 @@ def zto_actual_download(page):
).locator(".mini-tree-checkbox").click()
page.wait_for_timeout(300)
# 4. 触发查询与深度状态机判定
# 4. 触发查询并判断数据状态
print(">> 正在点击【查询】按钮并等待数据响应...")
arr_frame.locator("#searchbtn").click()
@@ -340,7 +341,7 @@ def zto_actual_download(page):
)
if empty_flag.is_visible():
print(
" ⚠️ 当前查询范围内【没有搜索到符合条件的数据记录】，终止并关闭环境。"
" ⚠️ 当前查询范围内没有数据，终止并关闭标签页。"
)
try:
page.locator(".mini-tab", has_text="到件扫描监控").locator(
@@ -351,7 +352,7 @@ def zto_actual_download(page):
return
arr_frame.locator("#page1").wait_for(state="visible", timeout=15000)
print(" ✅ 数据渲染完毕！(底部分页统计控件已就绪)")
print(" ✅ 数据已加载。(底部分页控件已就绪)")
# 5. 执行导出流程
arr_frame.locator("#exportExcel").click()
@@ -365,12 +366,12 @@ def zto_actual_download(page):
page.wait_for_timeout(300)
export_frame.locator(".mini-button-text", has_text="确定").click()
print(" >> [雷达扫描] 正在跨域动态追踪【温馨提示】所在的隐身视窗...")
print(" >> 正在查找【温馨提示】弹窗...")
ctx_alert = _wait_and_get_frame(page, "温馨提示")
ctx_alert.locator(
".mini-messagebox-buttons .mini-button-text", has_text="确定"
).click()
print(" ✅ 【温馨提示】已成功通过父级过滤并确认！")
print(" ✅ 已确认【温馨提示】弹窗。")
print(" >> 正在等待服务器建立后台离线任务...")
try:
@@ -380,26 +381,26 @@ def zto_actual_download(page):
ctx_tips.locator(".mini-tips-success").wait_for(
state="hidden", timeout=15000
)
print(" ✅ 成功提示框已平滑隐藏。")
print(" ✅ 成功提示框已消失。")
except Exception:
print(" ✅ 成功提示框及其容器已自动销毁。")
print(" ✅ 成功提示框已关闭。")
export_times.append(datetime.now())
# ====================================================================
# 🛡️ 优化优化：完成实到主表的数据导出提交后，立刻销毁本 Tab 恢复空态
# 完成实到数据导出提交后，关闭当前标签页
# ====================================================================
print(">> 📥 【到件扫描监控】任务已就绪，正在精准执行 Tab 窗口销毁复位...")
print(">> 【到件扫描监控】已完成，正在关闭标签页...")
try:
page.locator(".mini-tab", has_text="到件扫描监控").locator(
".mini-tab-close"
).click()
page.wait_for_timeout(500)
print(" ✅ 【到件扫描监控】Tab 已成功关闭。")
print(" ✅ 【到件扫描监控】标签页已关闭。")
except Exception as e:
print(f" ⚠️ 尝试关闭【到件扫描监控】Tab 时遇到轻微阻碍: {e}")
print(f" ⚠️ 关闭【到件扫描监控】标签页时出错: {e}")
# 核心逻辑交由统一的轮询下载引擎处理
# 交由统一的轮询下载流程处理
_zto_poll_and_download_tasks(
page,
export_times,
@@ -410,12 +411,13 @@ def zto_actual_download(page):
except Exception as e:
print(f"\n❌ 任务执行过程中发生异常: {e}")
return False
def _zto_poll_and_download_tasks(
page, export_times, target_task_title, download_dir, final_filename
):
"""进站中通任务专属轮询、验证及即选即隐下载引擎"""
"""中通离线任务的轮询与下载"""
print("\n>> 正在前往【导出任务管理】界面...")
zto_smart_menu_click(page, ["系统配置", "导出任务管理"])
@@ -425,7 +427,7 @@ def _zto_poll_and_download_tasks(
)
page.wait_for_timeout(1000)
print(">> 列表中已渲染，开始匹配并检查后端处理状态...")
print(">> 列表已加载，开始匹配并检查任务状态...")
total_expected = len(export_times)
while True:
@@ -464,23 +466,23 @@ def _zto_poll_and_download_tasks(
total_found = len(ready_timestamps) + len(processing_timestamps)
print(
f" 📊 状态盘点：期望任务数 [{total_expected}]，实际入表 [{total_found}] (就绪 [{len(ready_timestamps)}]，处理中 [{len(processing_timestamps)}])"
f" 📊 状态统计：期望 [{total_expected}]，已入表 [{total_found}] (就绪 [{len(ready_timestamps)}]，处理中 [{len(processing_timestamps)}])"
)
if total_found < total_expected or len(processing_timestamps) > 0:
print(" ⏳ 发现任务缺失或正在生成，执行【点击查询按钮】触发局部重载...")
print(" ⏳ 任务尚未齐全或仍在生成，点击查询刷新...")
try:
taskdone_frame.locator(
".mini-button-text", has_text="查询"
).first.click()
except Exception as e:
print(f" ⚠️ 局部刷新按钮失效，使用菜单后备刷新: {e}")
print(f" ⚠️ 查询按钮不可用，改用菜单刷新: {e}")
page.locator("li.leaf span.menu-name", has_text="导出任务管理").click()
page.wait_for_timeout(3000)
else:
target_task_timestamps = list(ready_timestamps)
print(">> ✅ 目标任务已全部生成，解除锁定，开始接收文件流...")
print(">> 所有目标任务已生成，开始下载...")
break
# 8. 下载逻辑
@@ -498,7 +500,7 @@ def _zto_poll_and_download_tasks(
)
.first
)
print(f" 🎯 触发下载 -> 任务 [{time_str}] ...")
print(f" 开始下载任务 [{time_str}] ...")
checkbox = target_row.locator(".mini-grid-checkbox")
if checkbox.is_visible():
@@ -513,32 +515,32 @@ def _zto_poll_and_download_tasks(
save_path = os.path.join(download_dir, download.suggested_filename)
download.save_as(save_path)
downloaded_files.append(save_path)
print(f" ⬇️ 文件已落盘: downloads/{download.suggested_filename}")
print(f" 已下载: downloads/{download.suggested_filename}")
if checkbox.is_visible():
checkbox.click()
print(" -> 已取消勾选，清理现场进入下一条")
print(" -> 已取消勾选，继续下一条")
page.wait_for_timeout(500)
except Exception as e:
print(f" ❌ 下载任务 [{time_str}] 失败: {e}")
# ====================================================================
# 🛡️ 优化优化：所有目标任务文件全部下载完毕后，立刻销毁“导出任务管理” Tab
# 所有目标文件下载完成后，关闭“导出任务管理”标签页
# ====================================================================
print(">> 📥 【导出任务管理】所有数据拉取流闭环，正在执行 Tab 窗口销毁复位...")
print(">> 【导出任务管理】下载完成，正在关闭标签页...")
try:
page.locator(".mini-tab", has_text="导出任务管理").locator(
".mini-tab-close"
).click()
page.wait_for_timeout(500)
print(" ✅ 【导出任务管理】Tab 已安全关闭。页面归档空态。")
print(" ✅ 【导出任务管理】标签页已关闭。")
except Exception as e:
print(f" ⚠️ 尝试收尾关闭【导出任务管理】Tab 时有波动: {e}")
print(f" ⚠️ 关闭【导出任务管理】标签页时出错: {e}")
# 9. 合并数据
if downloaded_files:
print("\n>> 🧪 正在开始执行扁平数据高能合并与清洗流程...")
print("\n>> 正在合并下载的数据...")
all_data_frames = []
for file_path in downloaded_files:
try:
@@ -553,10 +555,10 @@ def _zto_poll_and_download_tasks(
final_output_path = os.path.join(download_dir, final_filename)
combined_df.to_excel(final_output_path, index=False)
print(f"====================================================")
print(f" 🎉 恭喜！数据流闭环完成！")
print(f" 📁 最终输出成果归档至: {final_output_path}")
print(f" 合并完成。")
print(f" 📁 输出路径: {final_output_path}")
print(f"====================================================")
for file_path in downloaded_files:
os.remove(file_path)
print(" ✅ 临时缓存数据清理完毕。")
print(" 临时文件已清理。")