v1.1.0: 实现核心业务逻辑(扫描/任务书/派发/状态机)

新增(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>
This commit is contained in:
2026-09-14 14:17:43 +08:00
parent e2b45d690e
commit 06c053d7d7
20 changed files with 1796 additions and 82 deletions

45
scripts/util/constants.py Normal file
View File

@@ -0,0 +1,45 @@
"""技能标识、版本与阈值常量。
复制到新 skill 时必须修改:
- SKILL_SLUG 与 SKILL.md metadata.openclaw.slug 保持一致
- SKILL_VERSION 与 SKILL.md version 保持一致
"""
SKILL_SLUG = "alert-contract-expiring-hr"
SKILL_VERSION = "1.1.0"
LOG_LOGGER_NAME = "openclaw.skill.alert_contract_expiring_hr"
PLATFORM_KIT_MIN_VERSION = "1.2.11"
# 预警阈值(黄/橙/红):距到期日 <= yellow 天 = 黄色预警
# 顺序:red < orange < yellow
DEFAULT_THRESHOLDS = (90, 60, 30)
# 预警等级名称(与 thresholds 顺序一一对应)
ALERT_LEVELS = ("yellow", "orange", "red")
ALERT_LEVEL_LABELS = {"yellow": "黄色", "orange": "橙色", "red": "红色"}
# 续聘建议
SUGGESTIONS = ("renew", "observe", "non_renew")
SUGGESTION_LABELS = {
"renew": "建议续聘",
"observe": "常规续聘评审",
"non_renew": "不建议续聘",
}
# 合同状态
CONTRACT_STATUSES = ("可续聘", "预警中", "已续签", "已驳回", "已过期", "已撤回")
# 续签材料清单
DEFAULT_MATERIALS = [
"近 3 年年度考核表(盖章)",
"师德师风考评记录",
"学历学位证书复印件",
"岗位聘任登记表",
"劳动合同文本(一式三份)",
"个人述职报告(不少于 2000 字)",
]
# AI 续聘建议规则阈值
MORAL_FAIL_THRESHOLD = 60 # 师德 < 60 直接 non_renew
PERFORMANCE_FAIL_THRESHOLD = 60 # 单年考核 < 60 算不合格
PERFORMANCE_RENEW_MIN = 80 # 全部 >= 80 且逐年上升 → renew

64
scripts/util/date_util.py Normal file
View File

@@ -0,0 +1,64 @@
"""日期工具:距到期日天数、月份匹配、ISO 字符串解析。"""
from __future__ import annotations
from datetime import date, datetime
from typing import Tuple
def parse_iso_date(s: str) -> date:
"""解析 'YYYY-MM-DD' ISO 日期字符串。
容错:
- 已经是 date 实例 → 直接返回
- 含 'T' 的 ISO datetime → 取 date 部分
"""
if isinstance(s, date):
return s
s = str(s).strip()
if "T" in s:
s = s.split("T", 1)[0]
return datetime.strptime(s, "%Y-%m-%d").date()
def days_to(target: date, today: date | None = None) -> int:
"""计算距 target 还有多少天(负数 = 已过期)。"""
today = today or date.today()
return (target - today).days
def classify_alert(days: int, thresholds: Tuple[int, int, int] = (90, 60, 30)) -> str | None:
"""根据距到期天数返回预警等级。
thresholds = (yellow_max, orange_max, red_max),须 yellow_max > orange_max > red_max。
返回:'yellow' / 'orange' / 'red' / None(不在预警范围)。
"""
yellow_max, orange_max, red_max = thresholds
if days > yellow_max:
return None
if days <= red_max:
return "red"
if days <= orange_max:
return "orange"
return "yellow"
def month_match(date_to_check: date, month_str: str) -> bool:
"""判断日期是否落在 month_str('YYYY-MM')月份内。"""
if not month_str:
return True
try:
y, m = month_str.strip().split("-")
return date_to_check.year == int(y) and date_to_check.month == int(m)
except (ValueError, AttributeError):
return False
def today_iso() -> str:
"""返回今天日期的 ISO 字符串。"""
return date.today().isoformat()
def now_iso() -> str:
"""返回当前时间的 ISO 字符串。"""
return datetime.now().isoformat(timespec="seconds")

View File

@@ -0,0 +1,53 @@
"""logging 配置:降级到 stderr,避免污染 stdout(stdout 留给结构化输出如 JSON)。
优先用 jiangchang_skill_core 提供的 logging;本地无该包时降级到 stderr。
"""
from __future__ import annotations
import logging
import sys
from util.constants import LOG_LOGGER_NAME, SKILL_SLUG, SKILL_VERSION
_configured = False
def setup_skill_logging() -> logging.Logger:
"""初始化 skill logger(幂等)。"""
global _configured
logger = logging.getLogger(LOG_LOGGER_NAME)
if _configured:
return logger
# 优先用宿主提供的 skill logger 配置
try:
from jiangchang_skill_core.logging_config import setup_skill_logging as _host_setup
_host_setup(SKILL_SLUG, LOG_LOGGER_NAME)
_configured = True
return logging.getLogger(LOG_LOGGER_NAME)
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.setLevel(logging.INFO)
logger.addHandler(handler)
logger.propagate = False
_configured = True
logger.info("logging initialized (local fallback) v%s", SKILL_VERSION)
return logger
def get_skill_logger() -> logging.Logger:
"""获取 skill logger(不会重复 setup)。"""
if not _configured:
return setup_skill_logging()
return logging.getLogger(LOG_LOGGER_NAME)

View File

@@ -0,0 +1,82 @@
"""运行时路径解析。
本地环境:
- 仓库根 = scripts/main.py 的父目录的父目录
- data/ 在仓库根下
- output/ 在仓库根下
- .env 在仓库根下
宿主环境(jiangchang_skill_core 提供)会覆写这些路径到
{JIANGCHANG_DATA_ROOT}/{JIANGCHANG_USER_ID}/{slug}/...
"""
from __future__ import annotations
import os
from pathlib import Path
from util.constants import SKILL_SLUG
# util/runtime_paths.py → scripts/util/ → scripts/ → 仓库根(再上两级)
_THIS_FILE = Path(__file__).resolve()
_SCRIPTS_DIR = _THIS_FILE.parent.parent # scripts/
_SKILL_ROOT = _SCRIPTS_DIR.parent # alert-contract-expiring-hr/
def get_skill_root() -> Path:
"""返回 skill 仓库根目录。"""
# 优先尊重环境变量(宿主设置过)
env_root = os.environ.get("JIANGCHANG_SKILL_ROOT")
if env_root:
return Path(env_root)
return _SKILL_ROOT
def get_data_dir() -> Path:
"""返回 SQLite DB 与种子数据目录。"""
env_dir = os.environ.get("JIANGCHANG_DATA_DIR")
if env_dir:
return Path(env_dir)
return get_skill_root() / "data"
def get_db_path() -> Path:
"""返回 SQLite DB 文件路径。"""
env_db = os.environ.get("DB_PATH")
if env_db:
return Path(env_db)
return get_data_dir() / f"{SKILL_SLUG}.db"
def get_output_dir() -> Path:
"""返回 HTML 报告输出目录。"""
env_out = os.environ.get("HTML_OUTPUT_DIR")
if env_out:
return Path(env_out)
return get_skill_root() / "output"
def get_env_path() -> Path:
"""返回 .env 路径(仓库根 / 当前工作目录)。"""
return get_skill_root() / ".env"
def get_dispatch_dir() -> Path:
"""返回派发 payload JSON 输出目录。"""
return get_data_dir() / "dispatch"
def get_examples_dir() -> Path:
"""返回样例数据目录。"""
return get_skill_root() / "examples"
def get_templates_dir() -> Path:
"""返回 Jinja2 模板目录。"""
return _SCRIPTS_DIR.parent / "scripts" / "templates"
def ensure_dirs() -> None:
"""确保所有需要的目录存在。"""
for d in (get_data_dir(), get_output_dir(), get_dispatch_dir()):
d.mkdir(parents=True, exist_ok=True)