chun_qiu
4f7168a0e3
- server.py / main.py no longer reference fake_room at all (no import, no --fake-room flag, no _apply_fake_room) - Routing decision happens at connect time in DanmakuClient.start(): room_id == 1 dynamically imports fake_room.py and patches blivedm; any other room id returns immediately without touching fake_room - Switching back from room 1 to another room unpatches to real Bilibili - Production (packaged, no fake_room.py) is completely unaffected for normal room ids; room 1 without fake_room.py raises a clear error - Rebuild portable zip (83.9 MB)
223 lines
10 KiB
Python
223 lines
10 KiB
Python
"""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
|