"""把「认不出是谁」的 GitHub handle 匹配到员工 —— 画像富化 + 打分 + AI 复核。

上线导入后:1016 条成员资格里 **331 条认不出是谁**(33%)。那批人不在离职
自动回收的覆盖范围内,所以它是这套东西最大的一块治理缺口。

## 先说一个实测结论,它决定了这个模块长什么样

最初的设计只拿 handle 字符串本身做字面比对。**拿生产数据一跑,217 个认不出
的独立 handle 里,归一后与任何企业邮箱前缀字面相同的是 0 个。** 它们长成
``Ehco1996`` / ``zz-jason`` / ``hawkingrei`` —— handle 和员工姓名之间根本没有
可靠的字面关系,那一层几乎不产出任何东西。

信号在 **GitHub 公开个人资料**里。同一批抽样 30 个 handle:

* 21 个填了真实姓名(``Yajing Wang`` / ``Calvin Neo``);
* 10 个公开了邮箱,其中有的**直接就是企业邮箱**(``yajing.wang@pingcap.com``);
* 12 个填了 company,其中若干写着 ``@pingcap``。

而 company 写着别家公司或基金会的那些,回答了另一个更重要的问题。

## 「认不出是谁」必须能收敛成两堆

那 12 个填了 company 的样本里,有相当一部分写的是别家公司、基金会、高校
—— **他们本来就不是员工,永远也「认」不出来**。如果所有 331 条都留在一个
待办数里,那个数永远不会归零,而一个永远不归零的待办数等于没有待办数:
看的人第二周就不看了。

所以这里的输出有两种,不是一种:

1. **建议关联到某个员工** —— 人点一下确认;
2. **建议标记为外部人员** —— 人点一下确认,这一条就离开待办队列。

第二种不是把问题扫掉:被标记为外部的行仍然留在台账里,他们确实持有访问权。
它只是把「谁该管这条」说清楚 —— 离职回收管不了不在 HRIS 里的人。

## 这个模块唯一不肯让步的地方

**永不自动关联,任何置信度下都不。** 把一条成员资格挂到错误的人头上,后果是
那个人离职时**别人的**访问权被撤掉。漏判的代价是「有人多点一下」,错判的
代价是一次错误的权限回收 —— 两者不对等,所以宁可漏。

没配 LLM key 时退化为「画像 + 确定性打分」,这是设计而不是降级:画像里的
邮箱直接命中那一档本身就是最高置信度的一档,它不需要模型。
"""
from __future__ import annotations

import logging
import re
import unicodedata
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from typing import Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.identity.person_models import IdentityPerson
from app.integrations.driver_base import GithubUserProfile
from app.integrations.models import PlatformSetting

from . import constants as const
from .models import GithubOrg, GithubOrgMembership

logger = logging.getLogger(__name__)

#: 复用 AI 花费那套 LLM 配置,不新开一套 key。
_KEY_API = "ai_spending.llm.api_key"
_KEY_MODEL = "ai_spending.llm.model"
_KEY_REGION = "ai_spending.llm.region"

#: 打分分档。数字本身不重要,重要的是**顺序**与每一档配的理由 —— 界面上
#: 显示的是理由,人是靠理由判断的,不是靠分数。
#:
#: 前两档来自 GitHub 画像,后面几档来自 handle 字面。这个顺序不是审美:
#: 字面那几档在生产数据上实测产出为 0,画像那几档才是真正干活的。
SCORE_PROFILE_EMAIL_EXACT = 99
SCORE_PROFILE_EMAIL_LOCAL = 92
SCORE_PROFILE_NAME_EXACT = 88
SCORE_PROFILE_NAME_TOKENS = 86
SCORE_EMAIL_LOCAL_EXACT = 82
SCORE_DISPLAY_NAME_COMPACT = 78
SCORE_EMPLOYEE_NO = 74
SCORE_PROFILE_NAME_PARTIAL = 62
SCORE_EMAIL_LOCAL_CONTAINS = 58
SCORE_NAME_TOKEN = 54

