346 lines
13 KiB
Python
346 lines
13 KiB
Python
"""职称硬指标核验内核（唯一业务内核）。
|
||
|
||
核验规则（对齐《核心建设内容：各部门"Skill"深度需求规格书》人事处 Skill 001）：
|
||
R1 逐项比对：实有 < 最低要求 → 红（硬伤，附依据条款）
|
||
R2 边缘识别：达标但余量 < 10% → 黄（建议补充说明或再积累）
|
||
R3 安全判定：全部已核项 ≥ 要求 × 1.10 → 绿
|
||
R4 数据待核：指标缺失 → 灰（不判不达标，提示补材料）
|
||
R5 综合结论：任一红 → 硬伤（暂缓提交）；有黄/灰 → 边缘（建议补充）；全绿 → 安全
|
||
R6 基础条件：学历/任现职年限与门槛同口径比对
|
||
V1 门槛库完备性：指标无对应门槛 → 请人事处补充配置
|
||
|
||
核验仅针对硬指标形式审查，不构成评审结论。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
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
|
||
|
||
VERDICT_SAFE = "安全"
|
||
VERDICT_BORDER = "边缘"
|
||
VERDICT_RED = "硬伤"
|
||
|
||
_COLOR_PASS = "green"
|
||
_COLOR_BORDER = "yellow"
|
||
_COLOR_RED = "red"
|
||
_COLOR_PENDING = "gray"
|
||
|
||
_SAFE_MARGIN = 0.10
|
||
|
||
_METRIC_ALIASES = {
|
||
"teacherName": ("teacher_name", "姓名", "教师姓名"),
|
||
"department": ("department", "学院", "所属学院"),
|
||
"applyRank": ("apply_rank", "申报级别", "申报职称"),
|
||
"presidingProvincial": ("presiding_provincial", "主持省部级项目", "省部级项目"),
|
||
"corePapers": ("core_papers", "核心期刊论文", "核心论文"),
|
||
"firstAuthorCorePapers": ("first_author_core_papers", "核心期刊一作/通讯论文", "一作/通讯论文"),
|
||
"teachingHours": ("teaching_hours", "年均教学工作量", "教学工作量"),
|
||
"currentTitleYears": ("current_title_years", "任现职年限", "现职称年限"),
|
||
"monographs": ("monographs", "专著/教材", "专著"),
|
||
"awards": ("awards", "教学/科研获奖", "获奖"),
|
||
"degree": ("degree", "学历学位", "学历"),
|
||
}
|
||
|
||
|
||
def _log():
|
||
try:
|
||
return get_skill_logger()
|
||
except RuntimeError:
|
||
return None
|
||
|
||
|
||
def _camel(name: str) -> str:
|
||
parts = str(name).split("_")
|
||
if len(parts) <= 1:
|
||
return str(name)
|
||
head, *rest = parts
|
||
return head + "".join(p[:1].upper() + p[1:] for p in rest if p)
|
||
|
||
|
||
def _snake(name: str) -> str:
|
||
return "".join("_" + ch.lower() if ch.isupper() else ch for ch in str(name)).lstrip("_")
|
||
|
||
|
||
def norm_metrics(raw: dict[str, Any]) -> dict[str, Any]:
|
||
"""归一化申报指标键（兼容 camelCase / snake_case / 中文键，canonical 同时提供两种形式）。"""
|
||
out: dict[str, Any] = {}
|
||
for key, value in raw.items():
|
||
out.setdefault(str(key), value)
|
||
for canonical, names in _METRIC_ALIASES.items():
|
||
if str(key) in names or _camel(str(key)) == canonical:
|
||
out.setdefault(canonical, value)
|
||
out.setdefault(_snake(canonical), value)
|
||
return out
|
||
|
||
|
||
def _to_float(value: Any) -> Optional[float]:
|
||
if value is None or value == "":
|
||
return None
|
||
try:
|
||
return float(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def load_thresholds(rank: str) -> dict[str, dict[str, Any]]:
|
||
"""读取某级别的门槛（metric_key → 门槛行）。"""
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"SELECT metric_key, metric_name, min_value, unit, basis_doc, clause FROM threshold_rules WHERE rank = ?",
|
||
(rank,),
|
||
)
|
||
thresholds = {}
|
||
for metric_key, metric_name, min_value, unit, basis_doc, clause in cur.fetchall():
|
||
thresholds[str(metric_key)] = {
|
||
"metric_name": metric_name,
|
||
"min_value": float(min_value),
|
||
"unit": unit or "",
|
||
"basis_doc": basis_doc or "",
|
||
"clause": clause or "",
|
||
}
|
||
finally:
|
||
conn.close()
|
||
return thresholds
|
||
|
||
|
||
def verify_metrics(metrics_raw: dict[str, Any], thresholds: dict[str, dict[str, Any]]) -> dict[str, Any]:
|
||
"""对单个申报案例执行矩阵计算（纯计算，不落库）。"""
|
||
metrics = norm_metrics(metrics_raw)
|
||
teacher_name = str(metrics.get("teacherName") or "").strip() or "未命名申报人"
|
||
apply_rank = str(metrics.get("applyRank") or "").strip()
|
||
|
||
matrix: list[dict[str, Any]] = []
|
||
has_red = False
|
||
has_yellow_or_gray = False
|
||
|
||
for metric_key, threshold in thresholds.items():
|
||
actual_raw = metrics.get(metric_key)
|
||
required = float(threshold["min_value"])
|
||
unit = threshold["unit"]
|
||
basis = threshold["basis_doc"] + (threshold["clause"] or "")
|
||
display_actual = metrics.get(metric_key)
|
||
|
||
actual = _to_float(actual_raw)
|
||
if actual is None:
|
||
has_yellow_or_gray = True
|
||
matrix.append({
|
||
"metric": threshold["metric_name"],
|
||
"metric_key": metric_key,
|
||
"required": required,
|
||
"actual": display_actual,
|
||
"margin": None,
|
||
"color": _COLOR_PENDING,
|
||
"basis": basis,
|
||
"advice": f"数据待核：请补充{threshold['metric_name']}证明材料后重新核验（R4）。",
|
||
})
|
||
continue
|
||
|
||
margin = round((actual - required) / required, 4) if required > 0 else None
|
||
if actual < required:
|
||
has_red = True
|
||
advice = f"硬伤：{threshold['metric_name']}实有 {actual:g} {unit}，未达最低要求 {required:g} {unit}，须补足后方可提交"
|
||
color = _COLOR_RED
|
||
elif margin is not None and margin < _SAFE_MARGIN:
|
||
has_yellow_or_gray = True
|
||
advice = f"边缘项：仅超线 {margin * 100:.1f}%，建议补充说明或继续积累（R2）"
|
||
color = _COLOR_BORDER
|
||
else:
|
||
advice = ""
|
||
color = _COLOR_PASS
|
||
matrix.append({
|
||
"metric": threshold["metric_name"],
|
||
"metric_key": metric_key,
|
||
"required": required,
|
||
"actual": actual,
|
||
"margin": margin,
|
||
"color": color,
|
||
"basis": basis,
|
||
"advice": f"{advice}（依据：{basis}）" if advice else "",
|
||
})
|
||
|
||
# V1：申报指标中存在、但该级别门槛库未配置的指标。
|
||
# 覆盖判定以"该级别门槛键 + 其全部别名形式"为准——已知别名但本级别未配门槛
|
||
# 的指标同样要出 V1 灰行，绝不能当作已核验而漏掉。
|
||
identity_keys = {"teacher_name", "teacherName", "department", "apply_rank", "applyRank"}
|
||
|
||
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 covered_names:
|
||
continue
|
||
if metrics.get(key) in (None, ""):
|
||
continue
|
||
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
|
||
seen_v1.add(v1_key)
|
||
has_yellow_or_gray = True
|
||
display = str(_METRIC_ALIASES[canonical][0]) if canonical else str(key)
|
||
matrix.append({
|
||
"metric": display,
|
||
"metric_key": v1_key,
|
||
"required": None,
|
||
"actual": metrics.get(key),
|
||
"margin": None,
|
||
"color": _COLOR_PENDING,
|
||
"basis": "",
|
||
"advice": "门槛库未配置该指标的最低要求，请人事处补充（V1）。",
|
||
})
|
||
|
||
if has_red:
|
||
verdict = VERDICT_RED
|
||
summary = "存在不达门槛的硬伤项，暂缓提交；请按矩阵补足材料后重新核验。"
|
||
elif has_yellow_or_gray or not thresholds:
|
||
verdict = VERDICT_BORDER
|
||
if not thresholds:
|
||
summary = f"门槛库未配置「{apply_rank}」级别的门槛，本次无法完成硬指标核验，请人事处补充配置后重新核验（V1）。"
|
||
else:
|
||
summary = "无硬伤，但存在边缘项或待核数据，建议补充说明后提交。"
|
||
else:
|
||
verdict = VERDICT_SAFE
|
||
summary = "全部硬指标超标准 10% 以上，形式审查通过。"
|
||
|
||
return {
|
||
"teacher_name": teacher_name,
|
||
"department": str(metrics.get("department") or ""),
|
||
"apply_rank": apply_rank,
|
||
"verdict": verdict,
|
||
"summary": summary,
|
||
"matrix": matrix,
|
||
}
|
||
|
||
|
||
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
|
||
|
||
log = _log()
|
||
init_db()
|
||
if not cases:
|
||
raise ValueError("没有可核验的申报案例")
|
||
if len(cases) > 200:
|
||
raise ValueError(f"单批核验案例数过多（{len(cases)}），上限 200 个")
|
||
|
||
batch_no = batch_no or now_unix().to_bytes(5, "big").hex() + uuid.uuid4().hex[:6]
|
||
if log:
|
||
log.info("verify_kernel_start batch_no=%s count=%s stage=kernel", batch_no, len(cases))
|
||
|
||
results = []
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
for case in cases:
|
||
if not isinstance(case, dict):
|
||
raise ValueError("申报案例必须是 JSON 对象")
|
||
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
|
||
(batch_no, teacher_name, department, apply_rank, metrics_json, verdict, matrix_json)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
batch_no,
|
||
result["teacher_name"],
|
||
result["department"],
|
||
result["apply_rank"],
|
||
json.dumps(case, ensure_ascii=False),
|
||
result["verdict"],
|
||
json.dumps(result["matrix"], ensure_ascii=False),
|
||
),
|
||
)
|
||
case_id = int(cur.lastrowid)
|
||
result["case_id"] = case_id
|
||
result["batch_no"] = batch_no
|
||
report_file = write_report_for_kernel(result=result, batch_no=batch_no)
|
||
cur.execute("UPDATE verify_cases SET report_file = ? WHERE id = ?", (report_file, case_id))
|
||
result["report_file"] = report_file
|
||
results.append(result)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
verdict_counts: dict[str, int] = {}
|
||
for result in results:
|
||
verdict_counts[result["verdict"]] = verdict_counts.get(result["verdict"], 0) + 1
|
||
brief_file = build_batch_brief(batch_no=batch_no, results=results)
|
||
|
||
payload = {
|
||
"ok": True,
|
||
"batch_no": batch_no,
|
||
"count": len(results),
|
||
"verdict_counts": verdict_counts,
|
||
"brief_file": brief_file,
|
||
"disclaimer": "核验仅针对硬指标形式审查，不构成评审结论；门槛以人事处最新文件为准。",
|
||
"reports": results,
|
||
}
|
||
if log:
|
||
log.info(
|
||
"verify_kernel_done batch_no=%s count=%s red=%s yellow=%s green=%s stage=kernel",
|
||
batch_no,
|
||
len(results),
|
||
verdict_counts.get(VERDICT_RED, 0),
|
||
verdict_counts.get(VERDICT_BORDER, 0),
|
||
verdict_counts.get(VERDICT_SAFE, 0),
|
||
)
|
||
return payload
|