655 lines
24 KiB
Python
655 lines
24 KiB
Python
"""
|
|
BetterGI 弹幕联动主程序
|
|
========================
|
|
监听 B 站直播间弹幕 -> 关键词匹配 -> 调用 BetterGI.exe --startGroups 执行配置组
|
|
|
|
依赖: pip install websockets brotli
|
|
运行: python danmu_bettergi.py
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
import os
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
import zlib
|
|
from pathlib import Path
|
|
|
|
APP_DIR = Path(__file__).resolve().parent
|
|
if str(APP_DIR) not in sys.path:
|
|
sys.path.insert(0, str(APP_DIR))
|
|
|
|
from core.runtime_paths import APP_ROOT as PROJECT_ROOT, CONFIG_DIR, LOG_DIR, ensure_runtime_dirs, project_path
|
|
|
|
try:
|
|
import websockets
|
|
import brotli
|
|
except ImportError:
|
|
print("缺少依赖，请先运行: pip install websockets brotli")
|
|
sys.exit(1)
|
|
|
|
# ============== B站直播弹幕协议常量 ==============
|
|
HEADER_LEN = 16
|
|
OP_HEARTBEAT = 2 # 心跳请求
|
|
OP_HEARTBEAT_REPLY = 3 # 心跳响应(人气值)
|
|
OP_MESSAGE = 5 # 业务消息
|
|
OP_AUTH = 7 # 认证请求
|
|
OP_AUTH_REPLY = 8 # 认证响应
|
|
PROTO_JSON = 0 # 明文JSON
|
|
PROTO_ZLIB = 2 # zlib压缩
|
|
PROTO_BROTLI = 3 # brotli压缩
|
|
|
|
|
|
# ============== 弹幕服务器发现 ==============
|
|
def get_danmu_server(room_id: int) -> dict:
|
|
"""通过B站API获取弹幕服务器地址和token。优先使用host_server_list中的新服务器。"""
|
|
url = f"https://api.live.bilibili.com/room/v1/Danmu/getConf?room_id={room_id}"
|
|
req = urllib.request.Request(url)
|
|
req.add_header("User-Agent", "Mozilla/5.0")
|
|
resp = urllib.request.urlopen(req, timeout=10)
|
|
data = json.loads(resp.read())["data"]
|
|
token = data["token"]
|
|
host_list = data.get("host_server_list", [])
|
|
if host_list:
|
|
entry = host_list[0]
|
|
host = entry["host"]
|
|
port = entry.get("wss_port", 443)
|
|
else:
|
|
host = data["host"]
|
|
port = data.get("wss_port", 443)
|
|
return {
|
|
"ws_url": f"wss://{host}:{port}/sub",
|
|
"token": token,
|
|
"host": host,
|
|
"port": port,
|
|
}
|
|
|
|
|
|
def get_buvid3(sessdata: str = "") -> str:
|
|
"""通过B站finger/spi接口获取buvid3 (2024+协议认证必需)。"""
|
|
url = "https://api.bilibili.com/x/frontend/finger/spi"
|
|
req = urllib.request.Request(url)
|
|
req.add_header("User-Agent", "Mozilla/5.0")
|
|
if sessdata:
|
|
req.add_header("Cookie", f"SESSDATA={sessdata}")
|
|
try:
|
|
resp = urllib.request.urlopen(req, timeout=10)
|
|
data = json.loads(resp.read())
|
|
if data.get("code") == 0:
|
|
return data["data"]["b_3"]
|
|
except Exception:
|
|
pass
|
|
return ""
|
|
|
|
|
|
# ============== 配置管理 ==============
|
|
class Config:
|
|
def __init__(self, path: str):
|
|
self.path = Path(path)
|
|
self.data = {}
|
|
self.reload()
|
|
|
|
def reload(self):
|
|
with open(self.path, "r", encoding="utf-8") as f:
|
|
self.data = json.load(f)
|
|
|
|
@property
|
|
def room_id(self) -> int:
|
|
return int(self.data["bilibili"]["room_id"])
|
|
|
|
@property
|
|
def sessdata(self) -> str:
|
|
return self.data["bilibili"].get("sessdata", "")
|
|
|
|
@property
|
|
def bettergi_exe(self) -> str:
|
|
return self.data["bettergi"]["exe_path"]
|
|
|
|
@property
|
|
def bettergi_work_dir(self) -> str:
|
|
wd = self.data["bettergi"].get("work_dir", "")
|
|
return wd if wd else str(Path(self.bettergi_exe).parent)
|
|
|
|
@property
|
|
def default_cooldown(self) -> int:
|
|
return int(self.data["global"].get("default_cooldown", 30))
|
|
|
|
@property
|
|
def admin_uids(self) -> set:
|
|
return set(int(x) for x in self.data["global"].get("admin_uids", []))
|
|
|
|
@property
|
|
def rules(self) -> list:
|
|
return self.data.get("rules", [])
|
|
|
|
@property
|
|
def restart_mode(self) -> str:
|
|
"""重启模式: gentle(温和,跳过) / aggressive(激进,先杀再启)"""
|
|
return self.data["global"].get("restart_mode", "gentle")
|
|
|
|
|
|
# ============== 冷却管理 ==============
|
|
class CooldownManager:
|
|
"""按规则记录最后触发时间，防止同一指令被弹幕刷屏重复触发。"""
|
|
|
|
def __init__(self):
|
|
self._last_fire: dict = {} # keyword -> timestamp
|
|
|
|
def can_fire(self, keyword: str, cooldown: int) -> bool:
|
|
now = time.time()
|
|
last = self._last_fire.get(keyword, 0)
|
|
return (now - last) >= cooldown
|
|
|
|
def mark_fired(self, keyword: str):
|
|
self._last_fire[keyword] = time.time()
|
|
|
|
|
|
# ============== BetterGI 调用 ==============
|
|
class BetterGIRunner:
|
|
"""封装 BetterGI.exe --startGroups 调用。"""
|
|
|
|
def __init__(self, exe_path: str, work_dir: str, logger: logging.Logger,
|
|
restart_mode: str = "gentle"):
|
|
self.exe_path = exe_path
|
|
self.work_dir = work_dir
|
|
self.logger = logger
|
|
self.restart_mode = restart_mode # gentle / aggressive
|
|
self._busy = False # 标记是否正在执行任务
|
|
self._lock = asyncio.Lock()
|
|
# 缓存 cancelTaskHotkey (从 BetterGI Config.json 读)
|
|
self._cancel_hotkey = None
|
|
self._load_cancel_hotkey()
|
|
|
|
def _load_cancel_hotkey(self):
|
|
"""从 BetterGI 的 Config.json 读取取消任务热键。"""
|
|
try:
|
|
import os
|
|
cfg_path = os.path.join(self.work_dir, "User", "Config.json")
|
|
if os.path.exists(cfg_path):
|
|
with open(cfg_path, "r", encoding="utf-8") as f:
|
|
bgi_cfg = json.load(f)
|
|
hk = bgi_cfg.get("hotKeyConfig", {}).get("cancelTaskHotkey", "")
|
|
if hk:
|
|
self._cancel_hotkey = hk
|
|
self.logger.info(f"已读取 BetterGI 取消任务热键: {hk}")
|
|
except Exception as e:
|
|
self.logger.debug(f"读取 cancelTaskHotkey 失败(不影响使用): {e}")
|
|
|
|
async def _stop_bgi_gracefully(self):
|
|
"""优雅停止 BetterGI: 先按取消热键,再 taskkill 兜底。"""
|
|
# ① 模拟按取消热键 (让 BGI 内部收尾,保存进度)
|
|
if self._cancel_hotkey:
|
|
self.logger.info(f"[激进] 按取消热键: {self._cancel_hotkey}")
|
|
try:
|
|
await self._send_key(self._cancel_hotkey)
|
|
except Exception as e:
|
|
self.logger.warning(f"[激进] 按热键失败: {e}")
|
|
# 给 BetterGI 5 秒收尾时间
|
|
await asyncio.sleep(5)
|
|
|
|
# ② taskkill 强杀兜底
|
|
self.logger.info("[激进] taskkill /F /IM BetterGI.exe")
|
|
try:
|
|
proc = await asyncio.create_subprocess_exec(
|
|
"taskkill", "/F", "/IM", "BetterGI.exe",
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
# CREATE_NO_WINDOW = 0x08000000,避免弹黑窗
|
|
creationflags=0x08000000,
|
|
)
|
|
await proc.communicate()
|
|
except Exception as e:
|
|
self.logger.debug(f"[激进] taskkill 执行: {e}")
|
|
# 再等 1 秒让进程完全退出
|
|
await asyncio.sleep(1)
|
|
|
|
async def _send_key(self, key: str):
|
|
"""模拟按键。优先用 pywin32,其次 keyboard 库,都没有则跳过。"""
|
|
# 简单映射: BetterGI 的热键名通常是 "F9" "Ctrl+P" 这种
|
|
# 优先尝试 pywin32 (最稳定,不弹窗)
|
|
try:
|
|
import win32api
|
|
import win32con
|
|
# 简单支持单键 (F1-F24, 字母, 数字)
|
|
vk_map = {
|
|
"F1": win32con.VK_F1, "F2": win32con.VK_F2, "F3": win32con.VK_F3,
|
|
"F4": win32con.VK_F4, "F5": win32con.VK_F5, "F6": win32con.VK_F6,
|
|
"F7": win32con.VK_F7, "F8": win32con.VK_F8, "F9": win32con.VK_F9,
|
|
"F10": win32con.VK_F10, "F11": win32con.VK_F11, "F12": win32con.VK_F12,
|
|
"ESC": win32con.VK_ESCAPE, "ESCAPE": win32con.VK_ESCAPE,
|
|
}
|
|
# 处理 Ctrl+X Shift+X 这种组合键
|
|
parts = key.replace("+", " ").split()
|
|
main_key = parts[-1].upper()
|
|
ctrl = "CTRL" in [p.upper() for p in parts[:-1]]
|
|
shift = "SHIFT" in [p.upper() for p in parts[:-1]]
|
|
alt = "ALT" in [p.upper() for p in parts[:-1]]
|
|
|
|
vk = vk_map.get(main_key)
|
|
if vk is None and len(main_key) == 1:
|
|
vk = ord(main_key.upper()) # 字母键
|
|
|
|
if vk is None:
|
|
self.logger.warning(f"[激进] 不支持的热键: {key}, 跳过热键直接 taskkill")
|
|
return
|
|
|
|
if ctrl:
|
|
win32api.keybd_event(win32con.VK_CONTROL, 0, 0, 0)
|
|
if shift:
|
|
win32api.keybd_event(win32con.VK_SHIFT, 0, 0, 0)
|
|
if alt:
|
|
win32api.keybd_event(win32con.VK_MENU, 0, 0, 0)
|
|
|
|
win32api.keybd_event(vk, 0, 0, 0) # 按下
|
|
win32api.keybd_event(vk, 0, win32con.KEYEVENTF_KEYUP, 0) # 抬起
|
|
|
|
if ctrl:
|
|
win32api.keybd_event(win32con.VK_CONTROL, 0, win32con.KEYEVENTF_KEYUP, 0)
|
|
if shift:
|
|
win32api.keybd_event(win32con.VK_SHIFT, 0, win32con.KEYEVENTF_KEYUP, 0)
|
|
if alt:
|
|
win32api.keybd_event(win32con.VK_MENU, 0, win32con.KEYEVENTF_KEYUP, 0)
|
|
|
|
self.logger.info(f"[激进] 已模拟按键: {key}")
|
|
except ImportError:
|
|
self.logger.warning(
|
|
"[激进] 未安装 pywin32, 无法模拟热键。"
|
|
"可运行: pip install pywin32 (可选,不装则直接 taskkill)"
|
|
)
|
|
|
|
async def run_groups(self, groups: list) -> bool:
|
|
"""异步启动配置组。"""
|
|
async with self._lock:
|
|
if self._busy:
|
|
if self.restart_mode == "aggressive":
|
|
self.logger.warning(
|
|
f"[激进模式] BetterGI 正在执行任务,先杀再启: {groups}"
|
|
)
|
|
# 释放锁,执行停止 (停止可能耗时,不持有锁)
|
|
self._busy = False
|
|
else:
|
|
self.logger.warning(
|
|
f"[温和模式] BetterGI 正在执行任务，跳过本次触发: {groups}"
|
|
)
|
|
return False
|
|
else:
|
|
self._busy = True
|
|
|
|
# 激进模式: 先停止旧任务
|
|
if self.restart_mode == "aggressive":
|
|
await self._stop_bgi_gracefully()
|
|
async with self._lock:
|
|
self._busy = True
|
|
|
|
try:
|
|
cmd = [self.exe_path, "--startGroups"] + groups
|
|
self.logger.info(f"调用 BetterGI: {' '.join(cmd)}")
|
|
proc = await asyncio.create_subprocess_exec(
|
|
*cmd,
|
|
cwd=self.work_dir,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.PIPE,
|
|
)
|
|
stdout, stderr = await proc.communicate()
|
|
if proc.returncode in (0, 553):
|
|
self.logger.info(f"BetterGI 启动成功: {groups} (rc={proc.returncode})")
|
|
return True
|
|
else:
|
|
self.logger.error(
|
|
f"BetterGI 启动失败 rc={proc.returncode}: "
|
|
f"{stderr.decode('gbk', errors='replace')}"
|
|
)
|
|
return False
|
|
except Exception as e:
|
|
self.logger.exception(f"调用 BetterGI 异常: {e}")
|
|
return False
|
|
finally:
|
|
self._busy = False
|
|
|
|
|
|
# ============== 弹幕匹配引擎 ==============
|
|
class DanmuMatcher:
|
|
"""把弹幕文本匹配到对应规则。"""
|
|
|
|
def __init__(self, config: Config, cooldown: CooldownManager,
|
|
runner: BetterGIRunner, logger: logging.Logger):
|
|
self.config = config
|
|
self.cooldown = cooldown
|
|
self.runner = runner
|
|
self.logger = logger
|
|
|
|
async def handle(self, text: str, uid: int, uname: str):
|
|
text = (text or "").strip()
|
|
if not text:
|
|
return
|
|
admins = self.config.admin_uids
|
|
is_admin = uid in admins
|
|
|
|
for rule in self.config.rules:
|
|
if not self._match(text, rule):
|
|
continue
|
|
|
|
keyword = rule["keyword"]
|
|
groups = rule["groups"]
|
|
cooldown_sec = rule.get("cooldown", self.config.default_cooldown)
|
|
admin_only = rule.get("admin_only", False)
|
|
reply = rule.get("reply", "")
|
|
|
|
# 权限检查
|
|
if admin_only and not is_admin:
|
|
self.logger.info(
|
|
f"[权限不足] {uname}({uid}) 弹幕'{text}' 命中'{keyword}' 但需管理员"
|
|
)
|
|
return # 注意: 这里return了,继续检查下一条规则(如果有)
|
|
|
|
# 冷却检查
|
|
if not self.cooldown.can_fire(keyword, cooldown_sec):
|
|
remain = int(cooldown_sec - (time.time() - self.cooldown._last_fire.get(keyword, 0)))
|
|
self.logger.info(
|
|
f"[冷却中] '{keyword}' 还需 {remain}s, 来自 {uname}({uid})"
|
|
)
|
|
return # 冷却不触发,但日志已打印
|
|
|
|
# 触发
|
|
self.cooldown.mark_fired(keyword)
|
|
self.logger.info(
|
|
f"[触发] {uname}({uid}) 弹幕'{text}' -> 配置组 {groups}"
|
|
)
|
|
if reply:
|
|
self.logger.info(f"[回复提示] {reply}")
|
|
await self.runner.run_groups(groups)
|
|
return # 一条弹幕只触发第一个命中的规则
|
|
|
|
@staticmethod
|
|
def _match(text: str, rule: dict) -> bool:
|
|
keyword = rule["keyword"]
|
|
mtype = rule.get("match_type", "contains")
|
|
if mtype == "exact":
|
|
return text == keyword
|
|
elif mtype == "contains":
|
|
return keyword in text
|
|
elif mtype == "startswith":
|
|
return text.startswith(keyword)
|
|
elif mtype == "regex":
|
|
import re
|
|
return re.search(keyword, text) is not None
|
|
return False
|
|
|
|
|
|
# ============== B站弹幕协议 ==============
|
|
def make_packet(op: int, body: bytes = b"") -> bytes:
|
|
"""组装协议包。header 16字节 + body。"""
|
|
if isinstance(body, str):
|
|
body = body.encode("utf-8")
|
|
total = HEADER_LEN + len(body)
|
|
header = struct.pack(">IHHII", total, HEADER_LEN, 1, op, 1)
|
|
return header + body
|
|
|
|
|
|
def parse_packets(data: bytes):
|
|
"""一个 WebSocket 帧可能含多个协议包，循环切分。"""
|
|
offset = 0
|
|
packets = []
|
|
while offset < len(data):
|
|
if offset + HEADER_LEN > len(data):
|
|
break
|
|
total, header_len, proto_ver, op, seq = struct.unpack(
|
|
">IHHII", data[offset:offset + HEADER_LEN]
|
|
)
|
|
body = data[offset + header_len:offset + total]
|
|
packets.append((proto_ver, op, body))
|
|
offset += total
|
|
return packets
|
|
|
|
|
|
def decode_body(proto_ver: int, body: bytes) -> bytes:
|
|
"""按协议版本解压 body。"""
|
|
if proto_ver == PROTO_JSON:
|
|
return body
|
|
if proto_ver == PROTO_ZLIB:
|
|
return zlib.decompress(body)
|
|
if proto_ver == PROTO_BROTLI:
|
|
return brotli.decompress(body)
|
|
return body
|
|
|
|
|
|
def _clean_uid(value) -> int:
|
|
"""只接受纯十进制正整数 UID，匿名或含星号的值返回 0。"""
|
|
if isinstance(value, bool):
|
|
return 0
|
|
if isinstance(value, int):
|
|
return value if value > 0 else 0
|
|
if isinstance(value, str):
|
|
value = value.strip()
|
|
return int(value) if value.isdecimal() and int(value) > 0 else 0
|
|
return 0
|
|
|
|
|
|
def extract_danmu(body: bytes):
|
|
"""从消息体中提取弹幕(DANMU_MSG)。返回 [(text, uid, uname), ...]。"""
|
|
results = []
|
|
try:
|
|
decoded = decode_body(0, body) # body 已是解压后的，proto_ver 此处忽略
|
|
except Exception:
|
|
decoded = body
|
|
try:
|
|
msg = json.loads(decoded.decode("utf-8", errors="replace"))
|
|
except Exception:
|
|
return results
|
|
cmd = msg.get("cmd", "")
|
|
if cmd.startswith("DANMU_MSG"):
|
|
info = msg.get("info", [])
|
|
if not isinstance(info, list) or len(info) <= 2:
|
|
return results
|
|
text = str(info[1] if len(info) > 1 else "")
|
|
member = info[2]
|
|
if isinstance(member, (list, tuple)):
|
|
uid = _clean_uid(member[0] if len(member) > 0 else 0)
|
|
uname = str(member[1] if len(member) > 1 and member[1] is not None else "")
|
|
elif isinstance(member, dict):
|
|
uid = _clean_uid(member.get("uid_str") or member.get("uid") or member.get("mid"))
|
|
uname = str(member.get("uname") or member.get("name") or "")
|
|
else:
|
|
uid, uname = 0, ""
|
|
results.append((text, uid, uname))
|
|
return results
|
|
|
|
|
|
# ============== 弹幕客户端 ==============
|
|
class BliveClient:
|
|
"""B站直播弹幕 WebSocket 客户端,带自动重连。"""
|
|
|
|
def __init__(self, room_id: int, sessdata: str,
|
|
matcher: DanmuMatcher, logger: logging.Logger):
|
|
self.room_id = room_id
|
|
self.sessdata = sessdata
|
|
self.matcher = matcher
|
|
self.logger = logger
|
|
self._stop = False
|
|
self._buvid3 = ""
|
|
|
|
async def run(self):
|
|
"""主循环:断线自动重连,间隔递增。"""
|
|
# 启动时获取buvid3 (2024+协议认证必需)
|
|
self._buvid3 = get_buvid3(self.sessdata)
|
|
if self._buvid3:
|
|
self.logger.info(f"获取buvid3成功: {self._buvid3[:20]}...")
|
|
else:
|
|
self.logger.warning("获取buvid3失败,弹幕可能收不到! 将尝试匿名连接")
|
|
retry = 0
|
|
while not self._stop:
|
|
try:
|
|
# 每次连接前重新获取服务器地址(避免IP变化)
|
|
server_info = get_danmu_server(self.room_id)
|
|
ws_url = server_info["ws_url"]
|
|
token = server_info["token"]
|
|
self.logger.info(f"弹幕服务器: {server_info['host']}:{server_info['port']}")
|
|
await self._connect_once(ws_url, token)
|
|
retry = 0 # 连上后重置
|
|
except asyncio.CancelledError:
|
|
break
|
|
except Exception as e:
|
|
self.logger.warning(f"连接断开: {e}")
|
|
if self._stop:
|
|
break
|
|
retry += 1
|
|
wait = min(2 ** retry, 60) # 指数退避,最多60秒
|
|
self.logger.info(f"{wait}秒后重连(第{retry}次)...")
|
|
await asyncio.sleep(wait)
|
|
|
|
def stop(self):
|
|
self._stop = True
|
|
|
|
async def _connect_once(self, ws_url: str, token: str):
|
|
self.logger.info(f"连接直播间 room_id={self.room_id} ...")
|
|
# 禁用 websockets 自带 ping/pong (B站有自己的心跳协议)
|
|
async with websockets.connect(
|
|
ws_url,
|
|
max_size=None,
|
|
ping_interval=None,
|
|
ping_timeout=None,
|
|
close_timeout=5,
|
|
) as ws:
|
|
# 1. 发送认证包(使用API返回的token + buvid3)
|
|
auth = {
|
|
"uid": 0,
|
|
"roomid": self.room_id,
|
|
"protover": PROTO_BROTLI,
|
|
"platform": "web",
|
|
"type": 2,
|
|
"key": token or (self.sessdata or ""),
|
|
}
|
|
if self._buvid3:
|
|
auth["buvid"] = self._buvid3
|
|
await ws.send(make_packet(OP_AUTH, json.dumps(auth)))
|
|
self.logger.info("已发送认证包,等待响应...")
|
|
# 立即发送一个心跳,激活弹幕推送
|
|
await ws.send(make_packet(OP_HEARTBEAT))
|
|
|
|
# 2. 心跳任务
|
|
async def heartbeat():
|
|
while True:
|
|
await asyncio.sleep(15)
|
|
try:
|
|
await ws.send(make_packet(OP_HEARTBEAT))
|
|
except Exception:
|
|
break
|
|
|
|
hb_task = asyncio.create_task(heartbeat())
|
|
|
|
# 3. 接收循环
|
|
msg_counter = 0
|
|
async for raw in ws:
|
|
if isinstance(raw, str):
|
|
continue
|
|
for proto_ver, op, body in parse_packets(raw):
|
|
if op == OP_AUTH_REPLY:
|
|
# 检查认证是否真的成功
|
|
try:
|
|
auth_resp = json.loads(body.decode("utf-8", errors="replace"))
|
|
code = auth_resp.get("code", -1)
|
|
if code == 0:
|
|
self.logger.info(f"认证成功,开始监听弹幕 (响应: {auth_resp})")
|
|
else:
|
|
self.logger.error(f"认证失败! 响应: {auth_resp}")
|
|
except Exception:
|
|
self.logger.info(f"认证响应(原始): {body}")
|
|
elif op == OP_HEARTBEAT_REPLY:
|
|
# body 是 4 字节人气值(大端序int32)
|
|
if len(body) >= 4:
|
|
popularity = struct.unpack(">I", body[:4])[0]
|
|
self.logger.debug(f"人气值: {popularity}")
|
|
elif op == OP_MESSAGE:
|
|
# body 可能被压缩,按 proto_ver 解压后再切包
|
|
try:
|
|
decoded = decode_body(proto_ver, body)
|
|
except Exception as e:
|
|
self.logger.error(f"解压失败 proto={proto_ver} len={len(body)}: {e}")
|
|
continue
|
|
# 解压后可能内含多个子包
|
|
for sub_proto, sub_op, sub_body in parse_packets(decoded):
|
|
if sub_op == OP_MESSAGE:
|
|
msg_counter += 1
|
|
# 调试:打印每条消息的cmd类型
|
|
try:
|
|
msg_json = json.loads(sub_body.decode("utf-8", errors="replace"))
|
|
cmd = msg_json.get("cmd", "?")
|
|
self.logger.debug(f"#{msg_counter} cmd={cmd}")
|
|
except Exception:
|
|
self.logger.debug(f"#{msg_counter} 解析JSON失败")
|
|
continue
|
|
for text, uid, uname in extract_danmu(sub_body):
|
|
self.logger.info(f"[弹幕] {uname}({uid}): {text}")
|
|
await self.matcher.handle(text, uid, uname)
|
|
|
|
hb_task.cancel()
|
|
|
|
|
|
# ============== 日志 ==============
|
|
def setup_logger(config: Config) -> logging.Logger:
|
|
logger = logging.getLogger("danmu_bettergi")
|
|
logger.setLevel(getattr(logging, config.data["global"].get("log_level", "INFO")))
|
|
fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s",
|
|
datefmt="%H:%M:%S")
|
|
sh = logging.StreamHandler(sys.stdout)
|
|
sh.setFormatter(fmt)
|
|
logger.addHandler(sh)
|
|
log_file = config.data["global"].get("log_file")
|
|
if log_file:
|
|
LOG_DIR.mkdir(parents=True, exist_ok=True)
|
|
fh = logging.FileHandler(project_path(log_file, base=LOG_DIR), encoding="utf-8")
|
|
fh.setFormatter(fmt)
|
|
logger.addHandler(fh)
|
|
return logger
|
|
|
|
|
|
# ============== 入口 ==============
|
|
async def main():
|
|
ensure_runtime_dirs()
|
|
config_path = CONFIG_DIR / "config.json"
|
|
if not config_path.exists():
|
|
legacy_config_path = PROJECT_ROOT / "config.json"
|
|
if legacy_config_path.exists():
|
|
config_path = legacy_config_path
|
|
if not config_path.exists():
|
|
print("找不到 config.json,请先配置!")
|
|
sys.exit(1)
|
|
|
|
config = Config(str(config_path))
|
|
logger = setup_logger(config)
|
|
|
|
logger.info("=" * 50)
|
|
logger.info("BetterGI 弹幕联动启动")
|
|
logger.info(f"直播间: {config.room_id}")
|
|
logger.info(f"BetterGI: {config.bettergi_exe}")
|
|
logger.info(f"规则数: {len(config.rules)}")
|
|
logger.info(f"管理员UID: {config.admin_uids or '无'}")
|
|
logger.info(f"默认冷却: {config.default_cooldown}s")
|
|
logger.info(f"重启模式: {config.restart_mode}")
|
|
logger.info("=" * 50)
|
|
|
|
cooldown = CooldownManager()
|
|
runner = BetterGIRunner(
|
|
config.bettergi_exe, config.bettergi_work_dir, logger,
|
|
restart_mode=config.restart_mode,
|
|
)
|
|
matcher = DanmuMatcher(config, cooldown, runner, logger)
|
|
client = BliveClient(config.room_id, config.sessdata, matcher, logger)
|
|
|
|
try:
|
|
await client.run()
|
|
except KeyboardInterrupt:
|
|
logger.info("收到退出信号,正在停止...")
|
|
client.stop()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
asyncio.run(main())
|
|
except KeyboardInterrupt:
|
|
pass
|