"""点歌业务：MusicServer 搜索、Markdown 选歌菜单、点选后的歌曲卡片、歌词清洗。 指令：点歌（选歌菜单）、歌词、热搜（热词点一下直接搜）、榜单（榜单点开 直接是选歌菜单）、「<平台>下载 <歌名>」（如 酷狗下载 稻香，在指定平台 搜歌出选歌菜单——与点歌同款交互，点编号弹卡片再播放/下载）——平台与 指令名按 MusicServer /sources 动态生成，服务器加源即自动生效。菜单只负责 选歌：每页 N 首（MENU_PAGE_SIZE，默认 10）、编号回调按钮 + 上下页。点 编号 → 弹歌曲卡片（歌名/来源 + [▶播放][⬇下载] 回调按钮），卡片上再点才 执行。播放发语音、下载发文件，失败均回退直链文本。音频平台的节目级条目 （猫耳广播剧 d: 前缀 / 懒人书 / 蜻蜓频道）点编号先出选集菜单（/album 接口， 选集再点编号才弹卡片），不会把一个节目当单个音频播。 按钮回调进门先回应互动（防 QQ 客户端等超时），同一互动 uuid / 同群同按钮 10 秒内只处理一次（防抖）；播放/下载先回「正在获取」占位，结果再补发。 群消息去重两把钥匙：msg_id（同帧重推与 self-echo）加「群+发送人+文本」 10 秒短窗——全量模式下同一条 @消息会同时命中两个事件且 msg_id 不同 （2026-10-03 实测因此双发过）。 自动撤回（翻页撤旧菜单、选歌后撤菜单、播放/下载后撤卡片）由 WITHDRAW 开关统一控制，默认开。 配置了 PUBLIC_URL 时启用文件代理（见 proxy.py）：上传给 QQ 的与直链文本 里的 URL 换成 bot 自己的公开地址，请求到来时现场重解析直链再转发字节， 绕开 QQ 服务器抓不动风控直链（40093007）与 qs 本地流外网不可达的问题。""" import re import time import weakref from dataclasses import dataclass, replace from typing import Any from urllib.parse import quote, unquote from aiohttp import ClientError, ClientSession, ClientTimeout from loguru import logger from qqbotsdk import Bot, EventEmitter from qqbotsdk.api import ApiError, BotApi from qqbotsdk.events import Event from qqbotsdk.payloads import GroupAtMessage, Interaction, Ready from .config import BotConfig from .panel import sync_command_panel from .proxy import FileProxy _SONG_RE = re.compile(r"点歌\s+(\S.*)") _LYRIC_RE = re.compile(r"歌词\s+(\S.*)") # @机器人 前缀（"<@!12345> 点歌 稻香"）；开头的 "/"（斜杠指令）在 _normalize 剥 _MENTION_RE = re.compile(r"<@!?\w+>") _LRC_TS_RE = re.compile(r"\[\d+:\d+(?:\.\d+)?\]") _LRC_TAG_RE = re.compile(r"(?:\[[^\]]*\])+") _LYRIC_LIMIT = 600 # SDK 抛 ApiError；网络层（如请求超过 TIMEOUT）抛 TimeoutError/ClientError _API_FAILS = (ApiError, TimeoutError, ClientError) # QQ 服务器要现场抓取直链再入库，比常规 API 慢得多，SDK 默认 5s 必超时 _UPLOAD_TIMEOUT = ClientTimeout(total=120) # 群文件（file_type=4）上传仅限群主/管理员，语音（3）不受限；无权限时改发直链 _NO_FILE_PERMISSION = 40093005 _HELP = ( "♪ 音乐 bot\n" "/点歌 <歌名 [歌手]>：出选歌菜单，点编号选歌\n" "/歌词 <歌名>：查看歌词\n" "/热搜：大家都在搜什么，点词直接搜\n" "/榜单：热门榜单，点开就是选歌菜单\n" "/<平台>下载 <歌名>：在指定源搜歌出菜单，如 /酷狗下载 稻香\n" "输入框敲 / 也能唤出这些指令" ) # 搜索/取流失败时统一给用户的提示 _UNAVAILABLE = "音乐服务暂时不可用，稍后再试试" # 平台显示名兜底表：/sources 可达时以服务器报告的 name 为准 _SOURCE_NAMES = { "tx": "QQ音乐", "kw": "酷我", "kg": "酷狗", "wy": "网易云", "qs": "汽水音乐", "xm": "喜马拉雅", "me": "猫耳FM", "qt": "蜻蜓FM", "lr": "懒人听书", "mg": "咪咕音乐", "bd": "千千音乐", "joox": "JOOX", "bili": "哔哩哔哩", "5sing": "5sing", "apple": "Apple Music", } # 「<平台>下载」指令的惯用别名（服务器平台名之外的写法）； # apple 的「Apple Music」全名超出面板指令名的宽度上限，靠短名入面板 _EXTRA_ALIASES = {"wy": ("网易云",), "apple": ("Apple",)} _CJK_RE = re.compile(r"[\u4e00-\u9fff]") def _width(text: str) -> int: """QQ 的长度限制按宽度算：ASCII 计 1，其余（汉字、全角符号）计 2。""" return sum(1 if ord(c) < 128 else 2 for c in text) def _fit_width(text: str, limit: int) -> str: """截到宽度上限内（面板 desc 用；指令名不可截断，超宽只能换短名）。""" chars: list[str] = [] used = 0 for c in text: w = 1 if ord(c) < 128 else 2 if used + w > limit: break chars.append(c) used += w return "".join(chars) def _aliases_of(pid: str, name: str) -> list[str]: """平台指令别名，按 全名 → 短名 → 惯用名 排列：末位最常用。 「酷狗音乐」剥掉尾部「音乐」得短名「酷狗」；词干纯拉丁（如「QQ」） 不剥，保留全名。wy 惯用名「网易云」垫底。""" names = [name] if name.endswith("音乐"): stem = name[:-2] if len(stem) >= 2 and _CJK_RE.search(stem): names.append(stem) names.extend(_EXTRA_ALIASES.get(pid, ())) return names # /sources 不可达时的静态平台兜底（id/name/能力位，照服务器现状写死）； # 能力缺 search/url 的平台不参与下载指令与直链回退 _STATIC_PLATFORMS: dict[str, dict] = { pid: { "id": pid, "name": name, "capabilities": { "search": True, "url": True, "lyric": pid not in ("xm", "me", "qt", "lr"), "hotSearch": pid in ("kw", "kg", "tx", "wy"), "boards": pid in ("kw", "kg", "tx", "wy"), }, } for pid, name in ( ("tx", "QQ音乐"), ("kw", "酷我音乐"), ("kg", "酷狗音乐"), ("wy", "网易音乐"), ("qs", "汽水音乐"), ("xm", "喜马拉雅"), ("me", "猫耳FM"), ("qt", "蜻蜓FM"), ("lr", "懒人听书"), ("mg", "咪咕音乐"), ("bd", "千千音乐"), ("joox", "JOOX"), ("bili", "哔哩哔哩"), ("5sing", "5sing"), ("apple", "Apple Music"), ) } def _cap(platform: dict, cap: str) -> bool: """平台是否具备某能力位（capabilities 缺失按不具备算）。""" return bool((platform.get("capabilities") or {}).get(cap)) # 「<平台>下载」匹配状态：别名表 + 编译好的正则，/sources 刷新时重建 # （platforms() 换缓存时调用），进程启动先用静态表 _alias_to_source: dict[str, str] = {} _dl_re: re.Pattern[str] def _apply_aliases(table: dict[str, dict]) -> None: """按平台表（id → /sources 条目）重建下载指令的别名表与匹配正则。""" global _alias_to_source, _dl_re alias: dict[str, str] = {} for pid, platform in table.items(): name = platform.get("name") or _SOURCE_NAMES.get(pid, pid) for a in _aliases_of(pid, name): alias.setdefault(a, pid) # 撞名时先到先得（服务器顺序） _alias_to_source = alias alts = "|".join(re.escape(a) for a in sorted(alias, key=len, reverse=True)) _dl_re = re.compile(rf"({alts})下载\s+(\S.*)") _apply_aliases(_STATIC_PLATFORMS) # 内嵌键盘上限：5 行、每行 5 个按钮；歌曲占行 + 1 行翻页 → 每页最多 20 首 _MAX_ROWS = 5 _MAX_BUTTONS_PER_ROW = 5 _MAX_MENU_SONGS = (_MAX_ROWS - 1) * _MAX_BUTTONS_PER_ROW # 菜单条目的封面小图边长（卡片大图是 240px） _MENU_COVER_PX = 80 class MusicServerError(RuntimeError): """MusicServer 不可达或返回非 0 code。""" @dataclass(slots=True) class Song: source: str song_id: str name: str singer: str album: str | None = None interval: str | None = None album_id: str | None = None img: str | None = None # 专辑封面直链（部分源无此字段，如 kw/kg） # 每个群当前"可撤回"的 bot 消息 id（菜单或卡片）：发新消息前撤回上一张， # 播放/下载成功后撤回卡片。撤回与否由 config.withdraw 决定，失败仅记日志 _last_msg: dict[str, str] = {} # 文件代理实例（PUBLIC_URL 配置时由 register 创建、READY 时启动）； # None = 未启用，发送时用原直链 _file_proxy: FileProxy | None = None # 已处理的群消息 id：官方说明同一 msg_id 可能重复推送（需开发者自行排重， # 否则会给用户回两条）；bot 自己发出的消息 id 也记在这里，其回推 # （self-echo）按重复跳过 _seen_messages: dict[str, float] = {} _DEDUP_TTL = 300.0 # 秒；被动回复时效是分钟级，重复推送不会来得更晚 _DEDUP_MAX = 1024 # 条；字典按插入序，超限淘汰最旧 # 消息内容级去重：全量模式下同一条 @消息会同时命中 GROUP_AT_MESSAGE_CREATE # 与 GROUP_MESSAGE_CREATE，两个事件的 msg_id 不同（2026-10-03 实测同一命令 # 双占位双直链），按 id 拦不住；钥匙 = 群|发送人|归一化文本（@ 前缀与斜杠 # 已剥，两个事件归一后一致）。独立成表，不与按钮防抖互挤容量 _MSG_DUP_TTL = 10.0 # 秒；双事件两帧背靠背到达，窗口只需盖住抖动 _MSG_DUP_MAX = 1024 # 条；字典按插入序，超限淘汰最旧 _msg_dup_keys: dict[str, float] = {} # 按钮回调防抖：互动推送自带 uuid（interaction id），同一次点击可能被 # 平台重复推送（uuid 相同）；连点则是不同 uuid、按钮 data 相同——两把 # 钥匙任一在窗口内出现过都只处理第一次 _CLICK_TTL = 10.0 # 秒 _CLICK_MAX = 1024 # 条；字典按插入序，超限淘汰最旧 _recent_clicks: dict[str, float] = {} def _prune(table: dict[str, float], ttl: float, now: float) -> None: """剪掉 table 里过期记录（按插入序，头部最旧）。""" while table: oldest = next(iter(table)) if now - table[oldest] < ttl: break del table[oldest] def _mark(message_id: str) -> None: """记下一个 msg_id（已存在则只刷新时间），超出上限时淘汰最旧。""" now = time.monotonic() _prune(_seen_messages, _DEDUP_TTL, now) _seen_messages[message_id] = now if len(_seen_messages) > _DEDUP_MAX: del _seen_messages[next(iter(_seen_messages))] def _handled(message_id: str) -> bool: """这个 msg_id 近期是否已处理过；否者记下并返回 False。""" _prune(_seen_messages, _DEDUP_TTL, time.monotonic()) if message_id in _seen_messages: return True _mark(message_id) return False def _seen_key(table: dict[str, float], ttl: float, cap: int, key: str) -> bool: """这把钥匙近期（ttl 内）是否已处理过；未处理则记下并返回 False。 表按插入序，超上限淘汰最旧。""" now = time.monotonic() _prune(table, ttl, now) if key in table: return True table[key] = now if len(table) > cap: del table[next(iter(table))] return False def _click_seen(key: str) -> bool: """这个回调钥匙（互动 uuid 或 群+按钮 data）近期是否已处理过； 否则记下并返回 False。""" return _seen_key(_recent_clicks, _CLICK_TTL, _CLICK_MAX, key) def _sent_id(resp: Any) -> str | None: """从发送响应取消息 id（官方字段是 id，部分接口用 message_id）。""" if isinstance(resp, dict): new_id = resp.get("id") or resp.get("message_id") if new_id: return str(new_id) return None def _song_of(item: dict[str, Any]) -> Song | None: source = item.get("source") # id 约定（MusicServer）：kw/tx/wy/qs 用 songmid（qs 是 track id），kg 用 hash song_id = item.get("hash") if source == "kg" else item.get("songmid") if not source or not song_id or not item.get("name"): return None return Song( source=source, song_id=str(song_id), name=item["name"], singer=item.get("singer") or "", album=item.get("albumName"), interval=item.get("interval"), album_id=item.get("albumId"), img=item.get("img") or None, ) def _songs_of(items: list[dict[str, Any]], source: str = "") -> list[Song]: """条目列表转 Song。xm/me/qt/lr 的条目不带 source 字段，按本次请求的 平台补上（请求级信息是权威的）；聚合（source=all）里这类条目无法判源， 只能丢弃。""" out: list[Song] = [] for item in items or []: if source and source != "all" and not item.get("source"): item = {**item, "source": source} if song := _song_of(item): out.append(song) return out async def _get(http: ClientSession, config: BotConfig, path: str, **params: str) -> Any: """请求 MusicServer 并拆开统一信封，code≠0 抛 MusicServerError。""" try: async with http.get( f"{config.music_server_url}{path}", params={k: v for k, v in params.items() if v}, ) as resp: body = await resp.json() except (ClientError, TimeoutError) as e: raise MusicServerError(f"MusicServer 不可达: {e}") from e if not isinstance(body, dict) or body.get("code") != 0: msg = body.get("msg") if isinstance(body, dict) else body raise MusicServerError(f"{path} 响应异常: {msg}") return body.get("data") async def search( http: ClientSession, config: BotConfig, keyword: str, page: int = 1 ) -> Song | None: """按配置源搜索（默认 QQ 音乐），没有结果时聚合全源（source=all）兜底。""" sources = (config.music_source, "all") if config.music_source else ("all",) for source in sources: path = f"/search/{source}/{_q(keyword)}/{page}" data = await _get(http, config, path, limit="5") if songs := _songs_of(data.get("list"), source): return songs[0] return None async def search_list( http: ClientSession, config: BotConfig, keyword: str, page: int = 1 ) -> list[Song]: """按配置源搜索一页候选，配置源没结果时聚合全源兜底（菜单展示用）。""" sources = (config.music_source, "all") if config.music_source else ("all",) for source in sources: data = await _get( http, config, f"/search/{source}/{_q(keyword)}/{page}", limit=str(config.menu_page_size), ) if songs := _songs_of(data.get("list"), source): return songs return [] async def search_source_list( http: ClientSession, config: BotConfig, keyword: str, source: str, page: int = 1, ) -> list[Song]: """在指定平台搜一页候选（「<平台>下载」的菜单用）；不聚合兜底—— 用户点名了平台，兜底到别的源就失了本意。""" data = await _get( http, config, f"/search/{source}/{_q(keyword)}/{page}", limit=str(config.menu_page_size), ) return _songs_of(data.get("list"), source) async def _play_url(http: ClientSession, config: BotConfig, song: Song) -> str: # 附带歌名/歌手/时长：原源取流失败时，服务器按这些信息做酷我兜底匹配 # （kg 可由 hash 反查歌曲信息，只补专辑 id） params = { "quality": config.music_quality, "name": song.name, "singer": song.singer, "interval": song.interval or "", } if song.source == "kg" and song.album_id: params["albumId"] = song.album_id data = await _get(http, config, f"/url/{song.source}/{song.song_id}", **params) # 新版响应带实际命中信息：via=兜底通道（kw/script），type=实际音质（可能降级） via = data.get("via") if via and via != song.source: logger.info( f"「{song.name}」直链经兜底通道取得：{song.source}→{via}" f"（{data.get('type') or config.music_quality}）" ) return data["url"] async def _lyric(http: ClientSession, config: BotConfig, song: Song) -> str: data = await _get( http, config, f"/lyric/{song.source}/{song.song_id}", name=song.name, interval=song.interval or "", ) return data.get("lyric") or "" def _trim(text: str, limit: int = _LYRIC_LIMIT) -> str: """剥离 LRC 时间戳/元数据标签与空行，超长截断。""" lines = [] for raw in text.splitlines(): line = _LRC_TS_RE.sub("", raw).strip() if line and not _LRC_TAG_RE.fullmatch(line): lines.append(line) body = "\n".join(lines) if len(body) <= limit: return body return body[:limit].rstrip() + "\n……(歌词过长已截断)" def _caption(song: Song) -> str: return f"♪ {song.name} - {song.singer}" # Windows 文件名与 HTTP 头都不欢迎的字符，代理 URL 尾段前先清掉 _FILENAME_BAD = re.compile(r'[\\/:*?"<>|\r\n]') def _filename(song: Song) -> str: """代理 URL 尾段的下载文件名（带扩展名，展示与保存用）。""" base = f"{song.name} - {song.singer}".strip(" -") or song.song_id return _FILENAME_BAD.sub("_", base) + ".mp3" def _button(bid: str, label: str, data: str) -> dict: """内嵌回调按钮（type=1）：点击推 INTERACTION_CREATE 给 bot。""" return { "id": bid, "render_data": {"label": label, "visited_label": label, "style": 1}, "action": {"type": 1, "permission": {"type": 2}, "data": data}, } def _q(s: str) -> str: return quote(s, safe="") # 标准百分号编码时保留的 URL 结构字符（RFC 3986 reserved + %）；% 保留使 # 已编码的 URL 幂等，不会二次编码 _URL_SAFE = ":/?#[]@!$&'()*+,;=%" # Markdown 图片语法的 URL 里 ( ) 会提前截断链接、空格会截断属性，故不含 ( ) _MD_URL_SAFE = ":/?#[]@!$&'*+,;=%" def _encode_url(url: str, safe: str = _URL_SAFE) -> str: """把 URL 转成标准百分号编码：只转义空格、中文、引号等 URL 非法字符， 结构字符与已有 %XX 原样保留（幂等，重复调用结果一致）。发出链接前 统一走这里——源站直链/封面地址常带未编码字符，QQ 侧抓取与群友复制 都会被截断。""" return quote(url, safe=safe) def _cmd_button(bid: str, label: str, data: str) -> dict: """指令按钮（type=2）：点击把 data 填进聊天输入框，用户补上参数再发送， 等价于手敲斜杠指令——不产生回调，bot 收到的是普通消息。""" return { "id": bid, "render_data": {"label": label, "visited_label": label, "style": 1}, "action": {"type": 2, "permission": {"type": 2}, "data": data}, } async def _reply( api: BotApi, group_openid: str, content: str, msg_id: str | None = None, event_id: str | None = None, msg_seq: int = 1, ) -> None: if event_id: resp = await api.post_group_message( group_openid, content=content, event_id=event_id, msg_seq=msg_seq ) else: resp = await api.post_group_message( group_openid, content=content, msg_id=msg_id, msg_seq=msg_seq ) if sent := _sent_id(resp): _mark(sent) # 自己的消息若被回推，按重复跳过 async def _soft_reply( api: BotApi, group_openid: str, content: str, event_id: str | None = None, msg_seq: int = 1, ) -> None: """尽力回复：失败（如互动 event_id 失效）仅记日志，让按钮回应照常进行。""" try: await _reply(api, group_openid, content, event_id=event_id, msg_seq=msg_seq) except _API_FAILS as e: logger.warning(f"回复失败: {e!r}") async def _respond(api: BotApi, interaction_id: str) -> None: """回应互动（PUT /interactions）：收到回调几秒内必须应答，否则 QQ 客户端 一直 loading 直到超时；失败仅记日志，不阻塞后续处理。""" try: await api.respond_interaction(interaction_id) except _API_FAILS as e: logger.warning(f"回应互动失败: {e!r}") async def _withdraw(api: BotApi, group_openid: str, message_id: str) -> None: """撤回 bot 自己发的旧消息（超时限/无权限等失败仅记日志）。""" try: await api.withdraw_group_message(group_openid, message_id) except _API_FAILS as e: logger.warning(f"撤回消息失败: {e!r}") async def _retire(api: BotApi, config: BotConfig, group_openid: str) -> None: """撤回本群当前那张 bot 消息（菜单或卡片）并清记录；开关关闭时不撤回。""" old = _last_msg.pop(group_openid, None) if old and config.withdraw: await _withdraw(api, group_openid, old) async def _swap_message( api: BotApi, config: BotConfig, group_openid: str, resp: Any ) -> None: """记下刚发出的消息 id，并撤回同群上一张（菜单/卡片）。发送响应未返回 消息 id 时只跳过记录，不影响主流程。""" await _retire(api, config, group_openid) if new_id := _sent_id(resp): _last_msg[group_openid] = new_id _mark(new_id) # 全量模式下这张卡片可能被回推，别被当成用户消息 def _song_data(action: str, song: Song) -> str: """回调按钮携带的歌曲数据：自包含（歌曲 id/歌名/歌手/封面 % 转义， 避免撞上分隔符），bot 收到回调直接还原 Song，无需记忆菜单快照。 id 也要转义：bili 的 songmid 是「BV号|cid」，裸拼会把 data 拆散。""" return ( f"音乐|{action}|{song.source}|{_q(song.song_id)}" f"|{_q(song.name)}|{_q(song.singer)}|{_q(song.img or '')}" ) def _song_of_data(args: list[str]) -> Song: """从回调 data 的 [source, song_id, name, singer, img?] 还原 Song。 img 段是后加的：旧按钮的 data 只有 4 段，缺省按无封面处理。 song_id 段的 % 转义也是后加的（bili songmid 带 |）：旧按钮 id 不含 %，unquote 幂等，直接解不影响。""" source, song_id, name, singer = args[:4] img = unquote(args[4]) if len(args) > 4 and args[4] else None return Song( source=source, song_id=unquote(song_id), name=unquote(name), singer=unquote(singer), img=img, ) async def _fetchable(http: ClientSession, url: str) -> bool: """直链当前是否可被 bot 抓取（Range 探测：2xx 即视为可下载）。 tx 等源的直链常受 CDN 风控，bot 抓不动时 QQ 服务器侧现场下载同样失败 （40093007），故上传前先探测、不可抓则换下一候选源。""" try: async with http.get(url, headers={"Range": "bytes=0-0"}) as resp: return resp.status in (200, 206) except (ClientError, TimeoutError, OSError): return False def _card_keyboard(song: Song) -> dict: """歌曲卡片键盘：[▶ 播放][⬇ 下载] 回调按钮（点一下直接执行）。""" return { "content": { "rows": [ {"buttons": [ _button("song-play", "▶ 播放", _song_data("play", song)), _button("song-dl", "⬇ 下载", _song_data("dl", song)), ]} ] } } async def _post_markdown( api: BotApi, group_openid: str, lines: list[str], fallback: str, keyboard: dict, msg_id: str | None = None, event_id: str | None = None, ) -> Any: """发一条 Markdown 消息（带键盘），无权限等失败回退纯文本（键盘保留）； 两者都失败返回 None。""" try: return await api.post_group_message( group_openid, markdown="\n".join(lines), keyboard=keyboard, msg_id=msg_id, event_id=event_id, ) except _API_FAILS as e: logger.warning(f"Markdown 发送失败，回退文本: {e!r}") try: return await api.post_group_message( group_openid, content=fallback, keyboard=keyboard, msg_id=msg_id, event_id=event_id, ) except _API_FAILS as e: logger.warning(f"文本发送失败: {e!r}") return None def _menu_keyboard(songs: list[Song], nav: list[dict]) -> dict: """选歌键盘：编号回调按钮每行 5 个（点一下即弹卡片），nav 是调用方 组好的翻页按钮行（上一页/下一页，可为空）。""" rows: list[list[dict]] = [] row: list[dict] = [] for i, song in enumerate(songs[:_MAX_MENU_SONGS], 1): row.append(_button(f"pick-{i}", str(i), _song_data("pick", song))) if len(row) == _MAX_BUTTONS_PER_ROW: rows.append(row) row = [] if row: rows.append(row) if nav: rows.append(nav) return {"content": {"rows": [{"buttons": r} for r in rows]}} def _search_nav( keyword: str, page: int, more: bool, source: str | None = None ) -> list[dict]: """搜索菜单的翻页按钮（是否给下一页以"本页满员"判断）；source 非空是 「<平台>下载」的平台内菜单，翻页走 spage 动作、携带平台锁定同一源。""" action = "spage" if source else "page" head = f"音乐|{action}" + (f"|{source}" if source else "") nav = [] if page > 1: nav.append(_button("prev", "上一页", f"{head}|{_q(keyword)}|{page - 1}")) if more: nav.append(_button("next", "下一页", f"{head}|{_q(keyword)}|{page + 1}")) return nav def _board_nav( source: str, board_id: str, name: str, page: int, more: bool ) -> list[dict]: """榜单菜单的翻页按钮（携带榜单上下文，翻页后仍取同一榜单）。""" nav = [] if page > 1: prev = _board_data(source, board_id, name, page - 1) nav.append(_button("prev", "上一页", prev)) if more: nxt = _board_data(source, board_id, name, page + 1) nav.append(_button("next", "下一页", nxt)) return nav def _board_data(source: str, board_id: str, name: str, page: int) -> str: """榜单回调按钮携带的数据：源、榜单 id、榜单名（% 转义）与页码。""" return f"音乐|board|{source}|{board_id}|{_q(name)}|{page}" def _track_data(source: str, album_id: str, name: str, page: int) -> str: """选集翻页按钮携带的数据：源、节目 id、节目名（% 转义）与页码。""" return f"音乐|track|{source}|{album_id}|{_q(name)}|{page}" def _track_nav( source: str, album_id: str, name: str, page: int, more: bool ) -> list[dict]: """选集菜单的翻页按钮（携带节目上下文，翻页仍取同一节目的选集）。""" nav = [] if page > 1: prev = _track_data(source, album_id, name, page - 1) nav.append(_button("prev", "上一页", prev)) if more: nxt = _track_data(source, album_id, name, page + 1) nav.append(_button("next", "下一页", nxt)) return nav async def _post_menu( api: BotApi, config: BotConfig, group_openid: str, title: str, songs: list[Song], nav: list[dict], msg_id: str | None = None, event_id: str | None = None, ) -> None: """渲染并发送一张选歌菜单：Markdown 歌单（编号 + 歌名/歌手 + 封面小图） + 编号回调按钮 + 翻页行，Markdown 失败回退纯文本（按钮保留、不含图片 语法）；发送后撤同群上一张菜单/卡片。""" shown = songs[:_MAX_MENU_SONGS] lines = [f"## {title}"] texts: list[str] = [] # 纯文本回退行：不含封面图 for i, song in enumerate(shown, 1): entry = f"{song.name} - {song.singer}".strip(" -") texts.append(f"{i}. {entry}") lines.append(f"{i}. {entry}") if song.img: # 与卡片同款图片语法（QQ 后台代理并缓存图片），缩小为条目小图； # kw/kg 等无封面源的条目自然跳过；URL 转义见 _encode_url lines.append( f"![封面 #{_MENU_COVER_PX}px #{_MENU_COVER_PX}px]" f"({_encode_url(song.img, _MD_URL_SAFE)})" ) resp = await _post_markdown( api, group_openid, lines, "\n".join(texts) + "\n（点下方编号选歌）", _menu_keyboard(shown, nav), msg_id, event_id, ) if resp is None: return await _swap_message(api, config, group_openid, resp) logger.info(f"[{group_openid}] 菜单已发：{title}（{len(shown)}首）") async def send_menu( api: BotApi, http: ClientSession, config: BotConfig, group_openid: str, keyword: str, page: int = 1, msg_id: str | None = None, event_id: str | None = None, source: str | None = None, ) -> None: """搜索并展示选歌菜单（渲染见 _post_menu）。source 为空走配置源、无果 聚合兜底（点歌/热搜词）；source 非空是「<平台>下载」的平台内搜索， 不兜底，标题带平台名、翻页锁定该平台。""" try: if source: songs = await search_source_list(http, config, keyword, source, page) else: songs = await search_list(http, config, keyword, page) except MusicServerError as e: logger.warning(f"搜索失败: {e}") await _reply( api, group_openid, _UNAVAILABLE, msg_id, event_id ) return if not songs: if source: name = await _source_display(http, config, source) await _reply( api, group_openid, f"在{name}没找到「{keyword}」", msg_id, event_id ) else: await _reply(api, group_openid, f"没找到「{keyword}」", msg_id, event_id) return title = f"点歌：{keyword}" if source: title = f"{await _source_display(http, config, source)}：{keyword}" await _post_menu( api, config, group_openid, title, songs, _search_nav(keyword, page, len(songs) >= config.menu_page_size, source), msg_id, event_id, ) async def send_card( api: BotApi, http: ClientSession, config: BotConfig, group_openid: str, song: Song, msg_id: str | None = None, event_id: str | None = None, ) -> None: """选中歌曲后弹出卡片：Markdown 封面图 + 歌名/歌手/来源/专辑/时长 + [▶ 播放][⬇ 下载] 回调按钮；Markdown 失败回退纯文本（按钮保留）。""" title = f"{song.name} - {song.singer}".strip(" -") lines = [f"## {title}"] if song.img: # QQ 后台会代理并缓存图片到客户端（不落库），宽高按 #宽px #高px 指定； # URL 转义见 _encode_url lines.append(f"![封面 #240px #240px]({_encode_url(song.img, _MD_URL_SAFE)})") details = [f"来源：{_SOURCE_NAMES.get(song.source, song.source)}"] if song.album: details.append(f"专辑：{song.album}") if song.interval: details.append(f"时长：{song.interval}") lines.append("　".join(details)) resp = await _post_markdown( api, group_openid, lines, "\n".join(lines[1:]) + "\n（点下方按钮播放/下载）", _card_keyboard(song), msg_id, event_id, ) if resp is None: return await _swap_message(api, config, group_openid, resp) logger.info(f"[{group_openid}] 选歌：{title}") async def _upload_group_file( api: BotApi, group_openid: str, file_type: int, url: str ) -> dict: """上传群富媒体（同 BotApi.upload_group_file，但放宽该请求的超时； timeout 经 kwargs 透传给 aiohttp）。""" return await api.post( f"/v2/groups/{group_openid}/files", json={"file_type": file_type, "url": url, "srv_send_msg": False}, timeout=_UPLOAD_TIMEOUT, ) async def send_song( api: BotApi, http: ClientSession, config: BotConfig, group_openid: str, song: Song, keyword: str, voice: bool, msg_id: str | None = None, event_id: str | None = None, msg_seq: int = 1, ) -> None: """取可抓取的播放直链后发送。 voice=True 发语音（上传 file_type=3）。voice=False 的下载看 SEND_MODE： "file" 上传群文件（失败或 40093005 无权限则回退直链），"link" 直接发 直链、根本不尝试上传。 配置源直链抓不动时 QQ 服务器侧下载同样会失败（40093007），故先探测 可抓性、必要时换候选源；keyword 用于候选源搜索。文件代理启用时，最终 发出的 url（上传与直链文本同源）换成 bot 自己的公开地址。 msg_seq 是本事件被动回复的序号：调用方若已先回过占位消息，须传 2 （同一 msg_id/event_id 相同 msg_seq 的第二次回复会被 QQ 拒绝）。 卡片撤回不在这里做：调用方 handle_button 会把这次点击当成一次性操作， 无论成败都撤回卡片。""" found = await _fetchable_play(http, config, song, keyword) if found is None: logger.warning(f"[{group_openid}] 无可用直链：{song.name}") await _reply( api, group_openid, f"没拿到「{song.name}」的播放链接", msg_id, event_id, msg_seq, ) return _, url = found if _file_proxy is not None and _file_proxy.ready: # 换成 bot 自己的公开 URL：QQ 服务器抓风控直链 / qs 本地流会失败， # 经 bot 中转可抓。请求到来时现场重解析（直链有时效，刚探测的这条 # 不能复用）。服务没起来（端口被占）则不换链、仍发原直链。 url = _file_proxy.claim(song, keyword, _filename(song)) # 发给 QQ 前统一转标准百分号编码：源站直链常带空格/中文等未编码字符， # 上传与直链文本用同一条（代理 URL 里的文件名已是编码态，此处幂等） url = _encode_url(url) # 下载在 link 模式下跳过上传（群文件上传仅限群主/管理员，40093005） skip_upload = not voice and config.send_mode == "link" no_permission = False if not skip_upload: label, file_type = ("语音", 3) if voice else ("文件", 4) try: upload = await _upload_group_file(api, group_openid, file_type, url) resp = await api.post_group_message( group_openid, content=_caption(song), msg_id=msg_id, event_id=event_id, msg_seq=msg_seq, media=upload, ) if sent := _sent_id(resp): _mark(sent) logger.info(f"[{group_openid}] {label}已发：{_caption(song)}") return except _API_FAILS as e: no_permission = isinstance(e, ApiError) and e.code == _NO_FILE_PERMISSION if no_permission: logger.warning(f"[{group_openid}] 群文件上传无权限，改为直链文本") else: logger.warning(f"[{group_openid}] 以{label}发送失败: {e!r}") # 回退直链文本；文件权限问题多给一句说明 text = f"{_caption(song)}\n{url}" if no_permission: text = f"{_caption(song)}\n（本群文件上传仅限群主/管理员，先给直链）\n{url}" try: await _reply(api, group_openid, text, msg_id, event_id, msg_seq) logger.info(f"[{group_openid}] 直链已发：{_caption(song)}") except _API_FAILS as e: # 被动回复被拒（如互动 event_id 超时效）→ 降级为主动消息 logger.warning(f"[{group_openid}] 被动回复失败，降级为主动消息: {e!r}") try: await _reply(api, group_openid, text) except _API_FAILS as e: logger.warning(f"[{group_openid}] 文本链接发送失败: {e!r}") async def send_lyric( api: BotApi, http: ClientSession, config: BotConfig, group_openid: str, song: Song, msg_id: str | None = None, event_id: str | None = None, ) -> None: try: text = await _lyric(http, config, song) except MusicServerError as e: # xm/me/qt/lr 等音频平台没有歌词能力，服务器报非 0 code logger.warning(f"取歌词失败（{song.source}）: {e}") name = _SOURCE_NAMES.get(song.source, song.source) await _reply( api, group_openid, f"「{name}」这条没有歌词", msg_id, event_id, ) return if not text.strip(): await _reply( api, group_openid, f"没找到「{song.name}」的歌词", msg_id, event_id ) return await _reply(api, group_openid, _trim(text), msg_id, event_id) async def _send_help(api: BotApi, group_openid: str, msg_id: str) -> None: """发帮助：正文 + 指令按钮（输入框 / 菜单的入口）；按钮被拒回退纯文本。""" try: resp = await api.post_group_message( group_openid, content=_HELP, msg_id=msg_id, keyboard=_help_keyboard() ) except _API_FAILS as e: logger.warning(f"带指令按钮的帮助发送失败，回退纯文本: {e!r}") try: resp = await api.post_group_message( group_openid, content=_HELP, msg_id=msg_id ) except _API_FAILS as e: logger.warning(f"帮助发送失败: {e!r}") return if sent := _sent_id(resp): _mark(sent) def _help_keyboard() -> dict: """帮助键盘：指令按钮，点一下把斜杠指令填进输入框。""" return { "content": { "rows": [ {"buttons": [ _cmd_button("help-cmd-song", "🎤 点歌", "/点歌 "), _cmd_button("help-cmd-lyric", "📝 歌词", "/歌词 "), _cmd_button("help-cmd-hot", "🔥 热搜", "/热搜"), _cmd_button("help-cmd-board", "🏆 榜单", "/榜单"), ]} ] } } async def _source_display(http: ClientSession, config: BotConfig, source: str) -> str: """平台显示名：优先 /sources 报告的 name，缺省退回静态表。""" table = await platforms(http, config) platform = table.get(source) if platform and platform.get("name"): return platform["name"] return _SOURCE_NAMES.get(source, source) async def _feature_source( http: ClientSession, config: BotConfig, cap: str ) -> tuple[str, str] | None: """挑一个具备指定能力（hotSearch/boards）的平台：配置源优先，否则按 服务器顺序取第一个；/sources 不可达时退回静态表。返回 (平台 id, 显示名)。""" table = await platforms(http, config) or _STATIC_PLATFORMS if config.music_source in table and _cap(table[config.music_source], cap): pid = config.music_source else: pid = next((p for p in table if _cap(table[p], cap)), "") if not pid: return None name = table[pid].get("name") or _SOURCE_NAMES.get(pid, pid) return pid, name async def _log_login_status(http: ClientSession, config: BotConfig) -> None: """READY 时报告各平台登录态（部分源登录后才有高音质/完整曲库）。""" try: data = await _get(http, config, "/login/status") except MusicServerError: return # 老版 MusicServer 没有该接口，不报 logged = [ pid for pid, v in (data or {}).items() if isinstance(v, dict) and v.get("logged") ] state = " ".join(logged) if logged else "全部未登录" logger.info(f"MusicServer 登录态：{state}") async def send_hot( api: BotApi, http: ClientSession, config: BotConfig, group_openid: str, msg_id: str | None = None, event_id: str | None = None, ) -> None: """热搜：拉某平台的热搜词，编号列表 + 词按钮，点词直接出选歌菜单。""" picked = await _feature_source(http, config, "hotSearch") if picked is None: await _reply(api, group_openid, "当前音源不支持热搜", msg_id, event_id) return source, name = picked try: data = await _get(http, config, f"/hot-search/{source}") except MusicServerError as e: logger.warning(f"取热搜失败: {e}") await _reply(api, group_openid, _UNAVAILABLE, msg_id, event_id) return words = [w for w in data.get("list") or [] if w][:_MAX_MENU_SONGS] if not words: await _reply(api, group_openid, "没拿到热搜词", msg_id, event_id) return rows: list[list[dict]] = [] row: list[dict] = [] for i, word in enumerate(words, 1): row.append(_button(f"hot-{i}", word, f"音乐|hot|{_q(word)}")) if len(row) == _MAX_BUTTONS_PER_ROW: rows.append(row) row = [] if row: rows.append(row) lines = [f"## {name}热搜"] + [f"{i}. {w}" for i, w in enumerate(words, 1)] resp = await _post_markdown( api, group_openid, lines, "\n".join(lines[1:]) + "\n（点下方热词直接搜）", {"content": {"rows": [{"buttons": r} for r in rows]}}, msg_id, event_id, ) if resp is None: return await _swap_message(api, config, group_openid, resp) logger.info(f"[{group_openid}] 热搜已发：{name} {len(words)}词") async def send_boards( api: BotApi, http: ClientSession, config: BotConfig, group_openid: str, msg_id: str | None = None, event_id: str | None = None, ) -> None: """榜单：拉某平台的榜单列表，点榜单名直接出该榜单的选歌菜单。""" picked = await _feature_source(http, config, "boards") if picked is None: await _reply(api, group_openid, "当前音源不支持榜单", msg_id, event_id) return source, name = picked try: data = await _get(http, config, f"/boards/{source}") except MusicServerError as e: logger.warning(f"取榜单列表失败: {e}") await _reply(api, group_openid, _UNAVAILABLE, msg_id, event_id) return boards = [ b for b in data.get("list") or [] if b.get("id") and b.get("name") ][:_MAX_MENU_SONGS] if not boards: await _reply(api, group_openid, "没拿到榜单", msg_id, event_id) return rows: list[list[dict]] = [] row: list[dict] = [] for i, board in enumerate(boards, 1): label = board["name"][:8] # 按钮标签过长会被 QQ 截断，先自己裁 btn_data = _board_data(source, board["id"], board["name"], 1) row.append(_button(f"board-{i}", label, btn_data)) if len(row) == _MAX_BUTTONS_PER_ROW: rows.append(row) row = [] if row: rows.append(row) lines = [f"## {name}榜单"] + [f"{i}. {b['name']}" for i, b in enumerate(boards, 1)] resp = await _post_markdown( api, group_openid, lines, "\n".join(lines[1:]) + "\n（点下方榜单直接听）", {"content": {"rows": [{"buttons": r} for r in rows]}}, msg_id, event_id, ) if resp is None: return await _swap_message(api, config, group_openid, resp) logger.info(f"[{group_openid}] 榜单列表已发：{name} {len(boards)}个") def _album_id_of(song: Song) -> str | None: """节目级条目（一个节目多个音频）的选集 id；单曲级条目返回 None。 me 搜索里节目条目的 songmid 带 d: 前缀（d:44753 → 选集 id 44753）， 声音条目是普通 id；lr/qt 搜索返回的整条都是书/频道（id 即选集 id， 直接取链 502）；xm 返回单集级（带真实时长），不在此列。""" if song.source == "me" and song.song_id.startswith("d:"): return song.song_id[2:] if song.source in ("lr", "qt"): return song.song_id return None # /album 的翻页行为按源不同（2026-10-03 实测）：me 忽略 page 参数、一次给足 # limit（50 可用、200 报 422）→ 本地切片分页；lr/qt 服务端翻页正常 → 直取 # 当页。me 超过 50 集的节目会被截断（猫耳广播剧分季、很少超过） _ALBUM_LOCAL_PAGE = frozenset({"me"}) _ALBUM_FETCH_LIMIT = 50 async def send_track_menu( api: BotApi, http: ClientSession, config: BotConfig, group_openid: str, source: str, album_id: str, name: str, page: int = 1, event_id: str | None = None, ) -> None: """拉节目的选集列表并出选集菜单（渲染同选歌菜单）：点编号弹该集卡片、 再点播放/下载。选集条目的歌手就是节目名（展示冗余）、时长常为 0，都抹掉。""" try: if source in _ALBUM_LOCAL_PAGE: data = await _get( http, config, f"/album/{source}/{album_id}", limit=str(_ALBUM_FETCH_LIMIT), ) episodes = _songs_of(data.get("list"), source) # 服务器截断（>50 集）时以实拿为准，别让翻页按钮指向空页 total = data.get("total") if not (isinstance(total, int) and total <= len(episodes)): total = len(episodes) start = (page - 1) * config.menu_page_size songs = episodes[start:start + config.menu_page_size] more = total > page * config.menu_page_size else: data = await _get( http, config, f"/album/{source}/{album_id}/{page}", limit=str(config.menu_page_size), ) songs = _songs_of(data.get("list"), source) total = data.get("total") more = ( total > page * config.menu_page_size if isinstance(total, int) else len(songs) >= config.menu_page_size ) except MusicServerError as e: logger.warning(f"取选集失败（{source}）: {e}") await _reply(api, group_openid, _UNAVAILABLE, event_id=event_id) return if not songs: if page > 1: await _reply(api, group_openid, "没有更多了", event_id=event_id) else: await _reply( api, group_openid, f"「{name}」拿不到剧集列表（可能已下线或需要登录）", event_id=event_id, ) return songs = [ replace( s, singer="" if s.singer == name else s.singer, interval=None if s.interval in ("", "0") else s.interval, ) for s in songs ] title = f"选集：{name}" if isinstance(total, int): title += f"（共{total}集）" await _post_menu( api, config, group_openid, title, songs, _track_nav(source, album_id, name, page, more), event_id=event_id, ) async def send_board_menu( api: BotApi, http: ClientSession, config: BotConfig, group_openid: str, source: str, board_id: str, board_name: str, page: int, msg_id: str | None = None, event_id: str | None = None, ) -> None: """展示某榜单某页的选歌菜单（渲染同搜索菜单，翻页携带榜单上下文）。""" try: data = await _get( http, config, f"/boards/{source}/{board_id}/{page}", limit=str(config.menu_page_size), ) except MusicServerError as e: logger.warning(f"取榜单曲目失败: {e}") await _reply(api, group_openid, _UNAVAILABLE, msg_id, event_id) return songs = _songs_of(data.get("list"), source) if not songs: await _reply(api, group_openid, "榜单暂时拿不到", msg_id, event_id) return # 服务器给了总数就按总数判断翻页，没给则退回"本页满员"启发式 total = data.get("total") more = ( total > page * config.menu_page_size if isinstance(total, int) else len(songs) >= config.menu_page_size ) await _post_menu( api, config, group_openid, f"榜单：{board_name}", songs, _board_nav(source, board_id, board_name, page, more), msg_id, event_id, ) async def panel_items(http: ClientSession, config: BotConfig) -> list[dict]: """指令面板条目：基础指令 + 热搜/榜单（有平台支持才注册）+ 各平台 「<平台>下载」（search+url 齐备才注册），帮助垫底。平台表来自 /sources（不可达退静态表），服务器增删源自动跟着变。 接口长度限制按宽度算（ASCII 1、汉字 2）：name ≤14、desc ≤30。 指令名超宽时依次换更短的别名（如 Apple Music → Apple）， 换无可换只能放弃该条；「帮助」永远垫底，不被 20 条上限挤掉。""" table = await platforms(http, config) or _STATIC_PLATFORMS items = [ {"name": "点歌", "type": "command", "desc": "出选歌菜单，点编号选歌"}, {"name": "歌词", "type": "command", "desc": "查看歌词"}, ] if any(_cap(p, "hotSearch") for p in table.values()): items.append({ "name": "热搜", "type": "command", "desc": "大家都在搜什么，点词直接搜", }) if any(_cap(p, "boards") for p in table.values()): items.append({ "name": "榜单", "type": "command", "desc": "热门榜单，点开就是选歌菜单", }) for pid, platform in table.items(): if not (_cap(platform, "search") and _cap(platform, "url")): continue name = platform.get("name") or _SOURCE_NAMES.get(pid, pid) # 指令名取装得下面板宽度上限的最顺口别名（惯用名 > 短名 > 全名）， # 如 酷狗/QQ音乐/网易云/Apple cmd = next( (a for a in reversed(_aliases_of(pid, name)) if _width(f"{a}下载") <= 14), None, ) if cmd is None: continue items.append({ "name": f"{cmd}下载", "type": "command", "desc": f"在{name}搜歌出选歌菜单", }) # 面板接口上限 20 条：给「帮助」留最后一位，下载指令多时截前面 items = items[:19] items.append({"name": "帮助", "type": "command", "desc": "查看帮助"}) return [ {**item, "desc": _fit_width(item["desc"], 30)} for item in items if _width(item["name"]) <= 14 ] async def _search_source( http: ClientSession, config: BotConfig, keyword: str, source: str ) -> Song | None: """在指定源内搜索（不触发聚合兜底）。""" data = await _get(http, config, f"/search/{source}/{_q(keyword)}", limit="5") songs = _songs_of(data.get("list"), source) return songs[0] if songs else None # 直链回退源的优先顺序：tx 等源直链常受 CDN 风控抓不动，kw 实测可抓。 # 实际候选取 MusicServer /sources 报告的、具备 search+url 能力的平台： # 先按此优先序排，其余平台追加在后，服务器增删源无需改代码。 # tx 的 url 能力位只代表「服务器能给直链」，不代表直链抓得动——它是 # 风控惯犯（回退机制的本因），绝不作为候选源，否则回退末尾几乎必然 # 白探测一次。qs 同理不进候选：匿名只有 30s 试听，登录态直链是 # MusicServer 本地解密流代理（/stream/qs，QQ 服务器抓不到） _SOURCE_PREFERENCE = ("kw", "wy") _NO_FALLBACK_SOURCES = frozenset({"tx", "qs"}) # /sources 的进程内缓存，按 http 会话隔离（config 是 frozen 的挂不了缓存）。 # 值为 (到期时刻, 平台表)；平台表为 None 表示上游不可用，用短 TTL 防止 # 每次回退都白打一遍请求 _PLATFORMS_TTL = 300.0 _PLATFORMS_FAIL_TTL = 60.0 _platforms_cache: weakref.WeakKeyDictionary[ ClientSession, tuple[float, dict[str, dict] | None] ] = weakref.WeakKeyDictionary() async def platforms(http: ClientSession, config: BotConfig) -> dict[str, dict]: """MusicServer 支持的平台表（id → /sources 条目），带 TTL 缓存； 上游不可用时返回空表，调用方需自行退化。""" hit = _platforms_cache.get(http) now = time.monotonic() if hit and hit[0] > now: return hit[1] or {} try: data = await _get(http, config, "/sources") except MusicServerError as e: logger.warning(f"取 MusicServer 平台列表失败: {e}") _platforms_cache[http] = (now + _PLATFORMS_FAIL_TTL, None) return {} table = {p["id"]: p for p in data.get("platforms") or [] if p.get("id")} _platforms_cache[http] = (now + _PLATFORMS_TTL, table) # 顺带刷新「<平台>下载」别名表：服务器增删平台/改名即自动跟进 _apply_aliases(table) return table async def _fallback_sources( http: ClientSession, config: BotConfig, song: Song ) -> list[str]: """直链回退候选源：/sources 里具备 search+url 能力的平台，优先序在 前、其余追加在后，并排除原曲所在源；/sources 不可用时退回 _SOURCE_PREFERENCE。""" table = await platforms(http, config) capable = [ pid for pid, p in table.items() if _cap(p, "search") and _cap(p, "url") ] if not capable: return [s for s in _SOURCE_PREFERENCE if s != song.source] ordered = [s for s in _SOURCE_PREFERENCE if s in capable] ordered += [s for s in capable if s not in ordered] return [ s for s in ordered if s != song.source and s not in _NO_FALLBACK_SOURCES ] async def _fetchable_play( http: ClientSession, config: BotConfig, song: Song, keyword: str, ) -> tuple[Song, str] | None: """逐候选源取可直接抓取的播放直链：先原曲，抓不动再按候选源 （/sources 动态解析，kw 优先）逐个搜索探测（懒探测：原曲可用时 不产生候选源搜索）。全失败返回 None。""" fallbacks = iter(await _fallback_sources(http, config, song)) cand: Song | None = song while cand is not None: try: url = await _play_url(http, config, cand) except MusicServerError as e: logger.warning(f"取播放链接失败（{cand.source}）: {e}") else: if await _fetchable(http, url): return cand, url logger.warning(f"直链不可抓取（{cand.source}），尝试下一候选源") cand = None while cand is None: source = next(fallbacks, None) if source is None: break cand = await _search_source(http, config, keyword, source) logger.warning("所有候选源的直链都不可抓取") return None async def handle_button( api: BotApi, http: ClientSession, config: BotConfig, group_openid: str, interaction_id: str, data: str, event_id: str | None = None, ) -> bool: """处理内嵌按钮回调：page 翻页 / spage 平台内翻页（「<平台>下载」菜单）/ hot 热搜词搜歌 / board 榜单翻页 / track 选集翻页 / pick 选歌（歌曲直接 弹卡片；节目级条目先出选集菜单）/ play 发语音 / dl 发文件；返回是否为 本 bot 的按钮（非本 bot 的按钮不处理也不回应）。 interaction_id 仅用于回应互动（PUT /interactions）；消息被动回复的 event_id 须用事件最外层 id（"INTERACTION_CREATE:"），用 d.id 裸 uuid 会被 QQ 判为无效（40034025）。 进门先应答互动再干活（几秒内不应答客户端就报超时）；play/dl 取直链 可能要几十秒，先回一条「正在获取」占位，结果以 msg_seq=2 补发。同一 互动 uuid（重复推送）或同群同按钮 data（连点）在防抖窗口内只处理 第一次。""" parts = data.split("|") action, args = (parts[1] if len(parts) > 1 else ""), parts[2:] # 歌曲按钮的 args 为 [source, song_id, name, singer]（+ img，后加故兼容 4 段） is_song = action in ("pick", "play", "dl") and len(args) in (4, 5) known = ( (action == "page" and len(args) == 2) or (action == "spage" and len(args) == 3) or (action == "hot" and len(args) == 1) or (action == "board" and len(args) == 4) or (action == "track" and len(args) == 4) or is_song ) if not known: return False # 非本 bot 的按钮 # 防抖：uuid 重复推送说明已处理并应答过，直接跳过；连点是新 uuid， # 也要应答一次把客户端的 loading 关掉，只是不再重复执行 if _click_seen(interaction_id): logger.debug(f"[{group_openid}] 回调重复推送，跳过：{data[:40]}") return True if _click_seen(f"{group_openid}|{data}"): logger.info(f"[{group_openid}] 按钮连点，防抖窗口内已处理过，跳过") await _respond(api, interaction_id) return True # 先应答互动，再做慢活（取直链/换源探测/上传都可能在几秒开外） await _respond(api, interaction_id) seq = 1 if action in ("play", "dl"): # 同一事件多次回复需递增 msg_seq，先给占位消息把结果让到 seq=2 await _soft_reply( api, group_openid, f"正在获取「{_caption(_song_of_data(args))}」，请稍候…", event_id, ) seq = 2 try: if action == "page": keyword, page = unquote(args[0]), int(args[1]) await send_menu( api, http, config, group_openid, keyword, page, event_id=event_id, ) elif action == "spage": source, keyword, page = args[0], unquote(args[1]), int(args[2]) await send_menu( api, http, config, group_openid, keyword, page, event_id=event_id, source=source, ) elif action == "hot": await send_menu( api, http, config, group_openid, unquote(args[0]), event_id=event_id, ) elif action == "board": source, board_id = args[0], args[1] await send_board_menu( api, http, config, group_openid, source, board_id, unquote(args[2]), int(args[3]), event_id=event_id, ) elif action == "track": source, album_id = args[0], args[1] await send_track_menu( api, http, config, group_openid, source, album_id, unquote(args[2]), int(args[3]), event_id=event_id, ) elif action == "pick": song = _song_of_data(args) if album_id := _album_id_of(song): # 节目级条目（猫耳广播剧 / 懒人书 / 蜻蜓频道）：先出选集菜单 await send_track_menu( api, http, config, group_openid, song.source, album_id, song.name, event_id=event_id, ) else: await send_card( api, http, config, group_openid, song, event_id=event_id, ) else: # play / dl（进门已校验 is_song） song = _song_of_data(args) try: await send_song( api, http, config, group_openid, song, keyword=f"{song.name} {song.singer}".strip(), voice=action == "play", event_id=event_id, msg_seq=seq, ) finally: # 卡片是一次性 UI：这次点击已被消费，无论成败都撤回 # （放 finally 才能覆盖 send_song 抛错，如候选源搜索失败） await _retire(api, config, group_openid) except (MusicServerError, ValueError) as e: logger.warning(f"按钮处理失败: {e}") await _soft_reply( api, group_openid, _UNAVAILABLE, event_id, seq ) except _API_FAILS as e: logger.warning(f"按钮处理失败: {e!r}") return True async def _find( api: BotApi, http: ClientSession, config: BotConfig, group_openid: str, msg_id: str, keyword: str, ) -> Song | None: """搜索并就地回复失败情况（没找到 / 服务不可用），命中返回歌曲。""" try: song = await search(http, config, keyword) except MusicServerError as e: logger.warning(f"搜索失败: {e}") await _reply(api, group_openid, _UNAVAILABLE, msg_id=msg_id) return None if song is None: await _reply(api, group_openid, f"没找到「{keyword}」", msg_id=msg_id) return song def _normalize(content: str) -> str: """统一命令文本：剥掉 @机器人 前缀与开头的斜杠。 斜杠指令（开放平台「指令管理」配置后，群里输入框 / 唤出菜单触发） 推来的 content 形如 "/点歌 稻香"；@机器人是 "<@!bot> 点歌 稻香"； 全量模式是裸文本——三者归一后共用同一套命令解析。""" text = _MENTION_RE.sub("", content, count=1).strip() return text[1:].strip() if text.startswith("/") else text async def handle_command( api: BotApi, http: ClientSession, config: BotConfig, group_openid: str, msg_id: str, content: str, ) -> bool: """解析一条群消息里的音乐命令并执行；返回是否命中。 裸文本（点歌/歌词/热搜/榜单/帮助/<平台>下载）、@机器人 与斜杠指令 （/点歌 …）共用这里；选歌/播放/下载由内嵌按钮（回调）承担，这里只 处理文字命令。""" text = _normalize(content) if match := _SONG_RE.search(text): keyword = match.group(1).strip() logger.info(f"[{group_openid}] 点歌：{keyword}") await send_menu(api, http, config, group_openid, keyword, msg_id=msg_id) return True if text == "点歌": await _reply(api, group_openid, "用法：点歌 <歌名 [歌手]>", msg_id=msg_id) return True if match := _LYRIC_RE.search(text): keyword = match.group(1).strip() logger.info(f"[{group_openid}] 查歌词：{keyword}") if song := await _find(api, http, config, group_openid, msg_id, keyword): await send_lyric(api, http, config, group_openid, song, msg_id=msg_id) return True if text == "歌词": await _reply(api, group_openid, "用法：歌词 <歌名>", msg_id=msg_id) return True if match := _dl_re.search(text): # 「<平台>下载 <歌名>」：在指定平台搜歌出选歌菜单（与点歌同款交互， # 点编号弹卡片再播放/下载）；别名表来自 /sources（platforms 刷新时重建） alias, keyword = match.group(1), match.group(2).strip() source = _alias_to_source[alias] logger.info(f"[{group_openid}] 指定源点歌（{source}）：{keyword}") await send_menu( api, http, config, group_openid, keyword, msg_id=msg_id, source=source, ) return True if text.endswith("下载") and text[:-2] in _alias_to_source: await _reply( api, group_openid, f"用法：{text[:-2]}下载 <歌名>", msg_id=msg_id ) return True if text == "热搜": logger.info(f"[{group_openid}] 看热搜") await send_hot(api, http, config, group_openid, msg_id=msg_id) return True if text == "榜单": logger.info(f"[{group_openid}] 看榜单") await send_boards(api, http, config, group_openid, msg_id=msg_id) return True if text == "帮助" or "音乐帮助" in text: logger.info(f"[{group_openid}] 查看帮助") await _send_help(api, group_openid, msg_id) return True return False async def handle_group_message( api: BotApi, http: ClientSession, config: BotConfig, msg: GroupAtMessage, ) -> bool: """处理一条群消息（@机器人 与全量模式共用）：跳过机器人自己的消息和 重复推送（同 msg_id 重推，以及全量模式下 @消息双事件的不同 msg_id）， 再交给 handle_command；返回是否命中音乐命令。 全量模式需在开放平台开通「接收所有消息」，之后群里每条消息都会推送 GROUP_MESSAGE_CREATE（不限 @机器人）；未命中命令时保持沉默。该事件的 被动回复只能用 msg_id（官方 event_id 支持列表不含此事件）。 自己发出的消息会被回推（腾讯官方 Node SDK 的 self-echo 说明），拦截 有两层：author.bot 标记，以及发送成功时记下响应里的消息 id。""" if msg.author is not None and msg.author.bot: return False # 别的机器人与自己的消息都不理会，避免自触发 if not (msg.group_openid and msg.id and msg.content): return False if _handled(msg.id): logger.debug(f"跳过重复推送的消息: {msg.id}") return False # 同一条 @消息在双事件下 msg_id 不同，id 去重拦不住；两个事件归一化 # （剥 @ 与 /）后文本一致，再按「群+发送人+文本」短窗去重一把 dup_key = f"{msg.group_openid}|{msg.user_openid or ''}|{_normalize(msg.content)}" if _seen_key(_msg_dup_keys, _MSG_DUP_TTL, _MSG_DUP_MAX, dup_key): logger.debug(f"跳过跨事件重复的消息: {msg.id}") return False return await handle_command( api, http, config, msg.group_openid, msg.id, msg.content ) def register(ee: EventEmitter | Bot, config: BotConfig, http: ClientSession) -> None: """把点歌命令与按钮回调 handler 挂到事件总线（生产传 Bot 门面， 测试传裸 EventEmitter，二者 on 语义一致）；http 为 bot 自建连接池 （不受 SDK 的 TIMEOUT 约束），经闭包供 handler 使用——不能用形参注入， 否则拿到的是 SDK 那个 5s 超时的连接池。""" global _file_proxy _file_proxy = None logger.info( f"点歌已就绪：源={config.music_source or '聚合'} 音质={config.music_quality} " f"每页={config.menu_page_size} 下载={config.send_mode} " f"代理={config.public_url or '关'} " f"自动撤回={'开' if config.withdraw else '关'}" ) if config.file_proxy: proxy = FileProxy( config.public_url, http, lambda song, keyword: _fetchable_play(http, config, song, keyword), port=config.proxy_port, ) _file_proxy = proxy # 跟着连接走：READY（含重连）时启动，幂等；起失败仅记日志， # 发送时 ready=False 自动退回发原直链 @ee.on("READY") async def on_ready_file_proxy(_: Ready, api: BotApi) -> None: await proxy.start() # @机器人 的消息走前一个事件；开通「接收所有消息」后群内全部消息走 # 全量事件，两个事件字段一致，共用同一个 handler @ee.on("GROUP_AT_MESSAGE_CREATE") @ee.on("GROUP_MESSAGE_CREATE") async def on_group_message(msg: GroupAtMessage, api: BotApi) -> None: await handle_group_message(api, http, config, msg) @ee.on("INTERACTION_CREATE") async def on_button(act: Interaction, ev: Event, api: BotApi) -> None: if act.chat_type == 1 and act.id and act.group_openid: logger.info(f"[{act.group_openid}] 按钮：{act.button_data[:40]}") # 被动回复的 event_id 用网关帧最外层 id（Event.event_id）， # 裸 uuid（act.id）只用于回应互动 await handle_button( api, http, config, act.group_openid, act.id, act.button_data, event_id=ev.event_id or act.id, ) # 服务器发现 + 指令面板跟着连接走：每次建连（READY）先拉一次 /sources # （TTL 缓存内直读），日志报告平台与登录态、校验配置源，再按平台表 # 幂等同步指令面板（基础指令 + 热搜/榜单 + 各平台「<平台>下载」）。 # 顺序有讲究：面板条目由平台表生成，发现必须在前；RESUME 续传不重发 # READY，不会重复刷 @ee.on("READY") async def on_ready_services(_: Ready, api: BotApi) -> None: table = await platforms(http, config) if table: names = " ".join( f"{pid}({p.get('name') or pid})" for pid, p in table.items() ) logger.info(f"MusicServer 平台：{names}") if config.music_source and config.music_source not in table: logger.warning( f"配置源 {config.music_source} 不在上述平台里，点歌/歌词会报错" ) await _log_login_status(http, config) if config.command_panel: await sync_command_panel(api, await panel_items(http, config))