"""把「认不出是谁」匹配到员工 —— 画像、打分、AI 复核的边界。

这个模块的测试里最重要的不是「能不能匹配上」,而是三条不能破的:

1. **handle 字面那几档几乎不产出** —— 这是生产实测(217 个里 0 命中),
   所以真信号必须来自 GitHub 画像。有一条用例专门把这个事实钉住。
2. **模型不能造人。** 返回一个不在候选集里的 person_id 必须被丢掉。
3. **资料为空不等于外部人员。** 大多数人什么都不填,把「没填」当成外部
   会造成大量误标,而误标的后果是这个人离职时不会被回收。
"""
from __future__ import annotations

import uuid
from datetime import UTC, datetime, timedelta

import pytest

from app.domains.github_org import handle_matching as hm
from app.domains.github_org import service as gh
from app.identity.person_models import IdentityPerson
from app.integrations.driver_base import GithubUserProfile

_async = pytest.mark.asyncio


def _person(
    *, email="zhang.san@pingcap.com", name="Zhang San",
    personal=None, employee_no=None, status="active",
) -> IdentityPerson:
    return IdentityPerson(
        id=uuid.uuid4(), source_system="test",
        hris_person_id=f"P{uuid.uuid4().hex[:8]}",
        work_email=email, personal_email=personal,
        display_name=name, employment_status=status,
        employee_no=employee_no,
    )


# ── 打分:画像是真信号 ────────────────────────────────────────────


def test_profile_email_matching_the_work_email_is_the_strongest_signal():
    """GitHub 上公开的邮箱就是企业邮箱 —— 基本没有别的解释。

    这是生产上真实存在的形态:``anotherrachel`` 的资料里写着
    ``yajing.wang@pingcap.com``。
    """
    p = _person(email="yajing.wang@pingcap.com", name="Yajing Wang")
    profile = GithubUserProfile(
        login="anotherrachel", email="yajing.wang@pingcap.com",
        name="Yajing Wang",
    )
    out = hm.score_candidates("anotherrachel", [p], profile=profile)
    assert len(out) == 1
    assert out[0].score == hm.SCORE_PROFILE_EMAIL_EXACT
    assert any("公开的邮箱就是本人邮箱" in r for r in out[0].reasons)


def test_profile_email_also_matches_the_personal_email():
    """公开在 GitHub 上的往往是个人邮箱,而那恰恰是最硬的识别证据 ——
    不看 personal_email 会把这一档白白丢掉。"""
    p = _person(email="a@pingcap.com", personal="rachel@gmail.com")
    profile = GithubUserProfile(login="x", email="rachel@gmail.com")
    out = hm.score_candidates("x", [p], profile=profile)
    assert out and out[0].score == hm.SCORE_PROFILE_EMAIL_EXACT


def test_profile_name_matches_with_the_name_order_reversed():
    """中文姓名的英文写法两种顺序都见得到,它们是同一个人。"""
    p = _person(name="Yajing Wang")
    profile = GithubUserProfile(login="zz-jason", name="Wang Yajing")
    out = hm.score_candidates("zz-jason", [p], profile=profile)
    assert out and out[0].score == hm.SCORE_PROFILE_NAME_TOKENS


def test_one_letter_apart_is_not_a_match():
    """``zhangsan`` 与 ``zhangshan`` 是两个人。刻意不做编辑距离 ——
    一个说不出理由的相似度,在这里的代价是把成员资格挂错人。"""
    p = _person(email="zhangshan@pingcap.com", name="Zhangshan")
    out = hm.score_candidates("zhangsan", [p], profile=None)
    assert out == []


def test_the_real_production_handles_score_nothing_without_a_profile():
    """🔴 钉住那条决定本模块设计的实测事实。

    这些是生产上真实的「认不出是谁」的 handle。没有画像时,它们对任何
    候选人都打不出分 —— 217 个里 0 命中。如果哪天有人把「靠 handle 字面
    猜」的阈值放宽到能命中它们,这条会红,而那正是需要被拦住的改动。
    """
    people = [
        _person(email="liucong@pingcap.com", name="Cong Liu"),
        _person(email="jian.zhang@pingcap.com", name="Jian Zhang"),
        _person(email="lei.wang@pingcap.com", name="Lei Wang"),
    ]
    for handle in ("Ehco1996", "zz-jason", "hawkingrei", "suzaku", "innerr"):
        assert hm.score_candidates(handle, people, profile=None) == [], handle


