src/yahoo_client.py

import asyncio
import pandas as pd
from pydantic import BaseModel
import re
from typing import List, Mapping, Tuple, Union, Optional
import xml.etree.ElementTree as ET
from streamlit_oauth import OAuth2Component


class LeagueSettings(BaseModel):
    league_key: str
    league_id: str
    name: str
    num_teams: int
    current_week: int
    season: int
    playoff_start_week: int
    num_playoff_teams: int
    num_playoff_consolation_teams: int
    roster_positions: list[dict]

    @classmethod
    def from_xml(cls, xml_settings: ET.Element) -> "LeagueSettings":
        base_fields_list = [
            "league_key",
            "league_id",
            "name",
            "num_teams",
            "current_week",
            "season",
        ]
        base_fields_dict = {f: xml_settings.findtext(f"./league/{f}") for f in base_fields_list}

        settings_fields_list = [
            "playoff_start_week",
            "num_playoff_teams",
            "num_playoff_consolation_teams",
        ]
        settings_fields_dict = {f: xml_settings.findtext(f"./league/settings/{f}") for f in settings_fields_list}
        league_settings_dict = {**base_fields_dict, **settings_fields_dict}

        roster_fields_list = [
            "position",
            "position_type",
            "count",
            "is_starting_position",
        ]

        roster_positions = [
            {f: x.findtext(f"./{f}") for f in roster_fields_list}
            for x in xml_settings.findall("./league/settings/roster_positions/*")
        ]

        return cls(
            league_key=league_settings_dict["league_key"] or "",
            league_id=league_settings_dict["league_id"] or "",
            name=league_settings_dict["name"] or "",
            num_teams=int(league_settings_dict["num_teams"] or 0),
            current_week=int(league_settings_dict["current_week"] or 0),
            season=int(league_settings_dict["season"] or 0),
            playoff_start_week=int(league_settings_dict["playoff_start_week"] or 0),
            num_playoff_teams=int(league_settings_dict["num_playoff_teams"] or 0),
            num_playoff_consolation_teams=int(league_settings_dict["num_playoff_consolation_teams"] or 0),
            roster_positions=roster_positions,
        )


