"""点歌业务测试：假 MusicServer + Recording BotApi，覆盖选歌菜单与分页、
歌曲卡片、按钮回调、默认 tx 源与聚合兜底、直链候选源回退、错误路径。"""

from collections.abc import AsyncIterator
from dataclasses import replace
from pathlib import Path
from typing import Any
from urllib.parse import quote

import pytest
from aiohttp import ClientSession, web
from qqbotsdk import EventEmitter
from qqbotsdk.api import ApiError, BotApi

from musicbot import music
from musicbot.config import BotConfig
from musicbot.music import (
    handle_button,
    handle_command,
    register,
    search,
    send_menu,
)
from musicbot.panel import sync_command_panel


def make_api(
    fail_types: set[int] | None = None,
    fail_md: bool = False,
    fail_event: bool = False,
    fail_upload: bool = False,
    fail_file_perm: bool = False,
    fail_keyboard: bool = False,
):
    """替换 upload_group_file / Markdown 消息以模拟失败，request 记录其余调用。

    fail_event=True 时拒绝一切带 event_id 的被动回复（模拟互动回执失效）；
    fail_upload=True 时上传一律抛 TimeoutError（模拟 QQ 侧抓流超过请求超时）；
    fail_file_perm=True 时 file_type=4（文件）返回 40093005（仅群主/管理员可传）；
    fail_keyboard=True 时拒绝一切带 keyboard 的消息（模拟键盘无权限）。
    """
    calls: list[dict] = []

    class Recording(BotApi):
        async def request(self, method: str, path: str, **kwargs: Any) -> Any:
            body = kwargs.get("json", {})
            if fail_md and body.get("msg_type") == 2:
                raise ApiError(400, 304036, "无 Markdown 权限")
            if fail_keyboard and "keyboard" in body:
                raise ApiError(400, 304003, "无键盘权限")
            if fail_event and "event_id" in body:
                raise ApiError(400, 40034025, "请求参数event_id无效")
            if path.endswith("/files"):
                if fail_upload:
                    raise TimeoutError()
                if fail_file_perm and body.get("file_type") == 4:
                    raise ApiError(400, 40093005, "仅允许群主和管理员上传文件")
                if fail_types and body.get("file_type") in fail_types:
                    raise ApiError(400, "40054002", "上传失败")
                calls.append(
                    {
                        "method": method,
                        "path": path,
                        "file_type": body.get("file_type"),
                        "url": body.get("url"),
                        "timeout": kwargs.get("timeout"),
                    }
                )
                return {"file_info": f"f{body.get('file_type')}"}
            calls.append({"method": method, "path": path, **kwargs})
            return {"id": "MSG"} if path.endswith("/messages") else {}

    return Recording.__new__(Recording), calls


class FakeMusicServer:
    """模拟 MusicServer 的统一信封接口，记录搜索参数。

    /url 返回指向自身的直链；tx_denied=True 时 tx 的直链指向 403
    路由，模拟 CDN 风控（测直链探测与候选源回退）。
    """

    def __init__(self, tx_denied: bool = False) -> None:
        self.tx_denied = tx_denied
        self.runner: web.AppRunner | None = None
        self.port = 0
        self.queries: list[dict] = []

    async def start(self) -> None:
        app = web.Application()
        app.router.add_get("/search", self._search)
        app.router.add_get("/url/{source}/{song_id}", self._url)
        app.router.add_get("/lyric/{source}/{song_id}", self._lyric)
        app.router.add_get("/media/{tail:.+}", self._media)
        app.router.add_get("/denied/{tail:.+}", self._denied)
        self.runner = web.AppRunner(app)
        await self.runner.setup()
        site = web.TCPSite(self.runner, "127.0.0.1", 0)
        await site.start()
        # sockets 未列入 aiohttp 的类型存根，只能从私有属性取实际端口
        assert site._server is not None
        self.port = site._server.sockets[0].getsockname()[1]  # type: ignore[attr-defined]

    async def stop(self) -> None:
        assert self.runner is not None
        await self.runner.cleanup()

    async def _search(self, request: web.Request) -> web.Response:
        name = request.query["name"]
        source = request.query.get("source", "")
        self.queries.append({"name": name, "source": source})
        items: list[dict] = []
        if "空" not in name:
            if "兜底" in name:
                # tx 搜不到、聚合有 —— 测兜底路径
                items = [] if source == "tx" else [
                    {"source": "kw", "songmid": "S1", "name": name, "singer": "歌手A"}
                ]
            elif source == "kw":
                items = [
                    {"source": "kw", "songmid": "S1", "name": name, "singer": "歌手A"}
                ]
            elif source == "tx" or not source:
                items = [
                    {
                        "source": "tx",
                        "songmid": "T1",
                        "name": name,
                        "singer": "歌手T",
                        "albumName": "专辑T",
                        "interval": "04:00",
                        "img": "https://img.test/cover.jpg",
                    },
                    {"source": "kw", "songmid": "S1", "name": name, "singer": "歌手A"},
                ]
        return web.json_response({"code": 0, "data": {"list": items}})

    async def _url(self, request: web.Request) -> web.Response:
        info = request.match_info
        prefix = "denied" if info["source"] == "tx" and self.tx_denied else "media"
        return web.json_response(
            {
                "code": 0,
                "data": {
                    "type": "128k",
                    "url": f"http://127.0.0.1:{self.port}/{prefix}/{info['source']}/{info['song_id']}.mp3",
                },
            }
        )

    async def _media(self, _: web.Request) -> web.Response:
        return web.Response(body=b"fake-audio-bytes")

    async def _denied(self, _: web.Request) -> web.Response:
        return web.Response(status=403, text="forbidden")

    async def _lyric(self, request: web.Request) -> web.Response:
        lyric = "[kuwo:113]\n[ver:v1.0]\n[00:01.00]第一行\n[00:05.00]第二行\n"
        return web.json_response({"code": 0, "data": {"lyric": lyric}})


