diff --git a/.gitignore b/.gitignore index d7d24b3..f8744e8 100644 --- a/.gitignore +++ b/.gitignore @@ -31,4 +31,6 @@ data/exports/ .coverage.* htmlcov/ .pytest_cache/ +.pre-commit-cache/ +.pre-commit-home/ .ai/* \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index c3ce0ae..9030399 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -36,6 +36,7 @@ ffpy-pickem-web = "ffpy.pickem_web:main" ffpy-league-api = "ffpy.league_api:main" ffpy-web = "ffpy.unified_web:main" ffpy-sleeper = "ffpy.sleeper_web.main:main" +ffpy-ingest = "ffpy.ingest.cli:main" [build-system] requires = ["uv_build>=0.9.18,<0.10.0"] diff --git a/src/ffpy/ingest/__init__.py b/src/ffpy/ingest/__init__.py new file mode 100644 index 0000000..9e5326b --- /dev/null +++ b/src/ffpy/ingest/__init__.py @@ -0,0 +1,17 @@ +"""FFPy league data ingestion CLI. + +Usage: + ffpy-ingest espn [--season N] [--json|--csv] [--db PATH] [--swid ...] [--s2 ...] + ffpy-ingest yahoo [--season N] [--json|--csv] [--db PATH] [--token ...] + ffpy-ingest sleeper [--season N] [--json|--csv] [--db PATH] + ffpy-ingest yahoo-auth + ffpy-ingest yahoo-token --code CODE + ffpy-ingest leagues-list [--json|--csv] [--db PATH] + ffpy-ingest leagues-info [--json|--csv] [--db PATH] + ffpy-ingest roster [--json|--csv] [--db PATH] + ffpy-ingest matchups [--json|--csv] [--db PATH] +""" + +from . import auth, cli, espn, output, sleeper, yahoo + +__all__ = ["auth", "cli", "espn", "output", "sleeper", "yahoo"] diff --git a/src/ffpy/ingest/auth.py b/src/ffpy/ingest/auth.py new file mode 100644 index 0000000..ea1ca75 --- /dev/null +++ b/src/ffpy/ingest/auth.py @@ -0,0 +1,92 @@ +"""Auth helpers for ingest CLI: cookie files, token file I/O, env helpers.""" + +from __future__ import annotations + +import json +import logging +import os +import time +from pathlib import Path +from typing import Optional + +logger = logging.getLogger(__name__) + +TOKEN_DIR = Path.home() / ".ffpy" +TOKEN_FILE = TOKEN_DIR / "yahoo_token.json" +COOKIE_FILE = TOKEN_DIR / "espn_cookies.json" + + +def _ensure_dir() -> None: + TOKEN_DIR.mkdir(parents=True, exist_ok=True) + + +# --------------------------------------------------------------------------- +# Yahoo token file +# --------------------------------------------------------------------------- + + +def load_yahoo_token() -> Optional[dict]: + """Load Yahoo OAuth token from ~/.ffpy/yahoo_token.json if valid.""" + if not TOKEN_FILE.exists(): + return None + try: + data = json.loads(TOKEN_FILE.read_text()) + except (json.JSONDecodeError, OSError): + logger.warning("Corrupt Yahoo token file: %s", TOKEN_FILE) + return None + + expires_at = data.get("expires_at", 0) + if time.time() >= expires_at: + logger.info("Yahoo token expired — caller should refresh") + return data + + +def save_yahoo_token(token: dict) -> None: + """Persist Yahoo OAuth token to ~/.ffpy/yahoo_token.json.""" + _ensure_dir() + TOKEN_FILE.write_text(json.dumps(token, indent=2)) + TOKEN_FILE.chmod(0o600) + logger.info("Yahoo token saved to %s", TOKEN_FILE) + + +def delete_yahoo_token() -> None: + """Remove stored Yahoo token.""" + if TOKEN_FILE.exists(): + TOKEN_FILE.unlink() + logger.info("Yahoo token deleted") + + +# --------------------------------------------------------------------------- +# ESPN cookie file +# --------------------------------------------------------------------------- + + +def load_espn_cookies() -> tuple[str, str]: + """Load ESPN cookies from ~/.ffpy/espn_cookies.json or env vars.""" + swid = os.getenv("ESPN_SWID", "") + s2 = os.getenv("ESPN_S2", "") + + if not swid or not s2: + if COOKIE_FILE.exists(): + try: + data = json.loads(COOKIE_FILE.read_text()) + swid = data.get("swid", swid) + s2 = data.get("espn_s2", s2) + except (json.JSONDecodeError, OSError): + pass + + return swid, s2 + + +def save_espn_cookies(swid: str, espn_s2: str) -> None: + """Persist ESPN cookies to ~/.ffpy/espn_cookies.json.""" + _ensure_dir() + COOKIE_FILE.write_text(json.dumps({"swid": swid, "espn_s2": espn_s2}, indent=2)) + COOKIE_FILE.chmod(0o600) + logger.info("ESPN cookies saved to %s", COOKIE_FILE) + + +def delete_espn_cookies() -> None: + if COOKIE_FILE.exists(): + COOKIE_FILE.unlink() + logger.info("ESPN cookies deleted") diff --git a/src/ffpy/ingest/cli.py b/src/ffpy/ingest/cli.py new file mode 100644 index 0000000..98224d6 --- /dev/null +++ b/src/ffpy/ingest/cli.py @@ -0,0 +1,330 @@ +"""ffpy-ingest CLI — ingest league data from ESPN, Yahoo, and Sleeper. + +Usage: + ffpy-ingest espn [--season N] [--json|--csv] [--db PATH] [--swid ...] [--s2 ...] + ffpy-ingest yahoo [--season N] [--json|--csv] [--db PATH] [--token ...] + ffpy-ingest sleeper [--season N] [--json|--csv] [--db PATH] + ffpy-ingest yahoo-auth + ffpy-ingest yahoo-token --code CODE + ffpy-ingest leagues-list [--json|--csv] [--db PATH] + ffpy-ingest leagues-info [--json|--csv] [--db PATH] + ffpy-ingest roster [--json|--csv] [--db PATH] + ffpy-ingest matchups [--json|--csv] [--db PATH] +""" + +from __future__ import annotations + +import argparse +import logging +import sys +import time +from typing import Any, List, Optional + +from ffpy.database import FFPyDatabase +from ffpy.ingest import auth, espn, output, sleeper, yahoo + +logger = logging.getLogger(__name__) + + +def _get_db(db_path: Optional[str] = None) -> FFPyDatabase: + return FFPyDatabase(db_path=db_path) + + +def _format_and_exit(data: Any, fmt: str) -> None: + output.format_output(data, fmt) + sys.exit(0) + + +# --------------------------------------------------------------------------- +# Subcommand: ingest +# --------------------------------------------------------------------------- + + +def cmd_ingest_espn(args: argparse.Namespace) -> None: + data = espn.fetch_espn_league( + league_id=args.league_id, + season=args.season, + swid=args.swid, + espn_s2=args.s2, + ) + if args.db is not None: + output.persist_to_db(data, db_path=args.db or None) + _format_and_exit(data, args.format) + + +def cmd_ingest_yahoo(args: argparse.Namespace) -> None: + data = yahoo.fetch_yahoo_league( + league_id=args.league_id, + season=args.season, + access_token=args.token, + ) + if args.db is not None: + output.persist_to_db(data, db_path=args.db or None) + _format_and_exit(data, args.format) + + +def cmd_ingest_sleeper(args: argparse.Namespace) -> None: + data = sleeper.fetch_sleeper_league( + league_id=args.league_id, + season=args.season, + ) + if args.db is not None: + output.persist_to_db(data, db_path=args.db or None) + _format_and_exit(data, args.format) + + +# --------------------------------------------------------------------------- +# Subcommand: yahoo auth +# --------------------------------------------------------------------------- + + +def cmd_yahoo_auth(args: argparse.Namespace) -> None: + """Run the Yahoo OAuth flow interactively.""" + client_id, client_secret, redirect_uri = yahoo.get_client_credentials() + integration = yahoo.YahooIntegration(client_id, client_secret, redirect_uri) + + auth_url = integration.get_authorization_url(state="ffpy-ingest") + print("\n1. Visit this URL in your browser:\n") + print(f" {auth_url}\n") + print("2. Authorize the application") + print("3. Copy the full redirect URL and paste it below\n") + + redirect_response = input("Paste redirect URL or authorization code: ").strip() + if "code=" in redirect_response: + from urllib.parse import parse_qs, urlparse + + parsed = urlparse(redirect_response) + params = parse_qs(parsed.query) + code = params.get("code", [""])[0] + else: + code = redirect_response + + if not code: + print("Error: No authorization code provided") + sys.exit(1) + + token = integration.exchange_code(code) + token["expires_at"] = time.time() + token.get("expires_in", 3600) + auth.save_yahoo_token(token) + print(f"\nToken saved to {auth.TOKEN_FILE}") + print(f"Access token: {token['access_token'][:20]}... (expires at {token['expires_at']})") + + +def cmd_yahoo_token(args: argparse.Namespace) -> None: + """Exchange an OAuth code for a token (non-interactive).""" + client_id, client_secret, redirect_uri = yahoo.get_client_credentials() + integration = yahoo.YahooIntegration(client_id, client_secret, redirect_uri) + + code = args.code + token = integration.exchange_code(code) + token["expires_at"] = time.time() + token.get("expires_in", 3600) + auth.save_yahoo_token(token) + print(f"Token saved to {auth.TOKEN_FILE}") + + +# --------------------------------------------------------------------------- +# Subcommand: leagues +# --------------------------------------------------------------------------- + + +def cmd_leagues_list(args: argparse.Namespace) -> None: + db = _get_db(args.db or None) + try: + leagues = db.get_all_leagues() + finally: + db.close() + + if not leagues: + print("No imported leagues found.") + sys.exit(0) + + rows = [] + for lg in leagues: + rows.append( + { + "league_id": lg.get("league_id", ""), + "provider": lg.get("provider", ""), + "name": lg.get("league_name", ""), + "season": str(lg.get("season", "")), + "teams": str(lg.get("num_teams", "")), + } + ) + output.format_output(rows, args.format) + + +def cmd_leagues_info(args: argparse.Namespace) -> None: + db = _get_db(args.db or None) + try: + league = db.get_league_by_id(args.id) + finally: + db.close() + + if not league: + print(f"League '{args.id}' not found.", file=sys.stderr) + sys.exit(1) + + output.format_output(dict(league), args.format) + + +# --------------------------------------------------------------------------- +# Subcommand: roster +# --------------------------------------------------------------------------- + + +def cmd_roster(args: argparse.Namespace) -> None: + db = _get_db(args.db or None) + try: + teams = db.get_teams_for_league(args.league_id) + finally: + db.close() + + if not teams: + print(f"No teams found for league '{args.league_id}'", file=sys.stderr) + sys.exit(1) + + team = next((t for t in teams if t["team_id"] == args.team_id), None) + if not team: + print(f"Team '{args.team_id}' not found in league '{args.league_id}'", file=sys.stderr) + sys.exit(1) + + import json + + roster = json.loads(team.get("roster_json") or "[]") + if not roster: + print("Roster is empty") + sys.exit(0) + + output.format_output(roster, args.format) + + +# --------------------------------------------------------------------------- +# Subcommand: matchups +# --------------------------------------------------------------------------- + + +def cmd_matchups(args: argparse.Namespace) -> None: + db = _get_db(args.db or None) + try: + matchups = db.get_matchups_for_league(args.league_id, args.week) + finally: + db.close() + + if not matchups: + print(f"No matchups found for league '{args.league_id}' week {args.week}", file=sys.stderr) + sys.exit(1) + + rows = [] + for m in matchups: + rows.append( + { + "week": m.get("week", ""), + "home": m.get("home_team_id", ""), + "away": m.get("away_team_id", ""), + "home_score": m.get("home_score", ""), + "away_score": m.get("away_score", ""), + } + ) + output.format_output(rows, args.format) + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + + +def _common_parser() -> argparse.ArgumentParser: + """Shared flags that work after any subcommand.""" + common = argparse.ArgumentParser(add_help=False) + common.add_argument( + "--db", + nargs="?", + const="", + default=None, + help="Also persist to SQLite (optional path; default: ~/.ffpy/ffpy.db)", + ) + common.add_argument( + "--json", action="store_const", dest="format", const="json", default="table", help="Output as JSON" + ) + common.add_argument("--csv", action="store_const", dest="format", const="csv", help="Output as CSV") + return common + + +def build_parser() -> argparse.ArgumentParser: + common = _common_parser() + parser = argparse.ArgumentParser( + prog="ffpy-ingest", + description="Ingest fantasy league data from ESPN, Yahoo, and Sleeper.", + ) + sub = parser.add_subparsers(dest="command", required=True) + + # --- ingest espn --- + ingest_espn = sub.add_parser("espn", parents=[common], help="Ingest ESPN league data") + ingest_espn.add_argument("league_id", help="ESPN league ID (from URL)") + ingest_espn.add_argument("--season", type=int, default=2025, help="Season year (default: 2025)") + ingest_espn.add_argument("--swid", default=None, help="ESPN SWID cookie (for private leagues)") + ingest_espn.add_argument("--s2", default=None, help="ESPN S2 cookie (for private leagues)") + ingest_espn.set_defaults(func=cmd_ingest_espn) + + # --- ingest yahoo --- + ingest_yahoo = sub.add_parser("yahoo", parents=[common], help="Ingest Yahoo league data") + ingest_yahoo.add_argument("league_id", help="Yahoo league key (e.g. 389.l.12345)") + ingest_yahoo.add_argument("--season", type=int, default=2025, help="Season year (default: 2025)") + ingest_yahoo.add_argument("--token", default=None, help="Yahoo OAuth access token") + ingest_yahoo.set_defaults(func=cmd_ingest_yahoo) + + # --- ingest sleeper --- + ingest_sleeper = sub.add_parser("sleeper", parents=[common], help="Ingest Sleeper league data") + ingest_sleeper.add_argument("league_id", help="Sleeper league ID") + ingest_sleeper.add_argument("--season", type=int, default=2025, help="Season year (default: 2025)") + ingest_sleeper.set_defaults(func=cmd_ingest_sleeper) + + # --- yahoo auth --- + yahoo_auth = sub.add_parser("yahoo-auth", help="Run Yahoo OAuth flow") + yahoo_auth.set_defaults(func=cmd_yahoo_auth) + + # --- yahoo token --- + yahoo_token = sub.add_parser("yahoo-token", help="Exchange Yahoo OAuth code for token") + yahoo_token.add_argument("--code", required=True, help="OAuth authorization code") + yahoo_token.set_defaults(func=cmd_yahoo_token) + + # --- leagues list --- + leagues_list = sub.add_parser("leagues-list", parents=[common], help="List imported leagues from DB") + leagues_list.set_defaults(func=cmd_leagues_list) + + # --- leagues info --- + leagues_info = sub.add_parser("leagues-info", parents=[common], help="Show imported league details") + leagues_info.add_argument("id", help="League ID (prefixed, e.g. espn:123456)") + leagues_info.set_defaults(func=cmd_leagues_info) + + # --- roster --- + roster = sub.add_parser("roster", parents=[common], help="Show team roster from DB") + roster.add_argument("league_id", help="League ID (prefixed)") + roster.add_argument("team_id", help="Team ID (prefixed)") + roster.set_defaults(func=cmd_roster) + + # --- matchups --- + matchups = sub.add_parser("matchups", parents=[common], help="Show week matchups from DB") + matchups.add_argument("league_id", help="League ID (prefixed)") + matchups.add_argument("week", type=int, help="Week number (1-17)") + matchups.set_defaults(func=cmd_matchups) + + return parser + + +def main(argv: Optional[List[str]] = None) -> None: + parser = build_parser() + args = parser.parse_args(argv) + + logging.basicConfig( + level=logging.INFO, + format="%(levelname)s %(name)s: %(message)s", + ) + + if hasattr(args, "func"): + args.func(args) + else: + parser.print_help() + + +if __name__ == "__main__": + main() diff --git a/src/ffpy/ingest/espn.py b/src/ffpy/ingest/espn.py new file mode 100644 index 0000000..99e6fa7 --- /dev/null +++ b/src/ffpy/ingest/espn.py @@ -0,0 +1,143 @@ +"""ESPN league data fetcher with public/private auto-detect.""" + +from __future__ import annotations + +import logging +from typing import List, Optional + +import pandas as pd +import requests + +from ffpy.ingest.auth import load_espn_cookies, save_espn_cookies +from ffpy.integrations.espn_league import ESPNLeagueIntegration + +logger = logging.getLogger(__name__) + + +def _prompt_for_cookies() -> tuple[str, str]: + """Interactively prompt user for ESPN SWID and espn_s2 cookies.""" + print("\n--- ESPN Private League Authentication ---") + print("Your league appears to be private. Cookies can be found at:") + print(" Browser DevTools > Application > Cookies > https://www.espn.com\n") + swid = input("SWID cookie: ").strip() + s2 = input("espn_s2 cookie: ").strip() + save = input("Save cookies to ~/.ffpy/espn_cookies.json for reuse? [y/N]: ").strip().lower() + if save in ("y", "yes"): + save_espn_cookies(swid, s2) + return swid, s2 + + +def fetch_espn_league( + league_id: str, + season: int = 2025, + swid: Optional[str] = None, + espn_s2: Optional[str] = None, + interactive: bool = True, +) -> dict: + """Fetch league data from ESPN, trying public access first. + + Returns the normalized ``{league, teams, matchups}`` dict used by + ``ffpy.database.FFPyDatabase.store_user_league()``. + + * If the league is public, no cookies are needed. + * If private and cookies are provided via args/env/cookie-file, they are used. + * If private and no cookies available and ``interactive=True``, prompts the user. + """ + # Try public first + integration = ESPNLeagueIntegration(league_id=int(league_id), season=season) + integration.cookies = {} # ensure no cookies sent + + public = True + try: + info = integration.get_league_info() + teams = integration.get_all_teams() + all_rosters = integration.get_all_rosters() + except requests.HTTPError as exc: + status = getattr(getattr(exc, "response", None), "status_code", None) + if status not in (401, 403): + raise + public = False + logger.info("League %s requires auth (HTTP %s)", league_id, status) + info = teams = None + all_rosters = None + + if not public: + # Load cookies from args, env, or cookie file + resolved_swid = swid + resolved_s2 = espn_s2 + if not resolved_swid or not resolved_s2: + resolved_swid, resolved_s2 = load_espn_cookies() + + if (not resolved_swid or not resolved_s2) and interactive: + resolved_swid, resolved_s2 = _prompt_for_cookies() + + if not resolved_swid or not resolved_s2: + raise RuntimeError( + f"ESPN league {league_id} is private. " + "Provide ESPN_SWID and ESPN_S2 via env vars, cookie file, or --swid/--s2 flags." + ) + + integration = ESPNLeagueIntegration( + league_id=int(league_id), season=season, swid=resolved_swid, espn_s2=resolved_s2 + ) + info = integration.get_league_info() + teams = integration.get_all_teams() + all_rosters = integration.get_all_rosters() + + # Build matchups (loop weeks 1-17, stop on failure) + matchups: List[dict] = [] + for week in range(1, 18): + try: + week_matchups = integration.get_matchups(week) + except Exception: + break + for m in week_matchups: + matchups.append( + { + "week": week, + "home_team_id": f"espn:{league_id}:{m['home_team_id']}", + "away_team_id": f"espn:{league_id}:{m['away_team_id']}", + "home_score": m.get("home_score"), + "away_score": m.get("away_score"), + "is_playoff": 0, + "is_consolation": 0, + } + ) + + team_list = [] + for t in teams or []: + tid = t["id"] + roster = all_rosters.get(tid, pd.DataFrame()) if all_rosters is not None else pd.DataFrame() + team_list.append( + { + "team_id": f"espn:{league_id}:{tid}", + "name": t["name"], + "owner": t.get("owner", "Unknown"), + "wins": t.get("wins", 0), + "losses": t.get("losses", 0), + "ties": t.get("ties", 0), + "points_for": t.get("points_for", 0), + "points_against": t.get("points_against", 0), + "rank": t.get("rank"), + "roster": roster.to_dict(orient="records") + if isinstance(roster, pd.DataFrame) and not roster.empty + else [], + } + ) + + return { + "league": { + "league_id": f"espn:{league_id}", + "provider": "espn", + "name": (info or {}).get("name", "Unknown"), + "season": season, + "scoring_type": ((info or {}).get("scoring_type", "custom") or "custom") + .lower() + .replace("-", "_"), + "roster_size": (info or {}).get("size"), + "num_teams": len(team_list), + "playoff_teams": (info or {}).get("playoff_teams"), + }, + "teams": team_list, + "matchups": matchups, + } diff --git a/src/ffpy/ingest/output.py b/src/ffpy/ingest/output.py new file mode 100644 index 0000000..e2462c0 --- /dev/null +++ b/src/ffpy/ingest/output.py @@ -0,0 +1,81 @@ +"""Output formatters: table, JSON, CSV, and optional DB persistence.""" + +from __future__ import annotations + +import csv +import json +import logging +import sys +from typing import Any, Dict, List, Optional, TextIO + +from ffpy.database import FFPyDatabase + +logger = logging.getLogger(__name__) + + +def write_json(data: Any, file: TextIO = sys.stdout) -> None: + json.dump(data, file, indent=2, default=str) + file.write("\n") + + +def write_csv(data: List[Dict[str, Any]], file: TextIO = sys.stdout) -> None: + if not data: + return + writer = csv.DictWriter(file, fieldnames=data[0].keys()) + writer.writeheader() + writer.writerows(data) + + +def write_table(data: List[Dict[str, Any]], file: TextIO = sys.stdout) -> None: + """Simple aligned-column table output.""" + if not data: + return + headers = list(data[0].keys()) + rows = [[str(r.get(h, "")) for h in headers] for r in data] + col_widths = [max(len(h), max((len(r[i]) for r in rows), default=0)) for i, h in enumerate(headers)] + + sep = " " + header_line = sep.join(h.ljust(w) for h, w in zip(headers, col_widths)) + file.write(header_line + "\n") + file.write("-" * len(header_line) + "\n") + for row in rows: + file.write(sep.join(cell.ljust(w) for cell, w in zip(row, col_widths)) + "\n") + + +def persist_to_db(data: dict, user_id: str = "cli", db_path: Optional[str] = None) -> str: + """Store ingested league data in the FFPy SQLite database. + + Returns the league_id (prefixed string like ``espn:123456``). + """ + db = FFPyDatabase(db_path=db_path) + try: + league_id = db.store_user_league(user_id, data) + logger.info( + "Stored league %s (%d teams, %d matchups)", + league_id, + len(data.get("teams", [])), + len(data.get("matchups", [])), + ) + return league_id + finally: + db.close() + + +def format_output(data: Any, fmt: str, file: TextIO = sys.stdout) -> None: + """Dispatch to the correct formatter based on *fmt* (json|csv|table).""" + if fmt == "json": + write_json(data, file) + elif fmt == "csv": + if isinstance(data, dict): + write_csv([data], file) + elif isinstance(data, list): + write_csv(data, file) + else: + write_json(data, file) + else: + if isinstance(data, dict): + write_table([data], file) + elif isinstance(data, list): + write_table(data, file) + else: + file.write(str(data) + "\n") diff --git a/src/ffpy/ingest/sleeper.py b/src/ffpy/ingest/sleeper.py new file mode 100644 index 0000000..b3dd9b6 --- /dev/null +++ b/src/ffpy/ingest/sleeper.py @@ -0,0 +1,112 @@ +"""Sleeper league data fetcher — no auth needed.""" + +from __future__ import annotations + +import logging +from collections import defaultdict +from typing import List + +from ffpy.integrations.sleeper import SleeperIntegration + +logger = logging.getLogger(__name__) + + +def _load_sleeper_players() -> dict: + """Load Sleeper player map with caching fallback.""" + from ffpy.draft_strategy import load_sleeper_players + + return load_sleeper_players() + + +def fetch_sleeper_league(league_id: str, season: int = 2025) -> dict: + """Fetch league data from Sleeper (public API, no auth required). + + Returns the normalized ``{league, teams, matchups}`` dict used by + ``ffpy.database.FFPyDatabase.store_user_league()``. + """ + league = SleeperIntegration.get_league(league_id) + rosters = SleeperIntegration.get_rosters(league_id) + users = SleeperIntegration.get_league_users(league_id) + user_by_id = {u.get("user_id"): u for u in users} + players_map = _load_sleeper_players() + + teams: List[dict] = [] + for idx, r in enumerate(rosters): + players_list = r.get("players") or [] + owner_id = r.get("owner_id") or "" + if not owner_id and not players_list: + continue + roster_id = r.get("roster_id") + user = user_by_id.get(owner_id, {}) + metadata = user.get("metadata") or {} + fallback_id = str(roster_id) if roster_id is not None else owner_id or str(idx + 1) + team_name = ( + metadata.get("team_name") + or user.get("display_name") + or (f"Team {fallback_id}" if fallback_id else "Unknown") + ) + owner_display = user.get("display_name") or owner_id or "Unknown" + team_id_suffix = str(roster_id) if roster_id is not None else owner_id or str(idx + 1) + + teams.append( + { + "team_id": f"sleeper:{league_id}:{team_id_suffix}", + "name": team_name, + "owner": owner_display, + "wins": r.get("settings", {}).get("wins", 0), + "losses": r.get("settings", {}).get("losses", 0), + "ties": r.get("settings", {}).get("ties", 0), + "points_for": r.get("settings", {}).get("fpts", 0), + "points_against": r.get("settings", {}).get("fpts_against", 0), + "rank": None, + "roster": SleeperIntegration.enrich_roster(r.get("players", []), players_map), + } + ) + + teams.sort(key=lambda t: (-(t.get("wins") or 0), -(t.get("points_for") or 0))) + + matchups: List[dict] = [] + for week in range(1, 18): + try: + week_matchups = SleeperIntegration.get_matchups(league_id, week) + except Exception: + break + if not week_matchups: + break + by_matchup: dict = defaultdict(list) + for m in week_matchups: + matchup_id = m.get("matchup_id") + roster_id = m.get("roster_id") + if matchup_id is None or roster_id is None: + continue + by_matchup[matchup_id].append(m) + for group in by_matchup.values(): + if len(group) < 2: + continue + home, away = group[0], group[1] + matchups.append( + { + "week": week, + "home_team_id": f"sleeper:{league_id}:{home.get('roster_id')}", + "away_team_id": f"sleeper:{league_id}:{away.get('roster_id')}", + "home_score": home.get("points"), + "away_score": away.get("points"), + "is_playoff": 0, + "is_consolation": 0, + } + ) + + return { + "league": { + "league_id": f"sleeper:{league_id}", + "provider": "sleeper", + "name": league.get("name", "Unknown"), + "season": league.get("season", season), + "scoring_type": "custom", + "roster_size": None, + "num_teams": len(teams), + "playoff_teams": league.get("settings", {}).get("playoff_teams"), + }, + "teams": teams, + "matchups": matchups, + } diff --git a/src/ffpy/ingest/yahoo.py b/src/ffpy/ingest/yahoo.py new file mode 100644 index 0000000..ce6dd13 --- /dev/null +++ b/src/ffpy/ingest/yahoo.py @@ -0,0 +1,181 @@ +"""Yahoo league data fetcher with OAuth 2.0 token management.""" + +from __future__ import annotations + +import logging +import os +import time +from typing import List, Optional + +from ffpy.ingest.auth import load_yahoo_token, save_yahoo_token +from ffpy.integrations.yahoo import YahooIntegration + +logger = logging.getLogger(__name__) + + +def get_client_credentials() -> tuple[str, str, str]: + """Load Yahoo client credentials from env vars.""" + client_id = os.getenv("YAHOO_CLIENT_ID", "") + client_secret = os.getenv("YAHOO_CLIENT_SECRET", "") + redirect_uri = os.getenv("YAHOO_REDIRECT_URI", "http://127.0.0.1:8001") + if not client_id or not client_secret: + raise RuntimeError( + "Yahoo OAuth requires YAHOO_CLIENT_ID and YAHOO_CLIENT_SECRET in .env. " + "Create an app at https://developer.yahoo.com/apps/" + ) + return client_id, client_secret, redirect_uri + + +def _ensure_valid_token( + integration: YahooIntegration, + access_token: Optional[str] = None, + refresh_token: Optional[str] = None, +) -> dict: + """Return a valid token dict, refreshing or prompting if needed.""" + # Priority: 1) passed token 2) stored token file 3) interactive prompt + token = None + + if access_token: + token = { + "access_token": access_token, + "refresh_token": refresh_token or "", + "expires_at": time.time() + 3600, + } + + if not token: + token = load_yahoo_token() + + if token: + now = time.time() + expires_at = token.get("expires_at", 0) + if now >= expires_at: + rt = token.get("refresh_token", "") + if rt: + logger.info("Yahoo token expired — refreshing") + try: + new = integration.refresh_access_token(rt) + new["expires_at"] = time.time() + new.get("expires_in", 3600) + save_yahoo_token(new) + return new + except Exception as exc: + logger.warning("Token refresh failed: %s", exc) + logger.warning("Token expired and no refresh token available") + token = None + + if not token: + # Interactive OAuth flow + print("\n--- Yahoo OAuth 2.0 Authentication ---") + auth_url = integration.get_authorization_url(state="ffpy-ingest") + print(f"1. Visit this URL in your browser:\n {auth_url}") + print("2. Authorize the application") + print("3. Copy the full redirect URL and paste it below\n") + redirect_response = input("Paste redirect URL or authorization code: ").strip() + + # Extract code from URL if full URL pasted + if "code=" in redirect_response: + from urllib.parse import parse_qs, urlparse + + parsed = urlparse(redirect_response) + params = parse_qs(parsed.query) + code = params.get("code", [""])[0] + else: + code = redirect_response + + if not code: + raise RuntimeError("No authorization code provided") + + token = integration.exchange_code(code) + token["expires_at"] = time.time() + token.get("expires_in", 3600) + save_yahoo_token(token) + + return token + + +def fetch_yahoo_league( + league_id: str, + season: int = 2025, + access_token: Optional[str] = None, + refresh_token: Optional[str] = None, +) -> dict: + """Fetch league data from Yahoo Fantasy Sports via OAuth 2.0. + + Returns the normalized ``{league, teams, matchups}`` dict. + """ + client_id, client_secret, redirect_uri = get_client_credentials() + integration = YahooIntegration( + client_id=client_id, + client_secret=client_secret, + redirect_uri=redirect_uri, + ) + + token = _ensure_valid_token(integration, access_token, refresh_token) + token_str = token["access_token"] + + meta = integration.get_league_metadata(league_id, token_str) + standings = integration.get_standings(league_id, token_str) + + teams = [] + for s in standings: + team_key = s.get("team_key", "") + roster = integration.get_team_roster(team_key, token_str) if team_key else [] + teams.append( + { + "team_id": f"yahoo:{league_id}:{team_key}", + "name": s.get("name", "Unknown"), + "owner": s.get("manager", {}).get("nickname", "Unknown"), + "wins": s.get("standings", {}).get("outcome_totals", {}).get("wins", 0), + "losses": s.get("standings", {}).get("outcome_totals", {}).get("losses", 0), + "ties": s.get("standings", {}).get("outcome_totals", {}).get("ties", 0), + "points_for": s.get("standings", {}).get("points_for", 0), + "points_against": s.get("standings", {}).get("points_against", 0), + "rank": s.get("standings", {}).get("rank"), + "roster": roster if isinstance(roster, list) else [], + } + ) + + matchups: List[dict] = [] + for week in range(1, 18): + try: + week_matchups = integration.get_matchups(league_id, week, token_str) + except Exception: + break + for m in week_matchups: + teams_in = m.get("teams", {}) + home = away = None + for key, val in teams_in.items(): + if not isinstance(val, dict): + continue + t = val.get("team", []) + if isinstance(t, list) and len(t) > 1: + tk = t[0].get("team_key", "") + if home is None: + home = tk + else: + away = tk + if home and away: + matchups.append( + { + "week": week, + "home_team_id": f"yahoo:{league_id}:{home}", + "away_team_id": f"yahoo:{league_id}:{away}", + "home_score": None, + "away_score": None, + "is_playoff": 0, + "is_consolation": 0, + } + ) + + return { + "league": { + "league_id": f"yahoo:{league_id}", + "provider": "yahoo", + "name": meta.get("name", "Unknown"), + "season": season, + "scoring_type": "custom", + "roster_size": None, + "num_teams": meta.get("num_teams"), + "playoff_teams": None, + }, + "teams": teams, + "matchups": matchups, + } diff --git a/src/ffpy/integrations/espn_league.py b/src/ffpy/integrations/espn_league.py index 79e67cb..79d1f9d 100644 --- a/src/ffpy/integrations/espn_league.py +++ b/src/ffpy/integrations/espn_league.py @@ -56,16 +56,20 @@ def __init__( self.swid = swid or os.getenv("ESPN_SWID", "") self.espn_s2 = espn_s2 or os.getenv("ESPN_S2", "") - # Build cookies dict + # Only attach cookies when both SWID and espn_s2 are present. + # Sending an empty cookies dict can cause 401 on public leagues. self.cookies = {} - if self.swid: + has_auth = bool(self.swid) and bool(self.espn_s2) + if has_auth: self.cookies["swid"] = self.swid - if self.espn_s2: self.cookies["espn_s2"] = self.espn_s2 def _make_request(self, params: Dict[str, Any]) -> Dict: """ - Make authenticated request to ESPN API. + Make request to ESPN API. + + For public leagues, no cookies are sent. For private leagues, + SWID and espn_s2 cookies are included automatically. Args: params: Query parameters @@ -83,7 +87,11 @@ def _make_request(self, params: Dict[str, Any]) -> Dict: "Accept": "application/json", } - response = requests.get(url, params=params, headers=headers, cookies=self.cookies, timeout=10) + kwargs = {"params": params, "headers": headers, "timeout": 10} + if self.cookies: + kwargs["cookies"] = self.cookies + + response = requests.get(url, **kwargs) response.raise_for_status() return response.json() diff --git a/tests/test_cfbverse.py b/tests/test_cfbverse.py index fbdec5d..e85275b 100644 --- a/tests/test_cfbverse.py +++ b/tests/test_cfbverse.py @@ -4,6 +4,7 @@ import sqlite3 from pathlib import Path +from unittest.mock import patch import pandas as pd @@ -154,14 +155,17 @@ def test_normalize_cfb_plays_renames_key_columns(): assert row["home_team"] == "Alabama" -def test_cfb_roster_availability_message_for_unpublished_season(): +@patch("ffpy.cfbverse.get_cfb_roster_seasons", return_value=frozenset({2023, 2024})) +def test_cfb_roster_availability_message_for_unpublished_season(_mock_seasons): message = cfb_roster_availability_message(2025) assert message is not None assert "2025" in message assert "--skip-rosters" in message + assert "SEASON=2024" in message -def test_cfb_roster_availability_message_for_published_season(): +@patch("ffpy.cfbverse.get_cfb_roster_seasons", return_value=frozenset({2023, 2024})) +def test_cfb_roster_availability_message_for_published_season(_mock_seasons): assert cfb_roster_availability_message(2024) is None assert position_from_id("17") == "QB" assert position_from_id("45") == "WR" diff --git a/tests/test_ingest_auth.py b/tests/test_ingest_auth.py new file mode 100644 index 0000000..3eda1c9 --- /dev/null +++ b/tests/test_ingest_auth.py @@ -0,0 +1,88 @@ +"""Tests for ingest auth helpers.""" + +from __future__ import annotations + +import json + +from ffpy.ingest import auth + + +class TestYahooToken: + def test_save_and_load(self, tmp_path): + auth.TOKEN_DIR = tmp_path + auth.TOKEN_FILE = tmp_path / "yahoo_token.json" + + token = {"access_token": "test", "expires_at": 9999999999} + auth.save_yahoo_token(token) + + assert oct(auth.TOKEN_FILE.stat().st_mode & 0o777) == "0o600" + + loaded = auth.load_yahoo_token() + assert loaded["access_token"] == "test" + + def test_load_missing(self, tmp_path): + auth.TOKEN_DIR = tmp_path + auth.TOKEN_FILE = tmp_path / "yahoo_token.json" + + assert auth.load_yahoo_token() is None + + def test_load_corrupt(self, tmp_path): + auth.TOKEN_DIR = tmp_path + auth.TOKEN_FILE = tmp_path / "yahoo_token.json" + auth.TOKEN_FILE.write_text("{bad json") + + assert auth.load_yahoo_token() is None + + def test_expired_token(self, tmp_path): + auth.TOKEN_DIR = tmp_path + auth.TOKEN_FILE = tmp_path / "yahoo_token.json" + auth.save_yahoo_token({"access_token": "old", "expires_at": 0}) + + loaded = auth.load_yahoo_token() + assert loaded is not None # Still returned; caller should refresh + + def test_delete(self, tmp_path): + auth.TOKEN_DIR = tmp_path + auth.TOKEN_FILE = tmp_path / "yahoo_token.json" + auth.save_yahoo_token({"access_token": "test"}) + auth.delete_yahoo_token() + assert not auth.TOKEN_FILE.exists() + + +class TestEspnCookies: + def test_env_vars_take_priority(self, tmp_path, monkeypatch): + monkeypatch.setenv("ESPN_SWID", "{ENV-SWID}") + monkeypatch.setenv("ESPN_S2", "ENV-S2") + + auth.COOKIE_FILE = tmp_path / "espn_cookies.json" + auth.COOKIE_FILE.write_text(json.dumps({"swid": "{FILE-SWID}", "espn_s2": "FILE-S2"})) + + swid, s2 = auth.load_espn_cookies() + assert swid == "{ENV-SWID}" + assert s2 == "ENV-S2" + + def test_fallback_to_file(self, tmp_path, monkeypatch): + monkeypatch.delenv("ESPN_SWID", raising=False) + monkeypatch.delenv("ESPN_S2", raising=False) + + auth.COOKIE_FILE = tmp_path / "espn_cookies.json" + auth.COOKIE_FILE.write_text(json.dumps({"swid": "{FILE-SWID}", "espn_s2": "FILE-S2"})) + + swid, s2 = auth.load_espn_cookies() + assert swid == "{FILE-SWID}" + assert s2 == "FILE-S2" + + def test_save_and_delete(self, tmp_path): + auth.TOKEN_DIR = tmp_path + auth.COOKIE_FILE = tmp_path / "espn_cookies.json" + + auth.save_espn_cookies("{SWID}", "S2VAL") + assert auth.COOKIE_FILE.exists() + assert oct(auth.COOKIE_FILE.stat().st_mode & 0o777) == "0o600" + + data = json.loads(auth.COOKIE_FILE.read_text()) + assert data["swid"] == "{SWID}" + assert data["espn_s2"] == "S2VAL" + + auth.delete_espn_cookies() + assert not auth.COOKIE_FILE.exists() diff --git a/tests/test_ingest_cli.py b/tests/test_ingest_cli.py new file mode 100644 index 0000000..1ce9a67 --- /dev/null +++ b/tests/test_ingest_cli.py @@ -0,0 +1,89 @@ +"""Tests for the ffpy-ingest CLI argument parsing.""" + +from __future__ import annotations + +from ffpy.ingest.cli import build_parser + + +class TestCliArgparse: + def setup_method(self): + self.parser = build_parser() + + def test_espn_defaults(self): + args = self.parser.parse_args(["espn", "123456"]) + assert args.command == "espn" + assert args.league_id == "123456" + assert args.season == 2025 + assert args.format == "table" + assert args.swid is None + assert args.s2 is None + assert args.db is None + + def test_espn_custom_season(self): + args = self.parser.parse_args(["espn", "123456", "--season", "2024"]) + assert args.season == 2024 + + def test_espn_with_cookies(self): + args = self.parser.parse_args(["espn", "123456", "--swid", "{swid}", "--s2", "s2val"]) + assert args.swid == "{swid}" + assert args.s2 == "s2val" + + def test_yahoo_defaults(self): + args = self.parser.parse_args(["yahoo", "389.l.12345"]) + assert args.command == "yahoo" + assert args.league_id == "389.l.12345" + assert args.season == 2025 + + def test_sleeper_defaults(self): + args = self.parser.parse_args(["sleeper", "league123"]) + assert args.command == "sleeper" + assert args.league_id == "league123" + assert args.season == 2025 + + def test_json_flag_after_subcommand(self): + args = self.parser.parse_args(["espn", "123456", "--json"]) + assert args.format == "json" + + def test_csv_flag_after_subcommand(self): + args = self.parser.parse_args(["yahoo", "389.l.12345", "--csv"]) + assert args.format == "csv" + + def test_db_flag_with_path_after_subcommand(self): + args = self.parser.parse_args(["sleeper", "league1", "--db", "/tmp/test.db"]) + assert args.db == "/tmp/test.db" + + def test_db_flag_alone_means_default_path(self): + args = self.parser.parse_args(["sleeper", "league1", "--db"]) + assert args.db == "" + + def test_yahoo_auth(self): + args = self.parser.parse_args(["yahoo-auth"]) + assert args.command == "yahoo-auth" + + def test_yahoo_token(self): + args = self.parser.parse_args(["yahoo-token", "--code", "abc123"]) + assert args.command == "yahoo-token" + assert args.code == "abc123" + + def test_leagues_list(self): + args = self.parser.parse_args(["leagues-list"]) + assert args.command == "leagues-list" + + def test_leagues_info(self): + args = self.parser.parse_args(["leagues-info", "espn:123456"]) + assert args.command == "leagues-info" + assert args.id == "espn:123456" + + def test_roster(self): + args = self.parser.parse_args(["roster", "espn:123456", "espn:123456:1"]) + assert args.command == "roster" + assert args.league_id == "espn:123456" + assert args.team_id == "espn:123456:1" + + def test_matchups(self): + args = self.parser.parse_args(["matchups", "espn:123456", "3"]) + assert args.command == "matchups" + assert args.league_id == "espn:123456" + assert args.week == 3 + matchups_parser = self.parser.parse_args(["matchups", "espn:1", "1"]) + assert matchups_parser.week == 1 diff --git a/tests/test_ingest_espn.py b/tests/test_ingest_espn.py new file mode 100644 index 0000000..3e50444 --- /dev/null +++ b/tests/test_ingest_espn.py @@ -0,0 +1,198 @@ +"""Tests for ESPN ingest module.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import pytest +import requests + +from ffpy.ingest.espn import fetch_espn_league + + +def _make_mock_response(status=200, data=None): + resp = MagicMock(spec=requests.Response) + resp.status_code = status + resp.ok = status < 400 + resp.json.return_value = data or {} + if status >= 400: + resp.raise_for_status.side_effect = requests.HTTPError(f"{status} Error", response=resp) + return resp + + +class TestFetchEspnLeague: + @patch("ffpy.integrations.espn_league.requests.get") + def test_public_league(self, mock_get): + """Public league should work without cookies.""" + league_data = { + "settings": {"name": "Public League", "playoffTeamCount": 4}, + "teams": [ + { + "id": 1, + "name": "Team 1", + "primaryOwner": "Alice", + "record": { + "overall": {"wins": 5, "losses": 2, "ties": 0, "pointsFor": 800, "pointsAgainst": 700} + }, + } + ], + "schedule": [ + { + "matchupPeriodId": 1, + "home": {"teamId": 1, "totalPoints": 110}, + "away": {"teamId": 2, "totalPoints": 90}, + "winner": "home", + } + ], + } + side_effects = [] + for _ in range(3): + side_effects.append(_make_mock_response(200, league_data)) + for _ in range(17): + side_effects.append(_make_mock_response(200, {"schedule": []})) + + mock_get.side_effect = side_effects + + data = fetch_espn_league("123456", season=2024, interactive=False) + + assert data["league"]["provider"] == "espn" + assert data["league"]["league_id"] == "espn:123456" + assert data["league"]["name"] == "Public League" + assert len(data["teams"]) == 1 + assert data["teams"][0]["name"] == "Team 1" + + @patch("ffpy.integrations.espn_league.requests.get") + def test_private_league_with_cookies(self, mock_get): + """Private league should use cookies when provided.""" + league_data = { + "settings": {"name": "Private League", "playoffTeamCount": 4}, + "teams": [], + "schedule": [], + } + side_effects = [ + _make_mock_response(401), # public get_league_info fails immediately + ] + # Private attempt with cookies: mSettings, mTeam, mRoster + for _ in range(3): + side_effects.append(_make_mock_response(200, league_data)) + for _ in range(17): + side_effects.append(_make_mock_response(200, {"schedule": []})) + + mock_get.side_effect = side_effects + + data = fetch_espn_league( + "123456", + season=2024, + swid="{TEST-SWID}", + espn_s2="TEST-S2", + interactive=False, + ) + + assert data["league"]["provider"] == "espn" + assert data["league"]["name"] == "Private League" + + # Verify the first request after public failure had cookies + calls = mock_get.call_args_list + auth_calls = [c for c in calls if "cookies" in c.kwargs and c.kwargs["cookies"]] + assert len(auth_calls) > 0 + assert auth_calls[0].kwargs["cookies"]["swid"] == "{TEST-SWID}" + + @patch("ffpy.integrations.espn_league.requests.get") + def test_private_league_no_cookies_raises(self, mock_get): + """Private league with no cookies should raise.""" + mock_get.side_effect = [_make_mock_response(401)] + + with pytest.raises(RuntimeError, match="private"): + fetch_espn_league("123456", season=2024, interactive=False) + + @patch("ffpy.integrations.espn_league.requests.get") + def test_non_auth_http_error_is_reraised(self, mock_get): + """Non-auth failures should not be treated as a private league.""" + mock_get.side_effect = [_make_mock_response(500)] + + with pytest.raises(requests.HTTPError, match="500"): + fetch_espn_league("123456", season=2024, interactive=False) + + @patch("ffpy.integrations.espn_league.requests.get") + def test_normalized_data_shape(self, mock_get): + """Verify the output dict matches DB schema expectations.""" + league_data = { + "settings": {"name": "Test League", "playoffTeamCount": 6}, + "teams": [ + { + "id": 1, + "name": "Team A", + "abbrev": "TA", + "primaryOwner": "Alice", + "record": { + "overall": {"wins": 5, "losses": 2, "ties": 0, "pointsFor": 800, "pointsAgainst": 700} + }, + } + ], + "schedule": [], + } + side_effects = [] + for _ in range(3): + side_effects.append(_make_mock_response(200, league_data)) + for _ in range(17): + side_effects.append(_make_mock_response(200, {"schedule": []})) + + mock_get.side_effect = side_effects + + data = fetch_espn_league("123456", season=2024, interactive=False) + + league = data["league"] + assert league["league_id"] == "espn:123456" + assert league["provider"] == "espn" + assert league["season"] == 2024 + assert isinstance(league["scoring_type"], str) + + for team in data["teams"]: + assert "team_id" in team + assert "name" in team + assert "wins" in team + assert "losses" in team + assert "points_for" in team + assert "roster" in team + + for m in data["matchups"]: + assert "week" in m + assert "home_team_id" in m + assert "away_team_id" in m + assert "is_playoff" in m + + @patch("ffpy.integrations.espn_league.requests.get") + def test_matchups_stop_on_empty(self, mock_get): + """Matchup fetching should stop when ESPN returns empty/future weeks.""" + league_data = { + "settings": {"name": "Test"}, + "teams": [{"id": 1, "record": {"overall": {}}}], + "schedule": [], + } + responses = [] + # 3 calls for league info, teams, rosters + for _ in range(3): + responses.append(_make_mock_response(200, league_data)) + # Only week 1 has data + responses.append( + _make_mock_response( + 200, + { + "schedule": [ + { + "matchupPeriodId": 1, + "home": {"teamId": 1, "totalPoints": 100}, + "away": {"teamId": 2, "totalPoints": 90}, + } + ] + }, + ) + ) + # Week 2 is empty → stops + responses.append(_make_mock_response(200, {"schedule": []})) + + mock_get.side_effect = responses + + data = fetch_espn_league("123456", season=2024, interactive=False) + assert len(data["matchups"]) == 1 + assert data["matchups"][0]["week"] == 1 diff --git a/tests/test_ingest_output.py b/tests/test_ingest_output.py new file mode 100644 index 0000000..05fd28c --- /dev/null +++ b/tests/test_ingest_output.py @@ -0,0 +1,143 @@ +"""Tests for ingest output formatters.""" + +from __future__ import annotations + +import csv +import io +import json + +from ffpy.ingest.output import format_output, persist_to_db, write_csv, write_json, write_table + + +class TestWriteJson: + def test_dict_output(self): + buf = io.StringIO() + write_json({"a": 1, "b": 2}, buf) + result = json.loads(buf.getvalue()) + assert result == {"a": 1, "b": 2} + + def test_list_output(self): + buf = io.StringIO() + write_json([{"x": 1}, {"x": 2}], buf) + result = json.loads(buf.getvalue()) + assert len(result) == 2 + + +class TestWriteCsv: + def test_basic_csv(self): + buf = io.StringIO() + write_csv([{"name": "Alice", "score": "10"}, {"name": "Bob", "score": "20"}], buf) + buf.seek(0) + reader = csv.DictReader(buf) + rows = list(reader) + assert len(rows) == 2 + assert rows[0]["name"] == "Alice" + + def test_empty_list(self): + buf = io.StringIO() + write_csv([], buf) + assert buf.getvalue() == "" + + +class TestWriteTable: + def test_basic_table(self): + buf = io.StringIO() + write_table([{"name": "Alice", "wins": "5"}, {"name": "Bob", "wins": "3"}], buf) + output = buf.getvalue() + assert "Alice" in output + assert "name" in output + + def test_empty(self): + buf = io.StringIO() + write_table([], buf) + assert buf.getvalue() == "" + + +class TestFormatOutput: + def test_json_format(self): + buf = io.StringIO() + format_output({"key": "val"}, "json", buf) + assert json.loads(buf.getvalue()) == {"key": "val"} + + def test_csv_format_list(self): + buf = io.StringIO() + format_output([{"a": "1"}], "csv", buf) + assert "a" in buf.getvalue() + + def test_table_format_default(self): + buf = io.StringIO() + format_output([{"a": "1"}], "table", buf) + assert "a" in buf.getvalue() + + +class TestPersistToDb: + def test_store_and_retrieve(self, tmp_path): + db_path = str(tmp_path / "test.db") + data = { + "league": { + "league_id": "sleeper:test123", + "provider": "sleeper", + "name": "Test League", + "season": 2024, + "scoring_type": "ppr", + "roster_size": None, + "num_teams": 2, + "playoff_teams": None, + }, + "teams": [ + { + "team_id": "sleeper:test123:1", + "name": "Team A", + "owner": "Alice", + "wins": 5, + "losses": 2, + "ties": 0, + "points_for": 800.0, + "points_against": 700.0, + "rank": 1, + "roster": [], + }, + { + "team_id": "sleeper:test123:2", + "name": "Team B", + "owner": "Bob", + "wins": 3, + "losses": 4, + "ties": 0, + "points_for": 650.0, + "points_against": 750.0, + "rank": 2, + "roster": [], + }, + ], + "matchups": [ + { + "week": 1, + "home_team_id": "sleeper:test123:1", + "away_team_id": "sleeper:test123:2", + "home_score": 110.5, + "away_score": 98.3, + "is_playoff": 0, + "is_consolation": 0, + } + ], + } + league_id = persist_to_db(data, user_id="test_user", db_path=db_path) + assert league_id == "sleeper:test123" + + from ffpy.database import FFPyDatabase + + db = FFPyDatabase(db_path=db_path) + try: + leagues = db.get_user_leagues("test_user") + assert len(leagues) == 1 + assert leagues[0]["league_id"] == "sleeper:test123" + + teams = db.get_league_teams(league_id, "test_user") + assert len(teams) == 2 + + matchups = db.get_league_matchups(league_id, 1, "test_user") + assert len(matchups) == 1 + assert matchups[0]["home_score"] == 110.5 + finally: + db.close() diff --git a/tests/test_ingest_sleeper.py b/tests/test_ingest_sleeper.py new file mode 100644 index 0000000..ba7aa97 --- /dev/null +++ b/tests/test_ingest_sleeper.py @@ -0,0 +1,159 @@ +"""Tests for Sleeper ingest module.""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +import requests + +from ffpy.ingest.sleeper import fetch_sleeper_league + + +def _mock_json(data): + resp = MagicMock(spec=requests.Response) + resp.status_code = 200 + resp.ok = True + resp.json.return_value = data + return resp + + +class TestFetchSleeperLeague: + @patch("ffpy.integrations.sleeper.requests.get") + @patch("ffpy.ingest.sleeper._load_sleeper_players") + def test_basic_league(self, mock_players, mock_get): + """Verify normal Sleeper league import.""" + mock_players.return_value = { + "1234": {"full_name": "Patrick Mahomes", "position": "QB", "team": "KC"}, + "5678": {"full_name": "Justin Jefferson", "position": "WR", "team": "MIN"}, + } + + mock_get.side_effect = [ + _mock_json( + {"name": "Test League", "season": 2024, "total_rosters": 2, "settings": {"playoff_teams": 4}} + ), + _mock_json( + [ + { + "roster_id": 1, + "owner_id": "user_a", + "players": ["1234", "5678"], + "settings": {"wins": 5, "losses": 2, "ties": 0, "fpts": 800, "fpts_against": 700}, + }, + { + "roster_id": 2, + "owner_id": "user_b", + "players": [], + "settings": {"wins": 3, "losses": 4, "ties": 0, "fpts": 650, "fpts_against": 750}, + }, + ] + ), + _mock_json( + [ + {"user_id": "user_a", "display_name": "Alice", "metadata": {"team_name": "A-Team"}}, + {"user_id": "user_b", "display_name": "Bob", "metadata": {}}, + ] + ), + _mock_json( + [ + {"matchup_id": 1, "roster_id": 1, "points": 110.5}, + {"matchup_id": 1, "roster_id": 2, "points": 98.3}, + ] + ), + ] + + data = fetch_sleeper_league("test_league", season=2024) + + assert data["league"]["provider"] == "sleeper" + assert data["league"]["name"] == "Test League" + assert data["league"]["season"] == 2024 + assert data["league"]["num_teams"] == 2 + + assert len(data["teams"]) == 2 + team_a = data["teams"][0] + assert team_a["name"] == "A-Team" + assert team_a["owner"] == "Alice" + assert team_a["wins"] == 5 + assert len(team_a["roster"]) == 2 + assert team_a["roster"][0]["player"] == "Patrick Mahomes" + + assert len(data["matchups"]) == 1 + assert data["matchups"][0]["home_score"] == 110.5 + assert data["matchups"][0]["away_score"] == 98.3 + + @patch("ffpy.integrations.sleeper.requests.get") + @patch("ffpy.ingest.sleeper._load_sleeper_players") + def test_empty_roster_slot_skipped(self, mock_players, mock_get): + """Roster slots with no owner and no players should be skipped.""" + mock_players.return_value = {} + + mock_get.side_effect = [ + _mock_json({"name": "Test League", "season": 2024, "total_rosters": 3, "settings": {}}), + _mock_json( + [ + { + "roster_id": 1, + "owner_id": "user_a", + "players": ["1"], + "settings": {"wins": 1, "losses": 0, "ties": 0, "fpts": 100, "fpts_against": 50}, + }, + { + "roster_id": 2, + "owner_id": "", + "players": [], + "settings": {"wins": 0, "losses": 0, "ties": 0, "fpts": 0, "fpts_against": 0}, + }, + ] + ), + _mock_json([{"user_id": "user_a", "display_name": "Alice"}]), + _mock_json([]), + ] + + data = fetch_sleeper_league("test_league", season=2024) + assert len(data["teams"]) == 1 + assert len(data["matchups"]) == 0 + + @patch("ffpy.integrations.sleeper.requests.get") + @patch("ffpy.ingest.sleeper._load_sleeper_players") + def test_teams_sorted_by_record(self, mock_players, mock_get): + """Teams should be sorted by wins descending, then points_for.""" + mock_players.return_value = {} + + mock_get.side_effect = [ + _mock_json({"name": "Test", "season": 2024, "total_rosters": 3, "settings": {}}), + _mock_json( + [ + { + "roster_id": 1, + "owner_id": "a", + "players": ["1"], + "settings": {"wins": 5, "losses": 2, "ties": 0, "fpts": 800, "fpts_against": 700}, + }, + { + "roster_id": 2, + "owner_id": "b", + "players": ["2"], + "settings": {"wins": 5, "losses": 2, "ties": 0, "fpts": 750, "fpts_against": 720}, + }, + { + "roster_id": 3, + "owner_id": "c", + "players": ["3"], + "settings": {"wins": 3, "losses": 4, "ties": 0, "fpts": 650, "fpts_against": 750}, + }, + ] + ), + _mock_json( + [ + {"user_id": "a", "display_name": "Alice"}, + {"user_id": "b", "display_name": "Bob"}, + {"user_id": "c", "display_name": "Charlie"}, + ] + ), + _mock_json([]), + ] + + data = fetch_sleeper_league("test_league", season=2024) + assert len(data["teams"]) == 3 + assert data["teams"][0]["name"] == "Alice" + assert data["teams"][1]["name"] == "Bob" + assert data["teams"][2]["name"] == "Charlie"