240 lines
6.3 KiB
Python
240 lines
6.3 KiB
Python
from db import get_connection
|
|
|
|
|
|
def upsert_team(
|
|
external_id: str,
|
|
name: str,
|
|
short_name: str | None = None,
|
|
logo_url: str | None = None,
|
|
games: int | None = 0,
|
|
wins: int | None = 0,
|
|
goals: int | None = 0,
|
|
tournaments: int | None = 0,
|
|
) -> None:
|
|
query = """
|
|
INSERT INTO teams (
|
|
external_id,
|
|
name,
|
|
short_name,
|
|
logo_url,
|
|
games,
|
|
wins,
|
|
goals,
|
|
tournaments,
|
|
created_at,
|
|
updated_at
|
|
)
|
|
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, NOW(), NOW())
|
|
ON CONFLICT (external_id)
|
|
DO UPDATE SET
|
|
name = EXCLUDED.name,
|
|
short_name = EXCLUDED.short_name,
|
|
logo_url = EXCLUDED.logo_url,
|
|
games = EXCLUDED.games,
|
|
wins = EXCLUDED.wins,
|
|
goals = EXCLUDED.goals,
|
|
tournaments = EXCLUDED.tournaments,
|
|
updated_at = NOW();
|
|
"""
|
|
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
query,
|
|
(
|
|
external_id,
|
|
name,
|
|
short_name,
|
|
logo_url,
|
|
games,
|
|
wins,
|
|
goals,
|
|
tournaments,
|
|
),
|
|
)
|
|
conn.commit()
|
|
except Exception:
|
|
conn.rollback()
|
|
raise
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def get_team_external_id_by_name(name: str) -> str | None:
|
|
query = """
|
|
SELECT external_id
|
|
FROM teams
|
|
WHERE LOWER(name) = LOWER(%s)
|
|
LIMIT 1;
|
|
"""
|
|
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(query, (name,))
|
|
row = cur.fetchone()
|
|
return row[0] if row else None
|
|
finally:
|
|
conn.close()
|
|
|
|
def get_team_id_by_external_id(external_id: str) -> int | None:
|
|
external_id = str(external_id).strip()
|
|
|
|
query = """
|
|
SELECT id
|
|
FROM teams
|
|
WHERE TRIM(external_id) = %s
|
|
LIMIT 1;
|
|
"""
|
|
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
print(f"[get_team_id_by_external_id] search external_id = '{external_id}'")
|
|
cur.execute(query, (external_id,))
|
|
row = cur.fetchone()
|
|
print(f"[get_team_id_by_external_id] result = {row}")
|
|
return row[0] if row else None
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def search_teams_for_admin(q: str = "") -> list[dict]:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
if q.strip():
|
|
pattern = f"%{q.strip()}%"
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
t.id,
|
|
t.name,
|
|
t.full_name,
|
|
t.short_name_3,
|
|
t.city,
|
|
t.logo_path,
|
|
t.external_id
|
|
FROM teams t
|
|
WHERE
|
|
t.name ILIKE %s
|
|
OR COALESCE(t.full_name, '') ILIKE %s
|
|
OR COALESCE(t.short_name_3, '') ILIKE %s
|
|
OR COALESCE(t.city, '') ILIKE %s
|
|
OR COALESCE(t.external_id, '') ILIKE %s
|
|
ORDER BY t.name ASC, t.id ASC
|
|
LIMIT 200
|
|
""",
|
|
(pattern, pattern, pattern, pattern, pattern),
|
|
)
|
|
else:
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
t.id,
|
|
t.name,
|
|
t.full_name,
|
|
t.short_name_3,
|
|
t.city,
|
|
t.logo_path,
|
|
t.external_id
|
|
FROM teams t
|
|
ORDER BY t.id DESC
|
|
LIMIT 200
|
|
"""
|
|
)
|
|
rows = cur.fetchall()
|
|
|
|
return [
|
|
{
|
|
"id": row[0],
|
|
"name": row[1] or "",
|
|
"full_name": row[2] or "",
|
|
"short_name_3": row[3] or "",
|
|
"city": row[4] or "",
|
|
"logo_path": row[5] or "",
|
|
"external_id": row[6] or "",
|
|
}
|
|
for row in rows
|
|
]
|
|
finally:
|
|
conn.close()
|
|
|
|
def get_team_by_id(team_id: int) -> dict | None:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
SELECT
|
|
t.id,
|
|
t.name,
|
|
t.full_name,
|
|
t.short_name_3,
|
|
t.city,
|
|
t.logo_path,
|
|
t.external_id
|
|
FROM teams t
|
|
WHERE t.id = %s
|
|
LIMIT 1
|
|
""",
|
|
(team_id,),
|
|
)
|
|
row = cur.fetchone()
|
|
|
|
if not row:
|
|
return None
|
|
|
|
return {
|
|
"id": row[0],
|
|
"name": row[1] or "",
|
|
"full_name": row[2] or "",
|
|
"short_name_3": row[3] or "",
|
|
"city": row[4] or "",
|
|
"logo_path": row[5] or "",
|
|
"external_id": row[6] or "",
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
def update_team_admin(
|
|
team_id: int,
|
|
name: str = "",
|
|
full_name: str = "",
|
|
short_name_3: str = "",
|
|
city: str = "",
|
|
logo_path: str = "",
|
|
external_id: str = "",
|
|
) -> None:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
UPDATE teams
|
|
SET
|
|
name = %s,
|
|
full_name = %s,
|
|
short_name_3 = %s,
|
|
city = NULLIF(%s, ''),
|
|
logo_path = %s,
|
|
external_id = NULLIF(%s, '')
|
|
WHERE id = %s
|
|
""",
|
|
(
|
|
name.strip(),
|
|
full_name.strip(),
|
|
short_name_3.strip().upper(),
|
|
city.strip(),
|
|
logo_path.strip(),
|
|
external_id.strip(),
|
|
team_id,
|
|
),
|
|
)
|
|
conn.commit()
|
|
except Exception:
|
|
conn.rollback()
|
|
raise
|
|
finally:
|
|
conn.close() |