#!/usr/bin/env python3 """Local web server for FBK playoff tipping app. Features: - Static web app hosting - Live SHL playoff data proxy endpoint - Simple login (Fredrik/Johan/Pappa) - Shared tips storage in SQLite """ from __future__ import annotations import hashlib import hmac import json import os import secrets import ssl import sqlite3 import threading import time import traceback from datetime import datetime, timedelta, timezone from http import HTTPStatus from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any from urllib.parse import parse_qs, urlencode, urlparse from urllib.request import Request, urlopen try: from zoneinfo import ZoneInfo except ModuleNotFoundError: try: from backports.zoneinfo import ZoneInfo # type: ignore except ModuleNotFoundError: ZoneInfo = None # type: ignore SERIES_UUID = "qQ9-bb0bzEWUk" # SHL API_BASE = "https://www.shl.se/api" SITE_INSTANCE_ID = "shl1_shl" PLAYOFF_GAME_TYPE_CODE = "playoff" TRACKED_TEAM_CODES = ("FBK", "RBK") PRIMARY_TEAM_CODE = "FBK" SECONDARY_TEAM_CODE = "RBK" SERIES_LABEL = "Kvartsfinal: Färjestad vs Rögle" WINS_TO_ADVANCE = 4 def _resolve_local_tz() -> timezone | Any: if ZoneInfo is not None: try: return ZoneInfo("Europe/Stockholm") except Exception: pass # Fallback object; local conversion is handled explicitly below. return timezone.utc LOCAL_TZ = _resolve_local_tz() ROOT_DIR = Path(__file__).resolve().parent DB_PATH = ROOT_DIR / "shl_tips.db" ALLOWED_PERSON_IDS = ("fredrik", "johan", "pappa") ADMIN_PERSON_IDS = ("fredrik",) DEFAULT_CACHE_TTL_SECONDS = 60 LIVE_CACHE_TTL_SECONDS = 15 LIVE_REFRESH_WINDOW_BEFORE_START = timedelta(minutes=15) LIVE_REFRESH_WINDOW_AFTER_START = timedelta(hours=6) TIP_LOCK_GRACE_PERIOD = timedelta(minutes=5) SESSION_COOKIE_NAME = "fbk_kvartsfinal_session" SESSION_TTL_DAYS = 30 PBKDF2_ITERATIONS = 240_000 _cache_lock = threading.Lock() _cache_payload: dict[str, Any] | None = None _cache_timestamp = 0.0 _SSL_CTX_VERIFIED = ssl.create_default_context() _SSL_CTX_UNVERIFIED = ssl._create_unverified_context() def _now_utc() -> datetime: return datetime.now(timezone.utc) def _now_utc_iso() -> str: return _now_utc().isoformat() def _api_get_json( path: str, params: dict[str, Any] | None = None, extra_headers: dict[str, str] | None = None, allow_empty: bool = False, ) -> Any: url = f"{API_BASE}{path}" if params: url = f"{url}?{urlencode(params, doseq=True)}" headers = { "Accept": "application/json", "User-Agent": "SHL-Tips-App/1.0", } if extra_headers: headers.update(extra_headers) request = Request(url, headers=headers) def read_json(response: Any) -> Any: body = response.read().decode("utf-8") if allow_empty and body.strip() == "": return None return json.loads(body) try: with urlopen(request, timeout=25, context=_SSL_CTX_VERIFIED) as response: if response.status != HTTPStatus.OK: raise RuntimeError(f"API request failed: {response.status} ({url})") return read_json(response) except Exception as exc: # Synology/python builds can have outdated CA bundles. # Retry once without certificate verification to keep the app usable. if "CERTIFICATE_VERIFY_FAILED" not in str(exc): raise with urlopen(request, timeout=25, context=_SSL_CTX_UNVERIFIED) as response: if response.status != HTTPStatus.OK: raise RuntimeError(f"API request failed: {response.status} ({url})") return read_json(response) def _api_get(path: str, params: dict[str, Any] | None = None) -> Any: return _api_get_json(path, params=params) def _api_get_optional_json( path: str, params: dict[str, Any] | None = None, extra_headers: dict[str, str] | None = None, ) -> Any | None: return _api_get_json(path, params=params, extra_headers=extra_headers, allow_empty=True) def _parse_iso_utc(raw_value: str) -> datetime: return datetime.fromisoformat(raw_value.replace("Z", "+00:00")).astimezone(timezone.utc) def _parse_schedule_local(raw_value: str | None) -> datetime | None: if not raw_value: return None try: return datetime.strptime(raw_value.strip(), "%Y-%m-%d %H:%M:%S") except ValueError: return None def _last_sunday_of_month_utc(year: int, month: int) -> datetime: if month == 12: next_month = datetime(year + 1, 1, 1, tzinfo=timezone.utc) else: next_month = datetime(year, month + 1, 1, tzinfo=timezone.utc) candidate = next_month - timedelta(days=1) while candidate.weekday() != 6: candidate -= timedelta(days=1) return candidate.replace(hour=1, minute=0, second=0, microsecond=0) def _stockholm_offset_hours_for_utc(dt_utc: datetime) -> int: year = dt_utc.year dst_start = _last_sunday_of_month_utc(year, 3) dst_end = _last_sunday_of_month_utc(year, 10) return 2 if dst_start <= dt_utc < dst_end else 1 def _utc_to_stockholm_local(dt_utc: datetime) -> datetime: offset = timedelta(hours=_stockholm_offset_hours_for_utc(dt_utc)) return dt_utc.astimezone(timezone(offset)) def _pick_current_season_uuid(filter_payload: dict[str, Any]) -> tuple[str, str]: seasons = filter_payload.get("season") or [] if not seasons: raise RuntimeError("No seasons returned from season-series-game-types-filter.") def season_key(item: dict[str, Any]) -> int: code = str(item.get("code", "0")) try: return int(code) except ValueError: return 0 current = max(seasons, key=season_key) season_uuid = current.get("uuid") season_name = "" names = current.get("names") or [] if names: season_name = names[0].get("translation", "") if not season_uuid: raise RuntimeError("Could not resolve current season UUID.") return season_uuid, season_name def _pick_regular_game_type_uuid(filter_payload: dict[str, Any]) -> str: game_types = filter_payload.get("gameType") or [] for game_type in game_types: if game_type.get("code") == "regular": uuid = game_type.get("uuid") if uuid: return uuid raise RuntimeError("Could not resolve regular season gameType UUID.") def _pick_game_type_uuid(filter_payload: dict[str, Any], game_type_code: str) -> str: game_types = filter_payload.get("gameType") or [] for game_type in game_types: if game_type.get("code") == game_type_code: uuid = game_type.get("uuid") if uuid: return uuid raise RuntimeError(f"Could not resolve gameType UUID for {game_type_code}.") def _resolve_ssgt_uuid(filter_payload: dict[str, Any]) -> str: ssgt_uuid = filter_payload.get("ssgtUuid") if isinstance(ssgt_uuid, str) and ssgt_uuid: return ssgt_uuid if isinstance(ssgt_uuid, list) and ssgt_uuid: first = ssgt_uuid[0] if isinstance(first, str): return first raise RuntimeError("Could not resolve ssgtUuid from filter payload.") def _team_name_from_team_names(team_names: dict[str, Any], fallback: str) -> str: def trim_herr(name: str) -> str: clean = name.strip() suffix = " Herr" if clean.endswith(suffix): clean = clean[: -len(suffix)].strip() return clean for key in ("short_site", "shortSite", "long_site", "longSite", "short", "long", "code"): value = team_names.get(key) if isinstance(value, str) and value.strip(): return trim_herr(value) return fallback def _team_name_from_schedule(team_info: dict[str, Any], fallback: str) -> str: def trim_herr(name: str) -> str: clean = name.strip() suffix = " Herr" if clean.endswith(suffix): clean = clean[: -len(suffix)].strip() return clean names = team_info.get("names") or {} for key in ("shortSite", "longSite", "short", "long", "code"): value = names.get(key) if isinstance(value, str) and value.strip(): return trim_herr(value) return fallback def _team_code_from_schedule(team_info: dict[str, Any]) -> str: names = team_info.get("names") or {} for key in ("codeSite", "code_site", "code"): value = names.get(key) if isinstance(value, str) and value.strip(): return value.strip().upper() fallback = team_info.get("code") if isinstance(fallback, str) and fallback.strip(): return fallback.strip().upper() return "" def _to_int_or_none(value: Any) -> int | None: if value is None: return None if isinstance(value, int): return value if isinstance(value, str): value = value.strip() if value == "" or value.upper() == "N/A": return None try: return int(value) except ValueError: return None return None def _is_live_state(state: str) -> bool: normalized = (state or "").lower() return normalized not in {"pre-game", "post-game"} def _is_tip_locked(game: dict[str, Any], now_utc: datetime | None = None) -> bool: if bool(game.get("isFinished")): return True raw_start = game.get("startTimeUtc") or game.get("rawStartDateTime") if not raw_start: return bool(game.get("isLive")) try: start_utc = _parse_iso_utc(str(raw_start)) except ValueError: return bool(game.get("isLive")) return start_utc + TIP_LOCK_GRACE_PERIOD <= (now_utc or _now_utc()) def _is_game_in_live_refresh_window(game: dict[str, Any], now_utc: datetime | None = None) -> bool: raw_start = game.get("startTimeUtc") or game.get("rawStartDateTime") if not raw_start: return bool(game.get("isLive")) try: start_utc = _parse_iso_utc(str(raw_start)) except ValueError: return bool(game.get("isLive")) current = now_utc or _now_utc() return ( start_utc - LIVE_REFRESH_WINDOW_BEFORE_START <= current <= start_utc + LIVE_REFRESH_WINDOW_AFTER_START ) def _payload_cache_ttl_seconds(payload: dict[str, Any] | None) -> int: if not payload: return DEFAULT_CACHE_TTL_SECONDS tracked_games = payload.get("trackedGames") or [] now_utc = _now_utc() if any(_is_game_in_live_refresh_window(game, now_utc) for game in tracked_games): return LIVE_CACHE_TTL_SECONDS return DEFAULT_CACHE_TTL_SECONDS def _live_api_headers() -> dict[str, str]: return {"x-s8y-instance-id": SITE_INSTANCE_ID} def _schedule_state_from_live_overview(state: str) -> str | None: normalized = (state or "").strip() if normalized in {"Ongoing", "PeriodBreak", "Overtime", "Shootout"}: return "live-game" if normalized == "GameEnded": return "post-game" return None def _apply_live_overview_to_game(game: dict[str, Any], overview: dict[str, Any]) -> None: overview_state = str(overview.get("state") or "").strip() schedule_state = _schedule_state_from_live_overview(overview_state) if schedule_state: game["state"] = schedule_state game["isFinished"] = schedule_state == "post-game" game["isLive"] = schedule_state == "live-game" home_goals = _to_int_or_none(overview.get("homeGoals")) away_goals = _to_int_or_none(overview.get("awayGoals")) if home_goals is not None: game["homeScore"] = home_goals if away_goals is not None: game["awayScore"] = away_goals time_info = overview.get("time") or {} period = _to_int_or_none(time_info.get("period")) if overview_state in {"Overtime", "Shootout"} or (period is not None and period >= 4): game["wentToOvertime"] = True if overview_state == "Shootout" or (period is not None and period >= 5): game["wentToShootout"] = True def _apply_live_game_overviews(tracked_games: list[dict[str, Any]]) -> None: candidates = [game for game in tracked_games if _is_game_in_live_refresh_window(game)] for game in candidates: game_uuid = game.get("gameUuid") if not game_uuid: continue try: overview = _api_get_optional_json( f"/gameday/game-overview/{game_uuid}", extra_headers=_live_api_headers(), ) except Exception: continue if not isinstance(overview, dict) or not overview: continue _apply_live_overview_to_game(game, overview) def _normalize_person_id(value: str | None) -> str | None: if not value: return None person_id = value.strip().lower() if person_id in ALLOWED_PERSON_IDS: return person_id return None def _password_hash(password: str, salt_hex: str) -> str: digest = hashlib.pbkdf2_hmac( "sha256", password.encode("utf-8"), bytes.fromhex(salt_hex), PBKDF2_ITERATIONS, ) return digest.hex() def _winner_team_code(game: dict[str, Any]) -> str | None: home_score = _to_int_or_none(game.get("homeScore")) away_score = _to_int_or_none(game.get("awayScore")) if home_score is None or away_score is None or home_score == away_score: return None return game.get("homeTeamCode") if home_score > away_score else game.get("awayTeamCode") def _build_live_payload() -> dict[str, Any]: filter_payload = _api_get( "/sports-v2/season-series-game-types-filter", {"series": SERIES_UUID}, ) season_uuid, season_name = _pick_current_season_uuid(filter_payload) playoff_game_type_uuid = _pick_game_type_uuid(filter_payload, PLAYOFF_GAME_TYPE_CODE) schedule_payload = _api_get( "/sports-v2/game-schedule", { "seasonUuid": season_uuid, "seriesUuid": SERIES_UUID, "gameTypeUuid": playoff_game_type_uuid, "completeSeason": "all", "homeAway": "all", "allGames": "all", }, ) target_codes = set(TRACKED_TEAM_CODES) team_map: dict[str, dict[str, Any]] = {} tracked_games: list[dict[str, Any]] = [] for game in schedule_payload.get("gameInfo") or []: raw_start = game.get("rawStartDateTime") if not raw_start: continue home_info = game.get("homeTeamInfo") or {} away_info = game.get("awayTeamInfo") or {} home_code = _team_code_from_schedule(home_info) away_code = _team_code_from_schedule(away_info) if {home_code, away_code} != target_codes: continue start_utc = _parse_iso_utc(raw_start) start_local = _parse_schedule_local(game.get("startDateTime")) if start_local is None: start_local = _utc_to_stockholm_local(start_utc) home_name = _team_name_from_schedule(home_info, fallback=home_code or "Hemmalag") away_name = _team_name_from_schedule(away_info, fallback=away_code or "Bortalag") if home_code not in team_map: team_map[home_code] = { "teamCode": home_code, "teamName": home_name, "logo": home_info.get("icon"), } if away_code not in team_map: team_map[away_code] = { "teamCode": away_code, "teamName": away_name, "logo": away_info.get("icon"), } state_raw = str(game.get("state") or "pre-game") tracked_games.append( { "gameUuid": game.get("uuid"), "state": state_raw, "isFinished": state_raw == "post-game", "isLive": _is_live_state(state_raw), "startTimeUtc": start_utc.isoformat().replace("+00:00", "Z"), "startTimeLocal": start_local.isoformat(), "dateLocal": start_local.date().isoformat(), "timeLocal": start_local.strftime("%H:%M"), "homeTeamCode": home_code, "awayTeamCode": away_code, "homeTeamName": home_name, "awayTeamName": away_name, "homeLogo": home_info.get("icon"), "awayLogo": away_info.get("icon"), "homeScore": _to_int_or_none(home_info.get("score")), "awayScore": _to_int_or_none(away_info.get("score")), "wentToOvertime": bool(game.get("overtime")), "wentToShootout": bool(game.get("shootout")), "venue": (game.get("venueInfo") or {}).get("name"), } ) if not tracked_games: raise RuntimeError("Could not find any playoff games for Färjestad vs Rögle.") tracked_games.sort(key=lambda game: game["startTimeUtc"]) for index, game in enumerate(tracked_games, start=1): game["gameNumber"] = index game["round"] = index _apply_live_game_overviews(tracked_games) team1 = team_map.get(PRIMARY_TEAM_CODE) or { "teamCode": PRIMARY_TEAM_CODE, "teamName": "Färjestad", "logo": None, } team2 = team_map.get(SECONDARY_TEAM_CODE) or { "teamCode": SECONDARY_TEAM_CODE, "teamName": "Rögle", "logo": None, } actual_team1_wins = 0 actual_team2_wins = 0 finished_games = 0 for game in tracked_games: if game.get("isFinished"): finished_games += 1 winner_code = _winner_team_code(game) if winner_code == PRIMARY_TEAM_CODE: actual_team1_wins += 1 elif winner_code == SECONDARY_TEAM_CODE: actual_team2_wins += 1 remaining_games = [game for game in tracked_games if not game.get("isFinished")] next_game = remaining_games[0] if remaining_games else None winner_code = None if actual_team1_wins >= WINS_TO_ADVANCE: winner_code = PRIMARY_TEAM_CODE elif actual_team2_wins >= WINS_TO_ADVANCE: winner_code = SECONDARY_TEAM_CODE return { "fetchedAtLocal": _utc_to_stockholm_local(_now_utc()).isoformat(), "meta": { "seriesUuid": SERIES_UUID, "seasonUuid": season_uuid, "seasonName": season_name, "playoffGameTypeUuid": playoff_game_type_uuid, "seriesLabel": SERIES_LABEL, "scheduledGames": len(tracked_games), "trackedGames": len(tracked_games), "remainingGames": len(remaining_games), "winsToAdvance": WINS_TO_ADVANCE, }, "series": { "bestOf": len(tracked_games), "winsToAdvance": WINS_TO_ADVANCE, "team1": team1, "team2": team2, "actualTeam1Wins": actual_team1_wins, "actualTeam2Wins": actual_team2_wins, "finishedGames": finished_games, "remainingGames": len(remaining_games), "seriesFinished": bool(winner_code) or finished_games == len(tracked_games), "winnerCode": winner_code, "nextGameNumber": next_game.get("gameNumber") if next_game else None, "nextGameStartTimeLocal": next_game.get("startTimeLocal") if next_game else None, "nextGameDateLocal": next_game.get("dateLocal") if next_game else None, "nextGameTimeLocal": next_game.get("timeLocal") if next_game else None, }, "trackedGames": tracked_games, "sourceUrls": [ f"{API_BASE}/sports-v2/season-series-game-types-filter?series={SERIES_UUID}", ( f"{API_BASE}/sports-v2/game-schedule?seasonUuid={season_uuid}" f"&seriesUuid={SERIES_UUID}&gameTypeUuid={playoff_game_type_uuid}" "&completeSeason=all&homeAway=all&allGames=all" ), ], } def _get_payload_cached() -> dict[str, Any]: global _cache_payload, _cache_timestamp now = time.time() with _cache_lock: cache_ttl = _payload_cache_ttl_seconds(_cache_payload) if _cache_payload and now - _cache_timestamp < cache_ttl: return _cache_payload try: fresh_payload = _build_live_payload() except Exception: with _cache_lock: if _cache_payload: stale = dict(_cache_payload) stale["stale"] = True return stale raise with _cache_lock: _cache_payload = fresh_payload _cache_timestamp = time.time() return fresh_payload def _db_connect() -> sqlite3.Connection: connection = sqlite3.connect(DB_PATH) connection.row_factory = sqlite3.Row return connection def _init_db() -> None: with _db_connect() as connection: connection.execute( """ CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, person_id TEXT NOT NULL UNIQUE, password_salt TEXT NOT NULL, password_hash TEXT NOT NULL, created_at TEXT NOT NULL ) """ ) connection.execute( """ CREATE TABLE IF NOT EXISTS sessions ( token TEXT PRIMARY KEY, user_id INTEGER NOT NULL, created_at TEXT NOT NULL, expires_at TEXT NOT NULL, FOREIGN KEY (user_id) REFERENCES users(id) ) """ ) connection.execute( """ CREATE TABLE IF NOT EXISTS tips ( season_uuid TEXT NOT NULL, game_uuid TEXT NOT NULL, person_id TEXT NOT NULL, home_goals INTEGER, away_goals INTEGER, ot INTEGER NOT NULL DEFAULT 0, updated_at TEXT NOT NULL, updated_by_person TEXT NOT NULL, PRIMARY KEY (season_uuid, game_uuid, person_id) ) """ ) def _cleanup_sessions(connection: sqlite3.Connection) -> None: connection.execute( "DELETE FROM sessions WHERE expires_at < ?", (_now_utc_iso(),), ) def _tips_for_season(connection: sqlite3.Connection, season_uuid: str) -> dict[str, dict[str, dict[str, Any]]]: result: dict[str, dict[str, dict[str, Any]]] = {person_id: {} for person_id in ALLOWED_PERSON_IDS} rows = connection.execute( """ SELECT game_uuid, person_id, home_goals, away_goals, ot FROM tips WHERE season_uuid = ? """, (season_uuid,), ).fetchall() for row in rows: person_id = row["person_id"] if person_id not in result: continue result[person_id][row["game_uuid"]] = { "hg": row["home_goals"], "ag": row["away_goals"], "ot": bool(row["ot"]), } return result def _parse_cookie_header(header_value: str | None) -> dict[str, str]: cookies: dict[str, str] = {} if not header_value: return cookies for item in header_value.split(";"): if "=" not in item: continue key, value = item.split("=", 1) cookies[key.strip()] = value.strip() return cookies def _session_cookie(token: str) -> str: secure_attr = "; Secure" if os.environ.get("COOKIE_SECURE", "").strip() == "1" else "" max_age = int(timedelta(days=SESSION_TTL_DAYS).total_seconds()) return ( f"{SESSION_COOKIE_NAME}={token}; Path=/; HttpOnly; SameSite=Lax; " f"Max-Age={max_age}{secure_attr}" ) def _expired_session_cookie() -> str: return ( f"{SESSION_COOKIE_NAME}=; Path=/; HttpOnly; SameSite=Lax; " "Max-Age=0" ) class AppHandler(SimpleHTTPRequestHandler): def __init__(self, *args: Any, **kwargs: Any) -> None: super().__init__(*args, directory=str(ROOT_DIR), **kwargs) def do_GET(self) -> None: parsed = urlparse(self.path) if parsed.path in {"/admin", "/admin/"}: self.path = "/admin.html" super().do_GET() return if parsed.path == "/api/shl/current": self._handle_shl_current() return if parsed.path == "/api/auth/me": self._handle_auth_me() return if parsed.path == "/api/tips": self._handle_tips_get(parsed.query) return if parsed.path == "/healthz": self._write_json({"ok": True}) return super().do_GET() def do_POST(self) -> None: parsed = urlparse(self.path) if parsed.path == "/api/auth/login": self._handle_auth_login() return if parsed.path == "/api/auth/logout": self._handle_auth_logout() return if parsed.path == "/api/admin/reset-password": self._handle_admin_reset_password() return self._write_json({"error": "Not Found"}, status=HTTPStatus.NOT_FOUND) def do_PUT(self) -> None: parsed = urlparse(self.path) if parsed.path == "/api/tips": self._handle_tips_put(parsed.query) return self._write_json({"error": "Not Found"}, status=HTTPStatus.NOT_FOUND) def do_DELETE(self) -> None: parsed = urlparse(self.path) if parsed.path == "/api/tips": self._handle_tips_delete(parsed.query) return self._write_json({"error": "Not Found"}, status=HTTPStatus.NOT_FOUND) def _read_json_body(self) -> dict[str, Any]: content_length = int(self.headers.get("Content-Length", "0")) if content_length <= 0: return {} body = self.rfile.read(content_length) if not body: return {} try: return json.loads(body.decode("utf-8")) except json.JSONDecodeError: return {} def _current_user(self) -> dict[str, Any] | None: cookies = _parse_cookie_header(self.headers.get("Cookie")) token = cookies.get(SESSION_COOKIE_NAME) if not token: return None with _db_connect() as connection: _cleanup_sessions(connection) row = connection.execute( """ SELECT u.id, u.person_id, s.token FROM sessions s JOIN users u ON u.id = s.user_id WHERE s.token = ? AND s.expires_at >= ? """, (token, _now_utc_iso()), ).fetchone() if not row: return None return {"id": row["id"], "personId": row["person_id"], "sessionToken": row["token"]} def _require_auth(self) -> dict[str, Any] | None: user = self._current_user() if not user: self._write_json( {"error": "Du behöver logga in."}, status=HTTPStatus.UNAUTHORIZED, ) return None return user def _require_admin(self) -> dict[str, Any] | None: user = self._require_auth() if not user: return None if user["personId"] not in ADMIN_PERSON_IDS: self._write_json( {"error": "Du saknar behörighet för admin-funktionen."}, status=HTTPStatus.FORBIDDEN, ) return None return user def _season_uuid_from_query_or_current(self, query: str) -> str: params = parse_qs(query or "") season_uuid = (params.get("seasonUuid") or [None])[0] if season_uuid: return season_uuid payload = _get_payload_cached() return payload["meta"]["seasonUuid"] def _handle_shl_current(self) -> None: try: payload = _get_payload_cached() self._write_json(payload) except Exception as exc: traceback.print_exc() self._write_json( {"error": "Kunde inte hämta SHL-data just nu.", "details": str(exc)}, status=HTTPStatus.BAD_GATEWAY, ) def _handle_auth_me(self) -> None: user = self._current_user() if not user: self._write_json({"authenticated": False, "personId": None}) return self._write_json({"authenticated": True, "personId": user["personId"]}) def _handle_auth_login(self) -> None: payload = self._read_json_body() person_id = _normalize_person_id(payload.get("personId")) password = str(payload.get("password") or "") if not person_id: self._write_json({"error": "Ogiltig användare."}, status=HTTPStatus.BAD_REQUEST) return if len(password) < 4: self._write_json( {"error": "Lösenord måste vara minst 4 tecken."}, status=HTTPStatus.BAD_REQUEST, ) return with _db_connect() as connection: _cleanup_sessions(connection) row = connection.execute( "SELECT id, password_salt, password_hash FROM users WHERE person_id = ?", (person_id,), ).fetchone() if row: expected = row["password_hash"] actual = _password_hash(password, row["password_salt"]) if not hmac.compare_digest(expected, actual): self._write_json( {"error": "Fel lösenord."}, status=HTTPStatus.UNAUTHORIZED, ) return user_id = row["id"] else: salt_hex = secrets.token_hex(16) pwd_hash = _password_hash(password, salt_hex) cursor = connection.execute( """ INSERT INTO users (person_id, password_salt, password_hash, created_at) VALUES (?, ?, ?, ?) """, (person_id, salt_hex, pwd_hash, _now_utc_iso()), ) user_id = cursor.lastrowid session_token = secrets.token_urlsafe(32) expires_at = (_now_utc() + timedelta(days=SESSION_TTL_DAYS)).isoformat() connection.execute( """ INSERT INTO sessions (token, user_id, created_at, expires_at) VALUES (?, ?, ?, ?) """, (session_token, user_id, _now_utc_iso(), expires_at), ) connection.commit() self._write_json( {"ok": True, "personId": person_id}, set_cookies=[_session_cookie(session_token)], ) def _handle_auth_logout(self) -> None: cookies = _parse_cookie_header(self.headers.get("Cookie")) token = cookies.get(SESSION_COOKIE_NAME) if token: with _db_connect() as connection: connection.execute("DELETE FROM sessions WHERE token = ?", (token,)) connection.commit() self._write_json( {"ok": True}, set_cookies=[_expired_session_cookie()], ) def _handle_admin_reset_password(self) -> None: if not self._require_admin(): return payload = self._read_json_body() person_id = _normalize_person_id(payload.get("personId")) if person_id not in {"johan", "pappa"}: self._write_json( {"error": "Du kan bara återställa Johan eller Pappa här."}, status=HTTPStatus.BAD_REQUEST, ) return new_password = str(payload.get("newPassword") or "").strip() with _db_connect() as connection: row = connection.execute( "SELECT id FROM users WHERE person_id = ?", (person_id,), ).fetchone() if new_password: if len(new_password) < 4: self._write_json( {"error": "Nytt lösenord måste vara minst 4 tecken."}, status=HTTPStatus.BAD_REQUEST, ) return salt_hex = secrets.token_hex(16) pwd_hash = _password_hash(new_password, salt_hex) if row: user_id = row["id"] connection.execute( """ UPDATE users SET password_salt = ?, password_hash = ? WHERE id = ? """, (salt_hex, pwd_hash, user_id), ) else: cursor = connection.execute( """ INSERT INTO users (person_id, password_salt, password_hash, created_at) VALUES (?, ?, ?, ?) """, (person_id, salt_hex, pwd_hash, _now_utc_iso()), ) user_id = cursor.lastrowid mode = "updated" else: if row: user_id = row["id"] connection.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,)) connection.execute("DELETE FROM users WHERE id = ?", (user_id,)) mode = "cleared" if mode == "updated": connection.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,)) connection.commit() self._write_json({"ok": True, "personId": person_id, "mode": mode}) def _handle_tips_get(self, query: str) -> None: # Tips are publicly readable so everyone can compare predictions even when logged out. user = self._current_user() season_uuid = self._season_uuid_from_query_or_current(query) with _db_connect() as connection: tips = _tips_for_season(connection, season_uuid) self._write_json( { "seasonUuid": season_uuid, "tips": tips, "editablePersonId": user["personId"] if user else None, } ) def _handle_tips_put(self, query: str) -> None: user = self._require_auth() if not user: return body = self._read_json_body() game_uuid = str(body.get("gameUuid") or "").strip() if not game_uuid: self._write_json({"error": "gameUuid saknas."}, status=HTTPStatus.BAD_REQUEST) return payload = _get_payload_cached() tracked_games = { game["gameUuid"]: game for game in payload.get("trackedGames") or [] if game.get("gameUuid") } game = tracked_games.get(game_uuid) if not game: self._write_json( {"error": "Matchen finns inte i den spårade omgången."}, status=HTTPStatus.BAD_REQUEST, ) return if _is_tip_locked(game): self._write_json( {"error": "Matchen har redan börjat och tipset är låst."}, status=HTTPStatus.CONFLICT, ) return def parse_goal(value: Any) -> int | None: if value in ("", None): return None try: parsed = int(value) except (TypeError, ValueError): return None if parsed < 0 or parsed > 30: return None return parsed hg = parse_goal(body.get("hg")) ag = parse_goal(body.get("ag")) ot = bool(body.get("ot")) season_uuid = self._season_uuid_from_query_or_current(query) person_id = user["personId"] with _db_connect() as connection: if hg is None and ag is None and not ot: connection.execute( """ DELETE FROM tips WHERE season_uuid = ? AND game_uuid = ? AND person_id = ? """, (season_uuid, game_uuid, person_id), ) else: connection.execute( """ INSERT INTO tips (season_uuid, game_uuid, person_id, home_goals, away_goals, ot, updated_at, updated_by_person) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(season_uuid, game_uuid, person_id) DO UPDATE SET home_goals = excluded.home_goals, away_goals = excluded.away_goals, ot = excluded.ot, updated_at = excluded.updated_at, updated_by_person = excluded.updated_by_person """, ( season_uuid, game_uuid, person_id, hg, ag, 1 if ot else 0, _now_utc_iso(), person_id, ), ) connection.commit() self._write_json({"ok": True}) def _handle_tips_delete(self, query: str) -> None: user = self._require_auth() if not user: return season_uuid = self._season_uuid_from_query_or_current(query) payload = _get_payload_cached() removable_game_ids = [ game["gameUuid"] for game in payload.get("trackedGames") or [] if game.get("gameUuid") and not _is_tip_locked(game) ] deleted_count = 0 with _db_connect() as connection: if removable_game_ids: placeholders = ", ".join("?" for _ in removable_game_ids) cursor = connection.execute( f""" DELETE FROM tips WHERE season_uuid = ? AND person_id = ? AND game_uuid IN ({placeholders}) """, (season_uuid, user["personId"], *removable_game_ids), ) deleted_count = cursor.rowcount connection.commit() self._write_json({"ok": True, "deletedCount": deleted_count}) def _write_json( self, payload: dict[str, Any], status: HTTPStatus = HTTPStatus.OK, set_cookies: list[str] | None = None, ) -> None: body = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") if set_cookies: for cookie in set_cookies: self.send_header("Set-Cookie", cookie) self.end_headers() self.wfile.write(body) def run() -> None: _init_db() host = os.environ.get("HOST", "0.0.0.0") port = int(os.environ.get("PORT", "8080")) server = ThreadingHTTPServer((host, port), AppHandler) print(f"SHL Tips server running on http://{host}:{port}") server.serve_forever() if __name__ == "__main__": run()