Add services module: implemented ProfileService, GameService, and AnalysisService for user data, game management, and strategy analysis. Enhanced with schema-aware response handling and LLM-friendly output formatting.

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)