MoviePilot/tests/test_agent_image_support.py

"""
验证消息渠道的图片、音频和文件输入解析及 Agent 路由。
"""

import asyncio
import base64
import json
import tempfile
from dataclasses import replace
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock, patch
from urllib.parse import quote

# 此导入使用兼容包的惰性公开符号，Pylint 无法静态解析。
from app.agent import AgentChain, MoviePilotAgent  # pylint: disable=no-name-in-module

# 此导入使用兼容包的惰性公开符号，Pylint 无法静态解析。
from app.agent.llm import LLMHelper  # pylint: disable=no-name-in-module
from app.agent.tools.impl.send_local_file import SendLocalFileInput
from app.agent.tools.impl.send_message import SendMessageInput, SendMessageTool
from app.chain.message import MessageChain
from app.modules.discord import DiscordModule

# 此导入使用兼容包的惰性公开符号，Pylint 无法静态解析。
from app.modules.qqbot import QQBotModule  # pylint: disable=no-name-in-module
from app.modules.qqbot.qqbot import QQBot
from app.modules.slack import SlackModule
from app.modules.synologychat import SynologyChatModule

# 此导入使用兼容包的惰性公开符号，Pylint 无法静态解析。
from app.modules.telegram import TelegramModule  # pylint: disable=no-name-in-module
from app.modules.telegram.botapi import TelegramBotApi
from app.modules.telegram.telegram import Telegram
from app.modules.vocechat import VoceChatModule
from app.modules.wechat import WechatModule
from app.modules.wechat.wechatbot import WeChatBot
from app.runtime.config import settings
from app.runtime.loop import main_loop_registry

# 此导入使用兼容包的惰性公开符号，Pylint 无法静态解析。
from app.schemas import IncomingMessage, Message  # pylint: disable=no-name-in-module
from app.schemas.types import MessageType, NotificationChannel


def test_telegram_extract_audio_refs_returns_prefixed_file_ids():
    """
    Telegram 音频引用带渠道前缀，便于后续选择下载器。
    """
    audio_refs = TelegramModule._extract_audio_refs(
        {
            "voice": {"file_id": "voice-1"},
            "audio": {"file_id": "audio-1"},
        }
    )

    assert audio_refs == ["tg://voice_file_id/voice-1", "tg://audio_file_id/audio-1"]


def test_telegram_extract_images_returns_prefixed_file_ids():
    """
    Telegram 图片引用带渠道前缀，便于后续选择下载器。
    """
    images = TelegramModule._extract_images(
        {
            "photo": [{"file_id": "small"}, {"file_id": "large"}],
            "document": {
                "file_id": "doc-image",
                "mime_type": "image/png",
                "file_name": "poster.png",
            },
        }
    )

    assert [image.ref for image in images] == ["tg://file_id/large", "tg://file_id/doc-image"]
    assert images[0].mime_type == "image/jpeg"
    assert images[1].mime_type == "image/png"
    assert images[1].name == "poster.png"


def test_telegram_message_parser_accepts_double_encoded_body():
    """
    Telegram 消息解析兼容被重复 JSON 编码的请求正文。
    """
    module = TelegramModule()
    body = json.dumps(
        json.dumps(
            {
                "message": {
                    "from": {"id": 10001, "username": "tester"},
                    "chat": {"id": 10001, "type": "private"},
                    "photo": [{"file_id": "small"}, {"file_id": "large"}],
                }
            }
        )
    )

    with patch.object(
        module,
        "get_config",
        return_value=SimpleNamespace(name="telegram-test", config={}),
    ), patch.object(
        module,
        "get_instance",
        return_value=SimpleNamespace(bot_username=None),
    ):
        message = module.message_parser(
            source="telegram-test", body=body, form={}, args={}
        )

    assert message is not None
    assert [image.ref for image in message.images] == ["tg://file_id/large"]


def test_telegram_download_file_uses_configured_file_url():
    """
    Telegram 文件下载使用配置的文件服务地址。
    """
    telegram = Telegram.__new__(Telegram)
    # 自建 Bot API 地址：下载地址跟随该地址，且不走全局代理。
    bot = TelegramBotApi("token-123", api_url="https://tg-proxy.example")
    telegram._bot = bot

    with patch.object(
        bot, "get_file", return_value={"file_path": "photos/a.jpg"}
    ), patch(
        "app.modules.telegram.telegram.RequestUtils.get_res",
        return_value=SimpleNamespace(content=b"image-bytes"),
    ) as get_res:
        content = telegram.download_file("file-id-1")

    assert content == b"image-bytes"
    assert bot.proxy is None
    get_res.assert_called_once_with(
        "https://tg-proxy.example/file/bottoken-123/photos/a.jpg"
    )


def test_telegram_download_file_uses_default_file_url_with_proxy():
    """
    Telegram 默认文件服务地址保留代理配置。
    """
    telegram = Telegram.__new__(Telegram)
    proxy = {"https": "http://127.0.0.1:7890"}
    # 官方 Bot API 地址：下载请求沿用客户端的代理配置。
    bot = TelegramBotApi("token-123", proxy=proxy)
    telegram._bot = bot

    with patch.object(
        bot, "get_file", return_value={"file_path": "photos/a.jpg"}
    ), patch("app.modules.telegram.telegram.RequestUtils") as request_utils:
        request_utils.return_value.get_res.return_value = SimpleNamespace(
            content=b"image-bytes"
        )
        content = telegram.download_file("file-id-1")

    assert content == b"image-bytes"
    request_utils.assert_called_once_with(proxies=proxy, timeout=30)
    request_utils.return_value.get_res.assert_called_once_with(
        "https://api.telegram.org/file/bottoken-123/photos/a.jpg"
    )