def test_the_same_handles_do_match_once_the_profile_is_known():
    """同一批 handle,加上画像就认得出来 —— 这就是画像富化的全部价值。"""
    p = _person(email="liucong@pingcap.com", name="Cong Liu")
    profile = GithubUserProfile(
        login="innerr", name="Cong Liu", company="@pingcap",
    )
    out = hm.score_candidates("innerr", [p], profile=profile)
    assert out and out[0].score == hm.SCORE_PROFILE_NAME_EXACT


# ── 外部信号:宁可不判,不可乱判 ──────────────────────────────────


def test_a_different_company_is_an_external_signal():
    profile = GithubUserProfile(login="x", company="ByteDance")
    assert "ByteDance" in (hm.external_signal(profile) or "")


def test_a_bot_account_is_an_external_signal():
    profile = GithubUserProfile(login="tidbcloud-bot", account_type="Bot")
    assert "Bot" in (hm.external_signal(profile) or "")


def test_an_empty_profile_is_NOT_an_external_signal():
    """🔴 最重要的一条负面用例。

    大多数人什么都不填。把「没填 company」当成「是外部人员」,会把一半
    员工误标成外部 —— 而一条被误标成外部的成员资格,后果是这个人离职时
    不会被回收。那是这套东西最不想发生的事。
    """
    assert hm.external_signal(GithubUserProfile(login="x")) is None
    assert hm.external_signal(GithubUserProfile(login="x", company="")) is None
    assert hm.external_signal(GithubUserProfile(login="x", company="   ")) is None
    assert hm.external_signal(None) is None


def test_a_pingcap_company_is_not_external():
    for company in ("@pingcap", "PingCAP", "pingcap.com", "TiDB Cloud"):
        p = GithubUserProfile(login="x", company=company)
        assert hm.external_signal(p) is None, company


# ── AI 复核的护栏 ────────────────────────────────────────────────


class _FakeLLM:
    def __init__(self, reply) -> None:
        self._reply = reply
        self.calls = 0

    async def chat_json(self, **kwargs):
        self.calls += 1
        self._last = kwargs
        return self._reply


def _items(handle="octocat", candidates=None):
    return [
        {
            "handle": handle,
            "membership_ids": ["m1"],
            "org_slugs": ["pingcap"],
            "profile": {"name": "Octo Cat", "company": "ByteDance"},
            "external_hint": None,
            "affiliation_hint": None,
            "candidates": candidates
            if candidates is not None
            else [
                {
                    "person_id": "p-allowed", "person_label": "A",
                    "person_email": "a@x.test", "department": None,
                    "employment_status": "active", "score": 60,
                    "reasons": ["x"],
                }
            ],
        }
    ]


@_async
async def test_ai_cannot_invent_a_person_outside_the_candidate_set(
    db_session, monkeypatch,
):
    """🔴 幻觉护栏。

    一个不在候选集里的 person_id 会让界面上出现一个「AI 建议」按钮,点
    下去把成员资格挂到一个随机的人身上,而那个人离职时别人的访问权被撤。
    """
    fake = _FakeLLM(
        {"results": [
            {"handle": "octocat", "person_id": "p-HALLUCINATED",
             "confidence": "high", "reason": "编的"},
        ]}
    )
    monkeypatch.setattr(hm, "_llm_client", lambda db: _wrap(fake))
    out = await hm.ai_review(db_session, _items())
    assert out["octocat"]["person_id"] is None


@_async
async def test_ai_can_pick_a_person_inside_the_candidate_set(
    db_session, monkeypatch,
):
    fake = _FakeLLM(
        {"results": [
            {"handle": "octocat", "person_id": "p-allowed",
             "confidence": "high", "reason": "邮箱对上了"},
        ]}
    )
    monkeypatch.setattr(hm, "_llm_client", lambda db: _wrap(fake))
    out = await hm.ai_review(db_session, _items())
    assert out["octocat"]["person_id"] == "p-allowed"
    assert out["octocat"]["external"] is False


@_async
async def test_external_and_a_person_are_mutually_exclusive(
    db_session, monkeypatch,
):
    """模型同时说「是这个人」和「是外部人员」时,人优先、external 归 False。
    两个都为真的结论会让界面上两个按钮同时预选。"""
    fake = _FakeLLM(
        {"results": [
            {"handle": "octocat", "person_id": "p-allowed",
             "external": True, "confidence": "low", "reason": "自相矛盾"},
        ]}
    )
    monkeypatch.setattr(hm, "_llm_client", lambda db: _wrap(fake))
    out = await hm.ai_review(db_session, _items())
    assert out["octocat"]["person_id"] == "p-allowed"
    assert out["octocat"]["external"] is False


