diff --git a/config.example.yaml b/config.example.yaml index 6c4658f..4ff7b0d 100644 --- a/config.example.yaml +++ b/config.example.yaml @@ -20,6 +20,13 @@ debug: # 留空或不匹配时将回退为全量模式。 target_site: "" +# ---------------------------------------------------------------------------- +# 顺心捷达 (https://sxne.sxjdfreight.com) +# ---------------------------------------------------------------------------- +shunxin: + # 应到 / 实到数据的查询时间范围(单位:天,向前回溯 N 天至今天)。 + query_days: 1 + # ---------------------------------------------------------------------------- # 百世快运 (https://v5.800best.com) # ---------------------------------------------------------------------------- diff --git a/site_shunxin.py b/site_shunxin.py index 184a644..ae65111 100644 --- a/site_shunxin.py +++ b/site_shunxin.py @@ -2,10 +2,11 @@ import os import re -from datetime import datetime +import yaml +from datetime import datetime, timedelta import pandas as pd -from paths import DOWNLOAD_DIR +from paths import DOWNLOAD_DIR, CONFIG_PATH def _close_tab(page, tab_name): @@ -55,24 +56,102 @@ def shunxin_expected_download(page): page.get_by_role("button", name="强卸").wait_for(state="visible") print("✅ 车辆点到界面加载完毕") - # 2. 筛选条件:1天、状态=已发 -> 已到、查询 - print(">> 正在设置筛选条件: 选择【1天】...") - page.get_by_role("radio", name="1天").check() + # 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(">> 正在点击【查询】按钮...") - page.get_by_role("button", name="search 查询").click() - page.wait_for_timeout(2000) + # ==================================================================== + # 🛡️ 双重校验兜底机制:破除顺心表格“暂无数据”的旧状态遗留陷阱 + # ==================================================================== + 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 + + # ==================================================================== - # 3. 获取所有【运单列表】按钮并循环处理 - waybill_btns = page.get_by_role("button", name="运单列表") count = waybill_btns.count() print(f">> 共发现 {count} 个班次需要导出。") @@ -103,13 +182,9 @@ def shunxin_expected_download(page): page.get_by_role("tab", name="车辆点到").click() page.wait_for_timeout(500) - if count > 0: - print("✅ 所有班次的导出任务已成功提交!") - else: - print("⚠️ 未发现任何运单列表,直接跳转至下载环节。") + print("✅ 所有班次的导出任务已成功提交!") - # 5. 关闭【运单列表】与【车辆点到】标签页(这两个页面的工作均已完成) - # 运单列表为单一复用标签,所有班次导出完成后统一关闭即可。 + # 5. 关闭标签页 _close_tab(page, "运单列表") _close_tab(page, "车辆点到") @@ -191,7 +266,6 @@ def shunxin_expected_download(page): ).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) @@ -201,7 +275,7 @@ def shunxin_expected_download(page): except Exception as e: print(f" ❌ 下载任务 [{time_str}] 失败: {e}") - # 9. 合并数据 (修改了文件名前缀) + # 9. 合并数据 if downloaded_files: print("\n>> 正在合并下载的数据...") all_data_frames = [] @@ -225,10 +299,10 @@ def shunxin_expected_download(page): os.remove(file_path) print("✅ 临时文件已清理。") - # 10. 关闭【数据导出】标签页(该页面工作已完成) + # 10. 关闭标签页 _close_tab(page, "数据导出") - print("\n【顺心 - 应到货物数据下载】流程结束。") + return True except Exception as e: print(f"\n❌ 任务执行过程中发生异常: {e}") @@ -255,16 +329,92 @@ def shunxin_actual_download(page): page.get_by_role("radio", name="1天").wait_for(state="visible") print("✅ 卸车扫描记录界面加载完毕") - # 2. 筛选条件:1天、查询 - print(">> 正在设置筛选条件: 选择【1天】...") - page.get_by_role("radio", name="1天").check() + # 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 - print(">> 正在点击【查询】按钮...") - page.get_by_role("button", name="search 查询").click() - page.wait_for_timeout(2000) + 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(">> 正在发起导出请求...") + print(">> 正在发起全局数据导出请求...") page.get_by_role("button", name="export 导出").click() page.locator( @@ -283,7 +433,7 @@ def shunxin_actual_download(page): page.get_by_role("button", name="知道了").click() print("✅ 卸车扫描记录导出任务已成功提交!") - # 4. 关闭【卸车扫描记录】标签页(该页面工作已完成) + # 4. 关闭标签页 _close_tab(page, "卸车扫描记录") # 5. 前往数据导出页面去下载 @@ -364,7 +514,6 @@ def shunxin_actual_download(page): ).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) @@ -374,7 +523,7 @@ def shunxin_actual_download(page): except Exception as e: print(f" ❌ 下载任务 [{time_str}] 失败: {e}") - # 8. 合并数据 (修改了文件名前缀) + # 8. 合并数据 if downloaded_files: print("\n>> 正在合并下载的数据...") all_data_frames = [] @@ -398,10 +547,11 @@ def shunxin_actual_download(page): os.remove(file_path) print("✅ 临时文件已清理。") - # 9. 关闭【数据导出】标签页(该页面工作已完成) + # 9. 关闭标签页 _close_tab(page, "数据导出") print("\n【顺心 - 实到货物数据下载】流程结束。") + return True except Exception as e: print(f"\n❌ 任务执行过程中发生异常: {e}")