5b7a024fa8
- site_shunxin: replace the hardcoded 1天 radio with config-driven shunxin.query_days, set the date range via the ant-picker, and add a two-pass query/retry to clear stale "暂无数据" DOM state; return True on completion - config.example.yaml: add the shunxin section (query_days) so the example matches what the code reads Co-Authored-By: Claude <noreply@anthropic.com>
559 lines
23 KiB
Python
559 lines
23 KiB
Python
# site_shunxin.py
|
||
|
||
import os
|
||
import re
|
||
import yaml
|
||
from datetime import datetime, timedelta
|
||
import pandas as pd
|
||
|
||
from paths import DOWNLOAD_DIR, CONFIG_PATH
|
||
|
||
|
||
def _close_tab(page, tab_name):
|
||
"""关闭指定名称的标签页（Ant Design Tabs）。
|
||
|
||
通过标签文字定位标签容器，点击其右侧的关闭(×)按钮。
|
||
- tab_name 采用包含匹配，对“ 运单列表”这类带空格的标签同样有效。
|
||
- 关闭失败不会影响主流程，仅打印提示信息。
|
||
"""
|
||
try:
|
||
tab = page.locator(".ant-tabs-tab").filter(
|
||
has=page.get_by_role("tab", name=tab_name)
|
||
)
|
||
if tab.count() == 0:
|
||
print(
|
||
f" ℹ️ 未找到标签页【{tab_name.strip()}】(可能尚未打开或已关闭)，跳过。"
|
||
)
|
||
return
|
||
tab.first.locator(".ant-tabs-tab-remove").click()
|
||
print(f" 🗙 已关闭标签页【{tab_name.strip()}】")
|
||
page.wait_for_timeout(300)
|
||
except Exception as e:
|
||
print(f" ⚠️ 关闭标签页【{tab_name.strip()}】时出错: {e}")
|
||
|
||
|
||
def shunxin_expected_download(page):
|
||
"""顺心：应到货物数据下载"""
|
||
print("\n▶ 开始执行【顺心 - 应到货物数据下载】任务...")
|
||
|
||
# 初始化并创建下载目录
|
||
download_dir = DOWNLOAD_DIR
|
||
if not os.path.exists(download_dir):
|
||
os.makedirs(download_dir)
|
||
print(f">> 已创建下载目录: {download_dir}")
|
||
|
||
export_times = []
|
||
target_task_timestamps = []
|
||
|
||
try:
|
||
# 1. 导航与页面加载判断
|
||
print(">> 正在进入【车辆点到】界面...")
|
||
page.locator("span.ant-pro-menu-item-title:has-text('派件管理')").click()
|
||
page.locator("div.ant-pro-menu-item:has-text('车辆点到')").click()
|
||
|
||
page.get_by_role("button", name="点到").wait_for(state="visible")
|
||
page.get_by_role("button", name="打印交接单").wait_for(state="visible")
|
||
page.get_by_role("button", name="强卸").wait_for(state="visible")
|
||
print("✅ 车辆点到界面加载完毕")
|
||
|
||
# 2. 从配置读取时间偏移量并动态设置日期选择器
|
||
query_days = 1
|
||
try:
|
||
if os.path.exists(CONFIG_PATH):
|
||
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
|
||
config = yaml.safe_load(f) or {}
|
||
query_days = int(config.get("shunxin", {}).get("query_days", 1))
|
||
except Exception:
|
||
pass
|
||
|
||
today = datetime.now()
|
||
start_date = today - timedelta(days=(query_days - 1))
|
||
|
||
today_str = today.strftime("%Y-%m-%d")
|
||
start_date_str = start_date.strftime("%Y-%m-%d")
|
||
|
||
print(f">> 正在设置查询时间范围: [{start_date_str}] 至 [{today_str}]...")
|
||
|
||
# 分两步精准呼出和点击时间控件
|
||
print(" >> 设置起始时间...")
|
||
page.get_by_placeholder("开始时间").click()
|
||
page.wait_for_timeout(500)
|
||
page.locator(
|
||
f".ant-picker-dropdown:visible td[title='{start_date_str}']"
|
||
).first.click()
|
||
page.wait_for_timeout(300)
|
||
|
||
print(" >> 设置截止时间...")
|
||
page.get_by_placeholder("结束时间").click()
|
||
page.wait_for_timeout(500)
|
||
page.locator(
|
||
f".ant-picker-dropdown:visible td[title='{today_str}']"
|
||
).first.click()
|
||
page.wait_for_timeout(300)
|
||
|
||
# 确认日期
|
||
page.locator(".ant-picker-dropdown:visible button", has_text="确 定").click()
|
||
page.wait_for_timeout(500)
|
||
|
||
print(">> 正在展开【状态】下拉菜单...")
|
||
page.locator(
|
||
".ant-select-selection-item", has_text=re.compile(r"已发|已到")
|
||
).click()
|
||
print(">> 正在选择状态为【已到】...")
|
||
page.locator(".ant-select-item-option", has_text="已到").click()
|
||
page.wait_for_timeout(500)
|
||
|
||
# ====================================================================
|
||
# 🛡️ 双重校验兜底机制：破除顺心表格“暂无数据”的旧状态遗留陷阱
|
||
# ====================================================================
|
||
print(">> 正在发起查询与数据状态研判...")
|
||
has_data = False
|
||
waybill_btns = None
|
||
|
||
for attempt in range(2):
|
||
print(f" -> 第 {attempt + 1} 次点击【查询】按钮...")
|
||
page.get_by_role("button", name="search 查询").click()
|
||
|
||
# 尝试在极短时间内捕获加载小菊花
|
||
loading_spinner = page.locator(".ant-spin-dot-spin").first
|
||
try:
|
||
loading_spinner.wait_for(state="visible", timeout=800)
|
||
print(" ⏳ 捕捉到加载动画，等待数据渲染完成...")
|
||
loading_spinner.wait_for(state="hidden", timeout=15000)
|
||
except Exception:
|
||
print(" ⚡ 加载动画闪过过快或未出现，强制安全缓冲1秒...")
|
||
page.wait_for_timeout(1000)
|
||
|
||
# 解析查询结果
|
||
empty_desc = page.locator(
|
||
".ant-empty-description", has_text="暂无数据"
|
||
).first
|
||
waybill_btns = page.get_by_role("button", name="运单列表")
|
||
|
||
if waybill_btns.count() > 0:
|
||
has_data = True
|
||
print(" ✅ 数据已成功加载。")
|
||
break
|
||
elif empty_desc.is_visible():
|
||
print(" ⚠️ 当前表格显示【暂无数据】。")
|
||
if attempt == 0:
|
||
print(" -> 疑似前端 DOM 旧状态未刷新，触发二次兜底查询...")
|
||
page.wait_for_timeout(500)
|
||
else:
|
||
print(" -> 已二次确认为空数据环境。")
|
||
else:
|
||
# 没出现暂无数据，也没出现按钮，稳妥判定缓冲完毕
|
||
pass
|
||
|
||
if not has_data:
|
||
print(">> ⚠️ 本次查询区间内没有数据记录，提前结束流程。")
|
||
_close_tab(page, "车辆点到")
|
||
return True
|
||
|
||
# ====================================================================
|
||
|
||
count = waybill_btns.count()
|
||
print(f">> 共发现 {count} 个班次需要导出。")
|
||
|
||
for i in range(count):
|
||
print(f" ⏳ 正在处理第 {i+1}/{count} 个班次...")
|
||
|
||
waybill_btns.nth(i).click()
|
||
page.locator("label[title='运单查询']").wait_for(state="visible")
|
||
|
||
# 4. 执行导出流程
|
||
page.get_by_role("button", name="export 导出").click()
|
||
|
||
page.locator(
|
||
"span.ant-transfer-list-header-title:has-text('待选导出列')"
|
||
).wait_for(state="visible")
|
||
page.locator(".ant-transfer-list").first.locator(
|
||
".ant-transfer-list-header label"
|
||
).click()
|
||
|
||
page.get_by_label("导出").get_by_role("button", name="right").click()
|
||
page.get_by_role("button", name="export 导出数据").click()
|
||
|
||
export_times.append(datetime.now())
|
||
|
||
page.locator("text=任务添加成功!").wait_for(state="visible")
|
||
page.get_by_role("button", name="知道了").click()
|
||
|
||
page.get_by_role("tab", name="车辆点到").click()
|
||
page.wait_for_timeout(500)
|
||
|
||
print("✅ 所有班次的导出任务已成功提交！")
|
||
|
||
# 5. 关闭标签页
|
||
_close_tab(page, "运单列表")
|
||
_close_tab(page, "车辆点到")
|
||
|
||
# 6. 前往数据导出页面去下载
|
||
print(">> 正在前往【数据导出】界面...")
|
||
page.locator("a[href='/dataExport']").click()
|
||
|
||
page.get_by_role("columnheader", name="任务标题").wait_for(state="visible")
|
||
page.wait_for_timeout(2000)
|
||
|
||
page.get_by_role("button", name="search 查询").click()
|
||
page.wait_for_timeout(2000)
|
||
|
||
# 7. 轮询任务状态
|
||
print(">> 列表已加载，开始匹配并检查任务状态...")
|
||
while True:
|
||
rows = page.locator(".ant-table-tbody > tr.ant-table-row")
|
||
row_count = rows.count()
|
||
pending_tasks = 0
|
||
current_ready_timestamps = []
|
||
|
||
for i in range(row_count):
|
||
tds = rows.nth(i).locator("td")
|
||
if tds.count() < 9:
|
||
continue
|
||
|
||
submit_time_str = tds.nth(2).inner_text().strip()
|
||
title_str = tds.nth(5).inner_text().strip()
|
||
status_str = tds.nth(6).inner_text().strip()
|
||
|
||
if title_str == "发车管理运单列表":
|
||
try:
|
||
row_time = datetime.strptime(
|
||
submit_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_str != "执行完成":
|
||
pending_tasks += 1
|
||
if submit_time_str not in current_ready_timestamps:
|
||
print(
|
||
f" ⏳ 任务 [{submit_time_str}] 状态为【{status_str}】，数据生成中..."
|
||
)
|
||
else:
|
||
if submit_time_str not in current_ready_timestamps:
|
||
current_ready_timestamps.append(submit_time_str)
|
||
except Exception as e:
|
||
print(f" ⚠️ 解析时间时出错: {e}")
|
||
|
||
if pending_tasks > 0:
|
||
print(
|
||
f">> 共有 {pending_tasks} 个匹配任务还在处理中，等待 5 秒后刷新..."
|
||
)
|
||
page.wait_for_timeout(5000)
|
||
page.get_by_role("button", name="search 查询").click()
|
||
page.wait_for_timeout(2000)
|
||
else:
|
||
target_task_timestamps = current_ready_timestamps
|
||
if len(target_task_timestamps) > 0:
|
||
print(">> 所有目标任务已就绪，开始下载...")
|
||
break
|
||
|
||
# 8. 下载逻辑
|
||
downloaded_files = []
|
||
for time_str in target_task_timestamps:
|
||
try:
|
||
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}] ...")
|
||
|
||
with page.expect_download() as download_info:
|
||
target_row.locator("td").nth(8).locator(
|
||
"button", has_text=re.compile(r"下\s*载")
|
||
).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}")
|
||
except Exception as e:
|
||
print(f" ❌ 下载任务 [{time_str}] 失败: {e}")
|
||
|
||
# 9. 合并数据
|
||
if downloaded_files:
|
||
print("\n>> 正在合并下载的数据...")
|
||
all_data_frames = []
|
||
for file_path in downloaded_files:
|
||
try:
|
||
df = pd.read_excel(file_path, dtype=str)
|
||
if not df.empty:
|
||
all_data_frames.append(df)
|
||
except Exception as e:
|
||
pass
|
||
|
||
if all_data_frames:
|
||
combined_df = pd.concat(all_data_frames, ignore_index=True)
|
||
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"====================================================")
|
||
|
||
for file_path in downloaded_files:
|
||
os.remove(file_path)
|
||
print("✅ 临时文件已清理。")
|
||
|
||
# 10. 关闭标签页
|
||
_close_tab(page, "数据导出")
|
||
print("\n【顺心 - 应到货物数据下载】流程结束。")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"\n❌ 任务执行过程中发生异常: {e}")
|
||
return False
|
||
|
||
|
||
def shunxin_actual_download(page):
|
||
"""顺心：实到货物数据下载"""
|
||
print("\n▶ 开始执行【顺心 - 实到货物数据下载】任务...")
|
||
|
||
download_dir = DOWNLOAD_DIR
|
||
if not os.path.exists(download_dir):
|
||
os.makedirs(download_dir)
|
||
|
||
export_times = []
|
||
target_task_timestamps = []
|
||
|
||
try:
|
||
# 1. 导航与页面加载
|
||
print(">> 正在进入【卸车扫描记录】界面...")
|
||
page.locator("span.ant-pro-menu-item-title:has-text('派件管理')").click()
|
||
page.locator("div.ant-pro-menu-item:has-text('卸车扫描记录')").click()
|
||
|
||
page.get_by_role("radio", name="1天").wait_for(state="visible")
|
||
print("✅ 卸车扫描记录界面加载完毕")
|
||
|
||
# 2. 从配置读取时间并设置日期选择器
|
||
query_days = 1
|
||
try:
|
||
if os.path.exists(CONFIG_PATH):
|
||
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
|
||
config = yaml.safe_load(f) or {}
|
||
query_days = int(config.get("shunxin", {}).get("query_days", 1))
|
||
except Exception:
|
||
pass
|
||
|
||
today = datetime.now()
|
||
start_date = today - timedelta(days=(query_days - 1))
|
||
|
||
today_str = today.strftime("%Y-%m-%d")
|
||
start_date_str = start_date.strftime("%Y-%m-%d")
|
||
|
||
print(f">> 正在设置查询时间范围: [{start_date_str}] 至 [{today_str}]...")
|
||
|
||
# 分两步精准呼出和点击时间控件
|
||
print(" >> 设置起始时间...")
|
||
page.get_by_placeholder("开始时间").click()
|
||
page.wait_for_timeout(500)
|
||
page.locator(
|
||
f".ant-picker-dropdown:visible td[title='{start_date_str}']"
|
||
).first.click()
|
||
page.wait_for_timeout(300)
|
||
|
||
print(" >> 设置截止时间...")
|
||
page.get_by_placeholder("结束时间").click()
|
||
page.wait_for_timeout(500)
|
||
page.locator(
|
||
f".ant-picker-dropdown:visible td[title='{today_str}']"
|
||
).first.click()
|
||
page.wait_for_timeout(300)
|
||
|
||
# 确认日期
|
||
page.locator(".ant-picker-dropdown:visible button", has_text="确 定").click()
|
||
page.wait_for_timeout(500)
|
||
|
||
# ====================================================================
|
||
# 🛡️ 卸车扫描记录页面双重校验兜底机制
|
||
# ====================================================================
|
||
print(">> 正在发起查询与数据状态研判...")
|
||
has_data = False
|
||
|
||
for attempt in range(2):
|
||
print(f" -> 第 {attempt + 1} 次点击【查询】按钮...")
|
||
page.get_by_role("button", name="search 查询").click()
|
||
|
||
loading_spinner = page.locator(".ant-spin-dot-spin").first
|
||
try:
|
||
loading_spinner.wait_for(state="visible", timeout=800)
|
||
print(" ⏳ 捕捉到加载动画，等待数据渲染完成...")
|
||
loading_spinner.wait_for(state="hidden", timeout=15000)
|
||
except Exception:
|
||
print(" ⚡ 加载动画闪过过快或未出现，强制安全缓冲1秒...")
|
||
page.wait_for_timeout(1000)
|
||
|
||
empty_desc = page.locator(
|
||
".ant-empty-description", has_text="暂无数据"
|
||
).first
|
||
data_rows = page.locator(".ant-table-tbody > tr.ant-table-row")
|
||
|
||
if empty_desc.is_visible():
|
||
print(" ⚠️ 当前表格显示【暂无数据】。")
|
||
if attempt == 0:
|
||
print(" -> 疑似前端 DOM 旧状态未刷新，触发二次兜底查询...")
|
||
page.wait_for_timeout(500)
|
||
else:
|
||
print(" -> 已二次确认为空数据环境。")
|
||
elif data_rows.count() > 0:
|
||
has_data = True
|
||
print(" ✅ 数据已成功加载。")
|
||
break
|
||
else:
|
||
pass
|
||
|
||
if not has_data:
|
||
print(">> ⚠️ 本次查询未产生任何卸车记录，提前结束流程。")
|
||
_close_tab(page, "卸车扫描记录")
|
||
return True
|
||
|
||
# ====================================================================
|
||
|
||
# 3. 直接发起全局导出
|
||
print(">> 正在发起全局数据导出请求...")
|
||
page.get_by_role("button", name="export 导出").click()
|
||
|
||
page.locator(
|
||
"span.ant-transfer-list-header-title:has-text('待选导出列')"
|
||
).wait_for(state="visible")
|
||
page.locator(".ant-transfer-list").first.locator(
|
||
".ant-transfer-list-header label"
|
||
).click()
|
||
|
||
page.get_by_label("导出").get_by_role("button", name="right").click()
|
||
page.get_by_role("button", name="export 导出数据").click()
|
||
|
||
export_times.append(datetime.now())
|
||
|
||
page.locator("text=任务添加成功!").wait_for(state="visible")
|
||
page.get_by_role("button", name="知道了").click()
|
||
print("✅ 卸车扫描记录导出任务已成功提交！")
|
||
|
||
# 4. 关闭标签页
|
||
_close_tab(page, "卸车扫描记录")
|
||
|
||
# 5. 前往数据导出页面去下载
|
||
print(">> 正在前往【数据导出】界面...")
|
||
page.locator("a[href='/dataExport']").click()
|
||
|
||
page.get_by_role("columnheader", name="任务标题").wait_for(state="visible")
|
||
page.wait_for_timeout(2000)
|
||
|
||
page.get_by_role("button", name="search 查询").click()
|
||
page.wait_for_timeout(2000)
|
||
|
||
# 6. 轮询任务状态
|
||
print(">> 列表已加载，开始匹配并检查任务状态...")
|
||
while True:
|
||
rows = page.locator(".ant-table-tbody > tr.ant-table-row")
|
||
row_count = rows.count()
|
||
pending_tasks = 0
|
||
current_ready_timestamps = []
|
||
|
||
for i in range(row_count):
|
||
tds = rows.nth(i).locator("td")
|
||
if tds.count() < 9:
|
||
continue
|
||
|
||
submit_time_str = tds.nth(2).inner_text().strip()
|
||
title_str = tds.nth(5).inner_text().strip()
|
||
status_str = tds.nth(6).inner_text().strip()
|
||
|
||
if title_str.startswith("卸车扫描记录"):
|
||
try:
|
||
row_time = datetime.strptime(
|
||
submit_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_str != "执行完成":
|
||
pending_tasks += 1
|
||
if submit_time_str not in current_ready_timestamps:
|
||
print(
|
||
f" ⏳ 任务 [{submit_time_str}] 状态为【{status_str}】，数据生成中..."
|
||
)
|
||
else:
|
||
if submit_time_str not in current_ready_timestamps:
|
||
current_ready_timestamps.append(submit_time_str)
|
||
except Exception as e:
|
||
print(f" ⚠️ 解析时间时出错: {e}")
|
||
|
||
if pending_tasks > 0:
|
||
print(
|
||
f">> 共有 {pending_tasks} 个匹配任务还在处理中，等待 5 秒后刷新..."
|
||
)
|
||
page.wait_for_timeout(5000)
|
||
page.get_by_role("button", name="search 查询").click()
|
||
page.wait_for_timeout(2000)
|
||
else:
|
||
target_task_timestamps = current_ready_timestamps
|
||
if len(target_task_timestamps) > 0:
|
||
print(">> 目标任务已就绪，开始下载...")
|
||
break
|
||
|
||
# 7. 下载逻辑
|
||
downloaded_files = []
|
||
for time_str in target_task_timestamps:
|
||
try:
|
||
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}] ...")
|
||
|
||
with page.expect_download() as download_info:
|
||
target_row.locator("td").nth(8).locator(
|
||
"button", has_text=re.compile(r"下\s*载")
|
||
).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}")
|
||
except Exception as e:
|
||
print(f" ❌ 下载任务 [{time_str}] 失败: {e}")
|
||
|
||
# 8. 合并数据
|
||
if downloaded_files:
|
||
print("\n>> 正在合并下载的数据...")
|
||
all_data_frames = []
|
||
for file_path in downloaded_files:
|
||
try:
|
||
df = pd.read_excel(file_path, dtype=str)
|
||
if not df.empty:
|
||
all_data_frames.append(df)
|
||
except Exception as e:
|
||
pass
|
||
|
||
if all_data_frames:
|
||
combined_df = pd.concat(all_data_frames, ignore_index=True)
|
||
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"====================================================")
|
||
|
||
for file_path in downloaded_files:
|
||
os.remove(file_path)
|
||
print("✅ 临时文件已清理。")
|
||
|
||
# 9. 关闭标签页
|
||
_close_tab(page, "数据导出")
|
||
|
||
print("\n【顺心 - 实到货物数据下载】流程结束。")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"\n❌ 任务执行过程中发生异常: {e}")
|
||
return False
|