"""
验证飞书渠道的消息解析、长连接、卡片流式更新和媒体发送。
"""

import json
import tempfile
import threading
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import ANY, MagicMock, patch

import pytest

from app.testing.bootstrap import ensure_optional_stub

# 可选三方依赖在 CI / 全新环境可能未安装，补占位避免 app.modules.feishu 导入失败
ensure_optional_stub("psutil")
ensure_optional_stub("dateparser")
ensure_optional_stub("Pinyin2Hanzi", is_pinyin=lambda value: False)

# 先安装可选依赖占位，避免加载飞书及消息模块时缺少依赖。
from app.agent.callback import StreamingHandler  # noqa: E402

# 先安装可选依赖占位，避免加载飞书及消息模块时缺少依赖。
from app.modules.feishu.feishu import Feishu  # noqa: E402

# 先安装可选依赖占位，避免加载飞书及消息模块时缺少依赖。
from app.modules.feishu.module import FeishuModule  # noqa: E402

# 先安装可选依赖占位，避免加载飞书及消息模块时缺少依赖。
from app.modules.feishu.openapi import FeishuOpenApi, FeishuResponse  # noqa: E402

# 先安装可选依赖占位，避免加载飞书及消息模块时缺少依赖。
from app.schemas.message import (  # noqa: E402
    Message,
    MessageResponse,
)

# 先安装可选依赖占位，避免加载飞书及消息模块时缺少依赖。
from app.schemas.notification import (  # noqa: E402
    ChannelCapability,
    ChannelCapabilityManager,
)

# 先安装可选依赖占位，避免加载飞书及消息模块时缺少依赖。
from app.schemas.types import MessageType, NotificationChannel  # noqa: E402

# 返回 FeishuResponse 的 OpenAPI 方法；桩默认返回失败，用例须显式声明期望成功的调用。
_JSON_API_METHODS = (
    "create_message",
    "reply_message",
    "patch_message",
    "create_message_reaction",
    "delete_message_reaction",
    "upload_image",
    "upload_file",
    "create_card",
    "update_card_element_content",
    "update_card_settings",
)

_STREAM_METADATA_TEMPLATE = {
    "card_id": "card_stream",
    "element_id": Feishu.STREAM_CARD_BODY_ELEMENT_ID,
    "sequence": 0,
}


def _ok(**data) -> FeishuResponse:
    """
    构造开放平台成功响应。
    """
    return FeishuResponse(code=0, msg="success", data=data, log_id="log_test")


def _message_ok(message_id="om_test", chat_id="oc_test", msg_type="interactive") -> FeishuResponse:
    """
    构造发送/回复消息接口的成功响应。
    """
    return _ok(message_id=message_id, chat_id=chat_id, msg_type=msg_type)


def _build_api() -> MagicMock:
    """
    构造按 FeishuOpenApi 契约约束的桩，所有 JSON 接口默认返回失败。
    """
    api = MagicMock(spec=FeishuOpenApi)
    for method in _JSON_API_METHODS:
        getattr(api, method).return_value = FeishuResponse(code=99999, msg="未设置桩响应")
    return api


def _build_client(**kwargs) -> Feishu:
    """
    构造不会启动长连接、OpenAPI 为桩的飞书客户端。
    """
    with (
        patch.object(Feishu, "_build_api_client", return_value=_build_api()),
        patch.object(Feishu, "_start_ws_client"),
    ):
        return Feishu(
            FEISHU_APP_ID="cli_test_app_id",
            FEISHU_APP_SECRET="cli_test_app_secret",
            name="feishu-test",
            **kwargs,
        )


@pytest.fixture
def client() -> Feishu:
    """
    默认飞书客户端。
    """
    return _build_client()


@pytest.fixture
def api(client: Feishu) -> MagicMock:
    """
    客户端持有的 OpenAPI 桩。
    """
    return client._api_client


def _sent_content(call) -> dict:
    """
    解析一次 create_message / reply_message 调用中的消息内容 JSON。
    """
    return json.loads(call.kwargs["content"])


def _stream_metadata(**overrides) -> dict:
    """
    构造流式卡片 metadata，每个用例独立一份，避免共享可变状态。
    """
    return {"feishu_streaming": {**_STREAM_METADATA_TEMPLATE, **overrides}}


def _image_response(content: bytes = b"png-bytes", content_type: str = "image/png") -> MagicMock:
    """
    构造远程图片下载响应。
    """
    response = MagicMock()
    response.content = content
    response.headers = {"Content-Type": content_type}
    return response


def _event(event_type: str, event: dict) -> bytes:
    """
    构造长连接收到的 2.0 版事件报文。
    """
    return json.dumps(
        {"schema": "2.0", "header": {"event_type": event_type, "event_id": "ev_test"}, "event": event},
        ensure_ascii=False,
    ).encode("utf-8")


def _message_event(message: dict, open_id: str = "ou_user_evt", user_id=None) -> dict:
    """
    构造 im.message.receive_v1 事件体。
    """
    return {
        "sender": {"sender_id": {"open_id": open_id, "user_id": user_id}},
        "message": message,
    }


# ---------------------------------------------------------------- 入站消息解析


def test_parse_message_returns_callback_message(client):
    """
    卡片回调解析为带渠道、用户及回调数据的统一消息。
    """
    with patch(
        "app.modules.feishu.feishu.get_configured_user_channel_lookup",
        return_value=lambda **_bindings: None,
    ):
        result = client.parse_message(
            {
                "type": "cardAction",
                "callback_data": "approve",
                "message_id": "om_123",
                "chat_id": "oc_123",
                "sender": {
                    "open_id": "ou_user_1",
                    "user_id": "u_user_1",
                    "name": "tester",
                },
            }
        )

    assert result is not None
    assert result.channel == NotificationChannel.Feishu
    assert result.userid == "ou_user_1"
    assert result.text == "CALLBACK:approve"
    assert result.is_callback
    assert result.chat_id == "oc_123"


def test_extract_card_callback_data_supports_new_and_legacy_values():
    """
    卡片回调数据提取兼容新字段、旧字段和直接字符串。
    """
    assert Feishu._extract_card_callback_data({"callback_data": "approve"}) == "approve"
    assert Feishu._extract_card_callback_data({"value": "legacy"}) == "legacy"
    assert Feishu._extract_card_callback_data("direct") == "direct"
    assert Feishu._extract_card_callback_data({}, name="fallback") == "fallback"


def test_parse_message_blocks_non_admin_command():
    """
    非管理员发出的管理命令不会进入消息处理。
    """
    client = _build_client(FEISHU_ADMINS="ou_admin")

    with (
        patch(
            "app.modules.feishu.feishu.get_configured_user_channel_lookup",
            return_value=lambda **_bindings: None,
        ),
        patch.object(client, "send_text", return_value={"success": True}) as send_text,
    ):
        result = client.parse_message(
            {
                "type": "message",
                "text": "/help",
                "chat_id": "oc_chat_1",
                "sender": {
                    "open_id": "ou_user_2",
                    "user_id": "u_user_2",
                    "name": "tester",
                },
            }
        )

    assert result is None
    send_text.assert_called_once_with(
        "只有管理员才有权限执行此命令",
        userid="ou_user_2",
        chat_id="oc_chat_1",
        receive_id_type="open_id",
    )


def test_parse_message_maps_feishu_ids_to_moviepilot_username(client):
    """
    飞书稳定用户标识映射到绑定的 MoviePilot 用户名。
    """
    get_name = MagicMock(return_value="moviepilot-user")
    with patch(
        "app.modules.feishu.feishu.get_configured_user_channel_lookup",
        return_value=get_name,
    ):
        result = client.parse_message(
            {
                "type": "message",
                "text": "/ai 添加黑客帝国订阅",
                "sender": {
                    "open_id": "ou_bound_user",
                    "user_id": "u_bound_user",
                    "name": "ou_bound_user",
                },
            }
        )

    assert result is not None
    assert result.userid == "ou_bound_user"
    assert result.username == "moviepilot-user"
    get_name.assert_called_once_with(
        feishu_openid="ou_bound_user",
        feishu_userid="u_bound_user",
    )


