snowy-yang
324fe86962
- 同一互动 uuid（平台重复推送）或同群同按钮 10 秒内只处理一次；连点 仍应答互动，关掉客户端 loading - 播放/下载结果以 msg_seq=2 补发（同一事件相同 msg_seq 的二次回复会被 QQ 拒绝） - 一并入库此前未提交的 /sources 平台发现（直链回退链按服务器能力动态 解析）；uv.lock 随上游 qqbotsdk main 更新 sdist hash
977 lines
38 KiB
Python
977 lines
38 KiB
Python
"""点歌业务：MusicServer 搜索、Markdown 选歌菜单、点选后的歌曲卡片、歌词清洗。
|
||
|
||
菜单只负责选歌：每页 N 首（MENU_PAGE_SIZE，默认 10）、编号回调按钮 + 上下页。
|
||
点编号 → 弹歌曲卡片（歌名/来源 + [▶播放][⬇下载] 回调按钮），卡片上再点才执行。
|
||
播放发语音、下载发文件，失败均回退直链文本。
|
||
|
||
按钮回调进门先回应互动（防 QQ 客户端等超时），同一互动 uuid / 同群同按钮
|
||
10 秒内只处理一次（防抖）；播放/下载先回「正在获取」占位，结果再补发。
|
||
|
||
自动撤回（翻页撤旧菜单、选歌后撤菜单、播放/下载后撤卡片）由 WITHDRAW
|
||
开关统一控制，默认开。"""
|
||
|
||
import re
|
||
import time
|
||
import weakref
|
||
from dataclasses import dataclass
|
||
from typing import Any
|
||
from urllib.parse import quote, unquote
|
||
|
||
from aiohttp import ClientError, ClientSession, ClientTimeout
|
||
from loguru import logger
|
||
from qqbotsdk import 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
|
||
|
||
_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"
|
||
"输入框敲 / 也能唤出这些指令"
|
||
)
|
||
|
||
# MusicServer（lx-music 聚合源）实际支持的四个平台，没有 mg（咪咕）
|
||
_SOURCE_NAMES = {
|
||
"tx": "QQ音乐", "kw": "酷我", "kg": "酷狗", "wy": "网易云",
|
||
}
|
||
|
||
# 内嵌键盘上限：5 行、每行 5 个按钮；歌曲占行 + 1 行翻页 → 每页最多 20 首
|
||
_MAX_ROWS = 5
|
||
_MAX_BUTTONS_PER_ROW = 5
|
||
_MAX_MENU_SONGS = (_MAX_ROWS - 1) * _MAX_BUTTONS_PER_ROW
|
||
|
||
|
||
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] = {}
|
||
|
||
# 已处理的群消息 id：官方说明同一 msg_id 可能重复推送（需开发者自行排重，
|
||
# 否则会给用户回两条），全量模式下 @消息还可能同时命中两个事件，故统一
|
||
# 去重；bot 自己发出的消息 id 也记在这里，其回推（self-echo）按重复跳过
|
||
_seen_messages: dict[str, float] = {}
|
||
_DEDUP_TTL = 300.0 # 秒；被动回复时效是分钟级，重复推送不会来得更晚
|
||
_DEDUP_MAX = 1024 # 条；字典按插入序，超限淘汰最旧
|
||
|
||
# 按钮回调防抖：互动推送自带 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 _click_seen(key: str) -> bool:
|
||
"""这个回调钥匙（互动 uuid 或 群+按钮 data）近期是否已处理过；
|
||
否则记下并返回 False。"""
|
||
_prune(_recent_clicks, _CLICK_TTL, time.monotonic())
|
||
if key in _recent_clicks:
|
||
return True
|
||
_recent_clicks[key] = time.monotonic()
|
||
if len(_recent_clicks) > _CLICK_MAX:
|
||
del _recent_clicks[next(iter(_recent_clicks))]
|
||
return False
|
||
|
||
|
||
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 README）：kw/tx/wy/mg 用 songmid，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,
|
||
)
|
||
|
||
|
||
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 音乐），没有结果时聚合全源兜底。"""
|
||
for source in (config.music_source, "") if config.music_source else ("",):
|
||
data = await _get(
|
||
http, config, "/search",
|
||
name=keyword, source=source, page=str(page), limit="5",
|
||
)
|
||
for item in data.get("list") or []:
|
||
if song := _song_of(item):
|
||
return song
|
||
return None
|
||
|
||
|
||
async def search_list(
|
||
http: ClientSession, config: BotConfig, keyword: str, page: int = 1
|
||
) -> list[Song]:
|
||
"""按配置源搜索一页候选，配置源没结果时聚合全源兜底（菜单展示用）。"""
|
||
for source in (config.music_source, "") if config.music_source else ("",):
|
||
data = await _get(
|
||
http, config, "/search",
|
||
name=keyword, source=source, page=str(page),
|
||
limit=str(config.menu_page_size),
|
||
)
|
||
songs = [s for item in data.get("list") or [] if (s := _song_of(item))]
|
||
if songs:
|
||
return songs
|
||
return []
|
||
|
||
|
||
async def _play_url(http: ClientSession, config: BotConfig, song: Song) -> str:
|
||
params = {"quality": config.music_quality}
|
||
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)
|
||
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}"
|
||
|
||
|
||
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="")
|
||
|
||
|
||
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:
|
||
"""回调按钮携带的歌曲数据：自包含（歌名/歌手/封面 % 转义，避免撞上
|
||
分隔符），bot 收到回调直接还原 Song，无需记忆菜单快照。"""
|
||
return (
|
||
f"音乐|{action}|{song.source}|{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 段，缺省按无封面处理。"""
|
||
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=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)),
|
||
]}
|
||
]
|
||
}
|
||
}
|
||
|
||
|
||
def _menu_keyboard(
|
||
songs: list[Song], keyword: str, page: int, page_size: int
|
||
) -> dict:
|
||
"""选歌键盘：编号回调按钮每行 5 个（点一下即弹卡片），末行 [上一页][下一页]。
|
||
首页无上一页；是否给下一页以"本页满员"判断（搜索结果不足一页即到底）。"""
|
||
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)
|
||
nav = []
|
||
if page > 1:
|
||
nav.append(_button("prev", "上一页", f"音乐|page|{_q(keyword)}|{page - 1}"))
|
||
if len(songs) >= page_size:
|
||
nav.append(_button("next", "下一页", f"音乐|page|{_q(keyword)}|{page + 1}"))
|
||
if nav:
|
||
rows.append(nav)
|
||
return {"content": {"rows": [{"buttons": r} for r in rows]}}
|
||
|
||
|
||
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,
|
||
) -> None:
|
||
"""搜索并展示选歌菜单：Markdown 歌单（编号 + 歌名/歌手/来源）+ 编号回调按钮
|
||
（每行 5 个）+ 上下页。菜单只负责选歌——不放播放/下载，点编号后由 send_card
|
||
弹卡片。Markdown 失败回退纯文本（按钮保留）。"""
|
||
try:
|
||
songs = await search_list(http, config, keyword, page)
|
||
except MusicServerError as e:
|
||
logger.warning(f"搜索失败: {e}")
|
||
await _reply(
|
||
api, group_openid, "音乐服务暂时不可用，稍后再试试", msg_id, event_id
|
||
)
|
||
return
|
||
if not songs:
|
||
await _reply(api, group_openid, f"没找到「{keyword}」", msg_id, event_id)
|
||
return
|
||
lines = [f"## 点歌：{keyword}"]
|
||
for i, song in enumerate(songs, 1):
|
||
title = f"{song.name} - {song.singer}".strip(" -")
|
||
lines.append(f"{i}. {title}")
|
||
keyboard = _menu_keyboard(songs, keyword, page, config.menu_page_size)
|
||
try:
|
||
resp = 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:
|
||
resp = await api.post_group_message(
|
||
group_openid,
|
||
content="\n".join(lines[1:]) + "\n（点下方编号选歌）",
|
||
keyboard=keyboard,
|
||
msg_id=msg_id,
|
||
event_id=event_id,
|
||
)
|
||
except _API_FAILS as e:
|
||
logger.warning(f"文本菜单发送失败: {e!r}")
|
||
return
|
||
await _swap_message(api, config, group_openid, resp)
|
||
logger.info(f"[{group_openid}] 菜单已发：{keyword} 第{page}页 {len(songs)}首")
|
||
|
||
|
||
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 指定
|
||
lines.append(f"")
|
||
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))
|
||
keyboard = _card_keyboard(song)
|
||
try:
|
||
resp = 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:
|
||
resp = await api.post_group_message(
|
||
group_openid,
|
||
content="\n".join(lines[1:]) + "\n（点下方按钮播放/下载）",
|
||
keyboard=keyboard,
|
||
msg_id=msg_id,
|
||
event_id=event_id,
|
||
)
|
||
except _API_FAILS as e:
|
||
logger.warning(f"文本卡片发送失败: {e!r}")
|
||
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 用于候选源搜索。
|
||
|
||
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
|
||
# 下载在 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:
|
||
text = await _lyric(http, config, song)
|
||
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)
|
||
|
||
|
||
def _help_keyboard() -> dict:
|
||
"""帮助键盘：[🎤 点歌][📝 歌词] 指令按钮，点一下把斜杠指令填进输入框。"""
|
||
return {
|
||
"content": {
|
||
"rows": [
|
||
{"buttons": [
|
||
_cmd_button("help-cmd-song", "🎤 点歌", "/点歌 "),
|
||
_cmd_button("help-cmd-lyric", "📝 歌词", "/歌词 "),
|
||
]}
|
||
]
|
||
}
|
||
}
|
||
|
||
|
||
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)
|
||
|
||
|
||
async def _search_source(
|
||
http: ClientSession, config: BotConfig, keyword: str, source: str
|
||
) -> Song | None:
|
||
"""在指定源内搜索（不触发聚合兜底）。"""
|
||
data = await _get(http, config, "/search", name=keyword, source=source, limit="5")
|
||
for item in data.get("list") or []:
|
||
if song := _song_of(item):
|
||
return song
|
||
return None
|
||
|
||
|
||
# 直链回退源的优先顺序：tx 等源直链常受 CDN 风控抓不动，kw 实测可抓。
|
||
# 实际候选取 MusicServer /sources 报告的、具备 search+url 能力的平台：
|
||
# 先按此优先序排，其余平台追加在后，服务器增删源无需改代码。
|
||
# tx 的 url 能力位只代表「服务器能给直链」，不代表直链抓得动——它是
|
||
# 风控惯犯（回退机制的本因），绝不作为候选源，否则回退末尾几乎必然
|
||
# 白探测一次
|
||
_SOURCE_PREFERENCE = ("kw", "wy")
|
||
_NO_FALLBACK_SOURCES = frozenset({"tx"})
|
||
|
||
# /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)
|
||
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 all((p.get("capabilities") or {}).get(c) for c in ("search", "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 翻页 / pick 选歌弹卡片 / play 发语音 / dl 发文件；
|
||
返回是否为本 bot 的按钮（非本 bot 的按钮不处理也不回应）。
|
||
|
||
interaction_id 仅用于回应互动（PUT /interactions）；消息被动回复的
|
||
event_id 须用事件最外层 id（"INTERACTION_CREATE:<uuid>"），用 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)
|
||
if not ((action == "page" and len(args) == 2) or is_song):
|
||
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 == "pick":
|
||
await send_card(
|
||
api, http, config, group_openid, _song_of_data(args),
|
||
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, "音乐服务暂时不可用，稍后再试试", 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, "音乐服务暂时不可用，稍后再试试", 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 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:
|
||
"""处理一条群消息（@机器人 与全量模式共用）：跳过机器人自己的消息和
|
||
重复推送，再交给 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
|
||
return await handle_command(
|
||
api, http, config, msg.group_openid, msg.id, msg.content
|
||
)
|
||
|
||
|
||
def register(ee: EventEmitter, config: BotConfig, http: ClientSession) -> None:
|
||
"""把点歌命令与按钮回调 handler 挂到 emitter（群场景）；http 为 bot
|
||
自建连接池（不受 SDK 的 TIMEOUT 约束），经闭包供 handler 使用——不能
|
||
用形参注入，否则拿到的是 SDK 那个 5s 超时的连接池。"""
|
||
logger.info(
|
||
f"点歌已就绪：源={config.music_source or '聚合'} 音质={config.music_quality} "
|
||
f"每页={config.menu_page_size} 下载={config.send_mode} "
|
||
f"自动撤回={'开' if config.withdraw else '关'}"
|
||
)
|
||
|
||
# @机器人 的消息走前一个事件；开通「接收所有消息」后群内全部消息走
|
||
# 全量事件，两个事件字段一致，共用同一个 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）都幂等同步一次，
|
||
# 重连即自愈；RESUME 续传不重发 READY，不会重复刷
|
||
if config.command_panel:
|
||
|
||
@ee.on("READY")
|
||
async def on_ready(_: Ready, api: BotApi) -> None:
|
||
await sync_command_panel(api)
|
||
|
||
# 平台发现也跟着连接走：每次 READY 拉一次 /sources（TTL 缓存内直读），
|
||
# 日志报告服务器实际支持的源并校验配置源；重连即重新感知增删
|
||
@ee.on("READY")
|
||
async def on_ready_platforms(_: Ready, api: BotApi) -> None:
|
||
table = await platforms(http, config)
|
||
if not table:
|
||
return
|
||
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} 不在上述平台里，点歌/歌词会报错"
|
||
)
|