214 lines
8.9 KiB
Python
214 lines
8.9 KiB
Python
"""Playwright RPA：SmartCampus HR 平台合同台账落库 + 下载队列入队。
|
||
|
||
v1.1.31：从 platform_collector.py 拆分出来的 DB 持久化子模块。
|
||
包含 2 个函数：
|
||
- _persist_contracts（contracts 表 upsert + materials_content 写入 + sandbox_card 性能解析）
|
||
- _enqueue_download_tasks_safe（download_tasks 表入队，idempotent auto-create-table）
|
||
|
||
设计要点（保持 v1.1.30 不变）：
|
||
- 纯 DB 模块，无 Playwright 依赖
|
||
- 顶部 import db.repository / db.connection / db.migrations
|
||
- 异常处理返回 dict（_enqueue）而非 raise，便于 collect_all 容错
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sqlite3
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from db import repository as repo
|
||
from db.connection import get_connection
|
||
from db.migrations import ensure_materials_content_schema
|
||
from util.logging_setup import get_skill_logger
|
||
|
||
logger = get_skill_logger()
|
||
|
||
|
||
def _persist_contracts(contracts: list[dict]) -> int:
|
||
"""把抓到的合同 + 材料落到 SQLite。仅插入 contracts 表的占位行，
|
||
materials_content 由 parse-materials CLI 后续解析写入。
|
||
|
||
真正的合同台账来自 HR 系统 API（生产）或样例 CSV（本地），
|
||
本采集仅作为"快照"记录抓取时间与材料路径。
|
||
"""
|
||
if not contracts:
|
||
return 0
|
||
|
||
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 解析后文件路径）
|
||
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": []}
|
||
|
||
enqueued = 0
|
||
skipped = 0
|
||
failed = 0
|
||
errors: list[str] = []
|
||
|
||
owns_conn = conn is None
|
||
if owns_conn:
|
||
conn = get_connection()
|
||
# v1.1.29：idempotent 表创建 —— 用户没跑 init-db 时，download_tasks 表不存在，
|
||
# 首次采集直接建表（CREATE TABLE IF NOT EXISTS）而不是报 "no such table"
|
||
try:
|
||
conn.executescript(
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS download_tasks (
|
||
task_id TEXT PRIMARY KEY,
|
||
contract_id TEXT NOT NULL,
|
||
kind TEXT NOT NULL DEFAULT 'platform_button',
|
||
status TEXT NOT NULL DEFAULT 'pending',
|
||
attempts INTEGER NOT NULL DEFAULT 0,
|
||
error_msg TEXT,
|
||
enqueued_at TEXT NOT NULL,
|
||
started_at TEXT,
|
||
finished_at TEXT,
|
||
source_path TEXT,
|
||
result_path TEXT,
|
||
FOREIGN KEY (contract_id) REFERENCES contracts(contract_id)
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_dlq_status ON download_tasks(status);
|
||
CREATE INDEX IF NOT EXISTS idx_dlq_contract ON download_tasks(contract_id);
|
||
"""
|
||
)
|
||
conn.commit()
|
||
except Exception as ensure_err:
|
||
logger.warning(
|
||
"platform_collector: ensure download_tasks 表创建失败（继续走入队流程）: %s",
|
||
ensure_err,
|
||
)
|
||
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],
|
||
}
|