def test_parse_message_supports_image_and_file_payloads(client):
    """
    消息解析保留图片和通用文件的飞书资源引用。
    """
    with patch(
        "app.modules.feishu.feishu.get_configured_user_channel_lookup",
        return_value=lambda **_bindings: None,
    ):
        image_message = client.parse_message(
            {
                "type": "message",
                "text": "",
                "images": [{"ref": "feishu://image/img_v2_test"}],
                "message_id": "om_img",
                "chat_id": "oc_chat",
                "sender": {"open_id": "ou_user_5", "name": "tester"},
            }
        )
        file_message = client.parse_message(
            {
                "type": "message",
                "text": "",
                "files": [{"ref": "feishu://file/file_key/report.pdf", "name": "report.pdf"}],
                "message_id": "om_file",
                "chat_id": "oc_chat",
                "sender": {"open_id": "ou_user_6", "name": "tester"},
            }
        )

    assert image_message.images[0].ref == "feishu://image/img_v2_test"
    assert file_message.files[0].ref == "feishu://file/file_key/report.pdf"


# ---------------------------------------------------------------- 长连接事件分发


def test_event_handlers_cover_common_im_and_card_events(client):
    """
    长连接需要应答的 IM 与卡片事件都应有处理函数，避免飞书按未处理事件重推。
    """
    handlers = client._event_handlers()

    assert set(handlers) >= {
        "im.message.receive_v1",
        "im.message.message_read_v1",
        "im.message.reaction.created_v1",
        "im.message.reaction.deleted_v1",
        "im.message.recalled_v1",
        "im.chat.access_event.bot_p2p_chat_entered_v1",
        "card.action.trigger",
    }
    assert all(callable(handler) for handler in handlers.values())


def test_dispatch_event_forwards_received_message_to_message_chain(client):
    """
    收到飞书消息事件后，将规范化载荷交给消息链。
    """
    event = _message_event(
        {
            "message_id": "om_text_evt",
            "chat_id": "oc_chat_evt",
            "chat_type": "group",
            "message_type": "text",
            "content": json.dumps({"text": " /search 黑客帝国 "}, ensure_ascii=False),
        },
        open_id="ou_user_evt",
        user_id="u_user_evt",
    )

    with patch.object(client, "_forward_to_message_chain") as forward:
        result = client._dispatch_event(_event("im.message.receive_v1", event))

    assert result is None
    payload = forward.call_args.args[0]
    assert payload["type"] == "message"
    assert payload["source"] == "feishu-test"
    assert payload["message_id"] == "om_text_evt"
    assert payload["chat_id"] == "oc_chat_evt"
    assert payload["chat_type"] == "group"
    assert payload["text"] == "/search 黑客帝国"
    assert payload["sender"] == {"open_id": "ou_user_evt", "user_id": "u_user_evt", "name": "ou_user_evt"}
    # 收到消息后回复同一用户应落到其所在会话。
    assert client._user_chat_mapping["ou_user_evt"] == "oc_chat_evt"


def test_dispatch_card_action_returns_toast_and_forwards_callback(client):
    """
    卡片回调必须同步返回 toast（由长连接编码进应答帧），并转发 callback_data。
    """
    event = {
        "operator": {"open_id": None, "user_id": "u_operator"},
        "action": {"value": {"callback_data": "/subscribe 1"}, "name": "btn"},
        "context": {"open_message_id": "om_card", "open_chat_id": "oc_card"},
    }

    with patch.object(client, "_forward_to_message_chain") as forward:
        result = client._dispatch_event(_event("card.action.trigger", event))

    assert result == {"toast": {"type": "info", "content": "操作已提交"}}
    payload = forward.call_args.args[0]
    assert payload["type"] == "cardAction"
    assert payload["callback_data"] == "/subscribe 1"
    assert payload["message_id"] == "om_card"
    assert payload["chat_id"] == "oc_card"
    assert payload["sender"]["user_id"] == "u_operator"
    # 仅有 user_id 的操作者，后续回复应按 user_id 类型投递。
    assert client._resolve_target(userid="u_operator") == ("u_operator", "user_id")


@pytest.mark.parametrize(
    "event_type, event",
    [
        ("im.message.message_read_v1", {"reader": {"reader_id": {"open_id": "ou_r"}}, "message_id_list": ["om_1"]}),
        ("im.message.reaction.created_v1", {"message_id": "om_1", "reaction_type": {"emoji_type": "OK"}}),
        ("im.message.reaction.deleted_v1", {"message_id": "om_1", "reaction_type": {"emoji_type": "OK"}}),
        ("im.message.recalled_v1", {"message_id": "om_1", "chat_id": "oc_1"}),
        ("im.chat.access_event.bot_p2p_chat_entered_v1", {"chat_id": "oc_1", "operator_id": {"open_id": "ou_1"}}),
        ("contact.user.created_v3", {"object": {}}),
    ],
)
def test_dispatch_event_ignores_informational_and_unknown_events(client, event_type, event):
    """
    已读、表情、撤回等通知类事件与未知事件只记录日志，不进入消息链也不产生应答数据。
    """
    with patch.object(client, "_forward_to_message_chain") as forward:
        result = client._dispatch_event(_event(event_type, event))

    assert result is None
    forward.assert_not_called()


def test_on_message_wraps_feishu_image_ref_with_message_id(client):
    """
    图片引用包含原消息标识，供资源下载接口定位。
    """
    event = _message_event(
        {
            "message_id": "om_img_evt",
            "chat_id": "oc_chat_evt",
            "chat_type": "p2p",
            "message_type": "image",
            "content": json.dumps({"image_key": "img_v2_evt"}),
        }
    )

    with patch.object(client, "_forward_to_message_chain") as forward:
        client._on_message(event)

    payload = forward.call_args.args[0]
    assert payload["images"][0]["ref"] == "feishu://image/om_img_evt/img_v2_evt"


def test_on_message_wraps_feishu_audio_ref_with_message_id(client):
    """
    音频引用包含原消息标识、文件标识和名称。
    """
    event = _message_event(
        {
            "message_id": "om_audio_evt",
            "chat_id": "oc_chat_evt",
            "chat_type": "p2p",
            "message_type": "audio",
            "content": json.dumps({"file_key": "file_audio_evt", "file_name": "voice.opus"}),
        }
    )

    with patch.object(client, "_forward_to_message_chain") as forward:
        client._on_message(event)

    payload = forward.call_args.args[0]
    assert payload["audio_refs"] == ["feishu://file/om_audio_evt/file_audio_evt/voice.opus"]


# ---------------------------------------------------------------- 长连接生命周期


class _BlockingLongConnection:
    """
    替身长连接：run() 阻塞到 stop()，不访问网络；可配置为忽略 stop 模拟卡死线程。
    """

    instances = []

    def __init__(self, app_id, app_secret, on_event, domain, name, honor_stop=True):
        """
        记录构造参数，供用例核对客户端传入的配置与事件回调。
        """
        self.app_id = app_id
        self.app_secret = app_secret
        self.on_event = on_event
        self.domain = domain
        self.name = name
        self.honor_stop = honor_stop
        self.running = threading.Event()
        self.release = threading.Event()
        self.stop_calls = 0
        _BlockingLongConnection.instances.append(self)

    def run(self) -> None:
        """
        阻塞直到被释放。
        """
        self.running.set()
        self.release.wait(timeout=10)

    def stop(self) -> None:
        """
        记录停止请求，按配置决定是否让 run() 返回。
        """
        self.stop_calls += 1
        if self.honor_stop:
            self.release.set()


@pytest.fixture
def long_connection_factory():
    """
    把 Feishu 使用的 FeishuLongConnection 替换为阻塞替身，结束时释放所有线程。
    """
    _BlockingLongConnection.instances = []
    options = {"honor_stop": True}

    def _factory(*args, **kwargs):
        return _BlockingLongConnection(*args, honor_stop=options["honor_stop"], **kwargs)

    with patch("app.modules.feishu.feishu.FeishuLongConnection", side_effect=_factory):
        yield options
    for instance in _BlockingLongConnection.instances:
        instance.release.set()


def _start_client() -> Feishu:
    """
    构造会真实启动长连接线程的客户端（长连接为替身，OpenAPI 为桩）。
    """
    with patch.object(Feishu, "_build_api_client", return_value=_build_api()):
        return Feishu(
            FEISHU_APP_ID="cli_test_app_id",
            FEISHU_APP_SECRET="cli_test_app_secret",
            name="feishu-test",
        )