@_async
async def test_ai_may_answer_external_when_it_picks_nobody(
    db_session, monkeypatch,
):
    fake = _FakeLLM(
        {"results": [
            {"handle": "octocat", "person_id": None, "external": True,
             "confidence": "high", "reason": "company 写着 ByteDance"},
        ]}
    )
    monkeypatch.setattr(hm, "_llm_client", lambda db: _wrap(fake))
    out = await hm.ai_review(db_session, _items())
    assert out["octocat"]["external"] is True


@_async
async def test_no_llm_key_means_no_ai_not_an_error(db_session, monkeypatch):
    """没配 key 时确定性打分照常工作 —— 这是设计,不是降级路径的凑合。"""
    monkeypatch.setattr(hm, "_llm_client", lambda db: _wrap(None))
    assert await hm.ai_review(db_session, _items()) == {}


@_async
async def test_one_failing_batch_does_not_lose_the_others(
    db_session, monkeypatch,
):
    """AI 是加分项,不是必要条件。一批抛异常不该让整次建议为空。"""
    class _Boom:
        async def chat_json(self, **kwargs):
            raise RuntimeError("model unavailable")

    monkeypatch.setattr(hm, "_llm_client", lambda db: _wrap(_Boom()))
    assert await hm.ai_review(db_session, _items()) == {}


async def _wrap(v):
    return v


# ── 画像缓存 ──────────────────────────────────────────────────────


async def _org(db, slug="pingcap"):
    org = await gh.create_org(db, slug=slug, display_name=slug)
    await db.commit()
    return org


async def _membership(db, org, login):
    m = await gh.upsert_membership(
        db, org=org, github_login=login, person_id=None,
        state="active", source="reconcile", actor_kind="integration",
        trace_id="t",
    )
    await db.flush()
    return m


@_async
async def test_a_fresh_cached_profile_is_not_refetched(db_session):
    """一次匹配要看几百个 handle;现拉就是几百次 API 调用外加限流。"""
    org = await _org(db_session)
    m = await _membership(db_session, org, "octocat")
    m.github_profile = {"login": "octocat", "name": "Cached Name"}
    m.profile_fetched_at = datetime.now(UTC).replace(tzinfo=None)
    await db_session.flush()

    calls = []

    async def _fetch(login):
        calls.append(login)
        return GithubUserProfile(login=login, name="Fresh")

    out = await hm.enrich_profiles(
        db_session, {"octocat": [m]}, fetch=_fetch,
    )
    assert calls == []
    assert out["octocat"].name == "Cached Name"


@_async
async def test_a_stale_profile_is_refetched(db_session):
    org = await _org(db_session)
    m = await _membership(db_session, org, "octocat")
    m.github_profile = {"login": "octocat", "name": "Old"}
    m.profile_fetched_at = (
        datetime.now(UTC) - hm.PROFILE_TTL - timedelta(days=1)
    ).replace(tzinfo=None)
    await db_session.flush()

    async def _fetch(login):
        return GithubUserProfile(login=login, name="New")

    out = await hm.enrich_profiles(db_session, {"octocat": [m]}, fetch=_fetch)
    assert out["octocat"].name == "New"
    assert m.github_profile["name"] == "New"


@_async
async def test_a_failed_fetch_does_not_freeze_the_handle_for_a_month(
    db_session,
):
    """🔴 一次限流不能把这个 handle 冻在「已拉过、没内容」30 天。

    做法是失败时**不写** ``profile_fetched_at`` —— 写了就等于告诉下一次
    「这个已经查过了」。
    """
    org = await _org(db_session)
    m = await _membership(db_session, org, "octocat")
    await db_session.flush()
    assert m.profile_fetched_at is None

    async def _fetch(login):
        raise RuntimeError("rate limited")

    out = await hm.enrich_profiles(db_session, {"octocat": [m]}, fetch=_fetch)
    assert out["octocat"] is None
    assert m.profile_fetched_at is None
    assert m.github_profile is None


@_async
async def test_a_missing_github_account_is_recorded_not_retried_forever(
    db_session,
):
    """GitHub 说「没有这个人」(改名或注销)本身是一条结论。记下来,
    否则每次都会重拉一个注定 404 的 handle。"""
    org = await _org(db_session)
    m = await _membership(db_session, org, "ghost")
    await db_session.flush()

    async def _fetch(login):
        return None

    out = await hm.enrich_profiles(db_session, {"ghost": [m]}, fetch=_fetch)
    assert out["ghost"] is None
    assert m.github_profile == {"login": "ghost", "account_missing": True}
    assert m.profile_fetched_at is not None