@pytest.fixture
async def server():
    s = FakeMusicServer()
    await s.start()
    yield s
    await s.stop()


@pytest.fixture
async def env(server) -> AsyncIterator[tuple]:
    async with ClientSession() as http:
        config = BotConfig(
            music_server_url=f"http://127.0.0.1:{server.port}",
            music_quality="128k",
            music_source="tx",
        )
        yield http, config


@pytest.fixture(autouse=True)
def _reset_menu_state():
    """菜单/卡片消息记录与消息去重表是模块级状态，测试间清空避免串扰。"""
    music._last_msg.clear()
    music._seen_messages.clear()
    yield
    music._last_msg.clear()
    music._seen_messages.clear()


# 回调按钮 data：音乐|动作|源|song_id|歌名|歌手|封面（歌名/歌手/封面 % 转义）
IMG_URL = "https://img.test/cover.jpg"
PICK_1 = music._song_data("pick", music.Song("tx", "T1", "稻香", "歌手T", img=IMG_URL))
PLAY_TX = music._song_data("play", music.Song("tx", "T1", "稻香", "歌手T", img=IMG_URL))
DL_TX = music._song_data("dl", music.Song("tx", "T1", "稻香", "歌手T", img=IMG_URL))
# 旧版按钮 data（无封面段）—— 兼容性用例
PICK_1_LEGACY = "音乐|pick|tx|T1|%E7%A8%BB%E9%A6%99|%E6%AD%8C%E6%89%8BT"


def posts(calls: list[dict]) -> list[dict]:
    return [c for c in calls if c.get("path", "").endswith("/messages")]


def buttons(body: dict) -> list[dict]:
    """键盘的行列表；每行是 {"buttons": [...]}。"""
    return body["keyboard"]["content"]["rows"]


def labels(body: dict) -> list[list[str]]:
    return [[b["render_data"]["label"] for b in r["buttons"]] for r in buttons(body)]


async def test_search_prefers_tx_and_falls_back(server, env):
    http, config = env
    song = await search(http, config, "稻香")
    assert song is not None and (song.source, song.song_id) == ("tx", "T1")
    assert server.queries == [{"name": "稻香", "source": "tx"}]

    server.queries.clear()
    fallback = await search(http, config, "兜底")
    assert fallback is not None
    assert (fallback.source, fallback.song_id) == ("kw", "S1")  # tx 没有结果 → 聚合兜底
    assert [q["source"] for q in server.queries] == ["tx", ""]


async def test_menu_only_selects_songs(env):
    """搜索菜单只负责选歌：编号回调按钮，不放播放/下载、不放链接、不带音源。"""
    http, config = env
    api, calls = make_api()
    assert await handle_command(api, http, config, "G1", "m1", "点歌 稻香")
    body = posts(calls)[0]["json"]
    assert body["msg_type"] == 2
    md = body["markdown"]["content"]
    assert "## 点歌：稻香" in md
    assert "1. 稻香 - 歌手T" in md
    assert "QQ音乐" not in md and "酷我" not in md  # 每首歌后面不再标音源
    assert "http" not in md  # 菜单里没有任何链接
    # 编号回调按钮（type=1）：点一下弹卡片，data 自包含歌曲（含封面）
    rows = buttons(body)
    assert [b["render_data"]["label"] for b in rows[0]["buttons"]] == ["1", "2"]
    assert [b["action"]["type"] for b in rows[0]["buttons"]] == [1, 1]
    assert rows[0]["buttons"][0]["action"]["data"] == PICK_1
    # 首页无上一页，本页未满一页（2 首 < 10）→ 也不给下一页
    assert labels(body)[-1] == ["1", "2"]


async def test_menu_pagination_buttons(env):
    """满页给「下一页」，非首页给「上一页」，data 带页码与关键词。"""
    http, config = env
    api, calls = make_api()
    full = replace(config, menu_page_size=2)
    await send_menu(api, http, full, "G1", "稻香", page=2)
    body = posts(calls)[0]["json"]
    nav = [b["render_data"]["label"] for b in buttons(body)[-1]["buttons"]]
    assert nav == ["上一页", "下一页"]
    data = [b["action"]["data"] for b in buttons(body)[-1]["buttons"]]
    assert data[0] == f"音乐|page|{quote('稻香', safe='')}|1"
    assert data[1] == f"音乐|page|{quote('稻香', safe='')}|3"


async def test_menu_page_size_row_layout(env):
    """每行最多 5 个按钮，行数不超过键盘上限。"""
    http, config = env
    api, calls = make_api()
    big = replace(config, menu_page_size=20)
    await send_menu(api, http, big, "G1", "稻香")
    rows = buttons(posts(calls)[0]["json"])
    assert len(rows) <= 5
    for row in rows[:-1]:
        assert 1 <= len(row["buttons"]) <= 5


