taojianmin
06c053d7d7
新增（scripts/ 分层完整实现）：
- scripts/util/{constants,date_util,runtime_paths,logging_setup}.py
- scripts/db/{connection,seed_data,repository}.py（SQLite + 5 张表 + 50 份样例合同）
- scripts/service/{scanner,task_generator,dispatcher,html_renderer}.py
- scripts/cli/app.py（argparse 子命令）
- scripts/templates/task.html.j2（Jinja2 渲染，无 jinja2 时降级到 str.replace）
- examples/{contracts_sample,performance_sample}.csv（本地样例数据）
修改：
- scripts/main.py：argparse + jiangchang_skill_core 降级
- SKILL.md：version 1.0.1 → 1.1.0
- CHANGELOG.md：加 v1.1.0 条目
- .env.example：加 DB_PATH / HTML_OUTPUT_DIR / JIANGCHANG_DATA_DIR
本地验证（9 个 CLI 命令全部 exit code = 0）：
- health / version / config-path / init-db：基础 4 命令
- scan-expiring-contracts --thresholds 90,60,30：42 份可续聘 → 红 10 / 橙 10 / 黄 10，生成 30 任务
- render-html：写 output/<task_id>.html（4.9 KB）
- generate-renewal-task：单份合同任务书生成
- mark-contract-renewed：状态机更新 + 关闭 open alerts
- dispatch-task-notice：按单位/月份派发 payload（dry-run 模式）
业务规则（AI 续聘建议本地版，规则顺序判断）：
1. 师德 < 60 → non_renew
2. 近 3 年考核有不合格（< 60）→ observe
3. 全部 >= 80 且逐年上升 → renew
4. 其他 → observe
Co-Authored-By: Claude Code <noreply@anthropic.com>
89 lines
2.8 KiB
Python
89 lines
2.8 KiB
Python
"""扫描器：扫全校合同 → 分级 → 写 alerts + 生成 renewal_tasks。
|
||
|
||
核心业务入口，cron 每日凌晨 02:00 调用。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sqlite3
|
||
from datetime import date
|
||
from typing import Tuple
|
||
|
||
from db import repository as repo
|
||
from service.task_generator import generate_task_for_contract
|
||
from util.constants import ALERT_LEVELS, DEFAULT_THRESHOLDS
|
||
from util.date_util import classify_alert, days_to, parse_iso_date
|
||
from util.logging_setup import get_skill_logger
|
||
|
||
logger = get_skill_logger()
|
||
|
||
|
||
def scan(
|
||
conn: sqlite3.Connection,
|
||
thresholds: Tuple[int, int, int] = DEFAULT_THRESHOLDS,
|
||
today: date | None = None,
|
||
scan_id: str | None = None,
|
||
) -> dict:
|
||
"""扫描全校合同，返回结构化结果。
|
||
|
||
thresholds = (yellow_max, orange_max, red_max)，默认 (90, 60, 30)。
|
||
|
||
返回：
|
||
{
|
||
"scanned_count": <int>,
|
||
"yellow_count": <int>,
|
||
"orange_count": <int>,
|
||
"red_count": <int>,
|
||
"generated_task_count": <int>,
|
||
"scan_id": <str>,
|
||
"thresholds": [yellow_max, orange_max, red_max],
|
||
}
|
||
"""
|
||
today = today or date.today()
|
||
scan_id = scan_id or f"S{datetime_now_compact()}"
|
||
|
||
contracts = repo.list_active_contracts(conn)
|
||
counts = {f"{lvl}_count": 0 for lvl in ALERT_LEVELS}
|
||
generated = 0
|
||
|
||
logger.info("scan: start scan_id=%s thresholds=%s today=%s contracts=%d",
|
||
scan_id, thresholds, today.isoformat(), len(contracts))
|
||
|
||
with conn:
|
||
for c in contracts:
|
||
contract_end = parse_iso_date(c["contract_end"])
|
||
days = days_to(contract_end, today)
|
||
level = classify_alert(days, thresholds)
|
||
if level is None:
|
||
continue
|
||
|
||
# 写 alert（alert_id 包含 scan_id 避免同一合同被多次扫描时冲突）
|
||
alert_id = f"A-{c['contract_id']}-{scan_id}"
|
||
repo.insert_alert(conn, alert_id, c["contract_id"], level, days, scan_id)
|
||
counts[f"{level}_count"] += 1
|
||
|
||
# 生成任务书
|
||
task_id = f"T-{c['contract_id']}-{scan_id}"
|
||
task = generate_task_for_contract(conn, c, alert_id, task_id)
|
||
if task:
|
||
generated += 1
|
||
|
||
result = {
|
||
"scanned_count": len(contracts),
|
||
"yellow_count": counts["yellow_count"],
|
||
"orange_count": counts["orange_count"],
|
||
"red_count": counts["red_count"],
|
||
"generated_task_count": generated,
|
||
"scan_id": scan_id,
|
||
"thresholds": list(thresholds),
|
||
"today": today.isoformat(),
|
||
}
|
||
logger.info("scan: done %s", result)
|
||
return result
|
||
|
||
|
||
def datetime_now_compact() -> str:
|
||
"""生成紧凑时间字符串，用于 alert_id / task_id / scan_id。"""
|
||
from datetime import datetime
|
||
return datetime.now().strftime("%Y%m%d%H%M%S")
|