niuniu/gaokao-volunteer-system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(audit): validate majors in structured audit flow
Some checks failed
CI / pytest (Python 3.10) (push) Has been cancelled
Details
CI / pytest (Python 3.11) (push) Has been cancelled
Details
CI / pytest (Python 3.12) (push) Has been cancelled
Details

Browse Source
This commit is contained in:
Hermes Agent
2026-06-17 16:02:31 +08:00
parent 6b1157fe19
commit edc5b11230
5 changed files with 376 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
data/rules/audit_engine.py
View File

@@ -1,6 +1,9 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
from data.majors_catalog.loader import MajorsCatalogLoader
from .loader import RuleLoader
from .models import LoadedRule
@@ -17,8 +20,14 @@ class ProvinceRuleSnapshot:
class AuditEngine:
def __init__(self, loader: RuleLoader) -> None:
def __init__(
self,
loader: RuleLoader,
*,
majors_loader: MajorsCatalogLoader | None = None,
) -> None:
self._loader = loader
self._majors_loader = majors_loader
def get_province_snapshot(self, province: str) -> ProvinceRuleSnapshot:
rules = self._loader.list_province_rules(province)
@@ -32,6 +41,44 @@ class AuditEngine:
rules=rules,
)
def audit_plan(self, province: str, plan: dict[str, Any]) -> dict[str, object]:
issues: list[dict[str, object]] = []
if self._majors_loader is not None:
issues.extend(self._validate_majors(plan))
return {
"province": province,
"overall_pass": not issues,
"issues": issues,
}
def _validate_majors(self, plan: dict[str, Any]) -> list[dict[str, object]]:
if self._majors_loader is None:
return []
issues: list[dict[str, object]] = []
for item in plan.get("items", []):
school_name = item.get("school_name") or "未知院校"
for major_name in item.get("major_names", []):
major = self._majors_loader.lookup(str(major_name))
if major is None:
issues.append({
"rule_id": "MAJORS.not_found",
"severity": "warning",
"title": f"专业未在国家级目录中找到: {major_name}",
"message": f"{school_name} / {major_name} 未命中国家级目录",
"suggestion": "请人工核对专业名称或补充目录数据",
})
continue
if major.status != "active":
issues.append({
"rule_id": "MAJORS.non_active",
"severity": "critical",
"title": f"专业状态不是 active: {major.name}",
"message": f"{school_name} / {major.name} 当前状态为 {major.status}",
"suggestion": "请改用仍在招生/有效的专业，并核对最新目录",
})
return issues
def _scalar(rule: LoadedRule | None, key: str):
if rule is None:

22
data/rules/cli.py
View File

@@ -13,6 +13,8 @@ from data.majors_catalog.cli import (
build_status_payload as build_majors_status_payload,
build_verify_payload as build_majors_verify_payload,
)
from data.majors_catalog.loader import MajorsCatalogLoader
from data.rules.audit_engine import AuditEngine
from data.rules.loader import RuleLoader
@@ -76,6 +78,15 @@ def _build_parser() -> argparse.ArgumentParser:
school_verify.add_argument("--year", required=True, type=int)
school_verify.add_argument("--json", action="store_true")
audit_parser = subparsers.add_parser("audit", help="structured audit commands")
audit_sub = audit_parser.add_subparsers(dest="audit_command", required=True)
audit_run = audit_sub.add_parser("run", help="run structured audit on JSON plan")
audit_run.add_argument("--province", required=True)
audit_run.add_argument("--plan", required=True)
audit_run.add_argument("--truth-root", default=str(DEFAULT_RULES_TRUTH_ROOT))
audit_run.add_argument("--catalog-root", default=str(DEFAULT_CATALOG_ROOT))
audit_run.add_argument("--json", action="store_true")
return parser
@@ -84,7 +95,7 @@ def _emit(payload: dict[str, object], json_output: bool) -> int:
print(json.dumps(payload, ensure_ascii=False))
else:
print(payload)
return 0 if bool(payload.get("ok", True)) else 1
return 0 if bool(payload.get("ok", payload.get("overall_pass", True))) else 1
def main(argv: list[str] | None = None) -> int:
@@ -147,6 +158,15 @@ def main(argv: list[str] | None = None) -> int:
payload = build_school_verify_payload(Path(args.catalog_root), args.year)
return _emit(payload, args.json)
if args.command == "audit" and args.audit_command == "run":
truth_loader = RuleLoader.from_truth_root(Path(args.truth_root))
majors_loader = MajorsCatalogLoader.from_catalog_root(Path(args.catalog_root))
plan = json.loads(Path(args.plan).read_text(encoding="utf-8"))
payload = AuditEngine(truth_loader, majors_loader=majors_loader).audit_plan(
args.province, plan
)
return _emit(payload, args.json)
parser.error("unsupported command")
return 2

