- 全链路:多源JD接入/清洗交叉验证/结构化抽取(星火+离线双通道)/时序知识图谱/新岗位发现/演化分析/人岗匹配/学习路径 - WebUI 五视图:驾驶舱/图谱工作台/岗位演化/匹配(单人+批量)/学习路径 - 设置中心(config.json)/Token鉴权/SQLite状态库/语义向量通道(BGE-M3本地或联网API) - 71 个 pytest 用例;数据目录仅保留结构(.gitkeep),数据由 scripts/generate_dataset.py 复现
103 lines
4.4 KiB
Python
103 lines
4.4 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
|
||
|
||
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 len(list(RESUME_DIR.glob("*.txt"))) == 50
|
||
assert (RESUME_DIR / "gold.json").exists()
|
||
assert len(list(DEMO_DIR.glob("*.txt"))) == 12
|
||
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) == 50
|
||
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"])
|
||
|
||
|
||
def test_batch_demo_masked() -> None:
|
||
"""批量演示素材必须已脱敏:无 11 位手机号、邮箱仅呈 x***@domain 形态、姓名带 *。"""
|
||
files = list(DEMO_DIR.glob("*.txt"))
|
||
assert len(files) == 12
|
||
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
|
||
|
||
|
||
@pytest.mark.parametrize("dim", ["real_jd", "real_resume"])
|
||
def test_eval_real_dimensions(dim: str) -> None:
|
||
"""真实维度评测可运行且指标在合法区间(全量小数据集,直接实跑)。"""
|
||
evaluate = _load_evaluate()
|
||
result = evaluate.eval_real_jd() if dim == "real_jd" else 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 # 任务目标
|
||
else:
|
||
assert result["samples"] == 50
|
||
assert result["name_accuracy"] == 1.0 # 任务目标:脱敏后姓氏+*
|
||
assert result["phone_accuracy"] == 1.0
|