chun_qiu/bililive-touhou-tts
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Activity

feat: user-customizable event voice rules (gift/blind box/enter/follow/like/guard/super chat)

Browse Source 
- New event_rules.py: load/save event_rules.json, robust template renderer
  (missing vars -> empty, empty template -> muted), guard level mapping
- danmaku_handler.py: custom SEND_GIFT callback to extract blind_gift_name;
  implements all event callbacks feeding the TTS queue per user rules
- server.py: GET/POST /api/rules, load rules at startup, pass to client
- Web UI: 'Event Voice Rules' panel (per-event toggle + template + save)
- main.py CLI loads rules; build.py bundles event_rules.json
- skip_free_gift option to mute silver free gifts
- Rebuild portable zip (83.8 MB)
This commit is contained in:
chun_qiu
2026-08-08 16:17:28 +08:00
parent f4c99b1692
commit f3a71e908e
8 changed files with 468 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
danmaku_handler.py
+122 -14
View File
@@ -1,7 +1,8 @@
"""Bilibili Live Danmaku Handler.
Connects to a Bilibili live room and puts incoming danmaku messages into an asyncio queue.
Supports cookie-based login with browser header emulation.
Connects to a Bilibili live room, receives danmaku/gift/guard/super-chat/
interact events, and puts user-customizable speech text into an asyncio queue.
Rules come from event_rules.json (see event_rules.py).
"""
import asyncio
@@ -9,24 +10,55 @@ import logging
from typing import Optional
import blivedm
from blivedm.models.web import DanmakuMessage
from blivedm.models.web import DanmakuMessage, GiftMessage, GuardBuyMessage, \
SuperChatMessage, InteractWordV2Message, UserToastV2Message
from bili_login import build_danmaku_session
from event_rules import (
DEFAULT_RULES, GUARD_LEVEL_NAMES, event_text, is_enabled, load_rules,
)
logger = logging.getLogger(__name__)
DANMAKU_FORMAT = "{uname}\u8bf4\u3001 {msg}"
INTERACT_TYPE_NAMES = {
3: "\u5206\u4eab", 4: "\u7279\u522b\u5173\u6ce8", 5: "\u4e92\u7c89",
}
class DanmakuHandler(blivedm.BaseHandler):
"""Handles incoming danmaku from a Bilibili live room."""
"""Handles incoming danmaku and live events from a Bilibili live room."""
def __init__(self, queue: asyncio.Queue, message_format: str = DANMAKU_FORMAT):
_CMD_CALLBACK_DICT = blivedm.BaseHandler._CMD_CALLBACK_DICT.copy()
def __init__(self, queue: asyncio.Queue, rules: dict | None = None,
message_format: str = DANMAKU_FORMAT):
super().__init__()
self._queue = queue
self._rules = rules or dict(DEFAULT_RULES)
self._message_format = message_format
self._count = 0
# ── custom SEND_GIFT callback to capture blind-gift info ──────────
def __gift_callback(self, client: blivedm.BLiveClient, command: dict):
message = GiftMessage.from_command(command["data"])
blind = command.get("data", {}).get("blind_gift") or {}
blind_name = blind.get("blind_gift_name", "") if isinstance(blind, dict) else ""
self._on_gift_raw(client, message, blind_name)
_CMD_CALLBACK_DICT["SEND_GIFT"] = __gift_callback
# ── helpers ────────────────────────────────────────────────────────
def _put(self, text: str) -> None:
if not text:
return
try:
self._queue.put_nowait(text)
except asyncio.QueueFull:
logger.warning("Danmaku queue full, dropping: %s", text)
# ── danmaku ────────────────────────────────────────────────────────
def _on_danmaku(self, client: blivedm.BLiveClient, message: DanmakuMessage) -> None:
if not message.msg.strip():
return
@@ -40,17 +72,91 @@ class DanmakuHandler(blivedm.BaseHandler):
logger.info("Danmaku #%d: %s", self._count, text)
else:
logger.debug("Danmaku #%d: %s", self._count, text)
try:
self._queue.put_nowait(text)
except asyncio.QueueFull:
logger.warning("Danmaku queue full, dropping: %s", text)
self._put(text)
def _on_gift(self, client: blivedm.BLiveClient, message) -> None:
logger.debug("Gift: %s x%d from %s", message.gift_name, message.num, message.uname)
# ── gift ───────────────────────────────────────────────────────────
def _on_gift_raw(self, client: blivedm.BLiveClient, message: GiftMessage,
blind_box_gift: str) -> None:
if self._rules.get("skip_free_gift") and message.coin_type == "silver":
logger.debug("Skip free gift: %s x%d from %s",
message.gift_name, message.num, message.uname)
return
def _on_interact_word_v2(self, client: blivedm.BLiveClient, message) -> None:
logger.debug("Interact: %s (type=%d)", message.uname, message.msg_type)
if blind_box_gift:
text = event_text(self._rules, "blind_box",
viewer_name=message.uname,
gift_name=message.gift_name,
gift_num=message.num,
blind_box_gift=blind_box_gift)
else:
text = event_text(self._rules, "gift",
viewer_name=message.uname,
gift_name=message.gift_name,
gift_num=message.num,
gift_price=message.price,
gift_coin_type=message.coin_type)
logger.info("Gift: %s x%d from %s (blind=%s)",
message.gift_name, message.num, message.uname, blind_box_gift or "-")
self._put(text)
def _on_gift(self, client: blivedm.BLiveClient, message: GiftMessage) -> None:
# fallback in case _CMD_CALLBACK_DICT override is bypassed
self._on_gift_raw(client, message, "")
# ── buy guard ──────────────────────────────────────────────────────
def _on_buy_guard(self, client: blivedm.BLiveClient, message: GuardBuyMessage) -> None:
text = event_text(self._rules, "buy_guard",
ships_viewer_name=message.username,
ships_level=GUARD_LEVEL_NAMES.get(message.guard_level, str(message.guard_level)),
ships_num=message.num,
ships_price=message.price)
logger.info("Buy guard: %s level=%d", message.username, message.guard_level)
self._put(text)
# ── guard toast ────────────────────────────────────────────────────
def _on_user_toast_v2(self, client: blivedm.BLiveClient, message: UserToastV2Message) -> None:
text = event_text(self._rules, "guard_toast",
ships_viewer_name=message.username,
ships_level=GUARD_LEVEL_NAMES.get(message.guard_level, str(message.guard_level)),
guard_days=message.num,
toast_msg=message.toast_msg)
logger.info("Guard toast: %s level=%d", message.username, message.guard_level)
self._put(text)
# ── super chat ─────────────────────────────────────────────────────
def _on_super_chat(self, client: blivedm.BLiveClient, message: SuperChatMessage) -> None:
text = event_text(self._rules, "super_chat",
viewer_name=message.uname,
super_chat_content=message.message,
super_chat_price=message.price,
super_chat_duration=message.time)
logger.info("Super chat: %s (%d yuan): %s", message.uname, message.price, message.message)
self._put(text)
# ── interact (enter room / follow / like / share) ──────────────────
def _on_interact_word_v2(self, client: blivedm.BLiveClient, message: InteractWordV2Message) -> None:
msg_type = message.msg_type
if msg_type == 1:
text = event_text(self._rules, "enter_room",
viewer_name=message.username,
user_uid=message.uid)
elif msg_type == 2:
text = event_text(self._rules, "follow",
viewer_name=message.username,
user_uid=message.uid)
elif msg_type == 6:
text = event_text(self._rules, "like",
viewer_name=message.username,
user_uid=message.uid)
else:
text = event_text(self._rules, "interact_other",
viewer_name=message.username,
user_uid=message.uid,
interact_type=INTERACT_TYPE_NAMES.get(msg_type, str(msg_type)))
logger.debug("Interact: %s (type=%d)", message.username, msg_type)
self._put(text)
# ── client stopped ─────────────────────────────────────────────────
def on_client_stopped(self, client, exception: Optional[Exception]) -> None:
if exception is not None:
logger.error("Client stopped with exception: %s", exception)
@@ -66,11 +172,13 @@ class DanmakuClient:
room_id: int,
queue: asyncio.Queue,
cookies: Optional[dict[str, str]] = None,
rules: dict | None = None,
message_format: str = DANMAKU_FORMAT,
):
self._room_id = room_id
self._queue = queue
self._cookies = cookies
self._rules = rules or dict(DEFAULT_RULES)
self._message_format = message_format
self._client: Optional[blivedm.BLiveClient] = None
self._session: Optional["aiohttp.ClientSession"] = None
@@ -79,7 +187,7 @@ class DanmakuClient:
import aiohttp
self._session = await build_danmaku_session(self._cookies)
self._client = blivedm.BLiveClient(self._room_id, session=self._session)
handler = DanmakuHandler(self._queue, self._message_format)
handler = DanmakuHandler(self._queue, self._rules, self._message_format)
self._client.set_handler(handler)
self._client.start()
logger.info("Connected to room %d", self._room_id)