"""CLI 命令分发：argparse 子命令 → 调用 service 层。 v1.1.1：11 个子命令（在 9 个基础上新增） - collect-from-platform：Playwright RPA 抓 SmartCampus HR 平台数据 - parse-materials：解析已下载的材料 PDF/Word → 文本 原有 9 个： - health / version / config-path / init-db - scan-expiring-contracts / generate-renewal-task - dispatch-task-notice / mark-contract-renewed - render-html """ from __future__ import annotations import argparse import json import sys from typing import Callable, List, Optional from db.connection import get_connection, init_db from db import repository as repo from service import dispatcher, html_renderer, material_parser, platform_collector, scanner, task_generator from util.constants import DEFAULT_THRESHOLDS, HUMAN_WAIT_TIMEOUT, SKILL_VERSION from util.logging_setup import get_skill_logger, setup_skill_logging from util.runtime_paths import ( get_db_path, get_dispatch_dir, get_env_path, get_output_dir, ) logger = get_skill_logger() # ---------- 命令实现 ---------- def cmd_health(_args: argparse.Namespace) -> int: """健康检查。""" setup_skill_logging() db_path = get_db_path() db_exists = db_path.exists() db_writable = False error_msg = None try: conn = get_connection() try: conn.execute("SELECT 1").fetchone() db_writable = True finally: conn.close() except Exception as e: error_msg = str(e) output_dir = get_output_dir() dispatch_dir = get_dispatch_dir() checks = [ ("python_version", f"{sys.version_info.major}.{sys.version_info.minor}.{sys.version_info.micro}", True), ("db_path", str(db_path), True), ("db_exists", str(db_exists), db_exists), ("db_writable", str(db_writable), db_writable), ("output_dir", str(output_dir), True), ("dispatch_dir", str(dispatch_dir), True), ] if db_exists and db_writable: try: conn = get_connection() try: counts = { "contracts": conn.execute("SELECT COUNT(*) AS n FROM contracts").fetchone()["n"], "performance_records": conn.execute("SELECT COUNT(*) AS n FROM performance_records").fetchone()["n"], "alerts": conn.execute("SELECT COUNT(*) AS n FROM alerts").fetchone()["n"], "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"], } finally: conn.close() checks.append(("seed_loaded", f"contracts={counts['contracts']} perf={counts['performance_records']}", counts["contracts"] > 0)) except Exception as e: checks.append(("seed_check", f"ERROR: {e}", False)) all_ok = all(c[2] for c in checks) for name, val, ok in checks: marker = "OK " if ok else "FAIL" print(f"[{marker}] {name}: {val}") if error_msg: print(f"[FAIL] error: {error_msg}") if all_ok: print("OK: health check passed") return 0 print("ERROR: health check failed") return 1 def cmd_version(_args: argparse.Namespace) -> int: print(f"{SKILL_VERSION}") return 0 def cmd_config_path(_args: argparse.Namespace) -> int: print(get_env_path()) return 0 def cmd_init_db(_args: argparse.Namespace) -> int: counts = init_db() print(f"OK: init-db complete") for k, v in counts.items(): print(f" {k}: {v} rows") return 0 def cmd_scan(args: argparse.Namespace) -> int: thresholds = _parse_thresholds(args.thresholds) conn = get_connection() try: result = scanner.scan(conn, thresholds=thresholds) finally: conn.close() print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 def cmd_generate_task(args: argparse.Namespace) -> int: contract_id = args.contract_id task_id = args.task_id conn = get_connection() try: result = task_generator.regenerate_task_for_contract_id(conn, contract_id, task_id) conn.commit() except ValueError as e: print(f"ERROR: {e}") return 2 finally: conn.close() print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 def cmd_dispatch(args: argparse.Namespace) -> int: conn = get_connection() try: result = dispatcher.dispatch(conn, args.unit_id, args.month or "") finally: conn.close() print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 def cmd_mark_renewed(args: argparse.Namespace) -> int: conn = get_connection() try: from db import repository as repo contract = repo.get_contract(conn, args.contract_id) if contract is None: print(f"ERROR: contract_id={args.contract_id!r} 不存在") return 2 with conn: n = repo.update_contract_status(conn, args.contract_id, "已续签", args.material_id) closed = repo.close_open_alerts_for_contract(conn, args.contract_id) print(json.dumps({ "contract_id": args.contract_id, "status": "已续签", "material_id": args.material_id, "updated_rows": n, "closed_alerts": closed, }, ensure_ascii=False, indent=2)) finally: conn.close() return 0 def cmd_render_html(args: argparse.Namespace) -> int: conn = get_connection() try: output_path = html_renderer.render_task_html(conn, args.task_id) except ValueError as e: print(f"ERROR: {e}") return 2 except FileNotFoundError as e: print(f"ERROR: {e}") return 3 finally: conn.close() print(json.dumps({ "task_id": args.task_id, "output_path": str(output_path), "size_bytes": output_path.stat().st_size, }, ensure_ascii=False, indent=2)) return 0 def cmd_collect_platform(args: argparse.Namespace) -> int: """Playwright RPA 抓 SmartCampus HR 平台数据 → 落库。 首次需用户手工登录保存 cookie（默认 headless=False 便于交互）； 后续可用 headless=True 复用 profile。 退出码： 0 = 成功（即使抓到 0 行也返回 0，错误信息在 output["error"]） 1 = 通用异常 2 = RPA 结构化错误（LoginTimeout / MissingBrowser / TableNotFound 等） """ try: result = platform_collector.collect_all( limit=args.limit, headless=not args.show_browser, human_wait_timeout=args.human_wait_timeout or HUMAN_WAIT_TIMEOUT, ) except platform_collector.RPAError as e: # v1.1.6：结构化错误（ERROR:LOGIN_TIMEOUT 等）→ exit 2 print(json.dumps({ "error": e.code, "detail": str(e), "fetched_rows": 0, "saved_contracts": 0, "downloaded_materials": 0, "contracts": [], }, 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_parse_materials(args: argparse.Namespace) -> int: """对已落库的材料 PDF/Word 跑文本提取（pypdf / python-docx / OCR）。 --contract-id C001: 仅处理该合同 不传：处理所有 parse_method='pending' 或 extracted_text IS NULL 的材料 """ conn = get_connection() try: if args.contract_id: rows = repo.list_materials_for_contract(conn, args.contract_id) else: rows = repo.list_pending_materials(conn) results: list[dict] = [] for r in rows: from pathlib import Path fp = Path(r["file_path"]) text, method = material_parser.extract_text(fp) if text: repo.insert_material( conn, r["contract_id"], r["file_name"], r["file_path"], text, method, ) else: repo.insert_material( conn, r["contract_id"], r["file_name"], r["file_path"], "", "failed", ) results.append({ "contract_id": r["contract_id"], "file_name": r["file_name"], "parse_method": method, "text_length": len(text), }) conn.commit() print(json.dumps({ "processed": len(results), "results": results, }, ensure_ascii=False, indent=2)) return 0 finally: conn.close() # ---------- argparse ---------- def _parse_thresholds(s: str) -> tuple[int, int, int]: parts = [int(p.strip()) for p in s.split(",")] if len(parts) != 3 or sorted(parts, reverse=True) != parts: raise argparse.ArgumentTypeError( f"thresholds 必须是 3 个整数（黄,橙,红）且从大到小，例如 90,60,30；收到 {s!r}" ) return (parts[0], parts[1], parts[2]) def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="main.py", description=f"合同到期智能预警与续聘辅助 v{SKILL_VERSION}", ) sub = p.add_subparsers(dest="cmd", required=True) sub.add_parser("health", help="健康检查（Python / DB / 目录）").set_defaults(handler=cmd_health) sub.add_parser("version", help="查看版本").set_defaults(handler=cmd_version) sub.add_parser("config-path", help="查看 .env 路径").set_defaults(handler=cmd_config_path) sub.add_parser("init-db", help="建表 + 加载样例数据").set_defaults(handler=cmd_init_db) sp = sub.add_parser( "scan-expiring-contracts", help="扫描即将到期合同（async/cron）", ) sp.add_argument( "--thresholds", default="90,60,30", help="三级预警阈值（黄,橙,红）从大到小，默认 90,60,30", ) sp.set_defaults(handler=cmd_scan) sp = sub.add_parser( "generate-renewal-task", help="为指定合同生成续聘任务书", ) sp.add_argument("--contract-id", required=True, help="合同 ID") sp.add_argument("--task-id", default=None, help="可选：指定 task_id（默认自动生成）") sp.set_defaults(handler=cmd_generate_task) sp = sub.add_parser( "dispatch-task-notice", help="按单位/月份派发续聘通知（dry-run）", ) sp.add_argument("--unit-id", required=True, help="二级单位 ID") sp.add_argument("--month", default="", help="月份 YYYY-MM（留空 = 全部）") sp.set_defaults(handler=cmd_dispatch) sp = sub.add_parser( "mark-contract-renewed", help="标记合同已续签", ) sp.add_argument("--contract-id", required=True, help="合同 ID") sp.add_argument("--material-id", required=True, help="续签材料 ID") sp.set_defaults(handler=cmd_mark_renewed) sp = sub.add_parser( "render-html", help="渲染任务书 HTML", ) sp.add_argument("--task-id", required=True, help="任务 ID") sp.set_defaults(handler=cmd_render_html) sp = sub.add_parser( "collect-from-platform", help="Playwright RPA 抓 SmartCampus HR 平台（合同表 + 材料包）", ) sp.add_argument("--limit", type=int, default=0, help="限制抓取行数（0=全量）") sp.add_argument("--show-browser", action="store_true", help="显示浏览器（首次手工登录用，默认 headless）") sp.add_argument("--human-wait-timeout", type=int, default=None, help="手工登录 fallback 等待秒数（默认 180 = HUMAN_WAIT_TIMEOUT）") sp.set_defaults(handler=cmd_collect_platform) sp = sub.add_parser( "parse-materials", help="解析已下载的 PDF/Word 材料 → 文本", ) sp.add_argument("--contract-id", default="", help="仅解析该合同（留空=处理所有 pending）") sp.set_defaults(handler=cmd_parse_materials) return p def main(argv: Optional[List[str]] = None) -> int: argv = argv if argv is not None else sys.argv[1:] if not argv: print(f"合同到期智能预警与续聘辅助 v{SKILL_VERSION}") 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") return 1 setup_skill_logging() parser = build_parser() args = parser.parse_args(argv) handler: Callable[[argparse.Namespace], int] = args.handler return handler(args)