125 lines
4.7 KiB
Python
125 lines
4.7 KiB
Python
"""AcFun 直播录制通知接收插件（供 acfunlive 的「通知推送」功能使用）
|
||
|
||
acfunlive 会把开播 / 下播 / 开始录制 / 合并完成等事件以 JSON POST 过来，
|
||
本插件把它们转成群消息发出去。
|
||
|
||
安装：
|
||
把本文件复制到 Momoko 的 plugins 目录（和 upload_notify.py 放一起），重启 bot。
|
||
|
||
acfunlive 侧填写（web UI 的「通知推送地址」）：
|
||
http://127.0.0.1:3000/acfunlive/notify
|
||
（端口换成你 bot.py 里 nonebot.run(port=...) 的实际端口）
|
||
|
||
可选环境变量（写在 bot 的 .env 里）：
|
||
ACFUNLIVE_NOTIFY_GROUP=777163254 # 默认通知群，多个用逗号分隔；补底用，见下
|
||
ACFUNLIVE_NOTIFY_TOKEN=某个口令 # 填了之后会校验 acfunlive 推送时带的 Bearer 口令
|
||
|
||
通知发到哪个群：优先用 acfunlive 推送里带的 group 字段（在 acfunlive 的 web UI
|
||
「通知群号」里配，改完立即生效，不用重启 bot）；没配时才回退到环境变量
|
||
ACFUNLIVE_NOTIFY_GROUP；两边都没配就只记日志不发送。
|
||
|
||
acfunlive 推送的 JSON 字段：
|
||
event liveOn / liveOff / recordStart / mergeDone / test
|
||
uid 主播 uid
|
||
name 主播名字
|
||
title 直播标题
|
||
url 直播间地址
|
||
file 相关文件路径（录制开始时的视频、合并完成时的成片）
|
||
group 通知群号（acfunlive 页面里配的）
|
||
text 已经拼好的文本，可以直接发出去
|
||
time 事件发生的时刻
|
||
"""
|
||
|
||
import os
|
||
|
||
from fastapi import FastAPI, Request
|
||
from nonebot import get_bot, get_driver
|
||
from nonebot.log import logger
|
||
|
||
driver = get_driver()
|
||
app: FastAPI = driver.server_app
|
||
|
||
# 兜底的通知群，多个用逗号分隔；acfunlive 推送里带 group 时以推送的为准
|
||
GROUPS = [g.strip() for g in os.getenv("ACFUNLIVE_NOTIFY_GROUP", "").split(",") if g.strip()]
|
||
# 推送口令，留空表示不校验
|
||
TOKEN = os.getenv("ACFUNLIVE_NOTIFY_TOKEN", "").strip()
|
||
|
||
|
||
def resolve_groups(data: dict) -> list[str]:
|
||
"""决定这条通知发到哪些群：优先用推送里下发的群号，其次用环境变量"""
|
||
raw = str(data.get("group") or "").strip()
|
||
if not raw:
|
||
return GROUPS
|
||
return [g.strip() for g in raw.split(",") if g.strip()]
|
||
|
||
# 每种事件的前缀，让群里一眼能看出是什么事
|
||
EVENT_PREFIX = {
|
||
"liveOn": "🔴 开播",
|
||
"liveOff": "⚪ 下播",
|
||
"recordStart": "⏺ 开始录制",
|
||
"mergeDone": "✅ 合并完成",
|
||
"test": "🧪 测试通知",
|
||
}
|
||
|
||
|
||
def build_message(data: dict) -> str:
|
||
"""把推送内容拼成一条群消息"""
|
||
event = str(data.get("event", ""))
|
||
prefix = EVENT_PREFIX.get(event, "📢 " + (event or "通知"))
|
||
|
||
# text 是 acfunlive 拼好的正文，优先用它
|
||
body = str(data.get("text") or "").strip()
|
||
if not body:
|
||
parts = []
|
||
if data.get("name"):
|
||
parts.append(str(data["name"]))
|
||
if data.get("title"):
|
||
parts.append(str(data["title"]))
|
||
if data.get("url"):
|
||
parts.append(str(data["url"]))
|
||
body = "\n".join(parts)
|
||
|
||
lines = [prefix]
|
||
if body:
|
||
lines.append(body)
|
||
if data.get("time"):
|
||
lines.append(str(data["time"]))
|
||
return "\n".join(lines)
|
||
|
||
|
||
@app.post("/acfunlive/notify")
|
||
async def acfunlive_notify(data: dict, request: Request):
|
||
"""接收 acfunlive 推送的直播事件"""
|
||
if TOKEN:
|
||
auth = request.headers.get("authorization", "")
|
||
got = auth[7:].strip() if auth.lower().startswith("bearer ") else auth.strip()
|
||
if got != TOKEN:
|
||
logger.warning("AcFun 通知的口令不正确，已忽略")
|
||
return {"status": "error", "message": "口令不正确"}
|
||
|
||
message = build_message(data)
|
||
groups = resolve_groups(data)
|
||
logger.info(f"收到 AcFun 通知：{data.get('event')} {data.get('name', '')}")
|
||
|
||
if not groups:
|
||
logger.warning("推送里没有 group，也没有设置 ACFUNLIVE_NOTIFY_GROUP，只记录日志不发送")
|
||
return {"status": "success", "message": "未配置群号，已忽略"}
|
||
|
||
ok, failed = 0, []
|
||
for group_id in groups:
|
||
try:
|
||
bot = get_bot()
|
||
await bot.send_group_msg(group_id=int(group_id), message=message)
|
||
ok += 1
|
||
except Exception as e:
|
||
failed.append(f"{group_id}: {e}")
|
||
logger.error(f"向群 {group_id} 发送 AcFun 通知失败：{e}")
|
||
|
||
if failed:
|
||
return {"status": "error", "message": "；".join(failed)}
|
||
return {"status": "success", "message": f"已发送到 {ok} 个群"}
|
||
|
||
|
||
logger.info("AcFun 直播通知插件加载成功")
|
||
logger.info(f"HTTP 端点：POST /acfunlive/notify，默认群（推送里带 group 时以推送为准）：{GROUPS or '未配置'}")
|