scripts/test_data_ingestion.py

"""
test_data_ingestion.py — 测试公开岗位数据接入管线

使用内置 30 条模拟公开岗位样本，测试 normalize、filter、deduplicate、merge 全链路。
可离线运行，不依赖网络。
输出 data/public_jobs_sample.json 和 data/jobs_merged.json。
"""

import json
import os
import sys

ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, ROOT)

from src.public_job_ingestion import (
    normalize_public_job,
    infer_direction,
    infer_stage,
    _find_skills,
    _quality_score,
    deduplicate_jobs,
    merge_jobs,
    _safe_str,
)

# ---------------------------------------------------------------------------
# 内置 30 条模拟公开岗位样本（覆盖 AI/ML/LLM/Agent/推荐/后端/CV/NLP/数据分析）
# ---------------------------------------------------------------------------

FIXTURE_JOBS = [
    # ----- LLM / Agent 方向 (8 条) -----
    {"title": "LLM Application Engineer Intern", "company": "Anthropic", "location": "San Francisco",
     "description": "Build LLM-powered applications using Claude API. Implement RAG pipelines and agent workflows. Requirements: Python, PyTorch, LangChain, experience with prompt engineering and vector databases.",
     "employment_type": "Internship", "source": "greenhouse"},
    {"title": "大模型算法实习生", "company": "字节跳动", "location": "北京",
     "description": "负责大模型应用算法研发，包括 RAG 检索增强、Agent 工具调用、Prompt 优化。要求熟悉 Python、PyTorch、Transformer，有 LLM 项目经验优先。",
     "employment_type": "实习", "source": "fixture"},
    {"title": "AI Research Scientist - LLM", "company": "Meta", "location": "Menlo Park",
     "description": "Research and develop large language models. Fine-tune models using LoRA/QLoRA. Build evaluation benchmarks. Requirements: PhD in CS/NLP, publications at top venues.",
     "employment_type": "Full-time", "source": "lever"},
    {"title": "Generative AI Engineer", "company": "Stability AI", "location": "London",
     "description": "Develop generative AI applications with diffusion models and LLMs. Build RAG systems for document understanding. Requirements: Python, PyTorch, HuggingFace, LangChain.",
     "employment_type": "Full-time", "source": "ashby"},
    {"title": "Agent 应用开发工程师", "company": "Minimax", "location": "上海",
     "description": "开发基于 LLM 的 Agent 应用，实现多轮对话和工具调用。要求熟悉 LangChain Agent 框架，有 Function Calling 实践，了解 ReAct 工作流。",
     "employment_type": "社招", "source": "fixture"},
    {"title": "NLP 算法实习生（大模型方向）", "company": "百度", "location": "北京",
     "description": "参与文心大模型应用开发，进行 NER/文本分类/摘要等任务微调。要求：Python, PyTorch, BERT, HuggingFace Transformers，有 NLP 项目经验。",
     "employment_type": "实习", "source": "fixture"},
    {"title": "LLM Infrastructure Engineer", "company": "OpenAI", "location": "San Francisco",
     "description": "Build scalable infrastructure for training and serving large language models. Optimize inference latency. Requirements: Python, CUDA, Triton, Ray, Kubernetes.",
     "employment_type": "Full-time", "source": "greenhouse"},
    {"title": "AI Application Developer", "company": "Notion", "location": "San Francisco",
     "description": "Integrate AI features into Notion's product using LLMs. Build RAG-powered Q&A, summarization, and writing assistants. Requirements: TypeScript, Python, API design.",
     "employment_type": "Full-time", "source": "lever"},

    # ----- 推荐算法方向 (5 条) -----
    {"title": "推荐算法实习生", "company": "快手", "location": "北京",
     "description": "参与短视频推荐排序模型优化，包括召回、粗排、精排。要求熟悉推荐系统、排序模型、NDCG指标体系。Python、PyTorch、A/B Test 经验优先。",
     "employment_type": "实习", "source": "fixture"},
    {"title": "Machine Learning Engineer - Recommendations", "company": "Spotify", "location": "Stockholm",
     "description": "Build recommendation systems for music discovery. Work on collaborative filtering, content-based recommendations, and multi-objective ranking. Requirements: Python, TensorFlow, Scala.",
     "employment_type": "Full-time", "source": "greenhouse"},
    {"title": "搜索推荐算法工程师", "company": "阿里巴巴", "location": "杭州",
     "description": "负责电商搜索和推荐算法优化，提升 CTR/CVR。要求掌握排序模型(LTR/DNN)、召回策略、多目标优化。Python、TensorFlow 熟练。",
     "employment_type": "社招", "source": "fixture"},
    {"title": "推荐系统工程师", "company": "美团", "location": "北京",
     "description": "优化外卖推荐系统，从召回-粗排-精排全链路优化。要求熟悉 Wide&Deep、DeepFM、多任务学习。Python、Spark、Hive 熟练。",
     "employment_type": "社招", "source": "fixture"},
    {"title": "Junior Data Scientist - RecSys", "company": "Delivery Hero", "location": "Berlin",
     "description": "Develop recommendation models for food delivery. Apply collaborative filtering and deep learning to improve order recommendations. Requirements: Python, SQL, basic ML.",
     "employment_type": "Entry Level", "source": "ashby"},

    # ----- 后端/AI平台方向 (5 条) -----
    {"title": "后端研发实习生（Go 方向）", "company": "腾讯", "location": "深圳",
     "description": "参与 AI 平台后端服务开发，使用 Go 语言。要求掌握 Go、MySQL、Redis、消息队列，了解微服务架构和 gRPC。有分布式系统经验优先。",
     "employment_type": "实习", "source": "fixture"},
    {"title": "Backend Engineer - AI Platform", "company": "Stripe", "location": "Seattle",
     "description": "Build backend services for Stripe's ML platform. Design APIs for model serving and feature store. Requirements: Go/Java, gRPC, Kubernetes, experience with ML infrastructure.",
     "employment_type": "Full-time", "source": "lever"},
    {"title": "云原生后端开发工程师", "company": "华为", "location": "深圳",
     "description": "开发云原生 AI 平台后端服务，基于 Kubernetes/Docker 构建。要求 Go/Java/Python，熟悉微服务、容器化部署，了解 CI/CD 流程。",
     "employment_type": "社招", "source": "fixture"},
    {"title": "ML Platform Engineer", "company": "Databricks", "location": "San Francisco",
     "description": "Build the platform that powers ML training and serving at scale. Design model registry, feature store, and experiment tracking. Requirements: Python, Scala, K8s, Spark.",
     "employment_type": "Full-time", "source": "greenhouse"},
    {"title": "推荐平台后端实习生", "company": "小红书", "location": "上海",
     "description": "参与推荐平台后端开发，负责特征抽取服务和模型推理引擎。要求 Python/Go、gRPC、Docker，了解推荐系统架构。",
     "employment_type": "实习", "source": "fixture"},

    # ----- CV方向 (3 条) -----
    {"title": "计算机视觉算法实习生（检测方向）", "company": "商汤科技", "location": "上海",
     "description": "参与目标检测和图像分割算法研发，使用 YOLO/DETR/MaskRCNN 等模型。要求 Python、PyTorch、OpenCV，有 CV 项目经验优先。",
     "employment_type": "实习", "source": "fixture"},
    {"title": "Computer Vision Engineer", "company": "Tesla", "location": "Palo Alto",
     "description": "Develop computer vision algorithms for autonomous driving. Work on object detection, lane detection, and scene understanding. Requirements: Python, PyTorch, OpenCV.",
     "employment_type": "Full-time", "source": "lever"},
    {"title": "图像识别算法实习生", "company": "旷视科技", "location": "北京",
     "description": "参与图像分类、OCR 识别算法研发。要求 Python、PyTorch、CNN/ResNet/ViT 等模型经验，有模型轻量化和部署经验优先。",
     "employment_type": "实习", "source": "fixture"},

    # ----- NLP方向 (3 条) -----
    {"title": "NLP Research Intern", "company": "Google", "location": "Mountain View",
     "description": "Research in natural language processing. Work on multilingual models, text generation, and dialogue systems. Requirements: PyTorch, HuggingFace, publications.",
     "employment_type": "Internship", "source": "greenhouse"},
    {"title": "搜索 NLP 算法实习生", "company": "搜狗", "location": "北京",
     "description": "参与搜索 Query 理解和排序优化，包括意图识别、Query 纠错改写。要求 NLP 基础，Python，PyTorch/BERT，有搜索/对话系统项目优先。",
     "employment_type": "实习", "source": "fixture"},
    {"title": "文本生成算法工程师", "company": "阅文集团", "location": "上海",
     "description": "负责网络文学 AI 创作和文本生成模型优化。要求 NLP 背景，熟悉 GPT/BART 等生成式模型，有模型微调和部署经验。",
     "employment_type": "社招", "source": "fixture"},

    # ----- 数据分析方向 (3 条) -----
    {"title": "校招数据分析实习生", "company": "滴滴出行", "location": "北京",
     "description": "参与出行平台数据分析，设计指标体系、构建用户画像。要求 SQL 熟练，Python 数据分析，有 A/B 实验和可视化经验优先。",
     "employment_type": "校招", "source": "fixture"},
    {"title": "推荐数据分析实习生", "company": "Bilibili", "location": "上海",
     "description": "分析视频推荐效果指标，包括 CTR/CVR/用户停留时长。要求 SQL/Python 数据分析，了解推荐系统指标体系，有 Hadoop/Spark 经验优先。",
     "employment_type": "实习", "source": "fixture"},
    {"title": "Data Analyst - Growth", "company": "Canva", "location": "Sydney",
     "description": "Analyze user growth metrics and build dashboards. Design A/B tests and measure impact of product changes. Requirements: SQL, Python/R, Tableau.",
     "employment_type": "Full-time", "source": "lever"},

    # ----- 搜索方向 (2 条) -----
    {"title": "智能搜索算法实习生", "company": "知乎", "location": "北京",
     "description": "优化知乎内容搜索排序，包括向量检索、文本相关性排序。要求 NLP/搜索背景，Python/PyTorch，了解 ES/向量数据库。",
     "employment_type": "实习", "source": "fixture"},
    {"title": "校招人岗匹配算法实习生", "company": "BOSS直聘", "location": "北京",
     "description": "参与人岗匹配算法研发，包括简历-岗位语义匹配、Embedding 召回、排序模型优化。要求 NLP/推荐系统背景，有向量检索和排序项目经验。",
     "employment_type": "校招", "source": "fixture"},

    # ----- 游戏AI/其他方向 (1 条) -----
    {"title": "游戏 AI 算法实习生", "company": "网易游戏", "location": "广州",
     "description": "参与游戏 AI 研发，包括 NPC 行为决策、强化学习训练。要求 Python、PyTorch、强化学习基础，有游戏 AI 项目经验优先。",
     "employment_type": "实习", "source": "fixture"},
]