#: 低于它不作为候选给出去 —— 一份充满噪音的候选清单比没有候选更糟,因为
#: 人会开始不看理由直接点第一个。
MIN_SCORE = 50

#: 到这个分数以上才作为**默认建议**(界面上预选)。低于它仍然显示为候选,
#: 但要人自己挑 —— 预选一个 60 分的候选就是在诱导人点确认。
AUTO_SUGGEST_MIN = SCORE_PROFILE_NAME_EXACT

#: 一次最多让模型看多少条。331 条一次发过去既超上下文,也让一次失败的代价
#: 变成全部 —— 分批,每批独立成败。
LLM_BATCH_SIZE = 20

#: 画像缓存多久算新鲜。个人资料变动很少,而每次界面刷新都现拉几百个 handle
#: 会直接撞 GitHub 限流。30 天足够反映「换了公司」这种真正影响判断的变化。
PROFILE_TTL = timedelta(days=30)

#: 一次请求最多现拉多少个画像。剩下的用缓存或留空 —— 让一次页面加载去打
#: 几百个外部请求,页面会先超时,而且限流之后连已经拉到的都白拉。
MAX_PROFILE_FETCH_PER_RUN = 60

#: company / bio 里出现这些,说明这个人自称与我们相关 —— 是「值得认」的
#: 信号,不是「就是员工」的结论。
_AFFILIATION_MARKERS = ("pingcap", "tidb", "tikv", "ping cap")

ProfileFetcher = Callable[[str], Awaitable[GithubUserProfile | None]]


def _now() -> datetime:
    return datetime.now(UTC)


def _naive(dt: datetime) -> datetime:
    """列是 ``DateTime(timezone=True)``,但 MySQL/TiDB 的 DATETIME 不存时区,
    读回来是 naive 的。写入前和比较前都要统一,否则 aware 与 naive 相比
    直接 TypeError —— 而这里那条比较在「缓存过期了吗」的路径上,一进就炸。
    """
    return dt.replace(tzinfo=None) if dt.tzinfo is not None else dt


def _norm(text: str | None) -> str:
    """归一到「只剩小写字母数字」。

    handle 的分隔符是随意的(``zhang-san`` / ``zhang.san`` / ``zhangsan``
    是同一个人的三种写法),姓名里的空格与大小写同理。比对之前把这些差异
    全部抹掉,否则每一种写法都要单独写一条规则。
    """
    if not text:
        return ""
    t = unicodedata.normalize("NFKD", text)
    return re.sub(r"[^a-z0-9]", "", t.lower())


def _tokens(text: str | None) -> set[str]:
    """姓名切成词的集合。用集合而不是序列,是因为中文姓名的英文写法姓名
    顺序会颠倒(``Yajing Wang`` / ``Wang Yajing`` 是同一个人)。"""
    if not text:
        return set()
    raw = re.split(r"[\s·、,,/()()._-]+", text)
    return {t for t in (_norm(x) for x in raw) if len(t) >= 2}


def _email_local(email: str | None) -> str:
    return (email or "").split("@", 1)[0]


def _person_emails(p: IdentityPerson) -> list[str]:
    """这个人所有可用于比对的邮箱,小写去空。

    包含个人邮箱:GitHub 画像上公开的往往正是个人邮箱,而那恰恰是识别
    「这个 handle 是谁」最硬的一条证据。
    """
    out = []
    for v in (p.work_email, p.personal_email):
        v = (v or "").strip().lower()
        if v and "@" in v:
            out.append(v)
    return out


def external_signal(profile: GithubUserProfile | None) -> str | None:
    """这份画像有没有在说「这不是我们的员工」。返回一句中文理由或 ``None``。

    只在信号足够硬时才开口。``company`` 填了别家公司是硬信号,``company``
    为空**不是** —— 大多数人不填 company,把不填当成「外部」会把一半员工
    误标成外部人员,而那是一个人要花时间去撤销的错误。
    """
    if profile is None:
        return None
    if (profile.account_type or "").lower() == "bot":
        return "GitHub 账号类型是 Bot —— 机器账号,不对应某个员工"
    company = (profile.company or "").strip()
    if company:
        low = company.lower()
        if not any(m in low for m in _AFFILIATION_MARKERS):
            return f"GitHub 资料里 company 写的是「{company}」,不是本公司"
    return None


