358 lines
13 KiB
Python
358 lines
13 KiB
Python
"""
|
||
提供对 OverFast API (https://overfast-api.tekrop.fr) 的所有异步 HTTP 调用封装，
|
||
用于查询 Overwatch 玩家战绩、英雄信息等数据。
|
||
"""
|
||
|
||
import asyncio
|
||
import time
|
||
from typing import Any
|
||
from urllib.parse import quote
|
||
|
||
import aiohttp
|
||
|
||
try: # 统一使用 AstrBot 日志；独立单测时回退标准 logging
|
||
from astrbot.api import logger
|
||
except Exception: # pragma: no cover
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class OverFastAPIError(ValueError):
|
||
"""OverFast API 请求错误。
|
||
|
||
携带 HTTP 状态码、服务端返回的错误详情以及重试等待时间，
|
||
继承 ValueError 以兼容既有的错误捕获逻辑。
|
||
|
||
Attributes:
|
||
status_code: HTTP 状态码。
|
||
detail: 服务端返回的原始错误文本。
|
||
retry_after: 服务端建议的重试等待秒数（可能为 None）。
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
message: str,
|
||
status_code: int | None = None,
|
||
detail: str = "",
|
||
retry_after: int | None = None,
|
||
) -> None:
|
||
super().__init__(message)
|
||
self.status_code = status_code
|
||
self.detail = detail
|
||
self.retry_after = retry_after
|
||
|
||
|
||
# 进程级 GET 缓存：key -> (expire_ts, data)，降低限流风险
|
||
_CACHE: dict[str, tuple[float, Any]] = {}
|
||
|
||
|
||
def _cache_get(key: str) -> Any | None:
|
||
item = _CACHE.get(key)
|
||
if not item:
|
||
return None
|
||
expire, data = item
|
||
if expire < time.monotonic():
|
||
_CACHE.pop(key, None)
|
||
return None
|
||
return data
|
||
|
||
|
||
def _cache_set(key: str, data: Any, ttl: float) -> None:
|
||
if ttl <= 0:
|
||
return
|
||
# 简单容量保护
|
||
if len(_CACHE) > 512:
|
||
_CACHE.clear()
|
||
_CACHE[key] = (time.monotonic() + ttl, data)
|
||
|
||
|
||
def clear_cache() -> None:
|
||
_CACHE.clear()
|
||
|
||
|
||
class OverFastAPIClient:
|
||
"""OverFast API 异步客户端。
|
||
|
||
使用示例:
|
||
async with OverFastAPIClient() as client:
|
||
result = await client.search_players("TeKrop")
|
||
summary = await client.get_player_summary("TeKrop-2217")
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
base_url: str = "https://overfast-api.tekrop.fr",
|
||
timeout: float = 15.0,
|
||
) -> None:
|
||
"""初始化 OverFast API 客户端。
|
||
|
||
Args:
|
||
base_url: API 基础 URL，官方实例或自建实例（如 Docker 自部署）。
|
||
插件侧通过 `api_base_url` 配置项传入。
|
||
timeout: 请求超时时间（秒），默认 15 秒（聊天场景不宜过长）。
|
||
"""
|
||
self.base_url = base_url.rstrip("/")
|
||
self.timeout = aiohttp.ClientTimeout(total=timeout)
|
||
self._session: aiohttp.ClientSession | None = None
|
||
|
||
@property
|
||
def session(self) -> aiohttp.ClientSession:
|
||
"""获取当前 HTTP session。
|
||
|
||
Returns:
|
||
aiohttp.ClientSession 实例。
|
||
|
||
Raises:
|
||
RuntimeError: 如果 session 未初始化（未使用 async with）。
|
||
"""
|
||
if self._session is None or self._session.closed:
|
||
raise RuntimeError(
|
||
"HTTP session 未初始化。请使用 `async with OverFastAPIClient() as client:`"
|
||
)
|
||
return self._session
|
||
|
||
async def __aenter__(self) -> "OverFastAPIClient":
|
||
"""异步上下文管理器入口，创建 HTTP session。"""
|
||
self._session = aiohttp.ClientSession(timeout=self.timeout)
|
||
return self
|
||
|
||
async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
|
||
"""异步上下文管理器出口，确保 session 被关闭。"""
|
||
if self._session and not self._session.closed:
|
||
await self._session.close()
|
||
self._session = None
|
||
|
||
@staticmethod
|
||
def _quote_segment(seg: str) -> str:
|
||
return quote(seg, safe="-")
|
||
|
||
async def _request(
|
||
self,
|
||
endpoint: str,
|
||
params: dict | None = None,
|
||
*,
|
||
use_cache: bool = False,
|
||
cache_ttl: float = 300.0,
|
||
max_retries: int = 2,
|
||
) -> dict | list:
|
||
"""发送 GET 请求到 OverFast API（含缓存/限流重试）。
|
||
|
||
Args:
|
||
endpoint: API 端点路径（不含 base_url），如 `/players`。
|
||
params: 查询参数字典，值为 None 的键会被自动过滤。
|
||
use_cache: 是否启用进程级 GET 缓存。
|
||
cache_ttl: 缓存秒数。
|
||
max_retries: 429/503 时的最大重试次数。
|
||
|
||
Raises:
|
||
OverFastAPIError: 当返回 HTTP 4xx/5xx 错误时。
|
||
"""
|
||
filtered_params = {k: v for k, v in (params or {}).items() if v is not None}
|
||
url = f"{self.base_url}/{endpoint.lstrip('/')}"
|
||
logger.debug(f"OverFast API 请求: GET {url} params={filtered_params}")
|
||
|
||
cache_key = ""
|
||
if use_cache:
|
||
sorted_items = sorted((k, str(v)) for k, v in filtered_params.items())
|
||
cache_key = f"GET {url} {sorted_items}"
|
||
cached = _cache_get(cache_key)
|
||
if cached is not None:
|
||
logger.debug(f"OverFast API 缓存命中: {url}")
|
||
return cached
|
||
|
||
last_err: OverFastAPIError | None = None
|
||
for attempt in range(max_retries + 1):
|
||
async with self.session.get(url, params=filtered_params) as response:
|
||
if response.status >= 400:
|
||
error_text = ""
|
||
retry_after: int | None = None
|
||
try:
|
||
error_json = await response.json()
|
||
if isinstance(error_json, dict):
|
||
error_text = str(error_json.get("error", ""))
|
||
ra = error_json.get("retry_after")
|
||
if isinstance(ra, (int, float)):
|
||
retry_after = int(ra)
|
||
if not error_text and "detail" in error_json:
|
||
error_text = str(error_json["detail"])
|
||
if not error_text:
|
||
error_text = str(error_json)
|
||
else:
|
||
error_text = str(error_json)
|
||
except Exception:
|
||
error_text = await response.text() or f"HTTP {response.status}"
|
||
|
||
logger.debug(
|
||
f"OverFast API 错误: GET {url} params={filtered_params} "
|
||
f"-> HTTP {response.status}, error={error_text}, retry_after={retry_after}"
|
||
)
|
||
|
||
if response.status == 404:
|
||
raise OverFastAPIError(
|
||
f"未找到请求的资源: {error_text}",
|
||
status_code=404, detail=error_text, retry_after=retry_after,
|
||
)
|
||
if response.status in (429, 503) and attempt < max_retries:
|
||
wait = retry_after if retry_after else (2 ** attempt)
|
||
wait = min(max(wait, 1), 10)
|
||
logger.debug(f"OverFast API 限流，{wait}s 后重试 ({attempt + 1}/{max_retries})")
|
||
await asyncio.sleep(wait)
|
||
last_err = OverFastAPIError(
|
||
f"请求过于频繁: {error_text}",
|
||
status_code=response.status, detail=error_text,
|
||
retry_after=retry_after,
|
||
)
|
||
continue
|
||
if response.status == 429:
|
||
raise OverFastAPIError(
|
||
f"请求过于频繁，请稍后重试: {error_text}",
|
||
status_code=429, detail=error_text, retry_after=retry_after,
|
||
)
|
||
if response.status == 503:
|
||
raise OverFastAPIError(
|
||
f"服务暂时不可用（可能被限流）: {error_text}",
|
||
status_code=503, detail=error_text, retry_after=retry_after,
|
||
)
|
||
raise OverFastAPIError(
|
||
f"API 错误 (HTTP {response.status}): {error_text}",
|
||
status_code=response.status, detail=error_text, retry_after=retry_after,
|
||
)
|
||
|
||
try:
|
||
data = await response.json()
|
||
except aiohttp.ContentTypeError as e:
|
||
raw_text = await response.text()
|
||
raise ValueError(
|
||
f"无法解析 API 响应为 JSON: {e}. 原始响应: {raw_text[:500]}"
|
||
) from e
|
||
|
||
logger.debug(f"OverFast API 响应: {type(data).__name__}")
|
||
if use_cache:
|
||
_cache_set(cache_key, data, cache_ttl)
|
||
return data
|
||
|
||
assert last_err is not None
|
||
raise last_err
|
||
|
||
async def search_players(
|
||
self,
|
||
name: str,
|
||
order_by: str = "name:asc",
|
||
offset: int = 0,
|
||
limit: int = 20,
|
||
) -> dict:
|
||
"""搜索玩家。"""
|
||
return await self._request( # type: ignore[return-value]
|
||
"/players",
|
||
params={"name": name, "order_by": order_by, "offset": offset, "limit": limit},
|
||
use_cache=True,
|
||
cache_ttl=120,
|
||
)
|
||
|
||
async def get_player_summary(self, player_id: str) -> dict:
|
||
"""获取玩家摘要信息。"""
|
||
pid = self._quote_segment(player_id)
|
||
return await self._request(f"/players/{pid}/summary", use_cache=True, cache_ttl=120) # type: ignore[return-value]
|
||
|
||
async def get_player_full(self, player_id: str) -> dict:
|
||
"""获取玩家完整数据（摘要 + 统计数据）。"""
|
||
pid = self._quote_segment(player_id)
|
||
return await self._request(f"/players/{pid}", use_cache=True, cache_ttl=120) # type: ignore[return-value]
|
||
|
||
async def get_player_stats_summary(
|
||
self,
|
||
player_id: str,
|
||
gamemode: str | None = None,
|
||
platform: str | None = None,
|
||
) -> dict:
|
||
"""获取玩家统计摘要。"""
|
||
pid = self._quote_segment(player_id)
|
||
return await self._request( # type: ignore[return-value]
|
||
f"/players/{pid}/stats/summary",
|
||
params={"gamemode": gamemode, "platform": platform},
|
||
use_cache=True,
|
||
cache_ttl=120,
|
||
)
|
||
|
||
async def get_player_career_stats(
|
||
self,
|
||
player_id: str,
|
||
gamemode: str,
|
||
platform: str | None = None,
|
||
hero: str | None = None,
|
||
) -> dict:
|
||
"""获取玩家生涯统计（按英雄分类的详细数据）。"""
|
||
pid = self._quote_segment(player_id)
|
||
return await self._request( # type: ignore[return-value]
|
||
f"/players/{pid}/stats/career",
|
||
params={"gamemode": gamemode, "platform": platform, "hero": hero},
|
||
use_cache=True,
|
||
cache_ttl=120,
|
||
)
|
||
|
||
async def get_heroes_stats(
|
||
self,
|
||
platform: str,
|
||
gamemode: str,
|
||
region: str,
|
||
role: str | None = None,
|
||
map_key: str | None = None,
|
||
competitive_division: str | None = None,
|
||
order_by: str = "hero:asc",
|
||
) -> list:
|
||
"""获取英雄统计数据（全服英雄选取率/胜率排行榜）。
|
||
|
||
competitive_division 可选 bronze/silver/gold/platinum/emerald/diamond/master/grandmaster
|
||
（API 4.13 段位；ultimate 为玩家段位新值，榜单筛选暂不支持）。
|
||
"""
|
||
return await self._request( # type: ignore[return-value]
|
||
"/heroes/stats",
|
||
params={
|
||
"platform": platform,
|
||
"gamemode": gamemode,
|
||
"region": region,
|
||
"role": role,
|
||
"map": map_key,
|
||
"competitive_division": competitive_division,
|
||
"order_by": order_by,
|
||
},
|
||
use_cache=True,
|
||
cache_ttl=600,
|
||
)
|
||
|
||
async def get_hero_info(self, hero_key: str) -> dict:
|
||
"""获取英雄详细信息。"""
|
||
hk = self._quote_segment(hero_key.strip().lower())
|
||
return await self._request(f"/heroes/{hk}", use_cache=True, cache_ttl=3600) # type: ignore[return-value]
|
||
|
||
async def list_heroes(
|
||
self,
|
||
role: str | None = None,
|
||
gamemode: str | None = None,
|
||
) -> list:
|
||
"""获取英雄列表。"""
|
||
return await self._request( # type: ignore[return-value]
|
||
"/heroes",
|
||
params={"role": role, "gamemode": gamemode},
|
||
use_cache=True,
|
||
cache_ttl=3600,
|
||
)
|
||
|
||
async def list_roles(self) -> list:
|
||
"""获取所有角色类型。"""
|
||
return await self._request("/roles", use_cache=True, cache_ttl=3600) # type: ignore[return-value]
|
||
|
||
async def list_gamemodes(self) -> list:
|
||
"""获取所有游戏模式。"""
|
||
return await self._request("/gamemodes", use_cache=True, cache_ttl=3600) # type: ignore[return-value]
|
||
|
||
async def list_maps(self, gamemode: str | None = None) -> list:
|
||
"""获取所有地图。"""
|
||
return await self._request( # type: ignore[return-value]
|
||
"/maps",
|
||
params={"gamemode": gamemode},
|
||
use_cache=True,
|
||
cache_ttl=3600,
|
||
)
|