# ---------------------------------------------------------------------------
# 测试函数
# ---------------------------------------------------------------------------

def test_normalize():
    """测试 normalize_public_job 对每条 fixture 输出合法 schema。"""
    print("[TEST] normalize_public_job ...")
    required_fields = ["id", "title", "company", "city", "stage", "direction",
                       "skills", "project_signals", "jd", "interview_themes",
                       "source", "source_url", "posted_at", "data_quality_score"]
    passed = 0
    failed = 0
    for i, raw in enumerate(FIXTURE_JOBS):
        job = normalize_public_job(raw)
        if job is None:
            print(f"  [WARN] fixture[{i}] returned None")
            failed += 1
            continue
        missing = [f for f in required_fields if f not in job]
        if missing:
            print(f"  [FAIL] fixture[{i}] '{job.get('title', '?')}' missing fields: {missing}")
            failed += 1
        else:
            passed += 1

    ok = passed > 0 and failed == 0
    print(f"  [{'PASS' if ok else 'FAIL'}] normalize: {passed} passed, {failed} failed")


def test_direction():
    """测试方向推断。"""
    print("[TEST] infer_direction ...")
    cases = [
        ("大模型应用算法实习生", "负责 LLM 应用开发，包括 RAG 和 Agent 工作流。", "LLM"),
        ("推荐算法实习生", "优化推荐排序模型，提升 CTR 和 NDCG。", "推荐算法"),
        ("计算机视觉算法实习生", "使用 YOLO 进行目标检测和图像分割。", "计算机视觉"),
        ("后端研发实习生", "开发 Go 微服务，使用 gRPC 和 K8s。", "后端开发"),
        ("数据分析实习生", "分析用户数据，设计指标体系。", "数据分析"),
    ]
    passed = 0
    for title, desc, expected_dir in cases:
        direction = infer_direction(title, desc)
        if expected_dir in direction:
            passed += 1
        else:
            print(f"  [WARN] '{title}' -> '{direction}', expected '{expected_dir}'")
    ok = passed >= 4
    print(f"  [{'PASS' if ok else 'FAIL'}] direction infer: {passed}/{len(cases)}")


