"""data/real 真实数据集：文件齐全、gold Schema 合法、批量演示素材已脱敏、真实评测维度可运行。""" from __future__ import annotations import importlib.util import json import re import sys from pathlib import Path import pytest from pydantic import ValidationError from app.ingest.loader import load_jd_file from app.models import JDRecord ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) REAL = ROOT / "data" / "real" JD_DIR = REAL / "autoweaks_jd" RESUME_DIR = REAL / "aistudio_resumes" DEMO_DIR = REAL / "batch_demo_resumes" ESCO_DIR = REAL / "esco_reference" GOLD_JD_FIELDS = ("job_title", "responsibilities", "required_skills", "preferred_skills", "scenarios") def _load_evaluate(): """按路径加载 scripts/evaluate.py（scripts 不是包）。""" spec = importlib.util.spec_from_file_location("evaluate", ROOT / "scripts" / "evaluate.py") mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(mod) return mod def test_real_tree_complete() -> None: assert (REAL / "README.md").exists() assert (JD_DIR / "jd_sample.jsonl").exists() assert (JD_DIR / "gold_answers.json").exists() assert (JD_DIR / "jd_full.jsonl").exists() assert (JD_DIR / "jd_eval_500.jsonl").exists() assert (JD_DIR / "gold_answers_500.json").exists() assert len(list(RESUME_DIR.glob("*.txt"))) == 200 assert (RESUME_DIR / "gold.json").exists() assert len(list(DEMO_DIR.glob("*.txt"))) == 32 for name in ("labels_isco_codes.txt", "labels_isco_labels.txt", "labels_esco_code.txt", "labels_esco_occupation.txt", "README.md"): assert (ESCO_DIR / name).exists(), name def test_jd_gold_schema() -> None: records = [json.loads(ln) for ln in (JD_DIR / "jd_sample.jsonl").read_text(encoding="utf-8").splitlines() if ln.strip()] assert len(records) == 200 gold = json.loads((JD_DIR / "gold_answers.json").read_text(encoding="utf-8")) assert {r["jid"] for r in records} == set(gold) for r in records: for key in ("jid", "job_title", "company", "publish_date", "source", "url", "text"): assert key in r, (r["jid"], key) assert r["source"] == "autoweaks" assert r["company"] in ("腾讯", "京东") assert re.fullmatch(r"2019-\d{2}-\d{2}", r["publish_date"]), r["jid"] for jid, g in gold.items(): assert tuple(g) == GOLD_JD_FIELDS or set(g) == set(GOLD_JD_FIELDS), jid assert isinstance(g["job_title"], str) and g["job_title"] for f in ("responsibilities", "required_skills", "preferred_skills", "scenarios"): assert isinstance(g[f], list) and all(isinstance(s, str) for s in g[f]), (jid, f) def test_resume_gold_schema() -> None: gold = json.loads((RESUME_DIR / "gold.json").read_text(encoding="utf-8")) assert len(gold) == 200 for fname, g in gold.items(): assert (RESUME_DIR / fname).exists(), fname assert re.fullmatch(r"[一-龥]{2,4}", g["name"]), fname assert re.fullmatch(r"1[3-9]\d{9}", g["phone"]), fname assert isinstance(g["skills"], list) and all(isinstance(s, str) for s in g["skills"]) assert re.fullmatch(r"[0-9a-f]{12}", g["source_id"]), fname def test_batch_demo_masked() -> None: """批量演示素材必须已脱敏：无 11 位手机号、邮箱仅呈 x***@domain 形态、姓名带 *。""" files = list(DEMO_DIR.glob("*.txt")) assert len(files) == 32 phone_re = re.compile(r"1[3-9]\d{9}") email_re = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}") name_re = re.compile(r"姓名[:：]\s*([一-龥]{2,4})") # 未脱敏姓名（脱敏后含 *） for f in files: text = f.read_text(encoding="utf-8") assert not phone_re.search(text), f.name assert not email_re.search(text), f.name # 脱敏形态 x***@d.com 不匹配该正则 assert not name_re.search(text), f.name assert "****" in text, f.name # 手机号已打码 assert re.search(r"姓名[:：]\s*[一-龥]\*", text), f.name def test_jd_full_schema_and_contains_sample() -> None: """jd_full.jsonl：每行过 JDRecord 校验、全文去重、包含原 200 条样本（jid/原文一致）。""" records, errors = load_jd_file(JD_DIR / "jd_full.jsonl") assert errors == [] assert len(records) >= 200 sample = [json.loads(ln) for ln in (JD_DIR / "jd_sample.jsonl").read_text(encoding="utf-8").splitlines() if ln.strip()] by_jid = {r.jid: r for r in records} for s in sample: assert s["jid"] in by_jid, s["jid"] assert by_jid[s["jid"]].text == s["text"], s["jid"] texts = {r.text for r in records} assert len(texts) == len(records) # 全文哈希去重 for r in records: assert r.source == "autoweaks" assert re.fullmatch(r"2019-\d{2}-\d{2}", r.publish_date.isoformat()), r.jid assert len(r.text) >= 60 def test_jd_eval_500_schema_and_superset() -> None: """jd_eval_500.jsonl：500 条、含原 200 条样本、gold Schema 与正式 gold 一致。""" records, errors = load_jd_file(JD_DIR / "jd_eval_500.jsonl") assert errors == [] and len(records) == 500 full = load_jd_file(JD_DIR / "jd_full.jsonl")[0] full_ids = {r.jid for r in full} sample = [json.loads(ln) for ln in (JD_DIR / "jd_sample.jsonl").read_text(encoding="utf-8").splitlines() if ln.strip()] sample_ids = {s["jid"] for s in sample} eval_ids = {r.jid for r in records} assert sample_ids <= eval_ids <= full_ids gold = json.loads((JD_DIR / "gold_answers_500.json").read_text(encoding="utf-8")) assert set(gold) == eval_ids for jid, g in gold.items(): assert set(g) == set(GOLD_JD_FIELDS), jid assert isinstance(g["job_title"], str) and g["job_title"] for f in ("responsibilities", "required_skills", "preferred_skills", "scenarios"): assert isinstance(g[f], list) and all(isinstance(s, str) for s in g[f]), (jid, f) @pytest.mark.parametrize("dim", ["real_jd", "real_jd_500", "real_resume"]) def test_eval_real_dimensions(dim: str) -> None: """真实维度评测可运行且指标在合法区间（全量小数据集，直接实跑）。""" evaluate = _load_evaluate() if dim == "real_jd": result = evaluate.eval_real_jd() elif dim == "real_jd_500": result = evaluate.eval_real_jd_500() else: result = evaluate.eval_real_resume() assert 0.0 <= result["accuracy"] <= 1.0 if dim == "real_jd": assert result["samples"] == 200 and result["bad_lines"] == 0 assert result["fields"]["required_skills"]["f1"] >= 0.80 # 任务目标 elif dim == "real_jd_500": assert result["samples"] == 500 and result["bad_lines"] == 0 assert result["fields"]["required_skills"]["f1"] >= 0.80 # 任务目标 else: assert result["samples"] == 200 assert result["name_accuracy"] == 1.0 # 任务目标：脱敏后姓氏+* assert result["phone_accuracy"] == 1.0