完善虎牙账号管理并规范Cookie

This commit is contained in:
yml2213
2026-07-05 13:13:38 +08:00
parent 8f48b96fdc
commit 984096f758
15 changed files with 1320 additions and 59 deletions
+60
View File
@@ -0,0 +1,60 @@
"""虎牙 Cookie 规范化工具。"""
from __future__ import annotations
from collections.abc import Iterable, Mapping
import requests
def cookie_pairs(cookie: str) -> list[tuple[str, str]]:
"""把浏览器 Cookie 字符串拆成 key/value 对。"""
pairs: list[tuple[str, str]] = []
for part in (cookie or "").split(";"):
item = part.strip()
if not item or "=" not in item:
continue
key, value = item.split("=", 1)
key = key.strip()
if not key:
continue
pairs.append((key, value.strip()))
return pairs
def normalize_cookie_pairs(pairs: Iterable[tuple[str, str]]) -> str:
"""按 Cookie key 去重,保留最后一次出现的值。"""
ordered_keys: list[str] = []
values: dict[str, str] = {}
for raw_key, raw_value in pairs:
key = str(raw_key).strip()
if not key:
continue
if key not in values:
ordered_keys.append(key)
values[key] = "" if raw_value is None else str(raw_value).strip()
return "; ".join(f"{key}={values[key]}" for key in ordered_keys)
def normalize_huya_cookie(cookie: str | Mapping[str, str] | requests.cookies.RequestsCookieJar | None) -> str:
"""把 Cookie 字符串、dict 或 CookieJar 转成去重后的浏览器 Cookie 字符串。"""
if not cookie:
return ""
if isinstance(cookie, str):
return normalize_cookie_pairs(cookie_pairs(cookie))
return normalize_cookie_pairs(cookie.items())
def cookie_value(cookie: str | Mapping[str, str] | requests.cookies.RequestsCookieJar | None, key: str) -> str:
"""从 Cookie 中读取 key;有重复时以最后一次出现为准。"""
if not cookie or not key:
return ""
if isinstance(cookie, str):
pairs = cookie_pairs(cookie)
else:
pairs = list(cookie.items())
value = ""
for item_key, item_value in pairs:
if item_key == key:
value = "" if item_value is None else str(item_value).strip()
return value
+4 -11
View File
@@ -14,6 +14,7 @@ import urllib.parse
import urllib.request
from typing import Optional, Callable
from .cookie_utils import cookie_pairs, normalize_cookie_pairs, normalize_huya_cookie
from .taf_protocol import TafOutputStream, TafInputStream, TafType, TafStruct
from .wup_protocol import WupRequest, WupResponse
@@ -122,16 +123,8 @@ class HuyaHttpClient:
@staticmethod
def _normalize_cookie(cookie: str) -> str:
"""HTTP 业务 UserId.sCookie 需要带 huya_ua 前缀,避免重复写入。"""
import re
cookie = (cookie or "").strip()
normalized = f"huya_ua={HTTP_HUYA_UA}"
if re.search(r"(?:^|;\s*)huya_ua=", cookie):
return re.sub(r"(^|;\s*)huya_ua=[^;]*",
lambda m: f"{m.group(1)}{normalized}",
cookie, count=1)
if not cookie:
return normalized
return f"{normalized}; {cookie}"
pairs = [(key, value) for key, value in cookie_pairs(normalize_huya_cookie(cookie)) if key != "huya_ua"]
return normalize_cookie_pairs([("huya_ua", HTTP_HUYA_UA), *pairs])
@staticmethod
def _build_user(uid: int, guid: str, cookie: str):
@@ -176,7 +169,7 @@ class HuyaHttpClient:
user = ActivityUserId()
user.lUid = uid
user.sHuYaUA = HTTP_HUYA_UA
user.sCookie = cookie or ""
user.sCookie = normalize_huya_cookie(cookie)
return user
def call_rpc(self, service: str, method: str,
+3 -2
View File
@@ -16,6 +16,8 @@ from urllib.parse import quote, urlsplit, urlunsplit
import requests
from loguru import logger
from .cookie_utils import normalize_huya_cookie
APP_ID = "5002"
APP_VERSION = "2.6"
@@ -63,8 +65,7 @@ def password_sha1(password: str) -> str:
def cookie_string(cookies: requests.cookies.RequestsCookieJar | Mapping[str, str]) -> str:
"""把 CookieJar/dict 转成浏览器 Cookie 字符串。"""
items = cookies.items() if isinstance(cookies, Mapping) else cookies.items()
return "; ".join(f"{key}={value}" for key, value in items if value is not None)
return normalize_huya_cookie(cookies)
def cookie_mapping(cookie: Mapping[str, str] | str | None) -> dict[str, str]: