scripts/test_evidence.py

"""
test_evidence.py — 证据链模块测试脚本

测试目标：
1. builtin job 能生成 evidence
2. user pasted JD 能生成 evidence
3. evidence 字段存在且非空
4. 每条 evidence 都有 type/claim/evidence/source/confidence
"""

from __future__ import annotations

import json
import sys
from pathlib import Path

# 加入项目根目录到 sys.path
ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(ROOT))

from src.resume_parser import parse_resume
from src.jd_parser import parse_jd
from src.matcher import load_jobs, score_job, rank_job_list
from src.evidence import (
    build_jd_evidence,
    build_resume_evidence,
    build_gap_evidence,
    build_action_evidence,
    attach_evidence,
    attach_evidence_to_jobs,
)


# ---------------------------------------------------------------------------
# 测试数据
# ---------------------------------------------------------------------------

SAMPLE_RESUME = """张同学 | 计算机科学与技术 | 2026 届硕士
求职方向：大模型应用算法 / 推荐算法实习生，期望城市深圳或北京。
技能：Python、PyTorch、Transformer、RAG、Agent、Embedding、Faiss、LangChain、推荐系统、召回排序、NDCG、A/B Test、SQL。
项目经历：
1. GenAdRec 生成式广告推荐项目：基于 Transformer 建模用户行为序列，将广告候选集转化为生成式 likelihood rerank 问题；构建 Semantic ID 表示广告 item，结合多兴趣召回提升 NDCG@10。
2. LLM 求职助手 Demo：使用 DeepSeek API 和 bge embedding 实现 JD 检索、简历关键词诊断、Prompt 模板优化，支持输出岗位匹配解释。
3. MIND 多兴趣推荐复现：复现 capsule routing 用户多兴趣建模，在公开数据集上对比召回 HitRate 与 NDCG。
实习经历：
曾参与推荐系统离线评估脚本开发，负责样本构造、特征清洗和模型结果分析。
补充：希望找能结合 LLM、Agent、RAG 和推荐排序的算法岗位。"""

SAMPLE_JD = """职位名称：大模型应用算法实习生
公司：腾讯
城市：深圳
方向：大模型应用算法
阶段：实习
技能要求：LLM、RAG、Agent、Prompt Engineering、LangChain、Embedding、Faiss、Python
项目信号：RAG、Agent、Semantic Retrieval、Multi-turn Dialogue
JD 描述：
负责大模型应用算法研发，包括 RAG 系统搭建、Agent 工作流设计、Prompt 模板优化。
要求熟悉 LLM 应用开发，了解 RAG 技术栈（Embedding + Vector DB），有 Agent 框架使用经验者优先。
"""


# ---------------------------------------------------------------------------
# 工具函数
# ---------------------------------------------------------------------------

def _check_evidence_fields(ev: dict, context: str) -> list[str]:
    """检查单条 evidence 是否包含所有必填字段。返回错误列表。"""
    errors = []
    required_fields = ["type", "claim", "evidence", "source", "confidence"]
    for field in required_fields:
        if field not in ev:
            errors.append(f"{context}: 缺少字段 {field}")
    # confidence 必须是 high/medium/low 之一
    if "confidence" in ev and ev["confidence"] not in ("high", "medium", "low"):
        errors.append(f"{context}: confidence={ev['confidence']} 不在允许范围内")
    return errors


def _print_evidence_list(label: str, ev_list: list[dict]) -> None:
    """打印 evidence 列表，方便人工检查。"""
    print(f"\n  【{label}】（共 {len(ev_list)} 条）")
    for i, ev in enumerate(ev_list):
        print(f"    [{i+1}] type={ev.get('type','?')} | confidence={ev.get('confidence','?')}")
        print(f"        claim: {ev.get('claim','')}")
        print(f"        evidence: {ev.get('evidence','')}")
        print(f"        source: {ev.get('source','')}")


# ---------------------------------------------------------------------------
# 测试用例
# ---------------------------------------------------------------------------

