# -*- coding: utf-8 -*- """赛道匹配内核金标准用例（离线、确定性）。""" from __future__ import annotations import json import os import tempfile import unittest from _support import IsolatedDataRoot def _teacher(**overrides) -> dict: base = { "teacherName": "张教授", "department": "化材学院", "title": "教授", "researchTags": "钙钛矿,光伏,半导体,界面调控,稳定性", "recentPapers": "钙钛矿电池界面钝化研究", "projectHistory": "主持省自然科学基金", "sciQ1Count": 3, "corePaperCount": 8, } base.update(overrides) return base def _tracks() -> list[dict]: return [ {"year": 2027, "programName": "面上项目", "trackName": "光伏材料", "trackTags": "钙钛矿,光伏,半导体", "priorityKeywords": "界面调控,稳定性", "minTitle": "副高及以上"}, {"year": 2027, "programName": "青年项目", "trackName": "柔性电子", "trackTags": "二维材料,柔性电子,钙钛矿"}, {"year": 2027, "programName": "青年项目", "trackName": "农村经济", "trackTags": "乡村振兴,农村经济"}, {"year": 2027, "programName": "重点项目", "trackName": "量子计算", "trackTags": "量子计算,量子信息", "minTitle": "正高及以上", "minSciQ1": 5}, ] class MatchRuleTestBase(unittest.TestCase): def setUp(self) -> None: self._root = IsolatedDataRoot() self._root.__enter__() from db.connection import init_db init_db() def tearDown(self) -> None: self._root.__exit__(None, None, None) def _match(self, teacher=None, tracks=None): from service.match_service import match_teacher return match_teacher(teacher or _teacher(), tracks if tracks is not None else _tracks(), "2027") class TestHardThresholds(MatchRuleTestBase): def test_h1_title_threshold_excludes(self) -> None: result = self._match(_teacher(title="讲师")) excluded_tracks = {e["track"] for e in result["excluded"]} self.assertIn("重点项目·量子计算", excluded_tracks) self.assertIn("H1", result["excluded"][0]["reason"]) def test_h1_min_title_none_no_filter(self) -> None: result = self._match(_teacher(title="讲师")) # 无门槛赛道不受影响 self.assertGreater(result["eligible_count"], 0) def test_h2_sci_q1_threshold_excludes(self) -> None: result = self._match(_teacher(title="研究员", sciQ1Count=2)) excluded_tracks = {e["track"] for e in result["excluded"]} self.assertIn("重点项目·量子计算", excluded_tracks) self.assertIn("H2", result["excluded"][0]["reason"]) def test_e1_incomplete_profile_raises(self) -> None: with self.assertRaises(ValueError): self._match(_teacher(researchTags="", recentPapers="")) class TestScoring(MatchRuleTestBase): def test_s3_full_coverage_with_priority_hits(self) -> None: result = self._match() top = result["top3"][0] # 光伏材料：coverage 3/3=1.0 → 0.7 + priority 2 hits 0.3 = 1.0 self.assertEqual(top["track"], "面上项目·光伏材料") self.assertEqual(top["match_pct"], 100) self.assertEqual(top["tier"], "强烈推荐") def test_coverage_only_below_try_band(self) -> None: result = self._match(_teacher(researchTags="钙钛矿,光伏", recentPapers="")) entry = next(e for e in result["top3"] if e["track"] == "面上项目·光伏材料") # 仅覆盖率 2/3 → 0.7×0.6667=46.7% → 47%，低于 50% 可尝试档 → 不建议 self.assertEqual(entry["match_pct"], 47) self.assertEqual(entry["tier"], "不建议") def test_t1_try_band_with_supplement_hint(self) -> None: result = self._match(_teacher(researchTags="钙钛矿,光伏,界面调控")) entry = next(e for e in result["top3"] if e["track"] == "面上项目·光伏材料") # 覆盖率 2/3 → 0.7×0.6667=46.7% + 优先方向 1 命中 0.15 = 61.7% → 62%，落在可尝试档 self.assertEqual(entry["match_pct"], 62) self.assertEqual(entry["tier"], "可尝试") self.assertTrue(any("建议补充" in r for r in entry["reasons"])) def test_zero_match_reason_is_factual(self) -> None: result = self._match(_teacher(researchTags="量子计算", recentPapers="", projectHistory="")) entry = next(e for e in result["top3"] if e["track"] == "面上项目·光伏材料") reasons_text = "".join(entry["reasons"]) self.assertEqual(entry["match_pct"], 0) self.assertIn("暂无重合", reasons_text) self.assertNotIn("部分相关性", reasons_text) def test_t1_strong_boundary(self) -> None: # 恰好 80% 属强烈推荐 result = self._match(_teacher(researchTags="钙钛矿,光伏,半导体", recentPapers="")) entry = next(e for e in result["top3"] if e["track"] == "面上项目·光伏材料") # coverage 1.0 → 0.7 → 70% 可尝试 self.assertEqual(entry["tier"], "可尝试") self.assertEqual(entry["match_pct"], 70) def test_t2_top3_truncation_and_order(self) -> None: result = self._match() self.assertLessEqual(len(result["top3"]), 3) pcts = [e["match_pct"] for e in result["top3"]] self.assertEqual(pcts, sorted(pcts, reverse=True)) def test_reasons_traceable(self) -> None: result = self._match() top = result["top3"][0] self.assertTrue(any("命中需求标签" in r for r in top["reasons"])) self.assertTrue(any("优先资助方向" in r for r in top["reasons"])) class TestKernelAndPersistence(MatchRuleTestBase): def _run(self, teachers=None): from service.match_service import run_match return run_match(teachers or [_teacher()], _tracks(), "2027") def test_run_match_writes_reports_and_brief(self) -> None: result = self._run([_teacher(), _teacher(teacherName="李教授", researchTags="乡村振兴,农村金融", title="教授")]) self.assertEqual(result["count"], 2) self.assertTrue(result["brief_file"].endswith(".html")) for report in result["reports"]: self.assertTrue(report["report_file"].endswith(".html")) with open(report["report_file"], encoding="utf-8") as f: html_text = f.read() self.assertIn("申报策略建议书", html_text) self.assertIn("仅供参考", html_text) def test_empty_tracks_raises(self) -> None: from service.match_service import run_match with self.assertRaises(ValueError): run_match([_teacher()], [], "2027") def test_no_teachers_raises(self) -> None: from service.match_service import run_match with self.assertRaises(ValueError): run_match([], _tracks(), "2027") def test_upsert_teacher_and_track(self) -> None: from service.profile_service import upsert_teacher, upsert_track payload = upsert_teacher({"teacher_name": "张教授", "title": "教授"}) self.assertEqual(payload["action"], "created") payload = upsert_teacher({"teacher_name": "张教授", "sci_q1_count": 5}) self.assertEqual(payload["action"], "updated") with self.assertRaises(ValueError): upsert_teacher({"teacher_name": ""}) payload = upsert_track({"year": 2027, "track_name": "光伏材料", "track_tags": "钙钛矿,光伏,半导体"}) self.assertEqual(payload["action"], "created") with self.assertRaises(ValueError): upsert_track({"year": 2027, "track_name": "光伏材料", "min_title": "院士及以上"}) def test_stats(self) -> None: from service.profile_service import stats, upsert_track for track in _tracks(): upsert_track({ "year": track["year"], "track_name": track["trackName"], "program_name": track.get("programName", ""), "track_tags": track.get("trackTags", ""), }) self._run() payload = stats() self.assertEqual(payload["report_total"], 1) self.assertEqual(payload["tracks_by_year"].get("2027"), 4) class TestBatchSkipIncomplete(MatchRuleTestBase): """批量缺省跳过画像不完整教师（E1），指定单教师保持严格失败。""" def test_batch_skips_incomplete_teacher(self) -> None: from service.match_service import run_match incomplete = _teacher(teacherName="王讲师", researchTags="", recentPapers="") payload = run_match([_teacher(), incomplete], _tracks(), "2027") self.assertEqual(payload["ok"], True) self.assertEqual(payload["count"], 1) self.assertEqual(len(payload["skipped"]), 1) skipped = payload["skipped"][0] self.assertEqual(skipped["teacher_name"], "王讲师") self.assertIn("E1", skipped["reason"]) # 正常教师仍有报告 self.assertEqual(len(payload["reports"]), 1) self.assertEqual(payload["reports"][0]["teacher_name"], "张教授") def test_batch_brief_marks_skipped_teacher(self) -> None: from service.match_service import run_match incomplete = _teacher(teacherName="王讲师", researchTags="", recentPapers="") payload = run_match([_teacher(), incomplete], _tracks(), "2027") with open(payload["brief_file"], encoding="utf-8") as f: html_text = f.read() self.assertIn("画像待补录", html_text) self.assertIn("王讲师", html_text) def test_all_skipped_still_returns_success_payload(self) -> None: from service.match_service import run_match payload = run_match([_teacher(researchTags="", recentPapers="")], _tracks(), "2027") self.assertEqual(payload["ok"], True) self.assertEqual(payload["count"], 0) self.assertEqual(len(payload["skipped"]), 1) def test_strict_mode_raises_for_incomplete(self) -> None: from service.match_service import run_match with self.assertRaises(ValueError): run_match([_teacher(researchTags="", recentPapers="")], _tracks(), "2027", skip_incomplete=False) class TestTrackYearScope(MatchRuleTestBase): """缺省取赛道库最新年度（actions.json 契约）；显式年度不受影响。""" def _seed_two_years(self) -> None: from service.profile_service import upsert_track upsert_track({"year": 2026, "track_name": "旧方向", "program_name": "面上项目", "track_tags": "钙钛矿,光伏"}) upsert_track({"year": 2027, "track_name": "新方向", "program_name": "面上项目", "track_tags": "量子计算"}) def test_load_tracks_defaults_to_latest_year(self) -> None: from service.match_service import load_tracks self._seed_two_years() self.assertEqual({t["year"] for t in load_tracks()}, {2027}) def test_load_tracks_explicit_year(self) -> None: from service.match_service import load_tracks self._seed_two_years() self.assertEqual({t["year"] for t in load_tracks(year=2026)}, {2026}) def test_load_tracks_empty_library(self) -> None: from service.match_service import load_tracks self.assertEqual(load_tracks(), []) def test_kernel_default_matches_latest_year_only(self) -> None: from service.match_service import load_teachers, load_tracks, run_match from service.profile_service import upsert_teacher self._seed_two_years() upsert_teacher({"teacher_name": "张教授", "title": "教授", "research_tags": "量子计算,量子信息"}) payload = run_match(load_teachers(), load_tracks(), "2027") report = payload["reports"][0] all_tracks = [e["track"] for entries in report["tiers"].values() for e in entries] all_tracks += [e["track"] for e in report["top3"]] self.assertIn("面上项目·新方向", all_tracks) self.assertNotIn("面上项目·旧方向", all_tracks) class TestBatchInputCollection(MatchRuleTestBase): """input/ 目录收集：坏文件跳过并记录原因，不中断整批。""" def test_collect_input_teachers_skips_bad_files(self) -> None: from service.match_service import collect_input_teachers with tempfile.TemporaryDirectory() as tmp: with open(os.path.join(tmp, "a.json"), "w", encoding="utf-8") as f: json.dump([_teacher(), _teacher(teacherName="李教授")], f, ensure_ascii=False) with open(os.path.join(tmp, "single.json"), "w", encoding="utf-8") as f: json.dump({"teacherName": "王教授"}, 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) teachers, skipped = collect_input_teachers(tmp) names = [t.get("teacherName") for t in teachers] 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_teachers_empty_dir(self) -> None: from service.match_service import collect_input_teachers with tempfile.TemporaryDirectory() as tmp: teachers, skipped = collect_input_teachers(tmp) self.assertEqual(teachers, []) self.assertEqual(skipped, []) if __name__ == "__main__": unittest.main()