"""Outbound half of the assistant channel — how a reply gets back to Feishu.

Reuses the one Feishu client the platform already has (``FeishuApprovalClient``
— named for its first caller, in practice the shared open-API client) and the
one credential row (``chat_approval``). No new app, no new secret, no new
env var: the same reuse decision the Feishu SSO flow made.

Sending is best-effort by contract. A turn that produced a good answer but
could not deliver it must not raise into the event path and trigger a Feishu
redelivery of a message we already processed — the failure is logged and
surfaced through the turn's audit row instead.
"""
from __future__ import annotations

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

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

from app.approval.feishu_client import FeishuApprovalClient
from app.assistant.reply_hygiene import FEISHU_MENTION
from app.integrations.compat import get_feishu_config

logger = logging.getLogger(__name__)

# Feishu rejects oversized text payloads; a model answer is capped long before
# this, so truncation here is a guardrail rather than a routine path.
_MAX_TEXT_CHARS = 4000
_TRUNCATION_SUFFIX = "…"


@dataclass(frozen=True, slots=True)
class ReplyTarget:
    """Where one reply goes: a person's private chat, or a room.

    ## Why this is a value object rather than two keyword arguments

    Every reply the intake path sends — the answer, 「我还在处理你上一条」, the
    rate-limit notice, the unsupported-message line — has to go to the SAME
    place the question came from. With ``open_id=`` and ``chat_id=`` as two
    optional kwargs, that correctness lives at a dozen individual call sites
    and one missed site posts a group answer into a private chat (the employee
    is left thinking the bot ignored the room) or, worse, posts a private
    answer into a room. Resolving the target ONCE at the top of the handler and
    passing the value down makes the wrong thing unspellable.

    ``reply_to_message_id`` turns the send into a Feishu *reply* under the
    triggering message. In a two-person room a plain post is fine and reads
    less cluttered; in a busy one it is impossible to tell which question an
    answer belongs to, so group targets carry it and DM targets do not.

    ``thread`` additionally collapses the exchange into a Feishu thread.
    """

    open_id: str = ""
    chat_id: str = ""
    reply_to_message_id: str | None = None
    thread: bool = False

    @classmethod
    def dm(cls, open_id: str) -> ReplyTarget:
        """One employee's private chat with the bot."""
        return cls(open_id=open_id)

    @classmethod
    def chat(
        cls,
        chat_id: str,
        *,
        reply_to_message_id: str | None = None,
        thread: bool = False,
    ) -> ReplyTarget:
        """A room. Everyone in it reads whatever is sent."""
        return cls(
            chat_id=chat_id,
            reply_to_message_id=reply_to_message_id,
            thread=thread,
        )

    @property
    def is_group(self) -> bool:
        return bool(self.chat_id)

    @property
    def is_addressable(self) -> bool:
        return bool(self.open_id or self.chat_id)

    def describe(self) -> str:
        """For logs. Never carries message content."""
        return f"chat:{self.chat_id}" if self.chat_id else f"dm:{self.open_id}"


async def build_client(db: AsyncSession) -> FeishuApprovalClient | None:
    """The shared Feishu client, or ``None`` when the integration is unset."""
    try:
        cfg = await get_feishu_config(session=db)
    except Exception as exc:  # noqa: BLE001 — integration absent / unreadable
        logger.warning("assistant: feishu config unavailable: %s", exc)
        return None
    app_id = str(cfg.get("app_id") or "")
    app_secret = str(cfg.get("app_secret") or "")
    if not app_id or not app_secret:
        logger.warning("assistant: feishu integration has no app_id/app_secret")
        return None
    return FeishuApprovalClient(app_id=app_id, app_secret=app_secret)


def clamp_text(text: str) -> str:
    if len(text) <= _MAX_TEXT_CHARS:
        return text
    return text[: _MAX_TEXT_CHARS - len(_TRUNCATION_SUFFIX)] + _TRUNCATION_SUFFIX


async def send_text(
    db: AsyncSession,
    *,
    to: ReplyTarget,
    text: str,
    idempotency_key: str | None = None,
) -> bool:
    """Deliver one text reply. Returns whether Feishu accepted it."""
    if not to.is_addressable or not text.strip():
        return False
    client = await build_client(db)
    if client is None:
        return False
    try:
        text = await render_feishu_mentions(db, text=text, client=client)
        body = clamp_text(text)
        if to.reply_to_message_id:
            await client.reply_message(
                message_id=to.reply_to_message_id,
                msg_type="text",
                content={"text": body},
                reply_in_thread=to.thread,
                idempotency_key=idempotency_key,
            )
        elif to.is_group:
            await client.send_text_to_chat(
                chat_id=to.chat_id,
                text=body,
                idempotency_key=idempotency_key,
            )
        else:
            await client.send_text_to_user(
                open_id=to.open_id,
                text=body,
                idempotency_key=idempotency_key,
            )
    except Exception as exc:  # noqa: BLE001 — delivery must not break intake
        logger.warning(
            "assistant: reply delivery failed to %s: %s", to.describe(), exc,
        )
        return False
    return True


