snowy-yang
9bcdad7aa1
- 日志：控制台 + logs/musicbot.log（按天轮转、保留 7 天）， LOG_LEVEL/LOG_DIR 可配；目录不可写时退化为仅控制台； diagnose=False 避免把凭证写进日志 - 卡片用 Markdown 图片语法显示专辑封面（）， 封面取自 MusicServer 的 img 字段（tx/wy 有、kw/kg 无） - 按钮 data 加第 5 段（封面 URL），解码兼容旧 4 段，已发出的按钮不失效 - 菜单去掉每首歌后的（音源）标注；卡片保留"来源：" - 测试补：封面渲染、无封面源、旧 data 兼容、LOG_LEVEL、日志落盘
752 lines
28 KiB
Python
752 lines
28 KiB
Python
"""点歌业务：MusicServer 搜索、Markdown 选歌菜单、点选后的歌曲卡片、歌词清洗。
|
||
|
||
菜单只负责选歌：每页 N 首（MENU_PAGE_SIZE，默认 10）、编号回调按钮 + 上下页。
|
||
点编号 → 弹歌曲卡片（歌名/来源 + [▶播放][⬇下载] 回调按钮），卡片上再点才执行。
|
||
播放发语音、下载发文件，失败均回退直链文本。
|
||
|
||
自动撤回（翻页撤旧菜单、选歌后撤菜单、播放/下载后撤卡片）由 WITHDRAW
|
||
开关统一控制，默认开。"""
|
||
|
||
import re
|
||
import time
|
||
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
|
||
|
||
from .config import BotConfig
|
||
|
||
_SONG_RE = re.compile(r"点歌\s+(\S.*)")
|
||
_LYRIC_RE = re.compile(r"歌词\s+(\S.*)")
|
||
_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歌词 <歌名>：查看歌词"
|
||
|
||
_SOURCE_NAMES = {
|
||
"tx": "QQ音乐", "kw": "酷我", "kg": "酷狗", "wy": "网易云", "mg": "咪咕",
|
||
}
|
||
|
||
# 内嵌键盘上限：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 # 条；字典按插入序，超限淘汰最旧
|
||
|
||
|
||
def _prune(now: float) -> None:
|
||
"""剪掉过期记录（按插入序，头部最旧）。"""
|
||
while _seen_messages:
|
||
oldest = next(iter(_seen_messages))
|
||
if now - _seen_messages[oldest] < _DEDUP_TTL:
|
||
break
|
||
del _seen_messages[oldest]
|
||
|
||
|
||
def _mark(message_id: str) -> None:
|
||
"""记下一个 msg_id（已存在则只刷新时间），超出上限时淘汰最旧。"""
|
||
now = time.monotonic()
|
||
_prune(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(time.monotonic())
|
||
if message_id in _seen_messages:
|
||
return True
|
||
_mark(message_id)
|
||
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="")
|
||
|
||
|
||
async def _reply(
|
||
api: BotApi,
|
||
group_openid: str,
|
||
content: str,
|
||
msg_id: str | None = None,
|
||
event_id: str | None = None,
|
||
) -> None:
|
||
if event_id:
|
||
resp = await api.post_group_message(
|
||
group_openid, content=content, event_id=event_id
|
||
)
|
||
else:
|
||
resp = await api.post_group_message(
|
||
group_openid, content=content, msg_id=msg_id
|
||
)
|
||
if sent := _sent_id(resp):
|
||
_mark(sent) # 自己的消息若被回推，按重复跳过
|
||
|
||
|
||
async def _soft_reply(
|
||
api: BotApi, group_openid: str, content: str, event_id: str | None
|
||
) -> None:
|
||
"""尽力回复：失败（如互动 event_id 失效）仅记日志，让按钮回应照常进行。"""
|
||
try:
|
||
await _reply(api, group_openid, content, event_id=event_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)
|
||
|
||
|
||
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)
|
||
|
||
|
||
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,
|
||
) -> None:
|
||
"""取可抓取的播放直链后发送。
|
||
|
||
voice=True 发语音（上传 file_type=3）。voice=False 的下载看 SEND_MODE：
|
||
"file" 上传群文件（失败或 40093005 无权限则回退直链），"link" 直接发
|
||
直链、根本不尝试上传。
|
||
|
||
配置源直链抓不动时 QQ 服务器侧下载同样会失败（40093007），故先探测
|
||
可抓性、必要时换候选源；keyword 用于候选源搜索。
|
||
|
||
卡片撤回不在这里做：调用方 handle_button 会把这次点击当成一次性操作，
|
||
无论成败都撤回卡片。"""
|
||
found = await _fetchable_play(http, config, song, keyword)
|
||
if found is None:
|
||
await _reply(
|
||
api, group_openid, f"没拿到「{song.name}」的播放链接", msg_id, event_id
|
||
)
|
||
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,
|
||
media=upload,
|
||
)
|
||
if sent := _sent_id(resp):
|
||
_mark(sent)
|
||
return
|
||
except _API_FAILS as e:
|
||
no_permission = isinstance(e, ApiError) and e.code == _NO_FILE_PERMISSION
|
||
if no_permission:
|
||
logger.warning("群文件上传仅限群主/管理员，改为直链文本")
|
||
else:
|
||
logger.warning(f"以{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)
|
||
except _API_FAILS as e:
|
||
# 被动回复被拒（如互动 event_id 超时效）→ 降级为主动消息
|
||
logger.warning(f"被动回复失败，降级为主动消息: {e!r}")
|
||
try:
|
||
await _reply(api, group_openid, text)
|
||
except _API_FAILS as e:
|
||
logger.warning(f"文本链接发送失败: {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)
|
||
|
||
|
||
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 实测可抓
|
||
_DOWNLOAD_FALLBACK_SOURCES = ("kw", "wy", "mg")
|
||
|
||
|
||
async def _fetchable_play(
|
||
http: ClientSession,
|
||
config: BotConfig,
|
||
song: Song,
|
||
keyword: str,
|
||
) -> tuple[Song, str] | None:
|
||
"""逐候选源取可直接抓取的播放直链：先原曲，抓不动再按 kw/wy/mg
|
||
逐个搜索探测（懒探测：原曲可用时不产生候选源搜索）。全失败返回 None。"""
|
||
fallbacks = iter(s for s in _DOWNLOAD_FALLBACK_SOURCES if s != song.source)
|
||
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）。"""
|
||
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)
|
||
try:
|
||
if action == "page" and len(args) == 2:
|
||
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" and is_song:
|
||
await send_card(
|
||
api, http, config, group_openid, _song_of_data(args),
|
||
event_id=event_id,
|
||
)
|
||
elif action in ("play", "dl") and 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,
|
||
)
|
||
finally:
|
||
# 卡片是一次性 UI：这次点击已被消费，无论成败都撤回
|
||
# （放 finally 才能覆盖 send_song 抛错，如候选源搜索失败）
|
||
await _retire(api, config, group_openid)
|
||
else:
|
||
return False # 非本 bot 的按钮
|
||
except (MusicServerError, ValueError) as e:
|
||
logger.warning(f"按钮处理失败: {e}")
|
||
await _soft_reply(
|
||
api, group_openid, "音乐服务暂时不可用，稍后再试试", event_id
|
||
)
|
||
except _API_FAILS as e:
|
||
logger.warning(f"按钮处理失败: {e!r}")
|
||
|
||
# SDK 已容忍非 JSON Content-Type（回应接口返回 200 text/plain + {}），无需特判
|
||
try:
|
||
await api.respond_interaction(interaction_id)
|
||
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
|
||
|
||
|
||
async def handle_command(
|
||
api: BotApi,
|
||
http: ClientSession,
|
||
config: BotConfig,
|
||
group_openid: str,
|
||
msg_id: str,
|
||
content: str,
|
||
) -> bool:
|
||
"""解析一条群消息里的音乐命令并执行；返回是否命中。
|
||
|
||
选歌/播放/下载现在都由内嵌按钮（回调）承担，这里只处理文字命令。"""
|
||
if match := _SONG_RE.search(content):
|
||
keyword = match.group(1).strip()
|
||
await send_menu(api, http, config, group_openid, keyword, msg_id=msg_id)
|
||
return True
|
||
if match := _LYRIC_RE.search(content):
|
||
keyword = match.group(1).strip()
|
||
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 "音乐帮助" in content:
|
||
await _reply(api, group_openid, _HELP, msg_id=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 超时的连接池。"""
|
||
|
||
# @机器人 的消息走前一个事件；开通「接收所有消息」后群内全部消息走
|
||
# 全量事件，两个事件字段一致，共用同一个 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:
|
||
# 被动回复的 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,
|
||
)
|