def test_skills():
    """测试技能提取。"""
    print("[TEST] _find_skills ...")
    desc = ("Requirements: Python, PyTorch, LangChain, FAISS, Docker. "
            "Experience with RAG and Agent workflows. Knowledge of BERT, GPT, and Transformers.")
    skills = _find_skills(desc)
    expected_subset = ["python", "pytorch", "langchain", "faiss", "docker", "bert", "gpt"]
    hits = sum(1 for s in expected_subset if s in skills)
    ok = hits >= 5
    print(f"  [{'PASS' if ok else 'FAIL'}] Found {len(skills)} skills, {hits}/{len(expected_subset)} expected present")


def test_quality():
    """测试质量评分。"""
    print("[TEST] _quality_score ...")
    s1 = _quality_score("LLM Engineer", "Short desc.", "LLM", ["python"])
    s2 = _quality_score("大模型应用算法实习生", "这是一段很详细的岗位描述，" * 10, "LLM",
                        ["python", "pytorch", "langchain", "faiss"])
    ok = s2 > s1
    print(f"  [{'PASS' if ok else 'FAIL'}] short JD score={s1}, long JD score={s2}")


def test_dedup_move():
    """测试去重。"""
    print("[TEST] deduplicate_jobs ...")
    jobs = [
        {"title": "A", "company": "X", "city": "北京"},
        {"title": "A", "company": "X", "city": "北京"},
        {"title": "B", "company": "Y", "city": "上海"},
    ]
    result = deduplicate_jobs(jobs)
    ok = len(result) == 2
    print(f"  [{'PASS' if ok else 'FAIL'}] dedup: {len(jobs)} -> {len(result)}")


