Compare commits

...

3 Commits

Author SHA1 Message Date
1707b453c8 chore: auto release commit (2026-09-14 16:15:23) 2026-09-14 16:15:23 +08:00
01ba253bfe v1.1.2: CHANGELOG 添加 sandbox 卡片采集条目
Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-14 16:03:34 +08:00
4af2b4fbf3 v1.1.2: sandbox 卡卡片文本采集 + 解析为 performance 记录
sandbox 行为新发现:
1. 「续聘材料包」按钮不是下载文件,而是触发页面下方显示「续聘准备任务书」
   +「单位清单」卡片(含 checklist + 近三年考核表)
2. 进入 hr/contracts 页面后需要刷新(page.reload)才能看到表格行
3. 表格列序扩展为 10 列:工号 / 姓名 / 单位 / 岗位 / 合同类型 / 起止日期 / 剩余天数 / 预警 / 建议 / 材料包按钮
4. contract_id 改用工号(稳定,不随姓名变化)

改动:
- platform_collector.collect_all:
  * 进入 hr/contracts 后 page.reload 刷新触发 SPA mock 数据加载
  * 新列序映射:r[0]=工号 r[1]=姓名 r[2]=单位 r[3]=岗位 r[4]=合同类型
    r[5]=起止日期 r[6]=剩余天数 r[7]=预警 r[8]=建议
  * 合同 end_date 从「2024-03-01 ~ 2026-04-15」截取 ~ 后的部分
  * 点「续聘材料包」按钮 → 抓下方卡片 inner_text(含 checklist + 考核等次)
  * 把卡片文本存入 materials_content(parse_method='sandbox_card')
- _persist_contracts 增加 sandbox_card 路径分支:写 materials_content 后同步调
  repo.parse_sandbox_card_performance → 抽 year + 等次 → 写 performance_records
- repository.parse_sandbox_card_performance:双模式正则匹配
  * 行格式 "2024\t称职"
  * 叙述式 "2024年:等次「称职」"
- repository.upsert_performance_record + 等次 → 分数映射
  (优秀=90 称职=75 基本称职=60 不称职=30)

实测(sandbox 3 合同):
- RPA-T20190033-20260415 钱实验 → 解析 4 条 performance
- RPA-T20200055-20260630 孙助理 → 解析 6 条
- RPA-T20180021-20260831 王明远 → 解析 6 条
- generate-renewal-task 全部输出 observe + decision_source=rules

Co-Authored-By: Claude Code <noreply@anthropic.com>
2026-09-14 16:01:20 +08:00
6 changed files with 849 additions and 50 deletions

View File

@@ -2,6 +2,27 @@
本文件对应技能市场 **「更新日志」** Tab。
## 1.1.3
- **版本对齐修复**:v1.1.2 tag(commit `4af2b4f`)发布时 `SKILL.md version` / `scripts/util/constants.py SKILL_VERSION` 未同步升级(仍为 1.1.1),本次发布将 manifest 升至 1.1.3 并与代码现状对齐
- **内容同 v1.1.2**:sandbox 卡片采集功能(详见 1.1.2 条目)
> **说明**:v1.1.2 与 v1.1.3 实质内容相同(均含 sandbox 卡片采集)。v1.1.2 tag 视为已发布但 manifest 错位的历史遗留;v1.1.3 为首次 manifest 版本号与代码对齐的 release tag。
## 1.1.2
- **sandbox 卡片采集**:表格 10 列(工号/姓名/单位/岗位/合同类型/起止日期/剩余天数/预警/建议/材料包按钮),进入 hr/contracts 页面后 `page.reload()` 刷新触发 mock 数据加载
- **「续聘材料包」按钮行为**:不触发浏览器 download event,而是展开页面下方「续聘准备任务书」+「单位清单」卡片(含 checklist + 近三年考核表)。RPA 改为抓卡片 inner_text 存 `materials_content`(parse_method='sandbox_card')
- **卡片文本解析进 performance_records**:新增 `repository.parse_sandbox_card_performance`(双模式正则:行格式 `2024\t称职` + 叙述式 `2024年:等次「称职」`)+ `repository.upsert_performance_record`(等次→分数:优秀=90 称职=75 基本称职=60 不称职=30)
- **contract_id 改用工号**(稳定,不随姓名变化):`_gen_contract_id(emp_id, end_date)`
- **实测**(sandbox 3 合同):
| contract_id | 教师 | 解析 perf 条数 | AI 决策 |
|---|---|---|---|
| RPA-T20190033-20260415 | 钱实验 | 4 | observe |
| RPA-T20200055-20260630 | 孙助理 | 6 | observe |
| RPA-T20180021-20260831 | 王明远 | 6 | observe |
## 1.1.1
- **AI 续聘建议升级**:本地规则版 → LLM 优先 + 规则降级

View File

