snowy-yang
c364506925
代码测试 / ruff + pytest (push) Canceled after 0s
- 文件代理（PUBLIC_URL）：上传与直链文本的 URL 换成 bot 公开地址，请求 到来现场重解析直链再转发字节，绕开 QQ 抓风控直链失败（40093007）与 qs 本地流外网不可达；PROXY_PORT 指定监听端口 - 群消息去重补第二把钥匙：全量模式下 @消息同时命中两事件且 msg_id 不同， 只按 msg_id 拦不住，再加「群+发送人+文本」10 秒短窗（修双发） - 「<平台>下载 <歌名>」改出该平台的选歌菜单（与 /点歌 同款交互、spage 翻页锁定平台），不再直接下第一首 - 猫耳/懒人/蜻蜓节目级条目点编号先出选集菜单（/album）：猫耳选集接口 忽略翻页参数，一次拉 50 集本地切片；懒人/蜻蜓走服务端翻页；空选集 （已下线/需登录）给明确提示 - 指令面板条目按 /sources 现算、按宽度裁剪（修 30013 超限）；装配入口 改用 Bot 门面；qqbotsdk 依赖更名 qqbotsdk-py（撤 GitHub archive 源）
89 lines
3.4 KiB
Python
89 lines
3.4 KiB
Python
"""指令面板（群里输入框敲 / 唤出的指令菜单）自动注册。
|
|
|
|
调开放平台 2026-08 新增的 /v2/panels 接口，把点歌指令同步到 QQ：
|
|
启动与每次重连（READY）各同步一次——没有就创建（target_type=all
|
|
对全部群生效），内容有变就更新，一致则跳过，幂等可重入。
|
|
|
|
条目由调用方给定（music.panel_items 按 MusicServer /sources 现算：
|
|
基础指令 + 热搜/榜单 + 各平台「<平台>下载」），服务器增删源跟着变。
|
|
接口上限（每面板 20 条；name/desc 按宽度限 14/30——汉字计 2）由条目
|
|
生成方裁剪。失败只记日志不影响运行（如功能未对机器人开放）。
|
|
"""
|
|
|
|
from typing import Any
|
|
|
|
from aiohttp import ClientError
|
|
from loguru import logger
|
|
from qqbotsdk.api import ApiError, BotApi
|
|
|
|
_API_FAILS = (ApiError, TimeoutError, ClientError)
|
|
|
|
# 面板挂在群聊场景，对所有群生效
|
|
_SCOPE = "group"
|
|
_TARGET_TYPE = "all"
|
|
# remark 兼作身份标识：重连同步时按它找到自己注册过的面板
|
|
_REMARK = "音乐bot"
|
|
|
|
|
|
def _signature(items: list[dict]) -> list[tuple]:
|
|
"""面板条目的比较指纹：只取本模块管理的字段，忽略其余与键序。"""
|
|
return [
|
|
(
|
|
i.get("name"),
|
|
i.get("type"),
|
|
i.get("desc"),
|
|
i.get("only_admin", False),
|
|
i.get("link"),
|
|
)
|
|
for i in items
|
|
]
|
|
|
|
|
|
def _unwrap(body: Any, key: str) -> Any:
|
|
"""响应可能整体包一层 {"data": ...}，也可能直接是数据本体，两者都认。"""
|
|
if not isinstance(body, dict):
|
|
return body
|
|
inner = body.get("data")
|
|
return inner if key not in body and isinstance(inner, dict) else body
|
|
|
|
|
|
def _mine(records: list[Any]) -> dict | None:
|
|
"""在面板列表里找本 bot 注册过的那张（按 remark 识别）。"""
|
|
for record in records:
|
|
if not isinstance(record, dict):
|
|
continue
|
|
if (record.get("panel") or {}).get("remark") == _REMARK:
|
|
return record
|
|
return None
|
|
|
|
|
|
async def sync_command_panel(api: BotApi, items: list[dict]) -> None:
|
|
"""把给定条目同步成 QQ 指令面板（幂等）。任何失败只记日志，绝不抛出。"""
|
|
try:
|
|
resp = await api.get("/v2/panels", params={"scope": _SCOPE, "limit": "20"})
|
|
records = _unwrap(resp, "records").get("records") or []
|
|
mine = _mine(records)
|
|
unchanged = mine and _signature(mine["panel"].get("items") or []) == _signature(
|
|
items
|
|
)
|
|
if unchanged:
|
|
logger.info(f"指令面板已是最新（{len(items)} 条指令）")
|
|
return
|
|
if mine:
|
|
await api.put(
|
|
f"/v2/panels/{mine['panel_id']}",
|
|
json={"panel": {"items": items, "remark": _REMARK}},
|
|
)
|
|
logger.info(f"指令面板已更新（{len(items)} 条指令）")
|
|
else:
|
|
body = {
|
|
"scope": _SCOPE,
|
|
"panel": {"items": items, "remark": _REMARK},
|
|
"target_type": _TARGET_TYPE,
|
|
}
|
|
await api.post("/v2/panels", json=body)
|
|
logger.info(f"指令面板已创建（{len(items)} 条指令，对全部群生效）")
|
|
except _API_FAILS as e:
|
|
# 失败不影响运行；指令可改为在开放平台后台手工配置
|
|
logger.warning(f"指令面板同步失败: {e!r}")
|