40
docs/CURRENT_STATE.md
View File

@@ -1,24 +1,24 @@
# CURRENT_STATE
最后更新: 2026-06-17
状态词: 本地验证完成（v2.1 已交付增强 + Phase1 已完成 + Phase2 专业目录 MVP Batch1 已本地闭环）
状态词: 本地验证完成（Phase2 major_validation 已接入 + 目录/审核闭环已收口）
真相源优先级:
1. 本文件
2. docs/plans/2026-06-17-phase2-majors-catalog-implementation-plan.md（当前 Phase2 专业目录执行计划）
3. docs/PROJECT_PLANNING_REALIGNMENT_2026-06-16.md（本轮新增的规划/实现漂移审计）
4. docs/DESIGN_RULES_TRUSTED_CLI_2026-06-16.md（本轮新增的下一阶段设计：规则可信化 + CLI 能力层）
5. docs/RULES_SOURCE_OF_TRUTH.md（本轮新增的规则真相源索引）
6. docs/MAJOR_DATA_SOURCE_OF_TRUTH.md（本轮新增的专业目录真相源索引）
7. docs/CLI_API_MAPPING.md（本轮新增的 CLI/API 映射索引）
8. docs/FINAL_COMPLETION_REPORT_2026-06-13.md
9. docs/P0_P1_P2_REMEDIATION_PLAN_2026-06-14.md（当前整改板；2026-06-15 已做状态归一）
10. docs/FRONTEND_UI_AUDIT_2026-06-16.md
11. docs/FRONTEND_UI_EXECUTION_BOARD_2026-06-16.md
12. docs/ACTIVE_REMEDIATION_2026-06-13.md
13. product/PRD.md / product/ROADMAP.md / docs/IMPLEMENTATION_PLAN_v2.md
14. docs/PRODUCTION_DEPLOYMENT_CHECKLIST_2026-06-15.md
15. reports/PRODUCT_PLANNING_TECH_ALIGNMENT_REVIEW_2026-06-13.md（历史评审快照）
2. `docs/plans/2026-06-17-phase2-majors-catalog-implementation-plan.md`（当前 Phase2 专业目录执行计划）
3. `docs/DESIGN_RULES_TRUSTED_CLI_2026-06-16.md`（专业目录/CLI 总设计，当前被 Phase2 计划补充）
4. `docs/PROJECT_PLANNING_REALIGNMENT_2026-06-16.md`（规划/实现漂移审计）
5. `docs/RULES_SOURCE_OF_TRUTH.md`（规则真相源索引）
6. `docs/MAJOR_DATA_SOURCE_OF_TRUTH.md`（专业目录真相源索引）
7. `docs/CLI_API_MAPPING.md`（CLI/API 映射索引）
8. docs/P0_P1_P2_REMEDIATION_PLAN_2026-06-14.md（当前整改板；2026-06-15 已做状态归一）
9. docs/FRONTEND_UI_AUDIT_2026-06-16.md
10. docs/FRONTEND_UI_EXECUTION_BOARD_2026-06-16.md
11. docs/ACTIVE_REMEDIATION_2026-06-13.md
12. product/PRD.md / product/ROADMAP.md / docs/IMPLEMENTATION_PLAN_v2.md
13. docs/PRODUCTION_DEPLOYMENT_CHECKLIST_2026-06-15.md
14. reports/PRODUCT_PLANNING_TECH_ALIGNMENT_REVIEW_2026-06-13.md（历史评审快照）
---
@@ -28,9 +28,13 @@
- 已完成: 人工服务运营增强系统（闲鱼 / 微信 / 学校渠道）
- 已完成: 管理后台、订单、分享、渠道同步、AI 审核、CI/CD、性能与安全加固
- 未完成: 用户端 Web 自助支付 / 资料填写 / 站内交付的完整商业闭环（T12 进行中）
- 已完成且已本地验证: `mock` / `alipay_sim` / `alipay` 三层支付代码链与 notify/return 路由；退款状态闭环、portal token secret 分离、payment webhook fail-closed、删除/匿名化扩围、分享 allowlist、channel_sync DB 隔离与 DAO 真相收敛
- 未完成且仍阻塞线上验收: 真实支付宝商户凭据、公网 notify_url、备案域名、真实支付 acceptance
- 已完成/已验证:
- `Phase2 major_validation` 已接入 `AuditEngine`
- `gaokao-cli audit run --json` 已可基于专业目录做 major 校验
- 规则/专业目录/CLI 三层都已具备可回归验证入口
- 当前仍在推进:
- 已完成且已本地验证: `mock` / `alipay_sim` / `alipay` 三层支付代码链与 notify/return 路由；退款状态闭环、portal token secret 分离、payment webhook fail-closed、删除/匿名化扩围、分享 allowlist、channel_sync DB 隔离与 DAO 真相收敛
- 未完成且仍阻塞线上验收: 真实支付宝商户凭据、公网 notify_url、备案域名、真实支付 acceptance
一句话：

