"""「380838443567 这个云账号谁负责」—— 编号先落到行上,再交给模型。

## 这是在解决什么

生产实测(2026-09-07):员工问「380838443567 这个云账号谁负责」。那个号码是
``cmdb_cloud_scopes`` 里一行真实的 **AWS** 账号(`pingcap-dbaas`,prod,53 名成员),
状态 active、有负责人。助手的回答是
**另一个问题的答案** —— 它列出了「开通云账号(项目)申请」和「AWS/GCP/Azure/
Alibaba Production SSO Permission Application」两条**审批流程**的负责人,并注明
「平台根据审批记录推断」。

链路是这样断的:「谁负责」命中 :mod:`app.assistant.flow_grounding` 的问句模式,
于是 60 条审批流程被放进了这一轮的 prompt。模型看到一份现成的、名字里带「云账号」
的清单,就从里面挑了两条 —— 它压根没去查那个号码。#3512 修的是另一种形状(模型
调了 ``list_cloud_scopes`` 但不传 ``search``,然后声称「没有找到」);这一次它连
工具都没调。

## 为什么是预检索,而不是给模型加一句提示

和 ``flow_grounding`` 的开头写的是同一条道理:**system prompt 是请求,不是保证**。
「问题里出现编号时请先按编号查」这句话,模型可以照做,也可以像这次一样,在手边已经
有一份看起来很相关的清单时不照做。而编号是一种**判得准**的东西 —— 服务端自己抽出
来、自己查掉、把行摆进 prompt,模型就没有「挑一个错答案」的余地了:对的那一行已经
在它眼前。

反过来说,一个编号一行都没查到时,这里**什么都不加**(``None``),行为与改动前
完全一致 —— 该走审批流程接地、该给转人工/工单,都照旧。这个模块只在「确实拿到了
权威行」时才说话。

## RBAC 不写第二遍

查询一律通过 :func:`app.assistant.mcp_bridge.call_readonly_tool` 以**提问者本人**
的身份跑真实的只读 MCP 工具,而不是在这里写 SQL。每道门都已经在工具体里:
``list_cloud_scopes`` 对非云管理员只返回「自己拥有或参与」的 scope,``search_assets``
按资产域的范围过滤。抄一遍谓词就是给自己开一条越权读的路 —— 而且它会静默地对。

所以这里也**没有**「行存在但你看不见」这种提示:那本身就是在泄露存在性。看不见就
等于查不到,和改动前一样交给转人工。
"""
from __future__ import annotations

import json
import logging
import re
from dataclasses import dataclass
from typing import Any

logger = logging.getLogger(__name__)

#: 一条消息里最多认几个编号。粘一串编号进来是真实场景(「这几台机器谁的」),
#: 但每个编号都要跑真实工具,所以给一个明确的上界。
MAX_IDENTIFIERS = 3

#: 每个编号最多试几个工具、每个工具最多取几行。
MAX_TOOLS_PER_IDENTIFIER = 2
MAX_ROWS_PER_TOOL = 3

#: 整段接地文字的字符预算。行是完整放进去或者完全不放 —— 截一半的 JSON 行
#: 比不放更糟:模型会把被截断的字段当成真值读。
CHAR_BUDGET = 4200

KIND_CLOUD_ACCOUNT = "cloud_account"
KIND_CLOUD_RESOURCE = "cloud_resource"
KIND_IP = "ip"
KIND_DOMAIN = "domain"
KIND_SERIAL = "serial"

#: 每一类编号该问哪些只读工具,以及收窄参数叫什么(``search`` / ``q`` 两种拼法
#: 在注册表里都有,这里按工具逐个写死,不靠猜)。
#:
#: 顺序即优先级:先问最可能命中的那个。
TOOLS_BY_KIND: dict[str, tuple[tuple[str, str], ...]] = {
    # 云账号 / 订阅 / 项目:``list_cloud_scopes.search`` 同时匹配 display_name
    # 和 external_id,而 external_id 正是账号号码。
    KIND_CLOUD_ACCOUNT: (("list_cloud_scopes", "search"),),
    KIND_CLOUD_RESOURCE: (
        ("list_cloud_servers", "search"),
        ("cloud_resource_directory_list", "q"),
    ),
    KIND_IP: (("list_cloud_servers", "search"), ("list_dns_records", "search")),
    KIND_DOMAIN: (
        ("cloud_resource_directory_list", "q"),
        ("list_certificates", "search"),
    ),
    KIND_SERIAL: (("search_assets", "search"), ("list_mdm_devices", "q")),
}

