diff --git a/backend/package/yuxi/channel/extensions/bilibili/__init__.py b/backend/package/yuxi/channel/extensions/bilibili/__init__.py new file mode 100644 index 00000000..e6f11f68 --- /dev/null +++ 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) diff --git a/backend/package/yuxi/channel/extensions/bilibili/anti_risk.py b/backend/package/yuxi/channel/extensions/bilibili/anti_risk.py new file mode 100644 index 00000000..d94727f2 --- /dev/null +++ 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()) diff --git a/backend/package/yuxi/channel/extensions/bilibili/client.py b/backend/package/yuxi/channel/extensions/bilibili/client.py new file mode 100644 index 00000000..e7ebfb5e --- /dev/null +++ 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 可能不可用") diff --git a/backend/package/yuxi/channel/extensions/bilibili/config.py b/backend/package/yuxi/channel/extensions/bilibili/config.py new file mode 100644 index 00000000..8fe70edf --- /dev/null +++ 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() diff --git a/backend/package/yuxi/channel/extensions/bilibili/danmaku/__init__.py b/backend/package/yuxi/channel/extensions/bilibili/danmaku/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/package/yuxi/channel/extensions/bilibili/danmaku/handler.py b/backend/package/yuxi/channel/extensions/bilibili/danmaku/handler.py new file mode 100644 index 00000000..7f048038 --- /dev/null +++ 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 diff --git a/backend/package/yuxi/channel/extensions/bilibili/danmaku/monitor.py b/backend/package/yuxi/channel/extensions/bilibili/danmaku/monitor.py new file mode 100644 index 00000000..00b39d77 --- /dev/null +++ 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) diff --git a/backend/package/yuxi/channel/extensions/bilibili/danmaku/types.py b/backend/package/yuxi/channel/extensions/bilibili/danmaku/types.py new file mode 100644 index 00000000..cc69a933 --- /dev/null +++ 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 diff --git a/backend/package/yuxi/channel/extensions/bilibili/gateway.py b/backend/package/yuxi/channel/extensions/bilibili/gateway.py new file mode 100644 index 00000000..c70337b3 --- /dev/null +++ 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 diff --git a/backend/package/yuxi/channel/extensions/bilibili/outbound.py b/backend/package/yuxi/channel/extensions/bilibili/outbound.py new file mode 100644 index 00000000..145a2843 --- /dev/null +++ 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 diff --git a/backend/package/yuxi/channel/extensions/bilibili/pairing.py b/backend/package/yuxi/channel/extensions/bilibili/pairing.py new file mode 100644 index 00000000..8e6ed5ee --- /dev/null +++ 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() diff --git a/backend/package/yuxi/channel/extensions/bilibili/plugin.json b/backend/package/yuxi/channel/extensions/bilibili/plugin.json new file mode 100644 index 00000000..ae56b4b4 --- /dev/null +++ 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" +} \ No newline at end of file diff --git a/backend/package/yuxi/channel/extensions/bilibili/private_msg/__init__.py b/backend/package/yuxi/channel/extensions/bilibili/private_msg/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/package/yuxi/channel/extensions/bilibili/private_msg/poller.py b/backend/package/yuxi/channel/extensions/bilibili/private_msg/poller.py new file mode 100644 index 00000000..19656b66 --- /dev/null +++ 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 diff --git a/backend/package/yuxi/channel/extensions/bilibili/probe.py b/backend/package/yuxi/channel/extensions/bilibili/probe.py new file mode 100644 index 00000000..20430799 --- /dev/null +++ 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), + } diff --git a/backend/package/yuxi/channel/extensions/bilibili/security.py b/backend/package/yuxi/channel/extensions/bilibili/security.py new file mode 100644 index 00000000..a7a31d20 --- /dev/null +++ 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 diff --git a/backend/package/yuxi/channel/extensions/bilibili/status.py b/backend/package/yuxi/channel/extensions/bilibili/status.py new file mode 100644 index 00000000..1039eef0 --- /dev/null +++ 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() diff --git a/backend/package/yuxi/channel/extensions/bilibili/streaming.py b/backend/package/yuxi/channel/extensions/bilibili/streaming.py new file mode 100644 index 00000000..4ff7606f --- /dev/null +++ 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 diff --git a/backend/package/yuxi/channel/extensions/bilibili/types.py b/backend/package/yuxi/channel/extensions/bilibili/types.py new file mode 100644 index 00000000..4d8a7b31 --- /dev/null +++ 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) diff --git a/backend/package/yuxi/channel/extensions/bilibili/wbi.py b/backend/package/yuxi/channel/extensions/bilibili/wbi.py new file mode 100644 index 00000000..d9fb9e1c --- /dev/null +++ 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