def test_01_builtin_job_evidence() -> bool:
    """测试 1：builtin job 能生成 evidence。"""
    print("\n=== test_01_builtin_job_evidence ===")
    profile = parse_resume(SAMPLE_RESUME)
    jobs = load_jobs(ROOT / "data" / "jobs.json")
    job = jobs[0]  # 取第一个 builtin job

    # 对 job 做完整打分（会附加 evidence）
    scored = score_job(job, profile, SAMPLE_RESUME, "大模型应用算法", "深圳", "实习")

    # 检查 evidence 字段存在
    required_evidence_fields = ["jd_evidence", "resume_evidence", "gap_evidence", "action_evidence"]
    errors = []
    for field in required_evidence_fields:
        if field not in scored:
            errors.append(f"scored 缺少字段 {field}")
        elif not scored[field]:
            errors.append(f"scored[{field}] 为空")

    if errors:
        print("  [FAIL] " + "; ".join(errors))
        return False

    _print_evidence_list("jd_evidence", scored["jd_evidence"])
    _print_evidence_list("resume_evidence", scored["resume_evidence"])
    _print_evidence_list("gap_evidence", scored["gap_evidence"])
    print(f"  action_evidence: {scored['action_evidence']}")

    # 检查每条 evidence 的字段
    all_errors = []
    for ev in scored["jd_evidence"]:
        all_errors.extend(_check_evidence_fields(ev, "jd_evidence"))
    for ev in scored["resume_evidence"]:
        all_errors.extend(_check_evidence_fields(ev, "resume_evidence"))
    for ev in scored["gap_evidence"]:
        all_errors.extend(_check_evidence_fields(ev, "gap_evidence"))

    if all_errors:
        print("  [FAIL] " + "; ".join(all_errors))
        return False

    print("  [PASS] builtin job 证据链生成正常")
    return True


def test_02_user_pasted_jd_evidence() -> bool:
    """测试 2：user pasted JD 能生成 evidence。"""
    print("\n=== test_02_user_pasted_jd_evidence ===")
    profile = parse_resume(SAMPLE_RESUME)

    # 解析用户粘贴的 JD
    from src.jd_parser import parse_jd
    jd_result = parse_jd(SAMPLE_JD)
    if not jd_result:
        print("  [SKIP] parse_jd 返回空，可能是规则解析失败（非错误）")
        return True

    job = jd_result
    job.setdefault("source", "user_pasted")
    job.setdefault("stage", "实习")
    job.setdefault("city", "深圳")
    job.setdefault("direction", "大模型应用算法")
    job.setdefault("company", "测试公司")
    job.setdefault("title", "大模型应用算法实习生")
    job.setdefault("skills", job.get("skills", []))
    job.setdefault("project_signals", job.get("project_signals", []))

    # 对 job 做完整打分
    scored = score_job(job, profile, SAMPLE_RESUME, "大模型应用算法", "深圳", "实习")

    required_evidence_fields = ["jd_evidence", "resume_evidence", "gap_evidence", "action_evidence"]
    errors = []
    for field in required_evidence_fields:
        if field not in scored:
            errors.append(f"scored 缺少字段 {field}")

    if errors:
        print("  [FAIL] " + "; ".join(errors))
        return False

    _print_evidence_list("jd_evidence (user JD)", scored["jd_evidence"])
    _print_evidence_list("resume_evidence (user JD)", scored["resume_evidence"])
    _print_evidence_list("gap_evidence (user JD)", scored["gap_evidence"])

    print("  [PASS] user pasted JD 证据链生成正常")
    return True


def test_03_evidence_fields_complete() -> bool:
    """测试 3：每条 evidence 都有 type/claim/evidence/source/confidence。"""
    print("\n=== test_03_evidence_fields_complete ===")
    profile = parse_resume(SAMPLE_RESUME)
    jobs = load_jobs(ROOT / "data" / "jobs.json")
    job = jobs[0]

    scored = score_job(job, profile, SAMPLE_RESUME, "大模型应用算法", "深圳", "实习")

    all_errors = []
    for ev in scored.get("jd_evidence", []):
        all_errors.extend(_check_evidence_fields(ev, "jd_evidence"))
    for ev in scored.get("resume_evidence", []):
        all_errors.extend(_check_evidence_fields(ev, "resume_evidence"))
    for ev in scored.get("gap_evidence", []):
        all_errors.extend(_check_evidence_fields(ev, "gap_evidence"))

    if all_errors:
        print("  [FAIL] " + "; ".join(all_errors[:5]))
        return False

    print("  [PASS] 所有 evidence 字段完整")
    return True