163
tests/test_audit_cli_major_validation_phase2.py Normal file
View File

@@ -0,0 +1,163 @@
from __future__ import annotations
import json
import subprocess
import sys
from pathlib import Path
PROJECT_ROOT = Path(__file__).resolve().parents[1]
SCRIPT_PATH = PROJECT_ROOT / "scripts" / "gaokao-cli"
def _write_truth_and_catalog(root: Path) -> tuple[Path, Path, Path]:
truth_root = root / "truth"
province_dir = truth_root / "province"
province_dir.mkdir(parents=True, exist_ok=True)
(truth_root / "national.yaml").write_text(
"scope: national\nyear: 2026\nversion: '2026.1'\nrules: {}\n",
encoding="utf-8",
)
(province_dir / "hunan.yaml").write_text(
"scope: province\nprovince: 湖南\nyear: 2026\nversion: '2026.1'\nstatus: active\nrules:\n max_volunteers:\n title: 志愿上限\n severity: fatal\n value:\n max_volunteers: 45\n source_evidence_id: hunan-2026-max-volunteers\n effective_date: '2026-01-01'\n status: active\n",
encoding="utf-8",
)
catalog_root = root / "catalog"
national = catalog_root / "national"
national.mkdir(parents=True, exist_ok=True)
payload = {
"year": 2024,
"version": "2024.1",
"coverage_mode": "mvp_subset",
"source": "教育部普通高等学校本科专业目录（人工摘录校对）",
"source_url": "https://www.moe.gov.cn/",
"last_verified_at": "2026-06-17",
"majors": [
{
"code": "080901",
"name": "计算机科学与技术",
"discipline": "工学",
"category": "计算机类",
"degree": "工学学士",
"is_directional": False,
"status": "active",
"year_added": 1998,
"year_removed": None,
"notes": None,
},
{
"code": "120201K",
"name": "工商管理",
"discipline": "管理学",
"category": "工商管理类",
"degree": "管理学学士",
"is_directional": True,
"status": "deprecated",
"year_added": 1998,
"year_removed": 2024,
"notes": "示例停用",
},
],
}
for name in ("2024.json", "latest.json"):
(national / name).write_text(
json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8"
)
plan_path = root / "plan.json"
plan_path.write_text(
json.dumps(
{
"province": "湖南",
"items": [
{
"school_name": "北京大学",
"major_names": ["计算机科学与技术", "不存在专业", "工商管理"],
}
],
},
ensure_ascii=False,
indent=2,
),
encoding="utf-8",
)
return truth_root, catalog_root, plan_path
def test_audit_run_cli_reports_major_validation_issues(tmp_path: Path) -> None:
truth_root, catalog_root, plan_path = _write_truth_and_catalog(tmp_path)
result = subprocess.run(
[
sys.executable,
str(SCRIPT_PATH),
"audit",
"run",
"--province",
"湖南",
"--plan",
str(plan_path),
"--truth-root",
str(truth_root),
"--catalog-root",
str(catalog_root),
"--json",
],
cwd=PROJECT_ROOT,
capture_output=True,
text=True,
check=False,
)
assert result.returncode == 1, result.stderr
payload = json.loads(result.stdout)
rule_ids = {issue["rule_id"] for issue in payload["issues"]}
assert payload["overall_pass"] is False
assert "MAJORS.not_found" in rule_ids
assert "MAJORS.non_active" in rule_ids
def test_audit_run_cli_returns_zero_when_plan_majors_are_all_active(
tmp_path: Path,
) -> None:
truth_root, catalog_root, _ = _write_truth_and_catalog(tmp_path)
plan_path = tmp_path / "ok-plan.json"
plan_path.write_text(
json.dumps(
{
"province": "湖南",
"items": [
{"school_name": "北京大学", "major_names": ["计算机科学与技术"]}
],
},
ensure_ascii=False,
indent=2,
),
encoding="utf-8",
)
result = subprocess.run(
[
sys.executable,
str(SCRIPT_PATH),
"audit",
"run",
"--province",
"湖南",
"--plan",
str(plan_path),
"--truth-root",
str(truth_root),
"--catalog-root",
str(catalog_root),
"--json",
],
cwd=PROJECT_ROOT,
capture_output=True,
text=True,
check=False,
)
assert result.returncode == 0, result.stderr
payload = json.loads(result.stdout)
assert payload["overall_pass"] is True
assert payload["issues"] == []

