"""API 测试：TestClient 覆盖 health / jobs / graph / resume / match / learning-path / review。"""
from __future__ import annotations

import json

import pytest
from fastapi.testclient import TestClient

from app.api.routes import app
from app.config import ROOT


@pytest.fixture(scope="module")
def client(graph_path):
    with TestClient(app) as c:  # 触发 startup 加载图谱
        yield c


def test_health(client):
    r = client.get("/api/health")
    assert r.status_code == 200
    body = r.json()
    assert body["code"] == 0 and body["request_id"]
    assert body["data"]["graph"]["jobs"] >= 15


def test_graph_endpoint(client):
    r = client.get("/api/graph")
    assert r.status_code == 200
    data = r.json()["data"]
    assert len(data["nodes"]) > 100 and len(data["edges"]) > 100
    types = {n["type"] for n in data["nodes"]}
    assert "Job" in types and "Skill" in types
    # 行业维度：Industry 节点与 (Job)-[:BELONGS_TO]->(Industry) 边
    assert "Industry" in types
    industries = {n["title"] for n in data["nodes"] if n["type"] == "Industry"}
    assert {"人工智能", "互联网/软件", "金融科技", "智能制造"} <= industries
    assert any(e.get("rel") == "BELONGS_TO" for e in data["edges"])


def test_jobs_list_and_filter(client):
    r = client.get("/api/jobs")
    assert r.status_code == 200
    jobs = r.json()["data"]
    assert len(jobs) >= 15
    r2 = client.get("/api/jobs", params={"keyword": "Java"})
    titles = [j["title"] for j in r2.json()["data"]]
    assert titles == ["Java开发工程师"]


def test_job_detail_and_evolution(client):
    r = client.get("/api/jobs/job_Java开发工程师")
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["definition"]["title"] == "Java开发工程师"
    assert data["snapshot"]["skills"]
    r2 = client.get("/api/jobs/job_Java开发工程师/evolution")
    events = r2.json()["data"]["events"]
    assert events and {"新增", "退出"} & {e["change_type"] for e in events}
    assert all("数据源" in e["description"] for e in events)
    assert client.get("/api/jobs/不存在").status_code == 404


def test_resume_parse_masks_sensitive(client):
    text = "姓名：陈晓\n手机：13900001111\n邮箱：chenx@example.com\n4年经验，熟悉Java与MySQL"
    r = client.post("/api/resume/parse", json={"text": text})
    assert r.status_code == 200
    p = r.json()["data"]
    assert "13900001111" not in json.dumps(p, ensure_ascii=False)
    assert "陈晓" not in json.dumps(p, ensure_ascii=False)
    assert set(p["skills"]) == {"Java", "MySQL"}


def test_match_and_learning_path(client):
    r = client.post("/api/match", json={
        "job_id": "job_Java开发工程师",
        "skills": ["Java", "Spring Boot", "Spring Cloud", "MySQL", "Redis", "分布式系统"],
        "years": 5, "scenarios": ["电商", "金融"]})
    assert r.status_code == 200
    m = r.json()["data"]
    assert m["is_match"] and m["total_score"] >= 60
    assert m["dimension_scores"]["required"] == 100.0

    r2 = client.post("/api/learning-path", json={
        "job_id": "job_Java开发工程师", "known_skills": ["Java"]})
    assert r2.status_code == 200
    path = r2.json()["data"]["learning_path"]
    assert path and "missing_required" in r2.json()["data"]


