diff --git a/src/geoguessr_mcp/services/__init__.py b/src/geoguessr_mcp/services/__init__.py index e69de29..d7d1faf 100644 --- a/src/geoguessr_mcp/services/__init__.py +++ b/src/geoguessr_mcp/services/__init__.py @@ -0,0 +1,12 @@ +"""Services module for business logic.""" + +from .profile_service import ProfileService +from .game_service import GameService +from .analysis_service import AnalysisService, GameAnalysis + +__all__ = [ + "ProfileService", + "GameService", + "AnalysisService", + "GameAnalysis", +] \ No newline at end of file diff --git a/src/geoguessr_mcp/services/analysis_service.py b/src/geoguessr_mcp/services/analysis_service.py index f90f40e..2391497 100644 --- a/src/geoguessr_mcp/services/analysis_service.py +++ b/src/geoguessr_mcp/services/analysis_service.py @@ -1,30 +1,334 @@ -"""Analysis and statistics calculations.""" +""" +Analysis service for game statistics and strategy optimization. -from typing import List +This service provides comprehensive analysis capabilities with +dynamic data handling and LLM-friendly output formatting. +""" -from ..models.game import Game +import logging +from dataclasses import dataclass, field +from typing import Optional + +from ..api.client import GeoGuessrClient +from ..models.Game import Game +from ..monitoring.schema_manager import schema_registry +from .game_service import GameService +from .profile_service import ProfileService + +logger = logging.getLogger(__name__) + + +@dataclass +class GameAnalysis: + """Analysis results for a set of games.""" + games_analyzed: int = 0 + total_score: int = 0 + average_score: float = 0.0 + total_rounds: int = 0 + perfect_rounds: int = 0 + perfect_round_percentage: float = 0.0 + average_distance_meters: float = 0.0 + average_time_seconds: float = 0.0 + best_game_score: int = 0 + worst_game_score: int = 0 + score_trend: str = "stable" # improving, declining, stable + weak_areas: list = field(default_factory=list) + strong_areas: list = field(default_factory=list) + + def to_dict(self) -> dict: + """Convert to dictionary.""" + return { + "games_analyzed": self.games_analyzed, + "total_score": self.total_score, + "average_score": round(self.average_score, 2), + "total_rounds": self.total_rounds, + "perfect_rounds": self.perfect_rounds, + "perfect_round_percentage": round(self.perfect_round_percentage, 2), + "average_distance_meters": round(self.average_distance_meters, 2), + "average_time_seconds": round(self.average_time_seconds, 2), + "best_game_score": self.best_game_score, + "worst_game_score": self.worst_game_score, + "score_trend": self.score_trend, + "weak_areas": self.weak_areas, + "strong_areas": self.strong_areas, + } class AnalysisService: - """Service for analyzing game data.""" + """Service for game analysis and strategy optimization.""" + + def __init__( + self, + client: GeoGuessrClient, + game_service: Optional[GameService] = None, + profile_service: Optional[ProfileService] = None, + ): + self.client = client + self.game_service = game_service or GameService(client) + self.profile_service = profile_service or ProfileService(client) @staticmethod - def calculate_statistics(games: List[Game]) -> dict: - """Calculate aggregate statistics from games.""" + def analyze_games(games: list[Game]) -> GameAnalysis: + """ + Analyze a list of games and calculate statistics. + + Args: + games: List of Game objects to analyze + + Returns: + GameAnalysis with computed statistics + """ if not games: - return {"games_analyzed": 0, "total_score": 0, "average_score": 0, "perfect_rounds": 0} + return GameAnalysis() total_score = sum(g.total_score for g in games) - total_rounds = sum(len(g.rounds) for g in games) - perfect_rounds = sum(1 for g in games for r in g.rounds if r.score == 5000) + all_rounds = [r for g in games for r in g.rounds] + total_rounds = len(all_rounds) + perfect_rounds = sum(1 for r in all_rounds if r.score == 5000) - return { - "games_analyzed": len(games), - "total_score": total_score, - "average_score": total_score / len(games), - "total_rounds": total_rounds, - "perfect_rounds": perfect_rounds, - "perfect_round_percentage": ( + # Calculate averages + avg_distance = ( + sum(r.distance_meters for r in all_rounds) / total_rounds + if total_rounds > 0 + else 0 + ) + avg_time = ( + sum(r.time_seconds for r in all_rounds) / total_rounds + if total_rounds > 0 + else 0 + ) + + # Find best and worst + scores = [g.total_score for g in games] + best_score = max(scores) if scores else 0 + worst_score = min(scores) if scores else 0 + + # Determine trend (simple moving average comparison) + trend = "stable" + if len(games) >= 4: + first_half = sum(g.total_score for g in games[: len(games) // 2]) / ( + len(games) // 2 + ) + second_half = sum(g.total_score for g in games[len(games) // 2 :]) / ( + len(games) - len(games) // 2 + ) + if second_half > first_half * 1.05: + trend = "improving" + elif second_half < first_half * 0.95: + trend = "declining" + + # Identify weak/strong areas based on scores + weak_areas = [] + strong_areas = [] + + for game in games: + for round_guess in game.rounds: + if round_guess.score < 2000: + weak_areas.append({ + "game": game.token, + "round": round_guess.round_number, + "score": round_guess.score, + "distance": round_guess.distance_meters, + }) + elif round_guess.score >= 4500: + strong_areas.append({ + "game": game.token, + "round": round_guess.round_number, + "score": round_guess.score, + }) + + return GameAnalysis( + games_analyzed=len(games), + total_score=total_score, + average_score=total_score / len(games), + total_rounds=total_rounds, + perfect_rounds=perfect_rounds, + perfect_round_percentage=( (perfect_rounds / total_rounds * 100) if total_rounds > 0 else 0 ), + average_distance_meters=avg_distance, + average_time_seconds=avg_time, + best_game_score=best_score, + worst_game_score=worst_score, + score_trend=trend, + weak_areas=weak_areas[:10], # Limit to 10 + strong_areas=strong_areas[:10], + ) + + async def analyze_recent_games( + self, + count: int = 10, + session_token: Optional[str] = None, + ) -> dict: + """ + Analyze recent games and provide statistics summary. + + Args: + count: Number of recent games to analyze + session_token: Optional session token + + Returns: + Dictionary with analysis results and raw game data + """ + games = await self.game_service.get_recent_games(count, session_token) + analysis = self.analyze_games(games) + + return { + "analysis": analysis.to_dict(), + "games": [g.to_dict() for g in games], + "schema_info": { + "endpoints_used": ["/v4/feed/private", "/v3/games/{token}"], + "available_schemas": schema_registry.get_available_endpoints(), + }, } + + async def get_performance_summary( + self, + session_token: Optional[str] = None, + ) -> dict: + """ + Get a comprehensive performance summary. + + Combines profile stats, achievements, season info, and recent game analysis. + """ + results = { + "profile": None, + "stats": None, + "season": None, + "recent_games_analysis": None, + "explorer": None, + "objectives": None, + "api_status": schema_registry.get_schema_summary(), + "errors": [], + } + + # Get comprehensive profile + try: + results["profile"] = await self.profile_service.get_comprehensive_profile( + session_token + ) + except Exception as e: + results["errors"].append(f"Profile: {str(e)}") + + # Get season stats + try: + stats, response = await self.game_service.get_season_stats(session_token) + results["season"] = { + "data": { + "rank": stats.rank, + "rating": stats.rating, + "games_played": stats.games_played, + "division": stats.division, + }, + "raw_fields": response.available_fields, + } + except Exception as e: + results["errors"].append(f"Season: {str(e)}") + + # Analyze recent games + try: + results["recent_games_analysis"] = await self.analyze_recent_games( + 5, session_token + ) + except Exception as e: + results["errors"].append(f"Recent games: {str(e)}") + + # Get explorer progress + try: + response = await self.client.get( + self._create_endpoint("/v3/explorer"), session_token + ) + if response.is_success: + results["explorer"] = response.summarize() + except Exception as e: + results["errors"].append(f"Explorer: {str(e)}") + + # Get objectives + try: + response = await self.client.get( + self._create_endpoint("/v4/objectives"), session_token + ) + if response.is_success: + results["objectives"] = response.summarize() + except Exception as e: + results["errors"].append(f"Objectives: {str(e)}") + + return results + + async def get_strategy_recommendations( + self, + session_token: Optional[str] = None, + ) -> dict: + """ + Generate strategy recommendations based on performance analysis. + + This method analyzes the user's gameplay patterns and provides + actionable recommendations for improvement. + """ + # Get recent games for analysis + games = await self.game_service.get_recent_games(20, session_token) + analysis = self.analyze_games(games) + + recommendations = [] + + # Analyze perfect round rate + if analysis.perfect_round_percentage < 20: + recommendations.append({ + "category": "accuracy", + "priority": "high", + "recommendation": "Focus on improving pinpoint accuracy", + "detail": f"Your perfect round rate is {analysis.perfect_round_percentage:.1f}%. " + "Practice with familiar maps to build confidence.", + }) + + # Analyze time usage + if analysis.average_time_seconds < 30: + recommendations.append({ + "category": "time_management", + "priority": "medium", + "recommendation": "Consider taking more time per round", + "detail": f"Average time: {analysis.average_time_seconds:.0f}s. " + "Taking a bit more time can improve accuracy.", + }) + + # Analyze score trend + if analysis.score_trend == "declining": + recommendations.append({ + "category": "consistency", + "priority": "high", + "recommendation": "Your scores are trending downward", + "detail": "Consider taking breaks and reviewing your weak areas.", + }) + + # Check for weak areas pattern + if len(analysis.weak_areas) > 5: + recommendations.append({ + "category": "practice", + "priority": "medium", + "recommendation": "Practice specific regions", + "detail": f"You had {len(analysis.weak_areas)} rounds under 2000 points. " + "Consider using region-specific practice maps.", + }) + + return { + "analysis_summary": { + "games_analyzed": analysis.games_analyzed, + "average_score": round(analysis.average_score, 0), + "trend": analysis.score_trend, + "perfect_rate": f"{analysis.perfect_round_percentage:.1f}%", + }, + "recommendations": recommendations, + "data_sources": { + "endpoints_used": schema_registry.get_available_endpoints(), + "last_updated": schema_registry.get_schema_summary() + .get("endpoints", {}) + .get("/v4/feed/private", {}) + .get("last_updated"), + }, + } + + @staticmethod + def _create_endpoint(path: str): + """Create simple endpoint info for raw requests.""" + from ..api.endpoints import EndpointInfo + return EndpointInfo(path=path, description=f"Request to {path}") diff --git a/src/geoguessr_mcp/services/competitive_service.py b/src/geoguessr_mcp/services/competitive_service.py deleted file mode 100644 index e69de29..0000000 diff --git a/src/geoguessr_mcp/services/game_service.py b/src/geoguessr_mcp/services/game_service.py index e69de29..7028b77 100644 --- a/src/geoguessr_mcp/services/game_service.py +++ b/src/geoguessr_mcp/services/game_service.py @@ -0,0 +1,188 @@ +""" +Game service for game data operations. + +Handles game history, details, and competitive data with dynamic schema support. +""" + +import logging +from typing import Optional + +from ..api.client import GeoGuessrClient, DynamicResponse +from ..api.endpoints import Endpoints +from ..models.Game import Game +from ..models.SeasonStats import SeasonStats +from ..models.DailyChallenge import DailyChallenge + +logger = logging.getLogger(__name__) + + +class GameService: + """Service for game-related operations.""" + + def __init__(self, client: GeoGuessrClient): + self.client = client + + async def get_game_details( + self, + game_token: str, + session_token: Optional[str] = None, + ) -> tuple[Game, DynamicResponse]: + """ + Get details for a specific game. + + Args: + game_token: The game token/ID + session_token: Optional session token + + Returns: + Tuple of (Game, DynamicResponse) + """ + endpoint = Endpoints.GAMES.get_game_details(game_token) + response = await self.client.get(endpoint, session_token) + + if response.is_success: + game = Game.from_api_response(response.data) + return game, response + + raise ValueError(f"Failed to get game details: {response.data}") + + async def get_unfinished_games( + self, + session_token: Optional[str] = None, + ) -> DynamicResponse: + """Get list of unfinished games.""" + return await self.client.get(Endpoints.GAMES.GET_UNFINISHED_GAMES, session_token) + + async def get_streak_game( + self, + game_token: str, + session_token: Optional[str] = None, + ) -> DynamicResponse: + """Get streak game details.""" + endpoint = Endpoints.GAMES.get_streak_game(game_token) + return await self.client.get(endpoint, session_token) + + async def get_activity_feed( + self, + count: int = 10, + page: int = 0, + session_token: Optional[str] = None, + ) -> DynamicResponse: + """ + Get the activity feed. + + Args: + count: Number of items to fetch + page: Page number for pagination + session_token: Optional session token + + Returns: + DynamicResponse with feed data + """ + endpoint = Endpoints.SOCIAL.get_activity_feed(count, page) + return await self.client.get(endpoint, session_token) + + async def get_recent_games( + self, + count: int = 10, + session_token: Optional[str] = None, + ) -> list[Game]: + """ + Get recent games from the activity feed. + + Args: + count: Number of games to retrieve + session_token: Optional session token + + Returns: + List of Game objects + """ + feed_response = await self.get_activity_feed(count * 2, 0, session_token) + + if not feed_response.is_success: + return [] + + games = [] + entries = feed_response.data.get("entries", []) + + for entry in entries: + if len(games) >= count: + break + + entry_type = entry.get("type", "") + if entry_type in ["PlayedGame", "FinishedGame", "game"]: + payload = entry.get("payload", entry) + game_token = payload.get("gameToken", payload.get("token")) + + if game_token: + try: + game, _ = await self.get_game_details(game_token, session_token) + games.append(game) + except Exception as e: + logger.warning(f"Failed to fetch game {game_token}: {e}") + + return games + + async def get_season_stats( + self, + session_token: Optional[str] = None, + ) -> tuple[SeasonStats, DynamicResponse]: + """Get active season statistics.""" + response = await self.client.get( + Endpoints.COMPETITIVE.GET_ACTIVE_SEASON_STATS, session_token + ) + + if response.is_success: + stats = SeasonStats.from_api_response(response.data) + return stats, response + + raise ValueError(f"Failed to get season stats: {response.data}") + + async def get_daily_challenge( + self, + day: str = "today", + session_token: Optional[str] = None, + ) -> tuple[DailyChallenge, DynamicResponse]: + """ + Get daily challenge. + + Args: + day: "today", "yesterday", or specific date + session_token: Optional session token + + Returns: + Tuple of (DailyChallenge, DynamicResponse) + """ + endpoint = Endpoints.CHALLENGES.get_daily_challenge(day) + response = await self.client.get(endpoint, session_token) + + if response.is_success: + challenge = DailyChallenge.from_api_response(response.data) + return challenge, response + + raise ValueError(f"Failed to get daily challenge: {response.data}") + + async def get_battle_royale( + self, + game_id: str, + session_token: Optional[str] = None, + ) -> DynamicResponse: + """Get battle royale game details.""" + endpoint = Endpoints.GAME_SERVER.get_battle_royale(game_id) + return await self.client.get(endpoint, session_token) + + async def get_duel( + self, + duel_id: str, + session_token: Optional[str] = None, + ) -> DynamicResponse: + """Get duel game details.""" + endpoint = Endpoints.GAME_SERVER.get_duel(duel_id) + return await self.client.get(endpoint, session_token) + + async def get_tournaments( + self, + session_token: Optional[str] = None, + ) -> DynamicResponse: + """Get tournament information.""" + return await self.client.get(Endpoints.GAME_SERVER.GET_TOURNAMENTS, session_token) diff --git a/src/geoguessr_mcp/services/profile_service.py b/src/geoguessr_mcp/services/profile_service.py index 3fe471a..232a9d0 100644 --- a/src/geoguessr_mcp/services/profile_service.py +++ b/src/geoguessr_mcp/services/profile_service.py @@ -1,136 +1,184 @@ """ -Profile-related business logic. +Profile service for user data operations. + +This service handles profile, stats, and achievement data with +dynamic schema adaptation. """ +import logging from typing import Optional -from ..api.client import GeoguessrClient + +from ..api.client import GeoGuessrClient, DynamicResponse from ..api.endpoints import Endpoints -from ..models.profile import UserProfile, UserStats +from ..models.Achievement import Achievement +from ..models.UserProfile import UserProfile +from ..models.UserStats import UserStats + +logger = logging.getLogger(__name__) class ProfileService: - """Service for profile operations.""" + """Service for profile-related operations.""" - def __init__(self, client: GeoguessrClient): - """ - Initialize the profile service. - - Args: - client: GeoGuessr API client - """ + def __init__(self, client: GeoGuessrClient): self.client = client async def get_profile( self, - session_token: Optional[str] = None - ) -> UserProfile: + session_token: Optional[str] = None, + ) -> tuple[UserProfile, DynamicResponse]: """ - Get user profile. - - Args: - session_token: Optional session token for authentication + Get current user's profile. Returns: - UserProfile with user information - - Raises: - httpx.HTTPError: If the API request fails + Tuple of (UserProfile, DynamicResponse) for both structured and raw access """ - response = await self.client.get( - Endpoints.PROFILES.GET_PROFILE, - session_token - ) - data = response.json() - return UserProfile.from_api_response(data) + response = await self.client.get(Endpoints.PROFILES.GET_PROFILE, session_token) + + if response.is_success: + profile = UserProfile.from_api_response(response.data) + return profile, response + + raise ValueError(f"Failed to get profile: {response.data}") async def get_stats( self, - session_token: Optional[str] = None - ) -> UserStats: + session_token: Optional[str] = None, + ) -> tuple[UserStats, DynamicResponse]: """ Get user statistics. - Args: - session_token: Optional session token for authentication - Returns: - UserStats with user statistics - - Raises: - httpx.HTTPError: If the API request fails + Tuple of (UserStats, DynamicResponse) """ - response = await self.client.get( - Endpoints.PROFILES.GET_STATS, - session_token - ) - data = response.json() - return UserStats.from_api_response(data) + response = await self.client.get(Endpoints.PROFILES.GET_STATS, session_token) + + if response.is_success: + stats = UserStats.from_api_response(response.data) + return stats, response + + raise ValueError(f"Failed to get stats: {response.data}") async def get_extended_stats( self, - session_token: Optional[str] = None - ) -> dict: + session_token: Optional[str] = None, + ) -> DynamicResponse: """ - Get extended user statistics. + Get extended statistics. - Args: - session_token: Optional session token for authentication - - Returns: - Dictionary with extended statistics - - Raises: - httpx.HTTPError: If the API request fails + Returns raw DynamicResponse as extended stats have variable schema. """ - response = await self.client.get( - Endpoints.PROFILES.GET_EXTENDED_STATS, - session_token - ) - return response.json() + return await self.client.get(Endpoints.PROFILES.GET_EXTENDED_STATS, session_token) async def get_achievements( self, - session_token: Optional[str] = None - ) -> list: + session_token: Optional[str] = None, + ) -> tuple[list[Achievement], DynamicResponse]: """ Get user achievements. - Args: - session_token: Optional session token for authentication - Returns: - List of achievement dictionaries - - Raises: - httpx.HTTPError: If the API request fails + Tuple of (list of Achievement, DynamicResponse) """ - response = await self.client.get( - Endpoints.PROFILES.GET_ACHIEVEMENTS, - session_token - ) - return response.json() + response = await self.client.get(Endpoints.PROFILES.GET_ACHIEVEMENTS, session_token) + + if response.is_success: + achievements = [] + data = response.data + + # Handle different response formats + if isinstance(data, list): + achievements = [Achievement.from_api_response(a) for a in data] + elif isinstance(data, dict) and "achievements" in data: + achievements = [Achievement.from_api_response(a) for a in data["achievements"]] + + return achievements, response + + raise ValueError(f"Failed to get achievements: {response.data}") async def get_public_profile( self, user_id: str, - session_token: Optional[str] = None - ) -> UserProfile: + session_token: Optional[str] = None, + ) -> tuple[UserProfile, DynamicResponse]: + """Get another user's public profile.""" + endpoint = Endpoints.PROFILES.get_public_profile(user_id) + response = await self.client.get(endpoint, session_token) + + if response.is_success: + profile = UserProfile.from_api_response(response.data) + return profile, response + + raise ValueError(f"Failed to get public profile: {response.data}") + + async def get_user_maps( + self, + session_token: Optional[str] = None, + ) -> DynamicResponse: + """Get user's custom maps.""" + return await self.client.get(Endpoints.PROFILES.GET_USER_MAPS, session_token) + + async def get_comprehensive_profile( + self, + session_token: Optional[str] = None, + ) -> dict: """ - Get public profile of another user. + Get a comprehensive profile combining multiple endpoints. - Args: - user_id: User ID to fetch - session_token: Optional session token for authentication - - Returns: - UserProfile with public user information - - Raises: - httpx.HTTPError: If the API request fails + This method aggregates data from multiple sources and provides + a unified view with schema information for the LLM. """ - response = await self.client.get( - Endpoints.PROFILES.get_public_profile(user_id), - session_token - ) - data = response.json() - return UserProfile.from_api_response(data) + results = { + "profile": None, + "stats": None, + "extended_stats": None, + "achievements": None, + "schema_info": {}, + "errors": [], + } + + # Get profile + try: + profile, response = await self.get_profile(session_token) + results["profile"] = profile.to_dict() + results["schema_info"]["profile"] = response.available_fields + except Exception as e: + results["errors"].append(f"Profile: {str(e)}") + + # Get stats + try: + stats, response = await self.get_stats(session_token) + results["stats"] = stats.to_dict() + results["schema_info"]["stats"] = response.available_fields + except Exception as e: + results["errors"].append(f"Stats: {str(e)}") + + # Get extended stats + try: + response = await self.get_extended_stats(session_token) + if response.is_success: + results["extended_stats"] = response.summarize() + results["schema_info"]["extended_stats"] = response.available_fields + except Exception as e: + results["errors"].append(f"Extended stats: {str(e)}") + + # Get achievements summary + try: + achievements, response = await self.get_achievements(session_token) + unlocked = [a for a in achievements if a.unlocked] + results["achievements"] = { + "total": len(achievements), + "unlocked": len(unlocked), + "recent": [ + {"name": a.name, "unlocked_at": a.unlocked_at} + for a in sorted( + unlocked, + key=lambda x: x.unlocked_at or "", + reverse=True + )[:5] + ], + } + except Exception as e: + results["errors"].append(f"Achievements: {str(e)}") + + return results