async def test_menu_withdraws_previous(env):
    """同群出新菜单时撤回上一张菜单；不同群互不影响。"""
    http, config = env
    api, calls = make_api()
    assert await handle_command(api, http, config, "G1", "m1", "点歌 稻香")
    assert await handle_command(api, http, config, "G2", "m2", "点歌 稻香")
    assert not [c for c in calls if c["method"] == "DELETE"]
    assert await handle_command(api, http, config, "G1", "m3", "点歌 稻香")
    assert [c["path"] for c in calls if c["method"] == "DELETE"] == [
        "/v2/groups/G1/messages/MSG"
    ]


async def test_menu_fallback_keeps_keyboard(env):
    http, config = env
    api, calls = make_api(fail_md=True)
    await handle_command(api, http, config, "G1", "m1", "点歌 稻香")
    sent = posts(calls)
    assert len(sent) == 1  # Markdown 失败后回退文本 + 按钮仍带上
    body = sent[0]["json"]
    assert body["msg_type"] == 0
    assert "稻香 - 歌手T" in body["content"]
    assert "（点下方编号选歌）" in body["content"]
    assert "keyboard" in body


async def test_pick_button_sends_card(env):
    """点编号 → 弹歌曲卡片：封面图 + 歌名/来源 + [▶播放][⬇下载] 回调按钮。

    卡片由回调 data 还原（自包含 song 字段），故不含专辑/时长等菜单外信息。"""
    http, config = env
    api, calls = make_api()
    await handle_button(
        api, http, config, "G1", "I1", PICK_1, event_id="INTERACTION_CREATE:I1"
    )
    card = posts(calls)[0]["json"]
    assert card["msg_type"] == 2
    md = card["markdown"]["content"]
    assert "## 稻香 - 歌手T" in md
    assert "![封面 #240px #240px](https://img.test/cover.jpg)" in md  # 专辑封面
    assert "来源：QQ音乐" in md  # 卡片保留音源
    assert labels(card) == [["▶ 播放", "⬇ 下载"]]
    assert [b["action"]["type"] for b in buttons(card)[0]["buttons"]] == [1, 1]
    assert [b["action"]["data"] for b in buttons(card)[0]["buttons"]] == [
        PLAY_TX, DL_TX,
    ]
    # 已回应互动
    assert (calls[-1]["method"], calls[-1]["path"]) == ("PUT", "/interactions/I1")


async def test_card_without_img_when_source_has_none(env):
    """无封面源的歌曲（如 kw）：卡片不含图片，其余照常。"""
    http, config = env
    api, calls = make_api()
    no_img = "音乐|pick|kw|S1|%E7%A8%BB%E9%A6%99|%E6%AD%8C%E6%89%8BA|"
    await handle_button(api, http, config, "G1", "I1", no_img)
    md = posts(calls)[0]["json"]["markdown"]["content"]
    assert "![封面" not in md
    assert "## 稻香 - 歌手A" in md


async def test_legacy_button_data_without_img(env):
    """旧版 4 段 data 仍能处理（封面缺省），保证已发出的菜单按钮不失效。"""
    http, config = env
    api, calls = make_api()
    assert await handle_button(api, http, config, "G1", "I1", PICK_1_LEGACY)
    md = posts(calls)[0]["json"]["markdown"]["content"]
    assert "## 稻香 - 歌手T" in md
    assert "![封面" not in md


async def test_card_play_button_sends_voice(env):
    """卡片上的 [▶ 播放] 是回调按钮：点一下直接发语音。"""
    http, config = env
    api, calls = make_api()
    await handle_button(
        api, http, config, "G1", "I2", PLAY_TX, event_id="INTERACTION_CREATE:I2"
    )
    upload = [c for c in calls if c.get("file_type")][0]
    assert upload["file_type"] == 3
    assert upload["url"].endswith("/media/tx/T1.mp3")
    assert upload["timeout"].total == 120  # 上传放宽超时：QQ 侧抓流慢
    body = posts(calls)[-1]["json"]
    assert body["msg_type"] == 7
    assert body["event_id"] == "INTERACTION_CREATE:I2"


async def test_card_dl_button_sends_file(env):
    """SEND_MODE=file 时下载上传群文件（file_type=4）。"""
    http, config = env
    api, calls = make_api()
    file_cfg = replace(config, send_mode="file")
    await handle_button(
        api, http, file_cfg, "G1", "I3", DL_TX, event_id="INTERACTION_CREATE:I3"
    )
    upload = [c for c in calls if c.get("file_type")][0]
    assert upload["file_type"] == 4
    assert posts(calls)[-1]["json"]["msg_type"] == 7


async def test_default_link_mode_skips_upload(env):
    """默认 SEND_MODE=link：下载直接发直链，根本不上传（避开 40093005）。"""
    http, config = env
    assert config.send_mode == "link"
    api, calls = make_api()
    await handle_button(
        api, http, config, "G1", "I3", DL_TX, event_id="INTERACTION_CREATE:I3"
    )
    assert not [c for c in calls if c.get("file_type")]  # 无上传请求
    body = posts(calls)[-1]["json"]
    assert body["msg_type"] == 0
    assert "/media/tx/T1.mp3" in body["content"]


