619c00aa06
All checks were successful
CI/CD Pipeline / build-and-deploy (push) Successful in 12s
A gamer's periodic check would get permanently stuck if their stored token was revoked/expired: our stats API collapsed both "Lichess rejected the token" (401/403, permanent) and genuine transient errors into the same 502 response, so the bot treated an invalid token exactly like a network blip — retrying the same window forever at a capped 300s backoff, never advancing the checkpoint (observed in prod: Dor1zz stuck for 100+ consecutive errors over 8+ hours, admin alerts firing every 25 failures). Preserve the distinction that already existed one layer down (lichess_client.py already tells 401/403 apart from other failures) instead of collapsing it in stats_service.py: add PuzzleOfPeriodResponse.auth_failed, have main.py return 401 specifically for that case, and have the bot raise a distinct InvalidTokenError instead of returning None. On InvalidTokenError, the bot now clears the token for that pair, notifies the user to reconnect via /addtoken, and continues tracking games normally instead of stalling forever.
208 lines
No EOL
9.4 KiB
Python
208 lines
No EOL
9.4 KiB
Python
import aiohttp
|
|
import logging
|
|
from typing import Optional, Dict, Any
|
|
from config import LICHESS_API_BASE_URL, LICHESS_STATS_API_BASE_URL
|
|
from rate_limiter import get_rate_limiter
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class InvalidTokenError(Exception):
|
|
"""Raised when Lichess rejects a stored token (401) — permanent, not worth retrying."""
|
|
pass
|
|
|
|
|
|
class LichessAPI:
|
|
def __init__(self):
|
|
self.lichess_base_url = LICHESS_API_BASE_URL
|
|
self.stats_base_url = LICHESS_STATS_API_BASE_URL
|
|
self.rate_limiter = get_rate_limiter()
|
|
|
|
async def get_user_profile(self, token: str) -> Optional[Dict[str, Any]]:
|
|
"""Get user profile from Lichess API using token"""
|
|
await self.rate_limiter.wait_if_needed()
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
f"{self.lichess_base_url}/account",
|
|
headers=headers
|
|
) as response:
|
|
if response.status == 200:
|
|
return await response.json()
|
|
else:
|
|
error_text = await response.text()
|
|
logger.error(f"Failed to get user profile: {response.status} - {error_text}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error getting user profile: {e}")
|
|
import traceback
|
|
logger.error(traceback.format_exc())
|
|
return None
|
|
|
|
async def get_today_stats(self, username: str) -> Optional[Dict[str, Any]]:
|
|
"""Get today's statistics from our stats API"""
|
|
logger.info(f"🔍 LichessAPI.get_today_stats: username={username}, stats_base_url={self.stats_base_url}")
|
|
await self.rate_limiter.wait_if_needed()
|
|
url = f"{self.stats_base_url}/stats/{username}/today"
|
|
logger.info(f"🔍 Making request to: {url}")
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(url) as response:
|
|
logger.info(f"🔍 Response status: {response.status} for {username}")
|
|
if response.status == 200:
|
|
result = await response.json()
|
|
logger.info(f"🔍 Successfully got stats for {username}: {result.get('message', 'no message')}")
|
|
return result
|
|
else:
|
|
error_text = await response.text()
|
|
logger.error(f"❌ Failed to get today stats for {username}: status={response.status}, error={error_text[:200]}")
|
|
return None
|
|
except aiohttp.ClientError as e:
|
|
logger.error(f"❌ Client error getting today stats for {username}: {e}")
|
|
import traceback
|
|
logger.error(traceback.format_exc())
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"❌ Error getting today stats for {username}: {e}")
|
|
import traceback
|
|
logger.error(traceback.format_exc())
|
|
return None
|
|
|
|
async def get_yesterday_stats(self, username: str) -> Optional[Dict[str, Any]]:
|
|
"""Get yesterday's statistics from our stats API"""
|
|
await self.rate_limiter.wait_if_needed()
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
f"{self.stats_base_url}/stats/{username}/yesterday"
|
|
) as response:
|
|
if response.status == 200:
|
|
return await response.json()
|
|
else:
|
|
logger.error(f"Failed to get yesterday stats: {response.status}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error getting yesterday stats: {e}")
|
|
return None
|
|
|
|
async def get_week_stats(self, username: str) -> Optional[Dict[str, Any]]:
|
|
"""Get week's statistics from our stats API"""
|
|
await self.rate_limiter.wait_if_needed()
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
f"{self.stats_base_url}/stats/{username}/week"
|
|
) as response:
|
|
if response.status == 200:
|
|
return await response.json()
|
|
else:
|
|
logger.error(f"Failed to get week stats: {response.status}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error getting week stats: {e}")
|
|
return None
|
|
|
|
async def get_games_period(self, username: str, since: int, until: int, rated_only: Optional[bool] = None, include_games: bool = False) -> Optional[Dict[str, Any]]:
|
|
"""Get games for a specific period"""
|
|
await self.rate_limiter.wait_if_needed()
|
|
try:
|
|
url = f"{self.stats_base_url}/games/{username}/period"
|
|
params = {"since": since, "until": until}
|
|
if rated_only is not None:
|
|
params["rated_only"] = "true" if rated_only else "false"
|
|
if include_games:
|
|
params["include_games"] = "true"
|
|
logger.info(f"🔍 LichessAPI.get_games_period: URL={url}, params={params}")
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(url, params=params) as response:
|
|
logger.info(f"🔍 LichessAPI.get_games_period: response.status={response.status}")
|
|
if response.status == 200:
|
|
result = await response.json()
|
|
logger.info(f"🔍 LichessAPI.get_games_period: result={result}")
|
|
return result
|
|
else:
|
|
logger.error(f"Failed to get games period: {response.status}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error getting games period: {e}")
|
|
return None
|
|
|
|
async def get_puzzles_period(self, token: str, since: int, until: int, max_puzzles: int = 150) -> Optional[Dict[str, Any]]:
|
|
"""
|
|
Get puzzles for a specific period.
|
|
|
|
Raises InvalidTokenError on a 401 (Lichess rejected the token — permanent,
|
|
the caller should stop retrying with this token) instead of returning None,
|
|
so callers can tell that apart from a transient failure (network error,
|
|
Lichess hiccup) where retrying the same window later makes sense.
|
|
"""
|
|
await self.rate_limiter.wait_if_needed()
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
f"{self.stats_base_url}/puzzle/period",
|
|
headers=headers,
|
|
params={"since": since, "until": until, "max": max_puzzles}
|
|
) as response:
|
|
if response.status == 200:
|
|
return await response.json()
|
|
elif response.status == 401:
|
|
logger.warning(f"Puzzles period: token rejected by Lichess (401)")
|
|
raise InvalidTokenError("Lichess rejected the token")
|
|
else:
|
|
logger.error(f"Failed to get puzzles period: {response.status}")
|
|
return None
|
|
except InvalidTokenError:
|
|
raise
|
|
except Exception as e:
|
|
logger.error(f"Error getting puzzles period: {e}")
|
|
return None
|
|
|
|
async def check_user_exists(self, username: str) -> Optional[bool]:
|
|
"""
|
|
Check if user exists on Lichess.
|
|
|
|
Returns True/False for a definitive answer (200/404), or None if the
|
|
check itself failed (rate limit, network error, unexpected status) —
|
|
callers must not treat None as "not found", since the username may
|
|
well be valid.
|
|
"""
|
|
await self.rate_limiter.wait_if_needed()
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
f"{self.lichess_base_url}/user/{username}"
|
|
) as response:
|
|
if response.status == 200:
|
|
return True
|
|
elif response.status == 404:
|
|
logger.warning(f"User {username} not found on Lichess (404)")
|
|
return False
|
|
else:
|
|
logger.error(f"Failed to check user existence: {response.status}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error checking user existence: {e}")
|
|
return None
|
|
|
|
async def get_user_ratings(self, username: str) -> Optional[Dict[str, Any]]:
|
|
"""Get user ratings from Lichess API"""
|
|
await self.rate_limiter.wait_if_needed()
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(
|
|
f"{self.lichess_base_url}/user/{username}"
|
|
) as response:
|
|
if response.status == 200:
|
|
return await response.json()
|
|
else:
|
|
logger.error(f"Failed to get user ratings: {response.status}")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error getting user ratings: {e}")
|
|
return None |