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

139 statements  

1"""LICENSE 

2Copyright 2017 Hermann Krumrey <hermann@krumreyh.com> 

3 

4This file is part of bundesliga-tippspiel. 

5 

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. 

10 

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. 

15 

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""" 

19 

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 

32 

33 

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] = {} 

40 

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) 

51 

52 if last_update == 0: 

53 return True # Update on startup 

54 

55 now = datetime.utcnow() 

56 delta = time.time() - last_update 

57 

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] 

62 

63 if len(matches) == 0: 

64 app.logger.debug("Initial Filling of database") 

65 return True 

66 

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 

71 

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()) 

78 

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 

100 

101 

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 ) 

117 

118 

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}") 

131 

132 if league is None: 

133 league = Config.OPENLIGADB_LEAGUE 

134 if season is None: 

135 season = Config.OPENLIGADB_SEASON 

136 

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 

149 

150 for team_info in team_data: 

151 team = parse_team(team_info) 

152 db.session.merge(team) 

153 

154 for match_info in match_data: 

155 match = parse_match(match_info, league, int(season)) 

156 match = db.session.merge(match) 

157 

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 

163 

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 

170 

171 team_abbreviation = { 

172 1: match.home_team_abbreviation, 

173 -1: match.away_team_abbreviation 

174 }[goal_team] 

175 

176 goal.player_team_abbreviation = team_abbreviation 

177 home_score = goal.home_score 

178 player = parse_player(goal_data, team_abbreviation) 

179 

180 db.session.merge(player) 

181 db.session.merge(goal) 

182 

183 db.session.commit() 

184 

185 

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 

198 

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) 

210 

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") 

215 

216 home_team_abbreviation = get_team_data(match_data["Team1"]["TeamName"])[2] 

217 away_team_abbreviation = get_team_data(match_data["Team2"]["TeamName"])[2] 

218 

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 

238 

239 

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 

249 

250 minute = goal_data["MatchMinute"] 

251 

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 

256 

257 minute_et = 0 

258 if minute > 90: 

259 minute_et = minute - 90 

260 minute = 90 

261 

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 ) 

277 

278 if goal.home_score == 0 and goal.away_score == 0: 

279 return None 

280 else: 

281 return goal 

282 

283 

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 ) 

295 

296 

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 )