""" 懂球帝 API 客户端 - 支持 aiohttp / curl_cffi 双通道 - 自动指纹轮换、频率控制、重试 """ import asyncio import random import time from typing import Dict, Any, Optional from urllib.parse import urlencode import aiohttp from loguru import logger from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type from src.core.config import get_config from src.core.exceptions import ( NetworkError, RateLimitError, BlockedError, CaptchaError, ParsingError, ) from src.anti_detect.fingerprint import FingerprintPool class APIClient: """懂球帝 API 客户端 (aiohttp 通道)""" def __init__(self, fingerprint_pool: Optional[FingerprintPool] = None): self._cfg = get_config() self._fp_pool = fingerprint_pool self._session: Optional[aiohttp.ClientSession] = None self._request_count = 0 self._created_at: Optional[float] = None self._stats = { "total": 0, "success": 0, "fail": 0, "rate_limit": 0, "blocked": 0, "total_time": 0.0, } async def initialize(self): if self._session is None or self._should_rotate(): await self._new_session() async def _new_session(self): await self.close() timeout = aiohttp.ClientTimeout(total=self._cfg.crawler.request_timeout) self._session = aiohttp.ClientSession(timeout=timeout) self._request_count = 0 self._created_at = time.time() def _should_rotate(self) -> bool: scfg = self._cfg.crawler.session if self._request_count >= scfg.max_requests_per_session: return True if self._created_at and (time.time() - self._created_at) > scfg.rotate_interval: return True return False async def close(self): if self._session: await self._session.close() self._session = None async def _get_headers(self) -> tuple: """获取请求头，返回 (headers, fp_id)""" if self._fp_pool: fp = await self._fp_pool.acquire() return fp.get("http_headers", {}), fp.get("id") return { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", "Accept": "application/json, text/plain, */*", "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8", "Referer": "https://www.dongqiudi.com/", "Origin": "https://www.dongqiudi.com", }, None async def request( self, url: str, params: Optional[Dict[str, Any]] = None, proxy: Optional[str] = None, ) -> Dict[str, Any]: await self.initialize() headers, fp_id = await self._get_headers() self._request_count += 1 self._stats["total"] += 1 start = time.time() full_url = f"{url}?{urlencode(params)}" if params else url logger.info(f"[api] GET {full_url}") try: async with self._session.get( url, params=params, headers=headers, proxy=proxy, allow_redirects=True, ssl=False, ) as resp: elapsed = time.time() - start self._stats["total_time"] += elapsed return await self._handle_response(resp, url, fp_id, elapsed, params) except (RateLimitError, BlockedError, CaptchaError): self._stats["fail"] += 1 raise except aiohttp.ClientError as e: self._stats["fail"] += 1 if fp_id and self._fp_pool: await self._fp_pool.mark_failure(fp_id, "network") raise NetworkError(f"网络错误: {e}") from e except Exception as e: self._stats["fail"] += 1 raise NetworkError(f"请求异常: {e}") from e async def _handle_response( self, resp: aiohttp.ClientResponse, url: str, fp_id: Optional[str], elapsed: float, params: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: status = resp.status full_url = f"{url}?{urlencode(params)}" if params else url if status == 200: try: data = await resp.json(content_type=None) except Exception: text = await resp.text() raise ParsingError(f"JSON 解析失败，原始响应: {text[:500]}") resp_summary = str(data)[:300] if data else '(empty)' logger.info(f"[api] 200 OK {elapsed:.2f}s {full_url} resp={resp_summary}") self._stats["success"] += 1 if fp_id and self._fp_pool: await self._fp_pool.mark_success(fp_id) return data elif status == 429: logger.warning(f"[api] 429 RateLimit {elapsed:.2f}s {full_url}") self._stats["rate_limit"] += 1 retry_after = int(resp.headers.get("Retry-After", 60)) if fp_id and self._fp_pool: await self._fp_pool.mark_failure(fp_id, "rate_limit") raise RateLimitError(retry_after=retry_after) elif status == 403: logger.warning(f"[api] 403 Forbidden {elapsed:.2f}s {full_url}") self._stats["blocked"] += 1 if fp_id and self._fp_pool: await self._fp_pool.mark_failure(fp_id, "ip_blocked") raise BlockedError(f"403: {url}") elif status == 503: text = await resp.text() logger.warning(f"[api] 503 {elapsed:.2f}s {full_url}") if "captcha" in text.lower() or "验证码" in text: if fp_id and self._fp_pool: await self._fp_pool.mark_failure(fp_id, "captcha") raise CaptchaError("验证码") raise NetworkError(f"503: {url}") else: logger.warning(f"[api] HTTP {status} {elapsed:.2f}s {full_url}") raise NetworkError(f"HTTP {status}: {url}") # ── 懂球帝专用 API 方法 ── def _default_params(self, extra: Optional[Dict] = None) -> Dict[str, Any]: dp = self._cfg.dongqiudi.default_params params = {"app": dp.app, "platform": dp.platform, "version": dp.version, "language": dp.language} if extra: params.update(extra) return params async def get_standings(self, season_id: int) -> Dict[str, Any]: url = self._cfg.dongqiudi.base_url + self._cfg.dongqiudi.api.standings params = self._default_params({"season_id": season_id}) return await self.request(url, params) async def get_schedule(self, season_id: int, round_number: Optional[int] = None) -> Dict[str, Any]: url = self._cfg.dongqiudi.base_url + self._cfg.dongqiudi.api.schedule extra = {"season_id": season_id} if round_number: extra["round"] = round_number return await self.request(url, self._default_params(extra)) async def get_match_detail(self, match_id: int) -> Dict[str, Any]: url = self._cfg.dongqiudi.base_url + self._cfg.dongqiudi.api.match_detail return await self.request(url, self._default_params({"match_id": match_id})) async def get_match_menu(self) -> Dict[str, Any]: url = self._cfg.dongqiudi.web_url + self._cfg.dongqiudi.api.match_menu params = {"mark": "gif", "platform": "www", "version": "713"} return await self.request(url, params) async def get_live_matches(self) -> Dict[str, Any]: """从各联赛 tab API 汇总当前比赛数据""" all_matches = [] league_ids = set() for t in self._cfg.dongqiudi.targets: if not t.active: continue lid = self._get_league_id(t.league) if lid and lid not in league_ids: league_ids.add(lid) try: url = f"https://api.dongqiudi.com/data/tab/league/new/{lid}" raw = await self.request(url, None) items = raw.get("list", []) if isinstance(raw, dict) else [] all_matches.extend(items) except Exception as e: logger.warning(f"获取联赛 {t.league} 实时数据失败: {e}") return {"data": {"matches": all_matches}} _LEAGUE_ID_MAP = { "中超": 43, "中甲": 129, "亚冠": 226, "英超": 4, "西甲": 3, "德甲": 5, "意甲": 9, "法甲": 12, "欧冠": 6, "欧联": 14, } def _get_league_id(self, league_name: str) -> Optional[int]: return self._LEAGUE_ID_MAP.get(league_name) async def get_news(self, tab_id: int = 1, size: int = 30) -> Dict[str, Any]: """通过 tabs API 获取资讯列表""" url = f"{self._cfg.dongqiudi.web_url}/api/app/tabs/web/{tab_id}.json" params = {"size": size} if size != 20 else None return await self.request(url, params) async def get_video_list(self, tab_id: int = 233) -> Dict[str, Any]: """抓取视频列表页 HTML，返回 {'html': str, 'tab_id': int}""" await self.initialize() url = f"{self._cfg.dongqiudi.web_url}/videoList/{tab_id}" headers, fp_id = await self._get_headers() headers["Accept"] = "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8" self._request_count += 1 self._stats["total"] += 1 try: async with self._session.get(url, headers=headers, ssl=False) as resp: if resp.status == 200: html = await resp.text() self._stats["success"] += 1 return {"html": html, "tab_id": tab_id} raise NetworkError(f"HTTP {resp.status}: {url}") except aiohttp.ClientError as e: self._stats["fail"] += 1 raise NetworkError(f"网络错误: {e}") from e async def get_article_detail(self, article_id: int) -> Dict[str, Any]: """抓取文章详情页 HTML，返回 {'html': str}""" await self.initialize() url = f"{self._cfg.dongqiudi.web_url}/articles/{article_id}.html" headers, fp_id = await self._get_headers() headers["Accept"] = "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8" self._request_count += 1 self._stats["total"] += 1 try: async with self._session.get(url, headers=headers, ssl=False) as resp: if resp.status == 200: html = await resp.text() self._stats["success"] += 1 return {"html": html, "article_id": article_id} raise NetworkError(f"HTTP {resp.status}: {url}") except aiohttp.ClientError as e: self._stats["fail"] += 1 raise NetworkError(f"网络错误: {e}") from e @property def stats(self) -> Dict[str, Any]: total = self._stats["total"] return { **self._stats, "success_rate": self._stats["success"] / total if total else 1.0, "avg_time": self._stats["total_time"] / total if total else 0.0, } async def __aenter__(self): await self.initialize() return self async def __aexit__(self, *args): await self.close()