Parameterize Shunxin query window and document it in example config

- 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>
2026-06-21 10:43:55 +08:00
parent f4f9f520e1
commit 5b7a024fa8
2 changed files with 189 additions and 32 deletions
--- 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)
 # ----------------------------------------------------------------------------
--- 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}")