128 lines
3.7 KiB
Python
128 lines
3.7 KiB
Python
"""Preview the built admin UI against a fake local backend.
|
|
|
|
This script does not connect to Bilibili or BetterGI. It is only for checking
|
|
the admin page in a local browser while developing the Vue UI.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT / "app"))
|
|
|
|
from danmu_queue import Config, ServiceRegistry, WebServer # noqa: E402
|
|
|
|
|
|
class FakeRunner:
|
|
def is_bgi_running(self):
|
|
return False
|
|
|
|
async def kill_bgi(self):
|
|
return None
|
|
|
|
|
|
class FakeQueueManager:
|
|
def __init__(self):
|
|
self.state = {
|
|
"queue": [10001, 10002],
|
|
"current_admin_uid": 10001,
|
|
"current_group": "薄荷",
|
|
"group_start_time": None,
|
|
"admin_window_end": None,
|
|
"default_running": False,
|
|
"login_status": "confirming",
|
|
}
|
|
|
|
def _save(self):
|
|
return None
|
|
|
|
def leave_queue(self, uid):
|
|
if uid in self.state["queue"]:
|
|
self.state["queue"].remove(uid)
|
|
return {"success": True, "was_running": False}
|
|
|
|
|
|
class FakeUserManager:
|
|
def __init__(self):
|
|
self.users = {
|
|
"10001": {"uname": "测试用户A", "points": 18, "last_signin_date": ""},
|
|
"10002": {"uname": "测试用户B", "points": 7, "last_signin_date": ""},
|
|
}
|
|
|
|
async def _save(self):
|
|
return None
|
|
|
|
async def add_points(self, uid, points):
|
|
user = self.users.setdefault(str(uid), {"uname": f"用户{uid}", "points": 0})
|
|
user["points"] += int(points)
|
|
return user["points"]
|
|
|
|
|
|
class FakeSongRequestManager:
|
|
def __init__(self):
|
|
self.state = {
|
|
"queue": [
|
|
{"id": "1", "name": "测试歌曲", "artist": "测试歌手", "uname": "测试用户A"}
|
|
]
|
|
}
|
|
|
|
def remove_request(self, index=None, song_id=""):
|
|
queue = self.state.setdefault("queue", [])
|
|
if queue:
|
|
return {"success": True, "removed": queue.pop(0)}
|
|
return {"success": False, "msg": "empty"}
|
|
|
|
def clear_requests(self):
|
|
count = len(self.state.get("queue", []))
|
|
self.state["queue"] = []
|
|
return count
|
|
|
|
|
|
class FakeLogMonitor:
|
|
def set_current_group(self, group):
|
|
return None
|
|
|
|
|
|
class FakeSystem:
|
|
def __init__(self):
|
|
self.runner = FakeRunner()
|
|
self.queue_mgr = FakeQueueManager()
|
|
self.user_mgr = FakeUserManager()
|
|
self.song_request_mgr = FakeSongRequestManager()
|
|
self.log_monitor = FakeLogMonitor()
|
|
self.handler = type("FakeHandler", (), {"recent_danmu": [
|
|
{"uname": "测试用户A", "text": "排队"},
|
|
{"uname": "测试用户B", "text": "点歌 测试歌曲"},
|
|
]})()
|
|
self.broadcaster = None
|
|
self.health = ServiceRegistry()
|
|
self.health.set("主程序", ServiceRegistry.RUNNING, "preview")
|
|
self.health.set("Web后台服务", ServiceRegistry.RUNNING, "preview")
|
|
self.health.set("直播监听", ServiceRegistry.RECONNECTING, "preview reconnect")
|
|
|
|
def apply_config(self):
|
|
self.health.set("配置", ServiceRegistry.RUNNING, "preview save")
|
|
|
|
async def _start_default_group(self):
|
|
return None
|
|
|
|
|
|
async def main():
|
|
cfg = Config(str(ROOT / "config" / "config_queue.json"))
|
|
logger = logging.getLogger("preview_admin")
|
|
logger.setLevel(logging.INFO)
|
|
logger.addHandler(logging.StreamHandler(sys.stdout))
|
|
server = WebServer(FakeSystem(), cfg, logger, port=5190, host="127.0.0.1")
|
|
server_task = asyncio.create_task(server.start())
|
|
await asyncio.sleep(0.2)
|
|
print("Preview: http://127.0.0.1:5190/admin")
|
|
await server_task
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|