def test_review_flow(client, tmp_path, monkeypatch):
    """审核流：先造一个待审核候选，再 accept 写入图谱并留痕（SQLite 状态库重定向到临时目录）。"""
    import app.api.routes as routes
    from app.discover.cluster import save_pending
    from app.models import JobDefinition
    from app.storage import db as state_db
    # 把状态库/迁移种子/图谱路径重定向到 tmp，避免污染工程内已提交数据
    db_path = tmp_path / "state.db"
    monkeypatch.setitem(routes._cfg, "state_db_path", str(db_path))
    monkeypatch.setitem(routes._cfg, "audit_path", str(tmp_path / "audit.jsonl"))  # 旧留痕迁移种子（不存在则跳过）
    monkeypatch.setitem(routes._cfg, "graph_path", str(tmp_path / "graph.json"))
    cand = JobDefinition(job_id="cand_测试新岗位", title="测试新岗位",
                         required_skills=["Python"], status="pending",
                         source_count=10, company_count=6)
    save_pending([cand], db_path)
    r = client.post("/api/review", json={
        "candidate_id": "cand_测试新岗位", "action": "accept",
        "operator": "张老师", "reason": "证据充分"})
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["accepted"]["status"] == "active"
    assert data["review"]["before"]["status"] == "pending"
    assert data["review"]["operator"] == "张老师"
    # 审核后能在岗位列表查到
    assert client.get("/api/jobs/job_测试新岗位").status_code == 200
    # 留痕落在 SQLite（结构与旧 jsonl 行一致），候选已从待审核分区移除
    recs = state_db.get_audit(db_path)
    assert recs and recs[0]["operator"] == "张老师" and recs[0]["action"] == "accept"
    assert state_db.get_candidate(db_path, "cand_测试新岗位") is None
    # /api/audit 兼容查询
    r2 = client.get("/api/audit", params={"node_id": "cand_测试新岗位"})
    assert r2.status_code == 200 and r2.json()["data"][0]["candidate_id"] == "cand_测试新岗位"
    # 不存在的候选返回 404
    assert client.post("/api/review", json={
        "candidate_id": "cand_不存在", "action": "accept", "operator": "x"}).status_code == 404


def test_node_update_flow(client, tmp_path, monkeypatch):
    """节点编辑：白名单字段更新 + 图谱落盘 + 审核留痕（SQLite 状态库重定向到临时目录）。"""
    import app.api.routes as routes
    monkeypatch.setitem(routes._cfg, "graph_path", str(tmp_path / "graph.json"))
    monkeypatch.setitem(routes._cfg, "state_db_path", str(tmp_path / "state.db"))
    prev_nodes = client.get("/api/graph").json()["data"]["nodes"]
    prev_level = next(n for n in prev_nodes if n["id"] == "job_Java开发工程师").get("level")
    r = client.post("/api/graph/node/update", json={
        "node_id": "job_Java开发工程师",
        "updates": {"level": "高级", "weight": 0.9, "hack": "ignored"},
        "operator": "张老师", "reason": "webui 校正"})
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["node"]["level"] == "高级" and data["node"]["weight"] == 0.9
    assert "hack" not in data["node"]
    assert data["audit"]["before"]["level"] == prev_level
    # Job 节点编辑同步 JobDefinition：级别同步回期望年限，/api/jobs 一致
    assert data["node"]["expected_years"] == 6
    d = client.get("/api/jobs/job_Java开发工程师").json()["data"]["definition"]
    assert d["expected_years"] == 6
    assert (tmp_path / "graph.json").exists()
    # 留痕可查
    r2 = client.get("/api/audit", params={"node_id": "job_Java开发工程师"})
    assert r2.status_code == 200
    recs = r2.json()["data"]
    assert recs and recs[0]["operator"] == "张老师"
    # 不存在的节点 / 空更新
    assert client.post("/api/graph/node/update", json={
        "node_id": "不存在", "updates": {"level": "x"}, "operator": "x"}).status_code == 404
    assert client.post("/api/graph/node/update", json={
        "node_id": "job_Java开发工程师", "updates": {"hack": 1}, "operator": "x"}).status_code == 400


def test_jd_lookup(client):
    """JD 原文回查：合法 source_id 返回记录，非法输入 400/404。"""
    g = client.get("/api/graph").json()["data"]
    sid = next(s for e in g["edges"] for s in e.get("sources", []))
    r = client.get("/api/jd", params={"sid": sid})
    assert r.status_code == 200
    rec = r.json()["data"]
    assert rec["source_id"] == sid and rec["text"] and rec["job_title"]
    assert client.get("/api/jd", params={"sid": "无行号"}).status_code == 400
    assert client.get("/api/jd", params={"sid": "../evil.jsonl#1"}).status_code == 400
    assert client.get("/api/jd", params={"sid": "batch_2019.jsonl#99999"}).status_code == 404


