8a1d15911f
Version control Claude Code configuration including: - Global instructions (CLAUDE.md) - User settings (settings.json) - Custom agents (architect, designer, engineer, etc.) - Custom skills (create-skill templates and workflows) Excludes session data, secrets, cache, and temporary files per .gitignore. Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
714 lines
21 KiB
Python
Executable File
714 lines
21 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Major Domo API Client
|
|
|
|
Shared API client for all Major Domo (SBA) operations.
|
|
Provides methods for interacting with teams, players, standings, stats, transactions, and more.
|
|
|
|
Environment Variables:
|
|
API_TOKEN: Bearer token for API authentication (required)
|
|
DATABASE: 'prod' or 'dev' (default: prod)
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
from typing import Optional, Dict, List, Any, Literal
|
|
import requests
|
|
|
|
|
|
class MajorDomoAPI:
|
|
"""
|
|
Major Domo API client for SBA database access
|
|
|
|
Usage:
|
|
api = MajorDomoAPI(environment='prod')
|
|
|
|
# Get current season/week
|
|
current = api.get_current()
|
|
|
|
# Get a team
|
|
team = api.get_team(abbrev='CAR')
|
|
|
|
# List players
|
|
players = api.list_players(season=12, team_id=42)
|
|
|
|
# Get standings
|
|
standings = api.get_standings(season=12, division_abbrev='ALE')
|
|
"""
|
|
|
|
def __init__(self, environment: str = 'prod', token: Optional[str] = None, verbose: bool = False):
|
|
"""
|
|
Initialize API client
|
|
|
|
Args:
|
|
environment: 'prod' or 'dev'
|
|
token: API token (defaults to API_TOKEN env var)
|
|
verbose: Print request/response details
|
|
"""
|
|
self.env = environment.lower()
|
|
|
|
# Set base URL based on environment
|
|
if 'prod' in self.env:
|
|
self.base_url = 'https://api.sba.manticorum.com/v3'
|
|
else:
|
|
self.base_url = 'http://10.10.0.42:8000/api/v3' # Docker dev container
|
|
|
|
self.token = token or os.getenv('API_TOKEN')
|
|
self.verbose = verbose
|
|
|
|
if not self.token:
|
|
raise ValueError(
|
|
"API_TOKEN environment variable required. "
|
|
"Set it with: export API_TOKEN='your-token-here'"
|
|
)
|
|
|
|
self.headers = {
|
|
'Authorization': f'Bearer {self.token}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
def _log(self, message: str):
|
|
"""Print message if verbose mode enabled"""
|
|
if self.verbose:
|
|
print(f"[API] {message}")
|
|
|
|
def _build_url(self, endpoint: str, object_id: Optional[int] = None, **params) -> str:
|
|
"""Build API URL with query parameters"""
|
|
url = f'{self.base_url}/{endpoint}'
|
|
|
|
if object_id is not None:
|
|
url += f'/{object_id}'
|
|
|
|
# Add query parameters
|
|
if params:
|
|
param_parts = []
|
|
for key, value in params.items():
|
|
if value is not None:
|
|
if isinstance(value, bool):
|
|
param_parts.append(f'{key}={str(value).lower()}')
|
|
elif isinstance(value, list):
|
|
for item in value:
|
|
param_parts.append(f'{key}={item}')
|
|
else:
|
|
param_parts.append(f'{key}={value}')
|
|
|
|
if param_parts:
|
|
url += '?' + '&'.join(param_parts)
|
|
|
|
return url
|
|
|
|
# ====================
|
|
# Low-level HTTP methods
|
|
# ====================
|
|
|
|
def get(self, endpoint: str, object_id: Optional[int] = None, timeout: int = 10, **params) -> Any:
|
|
"""GET request to API"""
|
|
url = self._build_url(endpoint, object_id=object_id, **params)
|
|
self._log(f"GET {url}")
|
|
response = requests.get(url, headers=self.headers, timeout=timeout)
|
|
response.raise_for_status()
|
|
return response.json() if response.text else None
|
|
|
|
def post(self, endpoint: str, payload: Optional[Dict] = None, timeout: int = 10, **params) -> Any:
|
|
"""POST request to API"""
|
|
url = self._build_url(endpoint, **params)
|
|
self._log(f"POST {url}")
|
|
response = requests.post(url, headers=self.headers, json=payload, timeout=timeout)
|
|
response.raise_for_status()
|
|
return response.json() if response.text else {}
|
|
|
|
def patch(self, endpoint: str, object_id: Optional[int] = None, payload: Optional[Dict] = None, timeout: int = 10, **params) -> Any:
|
|
"""PATCH request to API"""
|
|
url = self._build_url(endpoint, object_id=object_id, **params)
|
|
self._log(f"PATCH {url}")
|
|
response = requests.patch(url, headers=self.headers, json=payload, timeout=timeout)
|
|
response.raise_for_status()
|
|
return response.json() if response.text else {}
|
|
|
|
def delete(self, endpoint: str, object_id: int, timeout: int = 10) -> str:
|
|
"""DELETE request to API"""
|
|
url = self._build_url(endpoint, object_id=object_id)
|
|
self._log(f"DELETE {url}")
|
|
response = requests.delete(url, headers=self.headers, timeout=timeout)
|
|
response.raise_for_status()
|
|
return response.text
|
|
|
|
# ====================
|
|
# Current Season/Week Operations
|
|
# ====================
|
|
|
|
def get_current(self, season: Optional[int] = None) -> Dict:
|
|
"""
|
|
Get current season/week status
|
|
|
|
Args:
|
|
season: Specific season (defaults to latest)
|
|
|
|
Returns:
|
|
Current status dict with season, week, trade_deadline, etc.
|
|
"""
|
|
return self.get('current', season=season)
|
|
|
|
def update_current(self, current_id: int, **updates) -> Dict:
|
|
"""
|
|
Update current season/week status (private endpoint)
|
|
|
|
Args:
|
|
current_id: Current record ID
|
|
**updates: Fields to update (week, season, freeze, etc.)
|
|
|
|
Returns:
|
|
Updated current status dict
|
|
"""
|
|
return self.patch('current', object_id=current_id, **updates)
|
|
|
|
# ====================
|
|
# Team Operations
|
|
# ====================
|
|
|
|
def get_team(self, team_id: Optional[int] = None, abbrev: Optional[str] = None, season: Optional[int] = None) -> Dict:
|
|
"""
|
|
Get a team by ID or abbreviation
|
|
|
|
Args:
|
|
team_id: Team ID
|
|
abbrev: Team abbreviation (e.g., 'CAR')
|
|
season: Season number (if using abbrev)
|
|
|
|
Returns:
|
|
Team dict
|
|
"""
|
|
if team_id:
|
|
return self.get('teams', object_id=team_id)
|
|
elif abbrev:
|
|
result = self.get('teams', team_abbrev=[abbrev.upper()], season=season)
|
|
teams = result.get('teams', [])
|
|
if not teams:
|
|
raise ValueError(f"Team '{abbrev}' not found" + (f" in season {season}" if season else ""))
|
|
return teams[0]
|
|
else:
|
|
raise ValueError("Must provide team_id or abbrev")
|
|
|
|
def list_teams(
|
|
self,
|
|
season: Optional[int] = None,
|
|
owner_id: Optional[List[int]] = None,
|
|
manager_id: Optional[List[int]] = None,
|
|
team_abbrev: Optional[List[str]] = None,
|
|
active_only: bool = False,
|
|
short_output: bool = False
|
|
) -> List[Dict]:
|
|
"""
|
|
List teams
|
|
|
|
Args:
|
|
season: Filter by season
|
|
owner_id: Filter by owner Discord ID(s)
|
|
manager_id: Filter by manager ID(s)
|
|
team_abbrev: Filter by team abbreviation(s)
|
|
active_only: Exclude IL and MiL teams
|
|
short_output: Return minimal data
|
|
|
|
Returns:
|
|
List of team dicts
|
|
"""
|
|
result = self.get(
|
|
'teams',
|
|
season=season,
|
|
owner_id=owner_id,
|
|
manager_id=manager_id,
|
|
team_abbrev=team_abbrev,
|
|
active_only=active_only,
|
|
short_output=short_output
|
|
)
|
|
return result.get('teams', [])
|
|
|
|
def get_team_roster(
|
|
self,
|
|
team_id: int,
|
|
which: Literal['current', 'next'] = 'current',
|
|
sort: Optional[str] = None
|
|
) -> Dict:
|
|
"""
|
|
Get team roster breakdown
|
|
|
|
Args:
|
|
team_id: Team ID
|
|
which: 'current' or 'next' week
|
|
sort: Sort method (e.g., 'wara-desc')
|
|
|
|
Returns:
|
|
Roster dict with active/shortil/longil player lists
|
|
"""
|
|
return self.get(f'teams/{team_id}/roster/{which}', sort=sort)
|
|
|
|
def update_team(self, team_id: int, **updates) -> Dict:
|
|
"""
|
|
Update team (private endpoint)
|
|
|
|
Args:
|
|
team_id: Team ID
|
|
**updates: Fields to update (manager1_id, stadium, color, etc.)
|
|
|
|
Returns:
|
|
Updated team dict
|
|
"""
|
|
return self.patch('teams', object_id=team_id, **updates)
|
|
|
|
# ====================
|
|
# Player Operations
|
|
# ====================
|
|
|
|
def get_player(
|
|
self,
|
|
player_id: Optional[int] = None,
|
|
name: Optional[str] = None,
|
|
season: Optional[int] = None,
|
|
short_output: bool = False
|
|
) -> Optional[Dict]:
|
|
"""
|
|
Get a player by ID or name
|
|
|
|
Args:
|
|
player_id: Player ID
|
|
name: Player name (exact match)
|
|
season: Season (required if using name)
|
|
short_output: Return minimal data
|
|
|
|
Returns:
|
|
Player dict or None if not found
|
|
"""
|
|
if player_id:
|
|
return self.get('players', object_id=player_id, short_output=short_output)
|
|
elif name and season:
|
|
result = self.get('players', season=season, name=name, short_output=short_output)
|
|
players = result.get('players', [])
|
|
return players[0] if players else None
|
|
else:
|
|
raise ValueError("Must provide player_id or (name and season)")
|
|
|
|
def list_players(
|
|
self,
|
|
season: int,
|
|
team_id: Optional[List[int]] = None,
|
|
pos: Optional[List[str]] = None,
|
|
strat_code: Optional[List[str]] = None,
|
|
is_injured: Optional[bool] = None,
|
|
sort: Optional[str] = None,
|
|
short_output: bool = False
|
|
) -> List[Dict]:
|
|
"""
|
|
List players with filters
|
|
|
|
Args:
|
|
season: Season number (required)
|
|
team_id: Filter by team ID(s)
|
|
pos: Filter by position(s) (e.g., ['1B', 'OF'])
|
|
strat_code: Filter by Strat code(s)
|
|
is_injured: Only injured players
|
|
sort: Sort method ('cost-asc', 'cost-desc', 'name-asc', 'name-desc')
|
|
short_output: Return minimal data
|
|
|
|
Returns:
|
|
List of player dicts
|
|
"""
|
|
result = self.get(
|
|
'players',
|
|
season=season,
|
|
team_id=team_id,
|
|
pos=pos,
|
|
strat_code=strat_code,
|
|
is_injured=is_injured,
|
|
sort=sort,
|
|
short_output=short_output
|
|
)
|
|
return result.get('players', [])
|
|
|
|
def search_players(
|
|
self,
|
|
query: str,
|
|
season: Optional[int] = None,
|
|
limit: int = 10,
|
|
short_output: bool = False
|
|
) -> List[Dict]:
|
|
"""
|
|
Fuzzy search players by name
|
|
|
|
Args:
|
|
query: Search query (partial name match)
|
|
season: Season (defaults to current)
|
|
limit: Maximum results (1-50)
|
|
short_output: Return minimal data
|
|
|
|
Returns:
|
|
List of player dicts
|
|
"""
|
|
result = self.get(
|
|
'players/search',
|
|
q=query,
|
|
season=season,
|
|
limit=limit,
|
|
short_output=short_output
|
|
)
|
|
return result.get('players', [])
|
|
|
|
def update_player(self, player_id: int, **updates) -> Dict:
|
|
"""
|
|
Update player (private endpoint)
|
|
|
|
Args:
|
|
player_id: Player ID
|
|
**updates: Fields to update (team_id, wara, il_return, etc.)
|
|
|
|
Returns:
|
|
Updated player dict
|
|
"""
|
|
return self.patch('players', object_id=player_id, **updates)
|
|
|
|
# ====================
|
|
# Standings Operations
|
|
# ====================
|
|
|
|
def get_standings(
|
|
self,
|
|
season: int,
|
|
team_id: Optional[List[int]] = None,
|
|
league_abbrev: Optional[str] = None,
|
|
division_abbrev: Optional[str] = None,
|
|
short_output: bool = False
|
|
) -> List[Dict]:
|
|
"""
|
|
Get league standings
|
|
|
|
Args:
|
|
season: Season number
|
|
team_id: Filter by team ID(s)
|
|
league_abbrev: Filter by league (e.g., 'AL', 'NL')
|
|
division_abbrev: Filter by division (e.g., 'ALE', 'NLW')
|
|
short_output: Return minimal data
|
|
|
|
Returns:
|
|
List of standings dicts (sorted by win percentage)
|
|
"""
|
|
result = self.get(
|
|
'standings',
|
|
season=season,
|
|
team_id=team_id,
|
|
league_abbrev=league_abbrev,
|
|
division_abbrev=division_abbrev,
|
|
short_output=short_output
|
|
)
|
|
return result.get('standings', [])
|
|
|
|
def get_team_standings(self, team_id: int) -> Dict:
|
|
"""
|
|
Get standings for a single team
|
|
|
|
Args:
|
|
team_id: Team ID
|
|
|
|
Returns:
|
|
Standings dict
|
|
"""
|
|
return self.get(f'standings/team/{team_id}')
|
|
|
|
def recalculate_standings(self, season: int) -> str:
|
|
"""
|
|
Recalculate standings for a season (private endpoint)
|
|
|
|
Args:
|
|
season: Season number
|
|
|
|
Returns:
|
|
Success message
|
|
"""
|
|
return self.post(f'standings/s{season}/recalculate')
|
|
|
|
# ====================
|
|
# Transaction Operations
|
|
# ====================
|
|
|
|
def get_transactions(
|
|
self,
|
|
season: int,
|
|
team_abbrev: Optional[List[str]] = None,
|
|
week_start: Optional[int] = None,
|
|
week_end: Optional[int] = None,
|
|
cancelled: Optional[bool] = None,
|
|
frozen: Optional[bool] = None,
|
|
player_name: Optional[List[str]] = None,
|
|
player_id: Optional[List[int]] = None,
|
|
move_id: Optional[str] = None,
|
|
short_output: bool = False
|
|
) -> List[Dict]:
|
|
"""
|
|
Get transactions with filters
|
|
|
|
Args:
|
|
season: Season number
|
|
team_abbrev: Filter by team abbreviation(s)
|
|
week_start: Start week (inclusive)
|
|
week_end: End week (inclusive)
|
|
cancelled: Filter by cancelled status
|
|
frozen: Filter by frozen status
|
|
player_name: Filter by player name(s)
|
|
player_id: Filter by player ID(s)
|
|
move_id: Filter by specific move ID
|
|
short_output: Return minimal data
|
|
|
|
Returns:
|
|
List of transaction dicts
|
|
"""
|
|
result = self.get(
|
|
'transactions',
|
|
season=season,
|
|
team_abbrev=team_abbrev,
|
|
week_start=week_start,
|
|
week_end=week_end,
|
|
cancelled=cancelled,
|
|
frozen=frozen,
|
|
player_name=player_name,
|
|
player_id=player_id,
|
|
move_id=move_id,
|
|
short_output=short_output
|
|
)
|
|
return result.get('transactions', [])
|
|
|
|
# ====================
|
|
# Statistics Operations (Advanced Views)
|
|
# ====================
|
|
|
|
def get_season_batting_stats(
|
|
self,
|
|
season: Optional[int] = None,
|
|
team_id: Optional[int] = None,
|
|
player_id: Optional[int] = None,
|
|
sbaplayer_id: Optional[int] = None,
|
|
min_pa: Optional[int] = None,
|
|
sort_by: str = 'woba',
|
|
sort_order: Literal['asc', 'desc'] = 'desc',
|
|
limit: int = 200,
|
|
offset: int = 0
|
|
) -> List[Dict]:
|
|
"""
|
|
Get season batting statistics
|
|
|
|
Args:
|
|
season: Season number (defaults to current)
|
|
team_id: Filter by team
|
|
player_id: Filter by player
|
|
sbaplayer_id: Filter by SBA player reference
|
|
min_pa: Minimum plate appearances
|
|
sort_by: Sort field (woba, avg, obp, slg, ops, hr, rbi, etc.)
|
|
sort_order: 'asc' or 'desc'
|
|
limit: Maximum results
|
|
offset: Offset for pagination
|
|
|
|
Returns:
|
|
List of batting stat dicts
|
|
"""
|
|
result = self.get(
|
|
'views/season-stats/batting',
|
|
season=season,
|
|
team_id=team_id,
|
|
player_id=player_id,
|
|
sbaplayer_id=sbaplayer_id,
|
|
min_pa=min_pa,
|
|
sort_by=sort_by,
|
|
sort_order=sort_order,
|
|
limit=limit,
|
|
offset=offset
|
|
)
|
|
return result.get('stats', [])
|
|
|
|
def get_season_pitching_stats(
|
|
self,
|
|
season: Optional[int] = None,
|
|
team_id: Optional[int] = None,
|
|
player_id: Optional[int] = None,
|
|
sbaplayer_id: Optional[int] = None,
|
|
min_outs: Optional[int] = None,
|
|
sort_by: str = 'era',
|
|
sort_order: Literal['asc', 'desc'] = 'asc',
|
|
limit: int = 200,
|
|
offset: int = 0
|
|
) -> List[Dict]:
|
|
"""
|
|
Get season pitching statistics
|
|
|
|
Args:
|
|
season: Season number (defaults to current)
|
|
team_id: Filter by team
|
|
player_id: Filter by player
|
|
sbaplayer_id: Filter by SBA player reference
|
|
min_outs: Minimum outs pitched
|
|
sort_by: Sort field (era, whip, k, bb, w, l, sv, etc.)
|
|
sort_order: 'asc' or 'desc'
|
|
limit: Maximum results
|
|
offset: Offset for pagination
|
|
|
|
Returns:
|
|
List of pitching stat dicts
|
|
"""
|
|
result = self.get(
|
|
'views/season-stats/pitching',
|
|
season=season,
|
|
team_id=team_id,
|
|
player_id=player_id,
|
|
sbaplayer_id=sbaplayer_id,
|
|
min_outs=min_outs,
|
|
sort_by=sort_by,
|
|
sort_order=sort_order,
|
|
limit=limit,
|
|
offset=offset
|
|
)
|
|
return result.get('stats', [])
|
|
|
|
def refresh_batting_stats(self, season: int) -> Dict:
|
|
"""
|
|
Refresh batting statistics for a season (private endpoint)
|
|
|
|
Args:
|
|
season: Season number
|
|
|
|
Returns:
|
|
Result dict with players_updated count
|
|
"""
|
|
return self.post(f'views/season-stats/batting/refresh', season=season)
|
|
|
|
def refresh_pitching_stats(self, season: int) -> Dict:
|
|
"""
|
|
Refresh pitching statistics for a season (private endpoint)
|
|
|
|
Args:
|
|
season: Season number
|
|
|
|
Returns:
|
|
Result dict with players_updated count
|
|
"""
|
|
return self.post(f'views/season-stats/pitching/refresh', season=season)
|
|
|
|
# ====================
|
|
# Schedule & Results Operations
|
|
# ====================
|
|
|
|
def get_schedules(
|
|
self,
|
|
season: int,
|
|
team_id: Optional[List[int]] = None,
|
|
week: Optional[int] = None,
|
|
short_output: bool = False
|
|
) -> List[Dict]:
|
|
"""
|
|
Get game schedules
|
|
|
|
Args:
|
|
season: Season number
|
|
team_id: Filter by team ID(s)
|
|
week: Filter by week
|
|
short_output: Return minimal data
|
|
|
|
Returns:
|
|
List of schedule dicts
|
|
"""
|
|
result = self.get(
|
|
'schedules',
|
|
season=season,
|
|
team_id=team_id,
|
|
week=week,
|
|
short_output=short_output
|
|
)
|
|
return result.get('schedules', [])
|
|
|
|
def get_results(
|
|
self,
|
|
season: int,
|
|
team_id: Optional[List[int]] = None,
|
|
week: Optional[int] = None,
|
|
short_output: bool = False
|
|
) -> List[Dict]:
|
|
"""
|
|
Get game results
|
|
|
|
Args:
|
|
season: Season number
|
|
team_id: Filter by team ID(s)
|
|
week: Filter by week
|
|
short_output: Return minimal data
|
|
|
|
Returns:
|
|
List of result dicts
|
|
"""
|
|
result = self.get(
|
|
'results',
|
|
season=season,
|
|
team_id=team_id,
|
|
week=week,
|
|
short_output=short_output
|
|
)
|
|
return result.get('results', [])
|
|
|
|
# ====================
|
|
# Utility Methods
|
|
# ====================
|
|
|
|
def health_check(self) -> bool:
|
|
"""
|
|
Check if API is accessible
|
|
|
|
Returns:
|
|
True if API is responding
|
|
"""
|
|
try:
|
|
self.get_current()
|
|
return True
|
|
except Exception as e:
|
|
if self.verbose:
|
|
print(f"Health check failed: {e}")
|
|
return False
|
|
|
|
def __repr__(self) -> str:
|
|
return f"MajorDomoAPI(env='{self.env}', base_url='{self.base_url}')"
|
|
|
|
|
|
def main():
|
|
"""CLI interface for testing"""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(description='Major Domo API Client')
|
|
parser.add_argument('--env', choices=['prod', 'dev'], default='prod', help='Environment')
|
|
parser.add_argument('--verbose', action='store_true', help='Verbose output')
|
|
args = parser.parse_args()
|
|
|
|
try:
|
|
api = MajorDomoAPI(environment=args.env, verbose=args.verbose)
|
|
|
|
print(f"Connected to {api.env} environment")
|
|
print(f"Base URL: {api.base_url}")
|
|
|
|
# Test connection
|
|
current = api.get_current()
|
|
print(f"\nCurrent Status:")
|
|
print(f" Season: {current['season']}")
|
|
print(f" Week: {current['week']}")
|
|
print(f" Freeze: {current['freeze']}")
|
|
print(f" Trade Deadline: Week {current['trade_deadline']}")
|
|
|
|
# List teams
|
|
teams = api.list_teams(season=current['season'], active_only=True)
|
|
print(f"\nActive Teams in Season {current['season']}: {len(teams)}")
|
|
for team in teams[:5]:
|
|
print(f" {team['abbrev']}: {team['lname']}")
|
|
if len(teams) > 5:
|
|
print(f" ... and {len(teams) - 5} more")
|
|
|
|
print("\n✅ API client working correctly!")
|
|
|
|
except Exception as e:
|
|
print(f"❌ Error: {e}")
|
|
sys.exit(1)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|