diff --git a/DEMO.md b/DEMO.md index fdf030a..6f572ad 100644 --- a/DEMO.md +++ b/DEMO.md @@ -12,5 +12,5 @@ demo_video_url: "" ## 演示二：师资科批量预审（数据管理） -1. 门槛库按最新文件核准；对话批量核验全院申报案例。 +1. 门槛库按最新文件核准；对话批量核验全院申报案例（无法解析的坏文件自动跳过并告警，不中断整批）。 2. 汇总页按结论排序：🔴硬伤清单（附差什么）供一次性反馈，🟢安全直接进入下一环节。 diff --git a/TUTORIAL.md b/TUTORIAL.md index ccda0ad..a52599c 100644 --- a/TUTORIAL.md +++ b/TUTORIAL.md @@ -2,7 +2,7 @@ ## 本技能做什么 -技能维护一份"评审门槛库"（按副教授/教授分级，每条门槛含最低要求与依据条款）。核验时把申报人的实有指标（省部级项目、核心论文、教学工作量、任现职年限等）逐项与门槛比对：不达标为🔴硬伤、达标但余量不足10%为🟡边缘、全部超标准10%以上为🟢安全，缺数据的标⚪待核，最终生成达标矩阵与《核验报告》。 +技能维护一份"评审门槛库"（按副教授/教授分级，每条门槛含最低要求与依据条款；教授含"核心期刊一作/通讯论文"独立门槛）。核验时把申报人的实有指标（省部级项目、核心论文、教学工作量、任现职年限等）逐项与门槛比对：不达标为🔴硬伤、达标但余量不足10%为🟡边缘、全部超标准10%以上为🟢安全，缺数据或门槛库未配置的标⚪待核，最终生成达标矩阵与《核验报告》。门槛库未配置的级别/指标一律不判"安全"。 ## 开始前 @@ -61,8 +61,9 @@ ## 常见卡住怎么办 -- **核验结论都是"边缘"**：门槛库数值可能不是学校现行标准；请管理员核准后重新核验 -- **某项显示"待核"**：该指标没有提供数值；补齐后重新核验 +- **核验结论都是"边缘"**：门槛库数值可能不是学校现行标准，或该级别门槛被清空（提示"门槛库未配置"）；请管理员核准后重新核验 +- **提示"申报级别只接受副教授、教授"**：申报级别填了其它值；核对级别或请管理员确认是否需扩展级别 +- **某项显示"待核"**：该指标没有提供数值，或门槛库未配置该指标；补齐数据/补配门槛后重新核验 - **提示"门槛库未配置该指标"**：申报人有门槛库之外的指标（如获奖）；请人事处在门槛库补充配置 - **看不到核验报告**：先在"核验案例"表点一次「生成核验报告」，再到左侧 HTML 报告区打开 -- **批量核验没有结果**：确认 input 目录 JSON 是申报案例对象数组 +- **批量核验没有结果**：确认 input 目录 JSON 是申报案例对象数组；无法解析的文件会被跳过并在进度中告警说明 diff --git a/scripts/db/seed_data.py b/scripts/db/seed_data.py index 446f796..5d2cd9c 100644 --- a/scripts/db/seed_data.py +++ b/scripts/db/seed_data.py @@ -16,20 +16,24 @@ _SEED_THRESHOLDS: tuple[tuple[str, str, str, float, str, str, str], ...] = ( ("副教授", "current_title_years", "任现职年限", 5.0, "年", _BASIS, "第X条"), ("教授", "presiding_provincial", "主持省部级项目", 2.0, "项", _BASIS, "第X条"), ("教授", "core_papers", "核心期刊论文", 6.0, "篇", _BASIS, "第X条"), - ("教授", "monographs", "专著/教材", 1.0, "部", _BASIS, "第X条"), + ("教授", "first_author_core_papers", "核心期刊一作/通讯论文", 3.0, "篇", _BASIS, "第X条（其中一作/通讯）"), + ("教授", "monographs", "专著/教材（或等效获奖，获奖由人事处认定）", 1.0, "部", _BASIS, "第X条"), ("教授", "current_title_years", "任现职年限", 5.0, "年", _BASIS, "第X条"), ) _KNOWN_METRIC_KEYS = ( "presiding_provincial", "core_papers", + "first_author_core_papers", "teaching_hours", "current_title_years", "monographs", "awards", "degree", ) -_KNOWN_RANKS = ("副教授", "教授") +KNOWN_RANKS = ("副教授", "教授") +# 兼容旧名 +_KNOWN_RANKS = KNOWN_RANKS def seed_thresholds(cursor: sqlite3.Cursor) -> int: diff --git a/scripts/service/report_service.py b/scripts/service/report_service.py index cbe6759..76a267d 100644 --- a/scripts/service/report_service.py +++ b/scripts/service/report_service.py @@ -22,7 +22,6 @@ def _log(): _TEMPLATE_REL = "assets/html/verify-report.html" -_VERDICT_BADGE = {"安全": "tier-strong", "边缘": "tier-try", "硬伤": "tier-strong"} _COLOR_BADGE = {"green": "tier-strong", "yellow": "tier-try", "red": "tag-red", "gray": "tier-not"} _COLOR_MARK = {"green": "🟢 达标", "yellow": "🟡 边缘", "red": "🔴 硬伤", "gray": "⚪ 待核"} @@ -132,7 +131,8 @@ def build_report_for_case_id(case_id: int) -> str: result["matrix"] = [] html_text = build_verify_html(result) path = write_html_page(f"report-{case_id}.html", html_text) - log.info("report_render_done case_id=%s path=%s stage=render", case_id, path) + if log: + log.info("report_render_done case_id=%s path=%s stage=render", case_id, path) return path diff --git a/scripts/service/task_service.py b/scripts/service/task_service.py index 529ee2f..1799d39 100644 --- a/scripts/service/task_service.py +++ b/scripts/service/task_service.py @@ -360,6 +360,7 @@ def cmd_verify_metric( ) -> int: """职称硬指标核验（主命令，同步）。""" log = _get_task_logger() + from db.seed_data import KNOWN_RANKS from service.verify_service import run_verify with job_context(skill=SKILL_SLUG): @@ -376,12 +377,21 @@ def cmd_verify_metric( metrics = _parse_json_arg(metrics_json, "--metrics-json") if not metrics: raise ValueError("申报指标 JSON 为空") - case = metrics[0] if len(metrics) == 1 else dict(metrics[0]) + if len(metrics) > 1: + raise ValueError("verify-metric 仅支持单个申报案例；多个案例请放入 input 目录走 verify-batch-input") + case = dict(metrics[0]) if teacher_name: case.setdefault("teacherName", teacher_name) if department: case.setdefault("department", department) case.setdefault("applyRank", apply_rank) + final_rank = str(case.get("applyRank") or "").strip() + if final_rank not in KNOWN_RANKS: + message = f"申报级别只接受 {'、'.join(KNOWN_RANKS)}：{final_rank or '（未提供）'}" + log.warning("task_failed task_type=verify_metric error_code=RANK_INVALID stage=verify") + finish(status="failed", message=message, skill=SKILL_SLUG, error_code="RANK_INVALID", stage="verify") + print(f"ERROR:RANK_INVALID {message}") + return 1 result = run_verify([case]) brief = result["reports"][0] @@ -419,9 +429,12 @@ def cmd_verify_metric( def cmd_verify_batch_input() -> int: - """批量核验（async Action 入口）：核验 input/ 目录中的申报案例 JSON。""" + """批量核验（async Action 入口）：核验 input/ 目录中的申报案例 JSON。 + + 坏文件（解析失败/非对象数组）跳过并告警，不中断整批；级别非法的案例由内核报错。 + """ log = _get_task_logger() - from service.verify_service import run_verify + from service.verify_service import collect_input_cases, run_verify from util.runtime_paths import get_input_dir with job_context(skill=SKILL_SLUG): @@ -436,26 +449,17 @@ def cmd_verify_batch_input() -> int: return 1 input_dir = get_input_dir() - files = sorted( - os.path.join(input_dir, name) - for name in os.listdir(input_dir) - if name.lower().endswith(".json") - ) - if not files: - message = f"input 目录没有申报案例 JSON：{input_dir}" + cases, skipped_files = collect_input_cases(input_dir) + for entry in skipped_files: + log.warning("batch_input_file_skipped file=%s reason=%s stage=batch", entry["file"], entry["reason"]) + emit(f"{entry['file']}：{entry['reason']}，该文件已跳过", type="warn", skill=SKILL_SLUG, stage="batch") + if not cases: + message = f"input 目录没有可用的申报案例 JSON（{len(skipped_files)} 个文件无法解析）：{input_dir}" log.warning("task_failed task_type=verify_batch error_code=METRIC_INVALID stage=batch") finish(status="failed", message=message, skill=SKILL_SLUG, error_code="METRIC_INVALID", stage="batch") print(f"ERROR:METRIC_INVALID {message}") return 1 - cases = [] - for path in files: - with open(path, encoding="utf-8") as f: - payload = json.load(f) - if isinstance(payload, dict): - payload = payload.get("cases") or [payload] - cases.extend(payload) - result = run_verify(cases) for index, brief in enumerate(result["reports"], start=1): emit( @@ -468,6 +472,7 @@ def cmd_verify_batch_input() -> int: "batch_no": result["batch_no"], "count": result["count"], "verdict_counts": result["verdict_counts"], + "skipped_files": skipped_files, "brief_file": result["brief_file"], }) finish(status="success", message=f"批量核验完成 {result['count']} 个案例", skill=SKILL_SLUG, stage="batch", batch_no=result["batch_no"]) diff --git a/scripts/service/threshold_service.py b/scripts/service/threshold_service.py index 2e70dc4..b2d9ed2 100644 --- a/scripts/service/threshold_service.py +++ b/scripts/service/threshold_service.py @@ -6,12 +6,11 @@ import json from typing import Any, Optional from db.connection import get_conn, init_db +from db.seed_data import KNOWN_RANKS as _KNOWN_RANKS from service.verify_service import _METRIC_ALIASES, _snake, norm_metrics from util.logging_config import get_skill_logger from util.timeutil import now_unix -_KNOWN_RANKS = ("副教授", "教授") - def _log(): try: diff --git a/scripts/service/verify_service.py b/scripts/service/verify_service.py index 52d071b..83446da 100644 --- a/scripts/service/verify_service.py +++ b/scripts/service/verify_service.py @@ -19,6 +19,7 @@ import uuid from typing import Any, Optional from db.connection import get_conn, init_db +from db.seed_data import KNOWN_RANKS from util.logging_config import get_skill_logger from util.timeutil import now_unix @@ -39,6 +40,7 @@ _METRIC_ALIASES = { "applyRank": ("apply_rank", "申报级别", "申报职称"), "presidingProvincial": ("presiding_provincial", "主持省部级项目", "省部级项目"), "corePapers": ("core_papers", "核心期刊论文", "核心论文"), + "firstAuthorCorePapers": ("first_author_core_papers", "核心期刊一作/通讯论文", "一作/通讯论文"), "teachingHours": ("teaching_hours", "年均教学工作量", "教学工作量"), "currentTitleYears": ("current_title_years", "任现职年限", "现职称年限"), "monographs": ("monographs", "专著/教材", "专著"), @@ -165,25 +167,43 @@ def verify_metrics(metrics_raw: dict[str, Any], thresholds: dict[str, dict[str, "advice": f"{advice}（依据：{basis}）" if advice else "", }) - # V1：申报指标中存在但门槛库未配置的键（排除标识字段与其别名形式） + # V1：申报指标中存在、但该级别门槛库未配置的指标。 + # 覆盖判定以"该级别门槛键 + 其全部别名形式"为准——已知别名但本级别未配门槛 + # 的指标同样要出 V1 灰行，绝不能当作已核验而漏掉。 identity_keys = {"teacher_name", "teacherName", "department", "apply_rank", "applyRank"} - known_keys = set(thresholds.keys()) | set(_METRIC_ALIASES.keys()) - unconfigured = [] + + def _alias_canonical(key: str) -> Optional[str]: + for canonical, names in _METRIC_ALIASES.items(): + if key == canonical or key == _snake(canonical) or key in names: + return canonical + return None + + covered_names: set[str] = set() + for tk in thresholds: + covered_names.add(tk) + covered_names.add(_camel(tk)) + for canonical, names in _METRIC_ALIASES.items(): + if canonical in thresholds or _snake(canonical) in thresholds: + covered_names.add(canonical) + covered_names.add(_snake(canonical)) + covered_names.update(names) + + seen_v1: set[str] = set() for key in metrics: - if key in identity_keys or key in known_keys: - continue - if _snake(key) in known_keys or _camel(key) in known_keys: + if key in identity_keys or key in covered_names: continue if metrics.get(key) in (None, ""): continue - if any(key in names for names in _METRIC_ALIASES.values()): + canonical = _alias_canonical(key) + v1_key = _snake(canonical) if canonical else str(key) + if v1_key in thresholds or v1_key in seen_v1: continue - unconfigured.append(str(key)) - for key in unconfigured: + seen_v1.add(v1_key) has_yellow_or_gray = True + display = str(_METRIC_ALIASES[canonical][0]) if canonical else str(key) matrix.append({ - "metric": str(_METRIC_ALIASES.get(key, (key,))[0]) if key in _METRIC_ALIASES else key, - "metric_key": key, + "metric": display, + "metric_key": v1_key, "required": None, "actual": metrics.get(key), "margin": None, @@ -195,9 +215,12 @@ def verify_metrics(metrics_raw: dict[str, Any], thresholds: dict[str, dict[str, if has_red: verdict = VERDICT_RED summary = "存在不达门槛的硬伤项，暂缓提交；请按矩阵补足材料后重新核验。" - elif has_yellow_or_gray: + elif has_yellow_or_gray or not thresholds: verdict = VERDICT_BORDER - summary = "无硬伤，但存在边缘项或待核数据，建议补充说明后提交。" + if not thresholds: + summary = f"门槛库未配置「{apply_rank}」级别的门槛，本次无法完成硬指标核验，请人事处补充配置后重新核验（V1）。" + else: + summary = "无硬伤，但存在边缘项或待核数据，建议补充说明后提交。" else: verdict = VERDICT_SAFE summary = "全部硬指标超标准 10% 以上，形式审查通过。" @@ -212,6 +235,35 @@ def verify_metrics(metrics_raw: dict[str, Any], thresholds: dict[str, dict[str, } +def collect_input_cases(input_dir: str) -> tuple[list[dict[str, Any]], list[dict[str, str]]]: + """读取 input/ 目录下的申报案例 JSON；坏文件跳过并记录原因，不中断整批。 + + 返回 (案例列表, 跳过文件列表)；跳过文件形如 {"file": 名称, "reason": 原因}。 + """ + import os + + files = sorted( + os.path.join(input_dir, name) + for name in os.listdir(input_dir) + if name.lower().endswith(".json") + ) + cases: list[dict[str, Any]] = [] + skipped_files: list[dict[str, str]] = [] + for path in files: + name = os.path.basename(path) + try: + with open(path, encoding="utf-8") as f: + payload = json.load(f) + if isinstance(payload, dict): + payload = payload.get("cases") or [payload] + if not isinstance(payload, list): + raise ValueError("内容不是申报案例对象数组") + cases.extend(item for item in payload if isinstance(item, dict)) + except (OSError, ValueError) as exc: + skipped_files.append({"file": name, "reason": str(exc)}) + return cases, skipped_files + + def run_verify(cases: list[dict[str, Any]], *, batch_no: str = "") -> dict[str, Any]: """核验内核：逐案例矩阵计算 → 写库 → 生成《核验报告》。""" from service.report_service import write_report_for_kernel, build_batch_brief @@ -234,7 +286,12 @@ def run_verify(cases: list[dict[str, Any]], *, batch_no: str = "") -> dict[str, for case in cases: if not isinstance(case, dict): raise ValueError("申报案例必须是 JSON 对象") - result = verify_metrics(case, load_thresholds(str(case.get("applyRank") or case.get("apply_rank") or "").strip())) + rank = str(case.get("applyRank") or case.get("apply_rank") or "").strip() + if rank not in KNOWN_RANKS: + raise ValueError( + f"申报级别只接受 {'、'.join(KNOWN_RANKS)}：{rank or '（未提供）'}（RANK_INVALID）" + ) + result = verify_metrics(case, load_thresholds(rank)) cur.execute( """ INSERT INTO verify_cases diff --git a/tests/test_verify_rules.py b/tests/test_verify_rules.py index c0b9251..ae51fa7 100644 --- a/tests/test_verify_rules.py +++ b/tests/test_verify_rules.py @@ -2,6 +2,9 @@ """职称硬指标核验内核金标准用例（离线、确定性）。""" from __future__ import annotations +import json +import os +import tempfile import unittest from _support import IsolatedDataRoot @@ -65,6 +68,33 @@ class TestMatrixColors(VerifyRuleTestBase): colors = {m["metric_key"]: m["color"] for m in result["matrix"]} self.assertEqual(colors["teaching_hours"], "green") + def test_yellow_boundary_at_109_percent(self) -> None: + # 1.09 倍（174.4 = 160×1.09，余量 9%）属黄 + result = self._verify(_metrics(teachingHours=174.4)) + colors = {m["metric_key"]: m["color"] for m in result["matrix"]} + self.assertEqual(colors["teaching_hours"], "yellow") + + def test_alias_metric_without_threshold_gets_v1(self) -> None: + # 副教授种子未配置 awards：提供该指标须出 V1 灰行，不得当作已核验跳过 + result = self._verify(_metrics(awards=2)) + v1 = [m for m in result["matrix"] if m["metric_key"] == "awards"] + self.assertEqual(len(v1), 1) + self.assertIn("V1", v1[0]["advice"]) + + def test_empty_thresholds_never_safe(self) -> None: + """该级别门槛库为空 → 一律不判"安全"（V1 灰行 + 边缘结论）。""" + from service.verify_service import verify_metrics + + result = verify_metrics(_metrics(), {}) + self.assertEqual(result["verdict"], "边缘") + self.assertIn("门槛库未配置", result["summary"]) + provided = [m for m in result["matrix"] if m["metric_key"] in ("presiding_provincial", "core_papers")] + self.assertTrue(provided and all(m["color"] == "gray" for m in provided)) + + bare = verify_metrics({"teacherName": "某人", "applyRank": "副教授"}, {}) + self.assertEqual(bare["verdict"], "边缘") + self.assertIn("门槛库未配置", bare["summary"]) + def test_r4_missing_metric_pending(self) -> None: result = self._verify(_metrics(corePapers=None)) self.assertEqual(result["verdict"], "边缘") @@ -106,6 +136,26 @@ class TestVerifyKernel(VerifyRuleTestBase): result = self._run([_metrics(rank="教授", presidingProvincial=2, corePapers=7, monographs=2)]) self.assertEqual(result["reports"][0]["verdict"], "边缘") # core 7/6 余量 16.7% 绿、presiding 2/2 黄 + def test_rank_invalid_raises(self) -> None: + with self.assertRaises(ValueError) as ctx: + self._run([_metrics(rank="院士", presidingProvincial=0)]) + self.assertIn("RANK_INVALID", str(ctx.exception)) + + def test_professor_first_author_threshold_enforced(self) -> None: + """教授"一作/通讯 ≥3"为独立门槛：缺数据判待核（边缘），提供后参与判定。""" + from service.verify_service import load_thresholds, verify_metrics + + base = _metrics(rank="教授", presidingProvincial=3, corePapers=7, monographs=2, currentTitleYears=6) + without = verify_metrics(base, load_thresholds("教授")) + self.assertEqual(without["verdict"], "边缘") + pending = next(m for m in without["matrix"] if m["metric_key"] == "first_author_core_papers") + self.assertEqual(pending["color"], "gray") + + full = verify_metrics(dict(base, firstAuthorCorePapers=4), load_thresholds("教授")) + fa = next(m for m in full["matrix"] if m["metric_key"] == "first_author_core_papers") + self.assertEqual(fa["color"], "green") + self.assertEqual(fa["required"], 3.0) + def test_set_threshold_changes_verdict(self) -> None: from service.threshold_service import upsert_threshold @@ -152,5 +202,37 @@ class TestThresholdService(VerifyRuleTestBase): self.assertTrue(payload["red_cases"]) +class TestBatchInputCollection(VerifyRuleTestBase): + """input/ 目录收集：坏文件跳过并记录原因，不中断整批。""" + + def test_collect_input_cases_skips_bad_files(self) -> None: + from service.verify_service import collect_input_cases + + with tempfile.TemporaryDirectory() as tmp: + with open(os.path.join(tmp, "a.json"), "w", encoding="utf-8") as f: + json.dump([_metrics(), _metrics(teacherName="李老师")], f, ensure_ascii=False) + with open(os.path.join(tmp, "single.json"), "w", encoding="utf-8") as f: + json.dump({"teacherName": "王老师", "applyRank": "副教授"}, f, ensure_ascii=False) + with open(os.path.join(tmp, "bad.json"), "w", encoding="utf-8") as f: + f.write("{not-json") + with open(os.path.join(tmp, "scalar.json"), "w", encoding="utf-8") as f: + json.dump("just-a-string", f, ensure_ascii=False) + + cases, skipped = collect_input_cases(tmp) + + names = [c.get("teacherName") for c in cases] + self.assertEqual(names, ["张老师", "李老师", "王老师"]) + self.assertEqual({e["file"] for e in skipped}, {"bad.json", "scalar.json"}) + self.assertTrue(all(e["reason"] for e in skipped)) + + def test_collect_input_cases_empty_dir(self) -> None: + from service.verify_service import collect_input_cases + + with tempfile.TemporaryDirectory() as tmp: + cases, skipped = collect_input_cases(tmp) + self.assertEqual(cases, []) + self.assertEqual(skipped, []) + + if __name__ == "__main__": unittest.main()