"""斗鱼 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, }