def affiliation_signal(profile: GithubUserProfile | None) -> str | None:
    """这份画像有没有在说「这大概是我们的人」。用于排序,不用于判定。"""
    if profile is None:
        return None
    blob = " ".join(
        x for x in (profile.company, profile.bio, profile.email) if x
    ).lower()
    for m in _AFFILIATION_MARKERS:
        if m in blob:
            return f"GitHub 资料里提到了「{m}」"
    return None


@dataclass
class Candidate:
    person_id: str
    person_label: str
    person_email: str | None
    department: str | None
    employment_status: str
    score: int
    #: 为什么它是候选 —— 界面上显示这个,人靠它判断,不是靠分数。
    reasons: list[str] = field(default_factory=list)


def score_candidates(
    handle: str,
    people: list[IdentityPerson],
    *,
    profile: GithubUserProfile | None = None,
    limit: int = 5,
) -> list[Candidate]:
    """打分。不调用任何模型。

    刻意**不**做模糊编辑距离:``zhangsan`` 与 ``zhangshan`` 差一个字母,
    但它们是两个人。字面命中给高分、部分命中给低分、拿不准交给下一层,
    比一个说不出理由的相似度更可用。
    """
    h = _norm(handle)
    p_email = (profile.email or "").strip().lower() if profile else ""
    p_email_local = _norm(_email_local(p_email))
    # 名字单独拿出来,不在下面反复写 ``profile.name`` —— 那些访问都在
    # 「p_name_norm 非空」的分支里(等价于 profile 非 None),但那个等价
    # 关系只有人看得出来。一个靠人脑推理才成立的守卫,下一次改动就会失效。
    p_name = profile.name if profile else None
    p_name_norm = _norm(p_name)
    p_name_tokens = _tokens(p_name)

    out: list[Candidate] = []
    for p in people:
        reasons: list[str] = []
        score = 0
        emails = _person_emails(p)

        # ── 画像:邮箱 ────────────────────────────────────────────────
        # 最硬的一条证据。GitHub 上公开的邮箱与员工邮箱相同,基本没有
        # 别的解释。
        if p_email and p_email in emails:
            score = max(score, SCORE_PROFILE_EMAIL_EXACT)
            reasons.append(f"GitHub 资料公开的邮箱就是本人邮箱({p_email})")
        elif p_email_local and any(
            _norm(_email_local(e)) == p_email_local for e in emails
        ):
            score = max(score, SCORE_PROFILE_EMAIL_LOCAL)
            reasons.append(
                f"GitHub 资料邮箱的前缀与本人邮箱前缀相同({p_email})",
            )

        # ── 画像:姓名 ────────────────────────────────────────────────
        dn_norm = _norm(p.display_name)
        dn_tokens = _tokens(p.display_name)
        if p_name_norm and dn_norm and p_name_norm == dn_norm:
            score = max(score, SCORE_PROFILE_NAME_EXACT)
            reasons.append(f"GitHub 资料姓名与本人姓名相同({p_name})")
        elif p_name_tokens and dn_tokens and p_name_tokens == dn_tokens:
            # 姓名顺序颠倒也算命中 —— 「Yajing Wang」和「Wang Yajing」
            # 是同一个人,而中英混排的显示名两种顺序都见得到。
            score = max(score, SCORE_PROFILE_NAME_TOKENS)
            reasons.append(
                f"GitHub 资料姓名与本人姓名词相同(顺序不同:{p_name})",
            )
        elif (
            p_name_tokens
            and dn_tokens
            and len(p_name_tokens & dn_tokens) >= 2
        ):
            score = max(score, SCORE_PROFILE_NAME_PARTIAL)
            reasons.append(
                "GitHub 资料姓名与本人姓名有多个词相同"
                f"({p_name})",
            )

        # ── handle 字面 ───────────────────────────────────────────────
        # 实测在生产数据上产出为 0,但它便宜、可解释、且在新数据上仍可能
        # 命中(新人多半用规范 handle),所以留着。
        if h:
            for e in emails:
                local = _norm(_email_local(e))
                if local and local == h:
                    score = max(score, SCORE_EMAIL_LOCAL_EXACT)
                    reasons.append(f"handle 与邮箱前缀完全相同({e})")
                    break
                if local and len(h) >= 4 and (h in local or local in h):
                    score = max(score, SCORE_EMAIL_LOCAL_CONTAINS)
                    reasons.append(f"handle 与邮箱前缀部分相同({e})")

            if dn_norm and dn_norm == h:
                score = max(score, SCORE_DISPLAY_NAME_COMPACT)
                reasons.append(f"handle 与显示名去空格后相同({p.display_name})")

            emp = _norm(p.employee_no)
            if emp and len(emp) >= 4 and emp in h:
                score = max(score, SCORE_EMPLOYEE_NO)
                reasons.append(f"handle 里含工号({p.employee_no})")

            for token in dn_tokens:
                if len(token) >= 4 and token in h:
                    score = max(score, SCORE_NAME_TOKEN)
                    reasons.append(f"姓名中的「{token}」出现在 handle 里")
                    break

        if score >= MIN_SCORE:
            out.append(
                Candidate(
                    person_id=str(p.id),
                    person_label=p.display_name or p.work_email or str(p.id),
                    person_email=p.work_email,
                    department=p.primary_department,
                    employment_status=p.employment_status,
                    score=score,
                    reasons=reasons,
                )
            )
    out.sort(key=lambda c: (-c.score, c.person_label))
    return out[:limit]