def _minimal_pdf(text: str) -> bytes:
    """构造含可提取文本的最小合法 PDF（ASCII 内容）。"""
    stream = f"BT /F1 12 Tf 72 720 Td ({text}) Tj ET".encode()
    objs = [
        b"<< /Type /Catalog /Pages 2 0 R >>",
        b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>",
        b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] "
        b"/Resources << /Font << /F1 4 0 R >> >> /Contents 5 0 R >>",
        b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>",
        b"<< /Length " + str(len(stream)).encode() + b" >>\nstream\n" + stream + b"\nendstream",
    ]
    out = b"%PDF-1.4\n"
    offsets = []
    for i, body in enumerate(objs, 1):
        offsets.append(len(out))
        out += f"{i} 0 obj\n".encode() + body + b"\nendobj\n"
    xref_pos = len(out)
    out += f"xref\n0 {len(objs)+1}\n".encode() + b"0000000000 65535 f \n"
    for off in offsets:
        out += f"{off:010d} 00000 n \n".encode()
    out += (f"trailer\n<< /Size {len(objs)+1} /Root 1 0 R >>\n"
            f"startxref\n{xref_pos}\n%%EOF").encode()
    return out


def test_resume_file_upload_docx_and_pdf(client, tmp_path):
    """简历文件上传：docx/pdf 均解析出技能并脱敏，非法格式与超限被拒。"""
    from docx import Document

    # docx：python-docx 真实生成
    doc = Document()
    doc.add_paragraph("姓名：李晓峰 手机：13912345678")
    doc.add_paragraph("6年 Java 后端经验，熟悉 Java、Spring Boot、MySQL、Redis")
    p = tmp_path / "r.docx"
    doc.save(str(p))
    r = client.post("/api/resume/parse/file",
                    files={"file": ("r.docx", p.read_bytes(), "application/octet-stream")})
    assert r.status_code == 200
    data = r.json()["data"]
    assert "Java" in data["skills"] and "MySQL" in data["skills"]
    assert "13912345678" not in data["raw_masked_text"]  # 已脱敏
    assert data["name"].startswith("李") and "晓峰" not in data["name"]

    # pdf：最小合法 PDF（ASCII 技能名，验证 pypdf 链路）
    pdf = _minimal_pdf("Java Spring Boot MySQL Redis Docker Kubernetes")
    r2 = client.post("/api/resume/parse/file",
                     files={"file": ("r.pdf", pdf, "application/pdf")})
    assert r2.status_code == 200
    skills2 = r2.json()["data"]["skills"]
    assert "Java" in skills2 and "Kubernetes" in skills2

    # txt 也走同一入口
    r3 = client.post("/api/resume/parse/file",
                     files={"file": ("r.txt", "熟悉 Python 与 TensorFlow".encode(), "text/plain")})
    assert r3.status_code == 200 and "Python" in r3.json()["data"]["skills"]

    # 非法格式 / 空文件
    bad = client.post("/api/resume/parse/file",
                      files={"file": ("r.exe", b"MZ", "application/octet-stream")})
    assert bad.status_code == 400
    empty = client.post("/api/resume/parse/file",
                        files={"file": ("r.pdf", b"", "application/pdf")})
    assert empty.status_code == 400


def test_match_normalizes_manual_skills(client):
    """手动录入技能经大小写/别名归一化后再匹配：redis→Redis、go→Go、K8s→Kubernetes。"""
    r = client.post("/api/match", json={
        "job_id": "job_Go开发工程师",
        "skills": ["go", "redis", "grpc", "K8s", "kafka", "微服务"],
        "years": 5})
    assert r.status_code == 200
    m = r.json()["data"]
    hit = set(m["matched_required"]) | set(m["matched_preferred"])
    assert {"Go", "Redis", "gRPC", "Kubernetes", "Kafka", "微服务"} <= hit
    assert "redis" not in hit and "K8s" not in hit  # 返回的是标准名


def test_skill_normalize_semantic_fallback():
    """normalize 三级容错：别名→包含匹配（取最长）→原样返回。"""
    from app.extract.base import SkillMatcher, load_lexicon
    m = SkillMatcher(load_lexicon())
    assert m.normalize("redis") == "Redis"          # 大小写
    assert m.normalize("K8s") == "Kubernetes"       # 别名
    assert m.normalize("微服务架构") == "微服务"      # 中文包含
    assert m.normalize("Java语言") == "Java"         # 后缀容错
    assert m.normalize("JavaScript") == "JavaScript"  # 最长优先，不被 Java 截胡
    assert m.normalize("222") == "222"               # 词表外原样返回


