# -*- 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" 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} 账号申请扫码登录 ...") 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 qr_path = Path(tempfile.gettempdir()) / f"bili-fav-{self.account}-{key[:8]}.png" write_qr_png(str(login_url), qr_path) print(f"[auth] 二维码已生成: {qr_path}") 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