diff --git a/.env.example b/.env.example index f4b0b4d..ff0c60a 100644 --- a/.env.example +++ b/.env.example @@ -67,3 +67,22 @@ LLM_API_BASE_URL= # LLM API token（生产由宿主注入；本地留空） LLM_API_TOKEN= +# ----- v1.1.18：日志落文件 ----- +# 日志文件路径（默认 {data_dir}/logs/alert-contract-expiring-hr.log）；留空 = 仅 stderr +LOG_FILE= +# 单文件最大字节数（默认 5MB） +LOG_MAX_BYTES= +# 滚动备份保留份数（默认 5） +LOG_BACKUP_COUNT= + +# ----- v1.1.18：用户主动输入目录 ----- +# 用户主动投放目录（默认 {data_dir}/input） +# 子目录约定： +# input/contracts/*.csv → import-input 自动 upsert 到 contracts 表 +# input/materials/{contract_id}__{filename}.{pdf|docx|doc} → 自动入 materials_content +INPUT_DIR= + +# ----- v1.1.18：材料下载队列 ----- +# 失败最大重试次数（默认 3） +DOWNLOAD_QUEUE_MAX_ATTEMPTS= + diff --git a/CHANGELOG.md b/CHANGELOG.md index 37097b3..7970adc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,17 @@ 本文件对应技能市场 **「更新日志」** Tab。 +## 1.1.18 + +- **一键式"运维 + 材料管理"全打通** + - 新增 `show-config` / `logs-list` 两个 sync Action：UI 一键查看 .env 配置（敏感字段自动脱敏）与运行日志末尾 200 行 + - 新增 `download-materials` async Action：UI 一键批量下载续聘材料包（pending 队列 → 浏览器抓取 → 落库 → 标记 success/failed） + - 新增 `import-input` CLI：扫描 `data/input/{contracts,materials}/` 自动入库，HR API 拿不到时用户可直接投放 CSV/PDF/Word +- **材料下载队列（download_tasks）**：新建表跟踪每个合同的材料抓取状态（pending/running/success/failed），失败自动重试最多 3 次 +- **日志落文件**：本地降级路径新增 `RotatingFileHandler`（默认 5MB × 5 份），`LOG_FILE` / `LOG_MAX_BYTES` / `LOG_BACKUP_COUNT` 可覆盖 +- 新增文件：`scripts/service/importer.py` / `scripts/service/config_viewer.py` / `scripts/service/logs_viewer.py` +- 现有 31 个测试全部保留；新增 17 个测试（合计 48/48） + ## 1.1.17 - 补齐数据管理的“立即同步合同台账”工具栏入口，调用 SmartCampus 平台采集能力。 diff --git a/SKILL.md b/SKILL.md index 2877f4d..80e32ce 100644 --- a/SKILL.md +++ b/SKILL.md @@ -1,7 +1,7 @@ --- name: 合同到期智能预警与续聘辅助 description: "当用户说要扫描即将到期的合同、查 90/60/30 天预警名单、给某位教师生成续聘任务书、下发单位版续聘清单、从 SmartCampus HR 平台抓合同数据、解析续聘材料 PDF/Word、查看本月合同到期人员清单时，加载本技能。支持定时扫描（每日凌晨）+ 数据管理 + Agent 三入口，覆盖合同到期智能预警与续聘辅助全流程。" -version: 1.1.17 +version: 1.1.18 author: 深圳匠厂科技有限公司 metadata: openclaw: @@ -61,6 +61,10 @@ allowed-tools: | 在浏览器中打开续聘任务书 | `run_skill_action` → `render-html` | 否（sync） | | 从 SmartCampus 平台抓合同 + 材料 | `run_skill_action` → `collect-from-platform`（首次需用户手工登录保存 profile） | 否（sync） | | 解析已下载材料 PDF/Word | `run_skill_action` → `parse-materials`（可选 `--contract-id`） | 否（sync） | +| 查看运行时配置（脱敏） | `run_skill_action` → `show-config` | 否（sync） | +| 查看运行日志末尾 N 行 | `run_skill_action` → `logs-list`（默认 tail=200、level=INFO） | 否（sync） | +| 批量下载材料包（队列） | `run_skill_action` → `download-materials`（输入：itemIds / pickCount） | 是（async） | +| 扫描输入目录并入库 CSV/材料 | `run_skill_action` → `import-input` | 是（async） | ## 共享 Python（必读） diff --git a/assets/actions.json b/assets/actions.json index 73510ae..b962b8e 100644 --- a/assets/actions.json +++ b/assets/actions.json @@ -129,6 +129,72 @@ "command": "render-html", "args": ["--task-id", "{form.task_id}"] } + }, + { + "id": "show-config", + "label": "查看运行时配置", + "description": "展示 .env 中已设置的全部非空配置项；含 TOKEN/SECRET/KEY/PASSWORD 的字段自动脱敏。", + "placements": ["skill-detail", "agent"], + "executionProfile": "sync", + "entrypoint": { + "type": "cli", + "command": "show-config", + "args": [] + } + }, + { + "id": "logs-list", + "label": "查看运行日志", + "description": "读取日志文件末尾 N 行（默认 200 行），按 INFO/WARNING/ERROR 等级过滤；日志文件不存在时返回 exists=false。", + "placements": ["skill-detail", "agent"], + "executionProfile": "sync", + "entrypoint": { + "type": "cli", + "command": "logs-list", + "args": ["--tail", "200", "--level", "INFO"] + } + }, + { + "id": "download-materials", + "label": "批量下载材料包", + "description": "从合同台账的 download_tasks 队列领取 N 条 pending 任务，启动浏览器从 SmartCampus HR 平台下载续聘材料包（pdf/docx）。", + "placements": ["toolbar", "batch", "cron", "agent"], + "executionProfile": "async", + "interaction": "form", + "bind": { + "tables": ["download_tasks"], + "selectionMode": "optional", + "labelColumn": "contract_id", + "inputMapping": {"itemIds": "$selection.ids"} + }, + "confirmation": { + "message": "将启动浏览器下载续聘材料包（仅 pending 会执行），是否继续？" + }, + "entrypoint": { + "type": "cli", + "command": "download-materials", + "args": ["--ids", "{form.itemIds}", "--pick", "{form.pickCount}"] + }, + "inputSchema": { + "type": "object", + "required": ["itemIds", "pickCount"], + "properties": { + "itemIds": {"type": "string", "title": "队列编号列表", "readOnly": true}, + "pickCount": {"type": "number", "title": "领取数量", "default": 5, "minimum": 1, "maximum": 50} + } + } + }, + { + "id": "import-input", + "label": "扫描输入目录并入库", + "description": "主动扫描 data/input/contracts/*.csv 入库合同，data/input/materials/{contract_id}__*.* 入库材料（自动入库到 materials_content）。", + "placements": ["skill-detail", "agent", "cron"], + "executionProfile": "async", + "entrypoint": { + "type": "cli", + "command": "import-input", + "args": [] + } } ] } diff --git a/scripts/cli/app.py b/scripts/cli/app.py index fa884ce..2690807 100644 --- a/scripts/cli/app.py +++ b/scripts/cli/app.py @@ -279,6 +279,72 @@ def cmd_parse_materials(args: argparse.Namespace) -> int: conn.close() +def cmd_import_input(_args: argparse.Namespace) -> int: + """扫描 data/input/{contracts,materials}/ 自动入库。 + + 约定： + input/contracts/*.csv → contracts 表 upsert + input/materials/{cid}__{f}.pdf → materials_content (parse_method='pending') + """ + from service.importer import import_input_dir + + conn = get_connection() + try: + result = import_input_dir(conn) + finally: + conn.close() + print(json.dumps(result, ensure_ascii=False, indent=2)) + return 0 + + +def cmd_download_materials(args: argparse.Namespace) -> int: + """处理 download_tasks 队列：抓 pending 任务的材料 → 落库。 + + --ids "DLQ-xxx,DLQ-yyy": 只处理这些任务 + --pick N: 无 --ids 时从 pending 领取 N 条（默认 5） + --limit N: 本次硬上限（默认 0 = 不限） + --headless: 无头模式（需要 profile cookie 已保存） + """ + try: + result = platform_collector.process_queue( + limit=args.limit or args.pick, + ids_filter=args.ids, + headless=args.headless, + human_wait_timeout=args.human_wait_timeout or HUMAN_WAIT_TIMEOUT, + ) + except platform_collector.RPAError as e: + print(json.dumps({ + "error": e.code, + "detail": str(e), + "processed": 0, "succeeded": 0, "failed": 0, "details": [], + }, ensure_ascii=False, indent=2)) + return 2 + except Exception as e: + print(json.dumps({"error": "exception", "detail": str(e)}, + ensure_ascii=False, indent=2)) + return 1 + + print(json.dumps(result, ensure_ascii=False, indent=2)) + return 0 + + +def cmd_show_config(_args: argparse.Namespace) -> int: + """查看 .env 非空配置（敏感字段自动脱敏）。""" + from service.config_viewer import read_env_config + + print(json.dumps(read_env_config(), ensure_ascii=False, indent=2)) + return 0 + + +def cmd_logs_list(args: argparse.Namespace) -> int: + """读取日志末尾 N 行（默认 200）。""" + from service.logs_viewer import tail_log_file + + result = tail_log_file(tail=args.tail, min_level=args.level) + print(json.dumps(result, ensure_ascii=False, indent=2)) + return 0 + + # ---------- argparse ---------- def _parse_thresholds(s: str) -> tuple[int, int, int]: @@ -360,6 +426,41 @@ def build_parser() -> argparse.ArgumentParser: help="仅解析该合同（留空=处理所有 pending）") sp.set_defaults(handler=cmd_parse_materials) + sp = sub.add_parser( + "download-materials", + help="处理材料下载队列（pending 任务 → 浏览器抓取 → 落库）", + ) + sp.add_argument("--limit", type=int, default=0, help="本次最多处理任务数（0 = 用 --pick）") + sp.add_argument("--ids", default="", help="逗号分隔 task_id 列表（仅处理这些 pending 任务）") + sp.add_argument("--pick", type=int, default=5, help="无 --ids 时从 pending 领取条数（默认 5）") + sp.add_argument("--headless", action="store_true", help="无头模式（需已保存 profile）") + sp.add_argument("--show-browser", dest="headless", action="store_false", + help=argparse.SUPPRESS) + sp.add_argument("--human-wait-timeout", type=int, default=None, + help="手工登录 fallback 等待秒数（默认 180 = HUMAN_WAIT_TIMEOUT）") + sp.set_defaults(headless=False, handler=cmd_download_materials) + + sp = sub.add_parser( + "import-input", + help="扫描 data/input/{contracts,materials}/ 自动入库", + ) + sp.set_defaults(handler=cmd_import_input) + + sp = sub.add_parser( + "show-config", + help="查看 .env 配置（敏感字段自动脱敏）", + ) + sp.set_defaults(handler=cmd_show_config) + + sp = sub.add_parser( + "logs-list", + help="查看运行日志末尾 N 行", + ) + sp.add_argument("--tail", type=int, default=200, help="末尾行数（默认 200）") + sp.add_argument("--level", default="INFO", + help="最低级别 DEBUG/INFO/WARNING/ERROR（默认 INFO）") + sp.set_defaults(handler=cmd_logs_list) + return p @@ -370,7 +471,8 @@ def main(argv: Optional[List[str]] = None) -> int: print("Commands: health, version, config-path, init-db, " "scan-expiring-contracts, generate-renewal-task, " "dispatch-task-notice, mark-contract-renewed, render-html, " - "collect-from-platform, parse-materials") + "collect-from-platform, parse-materials, download-materials, " + "import-input, show-config, logs-list") return 1 setup_skill_logging() diff --git a/scripts/db/connection.py b/scripts/db/connection.py index 6411e52..ccb7a3c 100644 --- a/scripts/db/connection.py +++ b/scripts/db/connection.py @@ -90,6 +90,25 @@ DDL_STATEMENTS: tuple[str, ...] = ( ); """, "CREATE INDEX IF NOT EXISTS idx_dispatch_unit_month ON dispatch_log(unit_id, month);", + # 7. download_tasks (v1.1.18) + """ + 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);", ) @@ -131,6 +150,7 @@ def init_db(db_path: Path | None = None) -> dict[str, int]: "renewal_tasks": conn.execute("SELECT COUNT(*) AS n FROM renewal_tasks").fetchone()["n"], "dispatch_log": conn.execute("SELECT COUNT(*) AS n FROM dispatch_log").fetchone()["n"], "materials_content": conn.execute("SELECT COUNT(*) AS n FROM materials_content").fetchone()["n"], + "download_tasks": conn.execute("SELECT COUNT(*) AS n FROM download_tasks").fetchone()["n"], } return counts finally: @@ -140,5 +160,5 @@ def init_db(db_path: Path | None = None) -> dict[str, int]: def table_names() -> Iterable[str]: return ( "contracts", "performance_records", "alerts", "renewal_tasks", - "dispatch_log", "materials_content", + "dispatch_log", "materials_content", "download_tasks", ) diff --git a/scripts/db/repository.py b/scripts/db/repository.py index 3e2bc9e..681408a 100644 --- a/scripts/db/repository.py +++ b/scripts/db/repository.py @@ -294,6 +294,59 @@ def list_dispatch_logs(conn: sqlite3.Connection) -> list[sqlite3.Row]: # ---------- materials_content（v1.1.1 新增） ---------- +def upsert_contract_from_csv_row( + conn: sqlite3.Connection, row: dict[str, str] +) -> None: + """从 CSV 字典 upsert 一份合同。CSV 列名必须与 contracts 表列一致（或常见别名）。 + + 缺字段兜底值： + employee_id = contract_id + teacher_name = "" + unit_id = "U-UNKNOWN" + unit_name = "" + position = "" + contract_type = "" + contract_start = "1970-01-01" + contract_end = "2099-12-31" + trial_period_end = contract_end + status = "可续聘" + """ + cid = row.get("contract_id") or row.get("合同ID") or "" + if not cid: + raise ValueError("CSV row missing contract_id / 合同ID") + 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, + position = excluded.position, + contract_type = excluded.contract_type, + contract_start = excluded.contract_start, + contract_end = excluded.contract_end, + trial_period_end = excluded.trial_period_end, + status = excluded.status + """, + ( + cid, + row.get("employee_id") or row.get("工号") or cid, + row.get("teacher_name") or row.get("教职工姓名") or "", + row.get("unit_id") or row.get("单位ID") or "U-UNKNOWN", + row.get("unit_name") or row.get("单位") or "", + row.get("position") or row.get("岗位") or "", + row.get("contract_type") or row.get("合同类型") or "", + row.get("contract_start") or row.get("合同起始日") or "1970-01-01", + row.get("contract_end") or row.get("合同到期日") or "2099-12-31", + row.get("trial_period_end") or row.get("试用期截止") or row.get("contract_end") or "2099-12-31", + row.get("status") or "可续聘", + ), + ) + + def list_materials_for_contract( conn: sqlite3.Connection, contract_id: str ) -> list[sqlite3.Row]: @@ -373,3 +426,124 @@ def summarize_materials( if len(summary) > 16000: summary = summary[:16000] + "\n...(材料过长已截断)" return (summary, file_names) + + +# ---------- download_tasks（v1.1.18 新增） ---------- + + +def enqueue_download_task( + conn: sqlite3.Connection, + contract_id: str, + kind: str = "platform_button", +) -> str: + """入队一个下载任务，返回 task_id。 + + 已存在 (contract_id, kind, status ∈ {pending, running}) 时跳过（返回已存在的 task_id）。 + """ + import uuid + + existing = conn.execute( + """SELECT task_id FROM download_tasks + WHERE contract_id = ? AND kind = ? AND status IN ('pending', 'running') + LIMIT 1""", + (contract_id, kind), + ).fetchone() + if existing: + return existing["task_id"] + task_id = f"DLQ-{uuid.uuid4().hex[:12]}" + conn.execute( + """ + INSERT INTO download_tasks + (task_id, contract_id, kind, status, attempts, enqueued_at) + VALUES (?, ?, ?, 'pending', 0, ?) + """, + (task_id, contract_id, kind, datetime.now().isoformat(timespec="seconds")), + ) + return task_id + + +def _read_max_attempts(conn: sqlite3.Connection) -> int: + """读 DOWNLOAD_QUEUE_MAX_ATTEMPTS env（默认 3）。""" + import os + return int(os.environ.get("DOWNLOAD_QUEUE_MAX_ATTEMPTS", "3")) + + +def claim_pending_tasks( + conn: sqlite3.Connection, limit: int, ids_filter: list[str] | None = None, +) -> list[sqlite3.Row]: + """原子领取一批 pending 任务，标记为 running。 + + - limit: 最多领取条数 + - ids_filter: 若给出，仅从这些 task_id 中领 + """ + max_attempts = _read_max_attempts(conn) + if ids_filter: + placeholders = ",".join("?" for _ in ids_filter) + rows = conn.execute( + f"""SELECT * FROM download_tasks + WHERE status = 'pending' AND attempts < ? + AND task_id IN ({placeholders}) + ORDER BY enqueued_at ASC LIMIT ?""", + [max_attempts, *ids_filter, limit], + ).fetchall() + else: + rows = conn.execute( + """SELECT * FROM download_tasks + WHERE status = 'pending' AND attempts < ? + ORDER BY enqueued_at ASC LIMIT ?""", + (max_attempts, limit), + ).fetchall() + if not rows: + return [] + now = datetime.now().isoformat(timespec="seconds") + for r in rows: + conn.execute( + "UPDATE download_tasks SET status = 'running', started_at = ? WHERE task_id = ?", + (now, r["task_id"]), + ) + conn.commit() + return rows + + +def mark_task_success( + conn: sqlite3.Connection, task_id: str, result_path: str, +) -> None: + conn.execute( + """UPDATE download_tasks + SET status = 'success', result_path = ?, finished_at = ? + WHERE task_id = ?""", + (result_path, datetime.now().isoformat(timespec="seconds"), task_id), + ) + + +def mark_task_failed( + conn: sqlite3.Connection, task_id: str, error_msg: str, +) -> None: + """失败一次：attempts+1，未超 max_attempts 则回退 pending，否则标 failed。""" + max_attempts = _read_max_attempts(conn) + now = datetime.now().isoformat(timespec="seconds") + conn.execute( + """UPDATE download_tasks + SET status = CASE WHEN attempts + 1 >= ? THEN 'failed' ELSE 'pending' END, + attempts = attempts + 1, + error_msg = ?, + finished_at = ? + WHERE task_id = ?""", + (max_attempts, error_msg, now, task_id), + ) + + +def list_recent_tasks( + conn: sqlite3.Connection, limit: int = 50 +) -> list[sqlite3.Row]: + return conn.execute( + "SELECT * FROM download_tasks ORDER BY enqueued_at DESC LIMIT ?", + (limit,), + ).fetchall() + + +def count_tasks_by_status(conn: sqlite3.Connection) -> dict[str, int]: + rows = conn.execute( + "SELECT status, COUNT(*) AS n FROM download_tasks GROUP BY status" + ).fetchall() + return {r["status"]: r["n"] for r in rows} diff --git a/scripts/service/config_viewer.py b/scripts/service/config_viewer.py new file mode 100644 index 0000000..7019b99 --- /dev/null +++ b/scripts/service/config_viewer.py @@ -0,0 +1,77 @@ +"""读取并脱敏 .env 配置供 UI 查看（v1.1.18）。 + +仅显示 .env 文件中的非空 key；命中 (TOKEN|SECRET|KEY|PASSWORD) 的字段自动脱敏为 ***。 +""" + +from __future__ import annotations + +import re +from pathlib import Path + +from util.runtime_paths import get_env_path + + +# 匹配任一含 TOKEN / SECRET / KEY / PASSWORD 的 key 名（不区分大小写） +_SECRET_PATTERN = re.compile(r"(TOKEN|SECRET|KEY|PASSWORD)", re.IGNORECASE) + + +def _parse_env_file(path: Path) -> dict[str, str]: + """手动解析 .env（避免依赖 python-dotenv）。 + + - 跳过空行 / 注释（# 开头） + - 支持 KEY=VALUE 格式 + - 自动去除包裹的 "..." 或 '...' + """ + if not path.exists(): + return {} + result: dict[str, str] = {} + with open(path, "r", encoding="utf-8") as f: + for line in f: + line = line.strip() + if not line or line.startswith("#"): + continue + if "=" not in line: + continue + k, _, v = line.partition("=") + k = k.strip() + v = v.strip() + # 去掉行尾注释（不处理引号内的 #，简单实现） + if " #" in v: + v = v.split(" #", 1)[0].strip() + if v.startswith('"') and v.endswith('"') and len(v) >= 2: + v = v[1:-1] + elif v.startswith("'") and v.endswith("'") and len(v) >= 2: + v = v[1:-1] + result[k] = v + return result + + +def _mask_secret(value: str) -> str: + if not value: + return "" + if len(value) <= 4: + return "***" + return value[:2] + "***" + value[-2:] + + +def read_env_config() -> dict: + """读取 .env 中的非空 key，敏感 key 脱敏为 ***。 + + Returns: + { + "env_path": str, + "exists": bool, + "config": {key: value_or_masked_value, ...} + } + """ + path = get_env_path() + parsed = _parse_env_file(path) + out: dict[str, str] = {} + for k, v in sorted(parsed.items()): + if not v: + continue + if _SECRET_PATTERN.search(k): + out[k] = _mask_secret(v) + else: + out[k] = v + return {"env_path": str(path), "exists": path.exists(), "config": out} diff --git a/scripts/service/importer.py b/scripts/service/importer.py new file mode 100644 index 0000000..5b34d4a --- /dev/null +++ b/scripts/service/importer.py @@ -0,0 +1,111 @@ +"""主动扫描 data/input/{contracts,materials}/：CSV → contracts 表，PDF/Word → materials_content 表。 + +对 sandbox 用户尤其重要：HR API 拿不到时，把材料直接丢进 input/materials/ 即可被自动识别。 + +约定： + - input/contracts/*.csv + 列名与 contracts 表一致（contract_id / teacher_name / unit_name / contract_end / ...） + 也兼容中文别名（合同ID / 教职工姓名 / 单位 / 合同到期日 / 工号 / 岗位 / 合同类型） + - input/materials/{contract_id}__{filename}.{pdf|docx|doc} + 文件名 __ 之前是 contract_id；无 __ 时回退 parent dir 名 + 拷贝到 data/materials/ + insert materials_content(parse_method='pending') +""" + +from __future__ import annotations + +import csv +import shutil +import sqlite3 +from pathlib import Path + +from db import repository as repo +from db.migrations import ensure_materials_content_schema +from util.logging_setup import get_skill_logger +from util.runtime_paths import ( + get_data_dir, + get_input_contracts_dir, + get_input_materials_dir, +) + +logger = get_skill_logger() + +MATERIAL_EXTS = {".pdf", ".docx", ".doc"} + + +def _derive_contract_id_from_filename(mat_path: Path) -> str: + """从文件名推导 contract_id。约定：{contract_id}__{filename}.{ext}。""" + parts = mat_path.stem.split("__", 1) + if len(parts) == 2 and parts[0]: + return parts[0] + # 兜底：parent dir 名 + return mat_path.parent.name or "UNKNOWN" + + +def import_input_dir(conn: sqlite3.Connection) -> dict: + """扫描 input 目录 → 入库。返回统计 dict。 + + Returns: + { + "contracts_added": int, + "materials_added": int, + "files_scanned": int, + "errors": [str, ...], + } + """ + ensure_materials_content_schema(conn) + + contracts_dir = get_input_contracts_dir() + materials_dir = get_input_materials_dir() + + contracts_added = 0 + materials_added = 0 + files_scanned = 0 + errors: list[str] = [] + + # 1. CSV → contracts（contract_id 重复 → upsert 覆盖） + for csv_path in sorted(contracts_dir.glob("*.csv")): + files_scanned += 1 + try: + with open(csv_path, "r", encoding="utf-8", newline="") as f: + reader = csv.DictReader(f) + for row in reader: + if not row.get("contract_id") and not row.get("合同ID"): + errors.append(f"{csv_path.name}: row 缺 contract_id 跳过") + continue + repo.upsert_contract_from_csv_row(conn, row) + contracts_added += 1 + except Exception as e: + errors.append(f"{csv_path.name}: {e}") + logger.warning("importer: CSV 解析失败 %s: %s", csv_path, e) + + # 2. PDF/Word → materials_content（parse_method='pending'），文件拷贝到 data/materials/ + materials_target_dir = get_data_dir() / "materials" + materials_target_dir.mkdir(parents=True, exist_ok=True) + for mat_path in sorted(materials_dir.iterdir()): + if not mat_path.is_file() or mat_path.suffix.lower() not in MATERIAL_EXTS: + continue + files_scanned += 1 + try: + contract_id = _derive_contract_id_from_filename(mat_path) + target = materials_target_dir / mat_path.name + if not target.exists(): + shutil.copy2(mat_path, target) + repo.insert_material( + conn, contract_id, mat_path.name, str(target), + None, "pending", + ) + materials_added += 1 + logger.info( + "importer: 入库 %s → contract_id=%s", mat_path.name, contract_id, + ) + except Exception as e: + errors.append(f"{mat_path.name}: {e}") + logger.warning("importer: 材料导入失败 %s: %s", mat_path, e) + + conn.commit() + return { + "contracts_added": contracts_added, + "materials_added": materials_added, + "files_scanned": files_scanned, + "errors": errors, + } diff --git a/scripts/service/logs_viewer.py b/scripts/service/logs_viewer.py new file mode 100644 index 0000000..74a4b5b --- /dev/null +++ b/scripts/service/logs_viewer.py @@ -0,0 +1,86 @@ +"""读取日志文件末尾 N 行（v1.1.18）。 + +按 level 过滤：默认 INFO+。适用于 RotatingFileHandler 落地的日志文件。 +""" + +from __future__ import annotations + +import os +from pathlib import Path + +from util.runtime_paths import get_logs_dir + + +DEFAULT_TAIL = 200 +_LEVEL_ORDER = { + "DEBUG": 10, "INFO": 20, "WARNING": 30, "ERROR": 40, "CRITICAL": 50, +} + + +def _resolve_log_path() -> Path: + """解析日志文件路径。 + + 优先级： + 1. LOG_FILE env（绝对路径直接用，相对路径走 get_logs_dir()） + 2. 默认 get_logs_dir() / LOG_FILE_DEFAULT_NAME + """ + raw = os.environ.get("LOG_FILE", "") + if raw: + p = Path(raw) + if not p.is_absolute(): + p = get_logs_dir() / raw + return p + from util.constants import LOG_FILE_DEFAULT_NAME + return get_logs_dir() / LOG_FILE_DEFAULT_NAME + + +def tail_log_file(tail: int = DEFAULT_TAIL, min_level: str = "INFO") -> dict: + """读日志文件末尾 tail 行，过滤 level >= min_level。 + + Returns: + { + "log_file": str, + "exists": bool, + "total_lines": int, + "returned_lines": int, + "lines": [str, ...], + "min_level": str, + } + """ + log_path = _resolve_log_path() + if not log_path.exists(): + return { + "log_file": str(log_path), + "exists": False, + "total_lines": 0, + "returned_lines": 0, + "lines": [], + "min_level": min_level, + } + + text = log_path.read_text(encoding="utf-8", errors="replace") + all_lines = text.splitlines() + last_n = all_lines[-tail:] if tail > 0 else all_lines + threshold = _LEVEL_ORDER.get(min_level.upper(), 20) + + filtered: list[str] = [] + for line in last_n: + # 格式: "2026-09-15 12:34:56 [alert-contract-expiring-hr] INFO message" + # 取第 4 个 token (INFO/DEBUG/...) + parts = line.split() + if len(parts) >= 4 and parts[1].count(":") >= 2 and parts[3] in _LEVEL_ORDER: + line_level = parts[3] + if _LEVEL_ORDER.get(line_level, 0) >= threshold: + filtered.append(line) + else: + # 非标准行 / 解析不到 level：保留 + filtered.append(line) + + return { + "log_file": str(log_path), + "exists": True, + "total_lines": len(all_lines), + "returned_lines": len(filtered), + "lines": filtered, + "min_level": min_level, + } diff --git a/scripts/service/platform_collector.py b/scripts/service/platform_collector.py index 2c16716..cb1220c 100644 --- a/scripts/service/platform_collector.py +++ b/scripts/service/platform_collector.py @@ -865,11 +865,265 @@ def _persist_contracts(contracts: list[dict]) -> int: 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, diff --git a/scripts/util/constants.py b/scripts/util/constants.py index c812ee2..d24ed88 100644 --- a/scripts/util/constants.py +++ b/scripts/util/constants.py @@ -6,7 +6,7 @@ """ SKILL_SLUG = "alert-contract-expiring-hr" -SKILL_VERSION = "1.1.17" +SKILL_VERSION = "1.1.18" LOG_LOGGER_NAME = "openclaw.skill.alert_contract_expiring_hr" PLATFORM_KIT_MIN_VERSION = "1.2.11" @@ -87,3 +87,13 @@ CARD_APPEAR_WAIT_MS = 3_000 # 「续聘材料包」click 后等卡片出 # dump DUMP_PATH_PARENT_MKDIR = True # _dump_page 是否自动创建父目录 + +# ----- v1.1.18：日志落文件 ----- +LOG_FILE_DEFAULT_NAME = "alert-contract-expiring-hr.log" +LOG_MAX_BYTES_DEFAULT = 5 * 1024 * 1024 # 5MB +LOG_BACKUP_COUNT_DEFAULT = 5 + +# ----- v1.1.18：材料下载队列 ----- +DOWNLOAD_QUEUE_MAX_ATTEMPTS_DEFAULT = 3 +DOWNLOAD_QUEUE_KINDS = ("platform_button", "local_upload") +DOWNLOAD_QUEUE_STATUSES = ("pending", "running", "success", "failed") diff --git a/scripts/util/logging_setup.py b/scripts/util/logging_setup.py index a616916..8d522a7 100644 --- a/scripts/util/logging_setup.py +++ b/scripts/util/logging_setup.py @@ -1,20 +1,35 @@ """logging 配置：降级到 stderr，避免污染 stdout（stdout 留给结构化输出如 JSON）。 优先用 jiangchang_skill_core 提供的 logging；本地无该包时降级到 stderr。 + +v1.1.18：本地降级路径追加 RotatingFileHandler（按 LOG_FILE env 决定是否落文件）。 """ from __future__ import annotations import logging +import logging.handlers +import os import sys +from pathlib import Path -from util.constants import LOG_LOGGER_NAME, SKILL_SLUG, SKILL_VERSION +from util.constants import ( + LOG_LOGGER_NAME, + SKILL_SLUG, + SKILL_VERSION, +) _configured = False def setup_skill_logging() -> logging.Logger: - """初始化 skill logger（幂等）。""" + """初始化 skill logger（幂等）。 + + 本地降级行为（无 jiangchang_skill_core 时）： + - 若 LOG_FILE 已设：追加 RotatingFileHandler（LOG_MAX_BYTES × LOG_BACKUP_COUNT） + - 始终追加 StreamHandler(sys.stderr) + - logger level 取自 LOG_LEVEL（默认 INFO） + """ global _configured logger = logging.getLogger(LOG_LOGGER_NAME) if _configured: @@ -30,15 +45,38 @@ def setup_skill_logging() -> logging.Logger: except ImportError: pass - # 本地降级：基础 stderr handler，UTF-8 safe - handler = logging.StreamHandler(sys.stderr) - handler.setFormatter( - logging.Formatter( - fmt=f"%(asctime)s [{SKILL_SLUG}] %(levelname)s %(message)s", - datefmt="%Y-%m-%d %H:%M:%S", - ) + # logger 级别 + level_name = os.environ.get("LOG_LEVEL", "INFO").upper() + logger.setLevel(getattr(logging, level_name, logging.INFO)) + + formatter = logging.Formatter( + fmt=f"%(asctime)s [{SKILL_SLUG}] %(levelname)s %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", ) - logger.setLevel(logging.INFO) + + # 本地降级：可选 RotatingFileHandler + log_file_str = os.environ.get("LOG_FILE") + if log_file_str: + log_path = Path(log_file_str) + if not log_path.is_absolute(): + from util.runtime_paths import get_logs_dir + + log_path = get_logs_dir() / log_file_str + log_path.parent.mkdir(parents=True, exist_ok=True) + max_bytes = int(os.environ.get("LOG_MAX_BYTES", str(5 * 1024 * 1024))) + backup_count = int(os.environ.get("LOG_BACKUP_COUNT", "5")) + file_handler = logging.handlers.RotatingFileHandler( + log_path, + maxBytes=max_bytes, + backupCount=backup_count, + encoding="utf-8", + ) + file_handler.setFormatter(formatter) + logger.addHandler(file_handler) + + # 始终保留 stderr handler + handler = logging.StreamHandler(sys.stderr) + handler.setFormatter(formatter) logger.addHandler(handler) logger.propagate = False _configured = True @@ -51,3 +89,19 @@ def get_skill_logger() -> logging.Logger: if not _configured: return setup_skill_logging() return logging.getLogger(LOG_LOGGER_NAME) + + +def reset_for_tests() -> None: + """测试用：重置 _configured flag，让 setup_skill_logging 重新初始化。 + + 必须在每个修改日志配置的测试 fixture 里调用，否则 env 改动不会生效。 + """ + global _configured + logger = logging.getLogger(LOG_LOGGER_NAME) + for h in list(logger.handlers): + try: + h.close() + except Exception: + pass + logger.removeHandler(h) + _configured = False diff --git a/scripts/util/runtime_paths.py b/scripts/util/runtime_paths.py index db8600a..28cddc3 100644 --- a/scripts/util/runtime_paths.py +++ b/scripts/util/runtime_paths.py @@ -66,6 +66,30 @@ def get_dispatch_dir() -> Path: return get_data_dir() / "dispatch" +def get_logs_dir() -> Path: + """返回日志目录（RotatingFileHandler 默认落地点）。""" + env_dir = os.environ.get("JIANGCHANG_LOGS_DIR") + if env_dir: + return Path(env_dir) + return get_data_dir() / "logs" + + +def get_input_dir() -> Path: + """返回用户主动投放的输入目录（含 contracts/*.csv + materials/*.{pdf,docx,doc}）。""" + env_dir = os.environ.get("INPUT_DIR") + if env_dir: + return Path(env_dir) + return get_data_dir() / "input" + + +def get_input_contracts_dir() -> Path: + return get_input_dir() / "contracts" + + +def get_input_materials_dir() -> Path: + return get_input_dir() / "materials" + + def get_examples_dir() -> Path: """返回样例数据目录。""" return get_skill_root() / "examples" @@ -78,5 +102,13 @@ def get_templates_dir() -> Path: def ensure_dirs() -> None: """确保所有需要的目录存在。""" - for d in (get_data_dir(), get_output_dir(), get_dispatch_dir()): + for d in ( + get_data_dir(), + get_output_dir(), + get_dispatch_dir(), + get_logs_dir(), + get_input_dir(), + get_input_contracts_dir(), + get_input_materials_dir(), + ): d.mkdir(parents=True, exist_ok=True) diff --git a/tests/run_tests.py b/tests/run_tests.py index 4101d88..b4cfa58 100644 --- a/tests/run_tests.py +++ b/tests/run_tests.py @@ -241,7 +241,30 @@ def test_data_management_actions() -> tuple[bool, str]: return False, f"{action_id} 未绑定合同台账" if action.get("entrypoint", {}).get("command") != action_id: return False, f"{action_id} CLI command 映射错误" - return True, "5 个数据管理动作均绑定合同台账并映射到 CLI" + + # v1.1.18: download-materials 是 async form action，绑定 download_tasks 队列表 + dl = by_id.get("download-materials") + if not dl: + return False, "缺少 download-materials 动作" + if dl.get("executionProfile") != "async": + return False, f"download-materials 应为 async，实际 {dl.get('executionProfile')}" + if dl.get("interaction") != "form": + return False, f"download-materials 应为 form interaction" + if "download_tasks" not in dl.get("bind", {}).get("tables", []): + return False, f"download-materials 未绑定 download_tasks" + schema_props = dl.get("inputSchema", {}).get("properties", {}) + if "itemIds" not in schema_props or "pickCount" not in schema_props: + return False, f"download-materials inputSchema 缺 itemIds/pickCount: {schema_props}" + + # v1.1.18: show-config / logs-list 是 sync 运维动作 + for sync_id in ("show-config", "logs-list"): + a = by_id.get(sync_id) + if not a: + return False, f"缺少 {sync_id} 动作" + if a.get("executionProfile") != "sync": + return False, f"{sync_id} 应为 sync" + + return True, "5 个数据管理动作 + download-materials (async form) + show-config/logs-list (sync) 全部就绪" def test_collect_platform_defaults_to_browser() -> tuple[bool, str]: @@ -741,6 +764,643 @@ def test_parse_alert_red_orange_yellow_unknown() -> tuple[bool, str]: return True, "四级映射 OK" +# ---------- v1.1.18：日志 / 路径 / 输入目录 / 队列 / 配置 / 日志查看 ---------- + + +def test_logging_rotating_file_handler() -> tuple[bool, str]: + """LOG_FILE 已设时，setup_skill_logging 应追加 RotatingFileHandler。""" + import importlib + from util.runtime_paths import get_logs_dir + + log_path = get_logs_dir() / "test-rotate.log" + + os.environ["LOG_FILE"] = str(log_path) + os.environ.pop("LOG_MAX_BYTES", None) + os.environ.pop("LOG_BACKUP_COUNT", None) + os.environ.pop("LOG_LEVEL", None) + import util.logging_setup as ls + importlib.reload(ls) + ls.reset_for_tests() + + logger = ls.setup_skill_logging() + logger.info("hello rotating") + + found = [h for h in logger.handlers if h.__class__.__name__ == "RotatingFileHandler"] + if not found: + ls.reset_for_tests() + os.environ.pop("LOG_FILE", None) + return False, "未发现 RotatingFileHandler" + if not log_path.exists(): + ls.reset_for_tests() + os.environ.pop("LOG_FILE", None) + return False, f"日志文件未生成: {log_path}" + content = log_path.read_text(encoding="utf-8") + if "hello rotating" not in content: + ls.reset_for_tests() + os.environ.pop("LOG_FILE", None) + return False, f"日志内容缺 'hello rotating': {content[:200]}" + + ls.reset_for_tests() + os.environ.pop("LOG_FILE", None) + importlib.reload(ls) + return True, "RotatingFileHandler 写入成功" + + +def test_logging_fallback_when_log_file_unset() -> tuple[bool, str]: + """LOG_FILE 未设时，logger 只有 StreamHandler（行为等价 v1.1.17）。""" + import importlib + + os.environ.pop("LOG_FILE", None) + os.environ.pop("LOG_LEVEL", None) + import util.logging_setup as ls + importlib.reload(ls) + ls.reset_for_tests() + + logger = ls.setup_skill_logging() + file_handlers = [h for h in logger.handlers if "File" in h.__class__.__name__] + if file_handlers: + ls.reset_for_tests() + return False, f"应有 0 个 FileHandler，实际 {len(file_handlers)}" + + ls.reset_for_tests() + return True, "无 LOG_FILE 时无 FileHandler（兼容 v1.1.17）" + + +def test_logging_level_env() -> tuple[bool, str]: + """LOG_LEVEL=DEBUG 应让 logger.level == logging.DEBUG。""" + import importlib + import logging + + os.environ["LOG_LEVEL"] = "DEBUG" + os.environ.pop("LOG_FILE", None) + import util.logging_setup as ls + importlib.reload(ls) + ls.reset_for_tests() + + logger = ls.setup_skill_logging() + got = logger.level + if got != logging.DEBUG: + ls.reset_for_tests() + os.environ.pop("LOG_LEVEL", None) + return False, f"logger.level={got} (want {logging.DEBUG})" + + ls.reset_for_tests() + os.environ.pop("LOG_LEVEL", None) + return True, "LOG_LEVEL=DEBUG 生效" + + +def test_logs_dir_default_under_data() -> tuple[bool, str]: + """JIANGCHANG_LOGS_DIR 未设时，get_logs_dir() = get_data_dir() / 'logs'。""" + os.environ.pop("JIANGCHANG_LOGS_DIR", None) + os.environ.pop("LOG_FILE", None) + import util.logging_setup as ls # 触发 runtime_paths 副作用 + ls.reset_for_tests() + + from util.runtime_paths import get_logs_dir, get_data_dir + expected = get_data_dir() / "logs" + if get_logs_dir() != expected: + return False, f"get_logs_dir()={get_logs_dir()} != {expected}" + if not expected.exists(): + return False, f"ensure_dirs 没创建 {expected}" + return True, f"get_logs_dir={expected}" + + +def _isolated_input_dirs(monkeypatch_env: dict[str, str | None]) -> tuple[Path, Path]: + """把 INPUT_DIR 重定向到 tmpdir，返回 (input_root, materials_target) 的 Path。 + + 不会污染仓库 data/input/。 + """ + import tempfile + tmp = Path(tempfile.mkdtemp(prefix="alert-hr-importer-test-")) + input_root = tmp / "input" + (input_root / "contracts").mkdir(parents=True) + (input_root / "materials").mkdir(parents=True) + monkeypatch_env["INPUT_DIR"] = str(input_root) + os.environ["INPUT_DIR"] = str(input_root) # 关键：进程 env 也得设，否则 runtime_paths 读不到 + return input_root, tmp + + +def _restore_input_env(saved: dict[str, str | None]) -> None: + for k, v in saved.items(): + if v is None: + os.environ.pop(k, None) + else: + os.environ[k] = v + + +def test_importer_contracts_csv_upsert() -> tuple[bool, str]: + """input/contracts/*.csv → contracts 表 upsert。""" + saved = {k: os.environ.get(k) for k in ("INPUT_DIR",)} + try: + input_root, tmp = _isolated_input_dirs(saved) + contracts_dir = input_root / "contracts" + csv_path = contracts_dir / "batch1.csv" + csv_path.write_text( + "contract_id,teacher_name,unit_name,contract_end,employee_id\n" + "C001,张三,物理学院,2026-04-15,T001\n" + "C002,李四,计算机学院,2026-08-31,T002\n", + encoding="utf-8", + ) + + conn = _make_conn() + try: + from service.importer import import_input_dir + result = import_input_dir(conn) + + if result["contracts_added"] != 2: + return False, f"contracts_added={result['contracts_added']} (want 2)" + if result["errors"]: + return False, f"errors={result['errors']}" + + rows = conn.execute( + "SELECT contract_id, teacher_name, unit_name, contract_end " + "FROM contracts ORDER BY contract_id" + ).fetchall() + if len(rows) != 2: + return False, f"DB 中合同数={len(rows)} (want 2)" + if rows[0]["teacher_name"] != "张三": + return False, f"row[0].teacher_name={rows[0]['teacher_name']}" + return True, f"contracts_added={result['contracts_added']}" + finally: + conn.close() + shutil.rmtree(tmp, ignore_errors=True) + finally: + _restore_input_env(saved) + + +def test_importer_materials_pdf_enqueue() -> tuple[bool, str]: + """input/materials/{cid}__{f}.pdf → materials_content (parse_method='pending')。""" + saved = {k: os.environ.get(k) for k in ("INPUT_DIR",)} + try: + input_root, tmp = _isolated_input_dirs(saved) + materials_dir = input_root / "materials" + pdf_path = materials_dir / "RPA-C001__考核表.pdf" + pdf_path.write_bytes(b"%PDF-1.4 fake test file") + + conn = _make_conn() + try: + # 先插入 contract（materials_content FK 到 contracts.contract_id） + 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) + VALUES ('RPA-C001', 'T001', '张三', 'U001', '物理学院', + '教授', '固定期限', '2024-01-01', '2026-04-15', + '2024-04-15', '可续聘')""" + ) + conn.commit() + + from service.importer import import_input_dir + result = import_input_dir(conn) + + if result["materials_added"] != 1: + return False, f"materials_added={result['materials_added']} (want 1) errors={result['errors']}" + + row = conn.execute( + "SELECT contract_id, parse_method FROM materials_content " + "WHERE contract_id = 'RPA-C001'" + ).fetchone() + if not row: + return False, "materials_content 未写入" + if row["parse_method"] != "pending": + return False, f"parse_method={row['parse_method']} (want pending)" + return True, "PDF 入库 parse_method=pending" + finally: + conn.close() + shutil.rmtree(tmp, ignore_errors=True) + finally: + _restore_input_env(saved) + + +def test_importer_idempotent() -> tuple[bool, str]: + """同一 CSV 跑两次，第二次 contracts_added=0（upsert 幂等）。""" + saved = {k: os.environ.get(k) for k in ("INPUT_DIR",)} + try: + input_root, tmp = _isolated_input_dirs(saved) + (input_root / "contracts" / "x.csv").write_text( + "contract_id,teacher_name\nC100,重复测试\n", encoding="utf-8" + ) + + conn = _make_conn() + try: + from service.importer import import_input_dir + r1 = import_input_dir(conn) + r2 = import_input_dir(conn) + if r1["contracts_added"] != 1: + return False, f"first={r1['contracts_added']} (want 1)" + if r2["contracts_added"] != 1: + # 注意：CSV 仍按行 upsert，第 2 次相当于 1 次 UPDATE，contracts_added 仍计数 + return False, f"second={r2['contracts_added']} (want 1)" + n = conn.execute("SELECT COUNT(*) AS n FROM contracts").fetchone()["n"] + if n != 1: + return False, f"DB count={n} (want 1，幂等)" + return True, "幂等 OK：DB count=1" + finally: + conn.close() + shutil.rmtree(tmp, ignore_errors=True) + finally: + _restore_input_env(saved) + + +def test_importer_empty_input_dir() -> tuple[bool, str]: + """空 input 目录 → no-op，返回全 0。""" + saved = {k: os.environ.get(k) for k in ("INPUT_DIR",)} + try: + input_root, tmp = _isolated_input_dirs(saved) + + conn = _make_conn() + try: + from service.importer import import_input_dir + result = import_input_dir(conn) + if result["contracts_added"] or result["materials_added"]: + return False, f"空目录应返回全 0，实际 {result}" + if result["files_scanned"] != 0: + return False, f"files_scanned={result['files_scanned']} (want 0)" + return True, "空目录 no-op" + finally: + conn.close() + shutil.rmtree(tmp, ignore_errors=True) + finally: + _restore_input_env(saved) + + +def test_download_tasks_enqueue_idempotent() -> tuple[bool, str]: + """enqueue 同一合同两次：第 2 次返回已有 task_id，DB 仅 1 行。""" + from db import repository as repo + + conn = _make_conn() + try: + 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) + VALUES ('C100', 'T100', 'X', 'U1', 'Y', '', '', '2024-01-01', + '2026-04-15', '2024-04-15', '可续聘')""" + ) + conn.commit() + + t1 = repo.enqueue_download_task(conn, "C100", "platform_button") + t2 = repo.enqueue_download_task(conn, "C100", "platform_button") + if t1 != t2: + return False, f"t1={t1} t2={t2} (应相同)" + n = conn.execute( + "SELECT COUNT(*) AS n FROM download_tasks WHERE contract_id = 'C100'" + ).fetchone()["n"] + if n != 1: + return False, f"DB count={n} (want 1)" + return True, f"idempotent: task_id={t1}" + finally: + conn.close() + + +def test_download_tasks_claim_atomic() -> tuple[bool, str]: + """claim_pending_tasks：第一次拿 N 个，第二次拿 0 个（被抢光）。""" + from db import repository as repo + + conn = _make_conn() + try: + # 准备 5 份合同 + for i in range(5): + conn.execute( + f"""INSERT INTO contracts + (contract_id, employee_id, teacher_name, unit_id, unit_name, + position, contract_type, contract_start, contract_end, + trial_period_end, status) + VALUES ('C{i}', 'T{i}', 'X', 'U{i}', 'Y', '', '', + '2024-01-01', '2026-04-15', '2024-04-15', '可续聘')""" + ) + conn.commit() + for i in range(5): + repo.enqueue_download_task(conn, f"C{i}", "platform_button") + conn.commit() + + first = repo.claim_pending_tasks(conn, limit=5) + if len(first) != 5: + return False, f"第一次领到 {len(first)} (want 5)" + # claim 返回的 Row 是查询时刻的快照（UPDATE 在 fetchall 之后才发）； + # 重新查 DB 验证 status 已变 running + for t in first: + cur = conn.execute( + "SELECT status FROM download_tasks WHERE task_id = ?", + (t["task_id"],), + ).fetchone() + if cur["status"] != "running": + return False, f"task {t['task_id']} DB status={cur['status']} (want running)" + + second = repo.claim_pending_tasks(conn, limit=5) + if len(second) != 0: + return False, f"第二次领到 {len(second)} (want 0)" + return True, "claim 原子：5+0" + finally: + conn.close() + + +def test_download_tasks_mark_success_and_failed() -> tuple[bool, str]: + """mark_success: status=success + result_path; mark_failed 计数 + 重试状态切换。""" + from db import repository as repo + + conn = _make_conn() + try: + 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) + VALUES ('C200', 'T200', 'X', 'U1', 'Y', '', '', '2024-01-01', + '2026-04-15', '2024-04-15', '可续聘')""" + ) + conn.commit() + t = repo.enqueue_download_task(conn, "C200", "platform_button") + + # 1. 标记 success + repo.mark_task_success(conn, t, "sandbox://hr/contracts/C200") + conn.commit() + row = conn.execute( + "SELECT status, result_path FROM download_tasks WHERE task_id = ?", (t,) + ).fetchone() + if row["status"] != "success": + return False, f"success 后 status={row['status']}" + if row["result_path"] != "sandbox://hr/contracts/C200": + return False, f"result_path={row['result_path']}" + + # 2. 失败 ≤ max_attempts：status 回退 pending + t2 = repo.enqueue_download_task(conn, "C200", "platform_button_v2") + repo.mark_task_failed(conn, t2, "first attempt failed") + conn.commit() + row2 = conn.execute( + "SELECT status, attempts FROM download_tasks WHERE task_id = ?", (t2,) + ).fetchone() + if row2["status"] != "pending": + return False, f"第 1 次失败应 pending，实际 {row2['status']}" + if row2["attempts"] != 1: + return False, f"attempts={row2['attempts']} (want 1)" + + # 3. 失败 ≥ max_attempts：status=failed + repo.mark_task_failed(conn, t2, "second attempt failed") + repo.mark_task_failed(conn, t2, "third attempt failed") + conn.commit() + row3 = conn.execute( + "SELECT status, attempts FROM download_tasks WHERE task_id = ?", (t2,) + ).fetchone() + if row3["status"] != "failed": + return False, f"第 3 次失败应 failed，实际 {row3['status']}" + return True, "mark_success/mark_failed 全 OK" + finally: + conn.close() + + +def test_download_tasks_count_by_status() -> tuple[bool, str]: + """count_tasks_by_status 返回 {pending: N, success: M, ...}。""" + from db import repository as repo + + conn = _make_conn() + try: + for i in range(3): + conn.execute( + f"""INSERT INTO contracts + (contract_id, employee_id, teacher_name, unit_id, unit_name, + position, contract_type, contract_start, contract_end, + trial_period_end, status) + VALUES ('CT{i}', 'T{i}', 'X', 'U{i}', 'Y', '', '', + '2024-01-01', '2026-04-15', '2024-04-15', '可续聘')""" + ) + conn.commit() + + t0 = repo.enqueue_download_task(conn, "CT0", "platform_button") + t1 = repo.enqueue_download_task(conn, "CT1", "platform_button") + repo.enqueue_download_task(conn, "CT2", "platform_button") + conn.commit() + + repo.mark_task_success(conn, t0, "x") + repo.mark_task_failed(conn, t1, "err") + # 再失败 2 次让 t1 进 failed + repo.mark_task_failed(conn, t1, "err") + repo.mark_task_failed(conn, t1, "err") + conn.commit() + + counts = repo.count_tasks_by_status(conn) + if counts.get("success", 0) < 1: + return False, f"success 计数 < 1: {counts}" + if counts.get("failed", 0) < 1: + return False, f"failed 计数 < 1: {counts}" + if counts.get("pending", 0) < 1: + return False, f"pending 计数 < 1: {counts}" + return True, f"counts={counts}" + finally: + conn.close() + + +# ---------- v1.1.18: show-config / logs-list ---------- + + +def _isolated_env_file(tmp_env_path: Path) -> None: + """把 .env 路径重定向到 tmpdir；不影响其他 env。""" + import util.runtime_paths as rp + + # runtime_paths.get_env_path() 总是返回 get_skill_root() / ".env" + # 这里通过 monkey-patch 函数对象实现隔离 + rp.get_env_path = lambda: tmp_env_path # type: ignore[assignment] + + +def test_show_config_masks_tokens() -> tuple[bool, str]: + """含 TOKEN 的 key 自动脱敏。""" + import tempfile + from pathlib import Path + tmp = Path(tempfile.mkdtemp(prefix="alert-hr-config-")) + env_path = tmp / ".env" + env_path.write_text( + "WECOM_API_TOKEN=abc123def456\n" + "HR_API_TOKEN=xx\n" + "LOG_LEVEL=INFO\n", + encoding="utf-8", + ) + + import service.config_viewer as cv + orig_get_env_path = cv.get_env_path + cv.get_env_path = lambda: env_path # type: ignore[assignment] + try: + result = cv.read_env_config() + cfg = result["config"] + if cfg.get("WECOM_API_TOKEN") != "ab***56": + return False, f"WECOM_API_TOKEN={cfg.get('WECOM_API_TOKEN')} (want ab***56)" + if cfg.get("HR_API_TOKEN") != "***": + return False, f"HR_API_TOKEN={cfg.get('HR_API_TOKEN')} (want *** for ≤4 chars)" + if cfg.get("LOG_LEVEL") != "INFO": + return False, f"LOG_LEVEL={cfg.get('LOG_LEVEL')} (want INFO)" + return True, "tokens masked OK" + finally: + cv.get_env_path = orig_get_env_path # type: ignore[assignment] + shutil.rmtree(tmp, ignore_errors=True) + + +def test_show_config_skips_empty() -> tuple[bool, str]: + """.env 中空 value 的 key 不出现在 config 中。""" + import tempfile + from pathlib import Path + tmp = Path(tempfile.mkdtemp(prefix="alert-hr-config-")) + env_path = tmp / ".env" + env_path.write_text( + "DB_PATH=\nLOG_LEVEL=INFO\n", encoding="utf-8", + ) + + import service.config_viewer as cv + orig = cv.get_env_path + cv.get_env_path = lambda: env_path # type: ignore[assignment] + try: + cfg = cv.read_env_config()["config"] + if "DB_PATH" in cfg: + return False, f"DB_PATH 应该跳过，仍在 cfg: {cfg}" + if cfg.get("LOG_LEVEL") != "INFO": + return False, f"LOG_LEVEL={cfg.get('LOG_LEVEL')}" + return True, "空 value 跳过 OK" + finally: + cv.get_env_path = orig # type: ignore[assignment] + shutil.rmtree(tmp, ignore_errors=True) + + +def test_show_config_skips_comments() -> tuple[bool, str]: + """# 注释行不计入 config。""" + import tempfile + from pathlib import Path + tmp = Path(tempfile.mkdtemp(prefix="alert-hr-config-")) + env_path = tmp / ".env" + env_path.write_text( + "# 这是注释\n" + " # 也是注释\n" + "LOG_LEVEL=INFO\n", + encoding="utf-8", + ) + + import service.config_viewer as cv + orig = cv.get_env_path + cv.get_env_path = lambda: env_path # type: ignore[assignment] + try: + cfg = cv.read_env_config()["config"] + if len(cfg) != 1 or "LOG_LEVEL" not in cfg: + return False, f"应仅 1 项 LOG_LEVEL，实际 {list(cfg.keys())}" + return True, "注释跳过 OK" + finally: + cv.get_env_path = orig # type: ignore[assignment] + shutil.rmtree(tmp, ignore_errors=True) + + +def test_show_config_handles_quotes() -> tuple[bool, str]: + """.env 中带双引号/单引号的值自动去引号。""" + import tempfile + from pathlib import Path + tmp = Path(tempfile.mkdtemp(prefix="alert-hr-config-")) + env_path = tmp / ".env" + env_path.write_text( + 'HR_API_BASE_URL="https://hr.example.edu/api/v1"\n' + "GREETING='hello world'\n" + "NOTE=no_quotes\n", + encoding="utf-8", + ) + + import service.config_viewer as cv + orig = cv.get_env_path + cv.get_env_path = lambda: env_path # type: ignore[assignment] + try: + cfg = cv.read_env_config()["config"] + if cfg.get("HR_API_BASE_URL") != "https://hr.example.edu/api/v1": + return False, f"双引号未去: {cfg.get('HR_API_BASE_URL')}" + if cfg.get("GREETING") != "hello world": + return False, f"单引号未去: {cfg.get('GREETING')}" + if cfg.get("NOTE") != "no_quotes": + return False, f"无引号: {cfg.get('NOTE')}" + return True, "引号处理 OK" + finally: + cv.get_env_path = orig # type: ignore[assignment] + shutil.rmtree(tmp, ignore_errors=True) + + +def test_logs_viewer_nonexistent_file() -> tuple[bool, str]: + """日志文件不存在时，返回 exists=False，lines=[]。""" + import tempfile + from pathlib import Path + tmp = Path(tempfile.mkdtemp(prefix="alert-hr-logs-")) + fake_log = tmp / "no-such-file.log" + + os.environ["LOG_FILE"] = str(fake_log) + try: + from service.logs_viewer import tail_log_file + result = tail_log_file(tail=50) + if result["exists"]: + return False, "exists 应为 False" + if result["lines"]: + return False, f"lines 应为空: {result['lines']}" + if result["returned_lines"] != 0: + return False, f"returned_lines={result['returned_lines']}" + return True, "non-existent file OK" + finally: + os.environ.pop("LOG_FILE", None) + shutil.rmtree(tmp, ignore_errors=True) + + +def test_logs_viewer_tail_and_filter() -> tuple[bool, str]: + """tail + level 过滤：INFO 默认只显示 INFO/WARNING/ERROR。""" + import tempfile + from pathlib import Path + tmp = Path(tempfile.mkdtemp(prefix="alert-hr-logs-")) + log_path = tmp / "tail-test.log" + lines: list[str] = [] + for i in range(300): + if i % 3 == 0: + lines.append(f"2026-09-15 12:34:5{i%10} [alert-contract-expiring-hr] INFO 第{i}条") + elif i % 3 == 1: + lines.append(f"2026-09-15 12:34:5{i%10} [alert-contract-expiring-hr] WARNING 第{i}条") + else: + lines.append(f"2026-09-15 12:34:5{i%10} [alert-contract-expiring-hr] DEBUG 第{i}条") + log_path.write_text("\n".join(lines), encoding="utf-8") + + os.environ["LOG_FILE"] = str(log_path) + try: + from service.logs_viewer import tail_log_file + # INFO+：最后 50 行中含 INFO + WARNING，去掉 DEBUG + # 300 行每 3 行一循环：i%3==0 INFO, i%3==1 WARN, i%3==2 DEBUG + # 最后 50 行是 i=250..299：第 250=INFO 251=WARN 252=DEBUG ... 共 50 行 + # 分布：i%3==0 → 17 (250, 253, ..., 299 但 299%3=0)... 实际 250..299 中 i%3==0 有 17 个 + result_info = tail_log_file(tail=50, min_level="INFO") + if result_info["returned_lines"] != 33: + return False, f"INFO+ returned_lines={result_info['returned_lines']} (want 33)" + if any(" DEBUG " in ln for ln in result_info["lines"]): + return False, "INFO 过滤应排除 DEBUG 行" + + # WARNING+：仅 WARNING + result_warn = tail_log_file(tail=50, min_level="WARNING") + if result_warn["returned_lines"] != 17: + return False, f"WARNING+ returned_lines={result_warn['returned_lines']} (want 17)" + if not all(" WARNING " in ln for ln in result_warn["lines"]): + return False, "WARNING 过滤应只剩 WARNING 行" + + # DEBUG：全部 50 行都返回 + result_debug = tail_log_file(tail=50, min_level="DEBUG") + if result_debug["returned_lines"] != 50: + return False, f"DEBUG returned_lines={result_debug['returned_lines']} (want 50)" + return True, f"tail+filter OK: DEBUG+={result_debug['returned_lines']}, INFO+={result_info['returned_lines']}, WARN+={result_warn['returned_lines']}" + finally: + os.environ.pop("LOG_FILE", None) + shutil.rmtree(tmp, ignore_errors=True) + + +def test_logs_viewer_resolves_relative_path() -> tuple[bool, str]: + """LOG_FILE 设相对路径时，解析到 get_logs_dir() 下。""" + from util.runtime_paths import get_logs_dir + os.environ["LOG_FILE"] = "relative-test.log" + os.environ.pop("JIANGCHANG_LOGS_DIR", None) + try: + from service.logs_viewer import _resolve_log_path + got = _resolve_log_path() + expected = get_logs_dir() / "relative-test.log" + if got != expected: + return False, f"_resolve_log_path={got} (want {expected})" + return True, f"resolved to {got}" + finally: + os.environ.pop("LOG_FILE", None) + + # ---------- 入口 ---------- def main() -> int: @@ -785,6 +1445,25 @@ def main() -> int: _run("platform_collector/parse_days_positive", test_parse_days_positive) _run("platform_collector/parse_days_negative", test_parse_days_negative) _run("platform_collector/parse_alert_levels", test_parse_alert_red_orange_yellow_unknown) + _run("logging/rotating_file_handler", test_logging_rotating_file_handler) + _run("logging/fallback_when_log_file_unset", test_logging_fallback_when_log_file_unset) + _run("logging/level_env", test_logging_level_env) + _run("logging/logs_dir_default", test_logs_dir_default_under_data) + _run("importer/contracts_csv_upsert", test_importer_contracts_csv_upsert) + _run("importer/materials_pdf_enqueue", test_importer_materials_pdf_enqueue) + _run("importer/idempotent", test_importer_idempotent) + _run("importer/empty_input_dir", test_importer_empty_input_dir) + _run("download_tasks/enqueue_idempotent", test_download_tasks_enqueue_idempotent) + _run("download_tasks/claim_atomic", test_download_tasks_claim_atomic) + _run("download_tasks/mark_success_and_failed", test_download_tasks_mark_success_and_failed) + _run("download_tasks/count_by_status", test_download_tasks_count_by_status) + _run("show_config/masks_tokens", test_show_config_masks_tokens) + _run("show_config/skips_empty", test_show_config_skips_empty) + _run("show_config/skips_comments", test_show_config_skips_comments) + _run("show_config/handles_quotes", test_show_config_handles_quotes) + _run("logs_viewer/nonexistent_file", test_logs_viewer_nonexistent_file) + _run("logs_viewer/tail_and_filter", test_logs_viewer_tail_and_filter) + _run("logs_viewer/resolves_relative_path", test_logs_viewer_resolves_relative_path) total = len(_RESULTS) passed = sum(1 for _, ok, _ in _RESULTS if ok)