新增斗鱼活动任务模块：绑定、宝典、鱼翅、积分、兑换等功能
- 新增 activity_client.py：封装斗鱼活动/兑换/充值/送礼接口 - 新增 cookie_utils.py：Cookie 解析与规范化工具 - 新增 douyu_service/douyu_runner：斗鱼任务服务层与批量执行器 - 新增 douyu 路由：任务类型查询、账号列表、配置管理、商品管理、批量任务、WebSocket 日志 - 新增 models/schemas：DouyuTask/DouyuConfig/DouyuGoodsSnapshot 模型，Account 扩展点数/鱼翅/绑定状态等字段 - 新增数据库迁移：斗鱼活动相关表与 accounts 字段补充 - 新增前端 DouyuTasksPage 任务操作台页面 - 兑换商品请求添加 sec-ch-ua 反检测头 - 兑换商品支持最多 8 次重试 + csrf_token 自动刷新 - 注册 douyu:task / douyu:config 权限点 - 侧边栏新增斗鱼分组与任务操作台菜单入口
This commit is contained in:
yml2213
@@ -5,6 +5,7 @@ from .login_api import LoginAPIStrategy
|
||||
from .login_api_wgapi import WgapiLoginAPI
|
||||
from .login_api_iframe import IframeLoginAPI
|
||||
from .email_verifier import EmailVerifier
|
||||
from .activity_client import DouyuActivityClient, DouyuActivityError
|
||||
|
||||
__all__ = [
|
||||
"DouyuLogin",
|
||||
@@ -13,4 +14,6 @@ __all__ = [
|
||||
"WgapiLoginAPI",
|
||||
"IframeLoginAPI",
|
||||
"EmailVerifier",
|
||||
"DouyuActivityClient",
|
||||
"DouyuActivityError",
|
||||
]
|
||||
|
||||
@@ -0,0 +1,465 @@
|
||||
"""斗鱼活动、兑换与充值接口客户端。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import random
|
||||
import string
|
||||
from collections.abc import Callable
|
||||
from typing import Any
|
||||
from urllib.parse import unquote
|
||||
|
||||
import requests
|
||||
|
||||
from .cookie_utils import cookie_pairs, cookie_value, normalize_cookie_pairs, normalize_douyu_cookie
|
||||
|
||||
|
||||
PC_UA = (
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||||
"Chrome/133.0.0.0 Safari/537.36"
|
||||
)
|
||||
|
||||
|
||||
class DouyuActivityError(RuntimeError):
|
||||
"""斗鱼活动接口异常。"""
|
||||
|
||||
|
||||
class DouyuActivityClient:
|
||||
"""封装斗鱼活动页、兑换、充值和送礼相关接口。"""
|
||||
|
||||
CSRF_API = "https://www.douyu.com/japi/carnival/nc/common/generateCsrf"
|
||||
ACF_CCN_API = "https://www.douyu.com/curl/csrfApi/getCsrfCookie?"
|
||||
ACF_CCN_FALLBACK_API = "https://www.douyu.com/curl/csrfNlApi/getCsrfCookie"
|
||||
SUBSCRIBE_API = "https://www.douyu.com/wgapi/livenc/asubscribe/userSubStatus"
|
||||
ROLE_PARAM_API = "https://www.douyu.com/japi/carnivalApi/tencent/roleParam"
|
||||
BIND_INFO_API = "https://www.douyu.com/japi/carnivalApi/tencent/bindInfo"
|
||||
BIND_INFO_V2_API = "https://www.douyu.com/japi/carnivalApi/v2/tencent/bindInfo"
|
||||
BIND_API = "https://www.douyu.com/japi/carnivalApi/tencent/bind"
|
||||
GOODS_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/storehome"
|
||||
EXCHANGE_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/exchange"
|
||||
EXCHANGE_LIST_API = "https://www.douyu.com/wgapi/ordnc/activity/peace/exchangeList"
|
||||
CREDIT_BALANCE_API = "https://www.douyu.com/japi/oms/web/peace/credit/balance"
|
||||
CREDIT_DIFF_API = "https://www.douyu.com/japi/oms/web/peace/credit/diff"
|
||||
PEACE_ITEM_API = "https://www.douyu.com/japi/oms/web/peace/item"
|
||||
PAY_QR_API = "https://www.douyu.com/japi/oms/web/pay/getQrCode"
|
||||
GOLD_QR_API = "https://cz.douyu.com/m/gold/getQrCode"
|
||||
GOLD_ACCOUNT_API = "https://cz.douyu.com/item/gold/account"
|
||||
EXCHANGE_BALANCE_API = "https://www.douyu.com/wjapi/nc/exchange/fim"
|
||||
DONATE_API = "https://www.douyu.com/japi/gift/donate/mainsite/v3"
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
cookie: str,
|
||||
*,
|
||||
logger: Callable[[str], None] | None = None,
|
||||
timeout: tuple[float, float] = (8, 20),
|
||||
):
|
||||
self.cookie = normalize_douyu_cookie(cookie)
|
||||
self.timeout = timeout
|
||||
self.logger = logger or (lambda _msg: None)
|
||||
self.session = requests.Session()
|
||||
self.session.trust_env = False
|
||||
self.session.headers.update({
|
||||
"User-Agent": PC_UA,
|
||||
"Accept": "application/json, text/plain, */*",
|
||||
"Accept-Language": "zh-CN,zh;q=0.9",
|
||||
})
|
||||
self._load_cookie(self.cookie)
|
||||
|
||||
@staticmethod
|
||||
def profile_from_cookie(cookie: str) -> dict[str, str]:
|
||||
"""从斗鱼 Cookie 提取账号基础信息。"""
|
||||
nickname = cookie_value(cookie, "acf_nickname")
|
||||
return {
|
||||
"uid": cookie_value(cookie, "acf_uid"),
|
||||
"nickname": unquote(nickname) if nickname else "",
|
||||
}
|
||||
|
||||
def _load_cookie(self, cookie: str) -> None:
|
||||
for key, value in cookie_pairs(cookie):
|
||||
self.session.cookies.set(key, value, domain=".douyu.com", path="/")
|
||||
|
||||
def _merge_response_cookie(self, response: requests.Response) -> None:
|
||||
if not response.cookies:
|
||||
return
|
||||
self.cookie = normalize_cookie_pairs([
|
||||
*cookie_pairs(self.cookie),
|
||||
*[(key, value) for key, value in response.cookies.items()],
|
||||
])
|
||||
self._load_cookie(self.cookie)
|
||||
|
||||
def _headers(self, referer: str = "https://www.douyu.com/") -> dict[str, str]:
|
||||
return {
|
||||
"User-Agent": PC_UA,
|
||||
"Accept": "application/json, text/plain, */*",
|
||||
"Accept-Language": "zh-CN,zh;q=0.9",
|
||||
"Origin": "https://www.douyu.com",
|
||||
"Referer": referer,
|
||||
"Cookie": self.cookie,
|
||||
}
|
||||
|
||||
def _request(
|
||||
self,
|
||||
method: str,
|
||||
url: str,
|
||||
*,
|
||||
source: str,
|
||||
headers: dict[str, str] | None = None,
|
||||
**kwargs,
|
||||
) -> requests.Response:
|
||||
req_headers = self._headers()
|
||||
if headers:
|
||||
req_headers.update(headers)
|
||||
response = self.session.request(
|
||||
method,
|
||||
url,
|
||||
headers=req_headers,
|
||||
timeout=kwargs.pop("timeout", self.timeout),
|
||||
**kwargs,
|
||||
)
|
||||
self._merge_response_cookie(response)
|
||||
self.logger(f"{source}: {response.status_code}")
|
||||
response.raise_for_status()
|
||||
return response
|
||||
|
||||
def _json(self, response: requests.Response, source: str) -> dict[str, Any]:
|
||||
text = response.text.strip()
|
||||
if not text:
|
||||
raise DouyuActivityError(f"{source} 返回为空")
|
||||
try:
|
||||
return response.json()
|
||||
except json.JSONDecodeError as exc:
|
||||
raise DouyuActivityError(f"{source} 返回不是 JSON: {text[:200]}") from exc
|
||||
|
||||
def _request_json(self, method: str, url: str, source: str, **kwargs) -> dict[str, Any]:
|
||||
return self._json(self._request(method, url, source=source, **kwargs), source)
|
||||
|
||||
def csrf_token(self) -> str:
|
||||
"""获取 cvl_csrf_token。"""
|
||||
payload = self._request_json(
|
||||
"post",
|
||||
self.CSRF_API,
|
||||
"生成 CSRF",
|
||||
data="{}",
|
||||
headers={
|
||||
"Content-Type": "application/json;charset=UTF-8",
|
||||
"Referer": "https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0",
|
||||
},
|
||||
)
|
||||
if payload.get("error") not in (0, "0", None):
|
||||
raise DouyuActivityError(payload.get("msg") or "生成 CSRF 失败")
|
||||
token = cookie_value(self.cookie, "cvl_csrf_token")
|
||||
if not token:
|
||||
raise DouyuActivityError("响应 Cookie 中没有 cvl_csrf_token")
|
||||
return token
|
||||
|
||||
def acf_ccn(self, *, refresh_subscribe: bool = True) -> str:
|
||||
"""获取 acf_ccn；必要时通过订阅接口刷新一次。"""
|
||||
response = self._request(
|
||||
"get",
|
||||
self.ACF_CCN_API,
|
||||
source="获取 acf_ccn",
|
||||
headers={
|
||||
"Origin": "",
|
||||
"Referer": "https://www.douyu.com/pages/ord-task-center?clientType=web&panelSource=1&rid=0",
|
||||
},
|
||||
)
|
||||
ctn = cookie_value(self.cookie, "acf_ccn")
|
||||
if not ctn:
|
||||
response = self._request(
|
||||
"get",
|
||||
self.ACF_CCN_FALLBACK_API,
|
||||
source="获取 acf_ccn 备用接口",
|
||||
headers={"Origin": ""},
|
||||
)
|
||||
ctn = cookie_value(self.cookie, "acf_ccn")
|
||||
if not ctn:
|
||||
raise DouyuActivityError(f"获取 acf_ccn 失败: {response.text[:200]}")
|
||||
|
||||
if refresh_subscribe:
|
||||
payload = self._request_json(
|
||||
"post",
|
||||
self.SUBSCRIBE_API,
|
||||
"刷新订阅状态",
|
||||
data={"subscribe": "cmn", "ctn": ctn},
|
||||
headers={
|
||||
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
|
||||
"Referer": "https://www.douyu.com/",
|
||||
},
|
||||
)
|
||||
if payload.get("error") not in (0, "0"):
|
||||
raise DouyuActivityError(payload.get("msg") or "刷新订阅状态失败")
|
||||
ctn = cookie_value(self.cookie, "acf_ccn") or ctn
|
||||
return ctn
|
||||
|
||||
def get_bind_qr(self, act_alias: str) -> dict[str, Any]:
|
||||
"""生成腾讯游戏绑定二维码链接。"""
|
||||
ctn = self.acf_ccn(refresh_subscribe=True)
|
||||
payload = self._request_json(
|
||||
"post",
|
||||
self.ROLE_PARAM_API,
|
||||
"获取绑定二维码",
|
||||
data={"actAlias": act_alias, "ctn": ctn},
|
||||
headers={
|
||||
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
|
||||
"Referer": "https://www.douyu.com/",
|
||||
},
|
||||
)
|
||||
url = (payload.get("data") or {}).get("url") or ""
|
||||
if not url:
|
||||
raise DouyuActivityError(payload.get("msg") or "获取绑定二维码失败")
|
||||
return {"url": url, "ctn": ctn, "raw": payload}
|
||||
|
||||
def bind_info(self, act_alias: str, *, v2: bool = False) -> dict[str, Any]:
|
||||
"""查询绑定游戏账号信息。"""
|
||||
url = self.BIND_INFO_V2_API if v2 else self.BIND_INFO_API
|
||||
payload = self._request_json(
|
||||
"get",
|
||||
url,
|
||||
"查询绑定信息",
|
||||
params={"actAlias": act_alias},
|
||||
headers={"Origin": "", "Referer": "https://www.douyu.com/"},
|
||||
)
|
||||
data = payload.get("data") or {}
|
||||
game_account = data.get("gameAccount") or {}
|
||||
game_role = data.get("gameRole") or {}
|
||||
return {
|
||||
"bind_account": data.get("bindAccount"),
|
||||
"bind_role": data.get("bindRole"),
|
||||
"bind_act": data.get("bindAct"),
|
||||
"nickname": game_account.get("nick") or "",
|
||||
"role_name": game_role.get("roleName") or "",
|
||||
"area_name": game_role.get("areaName") or "",
|
||||
"plat_name": game_role.get("platName") or "",
|
||||
"change_role_wait_time": data.get("changeRoleWaitTime"),
|
||||
"can_change_role": data.get("canChangeRole"),
|
||||
"raw": payload,
|
||||
}
|
||||
|
||||
def confirm_bind(self, act_alias: str) -> dict[str, Any]:
|
||||
"""确认绑定当前扫码选择的游戏角色。"""
|
||||
token = self.csrf_token()
|
||||
payload = self._request_json(
|
||||
"post",
|
||||
self.BIND_API,
|
||||
"确认绑定",
|
||||
data={"actAlias": act_alias, "csrfToken": token},
|
||||
headers={
|
||||
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
|
||||
"Referer": "https://www.douyu.com/",
|
||||
},
|
||||
)
|
||||
if payload.get("error") not in (0, "0"):
|
||||
raise DouyuActivityError(payload.get("msg") or "确认绑定失败")
|
||||
return {"csrf_token": token, "raw": payload}
|
||||
|
||||
def create_elite_qr(self, *, ctn: str, act_alias: str, amount: int, room_id: str) -> dict[str, Any]:
|
||||
"""生成精英宝典支付二维码。"""
|
||||
item_payload = self._request_json(
|
||||
"post",
|
||||
self.PEACE_ITEM_API,
|
||||
"校验精英宝典",
|
||||
json={"itemId": 1, "ctn": ctn},
|
||||
headers={"Content-Type": "application/json;charset=UTF-8"},
|
||||
)
|
||||
if item_payload.get("error") not in (0, "0"):
|
||||
raise DouyuActivityError(item_payload.get("msg") or "精英宝典校验失败")
|
||||
|
||||
payload = self._request_json(
|
||||
"post",
|
||||
self.PAY_QR_API,
|
||||
"生成精英宝典支付码",
|
||||
json={
|
||||
"bizType": "peace",
|
||||
"biz": {"actAlias": act_alias, "amount": amount, "roomId": int(room_id or 0)},
|
||||
"ctn": ctn,
|
||||
},
|
||||
headers={"Content-Type": "application/json;charset=UTF-8"},
|
||||
)
|
||||
qr = (payload.get("data") or {}).get("qr") or ""
|
||||
if not qr:
|
||||
raise DouyuActivityError(payload.get("msg") or "生成精英宝典支付码失败")
|
||||
return {"pay_url": qr, "ctn": ctn, "item": item_payload, "raw": payload}
|
||||
|
||||
def create_gold_qr(self, *, amount: int, pay_type: int = 1, product_id: str = "DYTV_Product_Gold_1") -> dict[str, Any]:
|
||||
"""生成鱼翅充值支付二维码。"""
|
||||
payload = self._request_json(
|
||||
"post",
|
||||
self.GOLD_QR_API,
|
||||
"生成鱼翅充值码",
|
||||
data={
|
||||
"pay_type": str(pay_type),
|
||||
"product_id": product_id,
|
||||
"number": str(amount),
|
||||
"toname": "",
|
||||
"source": "dy_box",
|
||||
},
|
||||
headers={
|
||||
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
|
||||
"Origin": "https://cz.douyu.com",
|
||||
"Referer": "https://cz.douyu.com/",
|
||||
},
|
||||
)
|
||||
data = payload.get("data") or {}
|
||||
pay_url = data.get("code_url") or payload.get("code_url") or ""
|
||||
token = data.get("token") or payload.get("token") or ""
|
||||
if not pay_url:
|
||||
raise DouyuActivityError(payload.get("msg") or "生成鱼翅充值码失败")
|
||||
return {"pay_url": pay_url, "token": token, "raw": payload}
|
||||
|
||||
def donate_elite_gift(
|
||||
self,
|
||||
*,
|
||||
gift_count: int,
|
||||
room_id: str,
|
||||
gift_id: str,
|
||||
skin_id: str,
|
||||
) -> dict[str, Any]:
|
||||
"""赠送精英令礼物。"""
|
||||
payload = self._request_json(
|
||||
"post",
|
||||
self.DONATE_API,
|
||||
"赠送精英令",
|
||||
data={
|
||||
"giftId": gift_id,
|
||||
"giftCount": str(gift_count),
|
||||
"roomId": room_id,
|
||||
"bizExt": json.dumps({"isMainGift": 1}, separators=(",", ":")),
|
||||
"skinId": skin_id,
|
||||
},
|
||||
headers={
|
||||
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
|
||||
"Referer": f"https://www.douyu.com/{room_id}",
|
||||
},
|
||||
)
|
||||
if payload.get("error") not in (0, "0") and payload.get("msg") != "success":
|
||||
raise DouyuActivityError(payload.get("msg") or "赠送精英令失败")
|
||||
return {"raw": payload}
|
||||
|
||||
def query_points(self, *, uid: str, ctn: str) -> dict[str, Any]:
|
||||
"""查询手册积分余额。"""
|
||||
payload = self._request_json(
|
||||
"post",
|
||||
self.CREDIT_BALANCE_API,
|
||||
"查询手册积分",
|
||||
json={"uid": str(uid), "ctn": ctn},
|
||||
headers={"Content-Type": "application/json;charset=UTF-8"},
|
||||
)
|
||||
data = payload.get("data") or {}
|
||||
return {"points": data.get("balance"), "raw": payload}
|
||||
|
||||
def query_points_diff(self, *, uid: str, ctn: str) -> dict[str, Any]:
|
||||
"""查询手册积分差额。"""
|
||||
payload = self._request_json(
|
||||
"post",
|
||||
self.CREDIT_DIFF_API,
|
||||
"查询手册积分差额",
|
||||
json={"userId": str(uid), "ctn": ctn},
|
||||
headers={"Content-Type": "application/json;charset=UTF-8"},
|
||||
)
|
||||
data = payload.get("data") or {}
|
||||
return {"diff": data.get("diff"), "raw": payload}
|
||||
|
||||
def list_goods(self, *, manual_id: str, rid: str) -> dict[str, Any]:
|
||||
"""刷新活动商品列表。"""
|
||||
payload = self._request_json(
|
||||
"get",
|
||||
self.GOODS_API,
|
||||
"刷新商品列表",
|
||||
params={"manualID": manual_id, "rid": rid},
|
||||
headers={
|
||||
"Origin": "",
|
||||
"Referer": f"https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0&roomId={rid}",
|
||||
},
|
||||
)
|
||||
goods = (payload.get("data") or {}).get("list") or []
|
||||
return {"goods": goods, "raw": payload}
|
||||
|
||||
def query_limited_goods(self, *, manual_id: str, rid: str) -> dict[str, Any]:
|
||||
"""查询当前账号限兑商品。"""
|
||||
result = self.list_goods(manual_id=manual_id, rid=rid)
|
||||
limited = [
|
||||
item for item in result["goods"]
|
||||
if str(item.get("status") or "") == "1"
|
||||
]
|
||||
return {"limited_goods": limited, "raw": result["raw"]}
|
||||
|
||||
def exchange_records(self, *, manual_id: str, limit: int = 100, offset: int = 0) -> dict[str, Any]:
|
||||
"""查询兑换记录。"""
|
||||
payload = self._request_json(
|
||||
"get",
|
||||
self.EXCHANGE_LIST_API,
|
||||
"查询兑换记录",
|
||||
params={"manualID": manual_id, "limit": limit, "offset": offset},
|
||||
headers={"Origin": "", "Referer": "https://www.douyu.com/"},
|
||||
)
|
||||
records = (payload.get("data") or {}).get("list") or []
|
||||
return {"records": records, "raw": payload}
|
||||
|
||||
def exchange_goods(self, *, manual_id: str, rid: str, commodity_id: str, ctn: str | None = None) -> dict[str, Any]:
|
||||
"""兑换活动商品。"""
|
||||
ctn_value = ctn or cookie_value(self.cookie, "acf_ccn") or self.acf_ccn(refresh_subscribe=False)
|
||||
token = self.csrf_token()
|
||||
randstr = "".join(random.choices(string.ascii_letters + string.digits, k=16))
|
||||
data = {
|
||||
"manualID": manual_id,
|
||||
"rid": rid,
|
||||
"commodityID": commodity_id,
|
||||
"ctn": ctn_value,
|
||||
"csrfToken": token,
|
||||
"randstr": randstr,
|
||||
"token[error][code]": "-1",
|
||||
"token[data]": "",
|
||||
}
|
||||
payload = self._request_json(
|
||||
"post",
|
||||
self.EXCHANGE_API,
|
||||
"兑换商品",
|
||||
data=data,
|
||||
headers={
|
||||
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
|
||||
"Referer": f"https://www.douyu.com/pages/live-peace-handbook/web/shop?ditchname=pass0&roomId={rid}",
|
||||
"sec-ch-ua": f'"Not_A Brand";v="8", "Chromium";v="120", "Google Chrome";v="120"',
|
||||
"sec-ch-ua-mobile": "?0",
|
||||
"sec-ch-ua-platform": '"Windows"',
|
||||
"Sec-Fetch-Site": "same-origin",
|
||||
"Sec-Fetch-Mode": "cors",
|
||||
"Sec-Fetch-Dest": "empty",
|
||||
},
|
||||
)
|
||||
if payload.get("error") not in (0, "0"):
|
||||
raise DouyuActivityError(payload.get("msg") or "兑换商品失败")
|
||||
return {
|
||||
"commodity_id": commodity_id,
|
||||
"ctn": ctn_value,
|
||||
"csrf_token": token,
|
||||
"randstr": randstr,
|
||||
"raw": payload,
|
||||
}
|
||||
|
||||
def gold_account(self) -> dict[str, Any]:
|
||||
"""查询 cz 侧鱼翅余额。"""
|
||||
payload = self._request_json(
|
||||
"get",
|
||||
self.GOLD_ACCOUNT_API,
|
||||
"查询鱼翅余额",
|
||||
headers={"Origin": "https://cz.douyu.com", "Referer": "https://cz.douyu.com/"},
|
||||
)
|
||||
data = payload.get("data") or {}
|
||||
return {"gold": data.get("gold"), "raw": payload}
|
||||
|
||||
def exchange_balance(self) -> dict[str, Any]:
|
||||
"""查询钱包兑换中心余额。"""
|
||||
payload = self._request_json(
|
||||
"get",
|
||||
self.EXCHANGE_BALANCE_API,
|
||||
"查询鱼刺余额",
|
||||
params={"appCode": "YJTX"},
|
||||
headers={
|
||||
"Origin": "",
|
||||
"Referer": "https://www.douyu.com/member/walletcenter",
|
||||
},
|
||||
)
|
||||
data = payload.get("data") or {}
|
||||
return {"count": data.get("count"), "raw": payload}
|
||||
@@ -0,0 +1,61 @@
|
||||
"""斗鱼 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 key:
|
||||
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_douyu_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 ""
|
||||
pairs = cookie_pairs(cookie) if isinstance(cookie, str) else 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
|
||||
Reference in New Issue
Block a user