RoyirinRosmontis
a10442d78c
- 全链路：多源JD接入/清洗交叉验证/结构化抽取(星火+离线双通道)/时序知识图谱/新岗位发现/演化分析/人岗匹配/学习路径 - WebUI 五视图：驾驶舱/图谱工作台/岗位演化/匹配(单人+批量)/学习路径 - 设置中心(config.json)/Token鉴权/SQLite状态库/语义向量通道(BGE-M3本地或联网API) - 71 个 pytest 用例；数据目录仅保留结构(.gitkeep)，数据由 scripts/generate_dataset.py 复现
77 lines
2.8 KiB
Python
77 lines
2.8 KiB
Python
"""清洗层测试：植入的抄袭簇必须被检出、噪声过滤、时间衰减公式。"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
from datetime import date
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.cleaning.decay import apply_decay, decay_weight
|
|
from app.cleaning.dedup import MinHash, find_near_duplicates, shingles
|
|
from app.cleaning.noise import filter_noise, is_noise
|
|
from app.ingest.loader import load_jd_dir
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def cleaned():
|
|
records, errors = load_jd_dir(ROOT / "data" / "jd_raw")
|
|
filter_noise(records)
|
|
clusters = find_near_duplicates(records)
|
|
meta = json.loads((ROOT / "data" / "jd_raw" / "_meta.json").read_text(encoding="utf-8"))
|
|
return records, errors, clusters, meta
|
|
|
|
|
|
def test_minhash_identical_text_similarity_one():
|
|
mh = MinHash()
|
|
s1 = mh.signature(shingles("熟练掌握Java与Spring Cloud，负责微服务开发"))
|
|
s2 = mh.signature(shingles("熟练掌握Java与Spring Cloud，负责微服务开发"))
|
|
assert mh.estimate(s1, s2) == 1.0
|
|
|
|
|
|
def test_minhash_distinguishes_different_text():
|
|
mh = MinHash()
|
|
s1 = mh.signature(shingles("熟练掌握Java与Spring Cloud，负责微服务开发"))
|
|
s2 = mh.signature(shingles("精通Verilog与FPGA，负责芯片逻辑设计与验证"))
|
|
assert mh.estimate(s1, s2) < 0.3
|
|
|
|
|
|
def test_planted_dup_clusters_detected(cleaned):
|
|
"""植入的抄袭簇必须全部被检出：副本被降权且指向最早来源。"""
|
|
records, _, clusters, meta = cleaned
|
|
by_jid = {r.jid: r for r in records}
|
|
assert len(clusters) >= len(meta["dup_groups"]) * 0.9
|
|
for keeper_jid, copy_jid in meta["dup_groups"]:
|
|
copy = by_jid[copy_jid]
|
|
assert copy.is_duplicate, f"抄袭副本未被检出: {copy_jid}"
|
|
assert copy.dup_of == keeper_jid
|
|
assert copy.weight < by_jid[keeper_jid].weight # 副本降权、最早来源保留
|
|
|
|
|
|
def test_planted_noise_detected(cleaned):
|
|
"""植入噪声：整条广告被标记，混排尾注被剔除。"""
|
|
records, _, _, meta = cleaned
|
|
by_jid = {r.jid: r for r in records}
|
|
for jid in meta["noise_jids"]:
|
|
rec = by_jid[jid]
|
|
assert rec.is_noise or not is_noise(rec.text), f"噪声未处理: {jid}"
|
|
|
|
|
|
def test_decay_formula():
|
|
w = decay_weight(date(2020, 1, 1), date(2026, 1, 1), lam=0.35)
|
|
assert abs(w - math.exp(-0.35 * (2191 / 365.25))) < 1e-3
|
|
assert decay_weight(date(2026, 1, 1), date(2026, 1, 1)) == 1.0
|
|
older = decay_weight(date(2019, 6, 1), date(2026, 6, 1))
|
|
newer = decay_weight(date(2025, 6, 1), date(2026, 6, 1))
|
|
assert older < newer # 越旧权重越低（时滞衰减）
|
|
|
|
|
|
def test_apply_decay_multiplies_existing(cleaned):
|
|
records, _, _, _ = cleaned
|
|
apply_decay(records, ref_date=date(2026, 12, 28), lam=0.35)
|
|
for r in records:
|
|
assert 0 < r.weight <= 1.0
|