def test_process_allows_image_only_message():
    """
    消息处理入口接受仅包含图片的消息。
    """
    chain = MessageChain()
    message = IncomingMessage(
        channel=NotificationChannel.Telegram,
        source="telegram-test",
        userid="10001",
        username="tester",
        images=["tg://file_id/image-1"],
    )

    with patch.object(chain, "message_parser", return_value=message), patch.object(
        chain, "handle_message"
    ) as handle_message:
        chain.process(body="{}", form={}, args={"source": "telegram-test"})

    handle_kwargs = handle_message.call_args.kwargs
    assert handle_kwargs["text"] == ""
    assert [image.ref for image in handle_kwargs["images"]] == ["tg://file_id/image-1"]


def test_process_allows_audio_only_message():
    """
    消息处理入口接受仅包含音频的消息。
    """
    chain = MessageChain()
    message = IncomingMessage(
        channel=NotificationChannel.Telegram,
        source="telegram-test",
        userid="10001",
        username="tester",
        audio_refs=["tg://voice_file_id/voice-1"],
    )

    with patch.object(chain, "message_parser", return_value=message), patch.object(
        chain, "handle_message"
    ) as handle_message:
        chain.process(body="{}", form={}, args={"source": "telegram-test"})

    handle_kwargs = handle_message.call_args.kwargs
    assert handle_kwargs["text"] == ""
    assert handle_kwargs["audio_refs"] == ["tg://voice_file_id/voice-1"]


def test_process_allows_file_only_message():
    """
    消息处理入口接受仅包含文件的消息。
    """
    chain = MessageChain()
    message = IncomingMessage(
        channel=NotificationChannel.Telegram,
        source="telegram-test",
        userid="10001",
        username="tester",
        files=[
            IncomingMessage.MessageAttachment(
                ref="tg://document_file_id/doc-1",
                name="note.txt",
                mime_type="text/plain",
                size=12,
            )
        ],
    )

    with patch.object(chain, "message_parser", return_value=message), patch.object(
        chain, "handle_message"
    ) as handle_message:
        chain.process(body="{}", form={}, args={"source": "telegram-test"})

    handle_kwargs = handle_message.call_args.kwargs
    assert handle_kwargs["text"] == ""
    assert handle_kwargs["files"][0].ref == "tg://document_file_id/doc-1"


def test_image_message_routes_to_agent_even_when_global_agent_is_disabled():
    """
    全局 Agent 开关关闭时，图片消息仍进入指定 Agent 路由。
    """
    chain = MessageChain()
    chain.runtime_config = replace(
        chain.runtime_config,
        ai_agent_enable=True,
        ai_agent_global=False,
    )

    with patch.object(chain, "load_cache", return_value={}), patch.object(
        chain.messagehelper, "put"
    ), patch.object(chain.messageoper, "add"), patch.object(
        chain, "_handle_ai_message"
    ) as handle_ai_message, patch.object(
        settings, "AI_AGENT_ENABLE", True
    ), patch.object(
        settings, "AI_AGENT_GLOBAL", False
    ):
        chain.handle_message(
            channel=NotificationChannel.Telegram,
            source="telegram-test",
            userid="10001",
            username="tester",
            text="",
            images=["tg://file_id/image-1"],
        )

    handle_ai_message.assert_called_once()


def test_audio_message_routes_to_agent_without_forcing_voice_reply():
    """
    音频消息进入 Agent，但不强制使用语音回复。
    """
    chain = MessageChain()
    chain.runtime_config = replace(
        chain.runtime_config,
        ai_agent_enable=True,
        ai_agent_global=False,
    )

    with patch.object(chain, "load_cache", return_value={}), patch.object(
        chain, "_transcribe_audio_refs", return_value="帮我推荐一部电影"
    ), patch.object(chain.messagehelper, "put"), patch.object(
        chain.messageoper, "add"
    ), patch.object(chain, "_handle_ai_message") as handle_ai_message:
        with patch.object(settings, "AI_AGENT_ENABLE", True), patch.object(
            settings, "AI_AGENT_GLOBAL", False
        ):
            chain.handle_message(
                channel=NotificationChannel.Telegram,
                source="telegram-test",
                userid="10001",
                username="tester",
                text="",
                audio_refs=["tg://voice_file_id/voice-1"],
            )

    handle_ai_message.assert_called_once()
    assert handle_ai_message.call_args.kwargs["text"] == "帮我推荐一部电影"
    assert handle_ai_message.call_args.kwargs["has_audio_input"]
    assert "reply_with_voice" not in handle_ai_message.call_args.kwargs


def test_file_message_routes_to_agent_even_when_global_agent_is_disabled():
    """
    全局 Agent 开关关闭时，文件消息仍进入指定 Agent 路由。
    """
    chain = MessageChain()
    chain.runtime_config = replace(
        chain.runtime_config,
        ai_agent_enable=True,
        ai_agent_global=False,
    )

    with patch.object(chain, "load_cache", return_value={}), patch.object(
        chain.messagehelper, "put"
    ), patch.object(chain.messageoper, "add"), patch.object(
        chain, "_handle_ai_message"
    ) as handle_ai_message, patch.object(
        settings, "AI_AGENT_ENABLE", True
    ), patch.object(
        settings, "AI_AGENT_GLOBAL", False
    ):
        chain.handle_message(
            channel=NotificationChannel.Telegram,
            source="telegram-test",
            userid="10001",
            username="tester",
            text="",
            files=[
                IncomingMessage.MessageAttachment(
                    ref="tg://document_file_id/doc-1",
                    name="report.txt",
                    mime_type="text/plain",
                )
            ],
        )

    handle_ai_message.assert_called_once()
    assert handle_ai_message.call_args.kwargs["files"][0].name == "report.txt"


