"""AI assistance for the bounty console — Bailian (Qwen) via the platform
LLM key.

Two jobs, both strictly advisory:

* :func:`parse_report` turns a pasted researcher mail into structured
  fields (submitter address, title, affected target, vuln type, suggested
  severity, a Chinese summary the business owner can actually read). The
  operator reviews and edits before anything is persisted — the parser
  never writes.
* :func:`draft_reply_intro` writes a short, report-specific opening
  paragraph to sit in front of a template body, so a form letter reads
  like a human answered it.

Degradation contract: no platform LLM key → ``AppException`` 503 with
``BAILIAN_LLM_NO_KEY``, which the console surfaces as "AI 解析不可用,请
手动填写". Every field the parser fills is one a human can type, so the
whole feature works without a key.

The model NEVER decides severity, never picks a template and never
touches money. It proposes ``suggested_severity`` — the same advisory
field a business owner fills — and the security team rules.
"""
from __future__ import annotations

import json
import logging
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import AppException
from app.domains.bounty import constants as const
from app.integrations.models import PlatformSetting

logger = logging.getLogger(__name__)

_KEY_API = "ai_spending.llm.api_key"
_KEY_MODEL = "ai_spending.llm.model"
_KEY_REGION = "ai_spending.llm.region"

#: Hard cap on what we hand the model. Researcher mails routinely carry
#: base64 screenshots and quoted history; the useful signal is at the
#: top. Truncating keeps the prompt (and the bill) bounded.
_PROMPT_CHARS = 12_000

_EMAIL_BOUNDARY_TABLE = str.maketrans({
    char: " " for char in "<>()[]{}\"'`,;:!?\\|/"
})
_EMAIL_LOCAL_CHARS = frozenset(
    "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789._%+-"
)
_EMAIL_DOMAIN_CHARS = frozenset(
    "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789.-"
)

#: Addresses that are never the researcher's — our own domains, and the
#: no-reply senders of the mail infrastructure the report passed through.
_OWN_DOMAINS = ("pingcap.com", "pingcap.net")


async def _setting(db: AsyncSession, key: str) -> str | None:
    row = await db.get(PlatformSetting, key)
    value = row.value if row else None
    return value if isinstance(value, str) else None


async def _build_client(db: AsyncSession):
    from app.domains.bailian.llm_client import BailianLLMClient

    api_key = (await _setting(db, _KEY_API) or "").strip()
    if not api_key:
        raise AppException(
            code="BAILIAN_LLM_NO_KEY",
            message=(
                "未配置百炼(DashScope) API Key,AI 解析不可用。请在 管理 → "
                "连接器 → AI / LLM 配置后重试;在此之前可手动填写各字段。"
            ),
            status_code=503,
        )
    return BailianLLMClient(
        api_key=api_key,
        model=await _setting(db, _KEY_MODEL),
        region=await _setting(db, _KEY_REGION), purpose="bounty.triage")


_PARSE_SYSTEM = (
    "你是一名资深应用安全工程师,负责处理外部白帽子(安全研究者)通过邮件提交的"
    "漏洞报告。给定一封邮件的原文,请抽取结构化信息。要求:"
    "① 只抽取邮件中真实出现的信息,**绝不推测、绝不编造**,拿不准就留空(null);"
    "② title 用一句话概括漏洞(中文,不超过 60 字),形如「某站点存在反射型 XSS」;"
    "③ summary 是给**业务负责人**看的中文说明(3-6 句):这是什么问题、影响到哪个"
    "系统、攻击者能做什么、判断是否真实需要看什么;不要照抄英文原文;"
    "④ reproduction 抽取复现步骤/PoC 原文(保留原文语言与格式),没有就留空;"
    "⑤ researcher_email 必须是**提交者本人**的邮箱(通常是发件人 From),不要填"
    "我方域名(pingcap.com / pingcap.net)的地址;"
    "⑥ suggested_severity 只能取给定枚举之一,是**初步建议**不是结论,"
    "无法判断就留空;"
    "⑦ vuln_type 只能取给定枚举之一;"
    "⑧ affected_urls 是邮件中出现的受影响 URL 列表(最多 10 条);"
    "⑨ affected_system 是受影响的系统/域名/产品的简称(如 tidbcloud.com)。"
    "严格输出单个 JSON 对象。"
)


