chesscom-mxcp

""" Chess.com player-related tools. """ from typing import Dict, Any, List from chess_client import make_api_request, get_cache_ttl from mxcp.runtime import db import json import logging logger = logging.getLogger(__name__) async def get_player_profile(username: str) -> Dict[str, Any]: """ Get a player's profile information from Chess.com. Args: username: The Chess.com username Returns: Player profile data including join date, status, followers, etc. """ endpoint = f"player/{username}" cache_ttl = get_cache_ttl("player_profile") try: data = await make_api_request(endpoint, cache_ttl=cache_ttl) # Store in analytics table for trending try: db.execute(""" INSERT INTO player_profile_views (username, viewed_at, follower_count, country) VALUES ($username, CURRENT_TIMESTAMP, $followers, $country) """, { "username": username, "followers": data.get("followers", 0), "country": data.get("country", "") }) except Exception as e: logger.debug(f"Could not store analytics: {e}") return data except Exception as e: return {"error": f"Failed to fetch player profile: {str(e)}"} async def get_player_stats(username: str) -> Dict[str, Any]: """ Get a player's chess statistics from Chess.com. Args: username: The Chess.com username Returns: Player statistics for different game types (chess_blitz, chess_rapid, etc.) """ endpoint = f"player/{username}/stats" cache_ttl = get_cache_ttl("player_stats") try: data = await make_api_request(endpoint, cache_ttl=cache_ttl) # Store rating history for analytics for game_type, stats in data.items(): if isinstance(stats, dict) and "last" in stats: rating_data = stats["last"] db.execute(""" INSERT INTO rating_history (username, game_type, rating, games_played, recorded_at) VALUES ($username, $game_type, $rating, $games, CURRENT_TIMESTAMP) """, { "username": username, "game_type": game_type, "rating": rating_data.get("rating", 0), "games": stats.get("record", {}).get("win", 0) + stats.get("record", {}).get("loss", 0) + stats.get("record", {}).get("draw", 0) }) return data except Exception as e: return {"error": f"Failed to fetch player stats: {str(e)}"} async def is_player_online(username: str) -> Dict[str, bool]: """ Check if a player is currently online on Chess.com. Args: username: The Chess.com username Returns: Dictionary with 'online' boolean value """ endpoint = f"player/{username}/is-online" cache_ttl = get_cache_ttl("current_games") # Short TTL for online status try: data = await make_api_request(endpoint, cache_ttl=cache_ttl) return data except Exception as e: return {"online": False, "error": str(e)} async def get_player_current_games(username: str) -> Dict[str, Any]: """ Get a list of a player's current games on Chess.com. Args: username: The Chess.com username Returns: List of ongoing games """ endpoint = f"player/{username}/games" cache_ttl = get_cache_ttl("current_games") try: data = await make_api_request(endpoint, cache_ttl=cache_ttl) # Store game count for activity tracking game_count = len(data.get("games", [])) db.execute(""" INSERT INTO player_activity (username, game_count, checked_at) VALUES ($username, $count, CURRENT_TIMESTAMP) """, {"username": username, "count": game_count}) return data except Exception as e: return {"games": [], "error": str(e)} async def get_player_games_by_month(username: str, year: int, month: int) -> Dict[str, Any]: """ Get a player's games for a specific month from Chess.com. Args: username: The Chess.com username year: Year (YYYY format) month: Month (1-12) Returns: List of games for the specified month """ month_str = str(month).zfill(2) endpoint = f"player/{username}/games/{year}/{month_str}" # Check if we have cached games for this month cached_games = db.execute(""" SELECT game_data FROM chess_games_cache WHERE username = $username AND year = $year AND month = $month ORDER BY cached_at DESC """, {"username": username, "year": year, "month": month}) if cached_games: # Return cached games games = [game["game_data"] for game in cached_games] return {"games": games} try: data = await make_api_request(endpoint, cache_ttl=None) # Don't use regular cache # Cache individual games games = data.get("games", []) for game in games: game_id = f"{game.get('white', {}).get('username', '')}_{game.get('black', {}).get('username', '')}_{game.get('end_time', '')}" db.execute(""" INSERT OR REPLACE INTO chess_games_cache (game_id, username, game_data, cached_at, year, month) VALUES ($game_id, $username, $game_data, CURRENT_TIMESTAMP, $year, $month) """, { "game_id": game_id, "username": username, "game_data": game, "year": year, "month": month }) return data except Exception as e: return {"games": [], "error": str(e)} async def get_player_game_archives(username: str) -> Dict[str, List[str]]: """ Get a list of available monthly game archives for a player on Chess.com. Args: username: The Chess.com username Returns: List of archive URLs """ endpoint = f"player/{username}/games/archives" cache_ttl = get_cache_ttl("game_archives") try: data = await make_api_request(endpoint, cache_ttl=cache_ttl) return data except Exception as e: return {"archives": [], "error": str(e)} async def download_player_games_pgn(username: str, year: int, month: int) -> str: """ Download PGN files for all games in a specific month from Chess.com. Args: username: The Chess.com username year: Year (YYYY format) month: Month (1-12) Returns: Multi-game PGN format text containing all games for the month """ month_str = str(month).zfill(2) endpoint = f"player/{username}/games/{year}/{month_str}/pgn" # Check PGN cache cached_pgn = db.execute(""" SELECT pgn_data FROM chess_games_cache WHERE username = $username AND year = $year AND month = $month AND pgn_data IS NOT NULL LIMIT 1 """, {"username": username, "year": year, "month": month}) if cached_pgn: return cached_pgn[0]["pgn_data"] try: pgn_data = await make_api_request(endpoint, accept_json=False) # Cache the PGN data db.execute(""" UPDATE chess_games_cache SET pgn_data = $pgn WHERE username = $username AND year = $year AND month = $month """, { "pgn": pgn_data, "username": username, "year": year, "month": month }) return pgn_data except Exception as e: return f"Error downloading PGN data: {str(e)}"