yvachot/GeoGuessrMCP
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Code cleanup: standardized imports, refined formatting for consistency, and resolved minor redundancies in services, models, monitoring, and tools modules.

Browse source
This commit is contained in:
Yûki VACHOT 2025-11-29 00:49:36 +01:00
parent e486d78e31
commit ec0fe38861
39 changed files with 222 additions and 239 deletions
Download patch file Download diff file Expand all files Collapse all files

4
src/geoguessr_mcp/__init__.py
View file

@ -8,6 +8,6 @@ with automatic API monitoring and dynamic schema adaptation.
__version__ = "0.2.0"
__author__ = "Yûki VACHOT"
from .main import mcp, main
from .main import main, mcp
__all__ = ["mcp", "main", "__version__"]
__all__ = ["mcp", "main", "__version__"]

6
src/geoguessr_mcp/api/__init__.py
View file

@ -1,7 +1,7 @@
"""API client module for GeoGuessr communication."""
from .client import GeoGuessrClient, DynamicResponse
from .endpoints import Endpoints, EndpointInfo, EndpointBuilder
from .client import DynamicResponse, GeoGuessrClient
from .endpoints import EndpointBuilder, EndpointInfo, Endpoints
__all__ = [
"GeoGuessrClient",
@ -9,4 +9,4 @@ __all__ = [
"Endpoints",
"EndpointInfo",
"EndpointBuilder",
]
]

13
src/geoguessr_mcp/api/client.py
View file

