602 lines
24 KiB
Python
602 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import base64
|
|
import ctypes
|
|
import ctypes.wintypes
|
|
import hashlib
|
|
import http.cookiejar
|
|
import http.cookies
|
|
import io
|
|
import json
|
|
import logging
|
|
import os
|
|
import secrets
|
|
import time
|
|
import urllib.parse
|
|
import urllib.error
|
|
import urllib.request
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
|
|
|
|
_PUBLIC_KEY_PEM = """-----BEGIN PUBLIC KEY-----
|
|
MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDLgd2OAkcGVtoE3ThUREbio0Eg
|
|
Uc/prcajMKXvkCKFCWhJYJcLkcM2DKKcSeFpD/j6Boy538YXnR6VhcuUJOhH2x71
|
|
nzPjfdTcqMz7djHum0qSZA0AyCBDABUqCrfNgCiJ00Ra7GmRj+YCK1NJEuewlb40
|
|
JNrRuoEUXpabUzGB8QIDAQAB
|
|
-----END PUBLIC KEY-----"""
|
|
_USER_AGENT = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
|
|
_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"
|
|
_QR_HEADERS = {
|
|
"Referer": "https://www.bilibili.com/",
|
|
"Origin": "https://www.bilibili.com",
|
|
}
|
|
|
|
|
|
class _DataBlob(ctypes.Structure):
|
|
_fields_ = [("cbData", ctypes.wintypes.DWORD), ("pbData", ctypes.POINTER(ctypes.c_byte))]
|
|
|
|
|
|
def _blob(data: bytes) -> tuple[_DataBlob, Any]:
|
|
buffer = ctypes.create_string_buffer(data)
|
|
return _DataBlob(len(data), ctypes.cast(buffer, ctypes.POINTER(ctypes.c_byte))), buffer
|
|
|
|
|
|
def _dpapi_encrypt(value: str) -> str:
|
|
if os.name != "nt":
|
|
raise RuntimeError("B站刷新令牌安全存储仅支持 Windows DPAPI")
|
|
source, source_buffer = _blob(value.encode("utf-8"))
|
|
entropy, entropy_buffer = _blob(b"Live-streaming:bilibili-refresh-token:v1")
|
|
output = _DataBlob()
|
|
ok = ctypes.windll.crypt32.CryptProtectData(
|
|
ctypes.byref(source), None, ctypes.byref(entropy), None, None, 0,
|
|
ctypes.byref(output),
|
|
)
|
|
_ = source_buffer, entropy_buffer
|
|
if not ok:
|
|
raise ctypes.WinError()
|
|
try:
|
|
encrypted = ctypes.string_at(output.pbData, output.cbData)
|
|
return base64.b64encode(encrypted).decode("ascii")
|
|
finally:
|
|
ctypes.windll.kernel32.LocalFree(output.pbData)
|
|
|
|
|
|
def _dpapi_decrypt(value: str) -> str:
|
|
if os.name != "nt":
|
|
raise RuntimeError("B站刷新令牌安全存储仅支持 Windows DPAPI")
|
|
source, source_buffer = _blob(base64.b64decode(value))
|
|
entropy, entropy_buffer = _blob(b"Live-streaming:bilibili-refresh-token:v1")
|
|
output = _DataBlob()
|
|
ok = ctypes.windll.crypt32.CryptUnprotectData(
|
|
ctypes.byref(source), None, ctypes.byref(entropy), None, None, 0,
|
|
ctypes.byref(output),
|
|
)
|
|
_ = source_buffer, entropy_buffer
|
|
if not ok:
|
|
raise ctypes.WinError()
|
|
try:
|
|
return ctypes.string_at(output.pbData, output.cbData).decode("utf-8")
|
|
finally:
|
|
ctypes.windll.kernel32.LocalFree(output.pbData)
|
|
|
|
|
|
class BilibiliCredentialStore:
|
|
def __init__(self, path: str | Path):
|
|
self.path = Path(path)
|
|
|
|
def save_refresh_token(self, refresh_token: str) -> None:
|
|
token = str(refresh_token or "").strip()
|
|
if not token:
|
|
raise ValueError("refresh_token 不能为空")
|
|
payload = {
|
|
"version": 1,
|
|
"provider": "windows_dpapi_current_user",
|
|
"refresh_token_protected": _dpapi_encrypt(token),
|
|
}
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = self.path.with_suffix(self.path.suffix + ".tmp")
|
|
with open(tmp, "w", encoding="utf-8", newline="\n") as handle:
|
|
json.dump(payload, handle, ensure_ascii=False, indent=2)
|
|
handle.write("\n")
|
|
tmp.replace(self.path)
|
|
|
|
def load_refresh_token(self) -> str:
|
|
if not self.path.exists():
|
|
return ""
|
|
with open(self.path, "r", encoding="utf-8") as handle:
|
|
payload = json.load(handle)
|
|
protected = str(payload.get("refresh_token_protected") or "")
|
|
return _dpapi_decrypt(protected) if protected else ""
|
|
|
|
def is_configured(self) -> bool:
|
|
if not self.path.exists():
|
|
return False
|
|
try:
|
|
with open(self.path, "r", encoding="utf-8") as handle:
|
|
payload = json.load(handle)
|
|
return bool(str(payload.get("refresh_token_protected") or ""))
|
|
except (OSError, ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
def _read_der_length(data: bytes, offset: int) -> tuple[int, int]:
|
|
first = data[offset]
|
|
offset += 1
|
|
if first < 0x80:
|
|
return first, offset
|
|
count = first & 0x7F
|
|
return int.from_bytes(data[offset:offset + count], "big"), offset + count
|
|
|
|
|
|
def _read_der_tlv(data: bytes, offset: int, expected_tag: int | None = None) -> tuple[int, bytes, int]:
|
|
tag = data[offset]
|
|
if expected_tag is not None and tag != expected_tag:
|
|
raise ValueError(f"DER tag 不匹配: expected={expected_tag:#x}, actual={tag:#x}")
|
|
length, content_offset = _read_der_length(data, offset + 1)
|
|
end = content_offset + length
|
|
return tag, data[content_offset:end], end
|
|
|
|
|
|
def _public_numbers() -> tuple[int, int]:
|
|
body = "".join(line for line in _PUBLIC_KEY_PEM.splitlines() if not line.startswith("-----"))
|
|
der = base64.b64decode(body)
|
|
_, spki, _ = _read_der_tlv(der, 0, 0x30)
|
|
_, _, offset = _read_der_tlv(spki, 0, 0x30)
|
|
_, bit_string, _ = _read_der_tlv(spki, offset, 0x03)
|
|
_, rsa_key, _ = _read_der_tlv(bit_string[1:], 0, 0x30)
|
|
_, modulus_bytes, rsa_offset = _read_der_tlv(rsa_key, 0, 0x02)
|
|
_, exponent_bytes, _ = _read_der_tlv(rsa_key, rsa_offset, 0x02)
|
|
return int.from_bytes(modulus_bytes, "big"), int.from_bytes(exponent_bytes, "big")
|
|
|
|
|
|
def _mgf1(seed: bytes, length: int) -> bytes:
|
|
result = bytearray()
|
|
counter = 0
|
|
while len(result) < length:
|
|
result.extend(hashlib.sha256(seed + counter.to_bytes(4, "big")).digest())
|
|
counter += 1
|
|
return bytes(result[:length])
|
|
|
|
|
|
def _rsa_oaep_sha256_encrypt(message: bytes) -> str:
|
|
modulus, exponent = _public_numbers()
|
|
key_size = (modulus.bit_length() + 7) // 8
|
|
digest_size = hashlib.sha256().digest_size
|
|
if len(message) > key_size - 2 * digest_size - 2:
|
|
raise ValueError("待加密内容过长")
|
|
label_hash = hashlib.sha256(b"").digest()
|
|
padding = b"\x00" * (key_size - len(message) - 2 * digest_size - 2)
|
|
data_block = label_hash + padding + b"\x01" + message
|
|
seed = secrets.token_bytes(digest_size)
|
|
data_mask = _mgf1(seed, key_size - digest_size - 1)
|
|
masked_data = bytes(left ^ right for left, right in zip(data_block, data_mask))
|
|
seed_mask = _mgf1(masked_data, digest_size)
|
|
masked_seed = bytes(left ^ right for left, right in zip(seed, seed_mask))
|
|
encoded = b"\x00" + masked_seed + masked_data
|
|
encrypted = pow(int.from_bytes(encoded, "big"), exponent, modulus)
|
|
return encrypted.to_bytes(key_size, "big").hex()
|
|
|
|
|
|
def _parse_cookie(cookie_text: str) -> dict[str, str]:
|
|
parsed = http.cookies.SimpleCookie()
|
|
parsed.load(str(cookie_text or "").replace("; ", ";"))
|
|
return {name: morsel.value for name, morsel in parsed.items()}
|
|
|
|
|
|
def _cookie_header(values: dict[str, str]) -> str:
|
|
return "; ".join(f"{name}={value}" for name, value in values.items() if value)
|
|
|
|
|
|
def _request_json(url: str, *, cookie: str = "", data: dict[str, str] | None = None,
|
|
opener: urllib.request.OpenerDirector | None = None,
|
|
headers: dict[str, str] | None = None,
|
|
retries: int = 0) -> tuple[dict, Any]:
|
|
body = urllib.parse.urlencode(data).encode("utf-8") if data is not None else None
|
|
request = urllib.request.Request(url, data=body, method="POST" if body is not None else "GET")
|
|
request.add_header("User-Agent", _USER_AGENT)
|
|
for name, value in (headers or {}).items():
|
|
request.add_header(name, value)
|
|
if cookie:
|
|
request.add_header("Cookie", cookie)
|
|
if body is not None:
|
|
request.add_header("Content-Type", "application/x-www-form-urlencoded")
|
|
retry_count = max(0, int(retries))
|
|
for attempt in range(retry_count + 1):
|
|
try:
|
|
response = (opener or urllib.request.build_opener()).open(request, timeout=15)
|
|
return json.loads(response.read().decode("utf-8")), response
|
|
except urllib.error.HTTPError:
|
|
raise
|
|
except (urllib.error.URLError, ConnectionError, TimeoutError, OSError) as exc:
|
|
if attempt >= retry_count:
|
|
raise RuntimeError("连接B站登录服务失败，请稍后重试") from exc
|
|
time.sleep(0.4 * (attempt + 1))
|
|
|
|
raise AssertionError("unreachable")
|
|
|
|
|
|
def _seed_cookie_jar(jar: http.cookiejar.CookieJar, values: dict[str, str]) -> None:
|
|
for name, value in values.items():
|
|
jar.set_cookie(http.cookiejar.Cookie(
|
|
version=0, name=name, value=value, port=None, port_specified=False,
|
|
domain=".bilibili.com", domain_specified=True, domain_initial_dot=True,
|
|
path="/", path_specified=True, secure=False, expires=None, discard=True,
|
|
comment=None, comment_url=None, rest={}, rfc2109=False,
|
|
))
|
|
|
|
|
|
def _jar_values(jar: http.cookiejar.CookieJar) -> dict[str, str]:
|
|
return {cookie.name: cookie.value for cookie in jar}
|
|
|
|
|
|
def _login_url_cookie_values(url: str) -> dict[str, str]:
|
|
values: dict[str, str] = {}
|
|
query = urllib.parse.urlparse(str(url or "")).query
|
|
for item in query.split("&"):
|
|
raw_name, separator, raw_value = item.partition("=")
|
|
if not separator:
|
|
continue
|
|
name = urllib.parse.unquote_plus(raw_name)
|
|
if name in {"SESSDATA", "bili_jct", "DedeUserID", "DedeUserID__ckMd5", "sid", "buvid3"}:
|
|
values[name] = raw_value
|
|
return values
|
|
|
|
|
|
def _render_qr_png(content: str) -> bytes:
|
|
try:
|
|
import qrcode
|
|
from qrcode.constants import ERROR_CORRECT_M
|
|
except ImportError as exc:
|
|
raise RuntimeError("缺少 qrcode 依赖，请重新安装 requirements.txt") from exc
|
|
|
|
qr = qrcode.QRCode(
|
|
version=None,
|
|
error_correction=ERROR_CORRECT_M,
|
|
box_size=8,
|
|
border=3,
|
|
)
|
|
qr.add_data(content)
|
|
qr.make(fit=True)
|
|
image = qr.make_image(fill_color="black", back_color="white")
|
|
output = io.BytesIO()
|
|
image.save(output, format="PNG")
|
|
return output.getvalue()
|
|
|
|
|
|
class BilibiliQrLogin:
|
|
"""服务端持有二维码密钥和 CookieJar，前端只获取二维码图片与状态。"""
|
|
|
|
_STATUS_MESSAGES = {
|
|
"idle": "尚未开始扫码登录",
|
|
"awaiting_scan": "请使用哔哩哔哩客户端扫码",
|
|
"awaiting_confirm": "已扫码，请在手机上确认登录",
|
|
"completed": "登录成功，Cookie 与刷新凭据已更新",
|
|
"expired": "二维码已过期，请重新生成",
|
|
"failed": "扫码登录失败",
|
|
}
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
credential_store: BilibiliCredentialStore,
|
|
update_cookie: Callable[[dict[str, str]], None],
|
|
logger: logging.Logger,
|
|
on_logged_in: Callable[[], Any] | None = None,
|
|
ttl_seconds: int = 180,
|
|
):
|
|
self.credential_store = credential_store
|
|
self.update_cookie = update_cookie
|
|
self.logger = logger
|
|
self.on_logged_in = on_logged_in
|
|
self.ttl_seconds = max(60, int(ttl_seconds))
|
|
self._lock = asyncio.Lock()
|
|
self._session: dict[str, Any] | None = None
|
|
|
|
def _snapshot(self) -> dict[str, Any]:
|
|
session = self._session or {}
|
|
state = str(session.get("state") or "idle")
|
|
expires_at = float(session.get("expires_at") or 0)
|
|
expires_in = max(0, int(expires_at - time.time())) if expires_at else 0
|
|
return {
|
|
"success": True,
|
|
"state": state,
|
|
"message": str(session.get("message") or self._STATUS_MESSAGES.get(state, "")),
|
|
"expires_in": expires_in,
|
|
"has_qr_image": state in {"awaiting_scan", "awaiting_confirm"} and expires_in > 0,
|
|
"credential_configured": self.credential_store.is_configured(),
|
|
"account": session.get("account"),
|
|
}
|
|
|
|
def snapshot(self) -> dict[str, Any]:
|
|
return self._snapshot()
|
|
|
|
def _start_sync(self) -> dict[str, Any]:
|
|
jar = http.cookiejar.CookieJar()
|
|
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
|
|
payload, _ = _request_json(
|
|
_QR_GENERATE_URL,
|
|
opener=opener,
|
|
headers=_QR_HEADERS,
|
|
retries=2,
|
|
)
|
|
if payload.get("code") != 0:
|
|
raise RuntimeError(f"B站二维码申请失败 code={payload.get('code')}")
|
|
data = payload.get("data") or {}
|
|
qr_url = str(data.get("url") or "").strip()
|
|
qr_key = str(data.get("qrcode_key") or "").strip()
|
|
if not qr_url or not qr_key:
|
|
raise RuntimeError("B站二维码响应缺少必要字段")
|
|
now = time.time()
|
|
return {
|
|
"state": "awaiting_scan",
|
|
"message": self._STATUS_MESSAGES["awaiting_scan"],
|
|
"created_at": now,
|
|
"expires_at": now + self.ttl_seconds,
|
|
"qr_url": qr_url,
|
|
"qr_key": qr_key,
|
|
"jar": jar,
|
|
"opener": opener,
|
|
"account": None,
|
|
}
|
|
|
|
async def start(self) -> dict[str, Any]:
|
|
async with self._lock:
|
|
try:
|
|
self._session = await asyncio.to_thread(self._start_sync)
|
|
except Exception as exc:
|
|
self._session = {
|
|
"state": "failed",
|
|
"message": str(exc) or type(exc).__name__,
|
|
"expires_at": 0,
|
|
}
|
|
self.logger.warning("[B站扫码登录] 二维码申请失败: %s", type(exc).__name__)
|
|
return self._snapshot()
|
|
self.logger.info("[B站扫码登录] 二维码已生成，等待扫码")
|
|
return self._snapshot()
|
|
|
|
def _poll_sync(self, session: dict[str, Any]) -> dict[str, Any]:
|
|
url = _QR_POLL_URL + "?" + urllib.parse.urlencode({"qrcode_key": session["qr_key"]})
|
|
payload, _ = _request_json(
|
|
url,
|
|
opener=session["opener"],
|
|
headers=_QR_HEADERS,
|
|
retries=2,
|
|
)
|
|
if payload.get("code") != 0:
|
|
return {"state": "failed", "message": f"B站扫码状态查询失败 code={payload.get('code')}"}
|
|
|
|
data = payload.get("data") or {}
|
|
status_code = int(data.get("code") or 0)
|
|
if status_code == 86101:
|
|
return {"state": "awaiting_scan", "message": self._STATUS_MESSAGES["awaiting_scan"]}
|
|
if status_code == 86090:
|
|
return {"state": "awaiting_confirm", "message": self._STATUS_MESSAGES["awaiting_confirm"]}
|
|
if status_code == 86038:
|
|
return {"state": "expired", "message": self._STATUS_MESSAGES["expired"]}
|
|
if status_code != 0:
|
|
return {"state": "failed", "message": str(data.get("message") or f"扫码失败 code={status_code}")}
|
|
|
|
refresh_token = str(data.get("refresh_token") or "").strip()
|
|
if not refresh_token:
|
|
return {"state": "failed", "message": "扫码成功响应缺少 refresh_token"}
|
|
cookie_values = _jar_values(session["jar"])
|
|
for name, value in _login_url_cookie_values(str(data.get("url") or "")).items():
|
|
cookie_values.setdefault(name, value)
|
|
if not cookie_values.get("SESSDATA") or not cookie_values.get("bili_jct"):
|
|
return {"state": "failed", "message": "扫码成功但登录 Cookie 不完整"}
|
|
|
|
cookie = _cookie_header(cookie_values)
|
|
nav, _ = _request_json(
|
|
"https://api.bilibili.com/x/web-interface/nav",
|
|
cookie=cookie,
|
|
headers={"Referer": "https://www.bilibili.com/"},
|
|
retries=2,
|
|
)
|
|
nav_data = nav.get("data") or {}
|
|
if nav.get("code") != 0 or not bool(nav_data.get("isLogin")):
|
|
return {"state": "failed", "message": "扫码 Cookie 登录验证失败"}
|
|
|
|
self.credential_store.save_refresh_token(refresh_token)
|
|
self.update_cookie(cookie_values)
|
|
return {
|
|
"state": "completed",
|
|
"message": self._STATUS_MESSAGES["completed"],
|
|
"account": {
|
|
"mid": str(nav_data.get("mid") or ""),
|
|
"uname": str(nav_data.get("uname") or "B站账号"),
|
|
},
|
|
}
|
|
|
|
async def poll(self) -> dict[str, Any]:
|
|
callback_needed = False
|
|
async with self._lock:
|
|
if not self._session:
|
|
return self._snapshot()
|
|
state = str(self._session.get("state") or "idle")
|
|
if state in {"completed", "expired", "failed"}:
|
|
return self._snapshot()
|
|
if time.time() >= float(self._session.get("expires_at") or 0):
|
|
self._session.update(state="expired", message=self._STATUS_MESSAGES["expired"])
|
|
return self._snapshot()
|
|
try:
|
|
result = await asyncio.to_thread(self._poll_sync, self._session)
|
|
except Exception as exc:
|
|
self.logger.warning("[B站扫码登录] 状态查询异常: %s", type(exc).__name__)
|
|
self._session.update(
|
|
state="failed",
|
|
message=str(exc) or f"扫码状态查询异常: {type(exc).__name__}",
|
|
)
|
|
return self._snapshot()
|
|
previous_state = state
|
|
self._session.update(result)
|
|
callback_needed = previous_state != "completed" and result.get("state") == "completed"
|
|
snapshot = self._snapshot()
|
|
|
|
if callback_needed:
|
|
self.logger.info("[B站扫码登录] 登录成功，Cookie 与刷新凭据已更新")
|
|
if self.on_logged_in:
|
|
callback_result = self.on_logged_in()
|
|
if asyncio.iscoroutine(callback_result):
|
|
await callback_result
|
|
return snapshot
|
|
|
|
async def qr_png(self) -> bytes:
|
|
async with self._lock:
|
|
if not self._session or self._session.get("state") not in {"awaiting_scan", "awaiting_confirm"}:
|
|
raise RuntimeError("当前没有可用的登录二维码")
|
|
if time.time() >= float(self._session.get("expires_at") or 0):
|
|
self._session.update(state="expired", message=self._STATUS_MESSAGES["expired"])
|
|
raise RuntimeError("登录二维码已过期")
|
|
content = str(self._session.get("qr_url") or "")
|
|
return await asyncio.to_thread(_render_qr_png, content)
|
|
|
|
|
|
class BilibiliCookieRefresher:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
credential_store: BilibiliCredentialStore,
|
|
get_cookie: Callable[[], str],
|
|
update_cookie: Callable[[dict[str, str]], None],
|
|
logger: logging.Logger,
|
|
check_interval_seconds: int = 6 * 60 * 60,
|
|
on_refreshed: Callable[[], Any] | None = None,
|
|
is_enabled: Callable[[], bool] | None = None,
|
|
get_check_interval_seconds: Callable[[], int] | None = None,
|
|
):
|
|
self.credential_store = credential_store
|
|
self.get_cookie = get_cookie
|
|
self.update_cookie = update_cookie
|
|
self.logger = logger
|
|
self.check_interval_seconds = max(3600, int(check_interval_seconds))
|
|
self.on_refreshed = on_refreshed
|
|
self.is_enabled = is_enabled or (lambda: True)
|
|
self.get_check_interval_seconds = get_check_interval_seconds
|
|
self._stop = False
|
|
self._wake = asyncio.Event()
|
|
|
|
def stop(self) -> None:
|
|
self._stop = True
|
|
self._wake.set()
|
|
|
|
def wake(self) -> None:
|
|
self._wake.set()
|
|
|
|
def _current_interval(self) -> int:
|
|
if not self.get_check_interval_seconds:
|
|
return self.check_interval_seconds
|
|
try:
|
|
return max(3600, int(self.get_check_interval_seconds()))
|
|
except (TypeError, ValueError):
|
|
return self.check_interval_seconds
|
|
|
|
def _check_and_refresh_sync(self) -> dict[str, Any]:
|
|
refresh_token = self.credential_store.load_refresh_token()
|
|
if not refresh_token:
|
|
return {"status": "disabled", "message": "未配置刷新令牌"}
|
|
current_cookie = self.get_cookie()
|
|
current_values = _parse_cookie(current_cookie)
|
|
csrf = current_values.get("bili_jct", "")
|
|
if not current_values.get("SESSDATA") or not csrf:
|
|
return {"status": "failed", "message": "当前 Cookie 缺少 SESSDATA 或 bili_jct"}
|
|
|
|
info, _ = _request_json(
|
|
"https://passport.bilibili.com/x/passport-login/web/cookie/info?" +
|
|
urllib.parse.urlencode({"csrf": csrf}),
|
|
cookie=current_cookie,
|
|
retries=2,
|
|
)
|
|
if info.get("code") != 0:
|
|
return {"status": "failed", "message": f"登录状态检查失败 code={info.get('code')}"}
|
|
if not bool((info.get("data") or {}).get("refresh")):
|
|
return {"status": "valid", "message": "Cookie 当前无需刷新"}
|
|
|
|
timestamp = str((info.get("data") or {}).get("timestamp") or "")
|
|
correspond_path = _rsa_oaep_sha256_encrypt(f"refresh_{timestamp}".encode("utf-8"))
|
|
request = urllib.request.Request(
|
|
f"https://www.bilibili.com/correspond/1/{correspond_path}",
|
|
headers={"User-Agent": _USER_AGENT, "Cookie": current_cookie},
|
|
)
|
|
html = urllib.request.urlopen(request, timeout=15).read().decode("utf-8", errors="replace")
|
|
match = __import__("re").search(r'<div\s+id=["\']1-name["\']>([^<]+)</div>', html)
|
|
if not match:
|
|
return {"status": "failed", "message": "未获取到 refresh_csrf"}
|
|
refresh_csrf = match.group(1).strip()
|
|
|
|
old_refresh_token = refresh_token
|
|
jar = http.cookiejar.CookieJar()
|
|
_seed_cookie_jar(jar, current_values)
|
|
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar))
|
|
refreshed, _ = _request_json(
|
|
"https://passport.bilibili.com/x/passport-login/web/cookie/refresh",
|
|
data={
|
|
"csrf": csrf,
|
|
"refresh_csrf": refresh_csrf,
|
|
"source": "main_web",
|
|
"refresh_token": old_refresh_token,
|
|
},
|
|
opener=opener,
|
|
)
|
|
if refreshed.get("code") != 0:
|
|
return {"status": "failed", "message": f"Cookie 刷新失败 code={refreshed.get('code')}"}
|
|
new_refresh_token = str((refreshed.get("data") or {}).get("refresh_token") or "")
|
|
if not new_refresh_token:
|
|
return {"status": "failed", "message": "刷新响应缺少新 refresh_token"}
|
|
new_values = dict(current_values)
|
|
new_values.update(_jar_values(jar))
|
|
new_csrf = new_values.get("bili_jct", "")
|
|
new_cookie = _cookie_header(new_values)
|
|
|
|
confirmed, _ = _request_json(
|
|
"https://passport.bilibili.com/x/passport-login/web/confirm/refresh",
|
|
cookie=new_cookie,
|
|
data={"csrf": new_csrf, "refresh_token": old_refresh_token},
|
|
)
|
|
if confirmed.get("code") != 0:
|
|
return {"status": "failed", "message": f"刷新确认失败 code={confirmed.get('code')}"}
|
|
|
|
nav, _ = _request_json(
|
|
"https://api.bilibili.com/x/web-interface/nav",
|
|
cookie=new_cookie,
|
|
retries=2,
|
|
)
|
|
if nav.get("code") != 0 or not bool((nav.get("data") or {}).get("isLogin")):
|
|
return {"status": "failed", "message": "新 Cookie 登录验证失败"}
|
|
self.credential_store.save_refresh_token(new_refresh_token)
|
|
self.update_cookie(new_values)
|
|
return {"status": "refreshed", "message": "Cookie 已刷新并验证"}
|
|
|
|
async def check_once(self) -> dict[str, Any]:
|
|
try:
|
|
result = await asyncio.to_thread(self._check_and_refresh_sync)
|
|
except Exception as exc:
|
|
self.logger.warning("[B站凭据] 自动检查异常: %s", type(exc).__name__)
|
|
return {"status": "failed", "message": type(exc).__name__}
|
|
status = result.get("status")
|
|
if status == "refreshed":
|
|
self.logger.info("[B站凭据] Cookie 已自动续期并完成登录验证")
|
|
if self.on_refreshed:
|
|
callback_result = self.on_refreshed()
|
|
if asyncio.iscoroutine(callback_result):
|
|
await callback_result
|
|
elif status == "valid":
|
|
self.logger.info("[B站凭据] Cookie 有效，当前无需续期")
|
|
elif status == "disabled":
|
|
self.logger.warning("[B站凭据] 自动续期未启用：未配置刷新令牌")
|
|
else:
|
|
self.logger.warning("[B站凭据] 自动续期失败：%s", result.get("message", "未知错误"))
|
|
return result
|
|
|
|
async def run(self) -> None:
|
|
while not self._stop:
|
|
if self.is_enabled():
|
|
await self.check_once()
|
|
try:
|
|
await asyncio.wait_for(self._wake.wait(), timeout=self._current_interval())
|
|
self._wake.clear()
|
|
except asyncio.TimeoutError:
|
|
continue
|