"""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_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_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 _wait_for_manual_login(page, wait_sec: int = HUMAN_WAIT_TIMEOUT) -> bool: """手工登录 fallback：轮询 _is_logged_in，每 HUMAN_POLL_INTERVAL 秒一次。 与 skill-template task_rpa.py::_ensure_logged_in 设计对齐： - 不自动破解，等用户在浏览器手工完成登录 - 屏幕打印 [登录] 提示 - 超时抛 LoginTimeoutError（ERROR:LOGIN_TIMEOUT） 自动 fill demo/123456 失败时调用此函数。 """ print( f"[登录] 自动登录失败，请在浏览器中完成登录" f"（sandbox demo/123456 或真实账号），最多等待 {wait_sec} 秒..." ) deadline = time.monotonic() + wait_sec attempt = 0 while time.monotonic() < deadline: attempt += 1 if _is_logged_in(page): print(f"[登录] 检测到登录成功（attempt {attempt}）") return True time.sleep(HUMAN_POLL_INTERVAL) raise LoginTimeoutError( f"自动登录失败，{wait_sec}s 内用户在浏览器中也未完成登录" ) 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.11：先用**最小侵入**方式检测已登录态 —— 直接 goto SPA_HOME_URL # （不要先跳 ROOT_URL + 自动 fill，会破坏已登录态） # v1.1.11 改进：探针结果 + 二次等待 + cookie 检测，三重判断避免误判 logger.info( "platform_collector: 检测登录态，goto SPA_HOME 先探 (当前 URL=%s)", page.url, ) _goto(page, SPA_HOME_URL) # v1.1.11：等 sandbox SPA 异步验证 cookie（POST_LOGIN_SETTLE_MS → SPA_HYDRATION_MS） page.wait_for_timeout(SPA_HYDRATION_MS) # v1.1.11：二次等待 —— sandbox SPA 有时首次访问会先跳一个中间过渡页 # （如 /select），等几秒看 URL 是否稳定到 /app/ first_url = page.url page.wait_for_timeout(SPA_HOME_RETRY_DELAY_MS) second_url = page.url if first_url != second_url: logger.info( "platform_collector: 探针 URL 过渡 %s → %s，等稳定", first_url, second_url, ) page.wait_for_timeout(SPA_HOME_RETRY_DELAY_MS) if _is_logged_in(page): logger.info( "platform_collector: ✅ 已登录态，**完全跳过登录流程**（不弹窗、不自动 fill、不打扰）" "，URL=%s", page.url, ) else: logger.info( "platform_collector: 未登录态（SPA_HOME 跳回 login, URL=%s），走登录分支", page.url, ) # 2. 未登录：直接 goto sandbox SPA login（绕过 /select 和 root 跳转） _goto(page, SPA_LOGIN_URL) # SPA hydration + framework 初始化时间（修复登录时间太短的核心点） page.wait_for_timeout(SPA_HYDRATION_MS) # 3. sandbox SPA 自动登录（demo/123456 已知可用） # 失败 fallback 手工登录等待（轮询 _is_logged_in + 180s 兜底） if not _fill_spa_login(page): logger.warning( "platform_collector: 自动 fill 失败（URL=%s），fallback 手工登录等待", page.url, ) if not headless: _wait_for_manual_login(page, wait_sec=human_wait_timeout) else: # headless 模式无法手工登录：抛 LoginTimeoutError 让 CLI 决定 raise LoginTimeoutError( f"headless 模式下自动 fill 失败，且无法手工登录" f"（请用 --show-browser 重试）" ) else: # 等 SPA 路由稳定（之前这里没有 wait，直接 goto home 会打断 pending 跳转） page.wait_for_timeout(POST_LOGIN_SETTLE_MS) # 4. 强制 goto SPA home（如果还停在 login，重试登录一次） for retry in range(SPA_HOME_RETRY_MAX): if _is_logged_in(page): break logger.info( "platform_collector: 强制 goto SPA_HOME (retry=%d), 当前 URL=%s", retry, page.url, ) _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) # 4.5. retry 后若仍未登录，headless 模式抛超时 if not _is_logged_in(page): if headless: raise LoginTimeoutError( "SPA_HOME retry 全部失败，URL 仍未离开 login 页" ) _wait_for_manual_login(page, wait_sec=human_wait_timeout) # 5. 等 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） # 修复：之前 15s 太短，sandbox mock 加载慢时直接 fallback 到 pushState → 表格加载不出来 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) return { "platform_url": PLATFORM_URL, "fetched_rows": fetched_rows, "saved_contracts": saved, "downloaded_materials": downloaded, "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（即使已抓到卡片也入队， # process_queue 失败时可用作重试锚点；enqueue_download_task 幂等） repo.enqueue_download_task(conn, c["contract_id"], "platform_button") return len(contracts) finally: conn.close() 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"