def _parse_prompt(raw_text: str, severities: list[str]) -> str:
    body = raw_text[:_PROMPT_CHARS]
    truncated = len(raw_text) > _PROMPT_CHARS
    return (
        "可选 vuln_type 枚举:" + ", ".join(const.VULN_TYPES) + "\n"
        "可选 suggested_severity 枚举:" + ", ".join(severities) + "\n\n"
        "邮件原文" + ("(已截断)" if truncated else "") + ":\n"
        "-----\n" + body + "\n-----\n\n"
        '请输出 JSON:{"title":"","summary":"","reproduction":null,'
        '"researcher_email":null,"researcher_name":null,'
        '"affected_system":null,"affected_urls":[],"vuln_type":null,'
        '"cve_id":null,"suggested_severity":null,'
        '"confidence":0}\n'
        "confidence 为 0-100 的整数,表示你对本次抽取整体可靠程度的判断。"
    )


def _fallback_email(raw_text: str) -> str | None:
    """First address in the text that is not one of ours.

    Runs regardless of the model's answer and is used when the model
    declined to give one — an intake form that guesses nothing forces the
    operator to hunt through the pasted mail for the address they can
    already see.
    """
    # Scan only the same bounded prefix sent to the model. Tokenization and
    # validation below are linear, avoiding backtracking on attacker supplied
    # report bodies while still handling ordinary mail punctuation.
    bounded = (raw_text or "")[:_PROMPT_CHARS]
    for token in bounded.translate(_EMAIL_BOUNDARY_TABLE).split():
        candidate = token.strip(".").lower()
        if candidate.count("@") != 1:
            continue
        local, domain = candidate.split("@", 1)
        if (
            not local
            or not domain
            or "." not in domain
            or domain.startswith(".")
            or domain.endswith(".")
            or any(char.isspace() for char in candidate)
            or any(char not in _EMAIL_LOCAL_CHARS for char in local)
            or any(char not in _EMAIL_DOMAIN_CHARS for char in domain)
            or len(domain.rsplit(".", 1)[-1]) < 2
            or not domain.rsplit(".", 1)[-1].isalpha()
        ):
            continue
        if any(candidate.endswith(f"@{d}") or candidate.endswith(f".{d}")
               for d in _OWN_DOMAINS):
            continue
        if candidate.startswith(("noreply@", "no-reply@", "mailer-daemon@")):
            continue
        return candidate
    return None


def _clean_str(value: Any, *, limit: int) -> str | None:
    if not isinstance(value, str):
        return None
    cleaned = value.strip()
    if not cleaned or cleaned.lower() in ("null", "none", "n/a", "unknown"):
        return None
    return cleaned[:limit]


async def parse_report(
    db: AsyncSession, *, raw_text: str, severity_keys: list[str],
) -> dict[str, Any]:
    """Extract intake fields from a pasted mail. Never persists."""
    client = await _build_client(db)
    payload: dict[str, Any]
    try:
        payload = await client.chat_json(
            system=_PARSE_SYSTEM,
            user=_parse_prompt(raw_text, severity_keys),
            max_tokens=2400,
            temperature=0.2,
        )
    except AppException:
        raise
    except Exception as exc:  # pragma: no cover - provider hiccup
        logger.warning("bounty ai parse failed: %s", exc)
        raise AppException(
            code="BOUNTY_AI_PARSE_FAILED",
            message="AI 解析失败,请手动填写字段(邮件原文已保留)。",
            status_code=502,
        ) from exc

    if not isinstance(payload, dict):
        payload = {}

    urls: list[str] = []
    raw_urls = payload.get("affected_urls")
    if isinstance(raw_urls, list):
        for item in raw_urls[:10]:
            cleaned = _clean_str(item, limit=2000)
            if cleaned:
                urls.append(cleaned)

    vuln_type = _clean_str(payload.get("vuln_type"), limit=32)
    if vuln_type not in const.VULN_TYPES:
        vuln_type = None

    suggested = _clean_str(payload.get("suggested_severity"), limit=24)
    if suggested not in severity_keys:
        suggested = None

    email = _clean_str(payload.get("researcher_email"), limit=320)
    if email:
        email = email.lower()
        if any(email.endswith(f"@{d}") for d in _OWN_DOMAINS):
            # The model picked our own security alias out of the headers.
            email = None
    if not email:
        email = _fallback_email(raw_text)

    raw_confidence = payload.get("confidence")
    try:
        confidence_int = max(0, min(100, int(str(raw_confidence).strip())))
    except (TypeError, ValueError):
        confidence_int = 0

    return {
        "title": _clean_str(payload.get("title"), limit=const.MAX_TITLE_CHARS),
        "summary": _clean_str(payload.get("summary"), limit=8000),
        "reproduction": _clean_str(payload.get("reproduction"), limit=20_000),
        "researcher_email": email,
        "researcher_name": _clean_str(payload.get("researcher_name"), limit=255),
        "affected_system": _clean_str(payload.get("affected_system"), limit=255),
        "affected_urls": urls,
        "vuln_type": vuln_type,
        "cve_id": _clean_str(payload.get("cve_id"), limit=64),
        "suggested_severity": suggested,
        "confidence": confidence_int,
        "truncated": len(raw_text) > _PROMPT_CHARS,
    }