#: 邮箱先整个抽掉,否则 ``@`` 后面的域名会被当成一个「员工问的域名」,而
#: 「zhang.san@pingcap.com 是谁」问的是人,不是 pingcap.com 这个域。
_EMAIL = re.compile(r"[\w.+-]+@[\w-]+(?:\.[\w-]+)+")

#: Azure 订阅 id / GCP 项目 uuid 的形状。
_UUID = re.compile(
    r"\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b",
    re.IGNORECASE,
)

#: 云上资源 id:``i-…`` / ``vpc-…`` / ``sg-…``。前缀白名单而不是「小写字母-数字」
#: 泛匹配 —— 后者会把「helix-3512」这样的普通词也认成资源 id。
_CLOUD_RESOURCE_ID = re.compile(
    r"\b(?:i|ins|instance|vpc|vsw|subnet|sg|eni|eip|vol|d|disk|lb|slb|clb|nlb|alb"
    r"|nat|snap|snapshot|ami|img|rds|redis|mongo|kafka|oss|cos|acl|rtb|igw)"
    r"-[0-9a-z]{6,}\b",
    re.IGNORECASE,
)

_IPV4 = re.compile(
    r"\b(?:(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)\.){3}"
    r"(?:25[0-5]|2[0-4]\d|1\d\d|[1-9]?\d)\b",
)

#: 域名。TLD 至少两个字母,所以 ``1.2.3.4`` 不会被当成域名(它先被 IPv4 认走了,
#: 这里是第二道)。
_DOMAIN = re.compile(
    r"\b(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,}\b", re.IGNORECASE,
)

#: 纯数字账号号码:AWS 12 位、阿里云 16 位、腾讯云 uin 9–11 位。下界 9 位是为了
#: 把日期(20260907,8 位)和小额数字挡在外面。
_DIGITS = re.compile(r"(?<![\w-])\d{9,24}(?![\w-])")

#: 资产序列号 / 设备编号:大写字母加数字混排。Mac 序列号(C02XY1234ABC)、
#: 资产标签(PC-2024-0157)都是这个形状。必须同时有字母和数字,否则就是纯数字,
#: 那一类由 ``_DIGITS`` 认走。
_SERIAL = re.compile(
    r"(?<![\w-])(?=[A-Z0-9-]{8,24}(?![\w-]))(?=[A-Z0-9-]*[A-Z])"
    r"(?=[A-Z0-9-]*\d)[A-Z0-9-]{8,24}(?![\w-])",
)

#: 抽取顺序 = 认领顺序。先具体后宽泛:一个 token 被前面的规则认走之后,后面的
#: 规则不再看它(否则 ``i-0abc123def`` 里的 ``0abc123def`` 会再被序列号认一遍,
#: 同一个编号查两遍不同的域)。
_EXTRACTORS: tuple[tuple[re.Pattern[str], str], ...] = (
    (_UUID, KIND_CLOUD_ACCOUNT),
    (_CLOUD_RESOURCE_ID, KIND_CLOUD_RESOURCE),
    (_IPV4, KIND_IP),
    (_DOMAIN, KIND_DOMAIN),
    (_DIGITS, KIND_CLOUD_ACCOUNT),
    (_SERIAL, KIND_SERIAL),
)


@dataclass(frozen=True)
class Identifier:
    """问题里的一个编号,以及它该去哪一类查。"""

    value: str
    kind: str