# ──────────────────────────────────────────────────────────────────────
# 画像富化
# ──────────────────────────────────────────────────────────────────────


def _profile_from_row(m: GithubOrgMembership) -> GithubUserProfile | None:
    raw = m.github_profile
    if not isinstance(raw, dict) or not raw:
        return None
    return GithubUserProfile(
        login=str(raw.get("login") or m.github_login),
        user_id=raw.get("user_id") if isinstance(raw.get("user_id"), int) else None,
        name=raw.get("name") or None,
        company=raw.get("company") or None,
        email=raw.get("email") or None,
        bio=raw.get("bio") or None,
        location=raw.get("location") or None,
        blog=raw.get("blog") or None,
        account_type=raw.get("account_type") or None,
    )


async def enrich_profiles(
    db: AsyncSession,
    rows_by_handle: dict[str, list[GithubOrgMembership]],
    *,
    fetch: ProfileFetcher | None,
    now: datetime | None = None,
    max_fetch: int = MAX_PROFILE_FETCH_PER_RUN,
) -> dict[str, GithubUserProfile | None]:
    """给每个 handle 备好画像:先用库里的缓存,过期或没有的现拉。

    拉到的写回**同一个 handle 的所有行**(一个人可能在多个组织里),这样
    第二次看的时候不用重拉,而且界面上五个组织显示的是同一份依据。

    ``fetch`` 为 ``None``(驱动没配好 / 拿不到凭证)时只用缓存 —— 那仍然
    比什么都没有好,而且不会让一个治理页面因为集成故障而整页报错。
    """
    now = _naive(now or _now())
    cutoff = now - PROFILE_TTL
    out: dict[str, GithubUserProfile | None] = {}
    stale: list[str] = []

    for handle_lower, rows in rows_by_handle.items():
        cached = None
        fetched_at = None
        for m in rows:
            got = _profile_from_row(m)
            if got is not None:
                cached = got
                fetched_at = m.profile_fetched_at
                break
        out[handle_lower] = cached
        if fetch is None:
            continue
        if cached is None or fetched_at is None or _naive(fetched_at) < cutoff:
            stale.append(handle_lower)

    if not fetch or not stale:
        return out

    fetched = 0
    for handle_lower in stale:
        if fetched >= max_fetch:
            break
        rows = rows_by_handle[handle_lower]
        login = rows[0].github_login
        try:
            profile = await fetch(login)
        except Exception as e:  # noqa: BLE001
            # 一个 handle 拉不到不影响其余。**不写 profile_fetched_at** ——
            # 否则一次限流会把这个 handle 冻在「已拉过、没内容」30 天。
            logger.warning(
                "github_org: profile fetch failed for %s: %s", login, e,
            )
            continue
        fetched += 1
        if profile is None:
            # GitHub 说没有这个人(改名或注销)。这本身是一条结论,值得
            # 记下来,否则每次都会重拉一个注定 404 的 handle。
            payload = {"login": login, "account_missing": True}
            out[handle_lower] = None
        else:
            payload = profile.as_dict()
            out[handle_lower] = profile
        for m in rows:
            m.github_profile = payload
            m.profile_fetched_at = now
    if fetched:
        await db.flush()
        logger.info("github_org: fetched %s GitHub profiles", fetched)
    return out


