Copy yahoo_client from yahoo-ff-dev

src/yahoo_client.py

@@ -0,0 +1,164 @@
+import pandas as pd
+from pydantic import BaseModel
+import re
+from typing import List, Mapping, Tuple, Union
+import xml.etree.ElementTree as ET
+from yahoo_oauth import OAuth2
+
+
+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
+
+    @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}
+        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),
+        )
+
+
+class YahooFantasyClient:
+    def get_new_oauth_from_json(self) -> OAuth2:
+        return OAuth2(None, None, from_file=self.client_json_path)
+
+    def __init__(self, client_json_path: str):
+        self.client_json_path = client_json_path
+        self.oauth: OAuth2 = self.get_new_oauth_from_json()
+
+    def authorize_yahoo_from_client_json(self) -> None:
+        if not self.oauth.token_is_valid():
+            self.oauth.refresh_access_token()
+
+    def yahoo_request_to_xml(self, url: str) -> Tuple[ET.Element, str]:
+        self.authorize_yahoo_from_client_json()
+        r = self.oauth.session.get(url)
+
+        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) -> List[str]:
+        url = "https://fantasysports.yahooapis.com/fantasy/v2/users;use_login=1/games;game_keys=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 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 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
+
+
+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