feat: auto-route room_id 1 to local fake_room.py for testing

- When room_id == 1, server/main automatically patch blivedm to connect
  to fake_room.py on 127.0.0.1:8081 (bot web UI port + 1)
- Any other room id restores real Bilibili (unpatch)
- fake_room.py (gitignored, not packaged): headless fake live room with
  bilibili-compatible init APIs (/x/web-interface/nav, /room/v1/Room/get_info,
  /xlive/web-room/v1/index/getDanmuInfo) and WS /sub protocol (AUTH/HEARTBEAT/
  SEND_MSG_REPLY), plus stdin/HTTP /emit control for danmaku/gift/blind box/
  guard/super chat/enter/follow
- Verified end-to-end: all events flow through blivedm and render rules
  (gift -> 感谢由Baka投喂的小电视, blind box, guard, super chat, custom
  danmaku template, enter room)
This commit is contained in:
2026-08-08 17:25:02 +08:00
parent 22ee613352
commit d7e940fe2e
3 changed files with 41 additions and 0 deletions
+3
View File
@@ -29,3 +29,6 @@ temp/
# Node.js # Node.js
node_modules/ node_modules/
# Dev/testing (not packaged, not committed)
fake_room.py
+13
View File
@@ -33,6 +33,8 @@ def parse_args() -> argparse.Namespace:
parser.add_argument("--no-numbers", action="store_true") parser.add_argument("--no-numbers", action="store_true")
parser.add_argument("--list-devices", action="store_true") parser.add_argument("--list-devices", action="store_true")
parser.add_argument("--debug", "-d", action="store_true") parser.add_argument("--debug", "-d", action="store_true")
parser.add_argument("--fake-room", type=int, default=0, metavar="PORT",
help="connect to local fake_room.py instead of real Bilibili (dev/testing only)")
return parser.parse_args() return parser.parse_args()
@@ -58,6 +60,17 @@ async def tts_worker(queue: asyncio.Queue, bridge: TTSBridge,
async def main_async(args: argparse.Namespace) -> int: async def main_async(args: argparse.Namespace) -> int:
if args.fake_room:
import importlib
fake = importlib.import_module("fake_room")
fake.patch_blivedm(args.fake_room)
logger.info("[dev] connected to fake room on port %d", args.fake_room)
elif args.room_id == 1:
import importlib
fake = importlib.import_module("fake_room")
fake.patch_blivedm(8081)
logger.info("[dev] room 1 -> fake room on port 8081")
if args.list_devices: if args.list_devices:
devices = get_output_devices() devices = get_output_devices()
print("Available audio output devices:") print("Available audio output devices:")
+25
View File
@@ -151,6 +151,21 @@ async def index_handler(request: web.Request) -> web.Response:
return web.FileResponse(STATIC_DIR / "index.html") return web.FileResponse(STATIC_DIR / "index.html")
FAKE_ROOM_PORT = 8081 # fake_room.py default = bot web UI port (8080) + 1
def _apply_fake_room(room_id: int) -> None:
"""room_id == 1 routes to the local fake room (dev/testing); otherwise restore real Bilibili."""
import importlib
fake = importlib.import_module("fake_room")
if room_id == 1:
fake.patch_blivedm(FAKE_ROOM_PORT)
logger.info("[dev] room 1 -> fake room on port %d", FAKE_ROOM_PORT)
else:
fake.unpatch_blivedm()
async def api_start(request: web.Request) -> web.Response: async def api_start(request: web.Request) -> web.Response:
try: try:
config = await request.json() config = await request.json()
@@ -173,6 +188,7 @@ async def api_start(request: web.Request) -> web.Response:
await tts_service.stop() await tts_service.stop()
except Exception: except Exception:
pass pass
_apply_fake_room(room_id)
tts_service.set_cookies(login_session.cookies if login_session.is_logged_in else None) tts_service.set_cookies(login_session.cookies if login_session.is_logged_in else None)
try: try:
await tts_service.start(config) await tts_service.start(config)
@@ -300,6 +316,8 @@ def parse_args() -> argparse.Namespace:
parser.add_argument("--port", "-p", type=int, default=8080) parser.add_argument("--port", "-p", type=int, default=8080)
parser.add_argument("--host", type=str, default="127.0.0.1") parser.add_argument("--host", type=str, default="127.0.0.1")
parser.add_argument("--debug", "-d", action="store_true") parser.add_argument("--debug", "-d", action="store_true")
parser.add_argument("--fake-room", type=int, default=0, metavar="PORT",
help="connect to local fake_room.py instead of real Bilibili (dev/testing only)")
return parser.parse_args() return parser.parse_args()
@@ -311,6 +329,13 @@ def main() -> int:
datefmt="%H:%M:%S", datefmt="%H:%M:%S",
) )
logging.getLogger("aiohttp.access").setLevel(logging.WARNING) logging.getLogger("aiohttp.access").setLevel(logging.WARNING)
if args.fake_room:
import importlib
fake = importlib.import_module("fake_room")
fake.patch_blivedm(args.fake_room)
print(f" [dev] connected to fake room on port {args.fake_room}")
app = create_app() app = create_app()
print(f"\n Bilibili Live → Yukkuri TTS Web UI") print(f"\n Bilibili Live → Yukkuri TTS Web UI")
print(f" Open: http://{args.host}:{args.port}\n") print(f" Open: http://{args.host}:{args.port}\n")