def test_04_action_evidence_format() -> bool:
    """测试 4：action_evidence 是 list[str]，且内容非空。"""
    print("\n=== test_04_action_evidence_format ===")
    profile = parse_resume(SAMPLE_RESUME)
    jobs = load_jobs(ROOT / "data" / "jobs.json")
    job = jobs[0]

    scored = score_job(job, profile, SAMPLE_RESUME, "大模型应用算法", "深圳", "实习")

    action_ev = scored.get("action_evidence", [])
    if not isinstance(action_ev, list):
        print(f"  [FAIL] action_evidence 类型错误：{type(action_ev)}")
        return False
    if not action_ev:
        print("  [FAIL] action_evidence 为空")
        return False
    for i, item in enumerate(action_ev):
        if not isinstance(item, str):
            print(f"  [FAIL] action_evidence[{i}] 不是 str：{type(item)}")
            return False
        if not item.strip():
            print(f"  [FAIL] action_evidence[{i}] 为空字符串")
            return False

    print(f"  [PASS] action_evidence 格式正确（{len(action_ev)} 条）")
    return True


def test_05_attach_evidence_to_jobs() -> bool:
    """测试 5：attach_evidence_to_jobs 能批量处理 job list。"""
    print("\n=== test_05_attach_evidence_to_jobs ===")
    profile = parse_resume(SAMPLE_RESUME)
    jobs = load_jobs(ROOT / "data" / "jobs.json")[:3]  # 只取前 3 个，加速

    scored_list = attach_evidence_to_jobs(jobs, SAMPLE_RESUME, profile)

    errors = []
    for i, job in enumerate(scored_list):
        for field in ["jd_evidence", "resume_evidence", "gap_evidence", "action_evidence"]:
            if field not in job:
                errors.append(f"job[{i}] 缺少 {field}")

    if errors:
        print("  [FAIL] " + "; ".join(errors))
        return False

    print(f"  [PASS] attach_evidence_to_jobs 批量处理 {len(scored_list)} 个岗位正常")
    return True


def test_06_evidence_grounded() -> bool:
    """测试 6：evidence 是 grounded 的（不能写空泛建议）。"""
    print("\n=== test_06_evidence_grounded ===")
    profile = parse_resume(SAMPLE_RESUME)
    jobs = load_jobs(ROOT / "data" / "jobs.json")
    job = jobs[0]

    scored = score_job(job, profile, SAMPLE_RESUME, "大模型应用算法", "深圳", "实习")

    # 检查 evidence 中的 claim/evidence 是否非空且有实质内容
    all_errors = []
    for ev in scored.get("jd_evidence", []):
        if not ev.get("claim", "").strip():
            all_errors.append("jd_evidence 中有空的 claim")
        if not ev.get("evidence", "").strip():
            all_errors.append("jd_evidence 中有空的 evidence")

    if all_errors:
        print("  [FAIL] " + "; ".join(all_errors))
        return False

    print("  [PASS] evidence 内容 grounded（非空泛建议）")
    return True


# ---------------------------------------------------------------------------
# 主函数
# ---------------------------------------------------------------------------

def main() -> None:
    print("=" * 60)
    print("证据链模块测试开始")
    print("=" * 60)

    results = []
    results.append(("builtin job 生成 evidence", test_01_builtin_job_evidence()))
    results.append(("user pasted JD 生成 evidence", test_02_user_pasted_jd_evidence()))
    results.append(("evidence 字段完整", test_03_evidence_fields_complete()))
    results.append(("action_evidence 格式正确", test_04_action_evidence_format()))
    results.append(("attach_evidence_to_jobs 批量处理", test_05_attach_evidence_to_jobs()))
    results.append(("evidence grounded", test_06_evidence_grounded()))

    print("\n" + "=" * 60)
    print("测试结果汇总")
    print("=" * 60)
    pass_count = 0
    for name, result in results:
        status = "[PASS]" if result else "[FAIL]"
        print(f"  {status}  {name}")
        if result:
            pass_count += 1
    print(f"\n总计：{pass_count}/{len(results)} 通过")
    if pass_count == len(results):
        print("[OK] 所有测试通过！")
    else:
        print("[WARN] 有测试失败，请检查 above。")


if __name__ == "__main__":
    main()