scripts/eval_corpus_quality.py

"""
Corpus Eval: 评估大库下检索质量。

Usage:
  python scripts/eval_corpus_quality.py

Outputs:
  reports/corpus_eval_report.md
"""

from __future__ import annotations

import json
import sys
import time
from collections import Counter
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))

from src.matcher import rank_job_list
from src.resume_parser import parse_resume


# 6-8 个代表性简历和目标方向
TEST_QUERIES = [
    {
        "name": "CV 学生（目标：计算机视觉）",
        "resume": """张伟
计算机视觉方向硕士生
技能：PyTorch, ResNet, YOLO, OpenCV, Python, 目标检测, 图像分类
项目：
1. 基于 YOLOv8 的实时目标检测系统（mAP 0.68）
2. 细粒度图像分类（ResNet + 注意力机制，Top1 准确率 89%）
""",
        "target_role": "计算机视觉",
        "target_city": "北京",
        "stage": "实习",
    },
    {
        "name": "LLM Agent 学生（目标：大模型应用算法）",
        "resume": """李明
大模型应用算法方向
技能：Python, PyTorch, LangChain, RAG, Agent, LLM, Embedding, Prompt
项目：
1. 基于 RAG 的知识库问答系统（召回率 85%，准确率 78%）
2. 多 Agent 协作任务规划框架（成功率达 72%）
""",
        "target_role": "大模型应用算法",
        "target_city": "深圳",
        "stage": "实习",
    },
    {
        "name": "推荐算法学生（目标：推荐算法）",
        "resume": """王芳
推荐算法方向
技能：Python, PyTorch, 推荐系统, 召回, 排序, NDCG, A/B Test, Embedding
项目：
1. 短视频推荐召回排序优化（线上 CTR + 12%）
2. 多目标推荐优化（CTR + CVR 联合优化）
""",
        "target_role": "推荐算法",
        "target_city": "北京",
        "stage": "校招",
    },
    {
        "name": "后端开发（目标：后端研发）",
        "resume": """赵强
后端研发方向
技能：Go, Python, Docker, Kubernetes, MySQL, Redis, FastAPI, 微服务
项目：
1. 高并发订单系统（QPS 5000+，P99 延迟 < 50ms）
2. 微服务治理平台（服务发现、熔断、限流）
""",
        "target_role": "后端研发",
        "target_city": "上海",
        "stage": "校招",
    },
    {
        "name": "数据分析（目标：数据分析）",
        "resume": """陈静
数据分析方向
技能：Python, SQL, Pandas, A/B Test, 指标体系, 可视化, Hive
项目：
1. 用户增长漏斗分析（识别 3 个关键流失点，转化率 + 18%）
2. A/B 实验平台搭建（支持 50+ 并行实验）
""",
        "target_role": "数据分析",
        "target_city": "杭州",
        "stage": "实习",
    },
    {
        "name": "NLP 转 LLM（目标：大模型应用算法）",
        "resume": """王磊
NLP 转大模型应用算法
技能：Python, PyTorch, BERT, 文本分类, 命名实体识别, LLM, Prompt
项目：
1. 基于 BERT 的文本分类系统（准确率 92%）
2. 命名实体识别模型（F1 值 0.87）
3. 正在学习 RAG 和 Agent，有一个简单的 demo
""",
        "target_role": "大模型应用算法",
        "target_city": "北京",
        "stage": "实习",
    },
    {
        "name": "弱项目背景（目标：推荐算法）",
        "resume": """刘洋
推荐算法方向
技能：Python, 机器学习, 数据分析
项目：
1. 课程作业：MovieLens 推荐系统（协同过滤）
2. 无实习经历，无论文发表
""",
        "target_role": "推荐算法",
        "target_city": "上海",
        "stage": "校招",
    },
]


def load_corpus() -> list[dict]:
    corpus_path = ROOT / "data" / "jobs_corpus.json"
    if not corpus_path.exists():
        print(f"[FAIL] {corpus_path} not found")
        return []
    with corpus_path.open("r", encoding="utf-8") as f:
        data = json.load(f)
    return data if isinstance(data, list) else []