def test_transcribe_audio_refs_supports_new_channel_refs():
    """
    音频转写支持新增渠道引用格式。
    """
    chain = MessageChain()
    audio_refs = [
        "slack://file/" + quote("https://files.slack.com/test.mp3", safe=""),
        "discord://file/" + quote("https://cdn.discordapp.com/voice.ogg", safe=""),
        "qq://file/" + quote("https://example.com/qq-voice.ogg", safe=""),
        "vocechat://file/%2Fuploads%2Fvoice.ogg",
        "synology://file/" + quote("https://example.com/synology-voice.wav", safe=""),
        "feishu://file/om_audio/file_audio/voice.opus",
    ]

    with patch(
        "app.chain.message.is_audio_input_available", return_value=True
    ), patch.object(
        chain,
        "run_module",
        side_effect=[b"slack", b"discord", b"qq", b"vocechat", b"synology", b"feishu"],
    ) as run_module, patch(
        "app.chain.message.transcribe_audio",
        side_effect=[
            "slack text",
            "discord text",
            "qq text",
            "vocechat text",
            "synology text",
            "feishu text",
        ],
    ) as transcribe_bytes:
        result = chain._transcribe_audio_refs(
            audio_refs=audio_refs,
            channel=NotificationChannel.Slack,
            source="mixed-source",
        )

    assert result == "slack text\ndiscord text\nqq text\nvocechat text\nsynology text\nfeishu text"
    assert [call.args[0] for call in run_module.call_args_list] == [
            "download_slack_file_bytes",
            "download_discord_file_bytes",
            "download_qq_file_bytes",
            "download_vocechat_file_bytes",
            "download_synologychat_file_bytes",
            "download_feishu_file_bytes",
        ]
    assert [call.kwargs["filename"] for call in transcribe_bytes.call_args_list] == [
            "test.mp3",
            "voice.ogg",
            "qq-voice.ogg",
            "voice.ogg",
            "synology-voice.wav",
            "voice.opus",
        ]


def test_agent_send_agent_message_keeps_default_text_reply():
    """
    发送 Agent 消息默认保持文本回复。
    """
    agent = MoviePilotAgent(
        session_id="session-1",
        user_id="user-1",
        channel=NotificationChannel.Telegram.value,
        source="telegram-test",
        username="tester",
    )

    with patch.object(
        AgentChain, "async_post_message", new_callable=AsyncMock
    ) as async_post_message:
        import asyncio

        asyncio.run(agent.send_agent_message("这是语音回复"))

    notification = async_post_message.await_args.args[0]
    assert notification.voice_path is None
    assert notification.voice_caption is None
    assert notification.text == "这是语音回复"


def test_agent_process_wraps_request_as_structured_json():
    """
    Agent 处理入口将请求包装为结构化 JSON。
    """
    agent = MoviePilotAgent(
        session_id="session-1",
        user_id="user-1",
        channel=NotificationChannel.Telegram.value,
        source="telegram-test",
        username="tester",
    )

    with patch(
        "app.agent.memory.memory_manager.get_agent_messages", return_value=[]
    ), patch.object(agent, "_execute_agent", new_callable=AsyncMock) as execute_agent:
        import asyncio

        asyncio.run(
            agent.process(
                "帮我总结这个文件",
                files=[
                    {
                        "name": "report.txt",
                        "local_path": "/tmp/report.txt",
                        "status": "ready",
                    }
                ],
            )
        )

    messages = execute_agent.await_args.args[0]
    human_message = messages[-1]
    content = human_message.content
    assert isinstance(content, list)
    payload = json.loads(content[0]["text"])
    assert payload["message"] == "帮我总结这个文件"
    assert payload["input"]["mode"] == "text"
    assert not payload["input"]["transcribed"]
    assert payload["files"][0]["local_path"] == "/tmp/report.txt"


def test_agent_process_marks_voice_input_in_structured_json():
    """
    语音输入应在结构化消息中标记为转写来源。
    """
    agent = MoviePilotAgent(
        session_id="session-1",
        user_id="user-1",
        channel=NotificationChannel.Telegram.value,
        source="telegram-test",
        username="tester",
    )

    with patch(
        "app.agent.memory.memory_manager.get_agent_messages", return_value=[]
    ), patch.object(agent, "_execute_agent", new_callable=AsyncMock) as execute_agent:
        asyncio.run(
            agent.process(
                "帮我推荐一部电影",
                has_audio_input=True,
            )
        )

    messages = execute_agent.await_args.args[0]
    payload = json.loads(messages[-1].content[0]["text"])
    assert payload["message"] == "帮我推荐一部电影"
    assert payload["input"]["mode"] == "voice"
    assert payload["input"]["transcribed"]


def test_llm_supports_image_input_respects_explicit_override():
    """
    图片输入能力优先遵循显式覆盖配置。
    """
    with patch.object(settings, "LLM_SUPPORT_IMAGE_INPUT", False):
        assert not LLMHelper.supports_image_input()


def test_llm_supports_image_input_uses_boolean_setting():
    """
    图片输入能力读取布尔配置值。
    """
    with patch.object(settings, "LLM_SUPPORT_IMAGE_INPUT", True):
        assert LLMHelper.supports_image_input()

    with patch.object(settings, "LLM_SUPPORT_IMAGE_INPUT", False):
        assert not LLMHelper.supports_image_input()