async def test_pick_swaps_menu_for_card(env):
    """点编号后：撤回旧菜单、发出卡片、记录换成卡片消息 id。"""
    http, config = env
    api, calls = make_api()
    assert await handle_command(api, http, config, "G1", "m1", "点歌 稻香")
    assert music._last_msg.get("G1") == "MSG"  # 菜单消息
    await handle_button(api, http, config, "G1", "I1", PICK_1)
    assert ("DELETE", "/v2/groups/G1/messages/MSG") in [
        (c["method"], c["path"]) for c in calls
    ]
    assert "## 稻香 - 歌手T" in posts(calls)[-1]["json"]["markdown"]["content"]


async def test_play_after_card_withdraws_card(env):
    """卡片上点播放成功后，撤回已完成使命的卡片。"""
    http, config = env
    api, calls = make_api()
    await handle_button(api, http, config, "G1", "I1", PICK_1)
    calls.clear()
    await handle_button(api, http, config, "G1", "I2", PLAY_TX)
    upload = [c for c in calls if c.get("file_type")][0]
    assert upload["file_type"] == 3
    assert upload["url"].endswith("/media/tx/T1.mp3")
    assert posts(calls)[-1]["json"]["msg_type"] == 7
    assert ("DELETE", "/v2/groups/G1/messages/MSG") in [
        (c["method"], c["path"]) for c in calls
    ]
    assert "G1" not in music._last_msg


async def test_card_withdrawn_even_on_failure(env):
    """卡片是一次性 UI：即使发送回退文本，卡片也要撤回（原实现只在成功时撤）。"""
    http, config = env
    api, calls = make_api(fail_types={3})  # 语音上传失败 → 回退文本
    await handle_button(api, http, config, "G1", "I1", PICK_1)
    calls.clear()
    await handle_button(api, http, config, "G1", "I2", PLAY_TX)
    assert ("DELETE", "/v2/groups/G1/messages/MSG") in [
        (c["method"], c["path"]) for c in calls
    ]
    assert "G1" not in music._last_msg


async def test_card_withdrawn_when_play_url_unavailable(env):
    """拿不到播放链接（MusicServer 不可达，send_song 直接 reply 后返回）时
    也要撤回卡片。"""
    http, config = env
    api, calls = make_api()
    await handle_button(api, http, config, "G1", "I1", PICK_1)
    calls.clear()
    broken = replace(config, music_server_url="http://127.0.0.1:1")
    await handle_button(api, http, broken, "G1", "I3", DL_TX)
    assert ("DELETE", "/v2/groups/G1/messages/MSG") in [
        (c["method"], c["path"]) for c in calls
    ]


async def test_file_upload_permission_denied_falls_back(env):
    """SEND_MODE=file 且群文件上传被拒（40093005）→ 回退直链文本 + 说明，并撤回卡片。"""
    http, config = env
    api, calls = make_api(fail_file_perm=True)
    file_cfg = replace(config, send_mode="file")
    await handle_button(api, http, file_cfg, "G1", "I1", PICK_1)
    calls.clear()
    await handle_button(api, http, file_cfg, "G1", "I3", DL_TX)
    body = posts(calls)[-1]["json"]
    assert body["msg_type"] == 0
    assert "/media/tx/T1.mp3" in body["content"]
    assert "仅限群主/管理员" in body["content"]
    assert ("DELETE", "/v2/groups/G1/messages/MSG") in [
        (c["method"], c["path"]) for c in calls
    ]


async def test_voice_upload_unaffected_by_file_permission(env):
    """40093005 只针对文件；语音上传不受影响（仍是语音消息）。"""
    http, config = env
    api, calls = make_api(fail_file_perm=True)
    await handle_button(api, http, config, "G1", "I1", PICK_1)
    calls.clear()
    await handle_button(api, http, config, "G1", "I2", PLAY_TX)
    upload = [c for c in calls if c.get("file_type")][0]
    assert upload["file_type"] == 3
    assert posts(calls)[-1]["json"]["msg_type"] == 7


async def test_withdraw_disabled_keeps_messages(env):
    """WITHDRAW 关闭时不出任何撤回请求（菜单/卡片都留着）。"""
    http, config = env
    api, calls = make_api()
    off = replace(config, withdraw=False)
    await send_menu(api, http, off, "G1", "稻香")
    await send_menu(api, http, off, "G1", "稻香")  # 出新菜单也不撤旧
    await handle_button(api, http, off, "G1", "I1", PICK_1)
    await handle_button(api, http, off, "G1", "I2", PLAY_TX)
    assert not [c for c in calls if c["method"] == "DELETE"]


async def test_play_voice_falls_back_to_text(env):
    """播放只发语音：语音失败直接回退直链文本，不再尝试文件。"""
    http, config = env
    api, calls = make_api(fail_types={3})
    await handle_button(api, http, config, "G1", "I2", PLAY_TX)
    body = posts(calls)[-1]["json"]
    assert body["msg_type"] == 0
    assert "/media/tx/T1.mp3" in body["content"]


