LichessStatTgWeb/LichessClientTG_bot/lichess_api.py

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, priority: str = "background") -> Optional[Dict[str, Any]]:
        """
        Get games for a specific period.

        priority="interactive" (e.g. the accuracy fetch behind /today, /yesterday,
        /week) jumps ahead of "background" (periodic_check) in the stats
        service's shared Lichess-token queue, so an on-demand command doesn't
        wait behind the whole periodic-check backlog.
        """
        await self.rate_limiter.wait_if_needed()
        try:
            url = f"{self.stats_base_url}/games/{username}/period"
            params = {"since": since, "until": until, "priority": priority}
            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