def test_handle_ai_message_routes_images_to_files_when_image_input_disabled():
    """
    模型不支持图片输入时，将图片按文件交给 Agent。
    """
    chain = MessageChain()
    chain.runtime_config = replace(
        chain.runtime_config,
        ai_agent_enable=True,
    )

    loop = Mock(**{"is_running.return_value": True, "is_closed.return_value": False})
    with patch.object(main_loop_registry, "require", return_value=loop), patch.object(
        settings, "AI_AGENT_ENABLE", True
    ), patch.object(
        settings, "LLM_SUPPORT_IMAGE_INPUT", False
    ), patch(
        "app.chain.message.supports_image_input", return_value=False
    ), patch.object(chain, "_get_or_create_session_id", return_value="session-1"), patch.object(
        chain, "_download_attachments_to_data_urls"
    ) as download_images, patch.object(
        chain,
        "_prepare_agent_files",
        return_value=[
            {
                "name": "image_1.jpg",
                "mime_type": "image/jpeg",
                "local_path": "/tmp/image_1.jpg",
                "status": "ready",
            }
        ],
    ) as prepare_files, patch(
        "app.chain.message.get_running_agent_manager"
    ) as get_running_manager, patch(
        "app.chain.message.asyncio.run_coroutine_threadsafe",
        side_effect=lambda coro, _loop: coro.close(),
    ) as run_coroutine_threadsafe:
        process_message = AsyncMock()
        get_running_manager.return_value.process_message = process_message
        chain._handle_ai_message(
            text="/ai 帮我看看这张图",
            channel=NotificationChannel.Telegram,
            source="telegram-test",
            userid="10001",
            username="tester",
            images=["tg://file_id/image-1"],
        )

    download_images.assert_not_called()
    prepare_files.assert_called_once()
    attachments = prepare_files.call_args.kwargs["files"]
    assert attachments[0].ref == "tg://file_id/image-1"
    assert attachments[0].mime_type == "image/jpeg"
    run_coroutine_threadsafe.assert_called_once()
    assert process_message.call_args.kwargs["images"] is None
    assert process_message.call_args.kwargs["files"][0]["local_path"] == "/tmp/image_1.jpg"


def test_handle_ai_message_forwards_voice_input_to_agent_manager():
    """
    AI消息入队时应保留语音输入标记。
    """
    chain = MessageChain()
    chain.runtime_config = replace(
        chain.runtime_config,
        ai_agent_enable=True,
    )

    loop = Mock(**{"is_running.return_value": True, "is_closed.return_value": False})
    with patch.object(main_loop_registry, "require", return_value=loop), patch.object(
        settings, "AI_AGENT_ENABLE", True
    ), patch.object(
        chain, "_get_or_create_session_id", return_value="session-1"
    ), patch(
        "app.chain.message.get_running_agent_manager"
    ) as get_running_manager, patch(
        "app.chain.message.asyncio.run_coroutine_threadsafe",
        side_effect=lambda coro, _loop: coro.close(),
    ):
        process_message = AsyncMock()
        get_running_manager.return_value.process_message = process_message
        chain._handle_ai_message(
            text="帮我推荐一部电影",
            channel=NotificationChannel.Telegram,
            source="telegram-test",
            userid="10001",
            username="tester",
            has_audio_input=True,
        )

    assert process_message.call_args.kwargs["has_audio_input"]


def test_slack_images_use_authenticated_data_url_download():
    """
    Slack 图片通过带认证的下载流程转换为数据 URL。
    """
    chain = MessageChain()

    with patch.object(
        chain,
        "run_module",
        return_value="data:image/png;base64,abc123",
    ) as run_module:
        images = chain._download_attachments_to_data_urls(
            attachments=["https://files.slack.com/files-pri/T1-F1/test.png"],
            channel=NotificationChannel.Slack,
            source="slack-test",
        )

    assert images == ["data:image/png;base64,abc123"]
    run_module.assert_called_once_with(
        "download_slack_file_to_data_url",
        file_url="https://files.slack.com/files-pri/T1-F1/test.png",
        source="slack-test",
    )


def test_slack_module_download_file_to_data_url():
    """
    Slack 模块将下载文件内容转换为数据 URL。
    """
    module = SlackModule()
    client = Mock()
    client.download_file.return_value = (b"png-binary", "image/png")

    with patch.object(
        module, "get_config", return_value=SimpleNamespace(name="slack-test")
    ), patch.object(module, "get_instance", return_value=client):
        data_url = module.download_slack_file_to_data_url(
            "https://files.slack.com/files-pri/T1-F1/test.png",
            "slack-test",
        )

    assert data_url == f"data:image/png;base64,{base64.b64encode(b'png-binary').decode()}"


def test_slack_extract_audio_refs_returns_private_file_refs():
    """
    Slack 音频解析保留私有文件引用。
    """
    audio_refs = SlackModule._extract_audio_refs(
        {
            "files": [
                {
                    "type": "audio",
                    "filetype": "mp3",
                    "mimetype": "audio/mpeg",
                    "url_private": "https://files.slack.com/files-pri/T1-F1/test.mp3",
                }
            ]
        }
    )

    assert audio_refs == [
            "slack://file/"
            + quote("https://files.slack.com/files-pri/T1-F1/test.mp3", safe="")
        ]


def test_send_message_input_accepts_image_only_payload():
    """
    消息发送参数允许只提供图片。
    """
    payload = SendMessageInput(
        image_url="https://example.com/poster.png",
    )

    assert payload.image_url == "https://example.com/poster.png"


def test_send_message_tool_uses_regular_notification_type():
    """
    发送消息工具应按普通通知消息登记。
    """

    async def _run():
        tool = SendMessageTool(session_id="session-1", user_id="10001")
        tool.set_message_attr(
            channel=NotificationChannel.Telegram.value,
            source="telegram-test",
            username="tester",
        )

        with patch(
            "app.agent.tools.base.ToolChain.async_post_message",
            new_callable=AsyncMock,
        ) as async_post_message:
            result = await tool.run(
                message="处理完成",
                title="智能体通知",
                image_url="https://example.com/poster.png",
            )
        return result, async_post_message

    result, async_post_message = asyncio.run(_run())
    notification = async_post_message.await_args.args[0]

    assert result == "消息已发送"
    assert notification.mtype == MessageType.Other
    assert notification.channel == NotificationChannel.Telegram
    assert notification.source == "telegram-test"
    assert notification.title == "智能体通知"
    assert notification.text == "处理完成"
    assert notification.image == "https://example.com/poster.png"
    assert notification.parse_mode is None


