import asyncio
import os
import sys
import time
import uuid
from collections.abc import Awaitable, Callable
from pathlib import Path
from typing import Any, cast
from urllib.parse import unquote

from requests import Response
from wechatpy.enterprise import WeChatClient, parse_message
from wechatpy.enterprise.crypto import WeChatCrypto
from wechatpy.enterprise.messages import ImageMessage, TextMessage, VoiceMessage
from wechatpy.exceptions import InvalidSignatureException
from wechatpy.messages import BaseMessage

from astrbot.api.event import MessageChain
from astrbot.api.message_components import File, Image, Plain, Record
from astrbot.api.platform import (
    AstrBotMessage,
    MessageMember,
    MessageType,
    Platform,
    PlatformMetadata,
    register_platform_adapter,
)
from astrbot.core import logger
from astrbot.core.platform.astr_message_event import MessageSesion
from astrbot.core.platform.webhook_server import FastAPIWebhookServer
from astrbot.core.utils.astrbot_path import get_astrbot_temp_path
from astrbot.core.utils.media_utils import (
    MEDIA_MIME_EXTENSIONS,
    MediaResolver,
    detect_image_mime_type_async,
)
from astrbot.core.utils.webhook_utils import log_webhook_info

from .wecom_event import WecomPlatformEvent
from .wecom_kf import WeChatKF
from .wecom_kf_message import WeChatKFMessage

if sys.version_info >= (3, 12):
    from typing import override
else:
    from typing_extensions import override


def _extract_wecom_media_filename(disposition: str | None) -> str | None:
    if not disposition:
        return None

    for part in disposition.split(";"):
        token = part.strip()
        token_lower = token.lower()
        if token_lower.startswith("filename*="):
            value = token.split("=", 1)[1].strip().strip('"')
            if value.lower().startswith("utf-8''"):
                value = value[7:]
            filename = Path(unquote(value).replace("\\", "/")).name
            return filename or None
        if token_lower.startswith("filename="):
            value = token.split("=", 1)[1].strip().strip('"')
            filename = Path(value.replace("\\", "/")).name
            return filename or None

    return None


class WecomServer:
    def __init__(self, event_queue: asyncio.Queue, config: dict) -> None:
        self.server = FastAPIWebhookServer("wecom-webhook")
        self.port = int(cast(str, config.get("port")))
        self.callback_server_host = config.get("callback_server_host", "0.0.0.0")
        self.server.add_url_rule(
            "/callback/command",
            view_func=self.verify,
            methods=["GET"],
        )
        self.server.add_url_rule(
            "/callback/command",
            view_func=self.callback_command,
            methods=["POST"],
        )
        self.event_queue = event_queue

        self.crypto = WeChatCrypto(
            config["token"].strip(),
            config["encoding_aes_key"].strip(),
            config["corpid"].strip(),
        )

        self.callback: Callable[[BaseMessage], Awaitable[None]] | None = None
        self.shutdown_event = asyncio.Event()

    async def verify(self, request):
        """内部服务器的 GET 验证入口"""
        return await self.handle_verify(request)

    async def handle_verify(self, request) -> str:
        """处理验证请求，可被统一 webhook 入口复用

        Args:
            request: FastAPI webhook request 对象

        Returns:
            验证响应
        """
        logger.info(f"验证请求有效性: {request.args}")
        args = request.args
        try:
            echo_str = self.crypto.check_signature(
                args.get("msg_signature"),
                args.get("timestamp"),
                args.get("nonce"),
                args.get("echostr"),
            )
            logger.info("验证请求有效性成功。")
            return echo_str
        except InvalidSignatureException:
            logger.error("验证请求有效性失败，签名异常，请检查配置。")
            raise

    async def callback_command(self, request):
        """内部服务器的 POST 回调入口"""
        return await self.handle_callback(request)

    async def handle_callback(self, request) -> str:
        """处理回调请求，可被统一 webhook 入口复用

        Args:
            request: FastAPI webhook request 对象

        Returns:
            响应内容
        """
        data = await request.get_data()
        msg_signature = request.args.get("msg_signature")
        timestamp = request.args.get("timestamp")
        nonce = request.args.get("nonce")
        try:
            xml = self.crypto.decrypt_message(data, msg_signature, timestamp, nonce)
        except InvalidSignatureException:
            logger.error("解密失败，签名异常，请检查配置。")
            raise
        else:
            msg = cast(BaseMessage, parse_message(xml))
            logger.info(f"解析成功: {msg}")

            if self.callback:
                await self.callback(msg)

        return "success"

    async def start_polling(self) -> None:
        logger.info(
            f"将在 {self.callback_server_host}:{self.port} 端口启动 企业微信 适配器。",
        )
        await self.server.run_task(
            host=self.callback_server_host,
            port=self.port,
            shutdown_trigger=self.shutdown_trigger,
        )

    async def shutdown_trigger(self) -> None:
        await self.shutdown_event.wait()