# ──────────────────────────────────────────────────────────────────────
# AI 复核
# ──────────────────────────────────────────────────────────────────────


async def _llm_client(db: AsyncSession):
    async def setting(key: str) -> str | None:
        return (
            await db.execute(
                select(PlatformSetting.value).where(PlatformSetting.key == key)
            )
        ).scalar_one_or_none()

    api_key = (await setting(_KEY_API) or "").strip()
    if not api_key:
        return None
    from app.domains.bailian.llm_client import BailianLLMClient  # noqa: PLC0415

    return BailianLLMClient(
        api_key=api_key,
        model=await setting(_KEY_MODEL),
        region=await setting(_KEY_REGION), purpose="github.handle_matching")


_SYSTEM = """你在帮一家公司(PingCAP,产品是 TiDB / TiKV)清理 GitHub 组织
成员台账。对每个 GitHub 用户名,你要回答两个问题之一:

A. 它对应给定候选人里的哪一个员工?
B. 它根本不是这家公司的员工(外部社区贡献者、其他公司的人、机器账号)?

这两个答案都有用。B 尤其有用 —— 这些组织是开源组织,里面本来就有大量
外部贡献者,把他们标记出来,清单才会收敛。

规则,一条都不能破:

1. 选人时**只能**从给定候选人里选,绝不能造一个不在列表里的 id。
2. 拿不准就两个都不选(person_id=null, external=false)。漏判的代价是
   「有人多看一眼」,错判的代价是「这条成员资格挂到错误的人头上,那个人
   离职时**别人**的访问权被撤掉」。两者不对等,所以宁可漏。
3. GitHub 资料里公开的邮箱等于候选人的邮箱 —— 这是最硬的证据。
4. 中文姓名的拼音写法有多种(zhangsan / zhang-san / sanzhang),姓名顺序
   也可能颠倒 —— 这些算命中。但只差一两个字母的不同名字(zhangsan vs
   zhangshan)算**不同的人**。
5. 判 external=true 需要正面证据:资料里 company 写着别家公司/基金会/
   高校,或者明显是机器账号(bot / ci / robot 结尾)。**资料为空不是
   证据** —— 大多数人什么都不填,把「没填」当成外部会造成大量误标。
6. 资料里出现 pingcap / tidb / tikv,说明这个人与我们相关,不要判
   external;但也不等于就是在职员工(可能是已离职的、或社区里的
   maintainer),认不出具体是谁就两个都不选。

对每一条输入返回:
{"handle": 原样, "person_id": 候选人 id 或 null, "external": true|false,
 "confidence": "high"|"medium"|"low", "reason": "一句中文说明依据"}
"""


def _render_batch(batch: list[dict[str, Any]]) -> str:
    lines: list[str] = []
    for it in batch:
        lines.append(f"\nGitHub 用户名: {it['handle']}")
        if it.get("org_slugs"):
            lines.append(f"  所在组织: {', '.join(it['org_slugs'])}")
        prof = it.get("profile") or {}
        if prof.get("account_missing"):
            lines.append("  GitHub 资料: 账号已不存在(改名或注销)")
        elif prof:
            bits = [
                f"{k}={prof[k]}"
                for k in ("name", "company", "email", "bio", "location", "account_type")
                if prof.get(k)
            ]
            lines.append(f"  GitHub 资料: {'; '.join(bits) if bits else '(全空)'}")
        else:
            lines.append("  GitHub 资料: (未取到)")
        if not it["candidates"]:
            lines.append("  候选人: (无 —— 字面比对没找到任何候选)")
            continue
        lines.append("  候选人:")
        for c in it["candidates"]:
            lines.append(
                f"    - id={c['person_id']} | {c['person_label']} | "
                f"{c.get('person_email')} | {c.get('department') or '—'} | "
                f"{c['employment_status']} | 依据: "
                f"{'; '.join(c['reasons']) or '—'}"
            )
    return "\n".join(lines)