#: 值得走富文本的最短回复。比这更短的基本是一句确认("好的,已记下"),
#: 套一层富文本没有任何收益。
_MIN_CHARS_FOR_POST = 20

#: markdown 标记 —— 一条回复里有没有值得保留的格式。没有就发纯文本:
#: 富文本消息在飞书里的气泡样式和纯文本一致,但多一次序列化和一层解析。
_MARKDOWN_HINTS = ("**", "](", "\n- ", "\n· ", "\n1. ", "`")


def wants_rich_text(text: str) -> bool:
    """这条回复里有没有「发成纯文本就会丢掉」的东西。

    🔴 **带 @ 提及的回复一律走纯文本。** 提及在飞书里是原生的 ``<at>`` 标签,
    而富文本的 ``md`` 元素不认它 —— 更糟的是 ``md`` 里的 ``<`` / ``>`` 必须转义
    (否则是内联 HTML 注入面),于是 ``<at user_id="ou_x">张三</at>`` 会被转成
    ``&lt;at …&gt;`` 原样显示给员工。纯文本消息**能**正确渲染 ``<at>``,所以
    有提及时退回那条路:少一点格式,好过一行可见的转义垃圾。
    (答案卡那一侧早就有同样的守卫 —— ``"[[helix_mention:" not in turn.reply``。)
    """
    if len(text.strip()) < _MIN_CHARS_FOR_POST:
        return False
    if FEISHU_MENTION.search(text):
        return False
    return any(hint in text for hint in _MARKDOWN_HINTS)


async def send_post(
    db: AsyncSession,
    *,
    to: ReplyTarget,
    markdown: str,
    idempotency_key: str | None = None,
) -> bool:
    """把模型写的 markdown 当富文本发出去,而不是先剥光再发。

    ## 为什么需要它

    在此之前,助手的普通回复走飞书 ``text`` 消息,而 ``to_feishu_text`` 会把模型
    写的加粗、链接、列表**全部剥掉** —— 因为纯文本消息确实渲染不了它们,留着
    标记比剥掉更难看。代价是助手的每一句话都是一坨没有层次的字,而它回答的常常
    是「你有 2 台电脑,分别是…」这种天生有结构的内容。

    ``post`` 消息的 ``md`` 元素能渲染 markdown 子集(实测:加粗、链接、列表、
    行内代码),于是格式可以原样送到。

    ## 转义与注入

    ``<`` / ``>`` 走 :func:`_escape_inline_html`,和答案卡同一条规矩:模型的
    上下文里有第三方文本(知识库正文、工单标题、工具结果),不能让其中一段
    渲染成内联 HTML。**链接**这一层与答案卡是同一个既有暴露面 —— 卡片路径早就
    在渲染模型写的 ``[文字](url)``,所以这里不引入新风险,只是不再比它更保守。

    返回是否被飞书接受;失败由调用方退回纯文本。
    """
    if not to.is_addressable or not markdown.strip():
        return False
    client = await build_client(db)
    if client is None:
        return False
    try:
        # 刻意**不**调 ``render_feishu_mentions``:它产出 ``<at>`` 标签,而下面
        # 那次转义(md 元素的注入面必须转义 ``<`` / ``>``)会把它变成可见的
        # ``&lt;at …&gt;``。带提及的回复由 :func:`wants_rich_text` 挡在门外,
        # 走纯文本那条能正确渲染 ``<at>`` 的路。
        body = _escape_inline_html(clamp_text(defang_foreign_links(markdown)))
        if to.reply_to_message_id:
            await client.reply_message(
                message_id=to.reply_to_message_id,
                msg_type="post",
                content={"zh_cn": {"title": "", "content": [
                    [{"tag": "md", "text": body}],
                ]}},
                reply_in_thread=to.thread,
                idempotency_key=idempotency_key,
            )
        elif to.is_group:
            await client.send_post_to_chat(
                chat_id=to.chat_id, markdown=body,
                idempotency_key=idempotency_key,
            )
        else:
            await client.send_post_to_user(
                open_id=to.open_id, markdown=body,
                idempotency_key=idempotency_key,
            )
    except Exception as exc:  # noqa: BLE001 — delivery must not break intake
        logger.warning(
            "assistant: rich-text reply delivery failed to %s: %s",
            to.describe(), exc,
        )
        return False
    return True


def _escape_inline_html(text: str) -> str:
    """飞书把 ``<`` / ``>`` 读作内联 HTML 起始 —— 同答案卡的 ``_safe_markdown``。"""
    return text.replace("<", "&lt;").replace(">", "&gt;")