@ -96,6 +96,7 @@ class DynamicResponse:
This reduces token usage while providing essential information.
"""
def summarize_value(value: Any, depth: int) -> Any:
if depth <= 0:
if isinstance(value, (dict, list)):
@ -103,10 +104,7 @@ class DynamicResponse:
return value
if isinstance(value, dict):
return {
k: summarize_value(v, depth - 1)
for k, v in list(value.items())[:10]
}
return {k: summarize_value(v, depth - 1) for k, v in list(value.items())[:10]}
if isinstance(value, list):
if len(value) == 0:
return []
@ -163,11 +161,7 @@ class GeoGuessrClient:
@staticmethod
def _get_base_url(endpoint: EndpointInfo) -> str:
"""Get the appropriate base URL for an endpoint."""
return (
settings.GAME_SERVER_URL
if endpoint.use_game_server
else settings.GEOGUESSR_API_URL
)
return settings.GAME_SERVER_URL if endpoint.use_game_server else settings.GEOGUESSR_API_URL
async def request(
self,
@ -199,6 +193,7 @@ class GeoGuessrClient:
logger.debug(f"{endpoint.method} {url}")
import time
start_time = time.time()
async with await self._get_authenticated_client(session_token) as client:

21
src/geoguessr_mcp/api/endpoints.py
View file

@ -13,6 +13,7 @@ from ..config import settings
@dataclass
class EndpointInfo:
"""Metadata about an API endpoint."""
path: str
method: str = "GET"
description: str = ""
@ -31,6 +32,7 @@ class Endpoints:
class AUTH:
"""Authentication endpoints."""
SIGNIN = EndpointInfo(
path="/v3/accounts/signin",
method="POST",
@ -40,6 +42,7 @@ class Endpoints:
class PROFILES:
"""User profile and stats endpoints."""
GET_PROFILE = EndpointInfo(
path="/v3/profiles",
description="Get current user profile",
@ -79,6 +82,7 @@ class Endpoints:
class GAMES:
"""Game-related endpoints."""
GET_UNFINISHED_GAMES = EndpointInfo(
path="/v3/social/events/unfinishedgames",
description="Get unfinished games",
@ -102,6 +106,7 @@ class Endpoints:
class GAME_SERVER:
"""Game server endpoints (different base URL)."""
GET_TOURNAMENTS = EndpointInfo(
path="/tournaments",
use_game_server=True,
@ -137,6 +142,7 @@ class Endpoints:
class COMPETITIVE:
"""Competitive and season-related endpoints."""
GET_ACTIVE_SEASON_STATS = EndpointInfo(
path="/v4/seasons/active/stats",
description="Get active season statistics",
@ -171,6 +177,7 @@ class Endpoints:
class SOCIAL:
"""Social and friends endpoints."""
GET_FRIENDS_SUMMARY = EndpointInfo(
path="/v3/social/friends/summary",
description="Get friends summary",
@ -194,9 +201,7 @@ class Endpoints:
)
@staticmethod
def get_friends_activities(
time_frame: str = "week", limit: int = 20
) -> EndpointInfo:
def get_friends_activities(time_frame: str = "week", limit: int = 20) -> EndpointInfo:
"""Get friends' activities."""
return EndpointInfo(
path="/v3/social/friends/activities",
@ -206,6 +211,7 @@ class Endpoints:
class MAPS:
"""Map-related endpoints."""
GET_PERSONALIZED_MAPS = EndpointInfo(
path="/v3/social/maps/browse/personalized",
description="Get personalized maps",
@ -243,6 +249,7 @@ class Endpoints:
class EXPLORER:
"""Explorer mode endpoints."""
GET_PROGRESS = EndpointInfo(
path="/v3/explorer",
description="Get explorer mode progress",
@ -250,6 +257,7 @@ class Endpoints:
class OBJECTIVES:
"""Objectives and rewards endpoints."""
GET_OBJECTIVES = EndpointInfo(
path="/v4/objectives",
description="Get current objectives",
@ -261,6 +269,7 @@ class Endpoints:
class SUBSCRIPTION:
"""Subscription-related endpoints."""
GET_INFO = EndpointInfo(
path="/v3/subscriptions",
description="Get subscription details",
@ -281,11 +290,7 @@ class EndpointBuilder:
Returns:
Complete URL string
"""
base = (
settings.GAME_SERVER_URL
if endpoint.use_game_server
else settings.GEOGUESSR_API_URL
)
base = settings.GAME_SERVER_URL if endpoint.use_game_server else settings.GEOGUESSR_API_URL
return f"{base}{endpoint.path}"
@staticmethod

5
src/geoguessr_mcp/auth/__init__.py
View file

@ -1,7 +1,8 @@
"""Auth module for GeoGuessr session."""
from .session import UserSession, SessionManager
from .session import SessionManager, UserSession
__all__ = [
"UserSession",
"SessionManager",
]
]

4
src/geoguessr_mcp/auth/session.py
View file

@ -213,9 +213,7 @@ class SessionManager:
try:
async with httpx.AsyncClient(timeout=30.0) as client:
client.cookies.set("_ncfa", cookie, domain=settings.GEOGUESSR_DOMAIN_NAME)
response = await client.get(
f"{settings.GEOGUESSR_API_URL}/v3/profiles"
)
response = await client.get(f"{settings.GEOGUESSR_API_URL}/v3/profiles")
if response.status_code == 200:
return response.json()
except Exception as e:

3
src/geoguessr_mcp/main.py
View file

@ -84,8 +84,7 @@ def main():
logger.info("Default authentication cookie configured from environment")
else:
logger.warning(
"No default authentication cookie set. "
"Users will need to login or provide a cookie."
"No default authentication cookie set. " "Users will need to login or provide a cookie."
)
# Run the server

1
src/geoguessr_mcp/models/Achievement.py
View file

@ -7,6 +7,7 @@ from typing import Optional
@dataclass
class Achievement:
"""Represents a user achievement."""
id: str
name: str
description: str = ""

5
src/geoguessr_mcp/models/DailyChallenge.py
View file

@ -7,6 +7,7 @@ from typing import Optional
@dataclass
class DailyChallenge:
"""Daily challenge information."""
token: str
map_name: str = ""
date: str = ""
@ -20,7 +21,9 @@ class DailyChallenge:
"""Create DailyChallenge from API response."""
return cls(
token=data.get("token", data.get("challengeToken", "")),
map_name=data.get("map", {}).get("name", "") if isinstance(data.get("map"), dict) else "",
map_name=(
data.get("map", {}).get("name", "") if isinstance(data.get("map"), dict) else ""
),
date=data.get("date", data.get("day", "")),
time_limit=data.get("timeLimit", 0),
completed=data.get("completed", data.get("played", False)),

1
src/geoguessr_mcp/models/Game.py
View file

@ -9,6 +9,7 @@ from .RoundGuess import RoundGuess
@dataclass
class Game:
"""Represents a complete game."""
token: str
map_name: str
mode: str

1
src/geoguessr_mcp/models/RoundGuess.py
View file

@ -7,6 +7,7 @@ from typing import Optional
@dataclass
class RoundGuess:
"""Represents a single round guess in a game."""
round_number: int
score: int
distance_meters: float

1
src/geoguessr_mcp/models/SeasonStats.py
View file

@ -6,6 +6,7 @@ from dataclasses import dataclass, field
@dataclass
class SeasonStats:
"""Competitive season statistics."""
season_id: str
season_name: str = ""
rank: int = 0

5
src/geoguessr_mcp/models/UserProfile.py
View file

@ -7,6 +7,7 @@ from typing import Optional
@dataclass
class UserProfile:
"""User profile information."""
id: str
nick: str
email: str = ""
@ -30,7 +31,9 @@ class UserProfile:
created=data.get("created", data.get("createdAt", "")),
is_verified=data.get("isVerified", data.get("verified", False)),
is_pro=data.get("isPro", data.get("isProUser", False)),
avatar_url=data.get("pin", {}).get("url") if isinstance(data.get("pin"), dict) else None,
avatar_url=(
data.get("pin", {}).get("url") if isinstance(data.get("pin"), dict) else None
),
raw_data=data,
)

1
src/geoguessr_mcp/models/UserStats.py
View file

@ -6,6 +6,7 @@ from dataclasses import dataclass, field
@dataclass
class UserStats:
"""User statistics from various endpoints."""
games_played: int = 0
rounds_played: int = 0
total_score: int = 0

13
src/geoguessr_mcp/models/__init__.py
View file

@ -1,12 +1,13 @@
"""Data models for GeoGuessr."""
from .UserProfile import UserProfile
from .UserStats import UserStats
from .RoundGuess import RoundGuess
from Game import Game
from Achievement import Achievement
from SeasonStats import SeasonStats
from DailyChallenge import DailyChallenge
from Game import Game
from SeasonStats import SeasonStats
from .RoundGuess import RoundGuess
from .UserProfile import UserProfile
from .UserStats import UserStats
__all__ = [
"UserProfile",
@ -16,4 +17,4 @@ __all__ = [
"Achievement",
"SeasonStats",
"DailyChallenge",
]
]

8
src/geoguessr_mcp/monitoring/__init__.py
View file

@ -1,9 +1,11 @@
"""Monitoring module for API endpoint tracking and schema detection."""
from .endpoint.EndpointMonitor import EndpointMonitor, endpoint_monitor, MONITORED_ENDPOINTS
from schema.EndpointSchema import EndpointSchema
from schema.SchemaRegistry import SchemaRegistry, schema_registry
from schema.SchemaDetector import SchemaDetector, SchemaField
from schema.SchemaRegistry import SchemaRegistry, schema_registry
from .endpoint.EndpointMonitor import (MONITORED_ENDPOINTS, EndpointMonitor,
endpoint_monitor)
__all__ = [
"EndpointMonitor",
@ -14,4 +16,4 @@ __all__ = [
"EndpointSchema",
"SchemaField",
"schema_registry",
]
]

2
src/geoguessr_mcp/monitoring/endpoint/EndpointDefinition.py
View file

@ -8,9 +8,11 @@ requirement, and additional parameters.
from dataclasses import dataclass, field
@dataclass
class EndpointDefinition:
"""Definition of an API endpoint to monitor."""
path: str
method: str = "GET"
requires_auth: bool = True

50
src/geoguessr_mcp/monitoring/endpoint/EndpointMonitor.py
View file

@ -13,15 +13,15 @@ Classes:
import asyncio
import logging
from datetime import datetime, UTC
from datetime import UTC, datetime
from typing import Optional
import httpx
from ...config import settings
from ..schema.SchemaRegistry import SchemaRegistry, schema_registry
from .EndpointDefinition import EndpointDefinition
from .EndpointMonitoringResult import MonitoringResult
from ..schema.SchemaRegistry import SchemaRegistry, schema_registry
logger = logging.getLogger(__name__)
@ -48,13 +48,11 @@ MONITORED_ENDPOINTS = [
path="/v3/profiles/maps",
description="User's custom maps",
),
# Game endpoints
EndpointDefinition(
path="/v3/social/events/unfinishedgames",
description="Unfinished games",
),
# Social endpoints
EndpointDefinition(
path="/v4/feed/private",
@ -73,19 +71,16 @@ MONITORED_ENDPOINTS = [
path="/v3/social/maps/browse/personalized",
description="Personalized map recommendations",
),
# Competitive endpoints
EndpointDefinition(
path="/v4/seasons/active/stats",
description="Active season statistics",
),
# Explorer endpoints
EndpointDefinition(
path="/v3/explorer",
description="Explorer mode progress",
),
# Objectives endpoints
EndpointDefinition(
path="/v4/objectives",
@ -95,19 +90,16 @@ MONITORED_ENDPOINTS = [
path="/v4/objectives/unclaimed",
description="Unclaimed objective rewards",
),
# Subscription endpoints
EndpointDefinition(
path="/v3/subscriptions",
description="Subscription information",
),
# Challenge endpoints
EndpointDefinition(
path="/v3/challenges/daily-challenges/today",
description="Today's daily challenge",
),
# Game server endpoints
EndpointDefinition(
path="/tournaments",
@ -116,6 +108,7 @@ MONITORED_ENDPOINTS = [
),
]
class EndpointMonitor:
"""
Monitors API endpoints for availability and schema changes.
@ -125,9 +118,9 @@ class EndpointMonitor:
"""
def __init__(
self,
registry: Optional[SchemaRegistry] = None,
ncfa_cookie: Optional[str] = None,
self,
registry: Optional[SchemaRegistry] = None,
ncfa_cookie: Optional[str] = None,
):
self.registry = registry or schema_registry
self.ncfa_cookie = ncfa_cookie or settings.DEFAULT_NCFA_COOKIE
@ -136,9 +129,9 @@ class EndpointMonitor:
self._task: Optional[asyncio.Task] = None
async def check_endpoint(
self,
endpoint: EndpointDefinition,
client: httpx.AsyncClient,
self,
endpoint: EndpointDefinition,
client: httpx.AsyncClient,
) -> MonitoringResult:
"""
Check a single endpoint and update its schema.
@ -151,9 +144,7 @@ class EndpointMonitor:
MonitoringResult with check details
"""
base_url = (
settings.GAME_SERVER_URL
if endpoint.use_game_server
else settings.GEOGUESSR_API_URL
settings.GAME_SERVER_URL if endpoint.use_game_server else settings.GEOGUESSR_API_URL
)
url = f"{base_url}{endpoint.path}"
@ -264,14 +255,16 @@ class EndpointMonitor:
except Exception as e:
logger.error(f"Error checking {endpoint.path}: {e}")
results.append(MonitoringResult(
endpoint=endpoint.path,
is_available=False,
response_code=0,
response_time_ms=0,
schema_changed=False,
error_message=str(e),
))
results.append(
MonitoringResult(
endpoint=endpoint.path,
is_available=False,
response_code=0,
response_time_ms=0,
schema_changed=False,
error_message=str(e),
)
)
self.results = results
return results
@ -334,8 +327,7 @@ class EndpointMonitor:
changed = [r for r in self.results if r.schema_changed]
avg_response_time = (
sum(r.response_time_ms for r in available) / len(available)
if available else 0
sum(r.response_time_ms for r in available) / len(available) if available else 0
)
return {

3
src/geoguessr_mcp/monitoring/endpoint/EndpointMonitoringResult.py
View file

@ -7,13 +7,14 @@ an endpoint, including its availability, response time, and any errors encounter
"""
from dataclasses import dataclass, field
from datetime import datetime, UTC
from datetime import UTC, datetime
from typing import Optional
@dataclass
class MonitoringResult:
"""Result of monitoring an endpoint."""
endpoint: str
is_available: bool
response_code: int

3
src/geoguessr_mcp/monitoring/schema/EndpointSchema.py
View file

@ -10,7 +10,7 @@ schema information.
import logging
from dataclasses import dataclass, field
from datetime import datetime, UTC
from datetime import UTC, datetime
from typing import Any, Optional
from .SchemaField import SchemaField
@ -21,6 +21,7 @@ logger = logging.getLogger(__name__)
@dataclass
class EndpointSchema:
"""Schema definition for an API endpoint."""
endpoint: str
method: str
fields: dict[str, SchemaField] = field(default_factory=dict)

14
src/geoguessr_mcp/monitoring/schema/SchemaDetector.py
View file

@ -60,7 +60,8 @@ class SchemaDetector:
def _is_uuid(value: str) -> bool:
"""Check if string is UUID format."""
import re
uuid_pattern = r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$'
uuid_pattern = r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$"
return bool(re.match(uuid_pattern, value.lower()))
@staticmethod
@ -86,13 +87,7 @@ class SchemaDetector:
self._analyze_object(data, fields, "", max_depth)
return fields
def _analyze_object(
self,
obj: dict,
fields: dict,
prefix: str,
remaining_depth: int
) -> None:
def _analyze_object(self, obj: dict, fields: dict, prefix: str, remaining_depth: int) -> None:
"""Recursively analyze an object and extract field information."""
if remaining_depth <= 0:
return
@ -121,7 +116,6 @@ class SchemaDetector:
def compute_schema_hash(fields: dict[str, SchemaField]) -> str:
"""Compute a hash of the schema for change detection."""
schema_repr = json.dumps(
{name: (f.field_type, f.nullable) for name, f in sorted(fields.items())},
sort_keys=True
{name: (f.field_type, f.nullable) for name, f in sorted(fields.items())}, sort_keys=True
)
return hashlib.sha256(schema_repr.encode()).hexdigest()[:16]

1
src/geoguessr_mcp/monitoring/schema/SchemaField.py
View file

@ -12,6 +12,7 @@ from typing import Any, Optional
@dataclass
class SchemaField:
"""Represents a single field in a schema."""
name: str
field_type: str
nullable: bool = False

32
src/geoguessr_mcp/monitoring/schema/SchemaRegistry.py
View file

@ -12,11 +12,10 @@ Classes:
import json
import logging
from datetime import datetime, UTC
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, Optional
from ...config import settings
from .EndpointSchema import EndpointSchema
from .SchemaDetector import SchemaDetector
@ -78,9 +77,7 @@ class SchemaRegistry:
try:
with open(self._get_schema_file(), "w") as f:
json.dump(
{ep: schema.to_dict() for ep, schema in self.schemas.items()},
f,
indent=2
{ep: schema.to_dict() for ep, schema in self.schemas.items()}, f, indent=2
)
with open(self._get_history_file(), "w") as f:
@ -90,17 +87,13 @@ class SchemaRegistry:
for ep, history in self.schema_history.items()
},
f,
indent=2
indent=2,
)
except Exception as e:
logger.error(f"Failed to save schemas: {e}")
def update_schema(
self,
endpoint: str,
response_data: Any,
response_code: int = 200,
method: str = "GET"
self, endpoint: str, response_data: Any, response_code: int = 200, method: str = "GET"
) -> tuple[EndpointSchema, bool]:
"""
Update schema for an endpoint based on response data.
@ -143,12 +136,7 @@ class SchemaRegistry:
return new_schema, schema_changed
def mark_unavailable(
self,
endpoint: str,
error_message: str,
response_code: int = 0
) -> None:
def mark_unavailable(self, endpoint: str, error_message: str, response_code: int = 0) -> None:
"""Mark an endpoint as unavailable."""
if endpoint in self.schemas:
self.schemas[endpoint].is_available = False
@ -191,7 +179,7 @@ class SchemaRegistry:
"response_code": schema.response_code,
}
for endpoint, schema in self.schemas.items()
}
},
}
def generate_dynamic_description(self, endpoint: str) -> str:
@ -228,14 +216,10 @@ class SchemaRegistry:
"""Truncate sample response for storage."""
if isinstance(data, dict):
return {
k: SchemaRegistry._truncate_sample(v, max_items)
for k, v in list(data.items())[:20]
k: SchemaRegistry._truncate_sample(v, max_items) for k, v in list(data.items())[:20]
}
if isinstance(data, list):
return [
SchemaRegistry._truncate_sample(item, max_items)
for item in data[:max_items]
]
return [SchemaRegistry._truncate_sample(item, max_items) for item in data[:max_items]]
if isinstance(data, str) and len(data) > 200:
return data[:200] + "..."
return data

6
src/geoguessr_mcp/services/__init__.py
View file

@ -1,12 +1,12 @@
"""Services module for business logic."""
from .profile_service import ProfileService
from .game_service import GameService
from .analysis_service import AnalysisService, GameAnalysis
from .game_service import GameService
from .profile_service import ProfileService
__all__ = [
"ProfileService",
"GameService",
"AnalysisService",
"GameAnalysis",
]
]

120
src/geoguessr_mcp/services/analysis_service.py
View file

@ -21,6 +21,7 @@ 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
@ -88,15 +89,9 @@ class AnalysisService:
# 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
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]
@ -106,9 +101,7 @@ class AnalysisService:
# 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
)
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
)
@ -124,18 +117,22 @@ class AnalysisService:
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,
})
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,
})
strong_areas.append(
{
"game": game.token,
"round": round_guess.round_number,
"score": round_guess.score,
}
)
return GameAnalysis(
games_analyzed=len(games),
@ -204,9 +201,7 @@ class AnalysisService:
# Get comprehensive profile
try:
results["profile"] = await self.profile_service.get_comprehensive_profile(
session_token
)
results["profile"] = await self.profile_service.get_comprehensive_profile(session_token)
except Exception as e:
results["errors"].append(f"Profile: {str(e)}")
@ -227,17 +222,13 @@ class AnalysisService:
# Analyze recent games
try:
results["recent_games_analysis"] = await self.analyze_recent_games(
5, session_token
)
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
)
response = await self.client.get(self._create_endpoint("/v3/explorer"), session_token)
if response.is_success:
results["explorer"] = response.summarize()
except Exception as e:
@ -245,9 +236,7 @@ class AnalysisService:
# Get objectives
try:
response = await self.client.get(
self._create_endpoint("/v4/objectives"), session_token
)
response = await self.client.get(self._create_endpoint("/v4/objectives"), session_token)
if response.is_success:
results["objectives"] = response.summarize()
except Exception as e:
@ -273,42 +262,50 @@ class AnalysisService:
# 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.",
})
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.",
})
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.",
})
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.",
})
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": {
@ -331,4 +328,5 @@ class AnalysisService:
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}")

60
src/geoguessr_mcp/services/game_service.py
View file

@ -7,11 +7,11 @@ 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.client import DynamicResponse, GeoGuessrClient
from ..api.endpoints import Endpoints
from ..models.DailyChallenge import DailyChallenge
from ..models.Game import Game
from ..models.SeasonStats import SeasonStats
from ..models.DailyChallenge import DailyChallenge
logger = logging.getLogger(__name__)
@ -23,9 +23,9 @@ class GameService:
self.client = client
async def get_game_details(
self,
game_token: str,
session_token: Optional[str] = None,
self,
game_token: str,
session_token: Optional[str] = None,
) -> tuple[Game, DynamicResponse]:
"""
Get details for a specific game.
@ -47,26 +47,26 @@ class GameService:
raise ValueError(f"Failed to get game details: {response.data}")
async def get_unfinished_games(
self,
session_token: Optional[str] = None,
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,
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,
self,
count: int = 10,
page: int = 0,
session_token: Optional[str] = None,
) -> DynamicResponse:
"""
Get the activity feed.
@ -83,9 +83,9 @@ class GameService:
return await self.client.get(endpoint, session_token)
async def get_recent_games(
self,
count: int = 10,
session_token: Optional[str] = None,
self,
count: int = 10,
session_token: Optional[str] = None,
) -> list[Game]:
"""
Get recent games from the activity feed.
@ -124,8 +124,8 @@ class GameService:
return games
async def get_season_stats(
self,
session_token: Optional[str] = None,
self,
session_token: Optional[str] = None,
) -> tuple[SeasonStats, DynamicResponse]:
"""Get active season statistics."""
response = await self.client.get(
@ -139,9 +139,9 @@ class GameService:
raise ValueError(f"Failed to get season stats: {response.data}")
async def get_daily_challenge(
self,
day: str = "today",
session_token: Optional[str] = None,
self,
day: str = "today",
session_token: Optional[str] = None,
) -> tuple[DailyChallenge, DynamicResponse]:
"""
Get daily challenge.
@ -163,26 +163,26 @@ class GameService:
raise ValueError(f"Failed to get daily challenge: {response.data}")
async def get_battle_royale(
self,
game_id: str,
session_token: Optional[str] = None,
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,
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,
self,
session_token: Optional[str] = None,
) -> DynamicResponse:
"""Get tournament information."""
return await self.client.get(Endpoints.GAME_SERVER.GET_TOURNAMENTS, session_token)

8
src/geoguessr_mcp/services/profile_service.py
View file

@ -8,7 +8,7 @@ dynamic schema adaptation.
import logging
from typing import Optional
from ..api.client import GeoGuessrClient, DynamicResponse
from ..api.client import DynamicResponse, GeoGuessrClient
from ..api.endpoints import Endpoints
from ..models.Achievement import Achievement
from ..models.UserProfile import UserProfile
@ -171,11 +171,7 @@ class ProfileService:
"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]
for a in sorted(unlocked, key=lambda x: x.unlocked_at or "", reverse=True)[:5]
],
}
except Exception as e:

36
src/geoguessr_mcp/tools/analysis_tools.py
View file

@ -1,4 +1,3 @@
@mcp.tool()
async def analyze_recent_games(count: int = 10) -> dict:
"""
@ -11,8 +10,7 @@ async def analyze_recent_games(count: int = 10) -> dict:
async with await get_async_session() as client:
# Get activity feed
feed_response = await client.get(
f"{GEOGUESSR_BASE_URL}/v4/feed/private",
params={"count": count * 2, "page": 0}
f"{GEOGUESSR_BASE_URL}/v4/feed/private", params={"count": count * 2, "page": 0}
)
feed_response.raise_for_status()
feed = feed_response.json()
@ -27,7 +25,9 @@ async def analyze_recent_games(count: int = 10) -> dict:
game_token = entry.get("payload", {}).get("gameToken")
if game_token:
try:
game_response = await client.get(f"{GEOGUESSR_BASE_URL}/v3/games/{game_token}")
game_response = await client.get(
f"{GEOGUESSR_BASE_URL}/v3/games/{game_token}"
)
if game_response.status_code == 200:
game = game_response.json()
@ -36,17 +36,19 @@ async def analyze_recent_games(count: int = 10) -> dict:
"map": game.get("map", {}).get("name", "Unknown"),
"mode": game.get("type", "Unknown"),
"total_score": 0,
"rounds": []
"rounds": [],
}
for round_data in game.get("player", {}).get("guesses", []):
round_score = round_data.get("roundScoreInPoints", 0)
game_info["total_score"] += round_score
game_info["rounds"].append({
"score": round_score,
"distance": round_data.get("distanceInMeters", 0),
"time": round_data.get("time", 0)
})
game_info["rounds"].append(
{
"score": round_score,
"distance": round_data.get("distanceInMeters", 0),
"time": round_data.get("time", 0),
}
)
total_rounds += 1
if round_score == 5000:
@ -63,8 +65,10 @@ async def analyze_recent_games(count: int = 10) -> dict:
"average_score": total_score / len(games_analyzed) if games_analyzed else 0,
"total_rounds": total_rounds,
"perfect_rounds": perfect_rounds,
"perfect_round_percentage": (perfect_rounds / total_rounds * 100) if total_rounds > 0 else 0,
"games": games_analyzed
"perfect_round_percentage": (
(perfect_rounds / total_rounds * 100) if total_rounds > 0 else 0
),
"games": games_analyzed,
}
@ -111,14 +115,16 @@ async def get_performance_summary() -> dict:
# Get achievements
try:
achievements_response = await client.get(f"{GEOGUESSR_BASE_URL}/v3/profiles/achievements")
achievements_response = await client.get(
f"{GEOGUESSR_BASE_URL}/v3/profiles/achievements"
)
achievements_response.raise_for_status()
achievements = achievements_response.json()
results["achievements_summary"] = {
"total": len(achievements) if isinstance(achievements, list) else 0,
"achievements": achievements
"achievements": achievements,
}
except Exception as e:
results["achievements_error"] = str(e)
return results
return results

6
src/geoguessr_mcp/tools/auth_tools.py
View file

@ -1,5 +1,7 @@
"""MCP tools for auth operations."""
import logging
from mcp.server.fastmcp import FastMCP
from ..auth.session import SessionManager
@ -44,7 +46,6 @@ def register_auth_tools(mcp: FastMCP, session_manager: SessionManager):
logger.error(f"Login error: {e}")
return {"success": False, "error": f"An unexpected error occurred: {str(e)}"}
@mcp.tool()
async def logout() -> dict:
"""
@ -63,7 +64,6 @@ def register_auth_tools(mcp: FastMCP, session_manager: SessionManager):
return {"success": False, "message": "No active session"}
@mcp.tool()
async def set_session_token(token: str) -> dict:
"""
@ -86,7 +86,6 @@ def register_auth_tools(mcp: FastMCP, session_manager: SessionManager):
return {"success": False, "error": "Invalid or expired session token"}
@mcp.tool()
async def set_ncfa_cookie(cookie: str) -> dict:
"""
@ -135,7 +134,6 @@ def register_auth_tools(mcp: FastMCP, session_manager: SessionManager):
"session_token": session_token,
}
@mcp.tool()
async def get_auth_status() -> dict:
"""

5
src/tests/conftest.py
View file

@ -31,10 +31,7 @@ def mock_profile_data():
"created": "2025-01-01T00:00:00.000Z",
"isVerified": True,
"level": 50,
"rating": {
"rating": 1500,
"deviation": 100
}
"rating": {"rating": 1500, "deviation": 100},
}

2
src/tests/integration/__init__.py
View file

@ -1 +1 @@
"""Integration tests for GeoGuessr MCP Server."""
"""Integration tests for GeoGuessr MCP Server."""

2
src/tests/integration/test_api_client.py
View file

@ -1 +1 @@
# TODO
# TODO

2
src/tests/integration/test_auth_flow.py
View file

@ -1 +1 @@
# TODO
# TODO

2
src/tests/unit/__init__.py
View file

@ -1 +1 @@
"""Unit tests for GeoGuessr MCP Server."""
"""Unit tests for GeoGuessr MCP Server."""

5
src/tests/unit/auth/test_session.py
View file

@ -19,10 +19,11 @@ TestSessionManager
login, logout, and session management operations in an async context.
"""
import pytest
from datetime import datetime, timedelta, UTC
from datetime import UTC, datetime, timedelta
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from geoguessr_mcp.auth.session import SessionManager, UserSession

4
src/tests/unit/monitoring/schema/test_schema_detector.py
View file

@ -8,8 +8,8 @@ objects, computation of schema hashes, and parsing of specific data formats
such as datetime strings, URLs, and UUIDs.
"""
from geoguessr_mcp.monitoring.schema.SchemaDetector import SchemaDetector
from geoguessr_mcp.monitoring.schema.EndpointSchema import SchemaField
from geoguessr_mcp.monitoring.schema.SchemaDetector import SchemaDetector
class TestSchemaDetector:
@ -94,7 +94,7 @@ class TestSchemaDetector:
"id": "123",
"profile": {
"name": "Test",
}
},
}
}

2
src/tests/unit/services/test_analysis_service.py
View file

@ -1 +1 @@
# TODO
# TODO

2
src/tests/unit/services/test_game_service.py
View file

@ -1 +1 @@
# TODO
# TODO

2
src/tests/unit/services/test_profile_service.py
View file

@ -1 +1 @@
# TODO
# TODO