def _batch_candidates(
    batch: list[dict[str, Any]], handle: str,
) -> list[dict[str, Any]]:
    for it in batch:
        if it["handle"].lower() == handle:
            return list(it["candidates"])
    return []


async def ai_review(
    db: AsyncSession, items: list[dict[str, Any]],
) -> dict[str, dict[str, Any]]:
    """让模型复核。返回 ``handle_lower -> 结论``。

    没配 LLM key 时返回空字典 —— 调用方照常给出确定性打分的结果。这不是
    降级路径的凑合:画像里的邮箱直接命中那一档是最高置信度的一档,而它
    不需要模型。
    """
    client = await _llm_client(db)
    if client is None or not items:
        return {}

    out: dict[str, dict[str, Any]] = {}
    for start in range(0, len(items), LLM_BATCH_SIZE):
        batch = items[start : start + LLM_BATCH_SIZE]
        try:
            reply = await client.chat_json(
                system=_SYSTEM,
                user=(
                    '请逐条判断,返回 {"results": [...]} 这一个对象。\n\n'
                    + _render_batch(batch)
                ),
                max_tokens=3000,
                temperature=0.1,
            )
        except Exception as e:  # noqa: BLE001
            # 一批失败不影响其余,也不影响确定性打分的结果 —— AI 是加分项,
            # 不是这条路的必要条件。
            logger.warning("github_org: AI handle review batch failed: %s", e)
            continue
        for r in (reply or {}).get("results") or []:
            handle = str(r.get("handle") or "").strip().lower()
            if not handle:
                continue
            pid = r.get("person_id")
            allowed = {c["person_id"] for c in _batch_candidates(batch, handle)}
            # 🔴 模型给了一个不在候选集里的 id —— 丢掉。它可能是幻觉,而一个
            # 幻觉的 person_id 会让界面上出现一个「AI 建议」按钮,点下去把
            # 成员资格挂到一个随机的人身上。
            if pid is not None and str(pid) not in allowed:
                logger.warning(
                    "github_org: AI returned person_id %s for %s which is not "
                    "in the candidate set — dropped",
                    pid, handle,
                )
                pid = None
            external = bool(r.get("external")) and pid is None
            out[handle] = {
                "person_id": (str(pid) if pid else None),
                "external": external,
                "confidence": str(r.get("confidence") or "low"),
                "reason": str(r.get("reason") or "")[:300],
            }
    return out


# ──────────────────────────────────────────────────────────────────────
# 对外入口
# ──────────────────────────────────────────────────────────────────────