def test_start_ws_client_runs_long_connection_in_background_thread(long_connection_factory):
    """
    启动长连接时在独立后台线程运行，并公开就绪状态。
    """
    client = _start_client()
    connection = _BlockingLongConnection.instances[0]

    assert connection.running.wait(timeout=2)
    assert client._ws_thread.is_alive()
    assert client._ws_thread is not threading.current_thread()
    assert client.get_state()
    assert (connection.app_id, connection.app_secret, connection.name) == (
        "cli_test_app_id",
        "cli_test_app_secret",
        "feishu-test",
    )
    # 长连接收到的事件应交给该客户端自己的分发入口。
    with patch.object(client, "_forward_to_message_chain"):
        toast = connection.on_event(
            _event("card.action.trigger", {"operator": {"open_id": "ou_1"}, "action": {"value": "ok"}})
        )
    assert toast["toast"]["content"] == "操作已提交"

    assert client.stop()


def test_stop_stops_long_connection_joins_thread_and_closes_api(long_connection_factory):
    """
    停止时关闭长连接，等待线程结束并释放接口客户端。
    """
    client = _start_client()
    connection = _BlockingLongConnection.instances[0]
    assert connection.running.wait(timeout=2)
    api = client._api_client

    assert client.stop() is True

    assert connection.stop_calls == 1
    assert not client._ws_thread.is_alive()
    assert not client.get_state()
    api.close.assert_called_once()


def test_stop_returns_false_when_thread_exceeds_join_budget(long_connection_factory):
    """
    停止等待超过线程回收预算时返回 False，保留未结束状态。
    """
    long_connection_factory["honor_stop"] = False
    client = _start_client()
    client._ws_join_timeout_seconds = 0.05
    connection = _BlockingLongConnection.instances[0]
    assert connection.running.wait(timeout=2)
    api = client._api_client

    assert client.stop() is False

    assert connection.stop_calls == 1
    assert client._ws_thread.is_alive()
    # 线程仍可能在使用 OpenAPI 客户端，未退出时不得关闭 HTTP 会话。
    api.close.assert_not_called()


def test_run_ws_client_marks_ready_only_while_running(client):
    """
    长连接仅在实际运行期间公开就绪状态。
    """
    observed = []
    client._ws_client = MagicMock()
    client._ws_client.run.side_effect = lambda: observed.append(client.get_state())

    client._run_ws_client()

    assert observed == [True]
    assert not client.get_state()


def test_run_ws_client_swallows_long_connection_error(client):
    """
    长连接运行异常不向外传播，结束后清除就绪状态。
    """
    client._ws_client = MagicMock()
    client._ws_client.run.side_effect = RuntimeError("fatal")

    client._run_ws_client()

    assert not client.get_state()


# ---------------------------------------------------------------- 发送通知与卡片


def test_send_notification_uses_direct_card_content(client, api):
    """
    普通通知直接发送卡片正文，不增加多余包装层。
    """
    api.create_message.return_value = _message_ok()

    result = client.send_notification(
        Message(
            title="测试标题",
            text="测试正文",
            buttons=[[{"text": "确认", "callback_data": "confirm"}]],
        ),
        userid="ou_user_3",
    )

    assert result["success"]
    call = api.create_message.call_args
    assert call.kwargs["receive_id_type"] == "open_id"
    assert call.kwargs["receive_id"] == "ou_user_3"
    assert call.kwargs["msg_type"] == "interactive"
    assert call.kwargs["uuid"]

    content = _sent_content(call)
    assert "card" not in content
    assert content["schema"] == "2.0"
    assert content["config"]["update_multi"]
    assert content["body"]["padding"] == "12px 12px 12px 12px"
    assert content["body"]["elements"][0]["text_size"] == "heading"
    assert content["body"]["elements"][0]["tag"] == "markdown"
    button = content["body"]["elements"][-1]["columns"][0]["elements"][0]
    assert button["tag"] == "button"
    assert "value" not in button
    assert button["behaviors"] == [{"type": "callback", "value": {"callback_data": "confirm"}}]


def test_send_notification_returns_failure_when_api_rejects(client, api):
    """
    飞书接口拒绝通知时返回失败结果。
    """
    api.create_message.return_value = FeishuResponse(code=230002, msg="bot not in chat")

    result = client.send_notification(Message(title="标题", text="正文"), userid="ou_user_3")

    assert result == {"success": False}


def test_send_notification_keeps_markdown_images_for_normal_card(client, api):
    """
    普通卡片保留正文中的 Markdown 图片语法。
    """
    api.create_message.return_value = _message_ok()

    result = client.send_notification(
        Message(
            title="普通通知",
            text="海报：![poster](https://example.com/poster.jpg)",
        ),
        userid="ou_user_img_md",
    )

    assert result["success"]
    content = _sent_content(api.create_message.call_args)
    assert content["body"]["elements"][1]["content"] == "海报：![poster](https://example.com/poster.jpg)"


def test_send_notification_embeds_remote_image_in_card(client, api):
    """
    远程图片下载上传后嵌入卡片，并清理临时文件。
    """
    uploaded = {}

    def _upload_image(file_path: Path):
        uploaded["path"] = file_path
        uploaded["bytes"] = file_path.read_bytes()
        return _ok(image_key="img_v2_remote")

    api.upload_image.side_effect = _upload_image
    api.create_message.return_value = _message_ok()
    response = _image_response()

    with patch("app.modules.feishu.feishu.RequestUtils") as request_utils:
        request_utils.return_value.get_res.return_value = response
        result = client.send_notification(
            Message(
                title="测试标题",
                text="测试正文",
                image="https://example.com/poster.png",
                buttons=[[{"text": "确认", "callback_data": "confirm"}]],
            ),
            userid="ou_user_img",
        )

    assert result["success"]
    response.close.assert_called_once()
    api.upload_image.assert_called_once()
    assert uploaded["bytes"] == b"png-bytes"
    assert uploaded["path"].suffix == ".png"
    # 下载到的临时图片上传后必须清理。
    assert not uploaded["path"].exists()
    call = api.create_message.call_args
    assert call.kwargs["msg_type"] == "interactive"
    content = _sent_content(call)
    assert content["body"]["padding"] == "0px 0px 0px 0px"
    image_element = content["body"]["elements"][0]
    assert image_element["tag"] == "img"
    assert image_element["img_key"] == "img_v2_remote"
    assert content["body"]["elements"][1]["margin"] == "12px 12px 0px 12px"
    assert content["body"]["elements"][2]["margin"] == "4px 12px 12px 12px"
    assert content["body"]["elements"][-1]["margin"] == "0px 12px 12px 12px"
    assert content["body"]["elements"][-1]["tag"] == "column_set"


def test_send_notification_supports_user_id_target(client, api):
    """
    通知接收目标支持飞书 user_id 标识类型。
    """
    api.create_message.return_value = _message_ok()

    client.send_notification(
        Message(title="测试标题", text="测试正文"),
        userid="u_user_4",
        receive_id_type="user_id",
    )

    call = api.create_message.call_args
    assert call.kwargs["receive_id_type"] == "user_id"
    assert call.kwargs["receive_id"] == "u_user_4"


def test_edit_message_uses_patch_api_for_cards(client, api):
    """
    普通卡片编辑通过消息更新接口提交完整卡片正文。
    """
    api.patch_message.return_value = _ok()

    success = client.edit_message(
        message_id="om_456",
        title="测试标题",
        text="测试正文",
        buttons=[[{"text": "确认", "callback_data": "confirm"}]],
    )

    assert success
    api.patch_message.assert_called_once()
    # 带按钮的普通卡片走 IM PATCH，不得误入 CardKit 流式更新。
    api.update_card_element_content.assert_not_called()

    message_id, card_json = api.patch_message.call_args.args
    assert message_id == "om_456"
    content = json.loads(card_json)
    assert "card" not in content
    assert content["schema"] == "2.0"
    assert content["config"]["update_multi"]
    assert content["body"]["elements"][0]["tag"] == "markdown"
    button = content["body"]["elements"][-1]["columns"][0]["elements"][0]
    assert button["behaviors"] == [{"type": "callback", "value": {"callback_data": "confirm"}}]


def test_send_notification_replies_when_original_message_id_is_present(client, api):
    """
    存在原消息标识时，通过回复接口发送通知卡片。
    """
    api.reply_message.return_value = _message_ok(message_id="om_reply")

    result = client.send_notification(
        Message(title="回复标题", text="回复正文"),
        userid="ou_user_9",
        original_message_id="om_origin",
    )

    assert result["success"]
    assert result["message_id"] == "om_reply"
    api.create_message.assert_not_called()
    api.reply_message.assert_called_once()
    call = api.reply_message.call_args
    assert call.kwargs["message_id"] == "om_origin"
    assert call.kwargs["msg_type"] == "interactive"
    assert call.kwargs["reply_in_thread"] is False
    assert _sent_content(call)["schema"] == "2.0"


