kairee
2641ba6f87
WSL and --no-browser never showed the PNG, and placeholder Chinese cookies crashed urllib latin-1 headers before the code was issued.
295 lines
12 KiB
Python
295 lines
12 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""扫码授权流程；仅由 ``python app.py auth`` 调用。
|
|
|
|
认证结果只写入指定的 ``bilibili.read`` 或 ``bilibili.write`` 段，两个账号的
|
|
身份及设备指纹 Cookie 绝不互相覆盖。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import struct
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
import webbrowser
|
|
import zlib
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Mapping
|
|
|
|
from .auth import (
|
|
IDENTITY_KEYS,
|
|
cookie_value,
|
|
cookies_from_set_cookie,
|
|
gen_uuid,
|
|
make_headers,
|
|
merge_cookie,
|
|
parse_cookie,
|
|
sessdata_expiry,
|
|
)
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
VENDOR = ROOT / "vendor"
|
|
if str(VENDOR) not in sys.path:
|
|
sys.path.insert(0, str(VENDOR))
|
|
import qrcodegen # noqa: E402
|
|
|
|
BASE = "https://api.bilibili.com"
|
|
PASSPORT = "https://passport.bilibili.com"
|
|
_QR_BLOCKS = {
|
|
(False, False): " ",
|
|
(True, False): "▀",
|
|
(False, True): "▄",
|
|
(True, True): "█",
|
|
}
|
|
_QR_ASCII_STYLE = "\033[48;5;231m\033[38;5;16m"
|
|
_QR_ASCII_RESET = "\033[0m"
|
|
|
|
|
|
def render_qr_ascii(text: str, *, border: int = 2) -> str:
|
|
"""把登录链接编成终端半块字符二维码（两行模块合成一行）。"""
|
|
if border < 0:
|
|
raise ValueError("border 必须 >= 0")
|
|
qr = qrcodegen.QrCode.encode_text(text, qrcodegen.QrCode.Ecc.MEDIUM)
|
|
size = qr.get_size()
|
|
dim = size + border * 2
|
|
|
|
def dark(x: int, y: int) -> bool:
|
|
return qr.get_module(x - border, y - border)
|
|
|
|
lines: list[str] = []
|
|
for y in range(0, dim, 2):
|
|
row: list[str] = []
|
|
bottom_y = y + 1
|
|
for x in range(dim):
|
|
top = dark(x, y)
|
|
bottom = dark(x, bottom_y) if bottom_y < dim else False
|
|
row.append(_QR_BLOCKS[(top, bottom)])
|
|
lines.append("".join(row))
|
|
return "\n".join(lines)
|
|
|
|
|
|
def format_qr_ascii(text: str, *, border: int = 2, color: bool | None = None) -> str:
|
|
"""终端二维码；TTY 下加白底黑字，方便深色终端扫码。"""
|
|
body = render_qr_ascii(text, border=border)
|
|
if color is None:
|
|
color = sys.stdout.isatty()
|
|
if not color:
|
|
return body
|
|
return "\n".join(f"{_QR_ASCII_STYLE}{line}{_QR_ASCII_RESET}" for line in body.split("\n"))
|
|
|
|
|
|
def write_qr_png(text: str, path: str | Path, scale: int = 8, border: int = 4) -> Path:
|
|
"""将登录链接编码为不依赖图像库的 8-bit 灰度 QR PNG。"""
|
|
qr = qrcodegen.QrCode.encode_text(text, qrcodegen.QrCode.Ecc.MEDIUM)
|
|
size = qr.get_size()
|
|
dim = (size + border * 2) * scale
|
|
rows: list[bytes] = []
|
|
for y in range(dim):
|
|
my = y // scale - border
|
|
row = bytearray([0])
|
|
for x in range(dim):
|
|
mx = x // scale - border
|
|
dark = 0 <= mx < size and 0 <= my < size and qr.get_module(mx, my)
|
|
row.append(0 if dark else 255)
|
|
rows.append(bytes(row))
|
|
|
|
def chunk(tag: bytes, data: bytes) -> bytes:
|
|
return struct.pack(">I", len(data)) + tag + data + struct.pack(
|
|
">I", zlib.crc32(tag + data) & 0xFFFFFFFF
|
|
)
|
|
|
|
output = Path(path)
|
|
output.write_bytes(
|
|
b"\x89PNG\r\n\x1a\n"
|
|
+ chunk(b"IHDR", struct.pack(">IIBBBBB", dim, dim, 8, 0, 0, 0, 0))
|
|
+ chunk(b"IDAT", zlib.compress(b"".join(rows), 9))
|
|
+ chunk(b"IEND", b"")
|
|
)
|
|
return output
|
|
|
|
|
|
def open_path(path: Path) -> bool:
|
|
if os.name == "nt":
|
|
os.startfile(path) # type: ignore[attr-defined]
|
|
return True
|
|
return webbrowser.open(path.as_uri())
|
|
|
|
|
|
def ensure_device_cookies(
|
|
credentials: dict[str, Any], get_json: Callable[[str, bool], Mapping[str, Any]]
|
|
) -> list[str]:
|
|
"""仅为当前账号补齐缺失设备字段，已有指纹不改动。"""
|
|
pairs = parse_cookie(str(credentials.get("cookie") or ""))
|
|
lower = {key.lower() for key, _ in pairs}
|
|
added: list[str] = []
|
|
if "buvid3" not in lower or "buvid4" not in lower:
|
|
try:
|
|
data = get_json(f"{BASE}/x/frontend/finger/spi", True).get("data") or {}
|
|
except Exception: # noqa: BLE001
|
|
data = {}
|
|
if data.get("b_3") and "buvid3" not in lower:
|
|
pairs.append(("buvid3", str(data["b_3"])))
|
|
lower.add("buvid3")
|
|
added.append("buvid3")
|
|
if data.get("b_4") and "buvid4" not in lower:
|
|
pairs.append(("buvid4", str(data["b_4"])))
|
|
lower.add("buvid4")
|
|
added.append("buvid4")
|
|
if "_uuid" not in lower:
|
|
pairs.append(("_uuid", gen_uuid()))
|
|
lower.add("_uuid")
|
|
added.append("_uuid")
|
|
if "b_nut" not in lower and "buvid3" in added:
|
|
pairs.append(("b_nut", str(int(time.time()))))
|
|
credentials["cookie"] = "; ".join(f"{key}={value}" for key, value in pairs)
|
|
return added
|
|
|
|
|
|
class AuthService:
|
|
def __init__(self, config_path: str | Path, *, account: str, logger=None) -> None:
|
|
if account not in {"read", "write"}:
|
|
raise ValueError(f"未知账号角色: {account}")
|
|
self.path = Path(config_path)
|
|
self.account = account
|
|
self.logger = logger
|
|
|
|
@property
|
|
def backup_path(self) -> Path:
|
|
return self.path.with_name(self.path.name + ".bak")
|
|
|
|
def _load_raw(self) -> dict[str, Any]:
|
|
with self.path.open(encoding="utf-8") as fh:
|
|
raw = json.load(fh)
|
|
if not isinstance(raw, dict):
|
|
raise ValueError("配置文件顶层必须是对象")
|
|
section = raw.get("bilibili")
|
|
if not isinstance(section, dict) or not isinstance(section.get(self.account), dict):
|
|
raise ValueError(f"缺少 bilibili.{self.account} 账号配置")
|
|
return raw
|
|
|
|
@staticmethod
|
|
def _request(credentials: Mapping[str, Any], url: str, *, auth: bool) -> tuple[dict[str, Any], list[str]]:
|
|
req = urllib.request.Request(url, headers=make_headers(dict(credentials), auth=auth))
|
|
with urllib.request.urlopen(req, timeout=30) as response:
|
|
return (
|
|
json.loads(response.read().decode("utf-8", "replace")),
|
|
response.headers.get_all("Set-Cookie") or [],
|
|
)
|
|
|
|
def _get_json(self, credentials: Mapping[str, Any], url: str, auth: bool) -> Mapping[str, Any]:
|
|
return self._request(credentials, url, auth=auth)[0]
|
|
|
|
def _write_raw(self, raw: Mapping[str, Any]) -> None:
|
|
tmp = self.path.with_name(self.path.name + ".tmp")
|
|
with tmp.open("w", encoding="utf-8") as fh:
|
|
json.dump(raw, fh, ensure_ascii=False, indent=2)
|
|
fh.write("\n")
|
|
os.replace(tmp, self.path)
|
|
|
|
def finish(self, raw: dict[str, Any], pairs: Mapping[str, str]) -> int:
|
|
"""写入单个账号、验证 nav；验证失败还原整个原配置文件。"""
|
|
credentials = raw["bilibili"][self.account]
|
|
old_cookie = str(credentials.get("cookie") or "")
|
|
merged = merge_cookie(old_cookie, dict(pairs))
|
|
if not cookie_value(merged, "SESSDATA") or not cookie_value(merged, "bili_jct"):
|
|
print("[auth] 合并后缺少 SESSDATA / bili_jct，中止")
|
|
return 2
|
|
credentials["cookie"] = merged
|
|
credentials["csrf"] = cookie_value(merged, "bili_jct")
|
|
added = ensure_device_cookies(
|
|
credentials,
|
|
lambda url, auth: self._get_json(credentials, url, auth),
|
|
)
|
|
new_mid = cookie_value(str(credentials["cookie"]), "DedeUserID")
|
|
if new_mid.isdigit():
|
|
credentials["mid"] = int(new_mid)
|
|
|
|
shutil.copy2(self.path, self.backup_path)
|
|
self._write_raw(raw)
|
|
try:
|
|
nav = self._get_json(credentials, f"{BASE}/x/web-interface/nav", True)
|
|
data = nav.get("data") or {}
|
|
ok = nav.get("code") == 0 and bool(data.get("isLogin"))
|
|
except Exception as exc: # noqa: BLE001
|
|
data, ok = {"uname": str(exc)}, False
|
|
if not ok:
|
|
shutil.copy2(self.backup_path, self.path)
|
|
print(f"[auth] {self.account} 登录校验失败，已回滚 config.json")
|
|
return 4
|
|
|
|
if isinstance(data.get("mid"), int):
|
|
credentials["mid"] = data["mid"]
|
|
self._write_raw(raw)
|
|
updated = sorted(key for key in pairs if key.lower() in IDENTITY_KEYS)
|
|
kept = [key for key in ("buvid3", "buvid4", "_uuid", "buvid_fp")
|
|
if cookie_value(str(credentials["cookie"]), key)]
|
|
print(f"[auth] {self.account} 已更新身份字段: {', '.join(updated)}")
|
|
if added:
|
|
print(f"[auth] {self.account} 补齐设备字段: {', '.join(added)}")
|
|
print(f"[auth] {self.account} 保留设备指纹: {', '.join(kept) if kept else '（无）'}")
|
|
expiry = sessdata_expiry(str(credentials["cookie"]))
|
|
if expiry:
|
|
print(f"[auth] 新 SESSDATA 到期: {time.strftime('%Y-%m-%d', time.localtime(expiry))}")
|
|
print(f"[auth] {self.account} 校验通过 ✅ mid={credentials['mid']} uname={data.get('uname')}")
|
|
return 0
|
|
|
|
def run(self, *, timeout: int, no_browser: bool) -> int:
|
|
raw = self._load_raw()
|
|
credentials = raw["bilibili"][self.account]
|
|
print(f"[auth] 为 {self.account} 账号申请扫码登录 ...", flush=True)
|
|
generated, _ = self._request(
|
|
credentials, f"{PASSPORT}/x/passport-login/web/qrcode/generate?source=main-fe-header", auth=False
|
|
)
|
|
data = generated.get("data") or {}
|
|
key, login_url = data.get("qrcode_key"), data.get("url")
|
|
if generated.get("code") != 0 or not key or not login_url:
|
|
print(f"[auth] 申请二维码失败: code={generated.get('code')} {generated.get('message')}")
|
|
return 2
|
|
login_url = str(login_url)
|
|
qr_path = Path(tempfile.gettempdir()) / f"bili-fav-{self.account}-{key[:8]}.png"
|
|
write_qr_png(login_url, qr_path)
|
|
print(f"[auth] 二维码已生成: {qr_path}", flush=True)
|
|
print("[auth] 请使用哔哩哔哩 App 扫描下方二维码", flush=True)
|
|
print(format_qr_ascii(login_url), flush=True)
|
|
print(f"[auth] 登录链接: {login_url}", flush=True)
|
|
if not no_browser:
|
|
try:
|
|
open_path(qr_path)
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"[auth] 打开图片失败: {exc}，请手动扫描终端二维码")
|
|
deadline, last = time.time() + timeout, None
|
|
while time.time() < deadline:
|
|
try:
|
|
payload, set_cookies = self._request(
|
|
credentials,
|
|
f"{PASSPORT}/x/passport-login/web/qrcode/poll?qrcode_key={urllib.parse.quote(str(key))}",
|
|
auth=False,
|
|
)
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"[auth] 轮询异常: {exc}")
|
|
time.sleep(3)
|
|
continue
|
|
poll = payload.get("data") or {}
|
|
state = poll.get("code")
|
|
if state == 0:
|
|
pairs = cookies_from_set_cookie(set_cookies)
|
|
for part in urllib.parse.urlparse(str(poll.get("url") or "")).query.split("&"):
|
|
key_name, _, value = part.partition("=")
|
|
if key_name and value:
|
|
pairs.setdefault(key_name, value)
|
|
return self.finish(raw, pairs)
|
|
if state == 86038:
|
|
print("[auth] 二维码已过期，请重新运行 auth")
|
|
return 3
|
|
if state != last:
|
|
print(f"[auth] {(poll.get('message') or {86101: '未扫码', 86090: '已扫码，等待确认'}.get(state, f'状态 {state}'))} ...")
|
|
last = state
|
|
time.sleep(2)
|
|
print("[auth] 等待超时，二维码已失效，请重新运行 auth")
|
|
return 3
|