class YahooFantasyClient:
    def __init__(self, oauth: OAuth2Component, token):
        self.oauth: OAuth2Component = oauth
        self.token = token

    def authorize_yahoo_from_client_json(self) -> None:
        self.token = self.oauth.refresh_token(self.token)

    def yahoo_request_to_xml(self, url: str) -> Tuple[ET.Element, str]:
        self.authorize_yahoo_from_client_json()
        client = self.oauth.client.get_httpx_client()
        r = asyncio.run(
            client.get(
                url,
                headers={"Authorization": f"Bearer {self.token.get('access_token')}"},
            )
        )

        xmlstring = r.text
        xmlstring = re.sub(' xmlns="[^"]+"', "", xmlstring, count=1)
        root = ET.fromstring(xmlstring)
        return root, xmlstring

    def find_all_leagues_for_logged_in_user(self, season: Optional[str]) -> List[str]:
        if season:
            season_str = f";seasons={season}"
        else:
            season_str = ""
        url = (
            f"https://fantasysports.yahooapis.com/fantasy/v2/users;use_login=1/games{season_str};game_codes=nfl/leagues"
        )
        root, _ = self.yahoo_request_to_xml(url)
        league_keys = list(
            filter(None, [x.text for x in root.findall("./users/user/games/game/leagues/league/league_key")])
        )
        return league_keys

    def get_guid_for_logged_in_user(self) -> str | None:
        url = "https://fantasysports.yahooapis.com/fantasy/v2/users;use_login=1/games;game_keys=nfl/teams"
        root, _ = self.yahoo_request_to_xml(url)
        user_guid = root.findtext("./users/user/guid")
        return user_guid

    def parse_matchup(self, matchup: ET.Element, match_index: int) -> List[Mapping[str, Union[str, float]]]:
        matchup_info = Matchup.from_xml(matchup, match_index)
        return matchup_info.to_list_team_dict()

    def parse_league_settings(self, league_key: str) -> LeagueSettings:
        url = f"https://fantasysports.yahooapis.com/fantasy/v2/league/{league_key}/settings"
        league_settings, _ = self.yahoo_request_to_xml(url)
        parsed_league_settings = LeagueSettings.from_xml(league_settings)
        return parsed_league_settings

    def get_draft(self, league_key: str):
        url = f"https://fantasysports.yahooapis.com/fantasy/v2/league/{league_key}/draftresults"
        draft_results_xml, _ = self.yahoo_request_to_xml(url)
        parsed_draft = [
            {
                "pick": x.findtext("./pick"),
                "round": x.findtext("./round"),
                "team_key": x.findtext("./team_key"),
                "player_key": x.findtext("./player_key"),
            }
            for x in draft_results_xml.findall("./league/draft_results/*")
        ]
        df = pd.DataFrame(parsed_draft)
        for col in ["round", "pick"]:
            if col in df:
                df[col] = df[col].astype(int)
        return df

    def parse_weeks_matchups(self, week: str, league_key: str) -> List[Mapping[str, Union[str, float]]]:
        url = f"https://fantasysports.yahooapis.com/fantasy/v2/leagues;league_keys={league_key}/scoreboard;week={week}"
        week_scoreboard, _ = self.yahoo_request_to_xml(url)
        week_matchups = week_scoreboard.findall("./leagues/league/scoreboard/matchups/matchup")
        weekly_scores = []
        for match_index, matchup in enumerate(week_matchups):
            matchup_result = self.parse_matchup(matchup, match_index)
            weekly_scores.extend(matchup_result)
        return weekly_scores

    def full_schedule_dataframe(self, league_key: str) -> pd.DataFrame:
        league_settings = self.parse_league_settings(league_key)
        all_weeks = ",".join([str(w) for w in range(1, league_settings.playoff_start_week)])
        df = pd.DataFrame(self.parse_weeks_matchups(week=all_weeks, league_key=league_key))
        return df

    def get_all_logged_in_user_league_settings(self, season: str) -> list[LeagueSettings]:
        all_league_keys = self.find_all_leagues_for_logged_in_user(season)
        return [self.parse_league_settings(lk) for lk in all_league_keys]


class MatchupTeam(BaseModel):
    team_id: str
    team_name: str
    team_points: float
    win_probability: float
    team_projected_points: float

    @classmethod
    def from_xml(cls, xml_matchup_team: ET.Element) -> "MatchupTeam":
        team_id = xml_matchup_team.findtext("./team_id")
        team_name = xml_matchup_team.findtext("./name")
        team_points = xml_matchup_team.findtext("./team_points/total")
        win_probability = xml_matchup_team.findtext("./win_probability")
        team_projected_points = xml_matchup_team.findtext("./team_projected_points/total")
        return cls(
            team_id=team_id or "",
            team_name=team_name or "",
            team_points=float(team_points or 0),
            win_probability=float(win_probability or 0),
            team_projected_points=float(team_projected_points or 0),
        )


class Matchup(BaseModel):
    matchup_status: str
    is_playoffs: str
    is_consolation: str
    week: float
    match_index: int
    matchup_teams: List[MatchupTeam]

    @classmethod
    def from_xml(cls, xml_matchup: ET.Element, match_index: int) -> "Matchup":
        matchup_status = xml_matchup.findtext("./status")
        is_playoffs = xml_matchup.findtext("./is_playoffs")
        is_consolation = xml_matchup.findtext("./is_consolation")
        week = xml_matchup.findtext("./week")

        xml_matchup_teams = xml_matchup.findall("./teams/team")
        matchup_teams = [MatchupTeam.from_xml(team) for team in xml_matchup_teams]
        return cls(
            matchup_status=matchup_status or "",
            is_playoffs=is_playoffs or "",
            is_consolation=is_consolation or "",
            week=float(week or 0),
            match_index=match_index,
            matchup_teams=matchup_teams,
        )

    def to_list_team_dict(self) -> List[Mapping[str, Union[str, float]]]:
        matchup_info_dict = self.dict(exclude={"matchup_teams"})
        out_list: List[Mapping[str, Union[str, float]]] = []
        for team in self.matchup_teams:
            team_dict = team.dict()
            team_dict.update(matchup_info_dict)
            out_list.append(team_dict)
        return out_list