async def test_dl_falls_back_to_text(env):
    """SEND_MODE=file 时文件上传失败 → 回退直链文本。"""
    http, config = env
    api, calls = make_api(fail_types={4})
    file_cfg = replace(config, send_mode="file")
    await handle_button(api, http, file_cfg, "G1", "I3", DL_TX)
    body = posts(calls)[-1]["json"]
    assert body["msg_type"] == 0
    assert "/media/tx/T1.mp3" in body["content"]


async def test_upload_timeout_falls_back_to_text(env):
    """上传超时（QQ 服务器抓流超过请求超时）→ 回退文本。"""
    http, config = env
    api, calls = make_api(fail_upload=True)
    await handle_button(api, http, config, "G1", "I2", PLAY_TX)
    body = posts(calls)[-1]["json"]
    assert "/media/tx/T1.mp3" in body["content"]


async def test_legacy_pick_text_is_ignored(env):
    """旧的「点歌播放 …」文字指令已废除，不再匹配（按钮回调取代）。"""
    http, config = env
    api, calls = make_api()
    assert not await handle_command(api, http, config, "G1", "m1", "点歌播放 1")
    assert calls == []


async def test_card_withdrawn_when_send_song_raises(env):
    """send_song 抛错（如候选源搜索时 MusicServer 挂掉）也要撤回卡片——
    这是 finally 存在的理由，否则卡片会永久残留。"""
    http, config = env
    api, calls = make_api()
    await handle_button(api, http, config, "G1", "I1", PICK_1)
    calls.clear()
    # 换一个不可达的 MusicServer：直链探测与候选源搜索都会失败
    broken = replace(config, music_server_url="http://127.0.0.1:1")
    await handle_button(api, http, broken, "G1", "I2", PLAY_TX)
    assert ("DELETE", "/v2/groups/G1/messages/MSG") in [
        (c["method"], c["path"]) for c in calls
    ]
    assert "G1" not in music._last_msg


async def test_card_play_falls_back_to_candidate_source(server):
    """tx 直链不可抓取时，卡片上的播放同样走候选源探测，用 kw 直链上传。"""
    s = FakeMusicServer(tx_denied=True)
    await s.start()
    try:
        async with ClientSession() as http:
            config = BotConfig(
                music_server_url=f"http://127.0.0.1:{s.port}",
                music_quality="128k",
                music_source="tx",
            )
            api, calls = make_api()
            await handle_button(api, http, config, "G1", "I2", PLAY_TX)
    finally:
        await s.stop()
    upload = [c for c in calls if c.get("file_type")][0]
    assert upload["file_type"] == 3
    assert "/media/kw/" in upload["url"]  # 绕开被风控的 tx 直链


async def test_page_button_new_menu(env):
    http, config = env
    api, calls = make_api()
    data = f"音乐|page|{quote('稻香', safe='')}|2"
    assert await handle_button(
        api, http, config, "G1", "I1", data, event_id="INTERACTION_CREATE:I1"
    )
    body = posts(calls)[0]["json"]
    assert body["msg_type"] == 2  # 翻页后发新菜单
    assert "点歌：稻香" in body["markdown"]["content"]
    assert (calls[-1]["method"], calls[-1]["path"]) == ("PUT", "/interactions/I1")

    api, calls = make_api()
    assert not await handle_button(api, http, config, "G1", "I1", "别的插件|xx")
    assert calls == []  # 非音乐按钮不处理也不回应


async def test_slash_commands_share_text_parsing(env):
    """斜杠指令与裸文本命令同一套解析：剥掉开头的 / 与 @机器人 前缀。"""
    http, config = env
    for content in ("点歌 稻香", "/点歌 稻香", "<@!BOT123> /点歌 稻香"):
        api, calls = make_api()
        assert await handle_command(api, http, config, "G1", "m1", content)
        body = posts(calls)[0]["json"]
        assert body["msg_type"] == 2
        assert "点歌：稻香" in body["markdown"]["content"]

    api, calls = make_api()
    assert await handle_command(api, http, config, "G1", "m1", "/歌词 稻香")
    assert posts(calls)[0]["json"]["content"] == "第一行\n第二行"


async def test_slash_help_sends_command_keyboard(env):
    """/帮助 出帮助正文 + type=2 指令按钮（点一下往输入框填斜杠指令）。"""
    http, config = env
    api, calls = make_api()
    assert await handle_command(api, http, config, "G1", "m1", "/帮助")
    body = posts(calls)[0]["json"]
    assert "点歌" in body["content"] and "歌词" in body["content"]
    rows = buttons(body)
    assert len(rows) == 1
    song_btn, lyric_btn = rows[0]["buttons"]
    assert song_btn["action"]["type"] == 2
    assert song_btn["action"]["data"] == "/点歌 "
    assert lyric_btn["action"]["data"] == "/歌词 "


async def test_slash_help_falls_back_to_plain_text(env):
    http, config = env
    api, calls = make_api(fail_keyboard=True)
    assert await handle_command(api, http, config, "G1", "m1", "/帮助")
    sent = posts(calls)
    assert len(sent) == 1  # 键盘被拒，回退为一条纯文本
    assert "keyboard" not in sent[0]["json"]
    assert "点歌" in sent[0]["json"]["content"]