def test_message_reaction_create_and_delete_use_official_api(client, api):
    """
    消息表情的添加与删除使用飞书官方接口。
    """
    api.create_message_reaction.return_value = _ok(reaction_id="reaction_1")
    api.delete_message_reaction.return_value = _ok()

    reaction_id = client.add_message_reaction("om_origin", Feishu.PROCESSING_REACTION_EMOJI)
    deleted = client.delete_message_reaction("om_origin", "reaction_1")

    assert reaction_id == "reaction_1"
    assert deleted
    api.create_message_reaction.assert_called_once_with("om_origin", Feishu.PROCESSING_REACTION_EMOJI)
    api.delete_message_reaction.assert_called_once_with("om_origin", "reaction_1")


def test_message_reaction_failures_return_empty_results(client, api):
    """
    消息表情操作失败时分别返回空标识或失败状态。
    """
    api.create_message_reaction.return_value = FeishuResponse(code=231001, msg="denied")
    api.delete_message_reaction.return_value = FeishuResponse(code=231001, msg="denied")

    assert client.add_message_reaction("om_origin", Feishu.PROCESSING_REACTION_EMOJI) is None
    assert client.delete_message_reaction("om_origin", "reaction_1") is False


# ---------------------------------------------------------------- 流式卡片


def test_send_notification_uses_streaming_card_for_agent_text(client, api):
    """
    Agent 文本通知创建流式卡片并返回后续更新所需元数据。
    """
    api.create_card.return_value = _ok(card_id="card_stream")
    api.create_message.return_value = _message_ok(message_id="om_stream", chat_id="oc_stream")

    result = client.send_notification(
        Message(mtype=MessageType.Agent, title="MoviePilot助手", text="第一帧内容"),
        userid="ou_user_stream",
    )

    assert result["success"]
    assert result["metadata"]["feishu_streaming"]["card_id"] == "card_stream"
    assert result["metadata"]["feishu_streaming"]["sequence"] == 0
    assert result["metadata"]["feishu_streaming"]["sent_image_urls"] == []
    card_payload = json.loads(api.create_card.call_args.args[0])
    assert card_payload["config"]["streaming_mode"]
    assert card_payload["body"]["elements"][-1]["element_id"] == Feishu.STREAM_CARD_BODY_ELEMENT_ID
    assert card_payload["body"]["elements"][-1]["content"] == "第一帧内容"
    call = api.create_message.call_args
    assert call.kwargs["msg_type"] == "interactive"
    assert _sent_content(call) == {"type": "card", "data": {"card_id": "card_stream"}}


def test_streaming_card_not_sent_when_card_creation_fails(client, api):
    """
    流式卡片创建失败时不继续发送消息。
    """
    api.create_card.return_value = FeishuResponse(code=200740, msg="invalid card")

    result = client.send_notification(
        Message(mtype=MessageType.Agent, title="MoviePilot助手", text="第一帧内容"),
        userid="ou_user_stream",
    )

    assert result == {"success": False}
    api.create_message.assert_not_called()


def test_streaming_card_sends_markdown_images_separately(client, api):
    """
    流式卡片将 Markdown 图片独立发送，正文保留可读替代文本。
    """
    api.create_card.return_value = _ok(card_id="card_stream")
    api.create_message.return_value = _message_ok(message_id="om_stream", chat_id="oc_stream")
    api.upload_image.return_value = _ok(image_key="img_v2_stream")

    with patch("app.modules.feishu.feishu.RequestUtils") as request_utils:
        request_utils.return_value.get_res.return_value = _image_response(content_type="image/jpeg")
        result = client.send_notification(
            Message(
                mtype=MessageType.Agent,
                title="MoviePilot助手",
                text="找到海报 ![poster](https://example.com/poster.jpg)\n[详情](https://example.com/detail)",
            ),
            userid="ou_user_stream",
        )

    assert result["success"]
    card_payload = json.loads(api.create_card.call_args.args[0])
    body_content = card_payload["body"]["elements"][-1]["content"]
    assert "![poster]" not in body_content
    assert "poster.jpg" not in body_content
    assert "poster" in body_content
    assert "[详情](https://example.com/detail)" in body_content
    assert api.upload_image.call_count == 1
    assert api.create_message.call_count == 2
    assert result["metadata"]["feishu_streaming"]["sent_image_urls"] == ["https://example.com/poster.jpg"]
    image_payload = _sent_content(api.create_message.call_args_list[-1])
    assert image_payload["body"]["elements"][0]["img_key"] == "img_v2_stream"


def test_streaming_card_sends_notification_image_separately(client, api):
    """
    流式通知的附带图片独立发送，并记录已发送地址。
    """
    api.create_card.return_value = _ok(card_id="card_stream")
    api.create_message.return_value = _message_ok(message_id="om_stream", chat_id="oc_stream")
    api.upload_image.return_value = _ok(image_key="img_v2_agent_image")

    with patch("app.modules.feishu.feishu.RequestUtils") as request_utils:
        request_utils.return_value.get_res.return_value = _image_response()
        result = client.send_notification(
            Message(
                mtype=MessageType.Agent,
                title="MoviePilot助手",
                text="第一帧内容",
                image="https://example.com/agent.png",
            ),
            userid="ou_user_stream",
        )

    assert result["success"]
    assert api.upload_image.call_count == 1
    assert api.create_message.call_count == 2
    assert result["metadata"]["feishu_streaming"]["sent_image_urls"] == ["https://example.com/agent.png"]
    stream_call, image_call = api.create_message.call_args_list
    assert _sent_content(stream_call)["data"]["card_id"] == "card_stream"
    assert _sent_content(image_call)["body"]["elements"][0]["img_key"] == "img_v2_agent_image"


def test_send_notification_replies_with_streaming_card_for_agent_text(client, api):
    """
    Agent 回复通过回复接口引用已创建的流式卡片。
    """
    api.create_card.return_value = _ok(card_id="card_stream")
    api.reply_message.return_value = _message_ok(message_id="om_reply", chat_id="oc_stream")

    result = client.send_notification(
        Message(mtype=MessageType.Agent, title="MoviePilot助手", text="第一帧内容"),
        userid="ou_user_stream",
        original_message_id="om_origin",
    )

    assert result["success"]
    api.create_message.assert_not_called()
    call = api.reply_message.call_args
    assert call.kwargs["message_id"] == "om_origin"
    assert call.kwargs["msg_type"] == "interactive"
    assert _sent_content(call)["data"]["card_id"] == "card_stream"
    assert result["metadata"]["feishu_streaming"]["sequence"] == 0


def test_edit_replied_streaming_card_uses_first_increment_sequence(client, api):
    """
    回复流式卡片的首次内容更新使用递增序号 1。
    """
    api.update_card_element_content.return_value = _ok()
    metadata = _stream_metadata(sent_image_urls=["https://example.com/poster.jpg"])

    success = client.edit_message(message_id="om_reply", text="补充内容", metadata=metadata)

    assert success
    api.patch_message.assert_not_called()
    assert api.update_card_element_content.call_args.kwargs["sequence"] == 1
    assert metadata["feishu_streaming"]["sequence"] == 1


def test_edit_message_uses_cardkit_content_for_streaming_card(client, api):
    """
    流式卡片编辑通过 CardKit 更新正文元素。
    """
    api.update_card_element_content.return_value = _ok()

    success = client.edit_message(
        message_id="om_stream",
        text="第二帧内容",
        metadata=_stream_metadata(sent_image_urls=["https://example.com/poster.jpg"]),
    )

    assert success
    api.update_card_element_content.assert_called_once()
    api.patch_message.assert_not_called()
    kwargs = api.update_card_element_content.call_args.kwargs
    assert kwargs["card_id"] == "card_stream"
    assert kwargs["element_id"] == Feishu.STREAM_CARD_BODY_ELEMENT_ID
    assert kwargs["content"] == "第二帧内容"
    assert kwargs["sequence"] == 1
    assert kwargs["uuid"]


