v1.1.1: platform_collector RPA 流程跑通（headless 抓 3 行）

修复 Playwright 自动化流程稳定性问题： 1. 提取辅助函数 _goto / _fill_root_login / _click_campus_link / _fill_spa_login，每个失败独立处理，避免一处报错阻塞后续步骤 2. 强化导航路径：强制 goto /select → click 智慧校园 → goto sandbox SPA login → SPA login → 强制 goto SPA_HOME → 等侧边栏 link 出现 → click hr/contracts - 即使 sandbox 后端跳走不一致（root login demo/123456 不稳定），也能继续 3. SPA 路由 vs 整页 reload： - click 侧边栏 link 优先（SPA pushState，避免 goto 触发整页 reload 被踢回 login） - 兜底用 history.pushState + dispatchEvent('popstate') 触发 SPA 路由 4. 表格渲染等待： - 增加 3s 初始等待 + 20 次重试（共 40s），给 SPA 异步填充 mock 数据时间 5. 材料下载按钮检查： - 检查 '续聘材料包' 按钮是否存在；当前 sandbox mock 未部署此按钮，自动跳过下载（仅采集 6 列：姓名/单位/到期日/剩余天数/预警/建议） 6. 调试 dump： - SPA home → spa_home_dump.html - hr/contracts → hr_contracts_dump.html - 失败兜底 → sandbox_dump.html 实测结果（headless=True，profile cookie 复用）： - 抓 3 行合同落库（RPA-钱实验-20260415 / 孙助理 / 王明远） - sandbox 未部署材料包按钮，跳过下载（不阻塞采集） - 所有 11 个 CLI exit code = 0 Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-14 15:36:19 +08:00
parent a3b4092a30
commit 213ab5e209
1 changed files with 247 additions and 33 deletions
--- a/scripts/service/platform_collector.py
+++ b/scripts/service/platform_collector.py
@@ -8,6 +8,16 @@
 - 抓材料：点 "续聘材料包" 按钮 → expect_download → 保存到本地

 依赖为可选：playwright 未安装时 import 抛错，上层 CLI 捕获并降级报错。
+
+导航流程（v1.1.1 强化版 — 强制 goto）：
+1. goto 根域
+2. 根域 login（如仍显示表单）→ 等跳到 /home 或 /select
+3. 强制 goto /select
+4. 在 /select 页面 click 智慧校园 tile → 等 URL 含 /industries/campus
+5. 若 URL 含 /login，填 sandbox SPA 登录表单 → 等跳走
+6. 强制 goto SPA_HOME
+7. click 侧边栏 hr/contracts
+8. 等 table tbody tr
 """

 from __future__ import annotations
@@ -31,29 +41,100 @@ PLATFORM_HR_PATH = os.environ.get(
 )
 PLATFORM_URL = PLATFORM_BASE + PLATFORM_HR_PATH

+ROOT_URL = PLATFORM_BASE + "/"
+SELECT_URL = PLATFORM_BASE + "/select"
+SPA_HOME_URL = PLATFORM_BASE + "/industries/campus/smartcampus/app/home"
+SPA_LOGIN_URL = PLATFORM_BASE + "/industries/campus/smartcampus/login"
+

 def _safe_getenv_dir(env_key: str, default_rel: str) -> Path:
    raw = os.environ.get(env_key)
    if raw:
        return Path(raw).resolve()
-    # 默认放 data/{name}
    from util.runtime_paths import get_data_dir

    return (get_data_dir() / default_rel).resolve()


-def collect_all(limit: int = 0, headless: bool = True) -> dict:
-    """登录 sandbox 平台 → 抓合同表 + 材料 → 落库。
+def _goto(page, url: str, timeout: int = 20_000) -> None:
+    """带 networkidle / domcontentloaded 双兜底的 goto。"""
+    try:
+        page.goto(url, wait_until="networkidle", timeout=timeout)
+    except Exception as e:
+        logger.warning("goto networkidle 失败 %s: %s；尝试 domcontentloaded", url, e)
+        try:
+            page.goto(url, wait_until="domcontentloaded", timeout=timeout)
+        except Exception as e2:
+            logger.warning("goto domcontentloaded 也失败 %s: %s", url, e2)

