"""Bilibili Live Danmaku Handler. 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 import logging import sys from typing import Optional import blivedm 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__) FAKE_ROOM_PORT = 8081 # fake_room.py default = bot web UI port (8080) + 1 def _maybe_apply_fake_room(room_id: int) -> None: """Connect-time routing: room_id == 1 uses the local fake_room.py (dev only). fake_room.py is gitignored and never packaged. For any other room id this returns immediately without ever importing fake_room, so production (which has no fake_room.py) is completely unaffected. """ fake = sys.modules.get("fake_room") if room_id == 1: if fake is None: import importlib fake = importlib.import_module("fake_room") fake.patch_blivedm(FAKE_ROOM_PORT) logger.info("[dev] room 1 -> fake room on port %d", FAKE_ROOM_PORT) elif fake is not None: # previously routed to fake room; switch back to real Bilibili fake.unpatch_blivedm() INTERACT_TYPE_NAMES = { 3: "\u5206\u4eab", 4: "\u7279\u522b\u5173\u6ce8", 5: "\u4e92\u7c89", } class DanmakuHandler(blivedm.BaseHandler): """Handles incoming danmaku and live events from a Bilibili live room.""" _CMD_CALLBACK_DICT = blivedm.BaseHandler._CMD_CALLBACK_DICT.copy() def __init__(self, queue: asyncio.Queue, rules: dict | None = None): super().__init__() self._queue = queue self._rules = rules or dict(DEFAULT_RULES) 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 text = event_text(self._rules, "danmaku", uname=message.uname, msg=message.msg, uid=message.uid) if not text: return self._count += 1 if self._count % 20 == 1: logger.info("Danmaku #%d: %s", self._count, text) else: logger.debug("Danmaku #%d: %s", self._count, text) self._put(text) # ── 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 if blind_box_gift: text = event_text(self._rules, "blind_box", viewer_name=message.uname, gift_name=message.gift_name, gift_num=message.num, gift_count=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_count=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 / 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) 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) else: logger.info("Client stopped normally.") class DanmakuClient: """Manages the BLiveClient connection for a single room.""" def __init__( self, room_id: int, queue: asyncio.Queue, cookies: Optional[dict[str, str]] = None, rules: dict | None = None, ): self._room_id = room_id self._queue = queue self._cookies = cookies self._rules = rules or dict(DEFAULT_RULES) self._client: Optional[blivedm.BLiveClient] = None self._session: Optional["aiohttp.ClientSession"] = None async def start(self) -> None: import aiohttp _maybe_apply_fake_room(self._room_id) self._session = await build_danmaku_session(self._cookies) self._client = blivedm.BLiveClient(self._room_id, session=self._session) handler = DanmakuHandler(self._queue, self._rules) self._client.set_handler(handler) self._client.start() logger.info("Connected to room %d", self._room_id) async def stop(self) -> None: if self._client is not None: logger.info("Disconnecting from room %d...", self._room_id) await self._client.stop_and_close() self._client = None if self._session is not None: await self._session.close() self._session = None @property def room_id(self) -> int: return self._room_id