107 lines
3.5 KiB
Python
107 lines
3.5 KiB
Python
"""斗鱼 Cookie 有效性检测服务。"""
|
||
|
||
from datetime import UTC, datetime
|
||
|
||
import requests
|
||
|
||
CHECK_UA = (
|
||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 "
|
||
"(KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36"
|
||
)
|
||
FISH_BALL_API = "https://www.douyu.com/wjapi/nc/exchange/fishBall"
|
||
USER_LEVEL_API = "https://www.douyu.com/japi/interactnc/web/userLevel/userLevelDetail"
|
||
|
||
|
||
def check_douyu_cookie(cookie: str) -> dict:
|
||
"""检测斗鱼 Cookie 有效性，鱼丸与等级接口均通过才算有效。"""
|
||
checked_at = datetime.now(UTC)
|
||
base = {
|
||
"checked_at": checked_at,
|
||
"fish_ball": None,
|
||
"nickname": None,
|
||
"level": None,
|
||
}
|
||
if not cookie:
|
||
return {**base, "valid": False, "message": "Cookie 为空"}
|
||
|
||
fish_ok = False
|
||
fish_msg = ""
|
||
fish_ball = None
|
||
try:
|
||
fish_data = requests.get(
|
||
FISH_BALL_API,
|
||
params={"appCode": "YJTX"},
|
||
headers={
|
||
"User-Agent": CHECK_UA,
|
||
"Accept": "application/json, text/plain, */*",
|
||
"Referer": "https://www.douyu.com/member/walletcenter",
|
||
"Cookie": cookie,
|
||
},
|
||
timeout=(5, 10),
|
||
).json()
|
||
if isinstance(fish_data, dict) and fish_data.get("error") in (0, "0"):
|
||
fish_ok = True
|
||
fish_ball = (
|
||
(fish_data.get("data") or {}).get("count")
|
||
if isinstance(fish_data.get("data"), dict)
|
||
else None
|
||
)
|
||
else:
|
||
fish_msg = (
|
||
str(fish_data.get("msg") or fish_data.get("error") or "响应异常")
|
||
if isinstance(fish_data, dict)
|
||
else "响应异常"
|
||
)
|
||
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
|
||
fish_msg = f"请求失败: {exc}"
|
||
|
||
level_ok = False
|
||
level_msg = ""
|
||
nickname = None
|
||
level = None
|
||
try:
|
||
level_data = requests.get(
|
||
USER_LEVEL_API,
|
||
params={"rid": "0"},
|
||
headers={
|
||
"User-Agent": CHECK_UA,
|
||
"Accept": "application/json, text/plain, */*",
|
||
"Referer": "https://www.douyu.com/pages/ord-user-level?clientType=web",
|
||
"Cookie": cookie,
|
||
},
|
||
timeout=(5, 10),
|
||
).json()
|
||
if isinstance(level_data, dict) and level_data.get("error") in (0, "0"):
|
||
level_ok = True
|
||
info_raw = level_data.get("data")
|
||
info = info_raw if isinstance(info_raw, dict) else {}
|
||
nickname = str(info.get("nn") or "") or None
|
||
level = info.get("lv")
|
||
else:
|
||
level_msg = (
|
||
str(level_data.get("msg") or level_data.get("error") or "响应异常")
|
||
if isinstance(level_data, dict)
|
||
else "响应异常"
|
||
)
|
||
except Exception as exc: # noqa: BLE001 外部接口与任务边界需要保留宽泛异常兜底
|
||
level_msg = f"请求失败: {exc}"
|
||
|
||
valid = fish_ok and level_ok
|
||
if valid:
|
||
message = "有效"
|
||
else:
|
||
message = "；".join(
|
||
[
|
||
f"鱼丸接口: {'ok' if fish_ok else (fish_msg or '失败')}",
|
||
f"等级接口: {'ok' if level_ok else (level_msg or '失败')}",
|
||
]
|
||
)
|
||
return {
|
||
**base,
|
||
"valid": valid,
|
||
"message": message[:200],
|
||
"fish_ball": fish_ball,
|
||
"nickname": nickname,
|
||
"level": level,
|
||
}
|