# -*- coding: utf-8 -*-
"""DingTalk Stream callback handler: message -> native dict -> reply."""

from __future__ import annotations

import asyncio
import logging
from typing import Any, Callable, Dict, List, Optional

import dingtalk_stream
from dingtalk_stream import CallbackMessage, ChatbotMessage
from qwenpaw.schemas import (
    TextContent,
)

from ..base import ContentType

from .content_utils import (
    conversation_id_from_chatbot_message,
    conversation_type_from_chatbot_message,
    dingtalk_content_from_type,
    get_type_mapping,
    sender_from_chatbot_message,
    session_param_from_webhook_url,
)

logger = logging.getLogger(__name__)

# Download filename hint by type (e.g. voice -> .amr).
FILENAME_HINT_BY_MAPPED = {
    "audio": "audio.amr",
    "image": "image.png",
    "video": "video.mp4",
}
DEFAULT_FILENAME_HINT = "file.bin"


class DingTalkChannelHandler(dingtalk_stream.ChatbotHandler):
    """Internal handler: convert DingTalk message to native dict, enqueue via
    manager (thread-safe), ACK immediately."""

    def __init__(
        self,
        main_loop: asyncio.AbstractEventLoop,
        enqueue_callback: Optional[Callable[[Any], None]],
        bot_prefix: str,
        download_url_fetcher,
        try_accept_message: Optional[Callable[[str], bool]] = None,
        require_mention: bool = False,
    ):
        super().__init__()
        self._main_loop = main_loop
        self._enqueue_callback = enqueue_callback
        self._bot_prefix = bot_prefix
        self._download_url_fetcher = download_url_fetcher
        self._try_accept_message = try_accept_message
        self._require_mention = require_mention

    def _emit_native_threadsafe(self, native: dict) -> None:
        if self._enqueue_callback:
            self._main_loop.call_soon_threadsafe(
                self._enqueue_callback,
                native,
            )

    def _fetch_download_url_and_content(
        self,
        download_code: str,
        robot_code: str,
        mapped: str,
        filename_hint: Optional[str] = None,
    ) -> Optional[Any]:
        """Fetch media by download_code; return Content to append or None."""
        hint = (filename_hint or "").strip() or FILENAME_HINT_BY_MAPPED.get(
            mapped,
            DEFAULT_FILENAME_HINT,
        )
        try:
            fut = asyncio.run_coroutine_threadsafe(
                self._download_url_fetcher(
                    download_code=download_code,
                    robot_code=robot_code,
                    filename_hint=hint,
                ),
                self._main_loop,
            )
            download_url = fut.result(timeout=15)
            return dingtalk_content_from_type(mapped, download_url)
        except Exception:
            return None

    @staticmethod
    def _extract_filename_hint(payload: Dict[str, Any]) -> Optional[str]:
        """Extract filename hint from DingTalk payload variants."""
        for key in ("fileName", "file_name", "filename", "name", "title"):
            val = payload.get(key)
            if isinstance(val, str) and val.strip():
                return val.strip()
        return None

    def _resolve_single_download(
        self,
        content_dict: Dict[str, Any],
        msg_dict: Dict[str, Any],
        robot_code: Optional[str],
    ) -> Optional[Any]:
        """Resolve a single downloadCode into a Content object.

        For voice messages (``mapped == "audio"``), prefer the built-in
        ``recognition`` text from DingTalk over downloading the AMR file.
        """
        dl_code = content_dict.get("downloadCode") or content_dict.get(
            "download_code",
        )
        if not dl_code or not robot_code:
            return None

        type_mapping = get_type_mapping()
        msgtype = (msg_dict.get("msgtype") or "").lower().strip()
        mapped = type_mapping.get(msgtype, msgtype or "file")
        if mapped not in ("image", "file", "video", "audio"):
            mapped = "file"

        # Voice messages from DingTalk include a ``recognition`` field
        # with the transcribed text.  Use it directly instead of
        # downloading the AMR file and running our own transcription.
        if mapped == "audio":
            recognition = (content_dict.get("recognition") or "").strip()
            if recognition:
                logger.info(
                    "Using DingTalk voice recognition: %s",
                    recognition[:80],
                )
                return TextContent(
                    type=ContentType.TEXT,
                    text=recognition,
                )

        filename_hint = self._extract_filename_hint(content_dict)
        return self._fetch_download_url_and_content(
            dl_code,
            robot_code,
            mapped,
            filename_hint=filename_hint,
        )

    def _parse_rich_content(
        self,
        incoming_message: Any,
    ) -> List[Any]:
        """Parse richText from incoming_message into runtime Content list."""
        content: List[Any] = []
        try:
            robot_code = getattr(
                incoming_message,
                "robot_code",
                None,
            ) or getattr(incoming_message, "robotCode", None)
            msg_dict = incoming_message.to_dict()
            c = msg_dict.get("content") or {}
            raw = c.get("richText")
            raw = raw or c.get("rich_text")
            rich_list = raw if isinstance(raw, list) else []
            type_mapping = get_type_mapping()
            for item in rich_list:
                if not isinstance(item, dict):
                    continue
                # Text may be under "text" or "content" (API variation).
                item_text = item.get("text") or item.get("content")
                if item_text is not None:
                    stripped = (item_text or "").strip()
                    if stripped:
                        content.append(
                            TextContent(
                                type=ContentType.TEXT,
                                text=stripped,
                            ),
                        )
                # Picture items may use pictureDownloadCode or downloadCode.
                dl_code = (
                    item.get("downloadCode")
                    or item.get("download_code")
                    or item.get("pictureDownloadCode")
                    or item.get("picture_download_code")
                )
                if not dl_code or not robot_code:
                    continue
                mapped = type_mapping.get(
                    item.get("type", "file"),
                    item.get("type", "file"),
                )
                filename_hint = self._extract_filename_hint(item)
                part_content = self._fetch_download_url_and_content(
                    dl_code,
                    robot_code,
                    mapped,
                    filename_hint=filename_hint,
                )
                if part_content is not None:
                    content.append(part_content)

            # -------- 2) single downloadCode (pure picture/file) --------
            if not content:
                part = self._resolve_single_download(c, msg_dict, robot_code)
                if part is not None:
                    content.append(part)

        except Exception:
            logger.exception("failed to fetch richText download url(s)")
        return content

    def _handle_quoted_media(
        self,
        replied_content: Any,
        replied_msg_type: str,
        robot_code: str,
        text_parts: List[str],
        content_parts: List[Any],
    ) -> None:
        """Handle quoted media message (picture/voice/audio/video/file)."""
        quoted_type_mapping = {
            "picture": "image",
            "voice": "audio",
            "audio": "audio",
            "video": "video",
            "file": "file",
        }
        mapped = quoted_type_mapping.get(replied_msg_type, "file")
        dl_code = ""
        if isinstance(replied_content, dict):
            dl_code = (
                replied_content.get("downloadCode")
                or replied_content.get("download_code")
                or ""
            ).strip()
        if not dl_code or not robot_code:
            text_parts.insert(0, f"[quoted {mapped} message]")
            return
        # Prefer recognition text for voice messages.
        if mapped == "audio" and isinstance(replied_content, dict):
            recognition = (replied_content.get("recognition") or "").strip()
            if recognition:
                text_parts.insert(
                    0,
                    f"[quoted voice message: {recognition}]",
                )
                return
        filename_hint = (
            self._extract_filename_hint(replied_content)
            if isinstance(replied_content, dict)
            else None
        )
        part = self._fetch_download_url_and_content(
            dl_code,
            robot_code,
            mapped,
            filename_hint=filename_hint,
        )
        if part is not None:
            content_parts.append(part)
        else:
            text_parts.insert(0, f"[quoted {mapped}: download failed]")

    def _handle_quoted_rich_text(
        self,
        replied_content: Any,
        robot_code: str,
        text_parts: List[str],
        content_parts: List[Any],
    ) -> None:
        """Handle quoted richText message (text + picture items)."""
        rich_list = []
        if isinstance(replied_content, dict):
            rich_list = replied_content.get("richText") or []
        has_content = False
        for item in rich_list:
            if not isinstance(item, dict):
                continue
            item_type = (item.get("msgType") or "").strip()
            if item_type == "text":
                item_text = (item.get("content") or "").strip()
                if item_text:
                    text_parts.insert(0, f"[quoted message: {item_text}]")
                    has_content = True
            elif item_type == "picture":
                dl_code = (
                    item.get("downloadCode")
                    or item.get("download_code")
                    or item.get("pictureDownloadCode")
                    or item.get("picture_download_code")
                    or ""
                ).strip()
                if dl_code and robot_code:
                    part = self._fetch_download_url_and_content(
                        dl_code,
                        robot_code,
                        "image",
                    )
                    if part is not None:
                        content_parts.append(part)
                        has_content = True
                    else:
                        text_parts.insert(
                            0,
                            "[quoted image: download failed]",
                        )
                        has_content = True
        if not has_content:
            text_parts.insert(0, "[quoted richText message]")

    def _process_quoted_message(
        self,
        raw_data: Dict[str, Any],
        text_parts: List[str],
        content_parts: List[Any],
    ) -> None:
        """Process quoted (replied-to) message from DingTalk callback.

        Only user messages carry retrievable content; bot messages
        (interactiveCard) have no content in the callback payload.
        """
        text_data = raw_data.get("text")
        if not isinstance(text_data, dict):
            return
        if not text_data.get("isReplyMsg"):
            return

        replied_msg = text_data.get("repliedMsg")
        if not isinstance(replied_msg, dict):
            return

        replied_msg_type = (replied_msg.get("msgType") or "").strip()
        replied_content = replied_msg.get("content")

        # Bot message (e.g. interactiveCard): no content in payload.
        if not replied_content:
            text_parts.insert(
                0,
                "[quoted bot message: content unavailable]",
            )
            return

        robot_code = (raw_data.get("robotCode") or "").strip()

        if replied_msg_type == "text":
            quoted_text = ""
            if isinstance(replied_content, dict):
                quoted_text = (replied_content.get("text") or "").strip()
            elif isinstance(replied_content, str):
                quoted_text = replied_content.strip()
            if quoted_text:
                text_parts.insert(0, f"[quoted message: {quoted_text}]")

        elif replied_msg_type in (
            "picture",
            "voice",
            "audio",
            "video",
            "file",
        ):
            self._handle_quoted_media(
                replied_content,
                replied_msg_type,
                robot_code,
                text_parts,
                content_parts,
            )

        elif replied_msg_type == "richText":
            self._handle_quoted_rich_text(
                replied_content,
                robot_code,
                text_parts,
                content_parts,
            )

        else:
            text_parts.insert(
                0,
                f"[quoted {replied_msg_type or 'unknown'} message]",
            )

    async def process(self, callback: CallbackMessage) -> tuple[int, str]:
        # pylint: disable=too-many-branches,too-many-statements
        try:
            # Raw msgId from channel callback for dedup (not assigned id).
            raw_data = getattr(callback, "data", None) or {}
            raw_msg_id = str(
                raw_data.get("msgId") or raw_data.get("msg_id") or "",
            ).strip()
            logger.info(
                "dingtalk raw callback: msgId=%r keys=%s",
                raw_msg_id or "(empty)",
                list(raw_data.keys()) if isinstance(raw_data, dict) else "?",
            )
            incoming_message = ChatbotMessage.from_dict(callback.data)

            logger.debug(
                "Dingtalk message received: %s",
                incoming_message.to_dict(),
            )
            content_parts: List[Any] = []
            text = ""
            if incoming_message.text:
                text = (incoming_message.text.content or "").strip()
            if text:
                content_parts.append(
                    TextContent(type=ContentType.TEXT, text=text),
                )
            # Always parse rich content so images/files are not dropped
            # when the message also contains text.
            content = self._parse_rich_content(incoming_message)
            # If text was extracted separately and rich content has no
            # text items, prepend the text so both text and media are
            # preserved. Do not prepend when top-level text is only a
            # placeholder (e.g. "\\n", "//n") so image+text from richText
            # is not overwritten.
            rich_has_text = any(
                item.type == "text" and (item.text or "").strip()
                for item in content
            )
            text_is_placeholder = not (text or "").strip() or (
                (text or "").strip() in ("\\n", "//n")
            )
            if (
                text
                and content
                and not rich_has_text
                and not text_is_placeholder
            ):
                content.insert(
                    0,
                    TextContent(type=ContentType.TEXT, text=text),
                )
            # Handle quoted (replied-to) message if present.
            quoted_text_parts: List[str] = []
            quoted_content_parts: List[Any] = []
            self._process_quoted_message(
                raw_data,
                quoted_text_parts,
                quoted_content_parts,
            )
            # Merge quoted parts into the main content list.
            target_parts = content if content else content_parts
            if quoted_text_parts:
                quoted_combined = "\n".join(quoted_text_parts)
                target_parts.insert(
                    0,
                    TextContent(type=ContentType.TEXT, text=quoted_combined),
                )
            if quoted_content_parts:
                # Insert quoted media after quoted text but before user text.
                insert_pos = 1 if quoted_text_parts else 0
                for part in reversed(quoted_content_parts):
                    target_parts.insert(insert_pos, part)

            # Use rich content (text + media with local paths) when present.
            parts_to_send = content if content else content_parts

            sender, skip = sender_from_chatbot_message(incoming_message)
            if skip:
                return dingtalk_stream.AckMessage.STATUS_OK, "ok"

            conversation_id = conversation_id_from_chatbot_message(
                incoming_message,
            )
            conversation_type = conversation_type_from_chatbot_message(
                incoming_message,
            )
            is_group = conversation_type == "group"

            is_bot_mentioned = bool(raw_data.get("isInAtList"))

            meta: Dict[str, Any] = {
                "conversation_type": conversation_type,
                "is_group": is_group,
                "sender_staff_id": getattr(
                    incoming_message,
                    "sender_staff_id",
                    None,
                )
                or getattr(incoming_message, "senderStaffId", None)
                or "",
                "sender_dingtalk_id": getattr(
                    incoming_message,
                    "sender_id",
                    None,
                )
                or getattr(incoming_message, "senderId", None)
                or "",
                "user_name": getattr(
                    incoming_message,
                    "sender_nick",
                    None,
                )
                or getattr(incoming_message, "senderNick", None)
                or "",
            }
            if is_bot_mentioned:
                meta["bot_mentioned"] = True
            if conversation_id:
                meta["conversation_id"] = conversation_id
            if raw_msg_id:
                meta["message_id"] = raw_msg_id
            sw = getattr(incoming_message, "sessionWebhook", None) or getattr(
                incoming_message,
                "session_webhook",
                None,
            )
            logger.debug(
                "dingtalk request: has_session_webhook=%s sender=%s",
                bool(sw),
                sender,
            )
            if sw:
                meta["session_webhook"] = sw
                sw_exp = getattr(
                    incoming_message,
                    "sessionWebhookExpiredTime",
                    None,
                ) or getattr(
                    incoming_message,
                    "session_webhook_expired_time",
                    None,
                )
                if sw_exp is not None:
                    meta["session_webhook_expired_time"] = int(sw_exp)
                logger.info(
                    "dingtalk recv: session_webhook present "
                    "session_from_url=%s "
                    "expired_time=%s",
                    session_param_from_webhook_url(sw),
                    sw_exp,
                )
            else:
                logger.debug(
                    "dingtalk recv: no sessionWebhook on incoming_message",
                )

            # Group mention check: skip if require_mention but not mentioned
            if is_group and self._require_mention and not is_bot_mentioned:
                logger.info(
                    "dingtalk group mention skip: sender=%s",
                    sender,
                )
                self.reply_text(" ", incoming_message)
                return dingtalk_stream.AckMessage.STATUS_OK, "ok"

            # Dedup by message_id only.
            if self._try_accept_message and not self._try_accept_message(
                raw_msg_id,
            ):
                logger.info(
                    "dingtalk duplicate ignored: raw_msg_id=%r from=%s",
                    raw_msg_id,
                    sender,
                )
                self.reply_text(" ", incoming_message)
                return dingtalk_stream.AckMessage.STATUS_OK, "ok"

            logger.info(
                "dingtalk accept: raw_msg_id=%r",
                raw_msg_id or "(empty)",
            )
            native = {
                "channel_id": "dingtalk",
                "sender_id": sender,
                "acl_sender_id": meta.get("sender_dingtalk_id") or sender,
                "content_parts": parts_to_send,
                "meta": meta,
            }
            if raw_msg_id:
                native["message_id"] = raw_msg_id
            if sw:
                native["session_webhook"] = sw
            logger.info(
                "dingtalk emit: native has_sw=%s meta_sw=%s",
                bool(native.get("session_webhook")),
                bool((native.get("meta") or {}).get("session_webhook")),
            )
            logger.info("recv from=%s text=%s", sender, text[:100])
            self._emit_native_threadsafe(native)

            # ACK immediately: all replies go through sessionWebhook or
            # AI Card, so we never need reply_text to carry real content.
            # This prevents DingTalk retry storms during long LLM runs.
            self.reply_text(" ", incoming_message)
            logger.info("dingtalk handler: ACK sent for sender=%s", sender)
            return dingtalk_stream.AckMessage.STATUS_OK, "ok"

        except Exception:
            logger.exception("process failed")
            return dingtalk_stream.AckMessage.STATUS_SYSTEM_EXCEPTION, "error"