_DRAFT_SYSTEM = (
    "你是代表公司安全团队回复外部白帽子的安全工程师。给定一条漏洞报告的上下文和"
    "一封模板回信,请写一段**放在模板正文最前面**的开场段落(英文,2-4 句),"
    "让这封回信读起来是人写的、针对这条具体报告的。要求:"
    "① 只使用给定上下文中的事实,不要提到金额、级别以外的内部信息,"
    "不要承诺任何未在上下文中出现的东西;"
    "② 语气专业、简洁、真诚,不要过度热情,不要用感叹号堆砌;"
    "③ 不要重复模板里已经写过的句子;"
    "④ 不要写称呼语(Dear ...)和签名 —— 模板里已经有了。"
    "严格输出单个 JSON 对象。"
)


async def draft_reply_intro(
    db: AsyncSession, *, report_context: dict[str, Any],
    template_name: str, template_body: str, instruction: str | None,
) -> dict[str, Any]:
    """Draft a report-specific opening paragraph for a template reply."""
    client = await _build_client(db)
    user = (
        "报告上下文(JSON):\n"
        + json.dumps(report_context, ensure_ascii=False, indent=2)
        + f"\n\n模板名称:{template_name}\n模板正文:\n-----\n"
        + template_body[:4000]
        + "\n-----\n"
        + (f"\n额外要求:{instruction}\n" if instruction else "")
        + '\n请输出 JSON:{"intro":"英文开场段落"}'
    )
    try:
        payload = await client.chat_json(
            system=_DRAFT_SYSTEM, user=user, max_tokens=800, temperature=0.5,
        )
    except AppException:
        raise
    except Exception as exc:  # pragma: no cover - provider hiccup
        logger.warning("bounty ai draft failed: %s", exc)
        raise AppException(
            code="BOUNTY_AI_DRAFT_FAILED",
            message="AI 起草失败,请直接使用模板正文。",
            status_code=502,
        ) from exc

    intro = _clean_str(
        (payload or {}).get("intro") if isinstance(payload, dict) else None,
        limit=2000,
    )
    if not intro:
        raise AppException(
            code="BOUNTY_AI_DRAFT_EMPTY",
            message="AI 未返回可用草稿,请直接使用模板正文。",
            status_code=502,
        )
    return {"intro": intro}


# ════════════════════════════════════════════════════════════════════
# Backfill assistance — grading and linking from the title alone
# ════════════════════════════════════════════════════════════════════
#
# Added 2026-08-25 for the 142 ungraded / 44 unlinked reports the FY27
# import left behind. It reads the TITLE, and only the title, because
# ``raw_report`` is empty on all 149 rows: the import deliberately kept
# researcher correspondence out of the database, so there is no body to
# read. Titles are unusually informative here ("Unauthenticated SQL
# Injection in TiDB Cloud Serverless", "Possible Subdomain Takeover
# (ai.pingcap.com)"), which is what makes this worth doing at all.
#
# It suggests. It never applies: the caller writes ``suggested_severity``
# and returns candidates, and a human adopts them. The failure mode being
# avoided is the HRIS one — a loose predicate silently created 26 accounts
# for system users because nothing asked a person first.

_BACKFILL_SYSTEM = (
    "You classify external vulnerability report TITLES for an internal "
    "security desk. You see only the title — no body exists. Be "
    "conservative: when a title is too vague to classify, say so rather "
    "than guessing. Reply with JSON only."
)


def _backfill_prompt(
    items: list[dict[str, str]],
    severities: list[str],
    hosts: list[str],
) -> str:
    lines = [
        "For each report below, using ONLY its title, return:",
        f'  severity: one of {severities}, or null when the title does not '
        "say enough to grade it.",
        f"  host: the affected hostname if the title names one, chosen from "
        f"{hosts} when it matches one of those, else the literal hostname, "
        "else null.",
        "  internal: true when this is our own hosted system; false when it "
        "is a finding in an open-source product (TiDB, TiKV, BR, PD, "
        "TiFlash), a public GitHub repository, or a CI workflow.",
        "  confidence: high | medium | low.",
        "",
        'Return {"items":[{"code":..,"severity":..,"host":..,'
        '"internal":..,"confidence":..}]} and nothing else.',
        "",
    ]
    for item in items:
        lines.append(f'  - code: {item["code"]}\n    title: {item["title"]}')
    return "\n".join(lines)