#: ``[文字](url)`` —— 与 email_layout 的 ``_MD_LINK`` 同一个形状。
_MD_LINK = re.compile(r"\[([^\]\n]{1,120})\]\((https?://[^)\s]{1,400})\)")


def defang_foreign_links(text: str) -> str:
    """把**非白名单主机**的 markdown 链接拆成「文字:URL」,不让它变成活链接。

    ## 为什么

    ``post`` 的 ``md`` 元素会把 ``[文字](url)`` 渲染成真链接,而这条通道送出去的
    是**模型写的字**;模型的上下文里有第三方文本(知识库正文、工单标题、工具
    结果)。于是一篇被人写坏的知识库文档,可以让助手用**自己的口气**发出
    ``[点此重置密码](https://evil)`` —— 锚文本说一件事、目标是另一件事,这是最
    典型的钓鱼形状。

    仓库对这件事已经有明确立场:`email_layout` 给邮件渠道渲染 markdown 链接时
    **刻意用了主机白名单,并且写明「比飞书卡片更窄」**,理由正是「不能把卡片那个
    已知隐患原样引进第二个渠道」。富文本回复是**第三个**渠道,所以照同一条规矩办
    —— 并且直接复用那份白名单,而不是再抄一份会漂移的。

    白名单内的链接原样保留(它们本来就是我们要给员工的入口);白名单外的降级成
    「文字:URL」——**目标仍然可见**,读者自己能判断,只是锚文本再也不能撒谎。
    """
    from app.notifications.email_layout import (  # noqa: PLC0415 — cycle
        _anchor_allowed,
    )

    def _replace(match: re.Match[str]) -> str:
        label, url = match.group(1), match.group(2)
        if _anchor_allowed(url):
            return match.group(0)
        return f"{label}:{url}"

    return _MD_LINK.sub(_replace, text)


async def render_feishu_mentions(
    db: AsyncSession,
    *,
    text: str,
    client: FeishuApprovalClient,
) -> str:
    """Resolve grounded person markers to native Feishu ``<at>`` tags.

    Directory or Feishu lookup failures are best-effort: the visible employee
    name is retained and the rest of the answer still gets delivered.
    """
    matches = list(FEISHU_MENTION.finditer(text or ""))
    if not matches:
        return text
    from app.identity.person_models import IdentityPerson

    ids = {m.group("person_id") for m in matches}
    rows = (await db.execute(
        select(IdentityPerson.id, IdentityPerson.work_email)
        .where(IdentityPerson.id.in_(ids), IdentityPerson.deleted_at.is_(None))
    )).all()
    open_ids: dict[str, str] = {}
    for row in rows:
        if not row.work_email:
            continue
        try:
            open_ids[str(row.id)] = await client.resolve_open_id_by_email(row.work_email)
        except Exception as exc:  # noqa: BLE001 — mention is best-effort
            logger.info("assistant: person mention unresolved id=%s: %s", row.id, exc)

    def replace(match: re.Match[str]) -> str:
        open_id = open_ids.get(match.group("person_id"))
        label = match.group("label")
        return f'<at user_id="{open_id}">{label}</at>' if open_id else label

    return FEISHU_MENTION.sub(replace, text)


async def send_card(
    db: AsyncSession,
    *,
    to: ReplyTarget,
    card: dict[str, Any],
    idempotency_key: str | None = None,
) -> bool:
    """Deliver one interactive card. Returns whether Feishu accepted it."""
    if not to.is_addressable or not card:
        return False
    client = await build_client(db)
    if client is None:
        return False
    try:
        if to.reply_to_message_id:
            try:
                await client.reply_message(
                    message_id=to.reply_to_message_id,
                    msg_type="interactive",
                    content=card,
                    reply_in_thread=to.thread,
                    idempotency_key=idempotency_key,
                )
            except Exception as exc:  # noqa: BLE001 — 退回平铺,别丢掉这张卡
                # 话题回复是**排版**,卡片是**内容**。飞书对「在话题里回一张
                # 交互卡片」的支持不是我们能保证的一件事,而一次排版上的拒绝
                # 不该让员工什么都收不到 —— 那会表现成「按了按钮没反应」。
                logger.info(
                    "assistant: card reply refused (%s) — posting it plainly",
                    exc,
                )
                await client.send_card_to_chat(
                    chat_id=to.chat_id, card=card,
                    idempotency_key=idempotency_key,
                )
        elif to.is_group:
            await client.send_card_to_chat(
                chat_id=to.chat_id, card=card,
                idempotency_key=idempotency_key,
            )
        else:
            await client.send_card_to_user(
                open_id=to.open_id, card=card,
                idempotency_key=idempotency_key,
            )
    except Exception as exc:  # noqa: BLE001 — delivery must not break intake
        logger.warning(
            "assistant: card delivery failed to %s: %s", to.describe(), exc,
        )
        return False
    return True