306 lines
14 KiB
Python
306 lines
14 KiB
Python
# -*- 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()
|