"""Playwright RPA:从 SmartCampus HR 平台抓合同表 + 续聘材料包。 设计要点(v1.1.6,按 skill-template examples/real_browser_rpa 设计模式): - 使用 launch_persistent_context 复用浏览器登录态,避免密码硬编码 - 首次使用需用户手工登录一次(CLI 默认 --headless false 便于交互) - 后续使用无头模式(profile 已存 cookie) - **登录策略**:自动 fill demo/123456(sandbox 已知凭据) + 失败时 fallback 手工登录等待(轮询 180s) - 抓表格:DOM eval_on_selector_all → JSON - 抓材料:点 "续聘材料包" 按钮 → 抓卡片 inner_text 依赖为可选:playwright 未安装时 import 抛错,上层 CLI 捕获并降级报错。 导航流程(v1.1.1 强化版 — 强制 goto): 1. goto 根域 2. 根域 login(如仍显示表单)→ 等跳到 /home 或 /select 3. 强制 goto sandbox SPA login URL 4. SPA login:自动 fill demo/123456,失败 fallback 手工等待(HUMAN_WAIT_TIMEOUT=180s) 5. 强制 goto SPA_HOME 6. click 侧边栏 hr/contracts 7. 等 table tbody tr(最长 ~40s) """ from __future__ import annotations import os import re import sqlite3 import time from datetime import datetime from pathlib import Path from db import repository as repo from util.constants import ( CARD_APPEAR_WAIT_MS, DEFAULT_PAGE_TIMEOUT_MS, FULL_RENDER_WAIT_MS, GOTO_TIMEOUT_MS, HUMAN_POLL_INTERVAL, HUMAN_PROMPT_INTERVAL, HUMAN_WAIT_TIMEOUT, POST_LOGIN_SETTLE_MS, POST_RELOAD_LOGIN_HYDRATION_MS, POST_RELOGIN_GOTO_WAIT_MS, PUSHSTATE_SETTLE_MS, RELOAD_DOMCONTENTLOADED_TIMEOUT_MS, RELOAD_NETWORKIDLE_TIMEOUT_MS, SIDEBAR_CLICK_URL_TIMEOUT_MS, SIDEBAR_LINK_TIMEOUT_MS, SPA_HOME_RETRY_DELAY_MS, SPA_HOME_RETRY_MAX, SPA_HYDRATION_MS, SPA_LOGIN_INPUT_TIMEOUT_MS, SPA_LOGIN_SETTLE_MS, SPA_LOGIN_URL_TIMEOUT_MS, TABLE_INITIAL_WAIT_MS, TABLE_POLL_INTERVAL_MS, TABLE_POLL_MAX_ATTEMPTS, TABLE_TR_TIMEOUT_MS, URL_STABLE_POLL_S, URL_STABLE_REQUIRED_S, URL_AFTER_LOGIN_TIMEOUT_MS, ) from util.logging_setup import get_skill_logger from util.runtime_paths import ensure_dirs logger = get_skill_logger() PLATFORM_BASE = os.environ.get("SMARTCAMPUS_BASE_URL", "https://sandbox.jc2009.com") PLATFORM_HR_PATH = os.environ.get( "SMARTCAMPUS_HR_PATH", "/industries/campus/smartcampus/app/hr/contracts" ) 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" # ---------- 错误码(与 skill-template 对齐) ---------- class RPAError(Exception): """RPA 流程错误基类。""" code: str = "RPA_ERROR" def __init__(self, message: str): super().__init__(message) self.message = message class LoginTimeoutError(RPAError): """手工登录超时(HUMAN_WAIT_TIMEOUT 秒内未检测到登录成功)。 错误码:ERROR:LOGIN_TIMEOUT(与 skill-template task_rpa.py 对齐) """ code = "ERROR:LOGIN_TIMEOUT" class MissingBrowserError(RPAError): """未检测到 Chrome/Edge。 错误码:ERROR:MISSING_BROWSER(与 skill-template browser_session.py 对齐) """ code = "ERROR:MISSING_BROWSER" class TableNotFoundError(RPAError): """40s 内未找到表格。 错误码:ERROR:TABLE_NOT_FOUND """ code = "ERROR:TABLE_NOT_FOUND" # ---------- 工具函数 ---------- def _safe_getenv_dir(env_key: str, default_rel: str) -> Path: raw = os.environ.get(env_key) if raw: return Path(raw).resolve() from util.runtime_paths import get_data_dir return (get_data_dir() / default_rel).resolve() def _goto(page, url: str, timeout: int = GOTO_TIMEOUT_MS) -> 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) def _dump_page(page, path: Path, label: str) -> None: """Dump 当前页面 HTML 到指定路径,自动创建父目录。修复:父目录可能不存在导致静默失败。""" try: path.parent.mkdir(parents=True, exist_ok=True) html = page.content() path.write_text(html, encoding="utf-8") logger.info( "platform_collector: %s dump → %s (%d 字节)", label, path, len(html), ) except Exception as e: logger.warning("dump %s 失败: %s", label, e) def _is_logged_in(page) -> bool: """检测 sandbox SPA 是否已登录(URL 含 /app/ 且不含 /login 或 /signin)。 与 skill-template real_browser_rpa _is_logged_in 概念对齐(这里只看 URL, 因为 sandbox SPA 登录后 URL 稳定跳到 /app/...)。 """ url_lower = page.url.lower() return ( "/app/" in url_lower and "/login" not in url_lower and "/signin" not in url_lower ) def _is_spa_routed(page) -> bool: """v1.1.22:检测 sandbox SPA 是否已路由到 SPA 页面(含 /select /app/)。""" url_lower = page.url.lower() return ( ("/app/" in url_lower or "/select" in url_lower) and "/login" not in url_lower and "/signin" not in url_lower ) def _try_auto_sign_in(page, wait_input_sec: int = 2) -> bool: """v1.1.21:检测登录页 demo/123456 是否已预填,有则自动点 Sign in。 v1.1.24:扩展到支持 SPA 二级登录页(/industries/campus/smartcampus/login), 按钮文本可能是中文「登录」而非英文「Sign in」。 适用场景: - 根域登录页(https://sandbox.jc2009.com/):按钮「Sign in」 - SPA 二级登录页(/industries/campus/smartcampus/login):按钮「登录」 默认有 demo/123456(或 18677308191 等真实账号)预填在 input。 行为: 1. 等 input 出现,最长 wait_input_sec 秒(默认 2s) 2. 检查 input 是否都有值 3. 有值 → 点 Sign in / 登录 → return True 4. 无值 → return False(交给 _wait_for_login_inputs_filled 接管) 不会自动 fill(如果没值就放弃,让用户自己输)。 """ try: page.wait_for_selector( "#jc-user, #jc-pass, input[type='tel'], input[type='password']", timeout=wait_input_sec * 1000, ) except Exception: logger.info( "platform_collector: %ds 内未出现登录 input,跳过自动登录", wait_input_sec, ) return False # 多种 input selector 优先级 user_input = None for sel in ("#jc-user", "input[type='tel']", "input[autocomplete='username']"): loc = page.locator(sel).first if loc.count() > 0: user_input = loc break if user_input is None: return False pass_input = None for sel in ("#jc-pass", "input[type='password']", "input[autocomplete='current-password']"): loc = page.locator(sel).first if loc.count() > 0: pass_input = loc break if pass_input is None: return False user_val = (user_input.input_value() or "").strip() pass_val = (pass_input.input_value() or "").strip() if not user_val or not pass_val: logger.info( "platform_collector: 登录页 input 为空(user=%r, pass=%r)," "等用户输入", bool(user_val), bool(pass_val), ) return False # 多种按钮文本优先级:英文 Sign in > 中文登录 > submit btn = None btn_text_used = "" for txt in ("Sign in", "登录", "Login", "signin"): loc = page.locator(f"button:has-text('{txt}')").first if loc.count() > 0: btn = loc btn_text_used = txt break if btn is None: btn = page.locator("button[type='submit']").first btn_text_used = "submit" if btn is None or btn.count() == 0: logger.warning("platform_collector: 未找到登录按钮") return False logger.info( "platform_collector: 登录页已预填账号(user=%r),自动点 %s 按钮", user_val, btn_text_used, ) btn.click() return True def _wait_for_login_inputs_filled(page, wait_sec: int = HUMAN_WAIT_TIMEOUT) -> bool: """v1.1.21:等用户输入完账号密码,然后自动点 Sign in。 - 每 HUMAN_POLL_INTERVAL 秒检查 #jc-user / #jc-pass 是否有值 - 两个都有值 → 自动点 Sign in → return True - 超时未填完 → return False """ deadline = time.monotonic() + wait_sec last_prompt_at = time.monotonic() last_state_log = time.monotonic() while time.monotonic() < deadline: cur_url = page.url user_input = page.locator("#jc-user").first pass_input = page.locator("#jc-pass").first # 已不在登录页(用户可能手动点了别的) if not _is_login_page_url_for_user(cur_url): logger.info( "platform_collector: URL 已离开登录页(%s),停止等输入", cur_url, ) return True if user_input.count() > 0 and pass_input.count() > 0: user_val = (user_input.input_value() or "").strip() pass_val = (pass_input.input_value() or "").strip() now = time.monotonic() if now - last_state_log >= HUMAN_PROMPT_INTERVAL: last_state_log = now logger.info( "platform_collector: input 状态 user=%r pass=%r", bool(user_val), bool(pass_val), ) if user_val and pass_val: logger.info( "platform_collector: ✅ 用户输入完成(user=%r),自动点 Sign in", user_val, ) btn = page.locator('button[type="submit"]:has-text("Sign in")').first if btn.count() > 0: btn.click() return True # 提醒用户 now = time.monotonic() if now - last_prompt_at >= HUMAN_PROMPT_INTERVAL: last_prompt_at = now remaining = int(deadline - now) print(f"[登录] 等待用户输入账号密码... 剩余 {remaining}s") time.sleep(HUMAN_POLL_INTERVAL) logger.warning( "platform_collector: %ds 内用户未完成输入", wait_sec, ) return False def _is_login_page_url_for_user(url: str) -> bool: """判断 URL 是否还在登录页(用于 _wait_for_login_inputs_filled 退出条件)。""" u = url.lower() if "/login" in u or "/signin" in u or "/auth/" in u: return True # 根域本身也是登录页(demo 数据未提交前的初始态) return u.rstrip("/").endswith("sandbox.jc2009.com") def _wait_for_manual_login(page, wait_sec: int = HUMAN_WAIT_TIMEOUT) -> bool: """手工登录 fallback(v1.1.19 重写,v1.1.20 加强,v1.1.21 修订): 等用户在浏览器完成登录。 判定「登录完成」的多个信号(满足任一即视为完成): 1. URL 含 `/app/` 且稳定 URL_STABLE_REQUIRED_S 秒(v1.1.19 旧逻辑) 2. URL 已离开 `/login` / `/signin` / `/auth/` 等登录路径(v1.1.20 新增) —— 即便 sandbox 未跳到 /app/,只要离开登录页就认为登录成功 3. 页面出现 sidebar 元素 `a[href$='/hr/contracts']`(v1.1.20 新增) —— 说明已加载到 SPA 主页 - **不做任何 goto / reload / click**,避免打断用户输入 - 每 HUMAN_PROMPT_INTERVAL 秒打印一次剩余时间(默认 15s) - 每 HUMAN_POLL_INTERVAL 秒检查一次(默认 5s) - 超时抛 LoginTimeoutError。 """ print( f"[登录] 请在浏览器中完成登录" f"(sandbox demo/123456 或真实账号),最多等待 {wait_sec} 秒..." ) deadline = time.monotonic() + wait_sec last_prompt_at = time.monotonic() last_url = page.url last_url_change_at = time.monotonic() def _is_login_page_url(url: str) -> bool: u = url.lower() return ( "/login" in u or "/signin" in u or "/auth/" in u or u.rstrip("/").endswith("sandbox.jc2009.com") ) def _sidebar_present() -> bool: try: return page.locator("a[href$='/hr/contracts']").count() > 0 except Exception: return False while True: remaining = deadline - time.monotonic() if remaining <= 0: break cur_url = page.url if cur_url != last_url: logger.info( "platform_collector: URL 变化 %s → %s", last_url, cur_url, ) last_url = cur_url last_url_change_at = time.monotonic() # 信号 1:URL 跳到 /app/... 并稳定 if _is_logged_in(page): stable_for = time.monotonic() - last_url_change_at if stable_for >= URL_STABLE_REQUIRED_S: print( f"[登录] ✅ 检测到登录完成(URL 含 /app/,稳定 {stable_for:.1f}s)" ) return True logger.info( "platform_collector: URL 含 /app/ 但还在过渡,等稳定 " "(current=%s, stable_for=%.1fs)", cur_url, stable_for, ) # 信号 2:URL 已离开登录页(即使不在 /app/,比如卡在中间过渡页也算) elif not _is_login_page_url(cur_url): stable_for = time.monotonic() - last_url_change_at if stable_for >= URL_STABLE_REQUIRED_S: print( f"[登录] ✅ 检测到登录完成(URL 已离开登录页 → {cur_url}," f"稳定 {stable_for:.1f}s)" ) return True logger.info( "platform_collector: URL 已离开登录页但还没稳定,等 " "(current=%s, stable_for=%.1fs)", cur_url, stable_for, ) # 信号 3:sidebar 已出现 elif _sidebar_present(): print( f"[登录] ✅ 检测到登录完成(sidebar 已出现,URL={cur_url})" ) return True else: # 还在登录页:每 HUMAN_PROMPT_INTERVAL 秒提醒用户一次 now = time.monotonic() if now - last_prompt_at >= HUMAN_PROMPT_INTERVAL: last_prompt_at = now print( f"[登录] 等待用户输入... 剩余 {int(remaining)}s " f"(当前 URL={cur_url})" ) time.sleep(HUMAN_POLL_INTERVAL) raise LoginTimeoutError( f"自动登录失败,{wait_sec}s 内用户在浏览器中也未完成登录" f"(最后 URL={page.url})" ) def _wait_for_url_stable( page, target_substr: str | tuple[str, ...], max_wait_sec: int = 30, ) -> bool: """v1.1.19:等 URL 含 target_substr,且连续 URL_STABLE_REQUIRED_S 秒不变。 v1.1.22:target_substr 支持 tuple(任一匹配即视为稳定)。 用于「已登录态直接跳到目标页面」后等待 SPA 路由稳定。 **不做任何 goto / reload**,只 listen URL 变化。 Returns: True = URL 已稳定到任一目标;False = 超时。 """ if isinstance(target_substr, str): targets = (target_substr,) else: targets = target_substr deadline = time.monotonic() + max_wait_sec last_url = page.url last_url_change_at = time.monotonic() while time.monotonic() < deadline: cur_url = page.url if cur_url != last_url: logger.info( "platform_collector._wait_for_url_stable: URL 变化 %s → %s", last_url, cur_url, ) last_url = cur_url last_url_change_at = time.monotonic() cur_lower = cur_url.lower() if any(t.lower() in cur_lower for t in targets): stable_for = time.monotonic() - last_url_change_at if stable_for >= URL_STABLE_REQUIRED_S: return True time.sleep(URL_STABLE_POLL_S) return False 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=URL_AFTER_LOGIN_TIMEOUT_MS, ) 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=SIDEBAR_LINK_TIMEOUT_MS // 2) except Exception as e: logger.warning("等 sandbox 路径超时: %s;当前 URL=%s", e, page.url) return True def _fill_spa_login(page, settle_ms: int = SPA_LOGIN_SETTLE_MS) -> bool: """sandbox SPA 自动登录(如还在登录页):fill demo/123456 → click 登录 → 等跳转。 返回是否执行了登录(True = 已点登录按钮且 URL 已跳离 login;调用方可继续后续流程)。 失败时**不抛异常**,由调用方决定是否 fallback 到 _wait_for_manual_login。 v1.1.10 关键修复:**先检测 _is_logged_in,已登录态直接 return True 不再 fill**。 之前没这个检查,profile 复用时会被强制 demo fill 破坏 session。 v1.1.11 关键修复:**sandbox SPA 登录 input 实际是 `type="tel"` + 无 autocomplete**, 之前 selector `input[autocomplete="username"]` 永远找不到,自动 fill 100% 失败。 改用 sandbox 实际 selector `#jc-user` / `#jc-pass`(来自 root 登录页)+ `button[type="submit"]`。 修复要点(v1.1.5/v1.1.6 登录时间太短导致采集不到数据): - 默认 settle_ms = SPA_LOGIN_SETTLE_MS(4000ms),给 SPA 异步渲染 + token 写入 cookie 时间 - 等 username input 的 timeout = SPA_LOGIN_INPUT_TIMEOUT_MS(20s) - click 按钮后 wait_for_url timeout = SPA_LOGIN_URL_TIMEOUT_MS(30s) - 多处过渡路径(/login, /signin, /auth/)一并判断 """ # v1.1.10:先检查是否已登录(profile cookie 复用场景),已登录直接 return True if _is_logged_in(page): logger.info( "platform_collector: 已登录态(profile cookie 复用),跳过自动 fill" ) return True url_lower = page.url.lower() if "/login" not in url_lower and "/signin" not in url_lower and "/auth/" not in url_lower: return False # v1.1.11:sandbox SPA 登录页 input 是 #jc-user / #jc-pass(与 root 登录页同 ID) # 等输入框出现(带超时,避免 sandbox mock 慢启动卡死) try: page.wait_for_selector( '#jc-user, input[type="tel"], input[autocomplete="username"]', timeout=SPA_LOGIN_INPUT_TIMEOUT_MS, ) except Exception as e: logger.warning("SPA login input 未出现: %s;当前 URL=%s", e, page.url) return False # v1.1.11:按优先级选 selector —— sandbox 实际用 #jc-user user_sel = None for sel in ('#jc-user', 'input[type="tel"]', 'input[autocomplete="username"]'): loc = page.locator(sel).first if loc.count() > 0: user_sel = sel break if not user_sel: logger.warning("SPA login input selector 全 miss(#jc-user / type=tel / autocomplete=username)") return False page.locator(user_sel).first.fill("demo") # 密码 selector 同样按优先级 pass_sel = None for sel in ('#jc-pass', 'input[type="password"]', 'input[autocomplete="current-password"]'): loc = page.locator(sel).first if loc.count() > 0: pass_sel = sel break if not pass_sel: logger.warning("SPA password input selector 全 miss") return False page.locator(pass_sel).first.fill("123456") page.locator( 'button:has-text("登录"), button[type="submit"]:has-text("Sign in"), ' 'button[type="submit"]' ).first.click() logger.info("platform_collector: sandbox SPA 自动登录 (user_sel=%s pass_sel=%s)", user_sel, pass_sel) page.wait_for_timeout(settle_ms) try: page.wait_for_url( lambda u: "/login" not in u.lower() and "/signin" not in u.lower() and "/auth/" not in u.lower(), timeout=SPA_LOGIN_URL_TIMEOUT_MS, ) except Exception as e: logger.warning("SPA 登录超时: %s;当前 URL=%s", e, page.url) return False # v1.1.10:fill 后再检测 _is_logged_in,避免 sandbox demo 凭据失效时 # 被误判为「已登录」(URL 跳 /app/home 但 cookie 没写入) if not _is_logged_in(page): logger.warning( "platform_collector: SPA URL 跳离 login 但 _is_logged_in=False,登录可能失败" ) return False logger.info("platform_collector: SPA 登录后 URL=%s", page.url) return True def collect_all( limit: int = 0, headless: bool = True, human_wait_timeout: int = HUMAN_WAIT_TIMEOUT, ) -> dict: """登录 sandbox 平台 → 抓合同表 + 材料 → 落库。 human_wait_timeout: 手工登录 fallback 等待秒数(默认 HUMAN_WAIT_TIMEOUT=180)。 失败时可能抛出 RPAError(LoginTimeoutError / MissingBrowserError / TableNotFoundError)。 """ try: from playwright.sync_api import sync_playwright except ImportError: logger.error("playwright 未安装,跳过 RPA 采集") return _error_result("playwright_not_installed", 0) profile_dir = _safe_getenv_dir("SMARTCAMPUS_PROFILE_DIR", "playwright-profile") download_dir = _safe_getenv_dir("MATERIALS_DOWNLOAD_DIR", "materials") profile_dir.mkdir(parents=True, exist_ok=True) download_dir.mkdir(parents=True, exist_ok=True) logger.info( "platform_collector: 启动 profile=%s download_dir=%s headless=%s limit=%d " "human_wait_timeout=%d", profile_dir, download_dir, headless, limit, human_wait_timeout, ) contracts: list[dict] = [] fetched_rows = 0 downloaded = 0 with sync_playwright() as p: # v1.1.9:headed 模式优先用系统 Edge(跟 Windows 桌面 session 同源, # 能真正弹窗让你看到并手工登录);找不到则 fallback Playwright 内置 chromium if not headless: try: ctx = p.chromium.launch_persistent_context( user_data_dir=str(profile_dir), headless=False, accept_downloads=True, channel="msedge", viewport={"width": 1440, "height": 900}, ) logger.info( "platform_collector: 使用系统 Edge(headed 弹窗)" ) except Exception as e: logger.warning( "platform_collector: 系统 Edge 启动失败(%s),fallback Playwright chromium", e, ) ctx = p.chromium.launch_persistent_context( user_data_dir=str(profile_dir), headless=False, accept_downloads=True, viewport={"width": 1440, "height": 900}, ) else: ctx = p.chromium.launch_persistent_context( user_data_dir=str(profile_dir), headless=True, accept_downloads=True, viewport={"width": 1440, "height": 900}, ) try: page = ctx.new_page() # v1.1.6:覆盖 Playwright 默认操作超时(之前无 set_default_timeout,导致 # 所有 wait_for_selector 默认 30s,调试时不直观) page.set_default_timeout(DEFAULT_PAGE_TIMEOUT_MS) # 1. v1.1.21:先 goto **根域**(sandbox 登录页),让 SPA hydrate 2 秒 logger.info( "platform_collector: 启动,goto ROOT_URL=%s (当前 URL=%s)", ROOT_URL, page.url, ) _goto(page, ROOT_URL) # 给 React/Vue 框架 2 秒 hydrate 时间,看登录 input 是否就绪 page.wait_for_timeout(2_000) if _is_logged_in(page): logger.info( "platform_collector: ✅ 已登录态(URL=%s),跳过登录流程", page.url, ) else: # 2. 等 #jc-user / #jc-pass 出现(最多 2 秒) # 若 sandbox 已预填 demo/123456 → 自动点 Sign in # 若 input 为空 → 等用户输入完成后再自动点 Sign in auto_clicked = _try_auto_sign_in(page, wait_input_sec=2) if auto_clicked: logger.info( "platform_collector: ✅ 自动登录成功(预填 demo),等 URL 跳到 /app/" ) elif not headless: # 3. input 为空:等用户输入 → 自动点 Sign in if _wait_for_login_inputs_filled(page, wait_sec=human_wait_timeout): logger.info( "platform_collector: ✅ 用户输入完成并点击,等 URL 跳到 /app/" ) else: # 超时未输入 → fallback 到原来的多信号等待 logger.warning( "platform_collector: 用户未输入完成,fallback 到 _wait_for_manual_login" ) _wait_for_manual_login(page, wait_sec=human_wait_timeout) else: raise LoginTimeoutError( f"headless 模式下未登录,且无法手工登录" f"(请用 --show-browser 重试)" ) # 4. 登录完成 → 等 URL 真正稳定到 /app/ 或 /select(最多 30s,**不做任何操作**) # v1.1.22:sandbox 登录后会先跳 /select,再自动跳 /app/home, # 所以两个 URL 都算"SPA 已路由" if not _wait_for_url_stable(page, ("/app/", "/select"), max_wait_sec=30): logger.warning( "platform_collector: 登录后 30s 内 URL 未稳定到 /app/或 /select(当前 URL=%s)", page.url, ) # 不强行 reload —— 让用户决定是否手工刷新 # 4.5. v1.1.24:sandbox 可能登录后把人踢回 SPA 二级登录页 # (/industries/campus/smartcampus/login)。如果 URL 落在 # 这个二级登录页,再走一次 _try_auto_sign_in(已经扩展支持 # 中文「登录」按钮);若 SPA 登录页 input 有预填账号 → 自动点; # 若 input 空 → 走 _wait_for_login_inputs_filled 等用户输入。 cur_url_after_stable = page.url.lower() if "/login" in cur_url_after_stable or "/signin" in cur_url_after_stable: logger.warning( "platform_collector: 登录后仍落在登录页(%s)," "v1.1.24 触发 SPA 二级登录兜底", page.url, ) spa_auto = _try_auto_sign_in(page, wait_input_sec=2) if spa_auto: logger.info( "platform_collector: ✅ SPA 二级登录页已自动登录," "等 URL 跳到 /app/ 或 /select" ) _wait_for_url_stable(page, ("/app/", "/select"), max_wait_sec=30) elif not headless: if _wait_for_login_inputs_filled(page, wait_sec=human_wait_timeout): _wait_for_url_stable(page, ("/app/", "/select"), max_wait_sec=30) else: _wait_for_manual_login(page, wait_sec=human_wait_timeout) else: raise LoginTimeoutError( f"SPA 二级登录页 input 空且 headless 无法手工登录(URL={page.url})" ) # 5. v1.1.22:显式跳到 SPA_HOME_URL(/app/home),等侧边栏 link 出现 # 避免 sandbox 在 /select 中间页时直接找 sidebar link 找不到 if "/app/home" not in page.url.lower(): logger.info( "platform_collector: 当前 URL=%s,跳到 SPA_HOME_URL=%s", page.url, SPA_HOME_URL, ) _goto(page, SPA_HOME_URL) page.wait_for_timeout(SPA_HYDRATION_MS) # 等 URL 稳定到 /app/home(最多 15s,**不做任何操作**) if not _wait_for_url_stable(page, "/app/home", max_wait_sec=15): logger.warning( "platform_collector: 15s 内未稳定到 /app/home(当前 URL=%s)", page.url, ) # 6. 等 SPA 渲染 + click 侧边栏 hr/contracts(SPA 路由,避免 goto 触发整页 reload 丢 session) try: # dump SPA home 一次(用于调试 selector) page.wait_for_timeout(POST_LOGIN_SETTLE_MS) _dump_page(page, download_dir.parent / "spa_home_dump.html", "SPA home") # 等 SPA 真正加载完(侧边栏 link 出现,最长 SIDEBAR_LINK_TIMEOUT_MS) try: page.wait_for_selector( "a[href$='/hr/contracts']", timeout=SIDEBAR_LINK_TIMEOUT_MS, state="attached", ) except Exception as e: 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=SIDEBAR_CLICK_URL_TIMEOUT_MS, ) 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(PUSHSTATE_SETTLE_MS) 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) # 5.5. v1.1.8:**首次登录后不再强制刷新**,避免打断用户手工登录态。 # 改为「等表格出现」兜底:给 SPA 一个初始等待窗口(TABLE_INITIAL_WAIT_MS*2), # 如果表格仍未出现,**才**刷新一次(sandbox mock 异步加载场景)。 logger.info( "platform_collector: 进入 hr/contracts 后**不刷新**,等待表格出现" "(首次登录态不要打断)" ) # 5.6. 兜底刷新:仅当等表格超时后才触发;刷新后可能被踢回 login,需重新登录 # 修复:之前没检查 URL,刷新后停在 login 页继续往后走,找不到表格就 dump 报错 def _try_reload_after_table_timeout(): logger.warning( "platform_collector: 表格未在 %dms 内出现,触发兜底刷新", TABLE_INITIAL_WAIT_MS * 2, ) try: page.reload( wait_until="networkidle", timeout=RELOAD_NETWORKIDLE_TIMEOUT_MS, ) except Exception as e: logger.warning("刷新 networkidle 失败: %s;尝试 domcontentloaded", e) try: page.reload( wait_until="domcontentloaded", timeout=RELOAD_DOMCONTENTLOADED_TIMEOUT_MS, ) except Exception as e2: logger.warning("刷新 domcontentloaded 也失败: %s", e2) if "/login" in page.url.lower() or "/signin" in page.url.lower(): logger.warning( "platform_collector: 刷新后回到登录页(%s),重新登录", page.url, ) _goto(page, SPA_LOGIN_URL) page.wait_for_timeout(POST_RELOAD_LOGIN_HYDRATION_MS) if not _fill_spa_login(page): if not headless: _wait_for_manual_login(page, wait_sec=human_wait_timeout) else: raise LoginTimeoutError( "刷新后回到 login 页且 headless 无法手工登录" ) else: page.wait_for_timeout(POST_LOGIN_SETTLE_MS) _goto(page, PLATFORM_URL) page.wait_for_timeout(POST_RELOGIN_GOTO_WAIT_MS) # 7. dump 当前 hr/contracts 页面(用于调试) _dump_page(page, download_dir.parent / "hr_contracts_dump.html", "hr/contracts") # 8. 等待表格加载(v1.1.8:先等,**表格超时未出现才触发兜底刷新**—— # 修复「首次登录后强制刷新打断手工登录态」问题) page.wait_for_timeout(TABLE_INITIAL_WAIT_MS) # 给 React 初始化 + mock 数据填充时间 found_table = False for attempt in range(TABLE_POLL_MAX_ATTEMPTS): page.wait_for_timeout(TABLE_POLL_INTERVAL_MS) cur_url = page.url try: page.wait_for_selector( "table tbody tr", timeout=TABLE_TR_TIMEOUT_MS, ) found_table = True break except Exception: logger.info( "platform_collector: URL=%s 等待表格加载 (attempt %d/%d)...", cur_url, attempt + 1, TABLE_POLL_MAX_ATTEMPTS, ) if not found_table: # v1.1.8:表格没出现,**先尝试兜底刷新一次**(sandbox mock 异步加载场景); # 刷新后再次轮询表格,仍找不到才报 TableNotFoundError。 logger.warning( "platform_collector: 首次轮询未找到表格,触发兜底刷新(sandbox mock 异步加载场景)" ) _try_reload_after_table_timeout() # 刷新后重新轮询一次(最多 5 次) for attempt2 in range(5): page.wait_for_timeout(TABLE_POLL_INTERVAL_MS) try: page.wait_for_selector( "table tbody tr", timeout=TABLE_TR_TIMEOUT_MS, ) found_table = True logger.info( "platform_collector: 兜底刷新后找到表格 (attempt %d)", attempt2 + 1 ) break except Exception: logger.info( "platform_collector: 兜底刷新后仍等待表格 (attempt %d/5)...", attempt2 + 1, ) if not found_table: _dump_page(page, download_dir.parent / "sandbox_dump.html", "sandbox") logger.warning("platform_collector: 表格未找到(兜底刷新后仍未出现)") raise TableNotFoundError( f"{TABLE_POLL_MAX_ATTEMPTS * TABLE_POLL_INTERVAL_MS // 1000}s " f"内未找到表格(兜底刷新后仍未出现) (URL={page.url})" ) # 表格已找到,等 tbody 完全填充 + dump 一份完整 DOM(用于诊断按钮 selector) try: page.wait_for_function( "document.querySelectorAll('table tbody tr').length > 0", timeout=DEFAULT_PAGE_TIMEOUT_MS, ) page.wait_for_timeout(FULL_RENDER_WAIT_MS) _dump_page( page, download_dir.parent / "hr_contracts_full_dump.html", "完整 hr/contracts", ) except Exception as e: logger.warning("等完整渲染超时: %s", e) rows = page.eval_on_selector_all( "table tbody tr", "rs => rs.map(r => Array.from(r.querySelectorAll('td')).map(td => td.innerText))", ) 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 # 抓「材料包」card 区段的 DOM(用于后续解析每行 click 后的卡片文本) # sandbox 行为:click 按钮后会在表格下方追加「续聘准备任务书」+「单位清单」卡片 for i, r in enumerate(rows): if limit and i >= limit: break # v1.1.8:表格列数自适应 —— sandbox 首次登录未刷新是 7 列 # (姓名/单位/到期日/剩余天数/预警/建议/按钮),刷新后才是 10 列 # (多 工号/岗位/合同类型/起止日期)。两种都支持。 ncols = len(r) if ncols < 6: logger.warning( "行 %d 列数=%d < 6,跳过: %s", i, ncols, r ) continue # 列序探测:判断首列是不是工号(纯数字 + 字母,纯姓名是中文) first_col = r[0].strip() import re as _re is_emp_id = bool(_re.match(r"^[A-Za-z]?\d+$", first_col)) if is_emp_id and ncols >= 9: # 10 列布局(含工号):0 工号 1 姓名 2 单位 3 岗位 4 合同类型 # 5 起止(含~) 6 剩余天数 7 预警 8 建议 9 按钮 emp_id = first_col teacher = r[1].strip() unit = r[2].strip() period = r[5].strip() contract_end = period.split("~")[-1].strip() if "~" in period else period days_raw = r[6].strip() alert_label = r[7].strip() suggestion_hint = r[8].strip() elif ncols >= 6: # 7 列布局(无工号):0 姓名 1 单位 2 到期日 3 剩余天数 4 预警 5 建议 6 按钮 teacher = first_col unit = r[1].strip() contract_end = r[2].strip() days_raw = r[3].strip() alert_label = r[4].strip() suggestion_hint = r[5].strip() emp_id = "" # 工号缺失,留空让 _gen_contract_id 用 teacher_name 兜底 else: logger.warning( "行 %d 列数=%d 无法识别布局,跳过: %s", i, ncols, r ) continue # 用工号生成 contract_id(稳定,不会随姓名变化) # v1.1.8:7 列布局无工号 → 用 teacher_name 派生(避免与 10 列布局冲突) if emp_id: contract_id = _gen_contract_id(emp_id, contract_end) else: contract_id = _gen_contract_id(teacher, contract_end) contract = { "contract_id": contract_id, "teacher_name": teacher, "unit_name": unit, "contract_end": _normalize_date(contract_end), "days_to_expire_hint": _parse_days(days_raw), "alert_level_hint": _parse_alert(alert_label), "ai_suggestion_hint": suggestion_hint, "materials": [], } # 点「续聘材料包」按钮 → 等卡片展开 → 抓卡片文本(含 checklist) if mat_btn_count > 0: try: btn = page.locator("table tbody tr").nth(i).locator( "button:has-text('续聘材料包')" ) btn.click() # 等卡片出现(页面下方追加 h2「续聘准备任务书」/「单位清单」) page.wait_for_timeout(CARD_APPEAR_WAIT_MS) # 抓表格下方所有 card 文本 cards_text = page.locator( "main .grid > div, main > div > div > div" ).all_inner_texts() materials_text = "\n\n".join( t.strip() for t in cards_text if t.strip() ) if materials_text: # 存到 materials_content(sandbox mock 卡片文本作为材料包内容) contract["materials"].append({ "kind": "sandbox_card", "file_name": f"{contract_id}_materials.txt", "text": materials_text, }) downloaded += 1 logger.info( "platform_collector: 抓到卡片文本 %d 字 (%s)", len(materials_text), contract_id, ) # dump 一次(i==0)方便调试 if i == 0: _dump_page( page, download_dir.parent / "after_click_dump.html", "首行 click 后", ) except Exception as e: logger.warning("抓卡片失败 %s: %s", contract_id, e) else: if i == 0: logger.info( "platform_collector: sandbox 未部署「续聘材料包」按钮," "跳过材料抓取(仅采集表格列)", ) contracts.append(contract) finally: ctx.close() saved = _persist_contracts(contracts) logger.info( "platform_collector: contracts 落库完成 saved=%d / total_fetched=%d", saved, len(contracts), ) # v1.1.18:每个合同入队 download_tasks(独立事务,失败不影响主流程) contract_ids = [c["contract_id"] for c in contracts if c.get("contract_id")] enqueue_summary = _enqueue_download_tasks_safe(contract_ids) return { "platform_url": PLATFORM_URL, "fetched_rows": fetched_rows, "saved_contracts": saved, "downloaded_materials": downloaded, "download_queue": enqueue_summary, "contracts": contracts, } def _persist_contracts(contracts: list[dict]) -> int: """把抓到的合同 + 材料落到 SQLite。仅插入 contracts 表的占位行, materials_content 由 parse-materials CLI 后续解析写入。 真正的合同台账来自 HR 系统 API(生产)或样例 CSV(本地), 本采集仅作为"快照"记录抓取时间与材料路径。 """ if not contracts: return 0 from db.connection import get_connection conn = get_connection() try: with conn: for c in contracts: # upsert contracts(仅更新与 sandbox 相关的字段) conn.execute( """ INSERT INTO contracts (contract_id, employee_id, teacher_name, unit_id, unit_name, position, contract_type, contract_start, contract_end, trial_period_end, status, last_material_id, renewed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, '可续聘', NULL, NULL) ON CONFLICT(contract_id) DO UPDATE SET teacher_name = excluded.teacher_name, unit_name = excluded.unit_name, contract_end = excluded.contract_end """, ( c["contract_id"], c["contract_id"], # employee_id 暂用 contract_id c["teacher_name"], "U-RPA", # unit_id 占位 c["unit_name"], "(RPA 采集)", "(RPA 采集)", "1970-01-01", c["contract_end"], c["contract_end"], ), ) # 写 materials_content(sandbox card 文本 + parse-materials 解析后文件路径) from db.migrations import ensure_materials_content_schema ensure_materials_content_schema(conn) for m in c["materials"]: if isinstance(m, dict) and m.get("kind") == "sandbox_card": # sandbox mock 卡片文本,直接写入 extracted_text conn.execute( """ INSERT OR REPLACE INTO materials_content (contract_id, file_name, file_path, extracted_text, parse_method, parsed_at) VALUES (?, ?, ?, ?, 'sandbox_card', ?) """, ( c["contract_id"], m["file_name"], f"sandbox://hr/contracts/{c['contract_id']}", m["text"], datetime.now().isoformat(timespec="seconds"), ), ) # 同步解析 → 抽 year + 等次 → 写入 performance_records perf_n = repo.parse_sandbox_card_performance( conn, c["contract_id"], m["text"], ) if perf_n > 0: logger.info( "platform_collector: %s 解析卡片 → performance %d 条", c["contract_id"], perf_n, ) else: # 本地文件路径(parse-materials 解析后写入) fp = m if isinstance(m, str) else m.get("file_path") fname = Path(fp).name if fp else "unknown" conn.execute( """ INSERT OR REPLACE INTO materials_content (contract_id, file_name, file_path, extracted_text, parse_method, parsed_at) VALUES (?, ?, ?, NULL, 'pending', ?) """, ( c["contract_id"], fname, fp, datetime.now().isoformat(timespec="seconds"), ), ) # v1.1.18 入队 download_tasks 已移到 collect_all 末尾的独立事务, # 见 _enqueue_download_tasks_safe() return len(contracts) finally: conn.close() def _enqueue_download_tasks_safe( contract_ids: list[str], conn: sqlite3.Connection | None = None, ) -> dict: """v1.1.18:每个合同入队 download_tasks。 在独立事务里完成,不影响 _persist_contracts 主流程(contracts / materials_content)。 任一入队失败 → 记 warning + 计入 failed,但不影响整体返回。 conn: 可选传入连接(用于测试 in-memory DB)。None → 走 get_connection()。 若传入 conn,调用方负责生命周期(不会 close)。 Returns: {enqueued: N, skipped: M, failed: K, errors: [...]} """ if not contract_ids: return {"enqueued": 0, "skipped": 0, "failed": 0, "errors": []} from db.connection import get_connection enqueued = 0 skipped = 0 failed = 0 errors: list[str] = [] owns_conn = conn is None if owns_conn: conn = get_connection() try: for cid in contract_ids: try: # 入队前查一下是否已存在 pending/running → 决定计入 skipped 还是 enqueued pre_existing = conn.execute( """SELECT 1 FROM download_tasks WHERE contract_id = ? AND kind = ? AND status IN ('pending', 'running') LIMIT 1""", (cid, "platform_button"), ).fetchone() repo.enqueue_download_task(conn, cid, "platform_button") if pre_existing: skipped += 1 else: enqueued += 1 except Exception as e: failed += 1 err = f"{cid}: {type(e).__name__}: {e}" errors.append(err) logger.warning( "platform_collector.enqueue: 入队 %s 失败: %s", cid, e, ) if owns_conn: conn.commit() finally: if owns_conn: conn.close() logger.info( "platform_collector.enqueue: 入队结果 enqueued=%d skipped=%d failed=%d (total=%d)", enqueued, skipped, failed, len(contract_ids), ) return { "enqueued": enqueued, "skipped": skipped, "failed": failed, "errors": errors[:10], } def _download_materials_for_row( page, row_idx: int, contract_id: str, mat_btn_count: int, ) -> list[dict]: """点续聘材料包按钮抓卡片文本(与 v1.1.17 collect_all 内联逻辑等价)。 返回 materials 列表(kind=sandbox_card,含 file_name + text)。 """ if mat_btn_count == 0: return [] materials: list[dict] = [] try: btn = page.locator("table tbody tr").nth(row_idx).locator( "button:has-text('续聘材料包')" ) btn.click() page.wait_for_timeout(CARD_APPEAR_WAIT_MS) cards_text = page.locator( "main .grid > div, main > div > div > div" ).all_inner_texts() materials_text = "\n\n".join(t.strip() for t in cards_text if t.strip()) if materials_text: materials.append({ "kind": "sandbox_card", "file_name": f"{contract_id}_materials.txt", "text": materials_text, }) logger.info( "platform_collector: 抓到卡片文本 %d 字 (%s)", len(materials_text), contract_id, ) except Exception as e: logger.warning("抓卡片失败 %s: %s", contract_id, e) return materials def _find_row_for_contract(page, contract_id: str) -> int | None: """在 hr/contracts 表格中找 contract_id 所在行号。 支持 7 列(用 teacher_name 派生 contract_id)与 10 列(用 emp_id)。 不匹配返回 None(调用方 mark_task_failed)。 """ rows = page.eval_on_selector_all( "table tbody tr", "rs => rs.map(r => Array.from(r.querySelectorAll('td')).map(td => td.innerText))", ) for i, r in enumerate(rows): ncols = len(r) if ncols < 6: continue first_col = r[0].strip() is_emp_id = bool(re.match(r"^[A-Za-z]?\d+$", first_col)) if is_emp_id and ncols >= 9: emp_id, contract_end = first_col, r[5].split("~")[-1].strip() elif ncols >= 6: emp_id, contract_end = first_col, r[2].strip() else: continue candidate = _gen_contract_id(emp_id, contract_end) if candidate == contract_id: return i return None def process_queue( limit: int = 5, headless: bool = True, human_wait_timeout: int = HUMAN_WAIT_TIMEOUT, ids_filter: str = "", ) -> dict: """从 download_tasks 队列取 pending 任务,对每个合同重新走材料抓取。 与 collect_all 重复登录流程,但每行只点「续聘材料包」按钮抓卡片,不重新抓合同表。 参数: limit: 本次最多处理任务数(0 = 不限) ids_filter: 逗号分隔 task_id 列表(只处理这些 pending) headless: 无头模式(需要 profile cookie 已保存) human_wait_timeout: 手工登录 fallback 等待秒数 返回 {processed, succeeded, failed, details}。 """ try: from playwright.sync_api import sync_playwright except ImportError: logger.error("playwright 未安装,跳过队列处理") return {"processed": 0, "succeeded": 0, "failed": 0, "details": [], "error": "playwright_not_installed"} from db.connection import get_connection # 1. 领取一批任务 ids_list = [s.strip() for s in ids_filter.split(",") if s.strip()] if ids_filter else None conn = get_connection() try: tasks = repo.claim_pending_tasks( conn, limit=limit if limit > 0 else 999, ids_filter=ids_list, ) finally: conn.close() if not tasks: return {"processed": 0, "succeeded": 0, "failed": 0, "details": []} profile_dir = _safe_getenv_dir("SMARTCAMPUS_PROFILE_DIR", "playwright-profile") download_dir = _safe_getenv_dir("MATERIALS_DOWNLOAD_DIR", "materials") profile_dir.mkdir(parents=True, exist_ok=True) download_dir.mkdir(parents=True, exist_ok=True) details: list[dict] = [] succeeded = 0 failed = 0 with sync_playwright() as p: ctx = p.chromium.launch_persistent_context( user_data_dir=str(profile_dir), headless=headless, accept_downloads=True, viewport={"width": 1440, "height": 900}, channel="msedge" if not headless else None, ) if not headless else p.chromium.launch_persistent_context( user_data_dir=str(profile_dir), headless=True, accept_downloads=True, viewport={"width": 1440, "height": 900}, ) try: page = ctx.new_page() page.set_default_timeout(DEFAULT_PAGE_TIMEOUT_MS) # 登录:复用 v1.1.17 collect_all 的探针 + 自动 fill + 手工 fallback 流程 logger.info("platform_collector.process_queue: 启动 profile=%s headless=%s 待处理 %d", profile_dir, headless, len(tasks)) _goto(page, SPA_HOME_URL) page.wait_for_timeout(SPA_HYDRATION_MS) if _is_logged_in(page): logger.info("platform_collector.process_queue: ✅ 已登录态") else: _goto(page, SPA_LOGIN_URL) page.wait_for_timeout(SPA_HYDRATION_MS) if not _fill_spa_login(page): if not headless: _wait_for_manual_login(page, wait_sec=human_wait_timeout) else: raise LoginTimeoutError("process_queue headless 自动 fill 失败") else: page.wait_for_timeout(POST_LOGIN_SETTLE_MS) # 强制 goto SPA_HOME + 侧边栏 hr/contracts for _ in range(SPA_HOME_RETRY_MAX): if _is_logged_in(page): break _goto(page, SPA_HOME_URL) page.wait_for_timeout(SPA_HOME_RETRY_DELAY_MS) if _fill_spa_login(page): page.wait_for_timeout(POST_LOGIN_SETTLE_MS) if not _is_logged_in(page): raise LoginTimeoutError("process_queue 仍未登录") # 等表格 page.wait_for_timeout(TABLE_INITIAL_WAIT_MS) found_table = False for _attempt in range(TABLE_POLL_MAX_ATTEMPTS): page.wait_for_timeout(TABLE_POLL_INTERVAL_MS) try: page.wait_for_selector( "table tbody tr", timeout=TABLE_TR_TIMEOUT_MS, ) found_table = True break except Exception: pass if not found_table: raise TableNotFoundError("process_queue 表格未找到") mat_btn_count = page.locator( "table tbody tr button:has-text('续聘材料包')" ).count() # 2. 逐任务处理 for t in tasks: contract_id = t["contract_id"] task_id = t["task_id"] conn = get_connection() try: row_idx = _find_row_for_contract(page, contract_id) if row_idx is None: repo.mark_task_failed(conn, task_id, "contract_not_found_in_table") failed += 1 details.append({"task_id": task_id, "contract_id": contract_id, "error": "contract_not_found_in_table"}) conn.commit() continue materials = _download_materials_for_row( page, row_idx, contract_id, mat_btn_count, ) from db.migrations import ensure_materials_content_schema ensure_materials_content_schema(conn) for m in materials: if isinstance(m, dict) and m.get("kind") == "sandbox_card": conn.execute( """ INSERT OR REPLACE INTO materials_content (contract_id, file_name, file_path, extracted_text, parse_method, parsed_at) VALUES (?, ?, ?, ?, 'sandbox_card', ?) """, ( contract_id, m["file_name"], f"sandbox://hr/contracts/{contract_id}", m["text"], datetime.now().isoformat(timespec="seconds"), ), ) repo.parse_sandbox_card_performance(conn, contract_id, m["text"]) result_path = ( f"sandbox://hr/contracts/{contract_id}" if materials else "no_materials_captured" ) repo.mark_task_success(conn, task_id, result_path) succeeded += 1 details.append({ "task_id": task_id, "contract_id": contract_id, "ok": True, "materials_count": len(materials), }) conn.commit() except Exception as e: repo.mark_task_failed(conn, task_id, str(e)) failed += 1 details.append({ "task_id": task_id, "contract_id": contract_id, "error": str(e), }) conn.commit() logger.warning("process_queue: task=%s 失败: %s", task_id, e) finally: conn.close() finally: ctx.close() return { "processed": len(tasks), "succeeded": succeeded, "failed": failed, "details": details, } def _error_result(reason: str, fetched: int) -> dict: return { "platform_url": PLATFORM_URL, "fetched_rows": fetched, "saved_contracts": 0, "downloaded_materials": 0, "contracts": [], "error": reason, } def _gen_contract_id(emp_id: str, end_date: str) -> str: """从 (工号, 到期日) 派 contract_id,确保幂等。 sandbox v1.1.1 起改用「工号」作主键(之前用姓名)。 """ safe_emp = re.sub(r"[^\w]", "", emp_id)[:10] or "T" safe_end = end_date.replace("-", "")[:8] return f"RPA-{safe_emp}-{safe_end}" def _normalize_date(s: str) -> str: """尽量把 sandbox 返回的日期归一为 YYYY-MM-DD。""" s = s.strip() if re.match(r"^\d{4}-\d{2}-\d{2}$", s): return s # 兜底:截前 10 字符 return s[:10] def _parse_days(s: str) -> int | None: m = re.search(r"-?\d+", s) return int(m.group(0)) if m else None def _parse_alert(s: str) -> str: s = s.strip() if "红" in s: return "red" if "橙" in s: return "orange" if "黄" in s: return "yellow" return "unknown"