def test_resume_structured_and_export(client):
    """结构化简历：上传 docx 提取照片/基本信息/分节；导出 DOCX 合法且含关键词。"""
    from app.config import ROOT as _ROOT
    p = _ROOT / "data" / "resumes" / "resume_full.docx"
    r = client.post("/api/resume/parse/file",
                    files={"file": ("resume_full.docx", p.read_bytes(), "application/octet-stream")})
    assert r.status_code == 200
    prof = r.json()["data"]
    assert prof["photo_b64"].startswith("data:image/")          # 照片识别
    assert prof["gender"] == "男" and prof["education"] == "本科"
    assert prof["address"].endswith("***")                       # 住址脱敏
    titles = [s["title"] for s in prof["sections"]]
    assert "教育经历" in titles and "项目经历" in titles          # 分节
    assert "13812345678" not in prof["raw_masked_text"]          # 手机脱敏

    # 匹配后导出 DOCX
    m = client.post("/api/match", json={
        "job_id": "job_Java开发工程师",
        "skills": prof["skills"], "years": prof["years"]}).json()["data"]
    r2 = client.post("/api/resume/export", json={"profile": prof, "match": m})
    assert r2.status_code == 200
    assert r2.headers["content-type"].startswith("application/vnd.openxmlformats")
    import io, zipfile
    z = zipfile.ZipFile(io.BytesIO(r2.content))
    assert "word/document.xml" in z.namelist()                   # 合法 docx
    assert any(n.startswith("word/media/") for n in z.namelist())  # 照片已嵌入


def test_runtime_settings_and_token_guard(client, monkeypatch, tmp_path):
    """设置读写 config.json（白名单+校验）、LLM 热生效判定、访问令牌鉴权。"""
    import app.config as cfgmod
    from app.api import routes
    monkeypatch.setattr(cfgmod, "CONFIG_JSON", tmp_path / "config.json")
    monkeypatch.setattr(routes, "load_runtime_settings", cfgmod.load_runtime_settings)

    # 保存设置：host/port 校验
    assert client.post("/api/config", json={"host": "8.8.8.8"}).status_code == 400
    assert client.post("/api/config", json={"port": 80}).status_code == 400
    r = client.post("/api/config", json={
        "host": "0.0.0.0", "port": 9012, "llm_api_key": "sk-test-1234", "mode": "demo"})
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["restart_required"] is True          # 保存的监听与运行中不同
    assert data["llm"]["configured"] is True         # LLM 热生效
    assert data["llm"]["api_key_masked"].endswith("1234")
    assert "sk-test-1234" not in r.text              # 密钥不落明文
    saved = cfgmod.load_runtime_settings()
    assert saved["llm_api_key"] == "sk-test-1234" and saved["port"] == 9012
    # 非白名单键不写入
    client.post("/api/config", json={})
    assert "hack" not in cfgmod.load_runtime_settings()

    # 令牌鉴权：设置令牌后 /api/* 需带 X-Access-Token，health 与 GET config 豁免
    monkeypatch.setitem(routes._cfg, "access_token", "tok-abc")
    assert client.get("/api/jobs").status_code == 401
    assert client.get("/api/jobs", headers={"X-Access-Token": "tok-abc"}).status_code == 200
    assert client.get("/api/health").status_code == 200
    assert client.get("/api/config").status_code == 200
    assert client.post("/api/config", json={}).status_code == 401
    monkeypatch.delitem(routes._cfg, "access_token")


def test_get_extractor_llm_key_from_config():
    """get_extractor：config.json 提供 llm_api_key 即启用 LLM 通道，无密钥回退离线。"""
    from app.extract.llm import LLMExtractor, get_extractor
    from app.extract.offline import OfflineExtractor
    e1 = get_extractor({"llm_api_key": "sk-x", "llm_base_url": "http://x/v1",
                        "llm_model": "m", "lexicon_path": None})
    assert isinstance(e1, LLMExtractor)
    e2 = get_extractor({"lexicon_path": None, "llm_api_key_env": "NO_SUCH_ENV_VAR"})
    assert isinstance(e2, OfflineExtractor)