def test_edit_streaming_card_sequence_keeps_increasing_after_failed_update(client, api, monkeypatch):
    """
    某次流式更新失败后，下一次仍须使用更大的 sequence，否则会被 CardKit 持续拒绝。
    """
    api.update_card_element_content.side_effect = [
        FeishuResponse(code=-1, msg="timeout"),
        _ok(),
    ]
    metadata = _stream_metadata(sequence=4, sent_image_urls=[])
    clock = [0.0]
    monkeypatch.setattr("app.modules.feishu.feishu.time.monotonic", lambda: clock[0])

    first = client.edit_message(message_id="om_stream", text="第一次", metadata=metadata)
    clock[0] += Feishu._STREAM_RETRY_INITIAL_SECONDS
    second = client.edit_message(message_id="om_stream", text="第二次", metadata=metadata)

    assert first is False
    assert second is True
    sequences = [call.kwargs["sequence"] for call in api.update_card_element_content.call_args_list]
    assert sequences == [5, 6]
    # 流式更新失败不得降级为普通卡片 PATCH。
    api.patch_message.assert_not_called()


@pytest.mark.parametrize("code", [200850, 300309])
def test_edit_streaming_card_reopens_and_retries_with_shared_sequence(client, api, code):
    """
    超过十分钟后的关闭错误应在同一卡片重开，每次请求使用新序号和 UUID。
    """
    api.update_card_element_content.side_effect = [
        FeishuResponse(code=code, msg="streaming mode is closed"), _ok(), _ok(),
    ]
    api.update_card_settings.return_value = _ok()
    metadata = _stream_metadata(sequence=480)

    assert client.edit_message("om_stream", text="完整累积内容", metadata=metadata)
    assert client.edit_message("om_stream", text="下一帧", metadata=metadata)

    requests = api.mock_calls
    assert [request[0] for request in requests] == [
        "update_card_element_content", "update_card_settings",
        "update_card_element_content", "update_card_element_content",
    ]
    assert [request.kwargs["sequence"] for request in requests] == [481, 482, 483, 484]
    assert len({request.kwargs["uuid"] for request in requests}) == 4
    assert {request.kwargs["card_id"] for request in requests} == {"card_stream"}
    assert json.loads(requests[1].kwargs["settings"]) == {"config": {"streaming_mode": True}}
    assert requests[2].kwargs["content"] == "完整累积内容"
    assert metadata["feishu_streaming"]["sequence"] == 484
    assert "retry_at" not in metadata["feishu_streaming"]
    api.patch_message.assert_not_called()


def test_edit_streaming_card_can_recover_on_later_expiration(client, api):
    """
    同一次长会话再次到期后仍可重开，收尾序号大于全部恢复请求。
    """
    closed = FeishuResponse(code=300309, msg="closed")
    api.update_card_element_content.side_effect = [closed, _ok(), closed, _ok()]
    api.update_card_settings.return_value = _ok()
    metadata = _stream_metadata()

    assert client.edit_message("om_stream", text="十分钟", metadata=metadata)
    assert client.edit_message("om_stream", text="二十分钟", metadata=metadata)
    assert client.close_streaming_card("card_stream", metadata["feishu_streaming"]["sequence"] + 1)

    assert [request.kwargs["sequence"] for request in api.mock_calls] == list(range(1, 8))
    assert json.loads(api.mock_calls[-1].kwargs["settings"]) == {"config": {"streaming_mode": False}}


@pytest.mark.parametrize("failure", ["settings", "content", "ordinary", "exception"])
def test_edit_streaming_card_failure_backs_off_and_recovers(client, api, monkeypatch, failure):
    """
    恢复失败或普通请求失败后，高频刷新不出站；到期重试最新文本并清除退避。
    """
    clock = [100.0]
    monkeypatch.setattr("app.modules.feishu.feishu.time.monotonic", lambda: clock[0])
    closed = FeishuResponse(code=300309, msg="closed")
    api.update_card_element_content.side_effect = {
        "settings": [closed],
        "content": [closed, closed],
        "ordinary": [FeishuResponse(code=-1, msg="timeout")],
        "exception": [RuntimeError("transport failure")],
    }[failure]
    api.update_card_settings.return_value = (
        _ok() if failure == "content" else FeishuResponse(code=-1, msg="timeout")
    )
    metadata = _stream_metadata(sequence=4)

    if failure == "exception":
        with pytest.raises(RuntimeError, match="transport failure"):
            client.edit_message("om_stream", text="未发送", metadata=metadata)
    else:
        assert not client.edit_message("om_stream", text="未发送", metadata=metadata)
    count = len(api.mock_calls)
    sequence = metadata["feishu_streaming"]["sequence"]
    assert count == {"settings": 2, "content": 3, "ordinary": 1, "exception": 1}[failure]
    for _ in range(100):
        assert not client.edit_message("om_stream", text="更多内容", metadata=metadata)
    assert len(api.mock_calls) == count
    assert metadata["feishu_streaming"]["sequence"] == sequence

    api.update_card_element_content.side_effect = None
    api.update_card_element_content.return_value = _ok()
    clock[0] += Feishu._STREAM_RETRY_INITIAL_SECONDS
    assert client.edit_message("om_stream", text="最新完整内容", metadata=metadata)
    assert api.update_card_element_content.call_args.kwargs["content"] == "最新完整内容"
    assert metadata["feishu_streaming"]["sequence"] == sequence + 1
    assert "retry_at" not in metadata["feishu_streaming"]
    assert "retry_delay" not in metadata["feishu_streaming"]
    api.patch_message.assert_not_called()


def test_edit_streaming_card_backoff_is_capped_and_card_local(client, api, monkeypatch):
    """
    持续失败的重试间隔递增到一分钟，不影响其他卡片。
    """
    clock = [0.0]
    monkeypatch.setattr("app.modules.feishu.feishu.time.monotonic", lambda: clock[0])
    api.update_card_element_content.return_value = FeishuResponse(code=-1, msg="timeout")
    metadata = _stream_metadata()
    with patch("app.modules.feishu.feishu.logger.warning") as warning:
        for delay in [2, 4, 8, 16, 32, 60, 60]:
            assert not client.edit_message("om_stream", text="待发送", metadata=metadata)
            assert metadata["feishu_streaming"]["retry_at"] == clock[0] + delay
            count = api.update_card_element_content.call_count
            clock[0] += delay - 0.1
            assert not client.edit_message("om_stream", text="待发送", metadata=metadata)
            assert api.update_card_element_content.call_count == count
            clock[0] += 0.1
        assert warning.call_count == 7

    api.update_card_element_content.return_value = _ok()
    assert client.edit_message("om_other", text="独立卡片", metadata=_stream_metadata(card_id="other"))
    assert api.update_card_element_content.call_args.kwargs["card_id"] == "other"
    api.update_card_settings.assert_not_called()
    api.patch_message.assert_not_called()


@pytest.mark.asyncio
@pytest.mark.parametrize("recover", [False, True])
async def test_agent_flush_preserves_unsent_text_during_feishu_backoff(client, api, monkeypatch, recover):
    """
    真实刷新链保留失败文本和共享序号；收尾只在全部送达后报告成功。
    """
    clock = [100.0]
    monkeypatch.setattr("app.modules.feishu.feishu.time.monotonic", lambda: clock[0])
    api.update_card_element_content.return_value = FeishuResponse(code=300309, msg="closed")
    api.update_card_settings.return_value = FeishuResponse(code=-1, msg="timeout")
    handler = StreamingHandler()
    handler._streaming_enabled = True
    handler._channel = NotificationChannel.Feishu.value
    handler._source = "feishu-test"
    handler._buffer = "完整回复"
    handler._sent_text = "旧回复"
    handler._message_response = MessageResponse(
        channel=NotificationChannel.Feishu, source="feishu-test", message_id="om_stream",
        chat_id="oc_test", success=True, metadata=_stream_metadata(),
    )
    response = handler._message_response
    chain = MagicMock()
    chain.edit_message.side_effect = lambda **kwargs: client.edit_message(
        kwargs["message_id"], text=kwargs["text"], metadata=kwargs["metadata"],
    )
    chain.finalize_message.side_effect = lambda result: client.close_streaming_card(
        result.metadata["feishu_streaming"]["card_id"],
        result.metadata["feishu_streaming"]["sequence"] + 1,
    )
    with patch("app.agent.callback._StreamChain", return_value=chain):
        await handler._flush()
        for _ in range(10):
            await handler._flush()
        assert api.update_card_element_content.call_count == 1
        assert handler._sent_text == "旧回复"
        assert response.metadata["feishu_streaming"]["sequence"] == 2

        if recover:
            clock[0] += Feishu._STREAM_RETRY_INITIAL_SECONDS
            api.update_card_element_content.return_value = _ok()
        api.update_card_settings.return_value = _ok()
        all_sent, final_text = await handler.stop_streaming()

    assert all_sent is recover
    assert final_text == ("完整回复" if recover else "")
    if not recover:
        assert await handler.take() == "完整回复"
    assert api.update_card_settings.call_args.kwargs["sequence"] == (4 if recover else 3)
    api.patch_message.assert_not_called()


