43d32e9b9d
- Refactor major-domo skill: api_client.py, cli.py, and CLI modules (admin, common, injuries, results, schedule, transactions) with significant simplification (-275 lines net) - Update CLI_REFERENCE.md and SKILL.md docs for major-domo - Update create-scheduled-task SKILL.md - Update plugins blocklist.json and known_marketplaces.json - Add patterns/ directory to repo - Update CLAUDE.md Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
722 lines
21 KiB
Python
Executable File
722 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 for write operations)
|
|
DATABASE: 'prod' or 'dev' (default: prod)
|
|
"""
|
|
|
|
import os
|
|
import sys
|
|
from typing import Optional, Dict, List, Any, Literal
|
|
import requests
|
|
|
|
_BASE_URLS = {
|
|
"prod": "https://api.sba.manticorum.com/v3",
|
|
"dev": "http://10.10.0.42:8000/api/v3",
|
|
}
|
|
|
|
|
|
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()
|
|
self.base_url = _BASE_URLS.get(self.env, _BASE_URLS["prod"])
|
|
|
|
self.token = token or os.getenv("API_TOKEN")
|
|
self.verbose = verbose
|
|
|
|
self.headers = {"Content-Type": "application/json"}
|
|
if self.token:
|
|
self.headers["Authorization"] = f"Bearer {self.token}"
|
|
|
|
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 _require_token(self, method: str):
|
|
"""Raise if no API token is set (required for write operations)."""
|
|
if not self.token:
|
|
raise ValueError(
|
|
f"{method} requires API_TOKEN. Set it with: export API_TOKEN='your-token-here'"
|
|
)
|
|
|
|
def post(
|
|
self, endpoint: str, payload: Optional[Dict] = None, timeout: int = 10, **params
|
|
) -> Any:
|
|
"""POST request to API"""
|
|
self._require_token("POST")
|
|
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"""
|
|
self._require_token("PATCH")
|
|
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"""
|
|
self._require_token("DELETE")
|
|
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 (Active), shortil (Injured List), longil (Minor League) 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_stats(
|
|
self,
|
|
stat_type: Literal["batting", "pitching"],
|
|
season: Optional[int] = None,
|
|
team_id: Optional[int] = None,
|
|
player_id: Optional[int] = None,
|
|
sbaplayer_id: Optional[int] = None,
|
|
min_threshold: Optional[int] = None,
|
|
sort_by: Optional[str] = None,
|
|
sort_order: Optional[Literal["asc", "desc"]] = None,
|
|
limit: int = 200,
|
|
offset: int = 0,
|
|
) -> List[Dict]:
|
|
"""Get season statistics for batting or pitching."""
|
|
threshold_key = "min_pa" if stat_type == "batting" else "min_outs"
|
|
result = self.get(
|
|
f"views/season-stats/{stat_type}",
|
|
season=season,
|
|
team_id=team_id,
|
|
player_id=player_id,
|
|
sbaplayer_id=sbaplayer_id,
|
|
sort_by=sort_by,
|
|
sort_order=sort_order,
|
|
limit=limit,
|
|
offset=offset,
|
|
**({threshold_key: min_threshold} if min_threshold is not None else {}),
|
|
)
|
|
return result.get("stats", [])
|
|
|
|
def get_season_batting_stats(
|
|
self,
|
|
*,
|
|
min_pa: Optional[int] = None,
|
|
sort_by: str = "woba",
|
|
sort_order: Literal["asc", "desc"] = "desc",
|
|
**kwargs,
|
|
) -> List[Dict]:
|
|
"""Get season batting statistics. See _get_season_stats for full kwargs."""
|
|
return self._get_season_stats(
|
|
"batting",
|
|
min_threshold=min_pa,
|
|
sort_by=sort_by,
|
|
sort_order=sort_order,
|
|
**kwargs,
|
|
)
|
|
|
|
def get_season_pitching_stats(
|
|
self,
|
|
*,
|
|
min_outs: Optional[int] = None,
|
|
sort_by: str = "era",
|
|
sort_order: Literal["asc", "desc"] = "asc",
|
|
**kwargs,
|
|
) -> List[Dict]:
|
|
"""Get season pitching statistics. See _get_season_stats for full kwargs."""
|
|
return self._get_season_stats(
|
|
"pitching",
|
|
min_threshold=min_outs,
|
|
sort_by=sort_by,
|
|
sort_order=sort_order,
|
|
**kwargs,
|
|
)
|
|
|
|
def refresh_stats(
|
|
self, stat_type: Literal["batting", "pitching"], season: int
|
|
) -> Dict:
|
|
"""Refresh batting or pitching statistics for a season (private endpoint)."""
|
|
return self.post(f"views/season-stats/{stat_type}/refresh", season=season)
|
|
|
|
def refresh_batting_stats(self, season: int) -> Dict:
|
|
"""Refresh batting statistics for a season."""
|
|
return self.refresh_stats("batting", season)
|
|
|
|
def refresh_pitching_stats(self, season: int) -> Dict:
|
|
"""Refresh pitching statistics for a season."""
|
|
return self.refresh_stats("pitching", 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()
|