feat(bilibili): 新增B站渠道插件，支持直播弹幕和私信交互

该提交实现了完整的B站渠道插件，包含以下核心功能： 1. 支持B站直播弹幕监听与处理，包含弹幕、SC、礼物等多种直播间事件 2. 支持B站私信的轮询接收与发送 3. 内置WBI签名算法，适配B站API鉴权要求 4. 提供账号配对、黑白名单等弹幕私信权限控制 5. 集成速率限制与防风险机制，降低账号封禁风险 6. 完善的配置管理与状态监控能力
2026-05-21 10:40:04 +08:00 · 2026-05-21 10:40:04 +08:00 · 415d66d1c6
commit 415d66d1c6
parent b8864ac12f
20 changed files with 2155 additions and 0 deletions
--- a/backend/package/yuxi/channel/extensions/bilibili/init.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/init.py
@ -0,0 +1,291 @@
+from __future__ import annotations
+
+import logging
+import re
+
+import httpx
+
+from yuxi.channel.capabilities import ChannelCapabilities
+from yuxi.channel.extensions.base import BaseChannelPlugin
+from yuxi.channel.extensions.bilibili.anti_risk import AntiRisk
+from yuxi.channel.extensions.bilibili.config import BilibiliConfigAdapter
+from yuxi.channel.extensions.bilibili.gateway import BilibiliGateway
+from yuxi.channel.extensions.bilibili.outbound import BilibiliOutbound
+from yuxi.channel.extensions.bilibili.pairing import (
+    generate_code,
+    normalize_allow_entry,
+    verify_code,
+)
+from yuxi.channel.extensions.bilibili.probe import BilibiliProbe
+from yuxi.channel.extensions.bilibili.security import BilibiliSecurity
+from yuxi.channel.extensions.bilibili.streaming import BilibiliBlockChunker
+from yuxi.channel.extensions.bilibili.types import BilibiliConfig
+from yuxi.channel.plugins.registry import ChannelPluginRegistry
+from yuxi.channel.protocols import ClassifiedError, ErrorSeverity
+
+logger = logging.getLogger(__name__)
+
+
+class BilibiliPlugin(BaseChannelPlugin):
+    id = "bilibili"
+    name = "Bilibili"
+    order = 95
+    label = "B站 (Bilibili)"
+    aliases = ["bilibili-chat", "b站"]
+
+    def __init__(self):
+        self._config_adapter = BilibiliConfigAdapter()
+        self._gateway = BilibiliGateway()
+        self._probe = BilibiliProbe()
+        self._outbound: BilibiliOutbound | None = None
+        self._security: BilibiliSecurity | None = None
+
+    @property
+    def capabilities(self) -> ChannelCapabilities:
+        return ChannelCapabilities(
+            chat_types=["danmaku", "direct"],
+            message_types=["text", "image"],
+            reactions=False,
+            typing_indicator=False,
+            threads=False,
+            edit=False,
+            unsend=False,
+            reply=True,
+            media=True,
+            native_commands=False,
+            polls=False,
+            streaming=True,
+            streaming_mode="block",
+            block_streaming=True,
+            block_streaming_chunk_min_chars=200,
+            block_streaming_chunk_max_chars=1200,
+            block_streaming_coalesce_min_chars=80,
+            block_streaming_coalesce_max_chars=400,
+            block_streaming_coalesce_idle_ms=500,
+            group_management=False,
+        )
+
+    def list_account_ids(self, config: dict) -> list[str]:
+        return self._config_adapter.list_account_ids(config)
+
+    async def resolve_account(self, account_id: str) -> dict:
+        return await self._config_adapter.resolve_account(account_id)
+
+    def is_configured(self, account: dict) -> bool:
+        return self._config_adapter.is_configured(account)
+
+    def is_enabled(self, account: dict) -> bool:
+        return self._config_adapter.is_enabled(account)
+
+    def describe_account(self, account: dict) -> dict:
+        return self._config_adapter.describe_account(account)
+
+    def config_schema(self) -> dict:
+        return self._config_adapter.config_schema()
+
+    def collect_warnings(self, config: dict, account_id: str | None = None, account: dict | None = None) -> list[str]:
+        security = self._security or BilibiliSecurity(
+            self._config_adapter.get_account_config(config, account_id or "default")
+        )
+        return security.collect_warnings(config, account_id, account)
+
+    async def start(self, ctx) -> object:
+        result = await self._gateway.start(ctx)
+
+        config = getattr(ctx, "config", {}) or {}
+        account_id = getattr(ctx, "account_id", "default")
+        acct_config = self._config_adapter.get_account_config(config, account_id)
+
+        if isinstance(result, dict) and result.get("running"):
+            self._outbound = BilibiliOutbound(
+                credential=self._gateway.credential,
+                anti_risk=self._gateway.anti_risk,
+                config=BilibiliConfig(accounts={"default": acct_config}),
+            )
+            self._security = BilibiliSecurity(acct_config)
+
+        return result
+
+    async def stop(self, ctx) -> None:
+        await self._gateway.stop(ctx)
+        if self._outbound:
+            await self._outbound.close()
+            self._outbound = None
+        self._security = None
+
+    async def send_text(
+        self,
+        target_id: str,
+        content: str,
+        *,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+        account_id: str | None = None,
+    ) -> None:
+        if self._outbound is None:
+            logger.warning("B站 outbound 未初始化，无法发送消息")
+            return
+        await self._outbound.send_text(to=target_id, text=content, account_id=account_id or "default")
+
+    async def send_media(
+        self,
+        target_id: str,
+        media_url: str,
+        media_type: str,
+        reply_to_id: str | None = None,
+        thread_id: str | None = None,
+    ) -> None:
+        if self._outbound is None:
+            return
+        if media_type == "image":
+            image_url = await self._outbound.upload_image(media_url)
+            if image_url:
+                await self._outbound.send_text(to=target_id, text=f"[图片] {image_url}")
+        else:
+            await self._outbound.send_text(to=target_id, text=media_url)
+
+    async def probe(self, account: dict) -> bool:
+        result = await self._probe.probe(account)
+        return result.get("ok", False)
+
+    def build_summary(self, snapshot: object) -> dict:
+        info: dict = {"channel": "bilibili"}
+        if self._outbound is not None:
+            info["outbound_ready"] = True
+        if self._security:
+            info["dm_policy"] = self._security.resolve_dm_policy()
+        return info
+
+    async def check_allowlist(self, peer_id: str, channel_type: str) -> bool:
+        if self._security:
+            if channel_type == "danmaku":
+                return True
+            return self._security.check_dm_policy(peer_id)
+        return True
+
+    def resolve_dm_policy(self) -> dict:
+        if self._security:
+            return {"mode": self._security.resolve_dm_policy(), "allow_from": []}
+        return {"mode": "pairing", "allow_from": []}
+
+    async def generate_code(self, peer_id: str) -> str:
+        return generate_code(peer_id)
+
+    async def verify_code(self, peer_id: str, code: str) -> bool:
+        return verify_code(peer_id, code)
+
+    def normalize_allow_entry(self, entry: str) -> str:
+        return normalize_allow_entry(entry)
+
+    @property
+    def streaming_mode(self) -> str:
+        return "block"
+
+    @property
+    def block_streaming_enabled(self) -> bool:
+        return True
+
+    def create_block_chunker(self) -> object:
+        anti_risk = self._gateway.anti_risk or AntiRisk(level="moderate", multiplier=1.0)
+        return BilibiliBlockChunker(
+            outbound=self._outbound,
+            target_id="",
+            anti_risk=anti_risk,
+        )
+
+    def resolve_session(self, msg: object):
+        from yuxi.channel.protocols import SessionResolution
+
+        if not hasattr(msg, "channel_type") or msg.channel_type != "bilibili":
+            return SessionResolution(kind="group", conversation_id="unknown")
+
+        metadata = getattr(msg, "metadata", None) or {}
+        source = metadata.get("source", "")
+
+        if source == "bilibili_danmaku":
+            group = getattr(msg, "group", None)
+            room_id = group.id if group and group.id else "unknown"
+            return SessionResolution(kind="danmaku", conversation_id=str(room_id))
+
+        if source == "bilibili_dm":
+            sender = getattr(msg, "sender", None)
+            sender_id = sender.id if sender else "unknown"
+            return SessionResolution(kind="direct", conversation_id=str(sender_id))
+
+        return SessionResolution(kind="group", conversation_id="unknown")
+
+    def resolve_target(
+        self,
+        to: str | None = None,
+        *,
+        config: dict | None = None,
+        allow_from: list[str] | None = None,
+        account_id: str | None = None,
+        mode: str | None = None,
+    ) -> tuple[bool, str]:
+        if not to:
+            return False, "target is required"
+        return True, to
+
+    def markdown_to_native(self, md_text: str) -> str:
+        text = re.sub(r"\*\*(.+?)\*\*", r"\1", md_text)
+        text = re.sub(r"\[([^\]]+)]\([^)]+\)", r"\1", text)
+        text = re.sub(r"`([^`]+)`", r"\1", text)
+        return text
+
+    def sanitize_text(self, text: str) -> str:
+        return re.sub(r"<[^>]*>", "", text)
+
+    def classify_error(self, error: BaseException) -> object:
+        if isinstance(error, httpx.HTTPStatusError):
+            code = error.response.status_code
+            if code == 429:
+                return ClassifiedError(
+                    severity=ErrorSeverity.RATE_LIMITED,
+                    original_error=error,
+                    error_message=str(error),
+                )
+            if 400 <= code < 500:
+                return ClassifiedError(
+                    severity=ErrorSeverity.FATAL,
+                    original_error=error,
+                    error_message=str(error),
+                )
+            return ClassifiedError(
+                severity=ErrorSeverity.RETRYABLE,
+                original_error=error,
+                error_message=str(error),
+            )
+        msg = str(error).lower()
+        if any(kw in msg for kw in ("rate", "limit", "throttle", "频繁")):
+            return ClassifiedError(
+                severity=ErrorSeverity.RATE_LIMITED,
+                original_error=error,
+                error_message=str(error),
+            )
+        if any(kw in msg for kw in ("timeout", "connection", "network")):
+            return ClassifiedError(
+                severity=ErrorSeverity.NETWORK,
+                original_error=error,
+                error_message=str(error),
+            )
+        if any(kw in msg for kw in ("forbidden", "auth", "expired", "permission")):
+            return ClassifiedError(
+                severity=ErrorSeverity.FORBIDDEN,
+                original_error=error,
+                error_message=str(error),
+            )
+        return super().classify_error(error)
+
+    @property
+    def ttl_seconds(self) -> int:
+        return 600
+
+    @property
+    def max_entries(self) -> int:
+        return 20000
+
+
+bilibili_plugin = BilibiliPlugin()
+ChannelPluginRegistry.register(bilibili_plugin)
+logger.info("Bilibili plugin registered: id=%s name=%s", bilibili_plugin.id, bilibili_plugin.name)
--- a/backend/package/yuxi/channel/extensions/bilibili/anti_risk.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/anti_risk.py
@ -0,0 +1,69 @@
+from __future__ import annotations
+
+import asyncio
+import random
+import time
+
+
+class RateLimitExceeded(Exception):
+    pass
+
+
+class AntiRisk:
+    RATE_LIMITS = {
+        "dm_send": (30, 60),
+        "dm_poll": (20, 60),
+        "danmaku_send": (5, 60),
+        "comment_reply": (10, 60),
+        "follow_user": (5, 60),
+    }
+
+    DELAY_RANGES = {
+        "dm_send": (1.5, 4.0),
+        "dm_poll": (3.0, 8.0),
+        "danmaku_send": (2.0, 5.0),
+        "comment_reply": (2.0, 5.0),
+        "follow_user": (3.0, 6.0),
+    }
+
+    def __init__(self, level: str = "moderate", multiplier: float = 1.5):
+        self._level = level
+        self._multiplier = multiplier
+        self._action_timestamps: dict[str, list[float]] = {}
+
+    async def throttle(self, action: str) -> None:
+        base_delay = self.DELAY_RANGES.get(action, (1.0, 2.0))
+        min_d, max_d = base_delay
+
+        level_mult = {"low": 0.5, "moderate": 1.0, "strict": 2.0}[self._level]
+        delay = random.uniform(min_d, max_d) * self._multiplier * level_mult
+
+        jitter = random.uniform(0.7, 1.3)
+        delay *= jitter
+
+        self._check_rate_limit(action)
+
+        await asyncio.sleep(delay)
+        self._record_action(action)
+
+    def random_delay(self, min_sec: float, max_sec: float) -> float:
+        base = random.uniform(min_sec, max_sec)
+        level_mult = {"low": 0.6, "moderate": 1.0, "strict": 2.0}[self._level]
+        return base * self._multiplier * level_mult
+
+    def _check_rate_limit(self, action: str) -> None:
+        limit = self.RATE_LIMITS.get(action, (100, 60))
+        max_count, window = limit
+
+        timestamps = self._action_timestamps.get(action, [])
+        now = time.time()
+
+        recent = [t for t in timestamps if now - t < window]
+        if len(recent) >= max_count:
+            raise RateLimitExceeded(f"操作 '{action}' 超过速率限制: {len(recent)}/{max_count} in {window}s")
+        self._action_timestamps[action] = recent
+
+    def _record_action(self, action: str) -> None:
+        if action not in self._action_timestamps:
+            self._action_timestamps[action] = []
+        self._action_timestamps[action].append(time.time())
--- a/backend/package/yuxi/channel/extensions/bilibili/client.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/client.py
@ -0,0 +1,88 @@
+from __future__ import annotations
+
+import logging
+
+from yuxi.channel.extensions.bilibili.types import BilibiliAccountConfig
+
+
+class BilibiliClientManager:
+    def __init__(self, account: BilibiliAccountConfig, logger: logging.Logger):
+        self._account = account
+        self._logger = logger
+        self._credential = None
+        self._img_key: str | None = None
+        self._sub_key: str | None = None
+
+    @property
+    def credential(self):
+        if self._credential is None:
+            from bilibili_api import Credential
+
+            self._credential = Credential(
+                sessdata=self._account.sessdata,
+                bili_jct=self._account.bili_jct,
+                buvid3=self._account.buvid3 or "",
+                dedeuserid=self._account.dedeuserid,
+            )
+        return self._credential
+
+    @property
+    def img_key(self) -> str | None:
+        return self._img_key
+
+    @property
+    def sub_key(self) -> str | None:
+        return self._sub_key
+
+    async def setup(self) -> None:
+        try:
+            from bilibili_api import select_client
+
+            select_client("curl_cffi")
+            try:
+                from bilibili_api import request_settings
+
+                request_settings.set("impersonate", "chrome131")
+            except Exception:
+                pass
+        except Exception:
+            self._logger.warning("curl_cffi 不可用，回退到 aiohttp")
+            try:
+                from bilibili_api import select_client
+
+                select_client("aiohttp")
+            except Exception:
+                pass
+
+        await self._fetch_wbi_keys()
+
+    async def verify_credential(self) -> bool:
+        try:
+            from bilibili_api import user
+
+            u = user.User(self.credential)
+            info = await u.get_user_info()
+            return info is not None
+        except Exception as e:
+            self._logger.error(f"Cookie 验证失败: {e}")
+            return False
+
+    async def _fetch_wbi_keys(self) -> None:
+        try:
+            import httpx
+
+            async with httpx.AsyncClient() as client:
+                resp = await client.get(
+                    "https://api.bilibili.com/x/web-interface/nav",
+                    cookies=self.credential.get_cookies(),
+                )
+                data = resp.json()
+                wbi_img = data.get("data", {}).get("wbi_img", {})
+                img_url = wbi_img.get("img_url", "")
+                sub_url = wbi_img.get("sub_url", "")
+                if img_url and sub_url:
+                    self._img_key = img_url.rsplit("/", 1)[-1].split(".")[0]
+                    self._sub_key = sub_url.rsplit("/", 1)[-1].split(".")[0]
+                    self._logger.debug("WBI keys 已获取")
+        except Exception:
+            self._logger.warning("WBI key 获取失败，部分 API 可能不可用")
--- a/backend/package/yuxi/channel/extensions/bilibili/config.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/config.py
@ -0,0 +1,176 @@
+from __future__ import annotations
+
+import logging
+
+from yuxi.channel.extensions.bilibili.types import (
+    BilibiliAccountConfig,
+    BilibiliConfig,
+    is_account_configured,
+)
+
+logger = logging.getLogger(__name__)
+
+_MERGE_FIELDS = (
+    "sessdata",
+    "bili_jct",
+    "dedeuserid",
+    "buvid3",
+    "buvid4",
+    "bili_ticket",
+    "require_mention",
+    "rate_limit_multiplier",
+)
+
+
+class BilibiliConfigAdapter:
+    def __init__(self):
+        self._config: dict = {}
+
+    @property
+    def _bl_cfg(self) -> dict:
+        return self._config.get("channels", {}).get("bilibili", {})
+
+    def list_account_ids(self, config: dict) -> list[str]:
+        self._config = config
+        accounts = self._bl_cfg.get("accounts", {})
+        if accounts:
+            return list(accounts.keys())
+        return ["default"]
+
+    async def resolve_account(self, account_id: str) -> dict:
+        return self._build_account(account_id)
+
+    def is_configured(self, account: dict) -> bool:
+        acct = self._dict_to_account_config(account)
+        return is_account_configured(acct)
+
+    def is_enabled(self, account: dict) -> bool:
+        return account.get("enabled", True)
+
+    def describe_account(self, account: dict) -> dict:
+        acct = self._dict_to_account_config(account)
+        return {
+            "account_id": acct.dedeuserid or "",
+            "uid": acct.dedeuserid,
+            "room_count": len(acct.room_ids),
+            "dm_enabled": acct.dm_enabled,
+            "configured": is_account_configured(acct),
+        }
+
+    def default_account_id(self) -> str:
+        return self._bl_cfg.get("default_account", "default")
+
+    def get_account_config(self, config: dict, account_id: str = "default") -> BilibiliAccountConfig:
+        cfg = self._parse_config(config)
+        account = cfg.accounts.get(account_id)
+        if account is None:
+            return BilibiliAccountConfig()
+
+        if account_id == "default":
+            merged = account.model_copy()
+            for field_name in _MERGE_FIELDS:
+                base_val = getattr(cfg, field_name, None)
+                is_empty = False
+                if isinstance(base_val, str) and not base_val:
+                    is_empty = True
+                elif isinstance(base_val, list) and not base_val:
+                    is_empty = True
+                if base_val is None or is_empty:
+                    continue
+                setattr(merged, field_name, base_val)
+            return merged
+        return account
+
+    def get_anti_risk_level(self, config: dict) -> str:
+        cfg = self._parse_config(config)
+        return cfg.anti_risk_level
+
+    def config_schema(self) -> dict:
+        return {
+            "$schema": "https://json-schema.org/draft-07/schema#",
+            "type": "object",
+            "title": "B站 渠道配置",
+            "properties": {
+                "sessdata": {
+                    "type": "string",
+                    "title": "SESSDATA",
+                    "description": "B站 Cookie SESSDATA",
+                    "x-ui-password": True,
+                },
+                "bili_jct": {
+                    "type": "string",
+                    "title": "bili_jct (CSRF Token)",
+                    "description": "B站 CSRF Token",
+                    "x-ui-password": True,
+                },
+                "dedeuserid": {
+                    "type": "string",
+                    "title": "DedeUserID",
+                    "description": "B站用户 UID",
+                },
+                "room_ids": {
+                    "type": "array",
+                    "items": {"type": "integer"},
+                    "title": "直播间 ID 列表",
+                },
+                "dm_policy": {
+                    "type": "string",
+                    "title": "DM 策略",
+                    "enum": ["pairing", "allowlist", "open", "disabled"],
+                    "default": "pairing",
+                },
+                "comment_enabled": {
+                    "type": "boolean",
+                    "title": "评论功能",
+                    "default": False,
+                    "x-ui-hidden": True,
+                },
+                "comment_oids": {
+                    "type": "array",
+                    "items": {"type": "integer"},
+                    "title": "评论 OID 列表",
+                    "x-ui-hidden": True,
+                },
+                "require_mention": {
+                    "type": "boolean",
+                    "title": "弹幕提及要求",
+                    "default": True,
+                },
+            },
+        }
+
+    def _parse_config(self, config_dict: dict | None) -> BilibiliConfig:
+        if config_dict is None:
+            return BilibiliConfig()
+        channel_cfg = config_dict.get("channels", {}).get("bilibili", {})
+        if isinstance(channel_cfg, dict):
+            accounts_raw = channel_cfg.get("accounts", {})
+            accounts = {}
+            for aid, acct in accounts_raw.items():
+                if isinstance(acct, BilibiliAccountConfig):
+                    accounts[aid] = acct
+                elif isinstance(acct, dict):
+                    accounts[aid] = BilibiliAccountConfig(**acct)
+            return BilibiliConfig(
+                enabled=channel_cfg.get("enabled", True),
+                name=channel_cfg.get("name"),
+                default_account=channel_cfg.get("default_account", "default"),
+                accounts=accounts,
+                anti_risk_level=channel_cfg.get("anti_risk_level", "moderate"),
+            )
+        if isinstance(channel_cfg, BilibiliConfig):
+            return channel_cfg
+        return BilibiliConfig()
+
+    def _build_account(self, account_id: str) -> dict:
+        return self._get_account_or_default(account_id).model_dump()
+
+    def _get_account_or_default(self, account_id: str) -> BilibiliAccountConfig:
+        cfg = self._parse_config(self._config)
+        return cfg.accounts.get(account_id) or BilibiliAccountConfig()
+
+    @staticmethod
+    def _dict_to_account_config(account: dict) -> BilibiliAccountConfig:
+        if isinstance(account, BilibiliAccountConfig):
+            return account
+        return BilibiliAccountConfig(**account) if account else BilibiliAccountConfig()
--- a/backend/package/yuxi/channel/extensions/bilibili/danmaku/init.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/danmaku/init.py
--- a/backend/package/yuxi/channel/extensions/bilibili/danmaku/handler.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/danmaku/handler.py
@ -0,0 +1,356 @@
+from __future__ import annotations
+
+import logging
+import time
+
+from yuxi.channel.extensions.bilibili.danmaku.types import DanmakuMessage, SuperChatEvent
+from yuxi.channel.extensions.bilibili.types import BilibiliAccountConfig
+from yuxi.channel.message.models import (
+    GroupContext,
+    MessageType,
+    PeerInfo,
+    UnifiedMessage,
+)
+from yuxi.channel.routing.models import PeerKind
+
+logger = logging.getLogger(__name__)
+
+_GUARD_LEVEL_NAME = {1: "总督", 2: "提督", 3: "舰长"}
+
+
+class DanmakuEventHandler:
+    def __init__(self, room_id: int, account, runtime, account_id: str = "default"):
+        self._room_id = room_id
+        self._account = account
+        self._runtime = runtime
+        self._account_id = account_id
+        self._bot_uid = str(account.dedeuserid)
+        self._room_users_seen: set[str] = set()
+
+    async def handle_danmaku(self, data: dict) -> None:
+        info = data["info"]
+        text = info[1]
+        user_info = info[2]
+        uid = str(user_info[0])
+        uname = user_info[1]
+
+        msg = DanmakuMessage(
+            room_id=self._room_id,
+            uid=uid,
+            uname=uname,
+            text=text,
+            timestamp=time.time(),
+            fans_medal_name=info[3][1] if len(info[3]) > 1 else "",
+            fans_medal_level=info[3][0] if len(info[3]) > 0 else 0,
+            ul_level=info[5][0] if len(info[5]) > 0 else 0,
+        )
+
+        if uid == self._bot_uid:
+            return
+
+        if self._account.require_mention and not self._should_respond(msg):
+            return
+
+        await self._dispatch_to_agent(msg)
+
+    async def handle_super_chat(self, data: dict) -> None:
+        event = SuperChatEvent(
+            room_id=self._room_id,
+            uid=str(data["uid"]),
+            uname=data["user_info"]["uname"],
+            message=data["message"],
+            price=data["price"],
+            start_time=data["start_time"],
+            end_time=data["end_time"],
+            timestamp=time.time(),
+        )
+        await self._dispatch_to_agent(event)
+
+    async def handle_gift(self, data: dict) -> None:
+        uname = data.get("uname", "")
+        uid = str(data.get("uid", ""))
+        num = data.get("num", 0)
+        gift_name = data.get("giftName", "")
+        price = data.get("price", 0)
+
+        logger.info("直播间 %s 礼物: %s 送出 %sx %s", self._room_id, uname, num, gift_name)
+
+        content = f"[礼物通知] {uname} 送出了 {num}x {gift_name}"
+        if price:
+            content += f" (总价 {price} 电池)"
+
+        await self._dispatch_event("gift", uid, uname, content)
+
+    async def handle_guard(self, data: dict) -> None:
+        username = data.get("username", "")
+        uid = str(data.get("uid", ""))
+        guard_level = data.get("guard_level", 0)
+        guard_name = _GUARD_LEVEL_NAME.get(guard_level, f"舰队Lv{guard_level}")
+
+        logger.info("直播间 %s 大航海: %s 开通 %s", self._room_id, username, guard_name)
+
+        content = f"[舰队通知] {username} 开通了{guard_name}！"
+        await self._dispatch_event("guard", uid, username, content)
+
+    async def handle_interact(self, data: dict) -> None:
+        msg_type = data.get("msg_type", 0)
+        if msg_type != 1:
+            return
+
+        uid = str(data.get("uid", ""))
+        uname = data.get("uname", "")
+
+        if uid in self._room_users_seen:
+            return
+        self._room_users_seen.add(uid)
+
+        content = f"[进房通知] {uname} 进入了直播间"
+        await self._dispatch_event("enter", uid, uname, content)
+
+    async def handle_live(self, data: dict) -> None:
+        logger.info("直播间 %s 开播", self._room_id)
+        await self._dispatch_status_event("live", "[直播状态] 开播了")
+
+    async def handle_preparing(self, data: dict) -> None:
+        logger.info("直播间 %s 下播", self._room_id)
+        await self._dispatch_status_event("preparing", "[直播状态] 下播了")
+
+    async def handle_cut_off(self, data: dict) -> None:
+        logger.warning("直播间 %s 被超管切断", self._room_id)
+        await self._dispatch_status_event("cut_off", "[直播状态] 直播间被超管切断")
+
+    async def handle_warning(self, data: dict) -> None:
+        logger.warning("直播间 %s 收到管理员警告", self._room_id)
+        await self._dispatch_status_event("warning", "[直播状态] 收到管理员警告")
+
+    async def handle_combo_send(self, data: dict) -> None:
+        uname = data.get("uname", "")
+        uid = str(data.get("uid", ""))
+        combo_num = data.get("combo_num", 0)
+        gift_name = data.get("gift_name", "")
+
+        logger.info("直播间 %s 连击: %s 连送 %s 个 %s", self._room_id, uname, combo_num, gift_name)
+
+        content = f"[连击礼物] {uname} 连送 {combo_num} 个 {gift_name}"
+        await self._dispatch_event("combo_send", uid, uname, content)
+
+    async def handle_sc_delete(self, data: dict) -> None:
+        message_id = data.get("message_id", "")
+        logger.info("直播间 %s SC 删除: message_id=%s", self._room_id, message_id)
+        content = f"[SC 删除] 消息 #{message_id} 已被删除"
+        await self._dispatch_status_event("sc_delete", content)
+
+    async def handle_user_toast(self, data: dict) -> None:
+        uid = str(data.get("uid", ""))
+        uname = data.get("username", "")
+        toast_msg = data.get("toast_msg", "")
+        guard_level = data.get("guard_level", 0)
+        price = data.get("price", 0)
+        num = data.get("num", 0)
+        gift_name = data.get("gift_name", "")
+
+        logger.info("直播间 %s Toast: %s - %s", self._room_id, uname, toast_msg)
+
+        if guard_level:
+            guard_name = _GUARD_LEVEL_NAME.get(guard_level, f"舰队Lv{guard_level}")
+            content = f"[上舰感谢] {uname} 开通{guard_name}！"
+        elif gift_name:
+            content = f"[礼物感谢] {uname} 赠送 {num}x {gift_name}"
+            if price:
+                content += f" (总价 {price} 电池)"
+        else:
+            content = f"[系统通知] {toast_msg}"
+
+        await self._dispatch_event("toast", uid, uname, content)
+
+    async def handle_welcome_guard(self, data: dict) -> None:
+        uid = str(data.get("uid", ""))
+        username = data.get("username", "")
+        guard_level = data.get("guard_level", 0)
+        guard_name = _GUARD_LEVEL_NAME.get(guard_level, f"舰队Lv{guard_level}")
+
+        logger.info("直播间 %s 舰长入场: %s (%s)", self._room_id, username, guard_name)
+        content = f"[舰长入场] {username} ({guard_name}) 进入了直播间"
+        await self._dispatch_event("welcome_guard", uid, username, content)
+
+    async def handle_entry_effect(self, data: dict) -> None:
+        uid = str(data.get("uid", ""))
+        uname = data.get("uname", "")
+        effect_id = data.get("effect_id", 0)
+
+        logger.info("直播间 %s 进场特效: %s (effect_id=%s)", self._room_id, uname, effect_id)
+        content = f"[进场特效] {uname} 进入了直播间"
+        await self._dispatch_event("entry_effect", uid, uname, content)
+
+    async def handle_super_chat_jpn(self, data: dict) -> None:
+        event = SuperChatEvent(
+            room_id=self._room_id,
+            uid=str(data.get("uid", "")),
+            uname=data.get("user_info", {}).get("uname", ""),
+            message=data.get("message", ""),
+            price=data.get("price", 0),
+            start_time=data.get("start_time", 0),
+            end_time=data.get("end_time", 0),
+            timestamp=time.time(),
+        )
+        await self._dispatch_to_agent(event)
+
+    async def handle_pk_start(self, data: dict) -> None:
+        pk_id = data.get("pk_id", "")
+        logger.info("直播间 %s PK 开始: pk_id=%s", self._room_id, pk_id)
+        content = "[PK通知] 大乱斗开始了！"
+        await self._dispatch_status_event("pk_start", content)
+
+    async def handle_pk_end(self, data: dict) -> None:
+        pk_id = data.get("pk_id", "")
+        logger.info("直播间 %s PK 结束: pk_id=%s", self._room_id, pk_id)
+        content = "[PK通知] 大乱斗结束了"
+        await self._dispatch_status_event("pk_end", content)
+
+    async def handle_pk_process(self, data: dict) -> None:
+        logger.debug("直播间 %s PK 进行中", self._room_id)
+
+    async def handle_anchor_lot_start(self, data: dict) -> None:
+        award_name = data.get("award_name", "")
+        logger.info("直播间 %s 天选时刻开始: %s", self._room_id, award_name)
+        content = f"[天选时刻] 抽奖开始 - {award_name}"
+        await self._dispatch_status_event("anchor_lot_start", content)
+
+    async def handle_anchor_lot_end(self, data: dict) -> None:
+        logger.info("直播间 %s 天选时刻结束", self._room_id)
+        content = "[天选时刻] 抽奖已结束"
+        await self._dispatch_status_event("anchor_lot_end", content)
+
+    async def handle_anchor_lot_award(self, data: dict) -> None:
+        award_name = data.get("award_name", "")
+        winners = data.get("award_users", [])
+        winner_names = [w.get("uname", "") for w in winners]
+        logger.info("直播间 %s 天选时刻开奖: %s -> %s", self._room_id, award_name, winner_names)
+        content = f"[天选时刻] 恭喜 {', '.join(winner_names)} 获得 {award_name}"
+        await self._dispatch_status_event("anchor_lot_award", content)
+
+    async def handle_notice_msg(self, data: dict) -> None:
+        msg_common = data.get("msg_common", "")
+        business_id = data.get("business_id", "")
+        msg_type = data.get("msg_type", 0)
+
+        logger.info("直播间 %s 系统通知: type=%s business=%s", self._room_id, msg_type, business_id)
+        content = f"[系统通知] {msg_common}"
+        await self._dispatch_status_event("notice", content)
+
+    async def handle_welcome(self, data: dict) -> None:
+        uname = data.get("uname", "")
+        logger.debug("直播间 %s 老爷入场: %s", self._room_id, uname)
+
+    async def handle_online_rank(self, data: dict) -> None:
+        logger.debug("直播间 %s 高能榜更新 (count=...)", self._room_id)
+
+    async def handle_watched_change(self, data: dict) -> None:
+        num = data.get("num", 0)
+        text_large = data.get("text_large", "")
+        logger.debug("直播间 %s 观看人数: %s (%s)", self._room_id, num, text_large)
+
+    async def handle_like_info(self, data: dict) -> None:
+        logger.debug("直播间 %s 点赞更新", self._room_id)
+
+    async def handle_room_change(self, data: dict) -> None:
+        title = data.get("title", "")
+        area_name = data.get("area_name", "")
+        logger.info("直播间 %s 信息变更: title=%s area=%s", self._room_id, title, area_name)
+
+    async def handle_voice_join(self, data: dict) -> None:
+        logger.debug("直播间 %s 语音连麦状态变更", self._room_id)
+
+    async def handle_red_pocket_start(self, data: dict) -> None:
+        logger.debug("直播间 %s 红包活动开始", self._room_id)
+
+    async def handle_red_pocket_winner(self, data: dict) -> None:
+        logger.debug("直播间 %s 红包中奖名单", self._room_id)
+
+    def _should_respond(self, msg: DanmakuMessage) -> bool:
+        text = msg.text.strip()
+        keywords = self._account.bot_keywords or []
+        return any(kw in text for kw in keywords) or len(text) > 0
+
+    async def _dispatch_to_agent(self, msg) -> None:
+        content = msg.text if hasattr(msg, "text") else msg.message
+        unified = UnifiedMessage(
+            msg_id=f"danmaku_{msg.room_id}_{msg.uid}_{int(msg.timestamp)}",
+            channel_type="bilibili",
+            account_id=self._account_id,
+            content=content,
+            sender=PeerInfo(
+                kind=PeerKind.GROUP,
+                id=msg.uid,
+                display_name=msg.uname,
+            ),
+            group=GroupContext(
+                id=str(msg.room_id),
+                route_peer_kind="group",
+                route_peer_id=str(msg.room_id),
+            ),
+            message_type=MessageType.TEXT,
+            timestamp=time.time(),
+            metadata={"source": "bilibili_danmaku"},
+        )
+
+        queue = getattr(self._runtime, "queue", None)
+        if queue:
+            await queue.put(unified)
+        else:
+            logger.warning("运行时队列不可用，无法分发弹幕消息")
+
+    async def _dispatch_event(self, event_type: str, uid: str, uname: str, content: str) -> None:
+        unified = UnifiedMessage(
+            msg_id=f"{event_type}_{self._room_id}_{uid}_{int(time.time())}",
+            channel_type="bilibili",
+            account_id=self._account_id,
+            content=content,
+            sender=PeerInfo(kind=PeerKind.GROUP, id=uid, display_name=uname),
+            group=GroupContext(
+                id=str(self._room_id),
+                route_peer_kind="group",
+                route_peer_id=str(self._room_id),
+            ),
+            message_type=MessageType.EVENT,
+            timestamp=time.time(),
+            metadata={"source": "bilibili_danmaku", "event_type": event_type},
+        )
+
+        queue = getattr(self._runtime, "queue", None)
+        if queue:
+            await queue.put(unified)
+        else:
+            logger.warning("运行时队列不可用，无法分发 %s 事件", event_type)
+
+    async def _dispatch_status_event(self, event_type: str, content: str) -> None:
+        unified = UnifiedMessage(
+            msg_id=f"live_status_{self._room_id}_{event_type}_{int(time.time())}",
+            channel_type="bilibili",
+            account_id=self._account_id,
+            content=content,
+            sender=PeerInfo(kind=PeerKind.GROUP, id="system", display_name="B站系统"),
+            group=GroupContext(
+                id=str(self._room_id),
+                route_peer_kind="group",
+                route_peer_id=str(self._room_id),
+            ),
+            message_type=MessageType.EVENT,
+            timestamp=time.time(),
+            metadata={"source": "bilibili_danmaku", "event_type": event_type},
+        )
+
+        queue = getattr(self._runtime, "queue", None)
+        if queue:
+            await queue.put(unified)
+        else:
+            logger.warning("运行时队列不可用，无法分发 %s 状态事件", event_type)
+
+
+def resolve_reply_target(msg: DanmakuMessage, account: BilibiliAccountConfig) -> str | None:
+    if account.danmaku_reply_enabled:
+        return f"danmaku:{msg.room_id}"
+    elif account.dm_enabled:
+        return f"dm:{msg.uid}"
+    else:
+        return None
--- a/backend/package/yuxi/channel/extensions/bilibili/danmaku/monitor.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/danmaku/monitor.py
@ -0,0 +1,291 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+
+from yuxi.channel.extensions.bilibili.danmaku.handler import DanmakuEventHandler
+
+logger = logging.getLogger(__name__)
+
+INITIAL_RECONNECT_DELAY = 2.0
+MAX_RECONNECT_DELAY = 60.0
+RECONNECT_BACKOFF = 2.0
+HEARTBEAT_INTERVAL = 30.0
+HEARTBEAT_TIMEOUT = 60.0
+
+
+class DanmakuMonitor:
+    def __init__(
+        self,
+        credential,
+        room_id: int,
+        account,
+        anti_risk,
+        runtime,
+        abort_signal: asyncio.Event,
+        account_id: str = "default",
+    ):
+        self._credential = credential
+        self._room_id = room_id
+        self._account = account
+        self._anti_risk = anti_risk
+        self._runtime = runtime
+        self._abort_signal = abort_signal
+        self._handler = DanmakuEventHandler(
+            room_id=room_id,
+            account=account,
+            runtime=runtime,
+            account_id=account_id,
+        )
+
+    async def run(self) -> None:
+        from bilibili_api.live import LiveDanmaku
+
+        reconnect_delay = INITIAL_RECONNECT_DELAY
+
+        while not self._abort_signal.is_set():
+            try:
+                logger.info("开始监听直播间 %s (重连延迟=%.1fs)", self._room_id, reconnect_delay)
+
+                danmaku = LiveDanmaku(self._room_id, credential=self._credential)
+                last_heartbeat = time.time()
+
+                @danmaku.on("DANMU_MSG")
+                async def on_danmaku(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    data = event["data"]
+                    await self._handler.handle_danmaku(data)
+
+                @danmaku.on("SUPER_CHAT_MESSAGE")
+                async def on_super_chat(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    data = event["data"]
+                    await self._handler.handle_super_chat(data)
+
+                @danmaku.on("SEND_GIFT")
+                async def on_gift(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    data = event["data"]
+                    await self._handler.handle_gift(data)
+
+                @danmaku.on("GUARD_BUY")
+                async def on_guard(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    data = event["data"]
+                    await self._handler.handle_guard(data)
+
+                @danmaku.on("INTERACT_WORD")
+                async def on_interact(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    data = event["data"]
+                    await self._handler.handle_interact(data)
+
+                @danmaku.on("LIVE")
+                async def on_live(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_live(event.get("data", {}))
+
+                @danmaku.on("PREPARING")
+                async def on_preparing(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_preparing(event.get("data", {}))
+
+                @danmaku.on("CUT_OFF")
+                async def on_cut_off(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_cut_off(event.get("data", {}))
+
+                @danmaku.on("WARNING")
+                async def on_warning(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_warning(event.get("data", {}))
+
+                @danmaku.on("COMBO_SEND")
+                async def on_combo_send(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_combo_send(event.get("data", {}))
+
+                @danmaku.on("SUPER_CHAT_MESSAGE_DELETE")
+                async def on_sc_delete(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_sc_delete(event.get("data", {}))
+
+                @danmaku.on("USER_TOAST_MSG")
+                async def on_user_toast(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_user_toast(event.get("data", {}))
+
+                @danmaku.on("WELCOME_GUARD")
+                async def on_welcome_guard(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_welcome_guard(event.get("data", {}))
+
+                @danmaku.on("ENTRY_EFFECT")
+                async def on_entry_effect(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_entry_effect(event.get("data", {}))
+
+                @danmaku.on("SUPER_CHAT_MESSAGE_JPN")
+                async def on_super_chat_jpn(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_super_chat_jpn(event.get("data", {}))
+
+                @danmaku.on("PK_START")
+                async def on_pk_start(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_pk_start(event.get("data", {}))
+
+                @danmaku.on("PK_END")
+                async def on_pk_end(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_pk_end(event.get("data", {}))
+
+                @danmaku.on("PK_PROCESS")
+                async def on_pk_process(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_pk_process(event.get("data", {}))
+
+                @danmaku.on("ANCHOR_LOT_START")
+                async def on_anchor_lot_start(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_anchor_lot_start(event.get("data", {}))
+
+                @danmaku.on("ANCHOR_LOT_END")
+                async def on_anchor_lot_end(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_anchor_lot_end(event.get("data", {}))
+
+                @danmaku.on("ANCHOR_LOT_AWARD")
+                async def on_anchor_lot_award(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_anchor_lot_award(event.get("data", {}))
+
+                @danmaku.on("NOTICE_MSG")
+                async def on_notice_msg(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_notice_msg(event.get("data", {}))
+
+                @danmaku.on("WELCOME")
+                async def on_welcome(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_welcome(event.get("data", {}))
+
+                @danmaku.on("ONLINE_RANK_COUNT")
+                async def on_online_rank_count(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_online_rank(event.get("data", {}))
+
+                @danmaku.on("ONLINE_RANK_V2")
+                async def on_online_rank_v2(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_online_rank(event.get("data", {}))
+
+                @danmaku.on("WATCHED_CHANGE")
+                async def on_watched_change(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_watched_change(event.get("data", {}))
+
+                @danmaku.on("LIKE_INFO_V3_CLICK")
+                async def on_like_info_click(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_like_info(event.get("data", {}))
+
+                @danmaku.on("LIKE_INFO_V3_UPDATE")
+                async def on_like_info_update(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_like_info(event.get("data", {}))
+
+                @danmaku.on("ROOM_CHANGE")
+                async def on_room_change(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_room_change(event.get("data", {}))
+
+                @danmaku.on("VOICE_JOIN_LIST")
+                async def on_voice_join_list(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_voice_join(event.get("data", {}))
+
+                @danmaku.on("VOICE_ROOM_STATUS")
+                async def on_voice_room_status(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_voice_join(event.get("data", {}))
+
+                @danmaku.on("POPULARITY_RED_POCKET_START")
+                async def on_red_pocket_start(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_red_pocket_start(event.get("data", {}))
+
+                @danmaku.on("POPULARITY_RED_POCKET_WINNER_LIST")
+                async def on_red_pocket_winner(event):
+                    nonlocal last_heartbeat
+                    last_heartbeat = time.time()
+                    await self._handler.handle_red_pocket_winner(event.get("data", {}))
+
+                async def heartbeat_checker(dmk):
+                    while not self._abort_signal.is_set():
+                        await asyncio.sleep(HEARTBEAT_INTERVAL)
+                        if time.time() - last_heartbeat > HEARTBEAT_TIMEOUT:
+                            gap = time.time() - last_heartbeat
+                            logger.warning("直播间 %s 心跳超时 (%.1fs 无消息)，主动断连", self._room_id, gap)
+                            try:
+                                await dmk.disconnect()
+                            except Exception:
+                                pass
+                            break
+
+                heartbeat_task = asyncio.create_task(heartbeat_checker(danmaku))
+
+                try:
+                    await danmaku.connect()
+                    reconnect_delay = INITIAL_RECONNECT_DELAY
+                finally:
+                    heartbeat_task.cancel()
+                    try:
+                        await heartbeat_task
+                    except asyncio.CancelledError:
+                        pass
+
+            except asyncio.CancelledError:
+                logger.info("直播间 %s 监听已取消", self._room_id)
+                break
+            except Exception as e:
+                logger.error("直播间 %s 监听异常 (%.1fs 后重连): %s", self._room_id, reconnect_delay, e)
+                try:
+                    await asyncio.wait_for(self._abort_signal.wait(), timeout=reconnect_delay)
+                    break
+                except TimeoutError:
+                    pass
+                reconnect_delay = min(reconnect_delay * RECONNECT_BACKOFF, MAX_RECONNECT_DELAY)
--- a/backend/package/yuxi/channel/extensions/bilibili/danmaku/types.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/danmaku/types.py
@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class DanmakuMessage:
+    room_id: int
+    uid: str
+    uname: str
+    text: str
+    timestamp: float
+    fans_medal_name: str = ""
+    fans_medal_level: int = 0
+    ul_level: int = 0
+
+
+@dataclass
+class SuperChatEvent:
+    room_id: int
+    uid: str
+    uname: str
+    message: str
+    price: float
+    start_time: int
+    end_time: int
+    timestamp: float
+
+
+@dataclass
+class GiftEvent:
+    room_id: int
+    uid: str
+    uname: str
+    gift_name: str
+    gift_num: int
+    price: float
+    timestamp: float
--- a/backend/package/yuxi/channel/extensions/bilibili/gateway.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/gateway.py
@ -0,0 +1,152 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from yuxi.channel.extensions.bilibili.anti_risk import AntiRisk
+from yuxi.channel.extensions.bilibili.client import BilibiliClientManager
+from yuxi.channel.extensions.bilibili.config import BilibiliConfigAdapter
+from yuxi.channel.extensions.bilibili.types import BilibiliAccountConfig
+
+logger = logging.getLogger(__name__)
+
+
+class BilibiliGateway:
+    def __init__(self):
+        self._config_adapter = BilibiliConfigAdapter()
+        self._client_manager: BilibiliClientManager | None = None
+        self._anti_risk: AntiRisk | None = None
+        self._danmaku_tasks: dict[int, asyncio.Task] = {}
+        self._dm_poll_task: asyncio.Task | None = None
+        self._running = False
+        self._abort_signal: asyncio.Event | None = None
+        self._queue: asyncio.Queue | None = None
+
+    @property
+    def credential(self):
+        if self._client_manager is None:
+            return None
+        return self._client_manager.credential
+
+    @property
+    def anti_risk(self) -> AntiRisk | None:
+        return self._anti_risk
+
+    async def start(self, ctx) -> object:
+        account, account_id = await self._resolve_account(ctx)
+
+        if not self._config_adapter.is_configured(account):
+            logger.warning("B站账户 %s 未配置，跳过启动", account_id)
+            return {"running": False, "reason": "not-configured"}
+
+        self._abort_signal = asyncio.Event()
+        self._queue = asyncio.Queue(maxsize=2000)
+        self._running = True
+
+        acct_config = self._config_adapter.get_account_config(ctx.config or {}, account_id)
+
+        self._client_manager = BilibiliClientManager(acct_config, ctx.logger or logger)
+        await self._client_manager.setup()
+
+        if not await self._client_manager.verify_credential():
+            raise RuntimeError("B站 Cookie 验证失败，可能已过期")
+
+        anti_risk_level = self._config_adapter.get_anti_risk_level(ctx.config or {})
+        self._anti_risk = AntiRisk(
+            level=anti_risk_level,
+            multiplier=acct_config.rate_limit_multiplier,
+        )
+
+        for room_id in acct_config.room_ids:
+            task = asyncio.create_task(
+                self._run_danmaku_monitor(room_id, acct_config, ctx, account_id),
+                name=f"bilibili-danmaku:{room_id}",
+            )
+            self._danmaku_tasks[room_id] = task
+
+        if acct_config.dm_enabled:
+            self._dm_poll_task = asyncio.create_task(
+                self._run_dm_poller(acct_config, ctx, account_id),
+                name="bilibili-dm-poller",
+            )
+
+        ctx.cancel_event_monitor = self._abort_signal
+
+        logger.info(
+            "B站网关已启动 account=%s room_count=%s dm=%s",
+            account_id,
+            len(acct_config.room_ids),
+            acct_config.dm_enabled,
+        )
+        return {"running": True, "account_id": account_id, "queue": self._queue}
+
+    async def stop(self, ctx) -> None:
+        self._running = False
+
+        if self._abort_signal:
+            self._abort_signal.set()
+
+        for room_id, task in self._danmaku_tasks.items():
+            task.cancel()
+        if self._dm_poll_task:
+            self._dm_poll_task.cancel()
+
+        for room_id, task in self._danmaku_tasks.items():
+            try:
+                await asyncio.wait_for(task, timeout=5.0)
+            except (TimeoutError, asyncio.CancelledError):
+                pass
+        if self._dm_poll_task:
+            try:
+                await asyncio.wait_for(self._dm_poll_task, timeout=5.0)
+            except (TimeoutError, asyncio.CancelledError):
+                pass
+
+        self._danmaku_tasks.clear()
+        self._dm_poll_task = None
+
+        if self._abort_signal:
+            self._abort_signal = None
+        self._queue = None
+
+        logger.info("B站网关已停止")
+
+    async def _run_danmaku_monitor(
+        self,
+        room_id: int,
+        account: BilibiliAccountConfig,
+        ctx,
+        account_id: str,
+    ) -> None:
+        from yuxi.channel.extensions.bilibili.danmaku.monitor import DanmakuMonitor
+
+        monitor = DanmakuMonitor(
+            credential=self._client_manager.credential,
+            room_id=room_id,
+            account=account,
+            anti_risk=self._anti_risk,
+            runtime=ctx,
+            abort_signal=self._abort_signal,
+            account_id=account_id,
+        )
+        await monitor.run()
+
+    async def _run_dm_poller(self, account: BilibiliAccountConfig, ctx, account_id: str) -> None:
+        from yuxi.channel.extensions.bilibili.private_msg.poller import DMPoller
+
+        poller = DMPoller(
+            credential=self._client_manager.credential,
+            account=account,
+            anti_risk=self._anti_risk,
+            runtime=ctx,
+            abort_signal=self._abort_signal,
+            account_id=account_id,
+        )
+        await poller.run()
+
+    async def _resolve_account(self, ctx) -> tuple[dict, str]:
+        config = getattr(ctx, "config", {}) or {}
+        account_id = getattr(ctx, "account_id", "default")
+        self._config_adapter.list_account_ids(config)
+        account = await self._config_adapter.resolve_account(account_id)
+        return account, account_id
--- a/backend/package/yuxi/channel/extensions/bilibili/outbound.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/outbound.py
@ -0,0 +1,157 @@
+from __future__ import annotations
+
+import json
+import logging
+import time
+
+import httpx
+
+from yuxi.channel.extensions.bilibili.types import BilibiliConfig
+
+logger = logging.getLogger(__name__)
+
+
+class BilibiliOutbound:
+    SEND_DM_URL = "https://api.vc.bilibili.com/web_im/v1/web_im/send_msg"
+
+    def __init__(self, credential, anti_risk, config: BilibiliConfig):
+        self._credential = credential
+        self._anti_risk = anti_risk
+        self._config = config
+        self._client: httpx.AsyncClient | None = None
+
+    async def _ensure_client(self) -> httpx.AsyncClient:
+        if self._client is None:
+            self._client = httpx.AsyncClient()
+        return self._client
+
+    async def close(self) -> None:
+        if self._client:
+            await self._client.aclose()
+            self._client = None
+
+    async def send_text(
+        self,
+        *,
+        to: str,
+        text: str,
+        account_id: str = "default",
+        **kwargs,
+    ) -> dict:
+        prefix, destination = self._parse_target(to)
+
+        if prefix == "dm":
+            return await self._send_dm(destination, text)
+        elif prefix == "danmaku":
+            return await self._send_danmaku(destination, text)
+        elif prefix == "comment":
+            return await self._send_comment(destination, text)
+        else:
+            return {"ok": False, "error": f"Unknown target type: {prefix}"}
+
+    async def _send_dm(self, receiver_uid: str, text: str) -> dict:
+        await self._anti_risk.throttle("dm_send")
+
+        client = await self._ensure_client()
+
+        content_json = json.dumps({"content": text})
+        form_data = {
+            "msg[sender_uid]": self._credential.dedeuserid,
+            "msg[receiver_id]": receiver_uid,
+            "msg[receiver_type]": 1,
+            "msg[msg_type]": 1,
+            "msg[content]": content_json,
+            "msg[timestamp]": int(time.time()),
+            "csrf": self._credential.bili_jct,
+        }
+
+        resp = await client.post(
+            self.SEND_DM_URL,
+            cookies=self._credential.get_cookies(),
+            data=form_data,
+        )
+        result = resp.json()
+        return {
+            "ok": result.get("code") == 0,
+            "message_id": result.get("data", {}).get("msg_key", ""),
+            "error": result.get("message") if result.get("code") != 0 else None,
+        }
+
+    async def _send_danmaku(self, room_id: str, text: str) -> dict:
+        account = self._config.accounts.get("default")
+        if account is None or not account.danmaku_reply_enabled:
+            return {"ok": False, "error": "Danmaku reply is disabled"}
+
+        from bilibili_api.live import LiveDanmaku
+
+        lm = LiveDanmaku(int(room_id), credential=self._credential)
+        result = await lm.send_danmaku(text[:50])
+        return {"ok": result, "message_id": ""}
+
+    async def _send_comment(self, oid: str, text: str) -> dict:
+        await self._anti_risk.throttle("comment_reply")
+
+        client = await self._ensure_client()
+        resp = await client.post(
+            "https://api.bilibili.com/x/v2/reply/add",
+            cookies=self._credential.get_cookies(),
+            data={
+                "oid": oid,
+                "type": 1,
+                "message": text,
+                "plat": 1,
+                "csrf": self._credential.bili_jct,
+            },
+        )
+        result = resp.json()
+        return {
+            "ok": result.get("code") == 0,
+            "message_id": str(result.get("data", {}).get("rpid", "")),
+            "error": result.get("message") if result.get("code") != 0 else None,
+        }
+
+    @staticmethod
+    def _parse_target(to: str) -> tuple[str, str]:
+        if ":" not in to:
+            raise ValueError(f"Invalid target format: {to}")
+        prefix, _, destination = to.partition(":")
+        return prefix, destination
+
+    async def upload_image(self, source: str) -> str | None:
+        if source.startswith(("http://", "https://")):
+            client = await self._ensure_client()
+            resp = await client.get(source)
+            resp.raise_for_status()
+            file_data = resp.content
+        else:
+            import aiofiles
+
+            async with aiofiles.open(source, "rb") as f:
+                file_data = await f.read()
+
+        client = await self._ensure_client()
+        resp = await client.post(
+            "https://api.vc.bilibili.com/api/v1/image/upload",
+            cookies=self._credential.get_cookies(),
+            files={"file_up": ("image.png", file_data, "image/png")},
+            data={"csrf": self._credential.bili_jct},
+        )
+        result = resp.json()
+        if result.get("code") == 0:
+            return result.get("data", {}).get("image_url")
+        return None
+
+    async def follow_user(self, uid: str) -> bool:
+        client = await self._ensure_client()
+        resp = await client.post(
+            "https://api.bilibili.com/x/relation/modify",
+            cookies=self._credential.get_cookies(),
+            data={
+                "fid": uid,
+                "act": 1,
+                "re_src": 11,
+                "csrf": self._credential.bili_jct,
+            },
+        )
+        result = resp.json()
+        return result.get("code") == 0
--- a/backend/package/yuxi/channel/extensions/bilibili/pairing.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/pairing.py
@ -0,0 +1,46 @@
+from __future__ import annotations
+
+import logging
+import random
+import string
+import time
+
+logger = logging.getLogger(__name__)
+
+_pairing_codes: dict[str, tuple[str, float]] = {}
+_pairing_verified: set[str] = set()
+CODE_TTL = 600
+
+
+def generate_code(peer_id: str) -> str:
+    code = "".join(random.choices(string.digits, k=6))
+    _pairing_codes[peer_id] = (code, time.time())
+    logger.info("为 %s 生成配对码: %s", peer_id, code)
+    return code
+
+
+def verify_code(peer_id: str, code: str) -> bool:
+    entry = _pairing_codes.get(peer_id)
+    if entry is None:
+        return False
+    stored_code, created_at = entry
+    if time.time() - created_at > CODE_TTL:
+        del _pairing_codes[peer_id]
+        return False
+    if stored_code == code:
+        del _pairing_codes[peer_id]
+        _pairing_verified.add(peer_id)
+        return True
+    return False
+
+
+def check_pairing(sender_uid: str, account) -> bool:
+    if hasattr(account, "dm_allow_from"):
+        allow_from = account.dm_allow_from or []
+        if str(sender_uid) in allow_from:
+            return True
+    return sender_uid in _pairing_verified
+
+
+def normalize_allow_entry(entry: str) -> str:
+    return entry.strip()
--- a/backend/package/yuxi/channel/extensions/bilibili/plugin.json
+++ b/backend/package/yuxi/channel/extensions/bilibili/plugin.json
@ -0,0 +1,35 @@
+{
+    "id": "bilibili",
+    "name": "B站 (Bilibili)",
+    "version": "0.1.0",
+    "description": "B站渠道插件，支持直播弹幕和私信交互",
+    "author": "ForcePilot Team",
+    "order": 95,
+    "dependencies": ["bilibili-api-python>=17.4.0"],
+    "capabilities": {
+        "chat_types": ["danmaku", "direct"],
+        "message_types": ["text", "image"],
+        "reactions": false,
+        "typing_indicator": false,
+        "threads": false,
+        "edit": false,
+        "unsend": false,
+        "reply": true,
+        "media": true,
+        "native_commands": false,
+        "polls": false,
+        "pins": false,
+        "group_management": false,
+        "adaptive_cards": false,
+        "streaming": true,
+        "streaming_mode": "block",
+        "block_streaming": true,
+        "block_streaming_chunk_min_chars": 200,
+        "block_streaming_chunk_max_chars": 1200,
+        "block_streaming_coalesce_min_chars": 80,
+        "block_streaming_coalesce_max_chars": 400,
+        "block_streaming_coalesce_idle_ms": 500
+    },
+    "enabled": true,
+    "python_requires": ">=3.12"
+}
--- a/backend/package/yuxi/channel/extensions/bilibili/private_msg/init.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/private_msg/init.py
--- a/backend/package/yuxi/channel/extensions/bilibili/private_msg/poller.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/private_msg/poller.py
@ -0,0 +1,198 @@
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+import time
+
+import httpx
+
+from yuxi.channel.message.models import (
+    MessageType,
+    PeerInfo,
+    UnifiedMessage,
+)
+from yuxi.channel.routing.models import PeerKind
+
+logger = logging.getLogger(__name__)
+
+
+class DMPoller:
+    BASE_URL = "https://api.vc.bilibili.com"
+    POLL_INTERVAL_MIN = 3.0
+    POLL_INTERVAL_MAX = 8.0
+    SESSION_LIST_URL = f"{BASE_URL}/session_svr/v1/session_svr/get_sessions"
+    FETCH_MSGS_URL = f"{BASE_URL}/svr_sync/v1/svr_sync/fetch_session_msgs"
+
+    def __init__(
+        self, credential, account, anti_risk, runtime, abort_signal: asyncio.Event, account_id: str = "default"
+    ):
+        self._credential = credential
+        self._account = account
+        self._anti_risk = anti_risk
+        self._runtime = runtime
+        self._abort_signal = abort_signal
+        self._account_id = account_id
+        self._client: httpx.AsyncClient | None = None
+        self._seen_msg_keys: set[str] = set()
+        self._session_seqnos: dict[int, int] = {}
+
+    async def run(self) -> None:
+        logger.info("DMPoller 已启动")
+        self._client = httpx.AsyncClient()
+
+        try:
+            while not self._abort_signal.is_set():
+                try:
+                    sessions = await self._fetch_session_list()
+                    new_msgs = []
+
+                    for session in sessions:
+                        session_key = session.get("talker_id", 0)
+                        unread = session.get("unread_count", 0)
+                        if unread <= 0:
+                            continue
+
+                        msgs = await self._fetch_session_msgs(
+                            int(session_key),
+                            self._session_seqnos.get(session_key, 0),
+                        )
+                        for msg in msgs:
+                            msg_key = msg.get("msg_key", "")
+                            if msg_key not in self._seen_msg_keys:
+                                self._seen_msg_keys.add(msg_key)
+                                new_msgs.append(msg)
+                                self._session_seqnos[session_key] = max(
+                                    self._session_seqnos.get(session_key, 0),
+                                    msg.get("msg_seqno", 0),
+                                )
+
+                    for msg in new_msgs:
+                        await self._handle_message(msg)
+
+                except Exception as e:
+                    logger.error("DMPoller 错误: %s", e)
+
+                delay = self._anti_risk.random_delay(self.POLL_INTERVAL_MIN, self.POLL_INTERVAL_MAX)
+                try:
+                    await asyncio.wait_for(self._abort_signal.wait(), timeout=delay)
+                    break
+                except TimeoutError:
+                    pass
+
+        finally:
+            if self._client:
+                await self._client.aclose()
+                self._client = None
+            logger.info("DMPoller 已停止")
+
+    async def _fetch_session_list(self) -> list[dict]:
+        resp = await self._client.get(
+            self.SESSION_LIST_URL,
+            cookies=self._credential.get_cookies(),
+            params={"session_type": 1},
+        )
+        data = resp.json()
+        if data.get("code") != 0:
+            return []
+        return data.get("data", {}).get("session_list", [])
+
+    async def _fetch_session_msgs(self, session_key: int, begin_seqno: int) -> list[dict]:
+        resp = await self._client.get(
+            self.FETCH_MSGS_URL,
+            cookies=self._credential.get_cookies(),
+            params={
+                "session_key": session_key,
+                "begin_seqno": begin_seqno,
+            },
+        )
+        data = resp.json()
+        if data.get("code") != 0:
+            return []
+        return data.get("data", {}).get("messages", [])
+
+    async def _handle_message(self, msg: dict) -> None:
+        sender_uid = str(msg.get("sender_uid", ""))
+        content = msg.get("content", "")
+        msg_type = msg.get("msg_type", 1)
+        msg_status = msg.get("msg_status", 0)
+
+        if msg_status == 1:
+            text = "[消息已撤回]"
+            message_type = MessageType.EVENT
+        elif msg_type == 1:
+            text = extract_dm_text(content)
+            message_type = MessageType.TEXT
+        elif msg_type == 2:
+            text = "[图片消息]"
+            message_type = MessageType.IMAGE
+            try:
+                content_obj = json.loads(content)
+                img_url = content_obj.get("url", "")
+                if img_url:
+                    text = img_url
+            except (json.JSONDecodeError, TypeError):
+                pass
+        elif msg_type == 5:
+            text = "[文件消息]"
+            message_type = MessageType.TEXT
+        elif msg_type == 6:
+            text = "[表情消息]"
+            message_type = MessageType.IMAGE
+        elif msg_type == 7:
+            text = "[分享链接]"
+            message_type = MessageType.TEXT
+        elif msg_type == 8:
+            text = "[语音消息]"
+            message_type = MessageType.TEXT
+        elif msg_type == 9:
+            text = "[视频消息]"
+            message_type = MessageType.TEXT
+        else:
+            text = f"[消息类型 {msg_type}]"
+            message_type = MessageType.TEXT
+
+        if not self._check_dm_policy(sender_uid):
+            return
+
+        unified = UnifiedMessage(
+            msg_id=msg.get("msg_key", ""),
+            channel_type="bilibili",
+            account_id=self._account_id,
+            content=text,
+            sender=PeerInfo(
+                kind=PeerKind.DIRECT,
+                id=sender_uid,
+            ),
+            message_type=message_type,
+            timestamp=time.time(),
+            metadata={"source": "bilibili_dm", "msg_type": msg_type, "msg_status": msg_status},
+        )
+
+        queue = getattr(self._runtime, "queue", None)
+        if queue:
+            await queue.put(unified)
+        else:
+            logger.warning("运行时队列不可用，无法分发私信消息")
+
+    def _check_dm_policy(self, sender_uid: str) -> bool:
+        policy = self._account.dm_policy
+        if policy == "disabled":
+            return False
+        if policy == "open":
+            return True
+        if policy == "allowlist":
+            return sender_uid in self._account.dm_allow_from
+        if policy == "pairing":
+            from yuxi.channel.extensions.bilibili.pairing import check_pairing
+
+            return check_pairing(sender_uid, self._account)
+        return False
+
+
+def extract_dm_text(raw_content: str) -> str:
+    try:
+        content_obj = json.loads(raw_content)
+        return content_obj.get("content", raw_content)
+    except (json.JSONDecodeError, TypeError):
+        return raw_content
--- a/backend/package/yuxi/channel/extensions/bilibili/probe.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/probe.py
@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import time
+
+import httpx
+
+
+class BilibiliProbe:
+    NAV_URL = "https://api.bilibili.com/x/web-interface/nav"
+
+    async def probe(self, account: dict) -> dict:
+        start = time.time()
+        try:
+            async with httpx.AsyncClient() as client:
+                resp = await client.get(
+                    self.NAV_URL,
+                    cookies={
+                        "SESSDATA": account.get("sessdata", ""),
+                        "bili_jct": account.get("bili_jct", ""),
+                        "DedeUserID": account.get("dedeuserid", ""),
+                    },
+                )
+                data = resp.json()
+            is_valid = data.get("code") == 0 and data.get("data", {}).get("isLogin")
+            latency = time.time() - start
+
+            return {
+                "ok": is_valid,
+                "latency_ms": int(latency * 1000),
+                "error": data.get("message") if not is_valid else None,
+            }
+        except Exception as e:
+            return {
+                "ok": False,
+                "latency_ms": int((time.time() - start) * 1000),
+                "error": str(e),
+            }
--- a/backend/package/yuxi/channel/extensions/bilibili/security.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/security.py
@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from yuxi.channel.extensions.bilibili.types import BilibiliAccountConfig
+
+
+class BilibiliSecurity:
+    def __init__(self, account: BilibiliAccountConfig | None = None):
+        self._account = account or BilibiliAccountConfig()
+
+    def check_dm_policy(self, sender_uid: str) -> bool:
+        policy = self._account.dm_policy
+        if policy == "disabled":
+            return False
+        if policy == "open":
+            return True
+        if policy == "allowlist":
+            return sender_uid in self._account.dm_allow_from
+        if policy == "pairing":
+            from yuxi.channel.extensions.bilibili.pairing import check_pairing
+
+            return check_pairing(sender_uid, self._account)
+        return False
+
+    def check_danmaku_policy(self, room_id: int) -> bool:
+        return room_id in self._account.room_ids
+
+    def resolve_dm_policy(self) -> str:
+        return self._account.dm_policy.value
+
+    def collect_warnings(self, config: dict, account_id: str | None = None, account: dict | None = None) -> list[str]:
+        warnings = []
+        acct = account or {}
+        if acct.get("dm_enabled") and acct.get("dm_policy") in ("open", "allowlist"):
+            warnings.append("私信功能已启用，使用非官方 API，存在账号风险")
+        if not acct.get("room_ids"):
+            warnings.append("未配置直播间 ID，弹幕功能不会启动")
+        return warnings
--- a/backend/package/yuxi/channel/extensions/bilibili/status.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/status.py
@ -0,0 +1,41 @@
+from __future__ import annotations
+
+from yuxi.channel.extensions.bilibili.types import BilibiliAccountConfig, is_account_configured
+
+
+async def get_status(ctx) -> dict:
+    account = await _resolve_account(ctx)
+    issues = []
+
+    if not is_account_configured(account):
+        issues.append({"severity": "error", "message": "Cookie 配置不完整"})
+
+    if not account.room_ids:
+        issues.append({"severity": "warning", "message": "未配置直播间 ID"})
+
+    if account.dm_enabled:
+        issues.append(
+            {
+                "severity": "warning",
+                "message": "私信功能已启用 — 使用非官方 API，存在账号风险",
+            }
+        )
+
+    return {
+        "running": getattr(ctx, "running", False),
+        "configured": is_account_configured(account),
+        "issues": issues,
+        "room_count": len(account.room_ids),
+        "dm_enabled": account.dm_enabled,
+    }
+
+
+async def _resolve_account(ctx) -> BilibiliAccountConfig:
+    config = getattr(ctx, "config", {}) or {}
+    channel_cfg = config.get("channels", {}).get("bilibili", {})
+    account_id = getattr(ctx, "account_id", "default")
+    accounts = channel_cfg.get("accounts", {})
+    account_dict = accounts.get(account_id, {})
+    if isinstance(account_dict, BilibiliAccountConfig):
+        return account_dict
+    return BilibiliAccountConfig(**account_dict) if account_dict else BilibiliAccountConfig()
--- a/backend/package/yuxi/channel/extensions/bilibili/streaming.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/streaming.py
@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import asyncio
+import time
+
+
+class BilibiliBlockChunker:
+    def __init__(
+        self,
+        outbound,
+        target_id: str,
+        anti_risk,
+        chunk_min_chars: int = 200,
+        chunk_max_chars: int = 1200,
+        coalesce_min_chars: int = 80,
+        coalesce_max_chars: int = 400,
+        coalesce_idle_ms: int = 500,
+    ):
+        self._outbound = outbound
+        self._target_id = target_id
+        self._anti_risk = anti_risk
+        self._chunk_min = chunk_min_chars
+        self._chunk_max = chunk_max_chars
+        self._coalesce_min = coalesce_min_chars
+        self._coalesce_max = coalesce_max_chars
+        self._coalesce_idle = coalesce_idle_ms / 1000.0
+        self._buffer = ""
+        self._last_flush = time.time()
+
+    async def append(self, text: str) -> None:
+        self._buffer += text
+        should_flush = len(self._buffer) >= self._chunk_max or (
+            len(self._buffer) >= self._chunk_min and time.time() - self._last_flush > self._coalesce_idle
+        )
+        if should_flush:
+            await self._flush()
+
+    async def finalize(self) -> None:
+        if self._buffer:
+            await self._flush()
+
+    async def _flush(self) -> None:
+        chunks = self._split_chunks(self._buffer, self._chunk_max)
+        for chunk in chunks:
+            await self._outbound.send_text(to=self._target_id, text=chunk)
+            await asyncio.sleep(0.5)
+        self._buffer = ""
+        self._last_flush = time.time()
+
+    def _split_chunks(self, text: str, limit: int) -> list[str]:
+        if len(text) <= limit:
+            return [text] if text else []
+
+        chunks = []
+        remaining = text
+        while len(remaining) > limit:
+            window = remaining[:limit]
+            last_para = window.rfind("\n\n")
+            if last_para > limit * 0.5:
+                chunks.append(window[:last_para].strip())
+                remaining = remaining[last_para + 2 :]
+                continue
+            for sep in ["。\n", "。", "！", "？", "\n", " "]:
+                idx = window.rfind(sep)
+                if idx > limit * 0.3:
+                    chunks.append(window[: idx + len(sep)].strip())
+                    remaining = remaining[idx + len(sep) :]
+                    break
+            else:
+                chunks.append(window.strip())
+                remaining = remaining[limit:]
+        if remaining.strip():
+            chunks.append(remaining.strip())
+        return chunks
--- a/backend/package/yuxi/channel/extensions/bilibili/types.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/types.py
@ -0,0 +1,43 @@
+from enum import StrEnum
+
+from pydantic import BaseModel
+
+
+class DMPolicy(StrEnum):
+    PAIRING = "pairing"
+    ALLOWLIST = "allowlist"
+    OPEN = "open"
+    DISABLED = "disabled"
+
+
+class BilibiliAccountConfig(BaseModel):
+    sessdata: str = ""
+    bili_jct: str = ""
+    dedeuserid: str = ""
+    buvid3: str | None = None
+    buvid4: str | None = None
+    bili_ticket: str | None = None
+    enabled: bool = True
+    room_ids: list[int] = []
+    dm_enabled: bool = False
+    dm_policy: DMPolicy = DMPolicy.PAIRING
+    dm_allow_from: list[str] = []
+    comment_enabled: bool = False
+    comment_oids: list[int] = []
+    danmaku_reply_enabled: bool = False
+    streaming_mode: str = "block"
+    rate_limit_multiplier: float = 1.5
+    require_mention: bool = True
+    bot_keywords: list[str] = []
+
+
+class BilibiliConfig(BaseModel):
+    enabled: bool = True
+    name: str | None = None
+    default_account: str = "default"
+    accounts: dict[str, BilibiliAccountConfig] = {}
+    anti_risk_level: str = "moderate"
+
+
+def is_account_configured(account: BilibiliAccountConfig) -> bool:
+    return bool(account.sessdata and account.bili_jct and account.dedeuserid)
--- a/backend/package/yuxi/channel/extensions/bilibili/wbi.py
+++ b/backend/package/yuxi/channel/extensions/bilibili/wbi.py
@ -0,0 +1,26 @@
+from __future__ import annotations
+
+import hashlib
+import time
+import urllib.parse
+
+_MIXIN_KEY_ENC_TAB = [
+    46, 47, 18, 2, 53, 8, 23, 32, 15, 50, 10, 31, 58, 3, 45, 35,
+    27, 43, 5, 49, 33, 9, 42, 19, 29, 28, 14, 39, 12, 38, 41, 13,
+    37, 48, 7, 16, 24, 55, 40, 61, 26, 17, 0, 1, 60, 51, 30, 4,
+    22, 25, 54, 21, 56, 59, 6, 63, 57, 62, 11, 36, 20, 34, 44, 52,
+]
+
+
+def get_mixin_key(raw_key: str) -> str:
+    return "".join(raw_key[i] for i in _MIXIN_KEY_ENC_TAB if i < len(raw_key))
+
+
+def sign_params(params: dict, img_key: str, sub_key: str) -> dict:
+    mixin_key = get_mixin_key(img_key + sub_key)[:32]
+    params["wts"] = int(time.time())
+    params = dict(sorted(params.items()))
+    query = urllib.parse.urlencode(params)
+    w_rid = hashlib.md5((query + mixin_key).encode()).hexdigest()
+    params["w_rid"] = w_rid
+    return params