alert-contract-expiring-hr/scripts/service/platform_collector.py

"""Playwright RPA：从 SmartCampus HR 平台抓合同表 + 续聘材料包。

设计要点：
- 使用 launch_persistent_context 复用浏览器登录态，避免密码硬编码
- 首次使用需用户手工登录一次（CLI 默认 --headless false 便于交互）
- 后续使用无头模式（profile 已存 cookie）
- 抓表格：DOM eval_on_selector_all → JSON
- 抓材料：点 "续聘材料包" 按钮 → expect_download → 保存到本地

依赖为可选：playwright 未安装时 import 抛错，上层 CLI 捕获并降级报错。
"""

from __future__ import annotations

import os
import re
import sqlite3
from datetime import datetime
from pathlib import Path

from db import repository as repo
from util.constants import SKILL_SLUG
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


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 平台 → 抓合同表 + 材料 → 落库。

    返回 dict 形如：
        {
            "platform_url": "...",
            "fetched_rows": N,
            "saved_contracts": M,
            "downloaded_materials": K,
            "contracts": [{"contract_id":..., "teacher_name":..., ...}],
        }
    """
    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",
        profile_dir, download_dir, headless, limit,
    )

    contracts: list[dict] = []
    fetched_rows = 0
    downloaded = 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},
        )
        try:
            page = ctx.new_page()
            page.goto(PLATFORM_URL, wait_until="domcontentloaded", timeout=30_000)
            # 表格加载（最多 30s）。Sandbox 是 mock，可能永远不出现。
            try:
                page.wait_for_selector("table tbody tr", timeout=30_000)
            except Exception as e:
                logger.warning("表格加载超时：%s；可能未登录或 sandbox mock 无数据", e)
                return _error_result("table_not_found", 0)

            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)

            for i, r in enumerate(rows):
                if limit and i >= limit:
                    break
                # 列序按截图：姓名/单位/到期日/剩余天数/预警/建议/材料包
                if len(r) < 6:
                    logger.warning("行 %d 列数 < 6，跳过: %s", i, r)
                    continue

                teacher = r[0].strip()
                unit = r[1].strip()
                contract_end = r[2].strip()
                days_raw = r[3].strip()
                alert_label = r[4].strip()
                suggestion_hint = r[5].strip()

                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": [],
                }

                # 点"续聘材料包"按钮
                try:
                    btn = page.locator("table tbody tr").nth(i).locator(
                        "button:has-text('续聘材料包')"
                    )
                    with page.expect_download(timeout=10_000) as dl_info:
                        btn.click()
                    dl = dl_info.value
                    fname = dl.suggested_filename or "materials.zip"
                    dest = download_dir / f"{contract_id}_{fname}"
                    dl.save_as(str(dest))
                    contract["materials"].append(str(dest))
                    downloaded += 1
                except Exception as e:
                    logger.warning("材料下载失败 %s: %s", contract_id, e)

                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（占位：file_path 写入，extracted_text 待 parse-materials）
                from db.migrations import ensure_materials_content_schema

                ensure_materials_content_schema(conn)

                for fp in c["materials"]:
                    fname = Path(fp).name
                    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"),
                        ),
                    )
        return len(contracts)
    finally:
        conn.close()


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(teacher: str, end_date: str) -> str:
    """从 (姓名, 到期日) 派 contract_id，确保幂等。"""
    safe_teacher = re.sub(r"[^\w]", "", teacher)[:8] or "T"
    safe_end = end_date.replace("-", "")[:8]
    return f"RPA-{safe_teacher}-{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"