122
tests/test_audit_engine_major_validation_phase2.py Normal file
View File

@@ -0,0 +1,122 @@
from __future__ import annotations
import json
from pathlib import Path
from typing import Any, cast
from data.majors_catalog.loader import MajorsCatalogLoader
from data.rules.audit_engine import AuditEngine
from data.rules.loader import RuleLoader
def _write_truth_and_catalog(root: Path) -> tuple[Path, Path, Path]:
truth_root = root / "truth"
province_dir = truth_root / "province"
province_dir.mkdir(parents=True, exist_ok=True)
(truth_root / "national.yaml").write_text(
"scope: national\nyear: 2026\nversion: '2026.1'\nrules: {}\n",
encoding="utf-8",
)
(province_dir / "hunan.yaml").write_text(
"scope: province\nprovince: 湖南\nyear: 2026\nversion: '2026.1'\nstatus: active\nrules:\n max_volunteers:\n title: 志愿上限\n severity: fatal\n value:\n max_volunteers: 45\n source_evidence_id: hunan-2026-max-volunteers\n effective_date: '2026-01-01'\n status: active\n",
encoding="utf-8",
)
catalog_root = root / "catalog"
national = catalog_root / "national"
national.mkdir(parents=True, exist_ok=True)
payload = {
"year": 2024,
"version": "2024.1",
"coverage_mode": "mvp_subset",
"source": "教育部普通高等学校本科专业目录（人工摘录校对）",
"source_url": "https://www.moe.gov.cn/",
"last_verified_at": "2026-06-17",
"majors": [
{
"code": "080901",
"name": "计算机科学与技术",
"discipline": "工学",
"category": "计算机类",
"degree": "工学学士",
"is_directional": False,
"status": "active",
"year_added": 1998,
"year_removed": None,
"notes": None,
},
{
"code": "120201K",
"name": "工商管理",
"discipline": "管理学",
"category": "工商管理类",
"degree": "管理学学士",
"is_directional": True,
"status": "deprecated",
"year_added": 1998,
"year_removed": 2024,
"notes": "示例停用",
},
],
}
for name in ("2024.json", "latest.json"):
(national / name).write_text(
json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8"
)
plan_path = root / "plan.json"
plan_path.write_text(
json.dumps(
{
"province": "湖南",
"items": [
{
"school_name": "北京大学",
"major_names": ["计算机科学与技术", "不存在专业", "工商管理"],
}
],
},
ensure_ascii=False,
indent=2,
),
encoding="utf-8",
)
return truth_root, catalog_root, plan_path
def test_audit_engine_major_validation_marks_missing_and_non_active_majors(
tmp_path: Path,
) -> None:
truth_root, catalog_root, plan_path = _write_truth_and_catalog(tmp_path)
loader = RuleLoader.from_truth_root(truth_root)
catalog = MajorsCatalogLoader.from_catalog_root(catalog_root)
engine = AuditEngine(loader, majors_loader=catalog)
plan = json.loads(plan_path.read_text(encoding="utf-8"))
result = engine.audit_plan("湖南", plan)
issues = cast(list[dict[str, Any]], result["issues"])
rule_ids = {issue["rule_id"] for issue in issues}
assert "MAJORS.not_found" in rule_ids
assert "MAJORS.non_active" in rule_ids
assert result["overall_pass"] is False
def test_audit_engine_major_validation_passes_when_all_majors_are_active(
tmp_path: Path,
) -> None:
truth_root, catalog_root, _ = _write_truth_and_catalog(tmp_path)
loader = RuleLoader.from_truth_root(truth_root)
catalog = MajorsCatalogLoader.from_catalog_root(catalog_root)
engine = AuditEngine(loader, majors_loader=catalog)
result = engine.audit_plan(
"湖南",
{
"province": "湖南",
"items": [{"school_name": "北京大学", "major_names": ["计算机科学与技术"]}],
},
)
assert result["issues"] == []
assert result["overall_pass"] is True