def test_send_message_tool_ignores_parse_mode_argument():
    """
    发送消息工具不再支持由 Agent 指定 Telegram parse_mode。
    """

    async def _run():
        tool = SendMessageTool(session_id="session-1", user_id="10001")
        tool.set_message_attr(
            channel=NotificationChannel.Telegram.value,
            source="telegram-test",
            username="tester",
        )

        with patch(
            "app.agent.tools.base.ToolChain.async_post_message",
            new_callable=AsyncMock,
        ) as async_post_message:
            result = await tool.run(
                message="<b>处理完成</b>",
                parse_mode="HTML",
            )
        return result, async_post_message

    result, async_post_message = asyncio.run(_run())
    notification = async_post_message.await_args.args[0]

    assert result == "消息已发送"
    assert notification.text == "<b>处理完成</b>"
    assert notification.parse_mode is None


def test_send_message_tool_marks_reply_sent_after_dispatch():
    """
    发送消息工具成功发送后应终止本轮回复。
    """

    async def _run():
        tool = SendMessageTool(session_id="session-1", user_id="10001")
        agent_context = {}
        tool.set_agent_context(agent_context)
        tool.set_message_attr(
            channel=NotificationChannel.Telegram.value,
            source="telegram-test",
            username="tester",
        )

        with patch(
            "app.agent.tools.base.ToolChain.async_post_message",
            new_callable=AsyncMock,
        ):
            result = await tool.run(message="处理完成")
        return result, agent_context

    result, agent_context = asyncio.run(_run())

    assert result == "消息已发送"
    assert agent_context["user_reply_sent"]
    assert agent_context["reply_mode"] == "send_message"


def test_send_local_file_input_accepts_file_payload():
    """
    本地文件发送参数接受文件载荷。
    """
    payload = SendLocalFileInput(
        file_path="/tmp/report.txt",
        message="请下载查看",
    )

    assert payload.file_path == "/tmp/report.txt"


def test_discord_extract_images_supports_attachment_content_type():
    """
    Discord 根据附件内容类型提取图片。
    """
    images = DiscordModule._extract_images(
        {
            "attachments": [
                {
                    "content_type": "image/png",
                    "url": "https://cdn.discordapp.com/test.png",
                }
            ]
        }
    )

    assert [image.ref for image in images] == ["https://cdn.discordapp.com/test.png"]
    assert images[0].mime_type == "image/png"


def test_discord_extract_audio_refs_supports_attachment_content_type():
    """
    Discord 根据附件内容类型提取音频引用。
    """
    audio_refs = DiscordModule._extract_audio_refs(
        {
            "attachments": [
                {
                    "content_type": "audio/ogg",
                    "filename": "voice.ogg",
                    "url": "https://cdn.discordapp.com/voice.ogg",
                }
            ]
        }
    )

    assert audio_refs == [
            "discord://file/"
            + quote("https://cdn.discordapp.com/voice.ogg", safe="")
        ]


def test_discord_extract_files_supports_non_media_attachment():
    """
    Discord 将非媒体附件保留为通用文件。
    """
    files = DiscordModule._extract_files(
        {
            "attachments": [
                {
                    "content_type": "application/pdf",
                    "filename": "guide.pdf",
                    "url": "https://cdn.discordapp.com/guide.pdf",
                    "size": 1024,
                }
            ]
        }
    )

    assert files[0].name == "guide.pdf"
    assert files[0].ref == ("discord://file/" + quote("https://cdn.discordapp.com/guide.pdf", safe=""))


def test_discord_send_direct_message_returns_chat_id():
    """
    Discord 私信发送返回会话 ID。
    """
    module = DiscordModule()
    client = Mock()
    client.send_msg.return_value = (
        True,
        {"message_id": "discord-msg-1", "chat_id": "discord-chat-1"},
    )

    with patch.object(
        module,
        "get_configs",
        return_value={"discord-test": SimpleNamespace(name="discord-test")},
    ), patch.object(
        module, "check_message", return_value=True
    ), patch.object(
        module, "get_instance", return_value=client
    ):
        response = module.send_direct_message(
            Message(title="hi", userid="user-1")
        )

    assert response is not None
    assert response.message_id == "discord-msg-1"
    assert response.chat_id == "discord-chat-1"


def test_download_images_routes_wechat_refs_to_module_downloader():
    """
    微信图片引用交给对应模块下载器处理。
    """
    chain = MessageChain()

    with patch.object(
        chain,
        "run_module",
        return_value="data:image/png;base64,wechat123",
    ) as run_module:
        images = chain._download_attachments_to_data_urls(
            attachments=["wxwork://media_id/media-1"],
            channel=NotificationChannel.Wechat,
            source="wechat-test",
        )

    assert images == ["data:image/png;base64,wechat123"]
    run_module.assert_called_once_with(
        "download_wechat_image_to_data_url",
        image_ref="wxwork://media_id/media-1",
        source="wechat-test",
    )


