other_tools.pyā¢3.75 kB
"""
Chess.com titled players and club tools.
"""
from typing import Dict, Any, List
from chess_client import make_api_request, get_cache_ttl
from mxcp.runtime import db
async def get_titled_players(title: str) -> Dict[str, List[str]]:
"""
Get a list of titled players from Chess.com.
Args:
title: Chess title (GM, WGM, IM, WIM, FM, WFM, NM, WNM, CM, WCM)
Returns:
Dictionary with 'players' list containing usernames
"""
valid_titles = ["GM", "WGM", "IM", "WIM", "FM", "WFM", "NM", "WNM", "CM", "WCM"]
if title not in valid_titles:
return {"error": f"Invalid title. Must be one of: {', '.join(valid_titles)}"}
endpoint = f"titled/{title}"
cache_ttl = get_cache_ttl("titled_players")
try:
data = await make_api_request(endpoint, cache_ttl=cache_ttl)
# Store in titled players table for analytics
players = data.get("players", [])
db.execute("""
INSERT INTO titled_players_snapshot
(title, player_count, players_list, snapshot_date)
VALUES ($title, $count, $players, CURRENT_DATE)
ON CONFLICT (title, snapshot_date) DO UPDATE
SET player_count = $count, players_list = $players
""", {
"title": title,
"count": len(players),
"players": players
})
return data
except Exception as e:
return {"players": [], "error": str(e)}
async def get_club_profile(url_id: str) -> Dict[str, Any]:
"""
Get information about a club on Chess.com.
Args:
url_id: The URL identifier of the club
Returns:
Club profile data including name, description, members count, etc.
"""
endpoint = f"club/{url_id}"
cache_ttl = get_cache_ttl("club_info")
try:
data = await make_api_request(endpoint, cache_ttl=cache_ttl)
# Track club statistics
db.execute("""
INSERT INTO club_statistics
(club_id, club_name, member_count, checked_at)
VALUES ($id, $name, $members, CURRENT_TIMESTAMP)
""", {
"id": url_id,
"name": data.get("name", ""),
"members": data.get("members_count", 0)
})
return data
except Exception as e:
return {"error": f"Failed to fetch club profile: {str(e)}"}
async def get_club_members(url_id: str) -> Dict[str, Any]:
"""
Get members of a club on Chess.com.
Args:
url_id: The URL identifier of the club
Returns:
Dictionary with lists of members by type (weekly, monthly, all_time)
"""
endpoint = f"club/{url_id}/members"
cache_ttl = get_cache_ttl("club_info")
try:
data = await make_api_request(endpoint, cache_ttl=cache_ttl)
# Store member information for tracking
all_members = set()
for member_type in ["weekly", "monthly", "all_time"]:
members = data.get(member_type, [])
all_members.update(member["username"] for member in members if isinstance(member, dict))
db.execute("""
INSERT INTO club_member_snapshots
(club_id, member_count, members_list, snapshot_date)
VALUES ($id, $count, $members, CURRENT_DATE)
ON CONFLICT (club_id, snapshot_date) DO UPDATE
SET member_count = $count, members_list = $members
""", {
"id": url_id,
"count": len(all_members),
"members": list(all_members)
})
return data
except Exception as e:
return {"error": f"Failed to fetch club members: {str(e)}"}