@@ -1,7 +1,7 @@
---
name: 合同到期智能预警与续聘辅助
description: "每日凌晨扫描全校教职工合同电子台账,对距离到期日 90/60/30 天的合同分级预警,并自动生成《续聘准备任务书》和《本月合同到期人员清单》。"
version: 1.1.1
version: 1.1.3
author: 深圳匠厂科技有限公司
metadata:
openclaw:

View File

@@ -59,6 +59,82 @@ def list_performance_by_contract(
).fetchall()
# 等次 → 分数映射(v1.1.1 sandbox card 解析)
_GRADE_TO_SCORE = {
"优秀": 90.0,
"称职": 75.0,
"基本称职": 60.0,
"不称职": 30.0,
"不合格": 30.0,
}
def upsert_performance_record(
conn: sqlite3.Connection,
contract_id: str,
year: int,
grade: str,
) -> bool:
"""upsert 一条业绩记录(用 grade 映射分数)。
sandbox card 文本不含真分数时使用。返回是否实际写入。
"""
score = _GRADE_TO_SCORE.get(grade)
if score is None:
return False
record_id = f"P-{contract_id}-{year}"
conn.execute(
"""
INSERT INTO performance_records
(record_id, contract_id, year, performance_score,
moral_teaching_score, research_score, teaching_score)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(record_id) DO UPDATE SET
performance_score = excluded.performance_score
""",
(record_id, contract_id, year, score, score, score, score),
)
return True
def parse_sandbox_card_performance(
conn: sqlite3.Connection,
contract_id: str,
card_text: str,
) -> int:
"""从 sandbox 卡片文本抽 (year, 等次) → upsert 到 performance_records。
文本片段形如:
year result
2024 称职
2025 优秀
2024年:等次「称职」。
返回写入条数。
"""
import re
if not card_text:
return 0
written = 0
# 模式 1:行格式 "2024\t称职"
for m in re.finditer(r"^(\d{4})\s*[ \t]+(优秀|称职|基本称职|不称职|不合格)\s*$",
card_text, re.MULTILINE):
year = int(m.group(1))
grade = m.group(2)
if upsert_performance_record(conn, contract_id, year, grade):
written += 1
# 模式 2:叙述式 "2024年:等次「称职」"
for m in re.finditer(r"(\d{4})\s*年[::]\s*等次[「\"']\s*(优秀|称职|基本称职|不称职|不合格)\s*[」\"']",
card_text):
year = int(m.group(1))
grade = m.group(2)
if upsert_performance_record(conn, contract_id, year, grade):
written += 1
return written
# ---------- alerts ----------
def insert_alert(

View File

@@ -243,6 +243,17 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
logger.info("platform_collector: 当前 URL=%s", page.url)
# 5.5. 进入页面后刷新(SPA 进入新路由时 mock 数据可能未加载,需刷新触发)
try:
logger.info("platform_collector: 进入 hr/contracts 后刷新页面")
page.reload(wait_until="networkidle", timeout=20_000)
except Exception as e:
logger.warning("刷新 networkidle 失败: %s;尝试 domcontentloaded", e)
try:
page.reload(wait_until="domcontentloaded", timeout=20_000)
except Exception as e2:
logger.warning("刷新 domcontentloaded 也失败: %s", e2)
# 7. dump 当前 hr/contracts 页面(用于调试)
try:
html = page.content()
@@ -284,6 +295,23 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
logger.warning("dump 失败: %s", e)
return _error_result("table_not_found_after_40s", 0)
# 表格已找到,等 tbody 完全填充 + dump 一份完整 DOM(用于诊断按钮 selector)
try:
page.wait_for_function(
"document.querySelectorAll('table tbody tr').length > 0",
timeout=10_000,
)
page.wait_for_timeout(1500)
html = page.content()
full_dump = download_dir.parent / "hr_contracts_full_dump.html"
full_dump.write_text(html, encoding="utf-8")
logger.info(
"platform_collector: 完整 hr/contracts dump → %s (%d 字节)",
full_dump, len(html),
)
except Exception as e:
logger.warning("等完整渲染超时: %s", e)
rows = page.eval_on_selector_all(
"table tbody tr",
"rs => rs.map(r => Array.from(r.querySelectorAll('td')).map(td => td.innerText))",
@@ -302,21 +330,33 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
except Exception:
mat_btn_count = 0
# 抓「材料包」card 区段的 DOM(用于后续解析每行 click 后的卡片文本)
# sandbox 行为:click 按钮后会在表格下方追加「续聘准备任务书」+「单位清单」卡片
for i, r in enumerate(rows):
if limit and i >= limit:
break
if len(r) < 6:
logger.warning("行 %d 列数 < 6,跳过: %s", i, r)
if len(r) < 7:
logger.warning("行 %d 列数 < 7(新列序工号+姓名+单位+岗位+合同类型+起止+剩余+预警+建议),跳过: %s", i, r)
continue
teacher = r[0].strip()
unit = r[1].strip()
contract_end = r[2].strip()
days_raw = r[3].strip()
alert_label = r[4].strip()
suggestion_hint = r[5].strip()
# 新列序(sandbox 加了工号 / 岗位 / 合同类型 / 起止日期):
# 0: 工号 1: 姓名 2: 单位 3: 岗位 4: 合同类型
# 5: 起止日期 (含 ~ ) 6: 剩余天数 7: 预警 8: 建议(含原因)
emp_id = r[0].strip()
teacher = r[1].strip()
unit = r[2].strip()
position = r[3].strip()
contract_type = r[4].strip()
period = r[5].strip()
# 起止日期形如 "2024-03-01 ~ 2026-04-15" → 取 ~ 后
contract_end = period.split("~")[-1].strip() if "~" in period else period
days_raw = r[6].strip()
alert_label = r[7].strip()
suggestion_hint = r[8].strip() if len(r) > 8 else ""
contract_id = _gen_contract_id(teacher, contract_end)
# 用工号生成 contract_id(稳定,不会随姓名变化)
contract_id = _gen_contract_id(emp_id, contract_end)
contract = {
"contract_id": contract_id,
"teacher_name": teacher,
@@ -328,32 +368,50 @@ def collect_all(limit: int = 0, headless: bool = True) -> dict:
"materials": [],
}
# 只有在按钮存在时才尝试下载(sandbox mock 当前未部署此按钮)
# 点「续聘材料包」按钮 → 等卡片展开 → 抓卡片文本(含 checklist)
if mat_btn_count > 0:
try:
btn = page.locator("table tbody tr").nth(i).locator(
"button:has-text('续聘材料包')"
)
with page.expect_download(timeout=10_000) as dl_info:
btn.click()
dl = dl_info.value
fname = dl.suggested_filename or "materials.zip"
dest = download_dir / f"{contract_id}_{fname}"
dl.save_as(str(dest))
contract["materials"].append(str(dest))
downloaded += 1
btn.click()
# 等卡片出现(页面下方追加 h2「续聘准备任务书」/「单位清单」)
page.wait_for_timeout(1500)
# 抓表格下方所有 card 文本
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_content(sandbox mock 卡片文本作为材料包内容)
contract["materials"].append({
"kind": "sandbox_card",
"file_name": f"{contract_id}_materials.txt",
"text": materials_text,
})
downloaded += 1
logger.info(
"platform_collector: 抓到卡片文本 %d 字 (%s)",
len(materials_text), contract_id,
)
# dump 一次(i==0)方便调试
if i == 0:
html_after = page.content()
click_dump = download_dir.parent / "after_click_dump.html"
click_dump.write_text(html_after, encoding="utf-8")
logger.info(
"platform_collector: 首行 click 后 dump → %s (%d 字节)",
click_dump, len(html_after),
)
except Exception as e:
logger.warning("材料下载失败 %s: %s", contract_id, e)
logger.warning("抓卡片失败 %s: %s", contract_id, e)
else:
logger.info(
"platform_collector: sandbox 未部署「续聘材料包」按钮,"
"跳过材料下载(仅采集表格列)",
)
# 仅警告一次
if i == 0:
logger.warning(
"platform_collector: 提示:材料包缺失时,"
"parse-materials CLI 将标记为「无材料」"
logger.info(
"platform_collector: sandbox 未部署「续聘材料包」按钮,"
"跳过材料抓取(仅采集表格列)",
)
contracts.append(contract)
@@ -413,27 +471,56 @@ def _persist_contracts(contracts: list[dict]) -> int:
),
)
# 写 materials_content(占位:file_path 写入,extracted_text 待 parse-materials)
# 写 materials_content(sandbox card 文本 + parse-materials 解析后文件路径)
from db.migrations import ensure_materials_content_schema
ensure_materials_content_schema(conn)
for fp in c["materials"]:
fname = Path(fp).name
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"),
),
)
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"),
),
)
return len(contracts)
finally:
conn.close()
@@ -450,11 +537,14 @@ def _error_result(reason: str, fetched: int) -> dict:
}
def _gen_contract_id(teacher: str, end_date: str) -> str:
"""从 (姓名, 到期日) 派 contract_id,确保幂等。"""
safe_teacher = re.sub(r"[^\w]", "", teacher)[:8] or "T"
def _gen_contract_id(emp_id: str, end_date: str) -> str:
"""从 (工号, 到期日) 派 contract_id,确保幂等。
sandbox v1.1.1 起改用「工号」作主键(之前用姓名)。
"""
safe_emp = re.sub(r"[^\w]", "", emp_id)[:10] or "T"
safe_end = end_date.replace("-", "")[:8]
return f"RPA-{safe_teacher}-{safe_end}"
return f"RPA-{safe_emp}-{safe_end}"
def _normalize_date(s: str) -> str:

