josh
3169d1a1ff
- utc_to_eastern: use zoneinfo instead of hardcoded EDT offset (-4) so start times are correct in both EST and EDT - standings: fetch before truncate so a failed API call doesn't wipe existing standings data - routes: call parse_games() once per request instead of three times - scheduler: wrap run_pending() in try/except so an unhandled exception doesn't kill the background thread Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
75 lines
1.9 KiB
Python
75 lines
1.9 KiB
Python
import logging
|
|
import sqlite3
|
|
|
|
import requests
|
|
|
|
from app.config import DB_PATH
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def create_standings_table(conn):
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS standings (
|
|
team_common_name TEXT,
|
|
league_sequence INTEGER,
|
|
league_l10_sequence INTEGER
|
|
)
|
|
""")
|
|
conn.commit()
|
|
|
|
|
|
def truncate_standings_table(conn):
|
|
cursor = conn.cursor()
|
|
cursor.execute("DELETE FROM standings")
|
|
conn.commit()
|
|
|
|
|
|
def insert_standings(conn, standings):
|
|
cursor = conn.cursor()
|
|
for team in standings:
|
|
cursor.execute(
|
|
"""
|
|
INSERT INTO standings (team_common_name, league_sequence, league_l10_sequence)
|
|
VALUES (?, ?, ?)
|
|
""",
|
|
(
|
|
team["team_common_name"],
|
|
team["league_sequence"],
|
|
team["league_l10_sequence"],
|
|
),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def fetch_standings():
|
|
url = "https://api-web.nhle.com/v1/standings/now"
|
|
try:
|
|
response = requests.get(url, timeout=10)
|
|
response.raise_for_status()
|
|
standings_data = response.json()
|
|
standings = []
|
|
for team in standings_data.get("standings", []):
|
|
standings.append(
|
|
{
|
|
"team_common_name": team["teamCommonName"]["default"],
|
|
"league_sequence": team["leagueSequence"],
|
|
"league_l10_sequence": team["leagueL10Sequence"],
|
|
}
|
|
)
|
|
return standings
|
|
except requests.RequestException as e:
|
|
logger.error("Failed to fetch standings: %s", e)
|
|
return None
|
|
|
|
|
|
def refresh_standings():
|
|
conn = sqlite3.connect(DB_PATH)
|
|
create_standings_table(conn)
|
|
standings = fetch_standings()
|
|
if standings:
|
|
truncate_standings_table(conn)
|
|
insert_standings(conn, standings)
|
|
conn.close()
|