def test_merge():
    """测试合并。"""
    print("[TEST] merge_jobs ...")
    builtin = [
        {"title": "大模型应用算法实习生", "company": "字节跳动", "city": "北京"},
        {"title": "推荐算法实习生", "company": "快手", "city": "北京"},
    ]
    public = [
        {"title": "LLM Engineer", "company": "OpenAI", "city": "SF"},
        {"title": "大模型应用算法实习生", "company": "字节跳动", "city": "北京"},  # 重复
    ]
    merged = merge_jobs(builtin, public)
    ok = len(merged) == 3
    print(f"  [{'PASS' if ok else 'FAIL'}] merge: {len(builtin)}+{len(public)} -> {len(merged)} (dup removed)")


def generate_sample_files():
    """生成非破坏性的测试样例文件，不覆盖正式岗位库。"""
    data_dir = os.path.join(ROOT, "data")
    os.makedirs(data_dir, exist_ok=True)

    # 标准化所有 fixture
    normalized = []
    for raw in FIXTURE_JOBS:
        job = normalize_public_job(raw)
        if job:
            normalized.append(job)

    # 去重
    deduped = deduplicate_jobs(normalized)

    # 至少保底取前 30 条（去重后可能少于30）
    sample = deduped[:min(len(deduped), 30)]

    # 保存到 test_* 文件，避免覆盖正式 data/public_jobs_sample.json。
    sample_path = os.path.join(data_dir, "test_public_jobs_sample.json")
    with open(sample_path, "w", encoding="utf-8") as f:
        json.dump(sample, f, ensure_ascii=False, indent=2)

    # 合并
    builtin_path = os.path.join(data_dir, "jobs.json")
    if os.path.exists(builtin_path):
        with open(builtin_path, "r", encoding="utf-8") as f:
            builtin = json.load(f)
    else:
        builtin = []

    merged = merge_jobs(builtin, sample)
    merged_path = os.path.join(data_dir, "test_jobs_merged.json")
    with open(merged_path, "w", encoding="utf-8") as f:
        json.dump(merged, f, ensure_ascii=False, indent=2)

    print(f"[OK] test_public_jobs_sample.json: {len(sample)} jobs")
    print(f"[OK] test_jobs_merged.json: {len(merged)} jobs (builtin {len(builtin)} + public {len(sample)} = {len(builtin)}+{len(sample)} -> {len(merged)} after dedup)")


# ---------------------------------------------------------------------------
# main
# ---------------------------------------------------------------------------

def main():
    print("=" * 60)
    print("  test_data_ingestion.py")
    print("=" * 60)

    test_normalize()
    test_direction()
    test_skills()
    test_quality()
    test_dedup_move()
    test_merge()
    generate_sample_files()

    print("\n[OK] All tests completed.")


if __name__ == "__main__":
    main()