@_async
async def test_the_profile_is_written_to_every_org_row_of_that_handle(
    db_session,
):
    """一个人可能在五个组织里。拉一次、写五行 —— 否则界面上五个组织
    显示的依据可能不一样,人就没法复核。"""
    a = await _org(db_session, "pingcap")
    b = await _org(db_session, "tikv")
    ma = await _membership(db_session, a, "octocat")
    mb = await _membership(db_session, b, "octocat")
    await db_session.flush()

    calls = []

    async def _fetch(login):
        calls.append(login)
        return GithubUserProfile(login=login, name="Octo")

    await hm.enrich_profiles(db_session, {"octocat": [ma, mb]}, fetch=_fetch)
    assert calls == ["octocat"]
    assert ma.github_profile["name"] == "Octo"
    assert mb.github_profile["name"] == "Octo"


@_async
async def test_no_fetcher_still_returns_the_cache_instead_of_erroring(
    db_session,
):
    """驱动没配好时治理页面仍然要能打开 —— 一个集成故障不该让整页报错。"""
    org = await _org(db_session)
    m = await _membership(db_session, org, "octocat")
    m.github_profile = {"login": "octocat", "name": "Cached"}
    m.profile_fetched_at = datetime.now(UTC).replace(tzinfo=None)
    await db_session.flush()
    out = await hm.enrich_profiles(db_session, {"octocat": [m]}, fetch=None)
    assert out["octocat"].name == "Cached"


@_async
async def test_the_fetch_budget_is_capped_per_run(db_session):
    """让一次页面加载去打几百个外部请求,页面会先超时,而且限流之后连
    已经拉到的都白拉。"""
    org = await _org(db_session)
    rows = {}
    for i in range(5):
        m = await _membership(db_session, org, f"user{i}")
        rows[f"user{i}"] = [m]
    await db_session.flush()

    calls = []

    async def _fetch(login):
        calls.append(login)
        return GithubUserProfile(login=login, name="X")

    await hm.enrich_profiles(
        db_session, rows, fetch=_fetch, max_fetch=2,
    )
    assert len(calls) == 2


# ── suggest_links:永不自动关联 ───────────────────────────────────


@_async
async def test_suggest_links_never_writes_person_id(db_session, monkeypatch):
    """🔴 这个模块唯一不肯让步的地方。

    把一条成员资格挂到错误的人头上,后果是那个人离职时**别人的**访问权
    被撤掉。所以任何置信度下都只产出建议。
    """
    org = await _org(db_session)
    p = _person(email="yajing.wang@pingcap.com", name="Yajing Wang")
    db_session.add(p)
    m = await _membership(db_session, org, "anotherrachel")
    m.github_profile = {
        "login": "anotherrachel", "email": "yajing.wang@pingcap.com",
        "name": "Yajing Wang",
    }
    m.profile_fetched_at = datetime.now(UTC).replace(tzinfo=None)
    await db_session.commit()

    monkeypatch.setattr(hm, "_llm_client", lambda db: _wrap(None))
    out = await hm.suggest_links(db_session, use_ai=False)

    item = next(i for i in out["items"] if i["handle"] == "anotherrachel")
    # 最高置信度的一档都命中了……
    assert item["suggested_person_id"] == str(p.id)
    assert item["suggestion_source"] == "deterministic"
    # ……台账依然没被动过。
    await db_session.refresh(m)
    assert m.person_id is None


@_async
async def test_confirmed_external_rows_leave_the_suggestion_queue(
    db_session, monkeypatch,
):
    """人做过的判断不能被重新捞出来问一遍 —— 那正是这个队列能收敛的原因。"""
    org = await _org(db_session)
    m = await _membership(db_session, org, "bytedancer")
    await db_session.commit()

    monkeypatch.setattr(hm, "_llm_client", lambda db: _wrap(None))
    before = await hm.suggest_links(db_session, use_ai=False)
    assert any(i["handle"] == "bytedancer" for i in before["items"])

    await gh.confirm_external(
        db_session, m, actor_person_id=None, note="ByteDance",
    )
    await db_session.commit()

    after = await hm.suggest_links(db_session, use_ai=False)
    assert not any(i["handle"] == "bytedancer" for i in after["items"])