def test_download_images_routes_feishu_refs_to_module_downloader():
    """
    飞书图片引用交给对应模块下载器处理。
    """
    chain = MessageChain()

    with patch.object(
        chain,
        "run_module",
        return_value="data:image/png;base64,feishu123",
    ) as run_module:
        data_urls = chain._download_attachments_to_data_urls(
            attachments=[
                IncomingMessage.MessageImage(
                    ref="feishu://image/img_v2_xxx",
                    mime_type="image/png",
                )
            ],
            channel=NotificationChannel.Feishu,
            source="feishu-test",
        )

    assert data_urls == ["data:image/png;base64,feishu123"]
    run_module.assert_called_once_with(
        "download_feishu_image_to_data_url",
        image_ref="feishu://image/img_v2_xxx",
        source="feishu-test",
    )


def test_download_message_file_bytes_supports_feishu_refs():
    """
    消息文件下载支持飞书引用并返回原始字节。
    """
    chain = MessageChain()

    with patch.object(chain, "run_module", return_value=b"feishu-file") as run_module:
        content = chain._download_message_file_bytes(
            file_ref="feishu://file/file_xxx/report.pdf",
            channel=NotificationChannel.Feishu,
            source="feishu-test",
        )

    assert content == b"feishu-file"
    run_module.assert_called_once_with(
        "download_feishu_file_bytes",
        file_ref="feishu://file/file_xxx/report.pdf",
        source="feishu-test",
    )


def test_wechat_message_parser_extracts_image_media_id():
    """
    企业微信消息解析提取图片的媒体标识。
    """
    module = WechatModule()
    xml_message = b"""
    <xml>
      <FromUserName><![CDATA[user-1]]></FromUserName>
      <MsgType><![CDATA[image]]></MsgType>
      <PicUrl><![CDATA[https://example.com/image.png]]></PicUrl>
      <MediaId><![CDATA[media-1]]></MediaId>
    </xml>
    """
    crypt = Mock()
    crypt.DecryptMsg.return_value = (0, xml_message)

    with patch.object(
        module,
        "get_config",
        return_value=SimpleNamespace(
            name="wechat-test",
            config={
                "WECHAT_TOKEN": "token",
                "WECHAT_ENCODING_AESKEY": "encoding",
                "WECHAT_CORPID": "corpid",
            },
        ),
    ), patch.object(
        module, "get_instance", return_value=SimpleNamespace(send_msg=Mock())
    ), patch(
        "app.modules.wechat.WXBizMsgCrypt",
        return_value=crypt,
    ):
        message = module.message_parser(
            source="wechat-test",
            body=b"encrypted",
            form={},
            args={"msg_signature": "sig", "timestamp": "1", "nonce": "n"},
        )

    assert message is not None
    assert [image.ref for image in message.images] == ["wxwork://media_id/media-1"]


def test_wechat_message_parser_extracts_file_media_id():
    """
    企业微信消息解析提取文件的媒体标识。
    """
    module = WechatModule()
    xml_message = b"""
    <xml>
      <FromUserName><![CDATA[user-1]]></FromUserName>
      <MsgType><![CDATA[file]]></MsgType>
      <MediaId><![CDATA[file-media-1]]></MediaId>
      <FileName><![CDATA[manual.pdf]]></FileName>
    </xml>
    """
    crypt = Mock()
    crypt.DecryptMsg.return_value = (0, xml_message)

    with patch.object(
        module,
        "get_config",
        return_value=SimpleNamespace(
            name="wechat-test",
            config={
                "WECHAT_TOKEN": "token",
                "WECHAT_ENCODING_AESKEY": "encoding",
                "WECHAT_CORPID": "corpid",
            },
        ),
    ), patch.object(
        module, "get_instance", return_value=SimpleNamespace(send_msg=Mock())
    ), patch(
        "app.modules.wechat.WXBizMsgCrypt",
        return_value=crypt,
    ):
        message = module.message_parser(
            source="wechat-test",
            body=b"encrypted",
            form={},
            args={"msg_signature": "sig", "timestamp": "1", "nonce": "n"},
        )

    assert message is not None
    assert message.files[0].ref == "wxwork://file_media_id/file-media-1"


def test_wechat_bot_parser_accepts_image_only_payload():
    """
    企业微信机器人解析器接受纯图片载荷。
    """
    module = WechatModule()
    body = json.dumps(
        {
            "body": {
                "from": {"userid": "wxbot-user"},
                "msgtype": "image",
                "image": {
                    "download_url": "https://example.com/encrypted-image",
                    "aeskey": "YWJjZGVmZw",
                },
            }
        }
    )

    with patch.object(
        module,
        "get_config",
        return_value=SimpleNamespace(
            name="wechat-bot-test", config={"WECHAT_MODE": "bot"}
        ),
    ), patch.object(
        module, "get_instance", return_value=SimpleNamespace(send_msg=Mock())
    ):
        message = module.message_parser(
            source="wechat-bot-test",
            body=body,
            form={},
            args={},
        )

    assert message is not None
    assert message.images[0].ref.startswith("wxbot://image/")


def test_wechat_bot_handles_image_only_callback():
    """
    企业微信机器人处理仅包含图片的回调。
    """
    bot = WeChatBot.__new__(WeChatBot)
    bot._config_name = "wechat-bot-test"
    bot._admins = []
    bot.send_msg = Mock()
    bot._remember_target = Mock()
    bot._forward_to_message_chain = Mock()

    payload = {
        "body": {
            "from": {"userid": "wxbot-user"},
            "msgtype": "image",
            "image": {
                "download_url": "https://example.com/encrypted-image",
                "aeskey": "YWJjZGVmZw",
            },
        }
    }

    bot._handle_callback_message(payload)

    bot._remember_target.assert_called_once_with("wxbot-user")
    bot._forward_to_message_chain.assert_called_once_with(payload)