async def test_bare_command_replies_usage(env):
    """斜杠菜单的参数可选时可能不带参发出来：只敲命令给用法提示。"""
    http, config = env
    for content, hint in (("点歌", "用法：点歌"), ("/歌词", "用法：歌词")):
        api, calls = make_api()
        assert await handle_command(api, http, config, "G1", "m1", content)
        assert posts(calls)[0]["json"]["content"].startswith(hint)


async def test_lyric_miss_help_and_no_hit(env):
    http, config = env
    api, calls = make_api()
    await handle_command(api, http, config, "G1", "m1", "歌词 稻香")
    assert posts(calls)[0]["json"]["content"] == "第一行\n第二行"

    api, calls = make_api()
    await handle_command(api, http, config, "G1", "m1", "点歌 空白歌")
    assert posts(calls)[0]["json"]["content"] == "没找到「空白歌」"

    broken = BotConfig(
        music_server_url="http://127.0.0.1:1",
        music_quality="128k",
        music_source="tx",
    )
    api, calls = make_api()
    await handle_command(api, http, broken, "G1", "m1", "点歌 稻香")
    assert posts(calls)[0]["json"]["content"] == "音乐服务暂时不可用，稍后再试试"

    api, calls = make_api()
    assert await handle_command(api, http, config, "G1", "m1", "音乐帮助")
    assert "点歌" in posts(calls)[0]["json"]["content"]
    assert not await handle_command(api, http, config, "G1", "m1", "早上好")


def test_trim_strips_tags_and_truncates():
    from musicbot.music import _trim

    text = "[kuwo:113]\n[ti:稻香]\n[00:01.00]正文一行\n[offset:0]\n"
    assert _trim(text) == "正文一行"
    long = "\n".join(f"[00:{i:02d}.00]第{i}行" for i in range(100))
    trimmed = _trim(long, limit=100)
    assert trimmed.endswith("……(歌词过长已截断)")
    assert len(trimmed) <= 100 + len("\n……(歌词过长已截断)")


class PanelRecording(BotApi):
    """预设 /v2/panels 响应并记录调用：fail_get=True 时查询抛 403。"""

    def __init__(self, records: list | None = None, fail_get: bool = False) -> None:
        self.records = records or []
        self.fail_get = fail_get
        self.calls: list[dict] = []

    async def request(self, method: str, path: str, **kwargs: Any) -> Any:
        if self.fail_get and path == "/v2/panels":
            raise ApiError(403, 11253, "无权限")
        self.calls.append({"method": method, "path": path, **kwargs})
        if method == "GET":
            # 响应整体包一层 {"data": ...}，顺带覆盖 _unwrap 的拆包路径
            return {"data": {"records": self.records, "is_end": True}}
        if method == "POST":
            return {"panel_id": "P1"}
        return {"version": 2}


async def test_panel_created_when_missing():
    api = PanelRecording()
    await sync_command_panel(api)
    assert [(c["method"], c["path"]) for c in api.calls] == [
        ("GET", "/v2/panels"),
        ("POST", "/v2/panels"),
    ]
    body = api.calls[1]["json"]
    assert body["scope"] == "group" and body["target_type"] == "all"
    assert [i["name"] for i in body["panel"]["items"]] == ["点歌", "歌词", "帮助"]


async def test_panel_skipped_when_identical():
    items = [
        {"name": "点歌", "type": "command", "desc": "出选歌菜单，点编号选歌"},
        {"name": "歌词", "type": "command", "desc": "查看歌词"},
        {"name": "帮助", "type": "command", "desc": "查看帮助"},
    ]
    api = PanelRecording(
        records=[{"panel_id": "P9", "panel": {"items": items, "remark": "音乐bot"}}]
    )
    await sync_command_panel(api)
    assert [(c["method"], c["path"]) for c in api.calls] == [("GET", "/v2/panels")]


async def test_panel_updated_when_differs():
    # 这条用响应不包 {"data"} 外层的形态，覆盖 _unwrap 的另一分支
    api = PanelRecording(
        records=[
            {
                "panel_id": "P9",
                "panel": {
                    "items": [{"name": "点歌", "type": "command", "desc": "旧"}],
                    "remark": "音乐bot",
                },
            }
        ]
    )
    await sync_command_panel(api)
    assert (api.calls[1]["method"], api.calls[1]["path"]) == ("PUT", "/v2/panels/P9")


async def test_panel_failure_swallowed():
    api = PanelRecording(fail_get=True)
    await sync_command_panel(api)  # 失败只记日志，绝不抛出
    assert api.calls == []


def test_config_send_mode(monkeypatch):
    """SEND_MODE 默认 link（避开文件上传权限），file 可显式开启，非法值回落 link。"""
    from musicbot.config import BotConfig

    monkeypatch.delenv("SEND_MODE", raising=False)
    assert BotConfig.load().send_mode == "link"

    monkeypatch.setenv("SEND_MODE", "FILE")  # 大小写不敏感
    assert BotConfig.load().send_mode == "file"

    monkeypatch.setenv("SEND_MODE", "随便写的")
    assert BotConfig.load().send_mode == "link"  # 非法值不报错，按默认处理


def test_config_log_level(monkeypatch):
    """LOG_LEVEL 默认 INFO，可显式设置（大小写不敏感），非法值回落 INFO。"""
    from musicbot.config import BotConfig

    monkeypatch.delenv("LOG_LEVEL", raising=False)
    assert BotConfig.load().log_level == "INFO"

    monkeypatch.setenv("LOG_LEVEL", "debug")
    assert BotConfig.load().log_level == "DEBUG"

    monkeypatch.setenv("LOG_LEVEL", "乱写")
    assert BotConfig.load().log_level == "INFO"


