Coverage for bundesliga_tippspiel/background/openligadb.py: 64%
Shortcuts on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
Shortcuts on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1"""LICENSE
2Copyright 2017 Hermann Krumrey <hermann@krumreyh.com>
4This file is part of bundesliga-tippspiel.
6bundesliga-tippspiel is free software: you can redistribute it and/or modify
7it under the terms of the GNU General Public License as published by
8the Free Software Foundation, either version 3 of the License, or
9(at your option) any later version.
11bundesliga-tippspiel is distributed in the hope that it will be useful,
12but WITHOUT ANY WARRANTY; without even the implied warranty of
13MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14GNU General Public License for more details.
16You should have received a copy of the GNU General Public License
17along with bundesliga-tippspiel. If not, see <http://www.gnu.org/licenses/>.
18LICENSE"""
20import time
21import json
22import requests
23from datetime import datetime
24from typing import Dict, Any, Optional, Tuple, List
25from jerrycan.base import db, app
26from bundesliga_tippspiel.db.match_data.Match import Match
27from bundesliga_tippspiel.db.match_data.Goal import Goal
28from bundesliga_tippspiel.db.match_data.Player import Player
29from bundesliga_tippspiel.db.match_data.Team import Team
30from bundesliga_tippspiel.Config import Config
31from bundesliga_tippspiel.utils.teams import get_team_data
34class UpdateTracker:
35 """
36 Class that keeps track of OpenLigaDB updates, which is useful to avoid
37 rate limiting
38 """
39 UPDATES: Dict[Tuple[str, int], int] = {}
41 @staticmethod
42 def update_required(league: str, season: int) -> bool:
43 """
44 Checks if a league requires updating
45 :param league: The league to check
46 :param season: The season to check
47 :return: True if update required, False otherwise
48 """
49 league_tuple = (league, season)
50 last_update = UpdateTracker.UPDATES.get(league_tuple, 0)
52 if last_update == 0:
53 return True # Update on startup
55 now = datetime.utcnow()
56 delta = time.time() - last_update
58 matches: List[Match] = Match.query.filter_by(
59 season=season, league=league
60 ).all()
61 unfinished_matches = [x for x in matches if not x.finished]
63 if len(matches) == 0:
64 app.logger.debug("Initial Filling of database")
65 return True
67 unfinished_matches.sort(key=lambda x: x.kickoff)
68 if len(unfinished_matches) == 0:
69 app.logger.debug("Season ended, no update required")
70 return False
72 started_matches = [
73 match for match in unfinished_matches
74 if match.has_started
75 ]
76 is_primary = \
77 league_tuple == (Config.OPENLIGADB_LEAGUE, Config.season())
79 if delta > 60 * 60 * 24: # Once a day minimum update
80 app.logger.debug("Minimum daily update")
81 return True
82 elif is_primary and delta > 60 * 60:
83 app.logger.debug("Minimum hourly update for primary league")
84 return True
85 elif len(started_matches) > 0:
86 last_started_match = started_matches[-1]
87 last_started_delta = now - last_started_match.kickoff_datetime
88 last_started_seconds_delta = last_started_delta.seconds
89 # Increase update frequency during and after matches
90 if last_started_seconds_delta < 60 * 180:
91 app.logger.debug("Updating because of matches in progress")
92 return True
93 else:
94 app.logger.warning("Some games have started 180+ "
95 "minutes ago but have not ended")
96 return False
97 else:
98 app.logger.debug("No update required")
99 return False
102def update_openligadb():
103 """
104 Updates all OpenLigaDB leagues in the configuration
105 :return: None
106 """
107 start = time.time()
108 app.logger.debug("Updating OpenLigaDB data")
109 for league, season in Config.all_leagues():
110 update_required = UpdateTracker.update_required(league, season)
111 if update_required:
112 UpdateTracker.UPDATES[(league, season)] = int(time.time())
113 update_match_data(league, str(season))
114 app.logger.debug(
115 f"Finished OpenLigaDB update in {time.time() - start:.2f}s"
116 )
119def update_match_data(
120 league: Optional[str] = None,
121 season: Optional[str] = None
122):
123 """
124 Updates the database with the match data for
125 the specified league and season using openligadb data
126 :param league: The league for which to update the data
127 :param season: The season for which to update the data
128 :return: None
129 """
130 app.logger.info(f"Updating match data for {league}/{season}")
132 if league is None:
133 league = Config.OPENLIGADB_LEAGUE
134 if season is None:
135 season = Config.OPENLIGADB_SEASON
137 # Fetch Data
138 base_url = "https://www.openligadb.de/api/{}/{}/{}"
139 try:
140 team_data = json.loads(requests.get(
141 base_url.format("getavailableteams", league, season)
142 ).text)
143 match_data = json.loads(requests.get(
144 base_url.format("getmatchdata", league, season)
145 ).text)
146 except (ConnectionError, requests.exceptions.ReadTimeout):
147 app.logger.warning("Failed to update match data due to failed request")
148 return
150 for team_info in team_data:
151 team = parse_team(team_info)
152 db.session.merge(team)
154 for match_info in match_data:
155 match = parse_match(match_info, league, int(season))
156 match = db.session.merge(match)
158 home_score = 0
159 for goal_data in match_info["Goals"]:
160 goal = parse_goal(goal_data, match)
161 if goal is None:
162 continue
164 if home_score < goal.home_score:
165 goal_team = 1
166 else:
167 goal_team = -1
168 if goal.own_goal:
169 goal_team *= -1
171 team_abbreviation = {
172 1: match.home_team_abbreviation,
173 -1: match.away_team_abbreviation
174 }[goal_team]
176 goal.player_team_abbreviation = team_abbreviation
177 home_score = goal.home_score
178 player = parse_player(goal_data, team_abbreviation)
180 db.session.merge(player)
181 db.session.merge(goal)
183 db.session.commit()
186def parse_match(match_data: Dict[str, Any], league: str, season: int) -> Match:
187 """
188 Parses a Match object from JSON match data
189 :param match_data: The match data to parse
190 :param league: The league
191 :param season: The season
192 :return: The generated Match object
193 """
194 ht_home = 0
195 ht_away = 0
196 ft_home = 0
197 ft_away = 0
199 for result in match_data["MatchResults"]:
200 if result["ResultName"] == "Halbzeit":
201 ht_home = result["PointsTeam1"]
202 ht_away = result["PointsTeam2"]
203 elif result["ResultName"] == "Endergebnis":
204 ft_home = result["PointsTeam1"]
205 ft_away = result["PointsTeam2"]
206 else: # pragma: no cover
207 pass
208 cur_home = max(ht_home, ft_home)
209 cur_away = max(ht_away, ft_away)
211 kickoff = match_data["MatchDateTimeUTC"]
212 kickoff = datetime.strptime(kickoff, "%Y-%m-%dT%H:%M:%SZ")
213 started = datetime.utcnow() > kickoff
214 kickoff = kickoff.strftime("%Y-%m-%d:%H-%M-%S")
216 home_team_abbreviation = get_team_data(match_data["Team1"]["TeamName"])[2]
217 away_team_abbreviation = get_team_data(match_data["Team2"]["TeamName"])[2]
219 match = Match(
220 home_team_abbreviation=home_team_abbreviation,
221 away_team_abbreviation=away_team_abbreviation,
222 season=season,
223 league=league,
224 matchday=match_data["Group"]["GroupOrderID"],
225 home_current_score=cur_home,
226 away_current_score=cur_away,
227 home_ht_score=ht_home,
228 away_ht_score=ht_away,
229 home_ft_score=ft_home,
230 away_ft_score=ft_away,
231 kickoff=kickoff,
232 started=started,
233 finished=match_data["MatchIsFinished"]
234 )
235 if match.has_started and match.minutes_since_kickoff > 1440: 235 ↛ 237line 235 didn't jump to line 237, because the condition on line 235 was never false
236 match.finished = True
237 return match
240def parse_goal(goal_data: Dict[str, Any], match: Match) -> Optional[Goal]:
241 """
242 Parses a goal JSON object and generates a Goal object
243 :param match: The match in which the goal was scored
244 :param goal_data: The goal data to parse
245 :return: The generated Goal object
246 """
247 if goal_data["GoalGetterID"] == 0:
248 return None
250 minute = goal_data["MatchMinute"]
252 # Minute defaults to 0 in case the minute data is missing.
253 # This keeps the entire thing from imploding.
254 if minute is None:
255 minute = 0
257 minute_et = 0
258 if minute > 90:
259 minute_et = minute - 90
260 minute = 90
262 goal = Goal(
263 home_team_abbreviation=match.home_team_abbreviation,
264 away_team_abbreviation=match.away_team_abbreviation,
265 season=match.season,
266 league=match.league,
267 matchday=match.matchday,
268 player_name=goal_data["GoalGetterName"],
269 player_team_abbreviation=None,
270 minute=minute,
271 minute_et=minute_et,
272 home_score=goal_data["ScoreTeam1"],
273 away_score=goal_data["ScoreTeam2"],
274 own_goal=goal_data["IsOwnGoal"],
275 penalty=goal_data["IsPenalty"]
276 )
278 if goal.home_score == 0 and goal.away_score == 0:
279 return None
280 else:
281 return goal
284def parse_player(goal_data: Dict[str, Any], team_abbreviation: str) -> Player:
285 """
286 Parses a Player object from a Goal JSON data object
287 :param goal_data: The data of a goal the player scored
288 :param team_abbreviation: The Team of the player
289 :return: The generated Player object
290 """
291 return Player(
292 team_abbreviation=team_abbreviation,
293 name=goal_data["GoalGetterName"]
294 )
297def parse_team(team_data: Dict[str, Any]) -> Team:
298 """
299 Parses team-related JSON data and generates a Team object from that
300 :param team_data: The team data to parse
301 :return: The generated Team object
302 """
303 name, short_name, abbrev, icons = get_team_data(team_data["TeamName"])
304 svg, png = icons
305 return Team(
306 name=name,
307 abbreviation=abbrev,
308 short_name=short_name,
309 icon_svg=svg,
310 icon_png=png
311 )