def check_topk(jobs: list[dict], query: dict, k: int = 5) -> dict:
    """检查 TopK 检索质量。"""
    profile = parse_resume(query["resume"])
    profile["_target_role"] = query["target_role"]
    profile["_city"] = query["target_city"]
    profile["_stage"] = query["stage"]

    start = time.time()
    ranked = rank_job_list(
        resume_text=query["resume"],
        profile=profile,
        target_role=query["target_role"],
        target_city=query["target_city"],
        stage=query["stage"],
        top_k=k,
        jobs=jobs,
    )
    elapsed = time.time() - start

    topk = ranked[:k]

    # 1. direction 命中
    direction_hit = sum(
        1 for j in topk if query["target_role"] in (j.get("direction") or "")
    )
    direction_recall = direction_hit / max(k, 1)

    # 2. city / stage 合理性
    city_hit = sum(
        1 for j in topk
        if j.get("city") == query["target_city"] or j.get("city") == "不限"
    )
    stage_hit = sum(
        1 for j in topk
        if j.get("stage") == query["stage"] or j.get("stage") == "不限"
    )

    # 3. 重复标题
    titles = [j.get("title", "") for j in topk]
    duplicate = k - len(set(titles))
    duplicate_ratio = duplicate / max(k, 1)

    # 4. 空字段
    empty_skills = sum(1 for j in topk if not j.get("skills"))
    empty_jd = sum(1 for j in topk if not j.get("jd") and not j.get("description"))

    # 5. source 多样性
    sources = [j.get("source", "unknown") for j in topk]
    source_counter = Counter(sources)
    top_source, top_source_count = source_counter.most_common(1)[0]
    source_diversity = 1.0 - (top_source_count / max(k, 1))

    # 6. 提取打分字段
    topk_detail = []
    for j in topk:
        topk_detail.append({
            "title": j.get("title", "?"),
            "company": j.get("company", "?"),
            "city": j.get("city", "?"),
            "direction": j.get("direction", "?"),
            "source": j.get("source", "?"),
            "match_score": j.get("match_score", "?"),
            "apply_priority": j.get("apply_priority", "?"),
            "pass_score": j.get("pass_score", "?"),
            "risk_score": j.get("risk_score", "?"),
        })

    return {
        "query_name": query["name"],
        "target_role": query["target_role"],
        "elapsed": round(elapsed, 3),
        "topk_detail": topk_detail,
        "direction_hit": direction_hit,
        "direction_recall": direction_recall,
        "city_hit": city_hit,
        "stage_hit": stage_hit,
        "duplicate": duplicate,
        "duplicate_ratio": duplicate_ratio,
        "empty_skills": empty_skills,
        "empty_jd": empty_jd,
        "source_diversity": source_diversity,
        "top_source": top_source,
    }


def generate_report(results: list[dict]) -> str:
    lines = []
    lines.append("# Corpus Eval Report\n")
    lines.append(f"Generated: {time.strftime('%Y-%m-%d %H:%M:%S')}\n")
    lines.append(f"**Corpus**: `data/jobs_corpus.json`\n")
    lines.append(f"**Queries**: {len(results)}\n")

    # 汇总指标
    total_direction_hit = sum(r["direction_hit"] for r in results)
    total_empty = sum(r["empty_skills"] + r["empty_jd"] for r in results)
    avg_diversity = sum(r["source_diversity"] for r in results) / max(len(results), 1)

    lines.append("## 汇总指标\n")
    lines.append(f"- **Top1 方向命中率**: {total_direction_hit}/{len(results)} ({total_direction_hit/max(len(results),1)*100:.1f}%)\n")
    lines.append(f"- **Top5 方向召回率**: 见各 case\n")
    lines.append(f"- **空字段数**: {total_empty}\n")
    lines.append(f"- **平均来源多样性**: {avg_diversity*100:.1f}%\n")

    # 每个 query 的 Top5 表格
    for r in results:
        lines.append(f"## {r['query_name']}\n")
        lines.append(f"- **Target**: {r['target_role']}\n")
        lines.append(f"- **Elapsed**: {r['elapsed']}s\n")
        lines.append(f"- **Direction Hit**: {r['direction_hit']}/5 ({r['direction_recall']*100:.1f}%)\n")
        lines.append(f"- **City Hit**: {r['city_hit']}/5\n")
        lines.append(f"- **Stage Hit**: {r['stage_hit']}/5\n")
        lines.append(f"- **Duplicate Titles**: {r['duplicate']}/5 ({r['duplicate_ratio']*100:.1f}%)\n")
        lines.append(f"- **Empty Skills**: {r['empty_skills']}\n")
        lines.append(f"- **Empty JD**: {r['empty_jd']}\n")
        lines.append(f"- **Source Diversity**: {r['source_diversity']*100:.1f}% (top: {r['top_source']})\n")

        lines.append("\n**Top 5 Results**:\n\n")
        lines.append("| Rank | Title | Company | Direction | City | Source | Match | Priority | Pass | Risk |\n")
        lines.append("|---|---|---|---|---|---|---|---|---|---|\n")
        for i, job in enumerate(r["topk_detail"]):
            lines.append(
                f"| {i+1} | {job['title']} | {job['company']} "
                f"| {job['direction']} | {job['city']} | {job['source']} "
                f"| {job['match_score']} | {job['apply_priority']} "
                f"| {job['pass_score']} | {job['risk_score']} |\n"
            )
        lines.append("\n")

    # 建议
    lines.append("## 建议\n")
    if total_empty > 0:
        lines.append("- [WARN] 发现空字段，检查数据源。\n")
    if avg_diversity < 0.4:
        lines.append("- [WARN] 来源多样性低，增加真实公开数据。\n")
    lines.append("\n---\n")
    lines.append("*Report generated by `scripts/eval_corpus_quality.py`*\n")

    return "".join(lines)


def main() -> None:
    print("[INFO] Loading job corpus...")
    jobs = load_corpus()
    if not jobs:
        return

    print(f"[INFO] Loaded {len(jobs)} jobs. Running corpus eval...")

    results = []
    for query in TEST_QUERIES:
        print(f"[INFO] Query: {query['name']}")
        result = check_topk(jobs, query, k=5)
        results.append(result)

    print("[INFO] Generating report...")
    report = generate_report(results)

    output_path = ROOT / "reports" / "corpus_eval_report.md"
    output_path.parent.mkdir(parents=True, exist_ok=True)
    output_path.write_text(report, encoding="utf-8")

    print(f"[OK] Report saved to {output_path}")
    print(f"[SUMMARY] Queries: {len(results)}, Empty Fields: {sum(r['empty_skills'] + r['empty_jd'] for r in results)}")


if __name__ == "__main__":
    main()