def test_setup_logging_writes_file(tmp_path: Path):
    """setup_logging 同时写控制台与文件；写不进去时退化为仅控制台不抛错。"""
    from loguru import logger as lg

    from musicbot.config import BotConfig
    from musicbot.main import setup_logging

    cfg = BotConfig(
        music_server_url="http://127.0.0.1:1",
        music_quality="128k",
        music_source="tx",
        log_level="INFO",
        log_dir=tmp_path,
    )
    try:
        setup_logging(cfg)
        lg.info("测试日志落盘")
        lg.complete()  # 等 enqueue 队列刷完
        logfile = tmp_path / "musicbot.log"
        assert logfile.is_file()
        assert "测试日志落盘" in logfile.read_text(encoding="utf-8")
    finally:
        lg.remove()  # 清掉本用例加的 handler，避免影响其他测试

    # 目录不可写（用一个文件当目录）→ 不抛错，仅告警
    blocked = tmp_path / "not-a-dir"
    blocked.write_text("x", encoding="utf-8")
    setup_logging(replace(cfg, log_dir=blocked))
    lg.remove()


def test_config_reload_flag(monkeypatch):
    """RELOAD 默认关（生产不受影响），on 可开启，非法值按关处理。"""
    from musicbot.config import BotConfig

    monkeypatch.delenv("RELOAD", raising=False)
    assert BotConfig.load().reload is False

    monkeypatch.setenv("RELOAD", "on")
    assert BotConfig.load().reload is True

    monkeypatch.setenv("RELOAD", "随便写")
    assert BotConfig.load().reload is False


# ---- 热重载：文件监听与进程监督 ----

def test_watched_files_and_skip_rules(tmp_path: Path):
    """监控 .py/.toml 与 .env；跳过 logs/、__pycache__/、.venv/ 与无关文件。"""
    from musicbot import reload

    (tmp_path / "musicbot").mkdir()
    (tmp_path / "musicbot" / "music.py").write_text("", encoding="utf-8")
    (tmp_path / "tests").mkdir()
    (tmp_path / "tests" / "test_x.py").write_text("", encoding="utf-8")
    (tmp_path / "intents.toml").write_text("", encoding="utf-8")
    (tmp_path / ".env").write_text("", encoding="utf-8")
    (tmp_path / "README.md").write_text("", encoding="utf-8")  # 无关后缀
    (tmp_path / "logs").mkdir()
    (tmp_path / "logs" / "musicbot.log").write_text("", encoding="utf-8")
    (tmp_path / "musicbot" / "__pycache__").mkdir()
    (tmp_path / "musicbot" / "__pycache__" / "x.py").write_text("", encoding="utf-8")

    watched = reload.iter_watched_files(tmp_path)
    names = {p.relative_to(tmp_path).as_posix() for p in watched}
    assert names == {"musicbot/music.py", "tests/test_x.py", "intents.toml", ".env"}


def test_changed_files_detects_edit_add_remove(tmp_path: Path):
    """快照比对：改动/新增/删除都能识别，未变则返回空。"""
    from musicbot import reload

    f = tmp_path / "musicbot"
    f.mkdir()
    mod = f / "music.py"
    mod.write_text("a", encoding="utf-8")

    before = reload.snapshot(tmp_path)
    assert reload.changed_files(before, reload.snapshot(tmp_path)) == []

    mod.write_text("bb", encoding="utf-8")
    assert reload.changed_files(before, reload.snapshot(tmp_path)) == [mod]


def test_is_child_guard(monkeypatch):
    """子进程标记生效：带标记的进程不再做监督，避免 --reload 无限自我复制。"""
    from musicbot import reload

    monkeypatch.delenv(reload.CHILD_ENV, raising=False)
    assert reload.is_child() is False
    monkeypatch.setenv(reload.CHILD_ENV, "1")
    assert reload.is_child() is True


def test_main_child_does_not_supervise(monkeypatch):
    """子进程即使带 --reload / RELOAD=on 也只跑 bot，不起监督
    （否则 --reload 会无限自我复制）。"""
    from musicbot import main as mb_main
    from musicbot import reload

    called: dict[str, bool] = {}

    async def fake_amain() -> None:
        called["bot"] = True

    monkeypatch.setattr(reload, "is_child", lambda: True)
    monkeypatch.setattr(mb_main, "amain", fake_amain)
    monkeypatch.setattr(
        reload, "run_supervisor",
        lambda *a, **k: called.setdefault("supervisor", True),
    )
    monkeypatch.setattr(mb_main.BotConfig, "load", staticmethod(
        lambda: replace(
            BotConfig(
                music_server_url="http://127.0.0.1:1",
                music_quality="128k",
                music_source="tx",
            ),
            reload=True,
        )
    ))

    mb_main.main(["--reload"])
    assert called == {"bot": True}  # 跑的是 bot，没有起监督


# ---- 全量模式（不 @ 也能用）与去重 ----