def test_edit_streaming_card_removes_markdown_image_syntax(client, api):
    """
    流式内容更新移除 Markdown 图片语法，保留图片替代文本。
    """
    api.update_card_element_content.return_value = _ok()

    success = client.edit_message(
        message_id="om_stream",
        text="第二帧 ![poster](https://example.com/poster.jpg)",
        metadata=_stream_metadata(sent_image_urls=["https://example.com/poster.jpg"]),
    )

    assert success
    api.patch_message.assert_not_called()
    assert api.update_card_element_content.call_args.kwargs["content"] == "第二帧 poster"


def test_edit_streaming_card_keeps_normal_markdown_links(client, api):
    """
    流式内容更新保留普通 Markdown 链接。
    """
    api.update_card_element_content.return_value = _ok()

    success = client.edit_message(
        message_id="om_stream",
        text="第二帧 [详情](https://example.com/detail)",
        chat_id="oc_stream",
        metadata=_stream_metadata(sent_image_urls=[]),
    )

    assert success
    api.patch_message.assert_not_called()
    api.upload_image.assert_not_called()
    assert api.update_card_element_content.call_args.kwargs["content"] == "第二帧 [详情](https://example.com/detail)"


def test_edit_streaming_card_hides_incomplete_markdown_image(client, api):
    """
    尚未生成完整的 Markdown 图片链接不在卡片正文展示。
    """
    api.update_card_element_content.return_value = _ok()

    success = client.edit_message(
        message_id="om_stream",
        text="第二帧 ![poster](https://example.com/poster",
        chat_id="oc_stream",
        metadata=_stream_metadata(sent_image_urls=[]),
    )

    assert success
    api.patch_message.assert_not_called()
    api.upload_image.assert_not_called()
    assert api.update_card_element_content.call_args.kwargs["content"] == "第二帧"


def test_edit_streaming_card_hides_incomplete_markdown_image_alt_text(client, api):
    """
    尚未生成完整的图片替代文本同样暂不展示。
    """
    api.update_card_element_content.return_value = _ok()

    success = client.edit_message(
        message_id="om_stream",
        text="第二帧 ![poster",
        chat_id="oc_stream",
        metadata=_stream_metadata(sent_image_urls=[]),
    )

    assert success
    api.patch_message.assert_not_called()
    api.upload_image.assert_not_called()
    assert api.update_card_element_content.call_args.kwargs["content"] == "第二帧"


def test_edit_streaming_card_sends_completed_markdown_image_once(client, api):
    """
    分块输出补齐的图片只发送一次，后续更新不重复上传。
    """
    api.update_card_element_content.return_value = _ok()
    api.create_message.return_value = _message_ok(message_id="om_img", chat_id="oc_stream")
    api.upload_image.return_value = _ok(image_key="img_v2_stream_edit")
    metadata = _stream_metadata(sent_image_urls=[])

    with patch("app.modules.feishu.feishu.RequestUtils") as request_utils:
        request_utils.return_value.get_res.return_value = _image_response(b"jpg-bytes", "image/jpeg")
        first_success = client.edit_message(
            message_id="om_stream",
            text="第二帧 ![poster](https://example.com/poster.jpg)",
            chat_id="oc_stream",
            metadata=metadata,
        )
        second_success = client.edit_message(
            message_id="om_stream",
            text="第二帧 ![poster](https://example.com/poster.jpg)",
            chat_id="oc_stream",
            metadata=metadata,
        )

    assert first_success
    assert second_success
    assert api.upload_image.call_count == 1
    assert metadata["feishu_streaming"]["sent_image_urls"] == ["https://example.com/poster.jpg"]
    api.create_message.assert_called_once()
    call = api.create_message.call_args
    assert call.kwargs["receive_id"] == "oc_stream"
    assert call.kwargs["receive_id_type"] == "chat_id"
    assert _sent_content(call)["body"]["elements"][0]["img_key"] == "img_v2_stream_edit"


def test_edit_streaming_card_skips_non_image_markdown_target(client, api):
    """
    Markdown 图片目标并非有效图片时不记录为已发送图片。
    """
    api.update_card_element_content.return_value = _ok()
    metadata = _stream_metadata(sent_image_urls=[])

    with patch("app.modules.feishu.feishu.RequestUtils") as request_utils:
        request_utils.return_value.get_res.return_value = _image_response(b"<html></html>", "text/html")
        success = client.edit_message(
            message_id="om_stream",
            text="第二帧 ![link](https://example.com/detail)",
            chat_id="oc_stream",
            metadata=metadata,
        )

    assert success
    api.upload_image.assert_not_called()
    assert metadata["feishu_streaming"]["sent_image_urls"] == []
    api.create_message.assert_not_called()


def test_close_streaming_card_updates_card_settings(client, api):
    """
    关闭流式卡片时更新设置以退出流式模式。
    """
    api.update_card_settings.return_value = _ok()

    success = client.close_streaming_card(card_id="card_stream", sequence=3)

    assert success
    kwargs = api.update_card_settings.call_args.kwargs
    assert kwargs["card_id"] == "card_stream"
    assert kwargs["sequence"] == 3
    assert json.loads(kwargs["settings"]) == {"config": {"streaming_mode": False}}


def test_close_streaming_card_reports_failure(client, api):
    """
    流式卡片关闭失败时向调用方返回 False。
    """
    api.update_card_settings.return_value = FeishuResponse(code=300309, msg="sequence")

    assert client.close_streaming_card(card_id="card_stream", sequence=3) is False


# ---------------------------------------------------------------- 文件、语音与下载


def test_feishu_channel_capabilities_enable_images_and_files():
    """
    飞书渠道声明支持图片与文件发送能力。
    """
    assert ChannelCapabilityManager.supports_capability(
        NotificationChannel.Feishu,
        ChannelCapability.IMAGES,
    )
    assert ChannelCapabilityManager.supports_capability(
        NotificationChannel.Feishu,
        ChannelCapability.FILE_SENDING,
    )


def test_send_file_uploads_image_then_sends_mixed_card(client, api):
    """
    图片文件上传后以包含图片及说明的卡片发送。
    """
    api.upload_image.return_value = _ok(image_key="img_v2_uploaded")
    api.create_message.return_value = _message_ok(message_id="om_image")

    with tempfile.NamedTemporaryFile(suffix=".png") as fp:
        fp.write(b"png-bytes")
        fp.flush()
        result = client.send_file(
            file_path=fp.name,
            userid="ou_user_7",
            title="图片标题",
            text="图片说明",
        )

    assert result["success"]
    api.upload_image.assert_called_once_with(Path(fp.name))
    api.upload_file.assert_not_called()
    call = api.create_message.call_args
    assert call.kwargs["msg_type"] == "interactive"
    content = _sent_content(call)
    assert content["body"]["padding"] == "0px 0px 0px 0px"
    assert content["body"]["elements"][0]["img_key"] == "img_v2_uploaded"
    assert content["body"]["elements"][1]["content"] == "图片标题"
    assert content["body"]["elements"][1]["margin"] == "12px 12px 0px 12px"
    assert content["body"]["elements"][2]["content"] == "图片说明"
    assert content["body"]["elements"][2]["margin"] == "4px 12px 12px 12px"


def test_send_file_keeps_non_image_file_message_and_caption(client, api):
    """
    非图片文件按文件消息发送，并保留标题和说明。
    """
    api.upload_file.return_value = _ok(file_key="file_doc")
    api.create_message.return_value = _message_ok(message_id="om_file")

    with (
        tempfile.NamedTemporaryFile(suffix=".txt") as fp,
        patch.object(client, "send_text", return_value={"success": True}) as send_text,
    ):
        fp.write(b"text-bytes")
        fp.flush()
        result = client.send_file(
            file_path=fp.name,
            userid="ou_user_7",
            title="文件标题",
            text="文件说明",
        )

    assert result["success"]
    api.upload_file.assert_called_once_with(
        Path(fp.name),
        file_type="stream",
        file_name=Path(fp.name).name,
        duration=None,
    )
    call = api.create_message.call_args
    assert call.kwargs["msg_type"] == "file"
    assert _sent_content(call) == {"file_key": "file_doc"}
    send_text.assert_called_once()
    assert "文件标题" in send_text.call_args.args[0]
    assert "文件说明" in send_text.call_args.args[0]