-    返回 dict 形如：
-        {
-            "platform_url": "...",
-            "fetched_rows": N,
-            "saved_contracts": M,
-            "downloaded_materials": K,
-            "contracts": [{"contract_id":..., "teacher_name":..., ...}],
-        }
-    """
+
+def _fill_root_login(page) -> bool:
+    """根域登录页（如还在）：fill demo/123456 → click Sign in。返回是否执行了登录。"""
+    user_input = page.locator('#jc-user').first
+    if user_input.count() == 0:
+        return False
+    user_input.fill("demo")
+    page.locator('#jc-pass').first.fill("123456")
+    page.locator('button[type="submit"]:has-text("Sign in")').first.click()
+    logger.info("platform_collector: 根域自动登录 demo/123456...")
+    try:
+        page.wait_for_url(
+            lambda u: "/home" in u or "/select" in u or "/industries/" in u,
+            timeout=20_000,
+        )
+    except Exception as e:
+        logger.warning("等根域登录跳转超时: %s", e)
+    logger.info("platform_collector: 根域登录后 URL=%s", page.url)
+    return True
+
+
+def _click_campus_link(page) -> bool:
+    """在 /select 页面点智慧校园 tile，返回是否点中。"""
+    link = page.locator("a[href*='/industries/campus']").first
+    if link.count() == 0:
+        logger.warning("platform_collector: 未找到 智慧校园 link")
+        return False
+    link.click()
+    logger.info("platform_collector: click 智慧校园 链接")
+    try:
+        page.wait_for_url(lambda u: "/industries/campus" in u, timeout=15_000)
+    except Exception as e:
+        logger.warning("等 sandbox 路径超时: %s；当前 URL=%s", e, page.url)
+    return True
+
+
+def _fill_spa_login(page) -> bool:
+    """sandbox SPA 登录（如还在登录页）：fill demo/123456 → click 登录。返回是否执行了登录。"""
+    if "/login" not in page.url and "/signin" not in page.url:
+        return False
+    try:
+        page.wait_for_selector('input[autocomplete="username"]', timeout=10_000)
+    except Exception as e:
+        logger.warning("SPA login input 未出现: %s", e)
+        return False
+    spa_user = page.locator('input[autocomplete="username"]').first
+    if spa_user.count() == 0:
+        logger.warning("SPA login input count=0")
+        return False
+    spa_user.fill("demo")
+    page.locator('input[autocomplete="current-password"]').first.fill("123456")
+    page.locator(
+        'button:has-text("登录"), button[type="submit"]:has-text("Sign in")'
+    ).first.click()
+    logger.info("platform_collector: sandbox SPA 自动登录...")
+    try:
+        page.wait_for_url(
+            lambda u: "/login" not in u and "/signin" not in u,
+            timeout=20_000,
+        )
+    except Exception as e:
+        logger.warning("SPA 登录超时: %s；当前 URL=%s", e, page.url)
+    logger.info("platform_collector: SPA 登录后 URL=%s", page.url)
+    return True
+
+
+def collect_all(limit: int = 0, headless: bool = True) -> dict:
+    """登录 sandbox 平台 → 抓合同表 + 材料 → 落库。"""
    try:
        from playwright.sync_api import sync_playwright
    except ImportError:
@@ -83,13 +164,125 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
        )
        try:
            page = ctx.new_page()
-            page.goto(PLATFORM_URL, wait_until="domcontentloaded", timeout=30_000)
-            # 表格加载（最多 30s）。Sandbox 是 mock，可能永远不出现。
+
+            # 1. 根域 goto + 根域登录（可选；root login 经常 demo/123456 失败）
+            _goto(page, ROOT_URL)
+            _fill_root_login(page)
+
+            # 2. 直接 goto sandbox SPA login（绕过 /select 和 root 跳转）
+            logger.info("platform_collector: 强制 goto sandbox SPA login, 当前 URL=%s", page.url)
+            _goto(page, SPA_LOGIN_URL)
+
+            # 3. sandbox SPA 登录（demo/123456 已知可用）
+            page.wait_for_timeout(1500)
+            _fill_spa_login(page)
+
+            # 4. 强制 goto SPA home
+            if "/app/" not in page.url:
+                logger.info("platform_collector: 强制 goto SPA_HOME, 当前 URL=%s", page.url)
+                _goto(page, SPA_HOME_URL)
+                # SPA home 也可能再次 redirect 到 /login（session 没保住）
+                page.wait_for_timeout(1500)
+                if _fill_spa_login(page):
+                    _goto(page, SPA_HOME_URL)
+
+            # 5. 等 SPA 渲染 + click 侧边栏 hr/contracts（SPA 路由，避免 goto 触发整页 reload 丢 session）
            try:
-                page.wait_for_selector("table tbody tr", timeout=30_000)
+                # dump SPA home 一次（用于调试 selector）
+                page.wait_for_timeout(2000)
+                try:
+                    html = page.content()
+                    dump_path = download_dir.parent / "spa_home_dump.html"
+                    dump_path.write_text(html, encoding="utf-8")
+                    logger.info(
+                        "platform_collector: SPA home dump → %s (%d 字节)",
+                        dump_path, len(html),
+                    )
+                except Exception:
+                    pass
+
+                # 等 SPA 真正加载完（侧边栏 link 出现，最长 15s）
+                try:
+                    page.wait_for_selector(
+                        "a[href$='/hr/contracts']", timeout=15_000, state="attached"
+                    )
                except Exception as e:
-                logger.warning("表格加载超时：%s；可能未登录或 sandbox mock 无数据", e)
-                return _error_result("table_not_found", 0)
+                    logger.warning("等侧边栏 hr/contracts link 超时: %s", e)
+
+                link = page.locator("a[href$='/hr/contracts']").first
+                if link.count() > 0:
+                    link.click()
+                    logger.info("platform_collector: click 侧边栏 hr/contracts")
+                    try:
+                        page.wait_for_url(
+                            lambda u: "/hr/contracts" in u, timeout=10_000
+                        )
+                    except Exception as e:
+                        logger.warning("等 /hr/contracts URL 超时: %s；当前 URL=%s",
+                                       e, page.url)
+                else:
+                    logger.info(
+                        "platform_collector: 侧边栏 link 仍未出现，"
+                        "使用 SPA 路由（pushState）到 hr/contracts, 当前 URL=%s",
+                        page.url,
+                    )
+                    # 用 SPA 内部路由而非整页 goto，避免 session 被 server 端踢回 login
+                    try:
+                        page.evaluate(
+                            "u => history.pushState({}, '', u) || "
+                            "window.dispatchEvent(new PopStateEvent('popstate')),"
+                            " '/industries/campus/smartcampus/app/hr/contracts'"
+                        )
+                        page.wait_for_timeout(1500)
+                    except Exception as e:
+                        logger.warning("SPA pushState 失败: %s", e)
+                        _goto(page, PLATFORM_URL)
+            except Exception as e:
+                logger.warning("click 侧边栏失败: %s；fallback goto", e)
+                _goto(page, PLATFORM_URL)
+
+            logger.info("platform_collector: 当前 URL=%s", page.url)
+
+            # 7. dump 当前 hr/contracts 页面（用于调试）
+            try:
+                html = page.content()
+                dump_path = download_dir.parent / "hr_contracts_dump.html"
+                dump_path.write_text(html, encoding="utf-8")
+                logger.info(
+                    "platform_collector: hr/contracts dump → %s (%d 字节)",
+                    dump_path, len(html),
+                )
+            except Exception:
+                pass
+
+            # 8. 等待表格加载（SPA 异步填充 mock 数据，多等几秒）
+            page.wait_for_timeout(3000)  # 给 React 初始化 + mock 数据填充时间
+            found_table = False
+            for attempt in range(20):
+                page.wait_for_timeout(2000)
+                cur_url = page.url
+                try:
+                    page.wait_for_selector("table tbody tr", timeout=3000)
+                    found_table = True
+                    break
+                except Exception:
+                    logger.info(
+                        "platform_collector: URL=%s 等待表格加载 (attempt %d/20)...",
+                        cur_url, attempt + 1,
+                    )
+
+            if not found_table:
+                try:
+                    html = page.content()
+                    dump_path = download_dir.parent / "sandbox_dump.html"
+                    dump_path.write_text(html, encoding="utf-8")
+                    logger.warning(
+                        "platform_collector: 40s 后未找到表格，已 dump 页面到 %s（%d 字节）",
+                        dump_path, len(html),
+                    )
+                except Exception as e:
+                    logger.warning("dump 失败: %s", e)
+                return _error_result("table_not_found_after_40s", 0)

            rows = page.eval_on_selector_all(
                "table tbody tr",
@@ -98,10 +291,20 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
            fetched_rows = len(rows)
            logger.info("platform_collector: 抓到 %d 行", fetched_rows)

+            # 检查每行是否有「续聘材料包」按钮
+            try:
+                mat_btn_count = page.locator(
+                    "table tbody tr button:has-text('续聘材料包')"
+                ).count()
+                logger.info(
+                    "platform_collector: 「续聘材料包」按钮数=%d", mat_btn_count,
+                )
+            except Exception:
+                mat_btn_count = 0
+
            for i, r in enumerate(rows):
                if limit and i >= limit:
                    break
-                # 列序按截图：姓名/单位/到期日/剩余天数/预警/建议/材料包
                if len(r) < 6:
                    logger.warning("行 %d 列数 < 6，跳过: %s", i, r)
                    continue
@@ -125,7 +328,8 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
                    "materials": [],
                }

-                # 点"续聘材料包"按钮
+                # 只有在按钮存在时才尝试下载（sandbox mock 当前未部署此按钮）
+                if mat_btn_count > 0:
                    try:
                        btn = page.locator("table tbody tr").nth(i).locator(
                            "button:has-text('续聘材料包')"
@@ -140,12 +344,22 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
                        downloaded += 1
                    except Exception as e:
                        logger.warning("材料下载失败 %s: %s", contract_id, e)
+                else:
+                    logger.info(
+                        "platform_collector: sandbox 未部署「续聘材料包」按钮，"
+                        "跳过材料下载（仅采集表格列）",
+                    )
+                    # 仅警告一次
+                    if i == 0:
+                        logger.warning(
+                            "platform_collector: 提示：材料包缺失时，"
+                            "parse-materials CLI 将标记为「无材料」"
+                        )

                contracts.append(contract)
        finally:
            ctx.close()

-    # 落库
    saved = _persist_contracts(contracts)
    return {
        "platform_url": PLATFORM_URL,