@dataclass(frozen=True)
class IdentifierGrounding:
    """一轮里,编号查到的权威行。"""

    #: 追加到 system prompt 的那一段。
    system_context: str
    #: ``[{identifier, kind, tool, rows}]`` —— 供评测取证和管理面回看。
    hits: list[dict[str, Any]]


#: 只扫问句的前这么多字符。飞书那一侧本来就截到 2000
#: (``feishu_events._MAX_QUESTION_CHARS``),这里给网页侧也画一条线:域名那条
#: 正则在「a.a.a.a…」这种病态串上是超线性的(实测 4000 字符 117ms),而粘一整段
#: 日志进来是真实场景。上界比飞书那条宽一倍,所以对正常问句没有任何影响。
_SCAN_CHARS = 4000


def extract_identifiers(question: str) -> list[Identifier]:
    """从一句话里抽出编号,按出现顺序,最多 :data:`MAX_IDENTIFIERS` 个。"""
    text = _EMAIL.sub(" ", (question or "")[:_SCAN_CHARS])
    if not text.strip():
        return []

    claimed: list[tuple[int, int]] = []
    found: list[tuple[int, Identifier]] = []
    seen: set[str] = set()

    def _overlaps(start: int, end: int) -> bool:
        return any(start < c_end and c_start < end for c_start, c_end in claimed)

    for pattern, kind in _EXTRACTORS:
        for match in pattern.finditer(text):
            start, end = match.span()
            if _overlaps(start, end):
                continue
            value = match.group(0)
            key = value.lower()
            if key in seen:
                # 同一个编号在一句话里出现两次,只查一次。
                claimed.append((start, end))
                continue
            seen.add(key)
            claimed.append((start, end))
            found.append((start, Identifier(value=value, kind=kind)))

    found.sort(key=lambda pair: pair[0])
    return [identifier for _position, identifier in found[:MAX_IDENTIFIERS]]


#: 装「行」的键名,按优先级。形状匹配(第一个元素是对象的列表)是兜底而不是
#: 首选:一个工具的返回里可能同时有 ``facets`` / ``filters`` 这类同样长得像行的
#: 列表,而挑错了就是把无关的东西当成权威行喂进 prompt。
_ROW_KEYS = (
    "items", "rows", "data", "results",
    "scopes", "servers", "assets", "devices", "records", "certificates",
    "resources",
)


def _rows_from(payload: str) -> tuple[list[dict[str, Any]], int]:
    """从一个工具的原始 JSON 里取出「行」,以及**一共有几行**。

    返回 ``(展开的行, 命中总数)``。总数单独返回是刻意的:只给前几行、却不说
    「一共几行」,正是这个模块在修的那一类缺陷 —— 一份被裁过的清单读起来和一份
    完整清单一模一样,而模型会据此回答「就这一个」。

    错误 / 超限 / 零行都返回空 —— 对这个模块来说它们是同一件事:没有拿到权威
    行,那就什么都不说。特别是超限:带了收窄条件之后还超时,桥层给的是
    ``partial_result`` 字符串而不是行,把半截 JSON 当事实读比不接地更糟。
    """
    try:
        data = json.loads(payload)
    except (TypeError, ValueError):  # pragma: no cover — tools return JSON
        return [], 0
    if not isinstance(data, dict):
        return [], 0
    if data.get("error") or data.get("truncated"):
        return [], 0

    rows: list[dict[str, Any]] | None = None
    for key in _ROW_KEYS:
        value = data.get(key)
        if isinstance(value, list) and value and all(
            isinstance(item, dict) for item in value
        ):
            rows = value
            break
    if rows is None:
        for value in data.values():
            if isinstance(value, list) and value and all(
                isinstance(item, dict) for item in value
            ):
                rows = value
                break
    if not rows:
        return [], 0

    # 工具报的 ``total`` 是全库命中数(可能大于这一页);没有这个字段时,这一页
    # 的长度就是我们知道的全部。取大的那个 —— 少报会让模型以为「就这些」。
    reported = data.get("total")
    total = max(
        len(rows), reported if isinstance(reported, int) and reported >= 0 else 0,
    )
    return rows[:MAX_ROWS_PER_TOOL], total