def test_send_file_fails_without_sending_when_upload_rejected(client, api):
    """
    文件上传被拒绝时返回失败，不再发送消息。
    """
    api.upload_file.return_value = FeishuResponse(code=234001, msg="too large")

    with tempfile.NamedTemporaryFile(suffix=".pdf") as fp:
        fp.write(b"pdf-bytes")
        fp.flush()
        result = client.send_file(file_path=fp.name, userid="ou_user_7")

    assert result == {"success": False}
    assert api.upload_file.call_args.kwargs["file_type"] == "pdf"
    api.create_message.assert_not_called()


def test_send_voice_uploads_audio_file_and_optionally_sends_caption(client, api):
    """
    语音文件上传后按音频消息发送，并按需发送说明。
    """
    api.upload_file.return_value = _ok(file_key="file_audio")
    api.create_message.return_value = _message_ok(message_id="om_audio")

    with tempfile.NamedTemporaryFile(suffix=".opus") as fp:
        fp.write(b"opus-bytes")
        fp.flush()
        with patch.object(client, "send_text", return_value={"success": True}) as send_text:
            result = client.send_voice(
                voice_path=fp.name,
                userid="ou_user_8",
                caption="这是说明",
            )

    assert result["success"]
    assert api.upload_file.call_args.kwargs["file_type"] == "opus"
    call = api.create_message.call_args
    assert call.kwargs["msg_type"] == "audio"
    assert _sent_content(call) == {"file_key": "file_audio"}
    send_text.assert_called_once()
    assert send_text.call_args.args[0] == "这是说明"


def test_download_helpers_return_bytes_and_metadata(client, api):
    """
    图片、文件与消息资源下载返回字节内容、名称和媒体类型。
    """
    api.download_image.return_value = (b"image-bytes", "poster.png", "image/png")
    api.download_file.return_value = (b"file-bytes", "report.txt", "text/plain")
    api.download_message_resource.return_value = (b"resource-bytes", "voice.opus", "audio/ogg")

    image_download = client.download_image_bytes("img_v2_test")
    file_download = client.download_file_bytes("file_test")
    resource_download = client.download_message_resource_bytes("om_test", "file_test", "audio")

    assert image_download == (b"image-bytes", "poster.png", "image/png")
    assert file_download == (b"file-bytes", "report.txt", "text/plain")
    assert resource_download == (b"resource-bytes", "voice.opus", "audio/ogg")
    api.download_image.assert_called_once_with("img_v2_test")
    api.download_file.assert_called_once_with("file_test")
    api.download_message_resource.assert_called_once_with("om_test", "file_test", "audio")


def test_download_helpers_skip_api_for_missing_keys(client, api):
    """
    资源标识缺失时直接返回空结果，不调用飞书接口。
    """
    assert client.download_image_bytes("") is None
    assert client.download_file_bytes("") is None
    assert client.download_message_resource_bytes("", "file_test", "audio") is None
    api.download_image.assert_not_called()
    api.download_file.assert_not_called()
    api.download_message_resource.assert_not_called()


# ---------------------------------------------------------------- 广播与目标解析


def test_send_notification_broadcasts_to_remembered_chats_without_target(client, api):
    """
    无显式目标且未配置默认目标时，应回退向最近互动过的会话发送。
    """
    api.create_message.return_value = _message_ok()
    client._user_chat_mapping = {
        "ou_user_a": "oc_group_a",
        "ou_user_b": "oc_group_a",
        "ou_user_c": "oc_p2p_c",
    }

    result = client.send_notification(Message(title="插件通知", text="无目标通知"))

    assert result["success"]
    assert api.create_message.call_count == 2
    calls = api.create_message.call_args_list
    assert sorted(call.kwargs["receive_id"] for call in calls) == ["oc_group_a", "oc_p2p_c"]
    assert all(call.kwargs["receive_id_type"] == "chat_id" for call in calls)
    # 每次发送携带独立幂等 uuid，避免飞书把不同会话的消息判为重复请求。
    assert len({call.kwargs["uuid"] for call in calls}) == 2


def test_send_without_target_raises_config_hint_when_nothing_available(client, api):
    """
    既无目标又无历史互动时，应报出含配置指引的明确错误。
    """
    api.create_message.return_value = _message_ok()

    with pytest.raises(ValueError) as exc_info:
        client._send_with_fallback_broadcast("text", {"text": "hello"})

    assert "FEISHU_OPEN_ID" in str(exc_info.value)
    assert "FEISHU_CHAT_ID" in str(exc_info.value)
    api.create_message.assert_not_called()


# ---------------------------------------------------------------- FeishuModule


def test_module_send_direct_message_prefers_open_id_target():
    """
    模块直接发送消息时优先使用 open_id 目标。
    """
    module = FeishuModule()
    module._channel = NotificationChannel.Feishu
    conf = SimpleNamespace(name="feishu-main")
    client = MagicMock()
    client.send_notification.return_value = {
        "success": True,
        "message_id": "om_789",
        "chat_id": "oc_789",
    }

    with (
        patch.object(module, "get_configs", return_value={"feishu-main": conf}),
        patch.object(module, "check_message", return_value=True),
        patch.object(module, "get_instance", return_value=client),
    ):
        response = module.send_direct_message(
            Message(
                targets={
                    "feishu_userid": "u_target",
                    "feishu_openid": "ou_target",
                }
            )
        )

    client.send_notification.assert_called_once_with(
        message=ANY,
        userid="ou_target",
        chat_id=None,
        receive_id_type="open_id",
        original_message_id=None,
    )
    assert response.success
    assert response.message_id == "om_789"
    assert response.chat_id == "oc_789"


def test_module_plain_direct_message_uses_literal_text_transport():
    """
    纯文本直发不得进入会解释密钥字符的 Markdown 卡片路径。
    """
    module = FeishuModule()
    module._channel = NotificationChannel.Feishu
    conf = SimpleNamespace(name="feishu-main")
    client = MagicMock()
    client.send_text.return_value = {
        "success": True,
        "message_id": "om_plain",
        "chat_id": "oc_plain",
    }
    literal_text = "G2A1_PROTECTED_MARKER_20260812\n**literal markdown**\n<img src=x>"

    with (
        patch.object(module, "get_configs", return_value={"feishu-main": conf}),
        patch.object(module, "check_message", return_value=True),
        patch.object(module, "get_instance", return_value=client),
    ):
        response = module.send_direct_message(
            Message(
                channel=NotificationChannel.Feishu,
                source="feishu-main",
                userid="ou_target",
                text=literal_text,
                private_delivery=True,
                parse_mode="plain",
            )
        )

    client.send_text.assert_called_once_with(
        text=literal_text,
        userid="ou_target",
        chat_id=None,
        receive_id_type=None,
        original_message_id=None,
    )
    client.send_notification.assert_not_called()
    assert response.success


def test_module_download_helpers_delegate_to_client():
    """
    模块下载入口委托飞书客户端，并转换为所需字节或数据 URL。
    """
    module = FeishuModule()
    client = MagicMock()
    client.download_image_bytes.return_value = (b"image", "poster.png", "image/png")
    client.download_file_bytes.return_value = (b"file", "note.txt", "text/plain")
    client.download_message_resource_bytes.return_value = (b"image", "poster.png", "image/png")

    with (
        patch.object(module, "get_config", return_value=SimpleNamespace(name="feishu-main")),
        patch.object(module, "get_instance", return_value=client),
    ):
        data_url = module.download_feishu_image_to_data_url("feishu://image/om_msg/img_v2_xxx", "feishu-main")
        file_bytes = module.download_feishu_file_bytes("feishu://file/file_xxx/note.txt", "feishu-main")
        audio_bytes = module.download_feishu_file_bytes(
            "feishu://file/om_audio/file_audio/voice.opus",
            "feishu-main",
        )

    assert data_url.startswith("data:image/png;base64,")
    assert file_bytes == b"file"
    assert audio_bytes == b"image"
    client.download_message_resource_bytes.assert_any_call(
        message_id="om_msg",
        file_key="img_v2_xxx",
        resource_type="image",
    )
    client.download_message_resource_bytes.assert_any_call(
        message_id="om_audio",
        file_key="file_audio",
        resource_type="audio",
    )


