153 lines
3.9 KiB
Python
153 lines
3.9 KiB
Python
from db import get_connection
|
|
from repositories.team_repository import get_team_id_by_external_id
|
|
|
|
|
|
def replace_standings_for_season(season: str, standings_rows: list[dict]) -> None:
|
|
delete_query = """
|
|
DELETE FROM standings
|
|
WHERE season = %s;
|
|
"""
|
|
|
|
insert_query = """
|
|
INSERT INTO standings (
|
|
team_id,
|
|
season,
|
|
played,
|
|
wins,
|
|
losses,
|
|
draws,
|
|
points_for,
|
|
points_against,
|
|
points,
|
|
position,
|
|
created_at,
|
|
updated_at
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW());
|
|
"""
|
|
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(delete_query, (season,))
|
|
|
|
for row in standings_rows:
|
|
team_external_id = row["team_external_id"]
|
|
team_id = get_team_id_by_external_id(team_external_id)
|
|
|
|
if team_id is None:
|
|
raise ValueError(f"Team not found by external_id: {team_external_id}")
|
|
|
|
cur.execute(
|
|
insert_query,
|
|
(
|
|
team_id,
|
|
season,
|
|
row.get("played", 0),
|
|
row.get("wins", 0),
|
|
row.get("losses", 0),
|
|
row.get("draws", 0),
|
|
row.get("points_for", 0),
|
|
row.get("points_against", 0),
|
|
row.get("points", 0),
|
|
row.get("position"),
|
|
),
|
|
)
|
|
|
|
conn.commit()
|
|
except Exception:
|
|
conn.rollback()
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_standings_by_season(season: str):
|
|
query = """
|
|
SELECT
|
|
s.id,
|
|
s.team_id,
|
|
t.name AS team_name,
|
|
s.season,
|
|
s.played,
|
|
s.wins,
|
|
s.losses,
|
|
s.draws,
|
|
s.points_for,
|
|
s.points_against,
|
|
s.points,
|
|
s.position,
|
|
s.created_at,
|
|
s.updated_at
|
|
FROM standings s
|
|
JOIN teams t ON t.id = s.team_id
|
|
WHERE s.season = %s
|
|
ORDER BY s.position ASC NULLS LAST, s.id ASC;
|
|
"""
|
|
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(query, (season,))
|
|
return cur.fetchall()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_standings_by_match_id(match_id: int) -> list[dict]:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT m.season, m.tour
|
|
FROM matches m
|
|
WHERE m.id = %s
|
|
""",
|
|
(match_id,),
|
|
)
|
|
base = cur.fetchone()
|
|
if not base:
|
|
return []
|
|
|
|
season = base[0]
|
|
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
s.position,
|
|
t.logo_url,
|
|
t.name,
|
|
s.played,
|
|
s.wins,
|
|
s.draws,
|
|
s.losses,
|
|
s.points_for,
|
|
s.points_against,
|
|
s.points
|
|
FROM standings s
|
|
JOIN teams t ON t.id = s.team_id
|
|
WHERE s.season = %s
|
|
ORDER BY s.position ASC, s.points DESC, s.team_id ASC
|
|
""",
|
|
(season,),
|
|
)
|
|
rows = cur.fetchall()
|
|
|
|
return [
|
|
{
|
|
"position": r[0],
|
|
"team_logo": r[1] or "",
|
|
"team_name": r[2] or "",
|
|
"played": r[3] or 0,
|
|
"wins": r[4] or 0,
|
|
"draws": r[5] or 0,
|
|
"losses": r[6] or 0,
|
|
"goals_for": r[7] or 0,
|
|
"goals_against": r[8] or 0,
|
|
"points": r[9] or 0,
|
|
}
|
|
for r in rows
|
|
]
|
|
finally:
|
|
conn.close() |