@register_platform_adapter("wecom", "wecom 适配器", support_streaming_message=False)
class WecomPlatformAdapter(Platform):
    WECHAT_KF_TEXT_CONTENT_DEDUP_TTL_SECONDS = 15

    def __init__(
        self,
        platform_config: dict,
        platform_settings: dict,
        event_queue: asyncio.Queue,
    ) -> None:
        super().__init__(platform_config, event_queue)
        self.settingss = platform_settings
        self.api_base_url = platform_config.get(
            "api_base_url",
            "https://qyapi.weixin.qq.com/cgi-bin/",
        )
        self.unified_webhook_mode = platform_config.get("unified_webhook_mode", False)

        if not self.api_base_url:
            self.api_base_url = "https://qyapi.weixin.qq.com/cgi-bin/"

        self.api_base_url = self.api_base_url.removesuffix("/")
        if not self.api_base_url.endswith("/cgi-bin"):
            self.api_base_url += "/cgi-bin"

        if not self.api_base_url.endswith("/"):
            self.api_base_url += "/"

        self.server = WecomServer(self._event_queue, self.config)
        self.agent_id: str | None = None
        self._wechat_kf_seen_text_messages: dict[str, float] = {}

        self.client = WeChatClient(
            self.config["corpid"].strip(),
            self.config["secret"].strip(),
        )

        # 微信客服
        self.kf_name = self.config.get("kf_name", None)
        if self.kf_name:
            # inject
            self.wechat_kf_api = WeChatKF(client=self.client)
            self.wechat_kf_message_api = WeChatKFMessage(self.client)
            self.client.__setattr__("kf", self.wechat_kf_api)
            self.client.__setattr__("kf_message", self.wechat_kf_message_api)

        self.client.__setattr__("API_BASE_URL", self.api_base_url)

        async def callback(msg: BaseMessage) -> None:
            if msg.type == "unknown" and msg._data["Event"] == "kf_msg_or_event":

                def get_latest_msg_item() -> dict | None:
                    token = msg._data["Token"]
                    kfid = msg._data["OpenKfId"]
                    has_more = 1
                    ret = {}
                    while has_more:
                        ret = self.wechat_kf_api.sync_msg(token, kfid)
                        has_more = ret["has_more"]
                    msg_list = ret.get("msg_list", [])
                    if msg_list:
                        return msg_list[-1]
                    return None

                msg_new = await asyncio.get_running_loop().run_in_executor(
                    None,
                    get_latest_msg_item,
                )
                if msg_new:
                    await self.convert_wechat_kf_message(msg_new)
                return
            await self.convert_message(msg)

        self.server.callback = callback

    def _is_duplicate_wechat_kf_text_message(self, session_id: str, text: str) -> bool:
        normalized_text = text.strip()
        if not normalized_text:
            return False

        now = time.monotonic()
        expired_keys = [
            key
            for key, expires_at in self._wechat_kf_seen_text_messages.items()
            if expires_at <= now
        ]
        for key in expired_keys:
            self._wechat_kf_seen_text_messages.pop(key, None)

        dedup_key = f"{session_id}:{normalized_text}"
        if dedup_key in self._wechat_kf_seen_text_messages:
            return True
        self._wechat_kf_seen_text_messages[dedup_key] = (
            now + self.WECHAT_KF_TEXT_CONTENT_DEDUP_TTL_SECONDS
        )
        return False

    @override
    async def send_by_session(
        self,
        session: MessageSesion,
        message_chain: MessageChain,
    ) -> None:
        # 企业微信客服不支持主动发送
        if hasattr(self.client, "kf_message"):
            logger.warning("企业微信客服模式不支持 send_by_session 主动发送。")
            await super().send_by_session(session, message_chain)
            return
        if not self.agent_id:
            logger.warning(
                f"send_by_session 失败：无法为会话 {session.session_id} 推断 agent_id。",
            )
            await super().send_by_session(session, message_chain)
            return

        message_obj = AstrBotMessage()
        message_obj.self_id = self.agent_id
        message_obj.session_id = session.session_id
        message_obj.type = session.message_type
        message_obj.sender = MessageMember(session.session_id, session.session_id)
        message_obj.message = []
        message_obj.message_str = ""
        message_obj.message_id = uuid.uuid4().hex
        message_obj.raw_message = {"_proactive_send": True}

        event = self.create_event(message_obj)
        await event.send(message_chain)
        await super().send_by_session(session, message_chain)

    @override
    def meta(self) -> PlatformMetadata:
        return PlatformMetadata(
            "wecom",
            "wecom 适配器",
            id=self.config.get("id", "wecom"),
            support_streaming_message=False,
            support_proactive_message=False,
        )

    @override
    async def run(self) -> None:
        loop = asyncio.get_running_loop()
        if self.kf_name:
            try:
                acc_list = (
                    await loop.run_in_executor(
                        None,
                        self.wechat_kf_api.get_account_list,
                    )
                ).get("account_list", [])
                logger.debug(f"获取到微信客服列表: {acc_list!s}")
                for acc in acc_list:
                    name = acc.get("name", None)
                    if name != self.kf_name:
                        continue
                    open_kfid = acc.get("open_kfid", None)
                    if not open_kfid:
                        logger.error("获取微信客服失败，open_kfid 为空。")
                    logger.debug(f"Found open_kfid: {open_kfid!s}")
                    kf_url = (
                        await loop.run_in_executor(
                            None,
                            self.wechat_kf_api.add_contact_way,
                            open_kfid,
                            "astrbot_placeholder",
                        )
                    ).get("url", "")
                    logger.info(
                        f"请打开以下链接，在微信扫码以获取客服微信: https://api.cl2wm.cn/api/qrcode/code?text={kf_url}",
                    )
            except Exception as e:
                logger.error(e)

        # 如果启用统一 webhook 模式，则不启动独立服务器
        webhook_uuid = self.config.get("webhook_uuid")
        if self.unified_webhook_mode and webhook_uuid:
            log_webhook_info(f"{self.meta().id}(企业微信)", webhook_uuid)
            # 保持运行状态，等待 shutdown
            await self.server.shutdown_event.wait()
        else:
            await self.server.start_polling()

    async def webhook_callback(self, request: Any) -> Any:
        """统一 Webhook 回调入口"""
        # 根据请求方法分发到不同的处理函数
        if request.method == "GET":
            return await self.server.handle_verify(request)
        else:
            return await self.server.handle_callback(request)

    async def convert_message(self, msg: BaseMessage) -> AstrBotMessage | None:
        abm = AstrBotMessage()
        if isinstance(msg, TextMessage):
            abm.message_str = msg.content
            abm.self_id = str(msg.agent)
            abm.message = [Plain(msg.content)]
            abm.type = MessageType.FRIEND_MESSAGE
            abm.sender = MessageMember(
                cast(str, msg.source),
                cast(str, msg.source),
            )
            abm.message_id = str(msg.id)
            abm.timestamp = int(cast(int | str, msg.time))
            abm.session_id = abm.sender.user_id
            abm.raw_message = msg
        elif isinstance(msg, ImageMessage):
            abm.message_str = "[图片]"
            abm.self_id = str(msg.agent)
            abm.message = [Image(file=msg.image, url=msg.image)]
            abm.type = MessageType.FRIEND_MESSAGE
            abm.sender = MessageMember(
                cast(str, msg.source),
                cast(str, msg.source),
            )
            abm.message_id = str(msg.id)
            abm.timestamp = int(cast(int | str, msg.time))
            abm.session_id = abm.sender.user_id
            abm.raw_message = msg
        elif isinstance(msg, VoiceMessage):
            resp: Response = await asyncio.get_running_loop().run_in_executor(
                None,
                self.client.media.download,
                msg.media_id,
            )
            temp_dir = get_astrbot_temp_path()
            path = os.path.join(temp_dir, f"wecom_{msg.media_id}.amr")
            await asyncio.to_thread(Path(path).write_bytes, resp.content)

            try:
                path_wav = await MediaResolver(
                    path,
                    media_type="audio",
                    default_suffix=".wav",
                ).to_path(target_format="wav")
            except Exception as e:
                logger.error(f"转换音频失败: {e}。如果没有安装 ffmpeg 请先安装。")
                path_wav = path
                return

            abm.message_str = ""
            abm.self_id = str(msg.agent)
            abm.message = [Record(file=path_wav, url=path_wav)]
            abm.type = MessageType.FRIEND_MESSAGE
            abm.sender = MessageMember(
                cast(str, msg.source),
                cast(str, msg.source),
            )
            abm.message_id = str(msg.id)
            abm.timestamp = int(cast(int | str, msg.time))
            abm.session_id = abm.sender.user_id
            abm.raw_message = msg
        else:
            logger.warning(f"暂未实现的事件: {msg.type}")
            return

        self.agent_id = abm.self_id
        logger.info(f"abm: {abm}")
        await self.handle_msg(abm)

    async def convert_wechat_kf_message(self, msg: dict) -> AstrBotMessage | None:
        msgtype = msg.get("msgtype")
        external_userid = cast(str, msg.get("external_userid"))
        abm = AstrBotMessage()
        abm.raw_message = msg
        abm.raw_message["_wechat_kf_flag"] = None  # 方便处理
        abm.self_id = msg["open_kfid"]
        abm.sender = MessageMember(external_userid, external_userid)
        abm.session_id = external_userid
        abm.type = MessageType.FRIEND_MESSAGE
        abm.message_id = msg.get("msgid", uuid.uuid4().hex[:8])
        abm.message_str = ""
        if msgtype == "text":
            text = msg.get("text", {}).get("content", "").strip()
            if self._is_duplicate_wechat_kf_text_message(abm.session_id, text):
                logger.debug(
                    "忽略 15 秒内重复微信客服文本消息 session_id=%s text=%s",
                    abm.session_id,
                    text,
                )
                return None
            abm.message = [Plain(text=text)]
            abm.message_str = text
        elif msgtype == "image":
            media_id = msg.get("image", {}).get("media_id", "")
            resp: Response = await asyncio.get_running_loop().run_in_executor(
                None,
                self.client.media.download,
                media_id,
            )
            temp_dir = get_astrbot_temp_path()
            mime_type = await detect_image_mime_type_async(
                resp.content,
                default_mime_type=None,
            )
            suffix = MEDIA_MIME_EXTENSIONS.get(mime_type or "", ".jpg")
            path = os.path.join(temp_dir, f"weixinkefu_{media_id}{suffix}")
            await asyncio.to_thread(Path(path).write_bytes, resp.content)
            abm.message = [Image(file=path, url=path)]
        elif msgtype == "voice":
            media_id = msg.get("voice", {}).get("media_id", "")
            resp: Response = await asyncio.get_running_loop().run_in_executor(
                None,
                self.client.media.download,
                media_id,
            )

            temp_dir = get_astrbot_temp_path()
            path = os.path.join(temp_dir, f"weixinkefu_{media_id}.amr")
            await asyncio.to_thread(Path(path).write_bytes, resp.content)

            try:
                path_wav = await MediaResolver(
                    path,
                    media_type="audio",
                    default_suffix=".wav",
                ).to_path(target_format="wav")
            except Exception as e:
                logger.error(f"转换音频失败: {e}。如果没有安装 ffmpeg 请先安装。")
                path_wav = path
                return

            abm.message = [Record(file=path_wav, url=path_wav)]
        elif msgtype == "file":
            media_id = msg.get("file", {}).get("media_id", "")
            if not media_id:
                logger.warning(f"微信客服文件消息缺少 media_id: {msg}")
                return

            resp: Response = await asyncio.get_running_loop().run_in_executor(
                None,
                self.client.media.download,
                media_id,
            )

            file_name = (
                _extract_wecom_media_filename(
                    resp.headers.get("Content-Disposition"),
                )
                or f"weixinkefu_{media_id}.bin"
            )
            temp_dir = Path(get_astrbot_temp_path())
            file_path = temp_dir / f"weixinkefu_{uuid.uuid4().hex}_{file_name}"
            file_path.write_bytes(resp.content)

            abm.message = [File(name=file_name, file=str(file_path))]
        else:
            logger.warning(f"未实现的微信客服消息事件: {msg}")
            return
        await self.handle_msg(abm)

    def create_event(self, message: AstrBotMessage) -> WecomPlatformEvent:
        """Creates a WeCom message event.

        Args:
            message: AstrBot message object to wrap.

        Returns:
            Created WeCom message event.
        """
        return WecomPlatformEvent(
            message_str=message.message_str,
            message_obj=message,
            platform_meta=self.meta(),
            session_id=message.session_id,
            client=self.client,
        )

    async def handle_msg(self, message: AstrBotMessage) -> None:
        self.commit_event(self.create_event(message))

    def get_client(self) -> WeChatClient:
        return self.client

    async def terminate(self) -> None:
        self.server.shutdown_event.set()
        try:
            await self.server.server.shutdown()
        except Exception as _:
            pass
        logger.info("企业微信 适配器已被关闭")