async def suggest_from_titles(
    db: AsyncSession,
    *,
    items: list[dict[str, str]],
    severity_keys: list[str],
    known_hosts: list[str],
) -> dict[str, dict[str, Any]]:
    """Title-only suggestions, keyed by report code. Never persists.

    Returns ``{}`` when the AI provider is not configured or the call
    fails — the caller must degrade to "no suggestion" rather than to a
    guess, because the whole point is that a human adopts these.
    """
    if not items:
        return {}
    try:
        client = await _build_client(db)
    except Exception:  # noqa: BLE001 - provider not configured is normal
        logger.info("bounty.ai: no provider configured for title suggestions")
        return {}

    try:
        payload = await client.chat_json(
            system=_BACKFILL_SYSTEM,
            user=_backfill_prompt(items, severity_keys, known_hosts),
            max_tokens=2400,
            temperature=0.1,
        )
    except Exception:  # noqa: BLE001
        logger.warning("bounty.ai: title suggestion failed", exc_info=True)
        return {}

    out: dict[str, dict[str, Any]] = {}
    for row in (payload or {}).get("items") or []:
        if not isinstance(row, dict):
            continue
        code = _clean_str(row.get("code"), limit=32)
        if not code:
            continue
        severity = _clean_str(row.get("severity"), limit=32)
        out[code] = {
            "severity": severity if severity in severity_keys else None,
            "host": _clean_str(row.get("host"), limit=255),
            # Default to "internal" being unknown rather than true: marking
            # a report as "no internal system applies" removes it from the
            # gap for good, so that claim needs the model to be explicit.
            "internal": (
                bool(row.get("internal"))
                if isinstance(row.get("internal"), bool) else None
            ),
            "confidence": _clean_str(row.get("confidence"), limit=16),
        }
    return out


#: The verdicts the model may return. Anything else is coerced to
#: ``needs_info`` — an unrecognised verdict must not read as a clean bill of
#: health, and "we could not tell" is the honest fallback.
ASSESS_VERDICTS: tuple[str, ...] = (
    "likely_valid",       # looks like a real finding as described
    "needs_info",         # not decidable as written
    "likely_invalid",     # described behaviour is not a vulnerability
    "out_of_scope",       # real, but not ours / not in programme scope
)

#: The model writes in the reader's language. Hardcoding Chinese put
#: Chinese prose under English headings for anyone whose console is in
#: English — half-translated is worse than either language on its own.
_ASSESS_LANGUAGE: dict[str, str] = {
    "zh": "用简体中文作答。",
    "en": "Answer in English.",
    "ja": "日本語で回答してください。",
}


def _assess_language(lang: str | None) -> str:
    code = (lang or "zh").split("-")[0].lower()
    return _ASSESS_LANGUAGE.get(code, _ASSESS_LANGUAGE["zh"])


_ASSESS_SYSTEM = (
    "你是一名资深应用安全工程师,负责对外部白帽子提交的漏洞报告做**初筛**。"
    "你的读者是安全值班同学:他要在 1 分钟内判断这份报告值不值得投入时间,"
    "以及下一步该做什么。要求:"
    "① **绝不编造**。只根据报告原文判断;原文没写的前提、版本、影响一律不要补。"
    "拿不准就用 needs_info,并把缺什么写进 questions。"
    "② verdict 只能取给定枚举之一。likely_valid = 按原文描述这确实是个漏洞;"
    "needs_info = 描述不足以判断(缺复现、缺前提、缺影响);"
    "likely_invalid = 描述的行为本身不构成漏洞(例如需要已经拿到管理员权限、"
    "或属于设计如此);out_of_scope = 是真问题但不属于我们的资产/项目范围。"
    "③ verdict_reason 用中文一到三句说**为什么**,要引用报告里的具体依据,"
    "不要复述结论。"
    "④ exploitability 用中文说清**攻击者需要什么前提**(是否需要登录、是否需要"
    "受害者交互、是否需要同租户……)—— 这是判断严重程度的关键。"
    "⑤ impact 用中文说清**最坏情况下能拿到什么/做到什么**,不要写「可能有风险」"
    "这类空话。"
    "⑥ what_to_check 是 2-4 条中文祈使句,是值班同学**为了确认这份报告要做的"
    "具体动作**(例如「用两个不同租户的账号调用 /api/x 看能否读到对方数据」),"
    "不是泛泛的「复现一下」。"
    "⑦ questions 是需要向研究者追问的点(中文),没有就给空数组。"
    "⑧ suggested_severity / vuln_type 只能取给定枚举之一,是**建议**不是结论,"
    "判断不了就留空。"
    "⑨ confidence 取 high / medium / low,反映你对 verdict 的确信程度。"
    "严格输出单个 JSON 对象。"
)


