405 lines
18 KiB
Python
405 lines
18 KiB
Python
"""Independent polling, request budget and persistent last-good snapshots.
|
|
|
|
No external coordinator or extra Python dependencies are required.
|
|
"""
|
|
from collections import deque
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime, timezone
|
|
from email.utils import parsedate_to_datetime
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qsl, urlencode, urlsplit, urlunsplit
|
|
from zoneinfo import ZoneInfo
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import math
|
|
import os
|
|
import random
|
|
import threading
|
|
import time
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PollingConfig:
|
|
max_requests_per_minute: int = 90
|
|
calendar_wait_seconds: float = 900
|
|
calendar_prepare_seconds: float = 120
|
|
prepare_minutes: float = 70
|
|
live_status_seconds: float = 1
|
|
box_score_seconds: float = 5
|
|
play_by_play_seconds: float = 5
|
|
prestart_status_seconds: float = 30
|
|
near_start_status_seconds: float = 10
|
|
metadata_seconds: float = 300
|
|
standings_seconds: float = 300
|
|
other_games_seconds: float = 30
|
|
other_games_demand_seconds: float = 90
|
|
finished_status_seconds: float = 30
|
|
finished_hold_seconds: float = 600
|
|
request_connect_timeout: float = 3
|
|
request_read_timeout: float = 5
|
|
retry_initial_seconds: float = 5
|
|
retry_max_seconds: float = 300
|
|
rate_limit_pause_seconds: float = 120
|
|
forbidden_pause_seconds: float = 600
|
|
startup_jitter_seconds: float = 30
|
|
snapshot_seconds: float = 30
|
|
timezone: str = "Europe/Moscow"
|
|
cache_directory: str = ".rfb_cache"
|
|
shotmap_max_images: int = 192
|
|
|
|
@classmethod
|
|
def load(cls, path, logger):
|
|
values = asdict(cls())
|
|
try:
|
|
overrides = json.loads(Path(path).read_text(encoding="utf-8-sig"))
|
|
except FileNotFoundError:
|
|
overrides = {}
|
|
if not isinstance(overrides, dict):
|
|
raise ValueError("polling_config.json must contain an object")
|
|
for name, value in overrides.items():
|
|
if name not in values:
|
|
raise ValueError(f"Unknown polling setting: {name}")
|
|
if isinstance(values[name], (int, float)):
|
|
if isinstance(value, bool) or not isinstance(value, (int, float)):
|
|
raise ValueError(f"{name} must be a number")
|
|
if not math.isfinite(value) or value < 0 or (value == 0 and name != "startup_jitter_seconds"):
|
|
raise ValueError(f"{name} must be positive")
|
|
elif not isinstance(value, str) or not value:
|
|
raise ValueError(f"{name} must be a nonempty string")
|
|
values[name] = value
|
|
for name in ("max_requests_per_minute", "shotmap_max_images"):
|
|
if int(values[name]) != values[name]:
|
|
raise ValueError(f"{name} must be an integer")
|
|
values[name] = int(values[name])
|
|
ZoneInfo(values["timezone"])
|
|
return cls(**values)
|
|
|
|
|
|
def root(payload):
|
|
if not isinstance(payload, dict):
|
|
return {}
|
|
if isinstance(payload.get("data"), dict) and not any(k in payload for k in ("items", "result")):
|
|
return payload["data"]
|
|
return payload
|
|
|
|
|
|
def game_state(payload):
|
|
value = root(payload)
|
|
value = value.get("result") if isinstance(value.get("result"), dict) else value
|
|
return str(value.get("gameStatus") or value.get("state") or value.get("status") or "").casefold()
|
|
|
|
|
|
FINAL_STATES = {"result", "resultconfirmed", "finished", "ended", "final", "game over"}
|
|
LIVE_STATES = {"online", "live"}
|
|
|
|
|
|
def valid_payload(name, payload):
|
|
value = root(payload)
|
|
if not value or "error" in value:
|
|
return False
|
|
if str(value.get("status", "")).casefold() in ("error", "fail", "no-status"):
|
|
return False
|
|
if str(value.get("message", "")).casefold() == "not found":
|
|
return False
|
|
if name in ("seasons", "calendar", "actual-standings", "season_stats", "league_stats"):
|
|
return isinstance(value.get("items"), list)
|
|
result = value.get("result")
|
|
if name == "play-by-play":
|
|
return isinstance(result, list)
|
|
if name == "live-status":
|
|
ls = result if isinstance(result, dict) else value
|
|
return str(ls.get("message", "")).casefold() != "not found" and any(
|
|
k in ls for k in ("gameStatus", "state", "period", "scoreA", "scoreB", "second")
|
|
)
|
|
if name in ("game", "box-score"):
|
|
if not isinstance(result, dict) or not isinstance(result.get("teams"), list):
|
|
return False
|
|
teams = result["teams"]
|
|
if not teams or any(not isinstance(team, dict) for team in teams):
|
|
return False
|
|
if not {team.get("teamNumber") for team in teams}.issuperset({1, 2}):
|
|
return False
|
|
if name == "game":
|
|
return (isinstance(result.get("game"), dict) and result["game"].get("id") is not None
|
|
and all(isinstance(result.get(key), dict) for key in ("team1", "team2")))
|
|
return all(isinstance(team.get("total"), dict) and "points" in team["total"]
|
|
for team in teams if team.get("teamNumber") in (1, 2))
|
|
if name == "pregame-full-stats":
|
|
return isinstance(result, dict) and all(isinstance(result.get(k), list) for k in ("team1PlayersStats", "team2PlayersStats"))
|
|
return isinstance(result, dict) and bool(result)
|
|
|
|
|
|
class FetchError(Exception):
|
|
def __init__(self, message, delay=5, status_code=None):
|
|
super().__init__(message)
|
|
self.delay, self.status_code = delay, status_code
|
|
|
|
|
|
class ApiClient:
|
|
"""One serialized transport and one budget for EVERY API call on this machine."""
|
|
def __init__(self, config, session, logger=None, clock=None):
|
|
self.config, self.session = config, session
|
|
self.logger = logger or logging.getLogger(__name__)
|
|
self.clock = clock or time
|
|
self.transport_lock = threading.Lock()
|
|
self.lock = threading.RLock()
|
|
self.starts = deque()
|
|
self.next_request_at = 0.0
|
|
self.blocked_until = 0.0
|
|
self.total_requests = 0
|
|
self.diagnostics = {}
|
|
self.failures = {}
|
|
self.notice_at = {}
|
|
|
|
def _wait_slot(self, stop):
|
|
while not stop.is_set():
|
|
now = self.clock.monotonic()
|
|
with self.lock:
|
|
while self.starts and self.starts[0] <= now - 60:
|
|
self.starts.popleft()
|
|
deadline = max(self.next_request_at, self.blocked_until)
|
|
if len(self.starts) >= self.config.max_requests_per_minute:
|
|
deadline = max(deadline, self.starts[0] + 60)
|
|
delay = deadline - now
|
|
if delay <= 0:
|
|
self.starts.append(now)
|
|
self.total_requests += 1
|
|
self.next_request_at = now + 60 / self.config.max_requests_per_minute
|
|
return
|
|
if stop.wait(min(delay, 1)):
|
|
break
|
|
raise FetchError("Polling stopped", 0)
|
|
|
|
def _retry_after(self, headers, fallback):
|
|
value = headers.get("Retry-After")
|
|
if value is None:
|
|
return fallback
|
|
try:
|
|
return max(0, float(value))
|
|
except (TypeError, ValueError):
|
|
try:
|
|
dt = parsedate_to_datetime(value)
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return max(0, dt.timestamp() - self.clock.time())
|
|
except (TypeError, ValueError, OverflowError):
|
|
return fallback
|
|
|
|
def fetch(self, name, url, stop):
|
|
code = None
|
|
try:
|
|
with self.transport_lock:
|
|
self._wait_slot(stop)
|
|
response = self.session.get(url, timeout=(self.config.request_connect_timeout,
|
|
self.config.request_read_timeout))
|
|
try:
|
|
code = response.status_code
|
|
if code == 429 or code in (401, 403):
|
|
fallback = self.config.rate_limit_pause_seconds if code == 429 else self.config.forbidden_pause_seconds
|
|
delay = max(fallback, self._retry_after(response.headers, fallback))
|
|
with self.lock:
|
|
self.blocked_until = max(self.blocked_until, self.clock.monotonic() + delay)
|
|
raise FetchError(f"HTTP {code}: shared pause on this machine", delay, code)
|
|
if not 200 <= code < 300:
|
|
raise FetchError(f"HTTP {code}", self.config.retry_initial_seconds, code)
|
|
payload = root(response.json())
|
|
if not valid_payload(name, payload):
|
|
raise FetchError("Data not ready or invalid response", max(30, self.config.retry_initial_seconds))
|
|
finally:
|
|
response.close()
|
|
except Exception as ex:
|
|
if stop.is_set():
|
|
raise FetchError("Polling stopped", 0) from ex
|
|
with self.lock:
|
|
n = min(self.failures.get((name, url), 0) + 1, 16)
|
|
self.failures[(name, url)] = n
|
|
delay = max(getattr(ex, "delay", 0), min(self.config.retry_max_seconds,
|
|
self.config.retry_initial_seconds * 2 ** (n - 1)))
|
|
self.diagnostics[name] = {"ok": False, "http_status": code, "message": str(ex),
|
|
"retry_seconds": delay}
|
|
now = self.clock.monotonic()
|
|
if now >= self.notice_at.get(name, -1):
|
|
self.logger.warning("[API] %s: %s; retry in %.1fs", name, ex, delay)
|
|
self.notice_at[name] = now + 300
|
|
raise FetchError(str(ex), delay, code) from ex
|
|
with self.lock:
|
|
self.failures.pop((name, url), None)
|
|
self.diagnostics[name] = {"ok": True, "http_status": code, "updated_at": datetime.now().isoformat()}
|
|
return payload
|
|
|
|
def health(self):
|
|
with self.lock:
|
|
now = self.clock.monotonic()
|
|
return {"total_requests": self.total_requests,
|
|
"requests_last_minute": sum(t > now - 60 for t in self.starts),
|
|
"limit_per_minute": self.config.max_requests_per_minute,
|
|
"pause_seconds": round(max(0, self.blocked_until - now), 1),
|
|
"sources": copy.deepcopy(self.diagnostics)}
|
|
|
|
|
|
class SnapshotStore:
|
|
def __init__(self, directory, identity, logger):
|
|
self.path = Path(directory) / (hashlib.sha256(identity.encode()).hexdigest()[:20] + ".json")
|
|
self.logger = logger
|
|
|
|
def load(self):
|
|
try:
|
|
value = json.loads(self.path.read_text(encoding="utf-8"))
|
|
if not isinstance(value, dict) or value.get("version") != 1:
|
|
return {}
|
|
if (not isinstance(value.get("latest_data", {}), dict)
|
|
or not isinstance(value.get("resource_game_ids", {}), dict)
|
|
or (value.get("calendar") is not None and not isinstance(value["calendar"], dict))):
|
|
raise ValueError("invalid snapshot structure")
|
|
checked_at = value.get("season_checked_at", 0)
|
|
if not isinstance(checked_at, (int, float)) or not math.isfinite(checked_at):
|
|
raise ValueError("invalid season timestamp")
|
|
return value
|
|
except FileNotFoundError:
|
|
return {}
|
|
except (OSError, ValueError) as ex:
|
|
self.logger.warning("[cache] snapshot ignored: %s", ex)
|
|
return {}
|
|
|
|
def save(self, value):
|
|
try:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
temporary = self.path.with_suffix(".tmp")
|
|
temporary.write_text(json.dumps(dict(version=1, **value), ensure_ascii=False), encoding="utf-8")
|
|
os.replace(temporary, self.path)
|
|
except (OSError, TypeError, ValueError) as ex:
|
|
self.logger.warning("[cache] snapshot not saved: %s", ex)
|
|
|
|
|
|
class MatchController:
|
|
"""Calendar-driven state machine, running on one existing server only."""
|
|
def __init__(self, namespace, config, client, store):
|
|
self.n, self.config, self.client, self.store = namespace, config, client, store
|
|
self.next_calendar = 0
|
|
self.next_seasons = 0
|
|
self.next_snapshot = 0
|
|
self.calendar_retry_at = 0
|
|
self.bootstrap_done = False
|
|
|
|
def restore(self):
|
|
saved = self.store.load()
|
|
with self.n["DATA_LOCK"]:
|
|
self.n["SEASON"] = saved.get("season")
|
|
self.n["CALENDAR"] = saved.get("calendar")
|
|
self.n["latest_data"].update(saved.get("latest_data") or {})
|
|
self.n["RESOURCE_GAME_IDS"].update(saved.get("resource_game_ids") or {})
|
|
self.n["SEASON_CHECKED_AT"] = saved.get("season_checked_at", 0)
|
|
status = self.n["latest_data"].get("live-status", {}).get("data")
|
|
status_gid = self.n["RESOURCE_GAME_IDS"].get("live-status")
|
|
if status_gid is not None and valid_payload("live-status", status):
|
|
self.n["GAME_STATUS_CACHE"][str(status_gid)] = {"payload": status}
|
|
self.next_seasons = time.monotonic() + 86400 if saved.get("season_checked_at", 0) > time.time() - 86400 else 0
|
|
|
|
def checkpoint(self):
|
|
with self.n["DATA_LOCK"]:
|
|
data = {k: copy.deepcopy(v) for k, v in self.n["latest_data"].items() if not k.startswith("excel")}
|
|
value = {"season": self.n["SEASON"], "calendar": copy.deepcopy(self.n["CALENDAR"]),
|
|
"latest_data": data, "resource_game_ids": dict(self.n["RESOURCE_GAME_IDS"]),
|
|
"season_checked_at": self.n.get("SEASON_CHECKED_AT", 0)}
|
|
self.store.save(value)
|
|
|
|
def tick(self, now=None, monotonic=None):
|
|
n, cfg = self.n, self.config
|
|
now = now or n["broadcast_now"]()
|
|
mono = time.monotonic() if monotonic is None else monotonic
|
|
if mono >= self.next_seasons:
|
|
try:
|
|
seasons = self.client.fetch("seasons", n["api_url"]("seasons"), n["stop_event"])
|
|
items = seasons.get("items") or []
|
|
if items:
|
|
n["SEASON"] = items[0]["season"]
|
|
n["SEASON_CHECKED_AT"] = time.time()
|
|
self.next_seasons = mono + 86400
|
|
except FetchError as ex:
|
|
self.next_seasons = mono + max(300, ex.delay)
|
|
if n["SEASON"] is None:
|
|
n["SEASON"] = now.year + (1 if now.month > 9 else 0)
|
|
if mono >= max(self.next_calendar, self.calendar_retry_at):
|
|
try:
|
|
calendar = self.client.fetch("calendar", n["api_url"]("calendar"), n["stop_event"])
|
|
with n["DATA_LOCK"]:
|
|
n["CALENDAR"] = calendar
|
|
self.calendar_retry_at = 0
|
|
self.bootstrap_done = True
|
|
except FetchError as ex:
|
|
self.calendar_retry_at = mono + max(60, ex.delay)
|
|
interval = cfg.calendar_prepare_seconds if n["CURRENT_THREADS_MODE"] in ("prestart", "live", "finished") else cfg.calendar_wait_seconds
|
|
self.next_calendar = mono + interval + random.uniform(0, interval * 0.05)
|
|
# Never activate an old restored schedule until a successful check in this process.
|
|
if not self.bootstrap_done:
|
|
return
|
|
gid, start, today, state = n["pick_game_for_team"](n["CALENDAR"], now=now)
|
|
# An active match belongs to this arena even when it continues past midnight.
|
|
current_gid = n.get("GAME_ID")
|
|
finishing = (n["CURRENT_THREADS_MODE"] == "finished" and n.get("OFFLINE_SWITCH_AT")
|
|
and time.time() < n["OFFLINE_SWITCH_AT"])
|
|
if current_gid is not None and (n["CURRENT_THREADS_MODE"] == "live" or finishing):
|
|
current = next((item.get("game") or {} for item in (n["CALENDAR"] or {}).get("items", [])
|
|
if str((item.get("game") or {}).get("id")) == str(current_gid)), {})
|
|
gid, start, today = current_gid, n.get("GAME_START_DT"), True
|
|
state = current.get("gameStatus") or ("Result" if finishing else "Online")
|
|
if not today or gid is None:
|
|
n["GAME_TODAY"] = False
|
|
n["GAME_SOON"] = False
|
|
n["GAME_START_DT"] = None
|
|
n["STATUS"] = "no_game_today"
|
|
n["start_waiting_threads"]()
|
|
# A previous game remains available from disk; no API request for it.
|
|
n["GAME_ID"] = n["get_cached_game_id"]() or gid
|
|
return
|
|
changed = str(gid) != str(n.get("GAME_ID"))
|
|
n["GAME_ID"], n["GAME_START_DT"], n["GAME_TODAY"] = gid, start, True
|
|
state = str(state or "").casefold()
|
|
delta = (start - now).total_seconds() if start else 0
|
|
n["GAME_SOON"] = delta <= cfg.prepare_minutes * 60
|
|
if changed:
|
|
n["OFFLINE_SWITCH_AT"] = None
|
|
n["start_waiting_threads"]()
|
|
n["STATUS"] = "today_not_started"
|
|
if str(gid) in n["FINISHED_GAME_IDS"]:
|
|
if n["OFFLINE_SWITCH_AT"] and time.time() >= n["OFFLINE_SWITCH_AT"]:
|
|
n["OFFLINE_SWITCH_AT"] = None
|
|
n["STATUS"] = "finished_today"
|
|
n["start_waiting_threads"]()
|
|
return
|
|
if state in FINAL_STATES:
|
|
n["begin_finished_mode"](gid)
|
|
elif n["CURRENT_THREADS_MODE"] == "live" and not changed:
|
|
pass # live-status has fresher information than calendar
|
|
elif state in LIVE_STATES:
|
|
n["STATUS"] = "live"
|
|
n["start_live_threads"](n["SEASON"], gid)
|
|
self.next_calendar = min(self.next_calendar, mono + cfg.calendar_prepare_seconds)
|
|
elif delta <= cfg.prepare_minutes * 60:
|
|
n["STATUS"] = "today_not_started"
|
|
n["start_preparation_threads"](n["SEASON"], gid, delta <= 300)
|
|
self.next_calendar = min(self.next_calendar, mono + cfg.calendar_prepare_seconds)
|
|
else:
|
|
n["STATUS"] = "today_not_started"
|
|
n["start_waiting_threads"]()
|
|
|
|
def run(self):
|
|
stop = self.n["stop_event"]
|
|
self.restore()
|
|
if stop.wait(random.uniform(0, self.config.startup_jitter_seconds)):
|
|
return
|
|
while not stop.is_set():
|
|
try:
|
|
self.tick()
|
|
now = time.monotonic()
|
|
if now >= self.next_snapshot:
|
|
self.checkpoint()
|
|
self.next_snapshot = now + self.config.snapshot_seconds
|
|
except Exception:
|
|
self.n["logger"].exception("[schedule] update failed; waiting before next attempt")
|
|
stop.wait(30)
|
|
stop.wait(2)
|
|
self.checkpoint()
|