def test_module_message_reaction_helpers_delegate_to_client():
    """
    模块消息表情入口委托对应客户端操作。
    """
    module = FeishuModule()
    client = MagicMock()
    client.add_message_reaction.return_value = "reaction_2"
    client.delete_message_reaction.return_value = True

    with (
        patch.object(module, "get_config", return_value=SimpleNamespace(name="feishu-main")),
        patch.object(module, "get_instance", return_value=client),
    ):
        reaction_id = module.add_feishu_message_reaction("om_x", "GLANCE", "feishu-main")
        deleted = module.delete_feishu_message_reaction("om_x", "reaction_2", "feishu-main")

    assert reaction_id == "reaction_2"
    assert deleted


def test_module_processing_status_uses_reaction_helpers():
    """
    处理中状态通过表情入口创建和移除，并保存表情标识。
    """
    module = FeishuModule()
    module._channel = NotificationChannel.Feishu

    with (
        patch.object(module, "add_feishu_message_reaction", return_value="reaction_processing") as add_reaction,
        patch.object(module, "delete_feishu_message_reaction", return_value=True) as delete_reaction,
    ):
        status = module.mark_message_processing_started(
            channel=NotificationChannel.Feishu,
            source="feishu-main",
            userid="ou_x",
            message_id="om_x",
            chat_id="oc_x",
            text="hello",
        )
        deleted = module.mark_message_processing_finished(
            channel=NotificationChannel.Feishu,
            source="feishu-main",
            userid="ou_x",
            status=status,
        )

    add_reaction.assert_called_once_with(
        message_id="om_x",
        emoji_type="GLANCE",
        source="feishu-main",
    )
    delete_reaction.assert_called_once_with(
        message_id="om_x",
        reaction_id="reaction_processing",
        source="feishu-main",
    )
    assert status["metadata"]["reaction_id"] == "reaction_processing"
    assert deleted


def test_module_finalize_message_closes_streaming_card():
    """
    消息完成入口使用元数据关闭对应流式卡片。
    """
    module = FeishuModule()
    module._channel = NotificationChannel.Feishu
    client = MagicMock()
    client.close_streaming_card.return_value = True

    with (
        patch.object(module, "get_config", return_value=SimpleNamespace(name="feishu-main")),
        patch.object(module, "get_instance", return_value=client),
    ):
        success = module.finalize_message(
            MessageResponse(
                message_id="om_stream",
                chat_id="oc_stream",
                channel=NotificationChannel.Feishu,
                source="feishu-main",
                metadata={"feishu_streaming": {"card_id": "card_stream", "sequence": 2}},
                success=True,
            )
        )

    assert success
    client.close_streaming_card.assert_called_once_with(card_id="card_stream", sequence=3)


def test_module_post_message_prefers_file_and_voice_paths():
    """
    模块发送入口优先按文件或语音载荷选择专用发送路径。
    """
    module = FeishuModule()
    conf = SimpleNamespace(name="feishu-main")
    client = MagicMock()

    with (
        patch.object(module, "get_configs", return_value={"feishu-main": conf}),
        patch.object(module, "check_message", return_value=True),
        patch.object(module, "get_instance", return_value=client),
    ):
        module.post_message(
            Message(
                file_path="/tmp/demo.txt",
                text="说明",
                title="标题",
                userid="ou_user",
            )
        )
        module.post_message(
            Message(
                voice_path="/tmp/demo.opus",
                voice_caption="语音说明",
                userid="ou_user",
            )
        )

    client.send_file.assert_called_once()
    client.send_voice.assert_called_once()


def test_module_post_message_sends_image_card_before_file_attachment():
    """
    通知含图片和附件时先发送图片卡片，再发送不重复说明的附件。
    """
    module = FeishuModule()
    conf = SimpleNamespace(name="feishu-main")
    client = MagicMock()

    with (
        patch.object(module, "get_configs", return_value={"feishu-main": conf}),
        patch.object(module, "check_message", return_value=True),
        patch.object(module, "get_instance", return_value=client),
    ):
        module.post_message(
            Message(
                file_path="/tmp/demo.txt",
                file_name="demo.txt",
                image="https://example.com/poster.png",
                text="说明",
                title="标题",
                userid="ou_user",
            )
        )

    client.send_notification.assert_called_once()
    sent_message = client.send_notification.call_args.kwargs["message"]
    assert sent_message.image == "https://example.com/poster.png"
    assert sent_message.file_path is None
    client.send_file.assert_called_once()
    assert client.send_file.call_args.kwargs.get("title") is None
    assert client.send_file.call_args.kwargs.get("text") is None


def test_module_send_direct_message_sends_image_card_before_file_attachment():
    """
    直接消息含图片和附件时先发送卡片，并返回卡片消息标识。
    """
    module = FeishuModule()
    module._channel = NotificationChannel.Feishu
    conf = SimpleNamespace(name="feishu-main")
    client = MagicMock()
    client.send_notification.return_value = {
        "success": True,
        "message_id": "om_card",
        "chat_id": "oc_card",
    }
    client.send_file.return_value = {"success": True, "message_id": "om_file"}

    with (
        patch.object(module, "get_configs", return_value={"feishu-main": conf}),
        patch.object(module, "check_message", return_value=True),
        patch.object(module, "get_instance", return_value=client),
    ):
        response = module.send_direct_message(
            Message(
                channel=NotificationChannel.Feishu,
                source="feishu-main",
                file_path="/tmp/demo.txt",
                file_name="demo.txt",
                image="https://example.com/poster.png",
                text="说明",
                title="标题",
                userid="ou_user",
            )
        )

    assert response.success
    assert response.message_id == "om_card"
    client.send_notification.assert_called_once()
    client.send_file.assert_called_once()


def test_module_post_message_passes_original_message_id_for_reply():
    """
    模块将原消息标识传给客户端，保留回复语义。
    """
    module = FeishuModule()
    conf = SimpleNamespace(name="feishu-main")
    client = MagicMock()

    with (
        patch.object(module, "get_configs", return_value={"feishu-main": conf}),
        patch.object(module, "check_message", return_value=True),
        patch.object(module, "get_instance", return_value=client),
    ):
        module.post_message(
            Message(
                title="标题",
                text="正文",
                userid="ou_user",
                original_message_id="om_source",
                original_chat_id="oc_source",
            )
        )

    client.send_notification.assert_called_once()
    assert client.send_notification.call_args.kwargs["original_message_id"] == "om_source"


def test_module_resolve_message_target_prefers_original_chat_id():
    """
    群聊@回复必须优先发回原会话，而不是按 open_id 发到私聊。
    """
    userid, chat_id, receive_id_type = FeishuModule._resolve_message_target(
        Message(userid="ou_user", original_chat_id="oc_group")
    )
    assert userid is None
    assert chat_id == "oc_group"
    assert receive_id_type == "chat_id"

    # 非回复类消息仍按原有逻辑优先 open_id。
    userid, chat_id, receive_id_type = FeishuModule._resolve_message_target(Message(userid="ou_user"))
    assert userid == "ou_user"
    assert chat_id is None
    assert receive_id_type == "open_id"

    # 无用户ID时回退 targets 中的飞书字段。
    userid, chat_id, receive_id_type = FeishuModule._resolve_message_target(
        Message(targets={"feishu_chat_id": "oc_config"})
    )
    assert userid is None
    assert chat_id == "oc_config"


def test_module_private_delivery_ignores_original_group_chat():
    """
    私聊投递只保留用户身份，并让客户端按已记录 ID 类型发送。
    """
    userid, chat_id, receive_id_type = FeishuModule._resolve_message_target(
        Message(
            userid="user_target",
            original_chat_id="oc_group",
            private_delivery=True,
        )
    )

    assert userid == "user_target"
    assert chat_id is None
    assert receive_id_type is None


def test_module_post_message_replies_to_original_chat_for_group_message():
    """
    携带原会话上下文的回复应定向到原会话（群聊）。
    """
    module = FeishuModule()
    conf = SimpleNamespace(name="feishu-main")
    client = MagicMock()

    with (
        patch.object(module, "get_configs", return_value={"feishu-main": conf}),
        patch.object(module, "check_message", return_value=True),
        patch.object(module, "get_instance", return_value=client),
    ):
        module.post_message(
            Message(
                title="标题",
                text="正文",
                userid="ou_user",
                original_chat_id="oc_group",
            )
        )

    client.send_notification.assert_called_once_with(
        message=ANY,
        userid=None,
        chat_id="oc_group",
        receive_id_type="chat_id",
        original_message_id=None,
    )