async def build_identifier_grounding(
    question: str, *, person: Any,
) -> IdentifierGrounding | None:
    """把问题里的编号查成行。没有一个编号查到东西就返回 ``None``。

    永不抛异常:接地失败的代价必须是「少一段证据」,不能是「这一轮挂掉」。
    """
    from app.assistant import mcp_bridge  # noqa: PLC0415 — cycle

    identifiers = extract_identifiers(question)
    if not identifiers:
        return None

    hits: list[dict[str, Any]] = []
    for identifier in identifiers:
        for tool_name, param in TOOLS_BY_KIND.get(identifier.kind, ())[
            :MAX_TOOLS_PER_IDENTIFIER
        ]:
            if not mcp_bridge.is_callable_tool(tool_name):
                # 工具被改名 / 变成写工具而从只读目录里消失 —— 不该让接地崩,
                # 也不该悄悄绕过桥层的允许清单。
                logger.warning(
                    "identifier grounding: %s is no longer read-only-callable",
                    tool_name,
                )
                continue
            try:
                raw = await mcp_bridge.call_readonly_tool(
                    tool_name,
                    {param: identifier.value, "page_size": 20},
                    person=person,
                )
            except Exception:  # noqa: BLE001 — 一个工具出问题只损失它自己
                logger.exception(
                    "identifier grounding: %s failed for %r",
                    tool_name, identifier.value,
                )
                continue
            rows, total = _rows_from(raw)
            if not rows:
                continue
            hits.append({
                "identifier": identifier.value,
                "kind": identifier.kind,
                "tool": tool_name,
                "rows": rows,
                "total": total,
            })
            # 命中就停:第二个工具是备选路径,不是补充数据。
            break

    if not hits:
        return None
    return IdentifierGrounding(
        system_context=_render(hits), hits=hits,
    )


def _render(hits: list[dict[str, Any]]) -> str:
    """把命中的行写成 prompt 段落,并且**说清楚这是权威数据**。

    那句「不要改用审批流程的负责人来回答」是照着生产上那次错答写的:模型手里
    同时有这段行和 60 条审批流程时,必须知道哪一份才是这个问题的答案。
    """
    lines = [
        "\n\n## 提问里的编号 —— 平台已按提问者的身份查过(权威数据)\n",
        "员工的问题里出现了下面这些编号。服务端已经以**他本人的权限**查了对应的"
        "台账,查到的行原样附在下面:\n\n",
    ]
    budget = CHAR_BUDGET
    for hit in hits:
        head = (
            f"### `{hit['identifier']}` — 查自 `{hit['tool']}`\n"
        )
        lines.append(head)
        budget -= len(head)
        kept = 0
        for row in hit["rows"]:
            blob = json.dumps(row, ensure_ascii=False, default=str)
            if len(blob) + 1 > budget:
                break
            lines.append(blob + "\n")
            budget -= len(blob) + 1
            kept += 1
        # 「一共几行」必须说,哪怕只有一行 —— 不说等于让模型把一份被裁过的
        # 清单当成完整清单读,而那正是这个模块在修的缺陷本身。
        total = int(hit.get("total") or kept)
        if total > kept:
            note = (
                f"(这个编号一共命中 {total} 行,以上展开了 {kept} 行;"
                f"要看其余的用 `{hit['tool']}` 带同样的收窄条件再查。"
                f"**不要**据此断言「只有这一个」)\n"
            )
            lines.append(note)
            budget -= len(note)
    lines.append(
        "\n**这几行就是这些编号的答案** —— 负责人、名称、状态、环境都在行里,"
        "请直接据此回答,并且照实说清楚是哪一个账号 / 哪一台机器 / 哪一台设备。\n"
        "不要用审批流程的负责人、也不要用别的域里名字相近的东西来代替回答:"
        "「谁负责这个账号」问的是这一行里的 owner,不是「开通这类账号该走哪条流程」"
        "的流程负责人。\n",
    )
    return "".join(lines)