View File

@@ -6,7 +6,7 @@
"""
SKILL_SLUG = "alert-contract-expiring-hr"
SKILL_VERSION = "1.1.1"
SKILL_VERSION = "1.1.3"
LOG_LOGGER_NAME = "openclaw.skill.alert_contract_expiring_hr"
PLATFORM_KIT_MIN_VERSION = "1.2.11"

612
tests/run_tests.py Normal file
View File

@@ -0,0 +1,612 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""默认测试套件:合同到期智能预警与续聘辅助。
设计原则:
- 仅用标准库(不依赖 pytest / jinja2),CI 与本地均能直接 `python tests/run_tests.py` 跑
- 用 sqlite3 ":memory:" 隔离 DB,不污染仓库 data/ 目录
- 覆盖:util.date_util 纯函数 / util.constants 版本对齐 / repository 新增功能
(parse_sandbox_card_performance + upsert_performance_record)/ scanner 分级 / task_generator
规则版 / html_renderer 降级路径
- 退出码:0 = 全部通过,1 = 有失败
- -v 开关:详细输出每个用例的状态
release.ps1 在打 tag 前会自动跑 `python tests/run_tests.py -v`,任何一个
test_* 返回 (False, msg) 即 abort 发布。
"""
from __future__ import annotations
import argparse
import os
import re
import shutil
import sqlite3
import sys
import tempfile
import traceback
from datetime import date
from pathlib import Path
from types import SimpleNamespace
from typing import Any, Callable
# 让 scripts/ 下的模块可导入
_REPO_ROOT = Path(__file__).resolve().parent.parent
_SCRIPTS_DIR = _REPO_ROOT / "scripts"
sys.path.insert(0, str(_SCRIPTS_DIR))
# ---------- 测试基础设施 ----------
# 测试结果收集器
_RESULTS: list[tuple[str, bool, str]] = []
_VERBOSE = False
def _run(name: str, fn: Callable[[], tuple[bool, str]]) -> None:
"""运行单个 test_*,记录结果。"""
try:
ok, msg = fn()
except Exception as e: # 测试本身抛异常也算失败
tb = traceback.format_exc(limit=3)
ok, msg = False, f"exception: {e}\n{tb}"
_RESULTS.append((name, ok, msg))
if _VERBOSE or not ok:
marker = "PASS" if ok else "FAIL"
print(f" [{marker}] {name}: {msg}", flush=True)
def _make_conn() -> sqlite3.Connection:
"""建一张 in-memory DB,应用所有 DDL + 迁移(不加载 seed)。"""
from db.connection import DDL_STATEMENTS
from db.migrations import apply_all
conn = sqlite3.connect(":memory:")
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON;")
for ddl in DDL_STATEMENTS:
conn.execute(ddl)
apply_all(conn)
conn.commit()
return conn
class _FakeRow:
"""Mock 一行 sqlite3.Row:同时支持下标访问 (r["x"]) 和属性访问 (r.x)。
production 代码里 task_generator.suggest() 用的是 r["moral_teaching_score"]
下标语法,所以 fake row 必须也支持。
"""
__slots__ = ("year", "performance_score", "moral_teaching_score",
"research_score", "teaching_score")
def __init__(self, year: int, score: float, moral: float):
self.year = year
self.performance_score = score
self.moral_teaching_score = moral
self.research_score = score
self.teaching_score = score
def __getitem__(self, key: str) -> Any:
return getattr(self, key)
def _fake_perf(year: int, score: float, moral: float | None = None) -> _FakeRow:
"""构造一个 performance_records-like 行(用于纯函数测试,不入 DB)。"""
m = moral if moral is not None else score
return _FakeRow(year=year, score=score, moral=m)
def _insert_stub_contract(conn: sqlite3.Connection, contract_id: str) -> None:
"""插入一份占位合同(performance_records 的 FK 目标)。"""
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)""",
(contract_id, f"E-{contract_id}", f"占位-{contract_id}", "U-TEST",
"测试系", "讲师", "正式", "2022-01-01", "2099-12-31", "2022-07-01"),
)
conn.commit()
# ---------- util.date_util ----------
def test_parse_iso_date() -> tuple[bool, str]:
from util.date_util import parse_iso_date
cases = [
("2024-03-15", date(2024, 3, 15)),
("2024-03-15T08:30:00", date(2024, 3, 15)),
(" 2024-03-15 ", date(2024, 3, 15)),
(date(2024, 3, 15), date(2024, 3, 15)), # 已为 date
]
for inp, want in cases:
got = parse_iso_date(inp)
if got != want:
return False, f"parse_iso_date({inp!r}) = {got}, want {want}"
return True, "4 cases ok"
def test_days_to() -> tuple[bool, str]:
from util.date_util import days_to
if days_to(date(2024, 3, 15), today=date(2024, 1, 1)) != 74:
return False, "days_to 正向计算错"
if days_to(date(2023, 12, 31), today=date(2024, 1, 1)) != -1:
return False, "days_to 已过期返回负数错"
# 默认 today=今天
if days_to(date(2099, 1, 1)) <= 0:
return False, "days_to 默认 today 应为今天"
return True, "正/负/默认 today 三种场景"
def test_classify_alert() -> tuple[bool, str]:
from util.date_util import classify_alert
t = (90, 60, 30)
cases = [
(100, None), # 超出预警范围
(91, None), # 边界外
(90, "yellow"),
(61, "yellow"),
(60, "orange"),
(31, "orange"),
(30, "red"),
(1, "red"),
(0, "red"),
(-5, "red"), # 已过期仍属 red
]
for days, want in cases:
got = classify_alert(days, t)
if got != want:
return False, f"classify_alert({days}) = {got}, want {want}"
return True, "10 边界 + 越界场景"
def test_month_match() -> tuple[bool, str]:
from util.date_util import month_match
d = date(2024, 3, 15)
if not month_match(d, "2024-03"):
return False, "同月应命中"
if month_match(d, "2024-04"):
return False, "不同月不应命中"
if not month_match(d, ""):
return False, "空 month_str 应命中全部"
if month_match(d, "garbage"):
return False, "非法 month_str 不应命中"
return True, "命中 / 未命中 / 空 / 非法 四种"
# ---------- util.constants 版本对齐(POLICY-RELEASE-001) ----------
def test_skill_md_version_alignment() -> tuple[bool, str]:
"""POLICY-RELEASE-001:SKILL.md version == SKILL_VERSION == CHANGELOG 顶部版本"""
from util.constants import SKILL_VERSION
skill_md = (_REPO_ROOT / "SKILL.md").read_text(encoding="utf-8")
m = re.search(r'(?m)^version:\s*["\']?(\d+\.\d+\.\d+)', skill_md)
if not m:
return False, "SKILL.md 缺顶层 version: X.Y.Z"
md_ver = m.group(1)
if md_ver != SKILL_VERSION:
return False, f"SKILL.md={md_ver} != SKILL_VERSION={SKILL_VERSION}"
# CHANGELOG 顶部应有 ## X.Y.Z
cl = (_REPO_ROOT / "CHANGELOG.md").read_text(encoding="utf-8")
cl_m = re.search(r'(?m)^##\s+(\d+\.\d+\.\d+)', cl)
if not cl_m:
return False, "CHANGELOG.md 缺顶层 ## X.Y.Z"
cl_ver = cl_m.group(1)
if cl_ver != SKILL_VERSION:
return False, f"CHANGELOG 顶部={cl_ver} != SKILL_VERSION={SKILL_VERSION}"
return True, f"SKILL.md={md_ver} == CHANGELOG={cl_ver} == SKILL_VERSION={SKILL_VERSION}"
def test_skill_md_slug_alignment() -> tuple[bool, str]:
"""SKILL.md metadata.openclaw.slug == SKILL_SLUG"""
from util.constants import SKILL_SLUG
skill_md = (_REPO_ROOT / "SKILL.md").read_text(encoding="utf-8")
m = re.search(r'(?m)^\s*slug:\s*["\']?([a-z0-9_-]+)', skill_md)
if not m:
return False, "SKILL.md 缺 metadata.openclaw.slug"
md_slug = m.group(1)
if md_slug != SKILL_SLUG:
return False, f"SKILL.md slug={md_slug} != SKILL_SLUG={SKILL_SLUG}"
return True, f"slug={md_slug}"
# ---------- db.repository:v1.1.2 新增的 sandbox card 解析 ----------
def test_parse_sandbox_card_row_format() -> tuple[bool, str]:
"""v1.1.2:行格式 '2024\\t称职'"""
from db import repository as repo
conn = _make_conn()
_insert_stub_contract(conn, "C001")
card_text = "year\tresult\n2023\t优秀\n2024\t称职\n2025\t基本称职\n"
n = repo.parse_sandbox_card_performance(conn, "C001", card_text)
if n != 3:
return False, f"写入 {n} 条,期望 3"
rows = repo.list_performance_by_contract(conn, "C001")
if len(rows) != 3:
return False, f"DB 行数 {len(rows)},期望 3"
by_year = {r["year"]: r["performance_score"] for r in rows}
expected = {2023: 90.0, 2024: 75.0, 2025: 60.0}
for y, want in expected.items():
if by_year.get(y) != want:
return False, f"{y} 年分数={by_year.get(y)},期望 {want}"
return True, "3 行(优秀/称职/基本称职)解析 + 写入正确"
def test_parse_sandbox_card_narrative_format() -> tuple[bool, str]:
"""v1.1.2:叙述式 '2024年:等次「称职」'"""
from db import repository as repo
conn = _make_conn()
_insert_stub_contract(conn, "C002")
card_text = "考核情况:\n2023年:等次「优秀」。\n2024年:等次「称职」。\n2025年:等次「不称职」。\n"
n = repo.parse_sandbox_card_performance(conn, "C002", card_text)
if n != 3:
return False, f"写入 {n} 条,期望 3"
rows = repo.list_performance_by_contract(conn, "C002")
by_year = {r["year"]: r["performance_score"] for r in rows}
expected = {2023: 90.0, 2024: 75.0, 2025: 30.0}
for y, want in expected.items():
if by_year.get(y) != want:
return False, f"{y} 年分数={by_year.get(y)},期望 {want}"
return True, "3 行(优秀/称职/不称职)叙述式解析正确"
def test_parse_sandbox_card_empty_and_invalid() -> tuple[bool, str]:
"""空字符串 / 无匹配应返回 0,不报错"""
from db import repository as repo
conn = _make_conn()
_insert_stub_contract(conn, "C001")
if repo.parse_sandbox_card_performance(conn, "C001", "") != 0:
return False, "空字符串应返回 0"
if repo.parse_sandbox_card_performance(conn, "C001", "没有任何年份/等次信息") != 0:
return False, "无匹配应返回 0"
rows = repo.list_performance_by_contract(conn, "C001")
if rows:
return False, "无匹配不应写入任何行"
return True, "空 / 无匹配场景"
def test_parse_sandbox_card_mixed_formats() -> tuple[bool, str]:
"""同卡片混用两种格式时按行去重(PK = P-<contract>-<year>)"""
from db import repository as repo
conn = _make_conn()
_insert_stub_contract(conn, "C003")
card_text = (
"2024年:等次「称职」。\n" # 叙述式
"2025\t优秀\n" # 行格式
"2026年:等次「基本称职」\n"
)
n = repo.parse_sandbox_card_performance(conn, "C003", card_text)
if n != 3:
return False, f"混合解析写入 {n} 条,期望 3"
return True, "叙述式 + 行格式混合解析"
def test_upsert_performance_record_grade_mapping() -> tuple[bool, str]:
"""grade → score 映射 + ON CONFLICT upsert"""
from db import repository as repo
conn = _make_conn()
_insert_stub_contract(conn, "C001")
if not repo.upsert_performance_record(conn, "C001", 2024, "优秀"):
return False, "优秀应返回 True"
if not repo.upsert_performance_record(conn, "C001", 2024, "优秀"):
return False, "upsert 已存在应仍返回 True"
if not repo.upsert_performance_record(conn, "C001", 2025, "称职"):
return False, "称职应返回 True"
if not repo.upsert_performance_record(conn, "C001", 2026, "基本称职"):
return False, "基本称职应返回 True"
if not repo.upsert_performance_record(conn, "C001", 2027, "不称职"):
return False, "不称职应返回 True"
if not repo.upsert_performance_record(conn, "C001", 2028, "不合格"):
return False, "不合格应返回 True"
if repo.upsert_performance_record(conn, "C001", 2029, "未知等次"):
return False, "未知等次应返回 False"
rows = repo.list_performance_by_contract(conn, "C001")
# 期望 5 行:2024/2025/2026/2027/2028
# 第二次 (2024, 优秀) 是 upsert 覆盖不新增;2029 未知等次不入库
if len(rows) != 5:
return False, f"DB 行数 {len(rows)},期望 5(upsert 同年不新增;未知等次不入库)"
by_year = {r["year"]: r["performance_score"] for r in rows}
expected = {2024: 90.0, 2025: 75.0, 2026: 60.0, 2027: 30.0, 2028: 30.0}
for y, want in expected.items():
if by_year.get(y) != want:
return False, f"{y} 年分数={by_year.get(y)},期望 {want}"
# 验证 2024 的分数确实被 upsert 覆盖(不是 30/60/75 等其他值)
if by_year[2024] != 90.0:
return False, f"upsert 未保持分数:2024={by_year[2024]},want 90.0"
return True, "grade 映射 + upsert 幂等 + 未知等次返回 False"
# ---------- service.task_generator.suggest 规则版 ----------
def test_suggest_empty_records() -> tuple[bool, str]:
"""无业绩记录 → observe(数据缺失)"""
from service.task_generator import suggest
s, r = suggest([])
if s != "observe":
return False, f"suggest([]) = {s}, want observe"
if "缺失" not in r:
return False, f"reason 应提示缺失,实际 {r!r}"
return True, "observe + '缺失'"
def test_suggest_low_moral() -> tuple[bool, str]:
"""师德 < 60 → non_renew"""
from service.task_generator import suggest
rows = [_fake_perf(2024, 85, moral=50)]
s, r = suggest(rows)
if s != "non_renew":
return False, f"suggest(师德 50) = {s}, want non_renew"
if "师德" not in r:
return False, f"reason 应含「师德」,实际 {r!r}"
return True, "non_renew + '师德'"
def test_suggest_fail_year() -> tuple[bool, str]:
"""任意一年考核 < 60 → observe(复核)"""
from service.task_generator import suggest
rows = [
_fake_perf(2023, 80, moral=85),
_fake_perf(2024, 50, moral=85), # 不及格
_fake_perf(2025, 85, moral=85),
]
s, r = suggest(rows)
if s != "observe":
return False, f"suggest(有不及格年) = {s}, want observe"
if "复核" not in r and "不合格" not in r:
return False, f"reason 应提示不合格/复核,实际 {r!r}"
return True, "observe + '不合格'"
def test_suggest_renew_upward() -> tuple[bool, str]:
"""全部 >= 80 且逐年上升 → renew"""
from service.task_generator import suggest
rows = [
_fake_perf(2023, 80, moral=85),
_fake_perf(2024, 85, moral=85),
_fake_perf(2025, 90, moral=85),
]
s, r = suggest(rows)
if s != "renew":
return False, f"suggest(全部 80+ 上升) = {s}, want renew"
return True, "renew + 上升趋势"
def test_suggest_observe_default() -> tuple[bool, str]:
"""达标但非逐年上升 → observe(常规评审)"""
from service.task_generator import suggest
rows = [
_fake_perf(2023, 85, moral=85),
_fake_perf(2024, 80, moral=85), # 回落
_fake_perf(2025, 85, moral=85),
]
s, r = suggest(rows)
if s != "observe":
return False, f"suggest(回落) = {s}, want observe"
return True, "observe(常规评审)"
# ---------- service.scanner 端到端 ----------
def test_scanner_no_alerts() -> tuple[bool, str]:
"""所有合同都在 90 天外 → 0 预警"""
from service import scanner
conn = _make_conn()
conn.executemany(
"""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)""",
[
("C-FAR-1", "T001", "张三", "U1", "数学系", "教授", "正式",
"2022-01-01", "2099-12-31", "2022-07-01"),
],
)
conn.commit()
result = scanner.scan(conn, today=date(2024, 1, 15))
if result["scanned_count"] != 1:
return False, f"scanned={result['scanned_count']}, want 1"
if result["yellow_count"] + result["orange_count"] + result["red_count"] != 0:
return False, f"应无预警,实测 yellow={result['yellow_count']} orange={result['orange_count']} red={result['red_count']}"
return True, "1 合同全在 90 天外 → 0 预警"
def test_scanner_three_levels() -> tuple[bool, str]:
"""3 份合同分别命中黄/橙/红"""
from service import scanner
conn = _make_conn()
conn.executemany(
"""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)""",
[
("C-YEL", "T001", "黄老师", "U1", "数学系", "教授", "正式",
"2022-01-01", "2024-05-15", "2022-07-01"), # 距 2024-02-15 = 90 天 → yellow
("C-ORA", "T002", "橙老师", "U1", "物理系", "讲师", "正式",
"2022-01-01", "2024-04-15", "2022-07-01"), # 距 2024-02-15 = 60 天 → orange
("C-RED", "T003", "红老师", "U1", "化学系", "副教授", "正式",
"2022-01-01", "2024-03-15", "2022-07-01"), # 距 2024-02-15 = 29 天 → red
],
)
conn.commit()
result = scanner.scan(conn, today=date(2024, 2, 15))
if result["scanned_count"] != 3:
return False, f"scanned={result['scanned_count']}, want 3"
if result["yellow_count"] != 1 or result["orange_count"] != 1 or result["red_count"] != 1:
return False, (f"分级数 yellow={result['yellow_count']} "
f"orange={result['orange_count']} red={result['red_count']}, "
f"want 1/1/1")
# 验证 alerts 表实际写入
alerts = conn.execute("SELECT contract_id, alert_level FROM alerts").fetchall()
by_cid = {a["contract_id"]: a["alert_level"] for a in alerts}
if by_cid.get("C-YEL") != "yellow":
return False, f"C-YEL 写入等级 {by_cid.get('C-YEL')}, want yellow"
if by_cid.get("C-ORA") != "orange":
return False, f"C-ORA 写入等级 {by_cid.get('C-ORA')}, want orange"
if by_cid.get("C-RED") != "red":
return False, f"C-RED 写入等级 {by_cid.get('C-RED')}, want red"
return True, "3 合同 → 1 黄 + 1 橙 + 1 红,alerts 表写入正确"
def test_scanner_skips_non_active_contracts() -> tuple[bool, str]:
"""status != '可续聘' 的合同不应被扫描"""
from service import scanner
conn = _make_conn()
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 ('C-DONE', 'T001', '已续签老师', 'U1', '数学系', '教授', '正式',
'2022-01-01', '2024-03-15', '2022-07-01', '已续签', NULL, NULL)"""
)
conn.commit()
result = scanner.scan(conn, today=date(2024, 2, 15))
if result["scanned_count"] != 0:
return False, f"已续签合同不应被扫描,scanned={result['scanned_count']}"
alerts = conn.execute("SELECT COUNT(*) AS n FROM alerts").fetchone()["n"]
if alerts != 0:
return False, f"应无 alerts 写入,实测 {alerts}"
return True, "已续签合同被跳过"
# ---------- service.html_renderer 降级路径 ----------
def test_html_render_basic() -> tuple[bool, str]:
"""HTML 渲染(无 jinja2 走 str.replace 降级路径)"""
from service import html_renderer
tmpdir = Path(tempfile.mkdtemp(prefix="alert-hr-test-"))
saved = os.environ.get("HTML_OUTPUT_DIR")
os.environ["HTML_OUTPUT_DIR"] = str(tmpdir)
try:
conn = _make_conn()
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 ('C-HTML', 'T100', '渲染测试', 'U9', '测试系', '教授', '正式',
'2022-01-01', '2024-06-30', '2022-07-01', '可续聘', NULL, NULL)"""
)
conn.execute(
"""INSERT INTO alerts
VALUES ('A-HTML', 'C-HTML', 'orange', 60, 'test',
'2024-04-01T00:00:00', 'open')"""
)
conn.execute(
"""INSERT INTO renewal_tasks
VALUES ('T-HTML', 'C-HTML', 'A-HTML', 'renew', '近 3 年考核良好',
'业绩良好', '["材料 A", "材料 B"]',
'2024-04-01T00:00:00', 'pending',
'rules', '')"""
)
conn.commit()
path = html_renderer.render_task_html(conn, "T-HTML")
if not path.exists():
return False, f"输出文件不存在 {path}"
text = path.read_text(encoding="utf-8")
if "渲染测试" not in text:
return False, "HTML 应含教师姓名"
if "续聘准备任务书" not in text:
return False, "HTML 应含 TITLE"
if "橙色" not in text:
return False, "HTML 应含「橙色」预警等级标签"
if "建议续聘" not in text:
return False, "HTML 应含「建议续聘」建议标签"
return True, f"HTML 渲染 OK → {path.name}"
finally:
if saved is None:
os.environ.pop("HTML_OUTPUT_DIR", None)
else:
os.environ["HTML_OUTPUT_DIR"] = saved
shutil.rmtree(tmpdir, ignore_errors=True)
def test_html_render_unknown_task_raises() -> tuple[bool, str]:
"""渲染不存在的 task_id 应抛 ValueError"""
from service import html_renderer
conn = _make_conn()
try:
html_renderer.render_task_html(conn, "T-NONEXISTENT")
except ValueError as e:
return True, f"ValueError OK: {e}"
return False, "应抛 ValueError,实际未抛"
# ---------- 入口 ----------
def main() -> int:
parser = argparse.ArgumentParser(description="默认 release 测试套件")
parser.add_argument("-v", "--verbose", action="store_true",
help="详细输出每个用例结果")
args = parser.parse_args()
global _VERBOSE
_VERBOSE = args.verbose
print("=== alert-contract-expiring-hr release tests ===", flush=True)
_run("util/parse_iso_date", test_parse_iso_date)
_run("util/days_to", test_days_to)
_run("util/classify_alert", test_classify_alert)
_run("util/month_match", test_month_match)
_run("constants/skill_md_version_alignment", test_skill_md_version_alignment)
_run("constants/skill_md_slug_alignment", test_skill_md_slug_alignment)
_run("db/parse_sandbox_card_row", test_parse_sandbox_card_row_format)
_run("db/parse_sandbox_card_narrative", test_parse_sandbox_card_narrative_format)
_run("db/parse_sandbox_card_empty_invalid", test_parse_sandbox_card_empty_and_invalid)
_run("db/parse_sandbox_card_mixed", test_parse_sandbox_card_mixed_formats)
_run("db/upsert_performance_record", test_upsert_performance_record_grade_mapping)
_run("task_generator/suggest_empty", test_suggest_empty_records)
_run("task_generator/suggest_low_moral", test_suggest_low_moral)
_run("task_generator/suggest_fail_year", test_suggest_fail_year)
_run("task_generator/suggest_renew", test_suggest_renew_upward)
_run("task_generator/suggest_observe", test_suggest_observe_default)
_run("scanner/no_alerts", test_scanner_no_alerts)
_run("scanner/three_levels", test_scanner_three_levels)
_run("scanner/skips_non_active", test_scanner_skips_non_active_contracts)
_run("html_renderer/basic", test_html_render_basic)
_run("html_renderer/unknown_task_raises", test_html_render_unknown_task_raises)
total = len(_RESULTS)
passed = sum(1 for _, ok, _ in _RESULTS if ok)
failed = total - passed
print()
print(f"=== Summary: {passed}/{total} passed, {failed} failed ===", flush=True)
if failed:
print("\nFailures:")
for name, ok, msg in _RESULTS:
if not ok:
print(f" - {name}: {msg}")
return 1
return 0
if __name__ == "__main__":
raise SystemExit(main())