def test_embed_model_status_and_config(client, tmp_path, monkeypatch):
    """语义权重状态：未配置/假目录/缺依赖时 ready=False 且状态字段正确；设置可保存。"""
    import app.config as cfgmod
    monkeypatch.setattr(cfgmod, "CONFIG_JSON", tmp_path / "config.json")  # 不写真实 config.json
    from app.match import semantic
    assert not semantic.weights_present("") and not semantic.weights_present(tmp_path)
    # 伪造一个"文件齐全"的权重目录
    for f in ["config.json", "model.safetensors", "tokenizer.json"]:
        (tmp_path / f).write_text("{}")
    assert semantic.weights_present(tmp_path)
    st = semantic.status({"embed_model_path": str(tmp_path)})
    assert st["configured"] and st["weights_present"]
    assert st["ready"] == st["deps_installed"]  # 本环境未装 sentence-transformers → False

    r = client.post("/api/config", json={"embed_model_path": str(tmp_path)})
    assert r.status_code == 200
    emb = r.json()["data"]["embed"]
    assert emb["configured"] and emb["weights_present"]
    # health 同步暴露
    h = client.get("/api/health").json()["data"]
    assert h["embed"]["configured"] is True
    # 清除
    r2 = client.post("/api/config", json={"embed_model_path": ""})
    assert r2.json()["data"]["embed"]["configured"] is False


def test_embed_dual_provider(client, tmp_path, monkeypatch):
    """语义双模式：api 模式有 Key 即就绪（无需权重）；local 模式需权重+依赖。"""
    import app.config as cfgmod
    monkeypatch.setattr(cfgmod, "CONFIG_JSON", tmp_path / "config.json")  # 不写真实 config.json
    from app.match import semantic
    st_api = semantic.status({"embed_provider": "api", "embed_api_key": "sk-1234"})
    assert st_api["ready"] is True and st_api["provider"] == "api"
    st_local = semantic.status({"embed_provider": "local", "embed_model_path": ""})
    assert st_local["ready"] is False
    # most_similar 在未就绪时安全返回 None
    assert semantic.most_similar("微服务", {"embed_provider": "local"}, {"skills": [{"name": "微服务"}]}) is None

    r = client.post("/api/config", json={
        "embed_provider": "api", "embed_api_key": "sk-dash-9999"})
    emb = r.json()["data"]["embed"]
    assert emb["ready"] is True and emb["api_key_masked"].endswith("9999")
    assert "sk-dash-9999" not in r.text
    # 还原：回到本地模式并清除 key
    r2 = client.post("/api/config", json={"embed_provider": "local", "embed_api_key": ""})
    assert r2.json()["data"]["embed"]["ready"] is False


def test_match_batch(client):
    """批量匹配：多份简历打分降序、坏文件记 error 不拖垮整批。"""
    from app.config import ROOT as _ROOT
    resumes = _ROOT / "data" / "resumes"
    files = []
    for fn in ["resume_full.txt", "resume_partial.txt", "resume_mismatch.txt"]:
        files.append(("files", (fn, (resumes / fn).read_bytes(), "text/plain")))
    files.append(("files", ("bad.exe", b"MZ", "application/octet-stream")))
    r = client.post("/api/match/batch",
                    data={"job_id": "job_Java开发工程师"}, files=files)
    assert r.status_code == 200
    data = r.json()["data"]
    assert data["total"] == 4 and data["ok"] == 3
    scores = [x["total_score"] for x in data["results"] if "error" not in x]
    assert scores == sorted(scores, reverse=True)          # 按总分降序
    assert any("error" in x for x in data["results"])      # 坏文件有错误记录
    top = data["results"][0]
    assert top["matched"] and top["name"]                  # 第一名有命中与脱敏姓名
    assert client.post("/api/match/batch",
                       data={"job_id": "不存在"}, files=files[:1]).status_code == 404


def test_semantic_setup_endpoints(client, monkeypatch):
    """一键配置：状态接口默认空闲；已就绪时幂等快速完成；重复启动不并发。"""
    import app.api.routes as routes
    from app.config import ROOT as _ROOT
    # 让运行配置处于"语义已就绪"状态，命中幂等快路径（避免测试里真跑 pip）
    monkeypatch.setitem(routes._cfg, "embed_provider", "local")
    monkeypatch.setitem(routes._cfg, "embed_model_path", str(_ROOT / "models" / "bge-m3"))
    routes._semantic_setup.update({"running": False, "stage": "", "log": [], "error": "", "done": False})
    st = client.get("/api/semantic/setup/status").json()["data"]
    assert st["running"] is False and "embed" in st
    r = client.post("/api/semantic/setup")
    assert r.status_code == 200
    import time
    st2 = st
    for _ in range(50):
        st2 = client.get("/api/semantic/setup/status").json()["data"]
        if not st2["running"]:
            break
        time.sleep(0.2)
    assert st2["done"] is True and st2["error"] == ""
    assert st2["embed"]["ready"] is True