def _assess_prompt(
    *, title: str, raw_text: str, severities: list[str],
    vuln_types: list[str], lang: str | None = None,
) -> str:
    return (
        f"{_assess_language(lang)}\n\n"
        f"报告标题:{title or '(无)'}\n\n"
        f"可选严重级别:{', '.join(severities) or '(无)'}\n"
        f"可选漏洞类型:{', '.join(vuln_types) or '(无)'}\n"
        f"verdict 只能取:{', '.join(ASSESS_VERDICTS)}\n\n"
        "报告原文如下(可能是英文,可能包含引用的往来):\n"
        "-----\n"
        f"{raw_text[:24000]}\n"
        "-----\n\n"
        "输出 JSON,字段:verdict, verdict_reason, confidence, "
        "exploitability, impact, what_to_check (数组), questions (数组), "
        "suggested_severity, vuln_type, affected_urls (数组), reproduction"
    )


async def assess_report(
    db: AsyncSession,
    *,
    title: str,
    raw_text: str,
    severity_keys: list[str],
    vuln_types: list[str],
    lang: str | None = None,
) -> dict[str, Any] | None:
    """Triage one report: is this actually a problem, and what next?

    Returns None when there is nothing to read or the model is unavailable
    — a missing assessment is honest, and a fabricated one on a security
    finding is worse than none. Never raises for the caller's benefit: this
    runs in bulk over 149 reports and one bad response must not stop the
    rest.
    """
    body = (raw_text or "").strip()
    if len(body) < 40:
        # Nothing to assess. The FY27 rows that carry only a title are
        # exactly this case, and guessing from a title is how a critical
        # finding gets called informational.
        return None

    try:
        client = await _build_client(db)
    except AppException:
        return None

    try:
        payload = await client.chat_json(
            system=_ASSESS_SYSTEM,
            user=_assess_prompt(
                title=title, raw_text=body,
                severities=severity_keys, vuln_types=vuln_types, lang=lang,
            ),
            max_tokens=2200,
        )
    except Exception:  # noqa: BLE001 — one bad reply must not stop a batch
        logger.warning("bounty.ai: assessment failed", exc_info=True)
        return None
    if not isinstance(payload, dict):
        return None

    verdict = _clean_str(payload.get("verdict"), limit=32)
    if verdict not in ASSESS_VERDICTS:
        # An unrecognised verdict must not read as a clean bill of health.
        verdict = "needs_info"

    def _list(key: str, *, limit: int, cap: int) -> list[str]:
        raw = payload.get(key)
        if not isinstance(raw, list):
            return []
        out = []
        for item in raw[:cap]:
            cleaned = _clean_str(item, limit=limit)
            if cleaned:
                out.append(cleaned)
        return out

    severity = _clean_str(payload.get("suggested_severity"), limit=32)
    vuln_type = _clean_str(payload.get("vuln_type"), limit=64)
    return {
        "verdict": verdict,
        "verdict_reason": _clean_str(payload.get("verdict_reason"), limit=800),
        "confidence": (
            _clean_str(payload.get("confidence"), limit=16)
            if _clean_str(payload.get("confidence"), limit=16)
            in ("high", "medium", "low") else None
        ),
        "exploitability": _clean_str(payload.get("exploitability"), limit=800),
        "impact": _clean_str(payload.get("impact"), limit=800),
        "what_to_check": _list("what_to_check", limit=300, cap=4),
        "questions": _list("questions", limit=300, cap=4),
        "suggested_severity": (
            severity if severity in severity_keys else None
        ),
        "vuln_type": vuln_type if vuln_type in vuln_types else None,
        "affected_urls": _list("affected_urls", limit=500, cap=10),
        "reproduction": _clean_str(payload.get("reproduction"), limit=6000),
        "model": getattr(client, "model", None),
        # Recorded so the console can say "this reading is in Chinese"
        # rather than silently mixing languages after a locale switch.
        "lang": (lang or "zh").split("-")[0].lower(),
    }