def test_vocechat_message_parser_extracts_image_file_payload():
    """
    VoceChat 消息解析保留图片文件载荷。
    """
    module = VoceChatModule()
    body = json.dumps(
        {
            "detail": {
                "type": "normal",
                "content_type": "vocechat/file",
                "content": "/uploads/poster.png",
                "properties": {"content_type": "image/png"},
            },
            "from_uid": 7910,
            "target": {"gid": 2},
        }
    )

    with patch.object(
        module,
        "get_config",
        return_value=SimpleNamespace(
            name="vocechat-test", config={"channel_id": "2"}
        ),
    ):
        message = module.message_parser(
            source="vocechat-test",
            body=body,
            form={},
            args={},
        )

    assert message is not None
    assert [image.ref for image in message.images] == ["vocechat://file/%2Fuploads%2Fposter.png"]
    assert message.images[0].mime_type == "image/png"


def test_vocechat_message_parser_extracts_audio_file_payload():
    """
    VoceChat 消息解析保留音频文件载荷。
    """
    module = VoceChatModule()
    body = json.dumps(
        {
            "detail": {
                "type": "normal",
                "content_type": "vocechat/file",
                "content": "/uploads/voice.ogg",
                "properties": {"content_type": "audio/ogg"},
            },
            "from_uid": 7910,
            "target": {"gid": 2},
        }
    )

    with patch.object(
        module,
        "get_config",
        return_value=SimpleNamespace(
            name="vocechat-test", config={"channel_id": "2"}
        ),
    ):
        message = module.message_parser(
            source="vocechat-test",
            body=body,
            form={},
            args={},
        )

    assert message is not None
    assert message.audio_refs == ["vocechat://file/%2Fuploads%2Fvoice.ogg"]


def test_vocechat_message_parser_extracts_generic_file_payload():
    """
    VoceChat 消息解析保留通用文件载荷。
    """
    module = VoceChatModule()
    body = json.dumps(
        {
            "detail": {
                "type": "normal",
                "content_type": "vocechat/file",
                "content": "/uploads/manual.pdf",
                "properties": {"content_type": "application/pdf"},
            },
            "from_uid": 7910,
            "target": {"gid": 2},
        }
    )

    with patch.object(
        module,
        "get_config",
        return_value=SimpleNamespace(
            name="vocechat-test", config={"channel_id": "2"}
        ),
    ):
        message = module.message_parser(
            source="vocechat-test",
            body=body,
            form={},
            args={},
        )

    assert message is not None
    assert message.files[0].ref == "vocechat://file/%2Fuploads%2Fmanual.pdf"


def test_vocechat_post_message_passes_image_and_correct_target():
    """
    VoceChat 发送消息时传递图片及正确的接收目标。
    """
    module = VoceChatModule()
    client = Mock()

    with patch.object(
        module,
        "get_configs",
        return_value={"vocechat-test": SimpleNamespace(name="vocechat-test")},
    ), patch.object(
        module, "check_message", return_value=True
    ), patch.object(
        module, "get_instance", return_value=client
    ):
        module.post_message(
            Message(
                title="poster",
                image="https://example.com/poster.png",
                targets={"vocechat_userid": "UID#100"},
            )
        )

    client.send_msg.assert_called_once_with(
        title="poster",
        text=None,
        image="https://example.com/poster.png",
        userid="UID#100",
        link=None,
    )


def test_slack_post_message_passes_local_file():
    """
    Slack 消息发送向客户端传递本地文件。
    """
    module = SlackModule()
    client = Mock()

    with tempfile.TemporaryDirectory() as tempdir:
        file_path = Path(tempdir) / "guide.pdf"
        file_path.write_bytes(b"pdf")

        with patch.object(
            module,
            "get_configs",
            return_value={"slack-test": SimpleNamespace(name="slack-test")},
        ), patch.object(
            module, "check_message", return_value=True
        ), patch.object(
            module, "get_instance", return_value=client
        ):
            module.post_message(
                Message(
                    title="手册",
                    text="请下载",
                    file_path=str(file_path),
                    file_name="guide.pdf",
                    userid="U123",
                )
            )

    client.send_file.assert_called_once()


def test_discord_post_message_passes_local_file():
    """
    Discord 消息发送向客户端传递本地文件。
    """
    module = DiscordModule()
    client = Mock()

    with tempfile.TemporaryDirectory() as tempdir:
        file_path = Path(tempdir) / "guide.pdf"
        file_path.write_bytes(b"pdf")

        with patch.object(
            module,
            "get_configs",
            return_value={
                "discord-test": SimpleNamespace(
                    name="discord-test",
                    type="discord",
                    enabled=True,
                )
            },
        ), patch.object(
            module, "check_message", return_value=True
        ), patch.object(
            module, "get_instance", return_value=client
        ):
            module.post_message(
                Message(
                    title="手册",
                    text="请下载",
                    file_path=str(file_path),
                    file_name="guide.pdf",
                    userid="user-1",
                )
            )

    client.send_file.assert_called_once()


def test_qq_message_parser_accepts_image_only_attachment():
    """
    QQ 消息解析支持纯图片附件。
    """
    module = QQBotModule()

    with patch.object(
        module,
        "get_config",
        return_value=SimpleNamespace(name="qq-test", config={}),
    ):
        message = module.message_parser(
            source="qq-test",
            body={
                "type": "C2C_MESSAGE_CREATE",
                "author": {"user_openid": "qq-user"},
                "attachments": [
                    {
                        "content_type": "image/png",
                        "url": "https://example.com/qq-image.png",
                    }
                ],
            },
            form={},
            args={},
        )

    assert message is not None
    assert [image.ref for image in message.images] == ["https://example.com/qq-image.png"]
    assert message.images[0].mime_type == "image/png"