async def suggest_links(
    db: AsyncSession,
    *,
    org_id: str | None = None,
    limit: int = 50,
    use_ai: bool = True,
    fetch_profile: ProfileFetcher | None = None,
) -> dict[str, Any]:
    """给「认不出是谁」的成员资格产出建议:关联到某人,或标记为外部人员。

    **只产出建议,永不自动写台账。** 见模块文档最后一节。
    """
    q = (
        select(GithubOrgMembership, GithubOrg)
        .join(GithubOrg, GithubOrg.id == GithubOrgMembership.org_id)
        .where(
            GithubOrgMembership.person_id.is_(None),
            # 已经被人确认是外部人员的不再问 —— 那正是这个队列能收敛的
            # 原因,重新把它们捞出来等于把人做过的判断丢掉。
            GithubOrgMembership.external_confirmed_at.is_(None),
            GithubOrgMembership.state.in_(const.MEMBERSHIP_LIVE_STATES),
            GithubOrg.deleted_at.is_(None),
        )
    )
    if org_id:
        q = q.where(GithubOrgMembership.org_id == str(org_id))
    rows = list(
        (await db.execute(q.order_by(GithubOrgMembership.github_login))).all()
    )

    # 同一个 handle 可能在多个组织里 —— 按 handle 聚合,一次判断覆盖所有
    # 组织的那几条,而不是让人对同一个人判断五次。
    by_handle: dict[str, dict[str, Any]] = {}
    rows_by_handle: dict[str, list[GithubOrgMembership]] = {}
    for m, org in rows:
        e = by_handle.setdefault(
            m.github_login_lower,
            {"handle": m.github_login, "membership_ids": [], "org_slugs": []},
        )
        e["membership_ids"].append(m.id)
        e["org_slugs"].append(org.slug)
        rows_by_handle.setdefault(m.github_login_lower, []).append(m)

    total = len(by_handle)
    if not total:
        return {
            "items": [], "ai_used": False, "total_unlinked_handles": 0,
            "profiles_available": 0,
        }

    selected = dict(list(by_handle.items())[:limit])
    profiles = await enrich_profiles(
        db,
        {k: rows_by_handle[k] for k in selected},
        fetch=fetch_profile,
    )

    people = list(
        (
            await db.execute(
                select(IdentityPerson).where(
                    IdentityPerson.deleted_at.is_(None),
                    # 已离职的人也要作为候选:一条挂到离职者身上的成员资格
                    # 正是治理待办要找的东西,把他们排除掉等于让最该被发现
                    # 的那批永远认不出来。
                    IdentityPerson.employment_status.in_(
                        ("active", "leaver_pending", "left"),
                    ),
                )
            )
        ).scalars().all()
    )

    items: list[dict[str, Any]] = []
    for key, e in selected.items():
        profile = profiles.get(key)
        cands = score_candidates(e["handle"], people, profile=profile)
        raw = rows_by_handle[key][0].github_profile
        items.append(
            {
                "handle": e["handle"],
                "membership_ids": e["membership_ids"],
                "org_slugs": sorted(set(e["org_slugs"])),
                "profile": raw if isinstance(raw, dict) else None,
                "external_hint": external_signal(profile),
                "affiliation_hint": affiliation_signal(profile),
                "candidates": [
                    {
                        "person_id": c.person_id,
                        "person_label": c.person_label,
                        "person_email": c.person_email,
                        "department": c.department,
                        "employment_status": c.employment_status,
                        "score": c.score,
                        "reasons": c.reasons,
                    }
                    for c in cands
                ],
            }
        )

    ai_map = await ai_review(db, items) if use_ai else {}
    for it in items:
        v = ai_map.get(it["handle"].lower())
        it["ai"] = v
        top = it["candidates"][0] if it["candidates"] else None
        if v and v.get("person_id"):
            it["suggested_person_id"] = v["person_id"]
            it["suggest_external"] = False
            it["suggestion_source"] = "ai"
        elif top and top["score"] >= AUTO_SUGGEST_MIN:
            # 确定性打分够硬时**不看 AI 的反对** —— 画像邮箱等于本人邮箱
            # 这种证据,比模型的一句「拿不准」可靠。
            it["suggested_person_id"] = top["person_id"]
            it["suggest_external"] = False
            it["suggestion_source"] = "deterministic"
        elif v and v.get("external"):
            it["suggested_person_id"] = None
            it["suggest_external"] = True
            it["suggestion_source"] = "ai"
        elif it["external_hint"] and not it["candidates"]:
            # 确定性的外部信号:company 写着别家公司,而且字面比对连一个
            # 候选都找不到。两个条件都要 —— 只有其中一个时留给人判断。
            it["suggested_person_id"] = None
            it["suggest_external"] = True
            it["suggestion_source"] = "deterministic"
        else:
            it["suggested_person_id"] = None
            it["suggest_external"] = False
            it["suggestion_source"] = None

    return {
        "items": items,
        "ai_used": bool(ai_map),
        "total_unlinked_handles": total,
        "profiles_available": sum(1 for p in profiles.values() if p is not None),
    }