def make_emitter(api, http, config) -> EventEmitter:
    """按 main.py 的装配方式注册 handler，手工注入 SDK 会在 run_loop 里
    登记的 BotApi；http 走闭包（与 main.py 一致）。"""
    ee = EventEmitter()
    register(ee, config, http)
    ee.services.update({BotApi: api})
    return ee


async def push(ee: EventEmitter, event: str, d: dict) -> None:
    await ee.emit({"op": 0, "t": event, "d": d})


def group_payload(msg_id: str, content: str, *, bot: bool = False) -> dict:
    return {
        "id": msg_id,
        "content": content,
        "group_openid": "G1",
        "author": {"member_openid": "U1", "bot": bot},
    }


async def test_no_at_message_sends_menu(env):
    """全量模式：群里不 @ 机器人也点歌（GROUP_MESSAGE_CREATE），
    被动回复仍用这条消息的 msg_id。"""
    http, config = env
    api, calls = make_api()
    ee = make_emitter(api, http, config)
    await push(ee, "GROUP_MESSAGE_CREATE", group_payload("m1", "点歌 稻香"))
    body = posts(calls)[0]["json"]
    assert body["msg_id"] == "m1"
    # 官方 event_id 支持列表不含 GROUP_MESSAGE_CREATE，只能用 msg_id
    assert "event_id" not in body
    assert "## 点歌：稻香" in body["markdown"]["content"]


async def test_at_message_still_handled(env):
    """@消息（默认事件）行为不变。"""
    http, config = env
    api, calls = make_api()
    ee = make_emitter(api, http, config)
    await push(ee, "GROUP_AT_MESSAGE_CREATE", group_payload("m1", "点歌 稻香"))
    assert "## 点歌：稻香" in posts(calls)[0]["json"]["markdown"]["content"]


async def test_group_message_ignores_own_and_other_chat(env):
    """机器人自己的消息（可能被回推）与非命令消息一律不回复。"""
    http, config = env
    api, calls = make_api()
    ee = make_emitter(api, http, config)
    await push(ee, "GROUP_MESSAGE_CREATE", group_payload("m1", "点歌 稻香", bot=True))
    await push(ee, "GROUP_MESSAGE_CREATE", group_payload("m2", ""))
    await push(ee, "GROUP_MESSAGE_CREATE", {
        "id": "m3", "content": "早上好", "group_openid": "G1",
        "author": {"member_openid": "U1"},
    })
    assert calls == []


async def test_duplicate_push_handled_once(env):
    """同一 msg_id 重复推送、或 @消息同时命中两个事件时只处理一次。"""
    http, config = env
    api, calls = make_api()
    ee = make_emitter(api, http, config)
    payload = group_payload("m1", "点歌 稻香")
    await push(ee, "GROUP_MESSAGE_CREATE", payload)
    await push(ee, "GROUP_MESSAGE_CREATE", payload)
    await push(ee, "GROUP_AT_MESSAGE_CREATE", payload)
    assert len(posts(calls)) == 1


async def test_no_at_command_without_content_is_ignored(env):
    """纯图片等无正文消息（全量模式同样推送）不触发任何请求。"""
    http, config = env
    api, calls = make_api()
    ee = make_emitter(api, http, config)
    await push(ee, "GROUP_MESSAGE_CREATE", {
        "id": "m1",
        "group_openid": "G1",
        "author": {"member_openid": "U1"},
        "attachments": [{"content_type": 1, "url": "https://x/1.png"}],
    })
    assert calls == []


def test_dedup_expires_and_bounds(monkeypatch):
    """去重表：过期记录可再次处理，条数达到上限时淘汰最旧。"""
    assert not music._handled("m1")
    assert music._handled("m1")

    # TTL 过去后同一条消息重新可处理
    monkeypatch.setattr(music, "_DEDUP_TTL", 0.0)
    assert not music._handled("m1")

    # 记录数超过上限时淘汰最旧的
    monkeypatch.setattr(music, "_DEDUP_TTL", 300.0)
    monkeypatch.setattr(music, "_DEDUP_MAX", 2)
    music._seen_messages.clear()
    for i in range(3):
        music._handled(f"n{i}")
    assert list(music._seen_messages) == ["n1", "n2"]


async def test_bot_own_reply_is_not_self_triggered(env):
    """bot 自己发的帮助文本里有「点歌」「音乐帮助」，被全量模式回推时不能
    自触发（否则递归出菜单）；作者无 bot 标记时靠发送记录兜底。"""
    http, config = env
    api, calls = make_api()
    ee = make_emitter(api, http, config)
    assert await handle_command(api, http, config, "G1", "m1", "音乐帮助")
    assert len(posts(calls)) == 1

    # 平台把这个已发出的消息（id=MSG）回推，author 不含 bot 字段
    await push(ee, "GROUP_MESSAGE_CREATE", {
        "id": "MSG", "content": "♪ 音乐 bot\n点歌 <歌名>：出菜单",
        "group_openid": "G1", "author": {"member_openid": "U1"},
    })
    assert len(posts(calls)) == 1  # 未再发任何消息

    # 同样内容的新消息（不同 id）来自真人 → 正常处理
    await push(ee, "GROUP_MESSAGE_CREATE", {
        "id": "m9", "content": "音乐帮助", "group_openid": "G1",
        "author": {"member_openid": "U2"},
    })
    assert len(posts(calls)) == 2