def test_qq_markdown_image_size_preserves_poster_ratio():
    """
    QQ Markdown 图片尺寸保留海报宽高比。
    """
    with patch.object(QQBot, "_get_image_size", return_value=(1000, 1500)):
        content, use_markdown = QQBot._format_message_markdown(
            title="poster",
            image="https://example.com/poster.jpg",
        )

    assert use_markdown
    assert "![image #341px #512px](https://example.com/poster.jpg)" in content


def test_qq_markdown_image_uses_poster_ratio_fallback():
    """
    QQ 无法取得图片尺寸时使用海报比例回退值。
    """
    with patch.object(QQBot, "_get_image_size", return_value=None):
        content, use_markdown = QQBot._format_message_markdown(
            image="https://example.com/poster.jpg",
        )

    assert use_markdown
    assert content == "![image #208px #320px](https://example.com/poster.jpg)"


def test_qq_message_parser_accepts_audio_only_attachment():
    """
    QQ 消息解析支持纯音频附件。
    """
    module = QQBotModule()

    with patch.object(
        module,
        "get_config",
        return_value=SimpleNamespace(name="qq-test", config={}),
    ):
        message = module.message_parser(
            source="qq-test",
            body={
                "type": "C2C_MESSAGE_CREATE",
                "author": {"user_openid": "qq-user"},
                "attachments": [
                    {
                        "content_type": "audio/ogg",
                        "filename": "voice.ogg",
                        "url": "https://example.com/qq-voice.ogg",
                    }
                ],
            },
            form={},
            args={},
        )

    assert message is not None
    assert message.audio_refs == ["qq://file/" + quote("https://example.com/qq-voice.ogg", safe="")]


def test_synology_message_parser_accepts_image_only_form():
    """
    群晖 Chat 消息解析接受仅图片的表单。
    """
    module = SynologyChatModule()

    with patch.object(
        module,
        "get_config",
        return_value=SimpleNamespace(name="synology-test", config={}),
    ), patch.object(
        module,
        "get_instance",
        return_value=SimpleNamespace(check_token=lambda token: token == "token-1"),
    ):
        message = module.message_parser(
            source="synology-test",
            body={},
            form={
                "token": "token-1",
                "user_id": "42",
                "username": "tester",
                "file_url": "https://example.com/image.png",
            },
            args={},
        )

    assert message is not None
    assert [image.ref for image in message.images] == ["https://example.com/image.png"]


def test_synology_message_parser_accepts_audio_only_form():
    """
    群晖 Chat 消息解析接受仅音频的表单。
    """
    module = SynologyChatModule()

    with patch.object(
        module,
        "get_config",
        return_value=SimpleNamespace(name="synology-test", config={}),
    ), patch.object(
        module,
        "get_instance",
        return_value=SimpleNamespace(check_token=lambda token: token == "token-1"),
    ):
        message = module.message_parser(
            source="synology-test",
            body={},
            form={
                "token": "token-1",
                "user_id": "42",
                "username": "tester",
                "attachments": json.dumps(
                    [
                        {
                            "url": "https://example.com/voice.ogg",
                            "content_type": "audio/ogg",
                        }
                    ]
                ),
            },
            args={},
        )

    assert message is not None
    assert message.audio_refs == ["synology://file/" + quote("https://example.com/voice.ogg", safe="")]


def test_synology_message_parser_accepts_generic_file_attachment():
    """
    群晖 Chat 消息解析保留通用文件附件。
    """
    module = SynologyChatModule()

    with patch.object(
        module,
        "get_config",
        return_value=SimpleNamespace(name="synology-test", config={}),
    ), patch.object(
        module,
        "get_instance",
        return_value=SimpleNamespace(check_token=lambda token: token == "token-1"),
    ):
        message = module.message_parser(
            source="synology-test",
            body={},
            form={
                "token": "token-1",
                "user_id": "42",
                "username": "tester",
                "attachments": json.dumps(
                    [
                        {
                            "url": "https://example.com/manual.pdf",
                            "content_type": "application/pdf",
                            "filename": "manual.pdf",
                        }
                    ]
                ),
            },
            args={},
        )

    assert message is not None
    assert message.files[0].ref == ("synology://file/" + quote("https://example.com/manual.pdf", safe=""))


def test_prepare_agent_files_saves_local_file():
    """
    为 Agent 准备文件时，将消息文件保存到本地。
    """
    chain = MessageChain()
    with tempfile.TemporaryDirectory() as tempdir:
        chain.runtime_config = replace(
            chain.runtime_config,
            temporary_path=Path(tempdir),
        )
        with patch.object(
            chain,
            "_download_message_file_bytes",
            return_value="你好，MoviePilot".encode("utf-8"),
        ):
            prepared = chain._prepare_agent_files(
                session_id="session-1",
                files=[
                    IncomingMessage.MessageAttachment(
                        ref="tg://document_file_id/doc-1",
                        name="note.txt",
                        mime_type="text/plain",
                    )
                ],
                channel=NotificationChannel.Telegram,
                source="telegram-test",
            )

            assert prepared[0]["status"] == "ready"
            assert Path(prepared[0]["local_path"]).exists()


def test_telegram_post_message_passes_file_to_client():
    """
    Telegram 消息发送向客户端传递文件参数。
    """
    module = TelegramModule()
    client = Mock()

    with tempfile.TemporaryDirectory() as tempdir:
        file_path = Path(tempdir) / "report.txt"
        file_path.write_text("hello", encoding="utf-8")

        with patch.object(
            module,
            "get_configs",
            return_value={"telegram-test": SimpleNamespace(name="telegram-test")},
        ), patch.object(
            module, "check_message", return_value=True
        ), patch.object(
            module, "get_instance", return_value=client
        ):
            module.post_message(
                Message(
                    title="报告",
                    text="请下载",
                    file_path=str(file_path),
                    file_name="report.txt",
                    userid="user-1",
                )
            )

    client.send_file.assert_called_once()