RoyirinRosmontis
1722c00249
- scripts/expand_real_dataset.py：AutoWeakS 全量 2215 条 + 500 条分层评测集 + 简历扩量（seed=2026 可复现） - pipeline 被动语料机制（passive_sources）：真实无日期 JD 参与接入/清洗/抽取验证（2915 条命中率 99.8%），不进图谱/演化/发现 - 新岗位发现第四指标'差异度'与双年窗口增长率口径 - 评测新增 real_jd_500 维度（99.91%）；测试会话隔离 jd_extra_dirs 防止本机配置拖慢套件
158 lines
7.0 KiB
Python
158 lines
7.0 KiB
Python
"""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
|