chun_qiu
fd5c732d8a
- Add web UI panel (dark theme) at / with REST API - Add Bilibili QR code scan login with browser header emulation - Add wanakana for English -> katakana conversion - Add volume control (0%-200%) - Add message format with '说、' pause marker - PyInstaller --onedir build for portable exe - Rate-limit danmaku logging (1/20) - Fix aiohttp.access log spam - Add README.md
227 lines
8.1 KiB
Python
227 lines
8.1 KiB
Python
"""Bilibili QR Code Login with full browser header emulation.
|
|
|
|
Mimics a modern Chrome browser to perform QR code scan login for Bilibili.
|
|
Returns cookies (SESSDATA, bili_jct, DedeUserID, etc.) that can be injected
|
|
into the danmaku client's HTTP session.
|
|
"""
|
|
|
|
import time
|
|
import logging
|
|
from typing import Optional
|
|
|
|
import aiohttp
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
BROWSER_HEADERS = {
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/131.0.0.0 Safari/537.36"
|
|
),
|
|
"Accept": "application/json, text/plain, */*",
|
|
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8,ja;q=0.7",
|
|
"Accept-Encoding": "gzip, deflate, br",
|
|
"sec-ch-ua": '"Google Chrome";v="131", "Chromium";v="131", "Not_A Brand";v="24"',
|
|
"sec-ch-ua-mobile": "?0",
|
|
"sec-ch-ua-platform": '"Windows"',
|
|
"Sec-Fetch-Dest": "empty",
|
|
"Sec-Fetch-Mode": "cors",
|
|
"Sec-Fetch-Site": "same-site",
|
|
}
|
|
|
|
_QR_GENERATE_URL = "https://passport.bilibili.com/x/passport-login/web/qrcode/generate"
|
|
_QR_POLL_URL = "https://passport.bilibili.com/x/passport-login/web/qrcode/poll"
|
|
_NAV_URL = "https://api.bilibili.com/x/web-interface/nav"
|
|
|
|
STATUS_NOT_SCANNED = 86101
|
|
STATUS_SCANNED_WAITING = 86090
|
|
STATUS_SUCCESS = 0
|
|
STATUS_EXPIRED = 86038
|
|
|
|
|
|
class BiliLoginSession:
|
|
"""Manages Bilibili QR code login with browser-like HTTP session."""
|
|
|
|
def __init__(self):
|
|
self._cookies: dict[str, str] = {}
|
|
self._qrcode_key: Optional[str] = None
|
|
self._is_logged_in = False
|
|
self._username: str = ""
|
|
self._uid: int = 0
|
|
|
|
@property
|
|
def is_logged_in(self) -> bool:
|
|
return self._is_logged_in
|
|
|
|
@property
|
|
def username(self) -> str:
|
|
return self._username
|
|
|
|
@property
|
|
def uid(self) -> int:
|
|
return self._uid
|
|
|
|
@property
|
|
def cookies(self) -> dict[str, str]:
|
|
return dict(self._cookies)
|
|
|
|
async def _create_session(self) -> aiohttp.ClientSession:
|
|
import http.cookies
|
|
h = dict(BROWSER_HEADERS)
|
|
h["Referer"] = "https://www.bilibili.com/"
|
|
h["Origin"] = "https://www.bilibili.com"
|
|
session = aiohttp.ClientSession(headers=h)
|
|
if self._cookies:
|
|
simple = http.cookies.SimpleCookie()
|
|
for name, value in self._cookies.items():
|
|
simple[name] = value
|
|
simple[name]["domain"] = ".bilibili.com"
|
|
session.cookie_jar.update_cookies(simple)
|
|
return session
|
|
|
|
async def generate_qrcode(self) -> dict:
|
|
session = await self._create_session()
|
|
try:
|
|
async with session.get(_QR_GENERATE_URL) as resp:
|
|
data = await resp.json()
|
|
if data.get("code") != 0:
|
|
raise RuntimeError(f"QR generate failed: {data}")
|
|
inner = data["data"]
|
|
self._qrcode_key = inner["qrcode_key"]
|
|
logger.info("QR code generated, key=%s", self._qrcode_key)
|
|
return {
|
|
"url": inner["url"],
|
|
"qrcode_key": inner["qrcode_key"],
|
|
}
|
|
finally:
|
|
await session.close()
|
|
|
|
async def poll_login(self) -> dict:
|
|
if not self._qrcode_key:
|
|
raise RuntimeError("No QR code generated. Call generate_qrcode() first.")
|
|
|
|
session = await self._create_session()
|
|
try:
|
|
params = {"qrcode_key": self._qrcode_key}
|
|
async with session.get(_QR_POLL_URL, params=params) as resp:
|
|
data = await resp.json()
|
|
if data.get("code") != 0:
|
|
logger.warning("QR poll API error: %s", data)
|
|
return {"status": -1, "message": f"API error: {data.get('message', 'unknown')}"}
|
|
|
|
inner = data["data"]
|
|
sc = inner["code"]
|
|
|
|
if sc == STATUS_SUCCESS:
|
|
self._is_logged_in = True
|
|
self._extract_cookies(session)
|
|
await self._fetch_user_info()
|
|
logger.info("QR login success, user=%s", self._username)
|
|
return {
|
|
"status": STATUS_SUCCESS,
|
|
"message": f"登录成功: {self._username}",
|
|
"cookies": dict(self._cookies),
|
|
"username": self._username,
|
|
"uid": self._uid,
|
|
}
|
|
if sc == STATUS_NOT_SCANNED:
|
|
return {"status": STATUS_NOT_SCANNED, "message": "请使用Bilibili客户端扫码"}
|
|
if sc == STATUS_SCANNED_WAITING:
|
|
return {"status": STATUS_SCANNED_WAITING, "message": "已扫码，请在手机上确认登录"}
|
|
if sc == STATUS_EXPIRED:
|
|
self._qrcode_key = None
|
|
return {"status": STATUS_EXPIRED, "message": "二维码已过期，请重新获取"}
|
|
|
|
return {"status": sc, "message": inner.get("message", f"未知状态: {sc}")}
|
|
finally:
|
|
await session.close()
|
|
|
|
def _extract_cookies(self, session: aiohttp.ClientSession) -> None:
|
|
jar = session.cookie_jar
|
|
for cookie in jar:
|
|
if cookie.key in ("SESSDATA", "bili_jct", "DedeUserID", "DedeUserID__ckMd5",
|
|
"sid", "buvid3", "buvid4", "b_nut"):
|
|
self._cookies[cookie.key] = cookie.value
|
|
logger.debug("Extracted cookies: %s", list(self._cookies.keys()))
|
|
|
|
async def _fetch_user_info(self) -> None:
|
|
session = await self._create_session()
|
|
try:
|
|
async with session.get(_NAV_URL) as resp:
|
|
data = await resp.json()
|
|
if data.get("code") == 0:
|
|
inner = data["data"]
|
|
self._username = inner.get("uname", "")
|
|
self._uid = inner.get("mid", 0)
|
|
logger.info("User info: %s (uid=%d)", self._username, self._uid)
|
|
else:
|
|
logger.warning("Failed to fetch user info: %s", data)
|
|
finally:
|
|
await session.close()
|
|
|
|
def clear(self) -> None:
|
|
self._cookies = {}
|
|
self._qrcode_key = None
|
|
self._is_logged_in = False
|
|
self._username = ""
|
|
self._uid = 0
|
|
|
|
async def save_cookies(self, path: str = "cookies.json") -> None:
|
|
import json
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump({
|
|
"cookies": self._cookies,
|
|
"username": self._username,
|
|
"uid": self._uid,
|
|
}, f, ensure_ascii=False, indent=2)
|
|
logger.info("Cookies saved to %s", path)
|
|
|
|
async def try_auto_login(self, path: str = "cookies.json") -> bool:
|
|
import json, os
|
|
if not os.path.exists(path):
|
|
return False
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
cookies = data.get("cookies", {})
|
|
if not cookies.get("SESSDATA"):
|
|
return False
|
|
result = await self.check_login(cookies)
|
|
if result["valid"]:
|
|
logger.info("Auto-login success: %s", self._username)
|
|
return True
|
|
else:
|
|
os.remove(path)
|
|
logger.info("Cached cookies expired, removed")
|
|
return False
|
|
except Exception as e:
|
|
logger.warning("Failed to load cached cookies: %s", e)
|
|
return False
|
|
|
|
|
|
async def build_danmaku_session(cookies: Optional[dict[str, str]] = None) -> aiohttp.ClientSession:
|
|
"""Build a browser-emulating aiohttp session for danmaku connections.
|
|
|
|
Args:
|
|
cookies: Optional dict of cookies to inject.
|
|
|
|
Returns:
|
|
aiohttp.ClientSession with browser headers and cookies configured.
|
|
"""
|
|
import http.cookies
|
|
|
|
h = dict(BROWSER_HEADERS)
|
|
h["Referer"] = "https://live.bilibili.com/"
|
|
h["Origin"] = "https://live.bilibili.com"
|
|
|
|
session = aiohttp.ClientSession(headers=h)
|
|
if cookies:
|
|
simple = http.cookies.SimpleCookie()
|
|
for key, value in cookies.items():
|
|
simple[key] = value
|
|
simple[key]["domain"] = ".bilibili.com"
|
|
session.cookie_jar.update_cookies(simple)
|
|
|
|
return session
|