import time from datetime import datetime, timedelta, timezone from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry import requests import json import os import tempfile import argparse import platform import sys import logging import pandas as pd import logging.config from typing import Any, Dict, List from zoneinfo import ZoneInfo import threading from concurrent.futures import ThreadPoolExecutor, as_completed HOST = "https://ref.russiabasket.org" APP_TZ = ZoneInfo("Europe/Moscow") MYHOST = platform.node() TELEGRAM_BOT_TOKEN = "7639240596:AAH0YtdQoWZSC-_R_EW4wKAHHNLIA0F_ARY" # TELEGRAM_CHAT_ID = 228977654 TELEGRAM_CHAT_ID = -4803699526 if not os.path.exists("logs"): os.makedirs("logs") _write_lock = threading.Lock() URLS = [ { "name": "seasons", "url": "{host}/api/abc/comps/seasons?Tag={league}&Lang={lang}", "interval": 86400, # раз в сутки }, { "name": "standings", "url": "{host}/api/abc/comps/standings?tag={league}&season={season}&lang={lang}", "interval": 1800, # раз в 30 минут }, { "name": "calendar", "url": "{host}/api/abc/comps/calendar?Tag={league}&Season={season}&Lang={lang}&MaxResultCount=1000", "interval": 86400, # раз в сутки }, { "name": "game", "url": "{host}/api/abc/games/game?Id={game_id}&Lang={lang}", "interval": 600, # раз в 10 минут }, { "name": "pregame", "url": "{host}/api/abc/games/pregame?tag={league}&season={season}&Id={game_id}&Lang={lang}", "interval": 86400, # раз в сутки }, { "name": "pregame-fullstats", "url": "{host}/api/abc/games/pregame-fullstats?tag={league}&season={season}&id={game_id}&lang={lang}", "interval": 600, # раз в 10 минут }, { "name": "live-status", "url": "{host}/api/abc/games/live-status?id={game_id}", "interval": 1, # каждую секунду }, { "name": "box-score", "url": "{host}/api/abc/games/box-score?Id={game_id}", "interval": 1, # каждую секунду }, { "name": "play-by-play", "url": "{host}/api/abc/games/play-by-play?Id={game_id}", "interval": 1, # каждую секунду }, ] LOG_CONFIG = { "version": 1, "handlers": { "telegram": { "class": "telegram_handler.TelegramHandler", "level": "INFO", "token": TELEGRAM_BOT_TOKEN, "chat_id": TELEGRAM_CHAT_ID, "formatter": "telegram", }, "console": { "class": "logging.StreamHandler", "level": "INFO", "formatter": "simple", "stream": "ext://sys.stdout", }, "file": { "class": "logging.FileHandler", "level": "DEBUG", "formatter": "simple", "filename": f"logs/GFX_{MYHOST}.log", "encoding": "utf-8", }, }, "loggers": { __name__: {"handlers": ["console", "file", "telegram"], "level": "DEBUG"}, }, "formatters": { "telegram": { "class": "telegram_handler.HtmlFormatter", "format": f"%(levelname)s [{MYHOST.upper()}] %(message)s", "use_emoji": "True", }, "simple": { "class": "logging.Formatter", "format": "%(asctime)s %(levelname)-8s %(funcName)s() - %(message)s", "datefmt": "%d.%m.%Y %H:%M:%S", }, }, } logging.config.dictConfig(LOG_CONFIG) logger = logging.getLogger(__name__) logger.handlers[2].formatter.use_emoji = True def create_session() -> requests.Session: session = requests.Session() retries = Retry( total=3, backoff_factor=0.5, status_forcelist=[429, 500, 502, 503, 504], ) session.mount("https://", HTTPAdapter(max_retries=retries)) session.headers.update( { "Connection": "keep-alive", "Accept": "application/json, */*", "Accept-Encoding": "gzip, deflate, br", "User-Agent": "game-watcher/1.0", } ) return session def get_json(session, url: str, name: str): resp = session.get(url, timeout=10) resp.raise_for_status() data = resp.json() atomic_write_json(data, f"api_{name}") return data def build_url(name: str, **kwargs) -> str: """ Собрать конечный URL по имени ручки из URLS. Пример: build_url("standings", host=..., league=..., season=..., lang=...) """ template = next((u["url"] for u in URLS if u["name"] == name), None) if not template: raise ValueError(f"Unknown URL name: {name}") return template.format(**kwargs) def atomic_write_json(data, name: str, out_dir: str = "static"): """ Сохраняет data в static/.json атомарно. Потокобезопасно (глобальный lock). """ os.makedirs(out_dir, exist_ok=True) filename = os.path.join(out_dir, f"{name}.json") with _write_lock: with tempfile.NamedTemporaryFile( "w", delete=False, dir=out_dir, encoding="utf-8" ) as tmp_file: json.dump(data, tmp_file, ensure_ascii=False, indent=2) tmp_file.flush() os.fsync(tmp_file.fileno()) tmp_name = tmp_file.name os.replace(tmp_name, filename) def get_items(data): for k, v in data.items(): if isinstance(v, list): return data[k] def poll_one_endpoint(session, endpoint_name, league, season, game_id, lang): """ Дёрнуть конкретный endpoint и вернуть (endpoint_name, data или None) """ if endpoint_name == "live-status": data = fetch_api_data( session, "live-status", host=HOST, game_id=game_id, ) return endpoint_name, data if endpoint_name == "box-score": data = fetch_api_data( session, "box-score", host=HOST, game_id=game_id, ) return endpoint_name, data if endpoint_name == "play-by-play": data = fetch_api_data( session, "play-by-play", host=HOST, game_id=game_id, ) return endpoint_name, data if endpoint_name == "game": data = fetch_api_data( session, "game", host=HOST, game_id=game_id, lang=lang, ) return endpoint_name, data if endpoint_name == "pregame-fullstats": data = fetch_api_data( session, "pregame-fullstats", host=HOST, league=league, season=season, game_id=game_id, lang=lang, ) return endpoint_name, data # fallback — вдруг добавим что-то ещё data = fetch_api_data(session, endpoint_name, host=HOST, game_id=game_id, lang=lang) return endpoint_name, data def fetch_api_data(session, name: str, name_save: str = None, **kwargs): """ Универсальная функция для получения данных с API: 1. Собирает URL по имени ручки 2. Получает JSON 3. Возвращает основной список данных (если есть) """ url = build_url(name, **kwargs) try: json_data = get_json(session, url, name_save or name) if json_data: items = get_items(json_data) return items if items is not None else json_data return None except Exception as ex: logger.error(f"{url} | {ex}") def parse_game_start_dt(item: dict) -> datetime: """ Достаёт дату/время начала матча из объекта расписания и приводит к APP_TZ. Приоритет полей: 1) game.defaultZoneDateTime — уже в "дефолтной зоне" лиги (например, +03:00) 2) game.scheduledTime — ISO 8601 с оффсетом (например, 2025-09-30T19:00:00+04:00) 3) game.startTime — если API когда-то его заполняет 4) (fallback) game.localDate + game.localTime — считаем, что это локальное время площадки, задаём tz=APP_TZ Возвращает aware-datetime в APP_TZ. """ g = item.get("game", {}) if "game" in item else item raw = g.get("defaultZoneDateTime") or g.get("scheduledTime") or g.get("startTime") if raw: try: dt = datetime.fromisoformat(raw) # ISO-8601 return dt.astimezone(APP_TZ) except Exception as e: raise RuntimeError(f"Ошибка парсинга ISO времени '{raw}': {e}") # Fallback: localDate + localTime (пример: "30.09.2025" + "19:00") ld, lt = g.get("localDate"), g.get("localTime") if ld and lt: try: naive = datetime.strptime(f"{ld} {lt}", "%d.%m.%Y %H:%M") aware = naive.replace(tzinfo=APP_TZ) return aware except Exception as e: raise RuntimeError(f"Ошибка парсинга localDate/localTime '{ld} {lt}': {e}") raise RuntimeError( "Не найдено ни одного подходящего поля времени (defaultZoneDateTime/scheduledTime/startTime/localDate+localTime)." ) def get_game_id(team_games: list[dict], team: str) -> tuple[dict | None, dict | None]: """ Принимаем все расписание и ищем для домашней команды game_id. Если сегодня нет матча, то берем game_id прошлой игры. """ now = datetime.now(APP_TZ) today = now.date() today_game = None last_played = None for g in team_games: start = parse_game_start_dt(g) status = g.get("game", {}).get("gameStatus", "").lower() if ( start.date() == today and today_game is None and g["team1"]["name"].lower() == team.lower() ): today_game = g last_played = None elif ( start <= now and status == "resultconfirmed" and g["team1"]["name"].lower() == team.lower() ): today_game = None last_played = g return today_game, last_played def is_game_live(game_obj: dict) -> bool: """ Пытаемся понять, идёт ли сейчас матч. game_obj ожидается как today_game["game"] (из calendar). """ status = (game_obj.get("gameStatus") or "").lower() # эвристика: # - "resultconfirmed" -> матч кончился # - "scheduled" / "notstarted" -> ещё не начался # всё остальное считаем лайвом if status in ("resultconfirmed", "finished", "final"): return False if status in ("scheduled", "notstarted", "draft"): return False return True def get_interval_by_name(name: str) -> int: """ вернуть interval из URLS по имени ручки """ for u in URLS: if u["name"] == name: return u["interval"] raise ValueError(f"interval not found for {name}") def run_live_loop( parent_session_unused, league: str, season: str, game_id: int, lang: str, game_meta: dict, ): """ Поток, который дергает онлайн API матча. По завершении матча говорит рендеру остановиться. """ logger.info( f"[LIVE_THREAD] start live loop for game_id={game_id} (league={league}, season={season})" ) # создаём свою сессию, чтобы не делить session между потоками session = create_session() # общий stop_event для live и render stop_event = threading.Event() # запускаем рендер-поток render_thread = threading.Thread( target=render_loop, args=(stop_event,), # только stop_event, out_name используем дефолт "ui_state" daemon=True, ) render_thread.start() logger.info("[LIVE_THREAD] render thread spawned") try: # крутим опрос API до конца матча poll_game_live( session=session, league=league, season=season, game_id=game_id, lang=lang, game_meta=game_meta, ) except Exception as e: logger.exception(f"[LIVE_THREAD] crash in live loop for game_id={game_id}: {e}") finally: # матч кончился -> выключаем рендер stop_event.set() logger.info(f"[LIVE_THREAD] stop live loop for game_id={game_id}") def poll_game_live( session, league: str, season: str, game_id: int, lang: str, game_meta: dict, stop_event: threading.Event, ): """ Онлайн-цикл: - "game" раз в 600 сек (pregame-fullstats можно вернуть позже) - "live-status", "box-score", "play-by-play" раз в 1 сек Цикл выходит, когда матч перестаёт быть live ИЛИ когда сработал stop_event. """ slow_endpoints = ["game"] fast_endpoints = ["live-status", "box-score", "play-by-play"] last_call = {ep: 0 for ep in slow_endpoints + fast_endpoints} # пул потоков живет весь матч with ThreadPoolExecutor(max_workers=5) as executor: while True: # внешняя принудительная остановка if stop_event.is_set(): logger.info(f"[LIVE_LOOP] stop_event set -> break live poll for game {game_id}") break now = time.time() # какие ручки надо дёрнуть прямо сейчас to_run = [] for ep in fast_endpoints + slow_endpoints: interval = get_interval_by_name(ep) if now - last_call[ep] >= interval: to_run.append(ep) futures = [] if to_run: for ep in to_run: futures.append( executor.submit( poll_one_endpoint, session, ep, league, season, game_id, lang, ) ) game_finished = False for fut in as_completed(futures): try: ep_name, data = fut.result() last_call[ep_name] = now if ep_name == "live-status": if isinstance(data, dict): st = ( data.get("status") or data.get("gameStatus") or "" ).lower() if st in ("resultconfirmed", "finished", "final"): logger.info( f"Game {game_id} finished by live-status -> stop loop" ) game_finished = True except Exception as e: logger.exception(f"poll endpoint error: {e}") # страховка по календарю (game_meta мог устареть, но лучше чем ничего) if not is_game_live(game_meta): logger.info( f"Game {game_id} no longer live by calendar meta -> stop loop" ) break if game_finished: break time.sleep(0.2) logger.debug("live poll tick ok") def get_data_API(session, league: str, team: str, lang: str): json_seasons = fetch_api_data( session, "seasons", host=HOST, league=league, lang=lang ) if not json_seasons: logger.error("Не удалось получить список сезонов") return season = json_seasons[0]["season"] # standings и calendar просто кешируем в файлы fetch_api_data( session, "standings", host=HOST, league=league, season=season, lang=lang ) json_calendar = fetch_api_data( session, "calendar", host=HOST, league=league, season=season, lang=lang ) if not json_calendar: logger.error("Не удалось получить список матчей") return today_game, last_played = get_game_id(json_calendar, team) # если есть завершённая последняя игра — просто сохраним статический срез и выходим if last_played and not today_game: game_id = last_played["game"]["id"] logger.info(f"Последний завершённый матч id={game_id}") fetch_api_data(session, "game", host=HOST, game_id=game_id, lang=lang) return # если есть матч сегодня if today_game: game_id = today_game["game"]["id"] logger.info(f"Онлайн матч id={game_id}") # базовые данные прямо сейчас fetch_api_data(session, "game", host=HOST, game_id=game_id, lang=lang) # если матч реально идёт -> запускаем live-петлю и рендер if is_game_live(today_game["game"]): # отдельная сессия для live-пула (как раньше) live_session = create_session() # единый stop_event для всего матча stop_event = threading.Event() # поток рендера render_thread = threading.Thread( target=render_loop, args=(stop_event, "ui_state"), # имя файла можешь менять daemon=False, ) render_thread.start() logger.info("[MAIN] render thread spawned") # поток live-пулинга API def live_worker(): try: poll_game_live( session=live_session, league=league, season=season, game_id=game_id, lang=lang, game_meta=today_game["game"], stop_event=stop_event, ) except Exception as e: logger.exception(f"[LIVE_THREAD] crash in live loop: {e}") live_thread = threading.Thread( target=live_worker, daemon=False, ) live_thread.start() logger.info("[MAIN] live thread spawned") # дожидаемся окончания live_thread (то есть завершения матча или ошибки) live_thread.join() logger.info("[MAIN] live thread finished") # говорим рендеру остановиться stop_event.set() # ждём корректного завершения рендера render_thread.join() logger.info("[MAIN] render thread finished") return logger.info("Для этой команды игр сегодня нет и нет завершённой последней игры.") def read_local_json(name: str, in_dir: str = "static"): """ Безопасно читает static/.json. Если файла нет или он в процессе записи -> вернёт None, но не упадёт. """ filename = os.path.join(in_dir, f"{name}.json") try: with open(filename, "r", encoding="utf-8") as f: return json.load(f) except FileNotFoundError: return None except json.JSONDecodeError: # файл мог быть в моменте перезаписи -> просто пропускаем этот тик return None except Exception as ex: logger.exception(f"read_local_json({name}) error: {ex}") return None def _now_iso() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") def build_render_state() -> dict: """ Читает сырые api_*.json и собирает удобный json для другой программы (графики и т.п.). Возвращает dict, который потом пишем в static/ui_state.json """ game_data = read_local_json("api_game") live_status_data = read_local_json("api_live-status") box_score_data = read_local_json("api_box-score") play_by_play_data = read_local_json("api_play-by-play") game_data = game_data["result"] # базовый безопасный каркас for index_team, team in enumerate(game_data["teams"][1:]): box_team = box_score_data["result"]["teams"][index_team] for player in team.get("starts", []): stat = next( ( s for s in box_team.get("starts", []) if s.get("startNum") == player.get("startNum") ), None, ) if stat: player["stats"] = stat team["total"] = box_team.get("total", {}) game_data["plays"] = play_by_play_data.get("result", []) game_data["scoreByPeriods"] = box_score_data["result"].get("scoreByPeriods", []) game_data["fullScore"] = box_score_data["result"].get("fullScore", {}) game_data["live_status"] = live_status_data["result"] merged: Dict[str, Any] = { "meta": { "generatedAt": _now_iso(), "sourceHints": { "boxScoreHas": "", "pbpLen": "", }, }, "result": game_data, } return merged def format_time(seconds: float | int) -> str: """ Форматирует время в секундах в строку "M:SS". Args: seconds (float | int): Количество секунд. Returns: str: Время в формате "M:SS". """ try: total_seconds = int(float(seconds)) minutes = total_seconds // 60 sec = total_seconds % 60 return f"{minutes}:{sec:02}" except (ValueError, TypeError): return "0:00" def Json_Team_Generation( merged: dict, *, out_dir: str = "static", who: str | None = None ) -> None: """ Единая точка: принимает уже нормализованный merged, делает нужные вычисления (если надо) и сохраняет в JSON. """ # Здесь можно делать любые расчёты/агрегации... # Пример предохранителя: сортировка плей-бай-плея по sequence # plays = merged.get("result", {}).get("plays", []) # if plays and isinstance(plays, list): # try: # plays.sort(key=lambda e: (e.get("sequence") is None, e.get("sequence"), e.get("time") or e.get("clock"))) # except Exception: # pass # Имя файла # print(merged) # merged = if who == "team1": for i in merged["result"]["teams"]: if i["teamNumber"] == 1: payload = i elif who == "team2": for i in merged["result"]["teams"]: if i["teamNumber"] == 2: payload = i # online = ( # True # if json_live_status # and "status" in json_live_status # and json_live_status["status"] == "Ok" # and json_live_status["result"]["gameStatus"] == "Online" # else False # ) online = False role_list = [ ("Center", "C"), ("Guard", "G"), ("Forward", "F"), ("Power Forward", "PF"), ("Small Forward", "SF"), ("Shooting Guard", "SG"), ("Point Guard", "PG"), ("Forward-Center", "FC"), ] starts = payload["starts"] team = [] for item in starts: player = { "id": (item["personId"] if item["personId"] else ""), "num": item["displayNumber"], "startRole": item["startRole"], "role": item["positionName"], "roleShort": ( [ r[1] for r in role_list if r[0].lower() == item["positionName"].lower() ][0] if any(r[0].lower() == item["positionName"].lower() for r in role_list) else "" ), "NameGFX": ( f"{item['firstName'].strip()} {item['lastName'].strip()}" if item["firstName"] is not None and item["lastName"] is not None else "Команда" ), "captain": item["isCapitan"], "age": item["age"] if item["age"] is not None else 0, "height": f'{item["height"]} cm' if item["height"] else 0, "weight": f'{item["weight"]} kg' if item["weight"] else 0, "isStart": (item["stats"]["isStart"] if item["stats"] else False), "isOn": ( "🏀" if item["stats"] and item["stats"]["isOnCourt"] is True else "" ), "flag": f"https://flagicons.lipis.dev/flags/4x3/{'ru' if item['countryId'] is None and item['countryName'] == 'Russia' else '' if item['countryId'] is None else item['countryId'].lower() if item['countryName'] is not None else ''}.svg", "pts": item["stats"]["points"] if item["stats"] else 0, "pt-2": ( f"{item['stats']['goal2']}/{item['stats']['shot2']}" if item["stats"] else 0 ), "pt-3": ( f"{item['stats']['goal3']}/{item['stats']['shot3']}" if item["stats"] else 0 ), "pt-1": ( f"{item['stats']['goal1']}/{item['stats']['shot1']}" if item["stats"] else 0 ), "fg": ( f"{item['stats']['goal2'] + item['stats']['goal3']}/{item['stats']['shot2'] + item['stats']['shot3']}" if item["stats"] else 0 ), "ast": item["stats"]["assist"] if item["stats"] else 0, "stl": item["stats"]["steal"] if item["stats"] else 0, "blk": item["stats"]["block"] if item["stats"] else 0, "blkVic": item["stats"]["blocked"] if item["stats"] else 0, "dreb": item["stats"]["defReb"] if item["stats"] else 0, "oreb": item["stats"]["offReb"] if item["stats"] else 0, "reb": ( item["stats"]["defReb"] + item["stats"]["offReb"] if item["stats"] else 0 ), "to": item["stats"]["turnover"] if item["stats"] else 0, "foul": item["stats"]["foul"] if item["stats"] else 0, "foulT": item["stats"]["foulT"] if item["stats"] else 0, "foulD": item["stats"]["foulD"] if item["stats"] else 0, "foulC": item["stats"]["foulC"] if item["stats"] else 0, "foulB": item["stats"]["foulB"] if item["stats"] else 0, "fouled": item["stats"]["foulsOn"] if item["stats"] else 0, "plusMinus": item["stats"]["plusMinus"] if item["stats"] else 0, "dunk": item["stats"]["dunk"] if item["stats"] else 0, "kpi": ( item["stats"]["points"] + item["stats"]["defReb"] + item["stats"]["offReb"] + item["stats"]["assist"] + item["stats"]["steal"] + item["stats"]["block"] + item["stats"]["foulsOn"] + (item["stats"]["goal1"] - item["stats"]["shot1"]) + (item["stats"]["goal2"] - item["stats"]["shot2"]) + (item["stats"]["goal3"] - item["stats"]["shot3"]) - item["stats"]["turnover"] - item["stats"]["foul"] if item["stats"] else 0 ), "time": (format_time(item["stats"]["second"]) if item["stats"] else "0:00"), "pts1q": 0, "pts2q": 0, "pts3q": 0, "pts4q": 0, "pts1h": 0, "pts2h": 0, "Name1GFX": (item["firstName"].strip() if item["firstName"] else ""), "Name2GFX": (item["lastName"].strip() if item["lastName"] else ""), "photoGFX": ( os.path.join( "D:\\Photos", merged["result"]["league"]["abcName"], merged["result"][who]["name"], # LEAGUE, # data[who], f"{item['displayNumber']}.png", ) if item["startRole"] == "Player" else "" ), # "season": text, "isOnCourt": (item["stats"]["isOnCourt"] if item["stats"] else False), # "AvgPoints": ( # row_player_season_avg["points"] # if row_player_season_avg # and row_player_season_avg["points"] != "" # else "0.0" # ), # "AvgAssist": ( # row_player_season_avg["assist"] # if row_player_season_avg # and row_player_season_avg["assist"] != "" # else "0.0" # ), # "AvgBlocks": ( # row_player_season_avg["blockShot"] # if row_player_season_avg # and row_player_season_avg["blockShot"] != "" # else "0.0" # ), # "AvgDefRebound": ( # row_player_season_avg["defRebound"] # if row_player_season_avg # and row_player_season_avg["defRebound"] != "" # else "0.0" # ), # "AvgOffRebound": ( # row_player_season_avg["offRebound"] # if row_player_season_avg # and row_player_season_avg["offRebound"] != "" # else "0.0" # ), # "AvgRebound": ( # row_player_season_avg["rebound"] # if row_player_season_avg # and row_player_season_avg["rebound"] != "" # else "0.0" # ), # "AvgSteal": ( # row_player_season_avg["steal"] # if row_player_season_avg # and row_player_season_avg["steal"] != "" # else "0.0" # ), # "AvgTurnover": ( # row_player_season_avg["turnover"] # if row_player_season_avg # and row_player_season_avg["turnover"] != "" # else "0.0" # ), # "AvgFoul": ( # row_player_season_avg["foul"] # if row_player_season_avg # and row_player_season_avg["foul"] != "" # else "0.0" # ), # "AvgOpponentFoul": ( # row_player_season_avg["foulsOnPlayer"] # if row_player_season_avg # and row_player_season_avg["foulsOnPlayer"] != "" # else "0.0" # ), # "AvgPlusMinus": ( # row_player_season_avg["plusMinus"] # if row_player_season_avg # and row_player_season_avg["plusMinus"] != "" # else "0.0" # ), # "AvgDunk": ( # row_player_season_avg["dunk"] # if row_player_season_avg # and row_player_season_avg["dunk"] != "" # else "0.0" # ), # "AvgKPI": "0.0", # "AvgPlayedTime": ( # row_player_season_avg["playedTime"] # if row_player_season_avg # and row_player_season_avg["playedTime"] != "" # else "0:00" # ), # "Shot1Percent": calc_shot_percent_by_type( # sum_stat, item["stats"], online, shot_types=1 # ), # "Shot2Percent": calc_shot_percent_by_type( # sum_stat, item["stats"], online, shot_types=2 # ), # "Shot3Percent": calc_shot_percent_by_type( # sum_stat, item["stats"], online, shot_types=3 # ), # "Shot23Percent": calc_shot_percent_by_type( # sum_stat, item["stats"], online, shot_types=[2, 3] # ), # "TPoints": sum_stat_with_online( # "points", sum_stat, item["stats"], online # ), # "TShots1": calc_total_shots_str( # sum_stat, item["stats"], online, 1 # ), # "TShots2": calc_total_shots_str( # sum_stat, item["stats"], online, 2 # ), # "TShots3": calc_total_shots_str( # sum_stat, item["stats"], online, 3 # ), # "TShots23": calc_total_shots_str( # sum_stat, item["stats"], online, [2, 3] # ), # "TShot1Percent": calc_shot_percent_by_type( # sum_stat, item["stats"], online, shot_types=1 # ), # "TShot2Percent": calc_shot_percent_by_type( # sum_stat, item["stats"], online, shot_types=2 # ), # "TShot3Percent": calc_shot_percent_by_type( # sum_stat, item["stats"], online, shot_types=3 # ), # "TShot23Percent": calc_shot_percent_by_type( # sum_stat, item["stats"], online, shot_types=[2, 3] # ), # "TAssist": sum_stat_with_online( # "assist", sum_stat, item["stats"], online # ), # "TBlocks": sum_stat_with_online( # "blockShot", sum_stat, item["stats"], online # ), # "TDefRebound": sum_stat_with_online( # "defRebound", sum_stat, item["stats"], online # ), # "TOffRebound": sum_stat_with_online( # "offRebound", sum_stat, item["stats"], online # ), # "TRebound": ( # sum_stat_with_online( # "defRebound", sum_stat, item["stats"], online # ) # + sum_stat_with_online( # "offRebound", sum_stat, item["stats"], online # ) # ), # "TSteal": sum_stat_with_online( # "steal", sum_stat, item["stats"], online # ), # "TTurnover": sum_stat_with_online( # "turnover", sum_stat, item["stats"], online # ), # "TFoul": sum_stat_with_online( # "foul", sum_stat, item["stats"], online # ), # "TOpponentFoul": sum_stat_with_online( # "foulsOnPlayer", sum_stat, item["stats"], online # ), # "TPlusMinus": 0, # "TDunk": sum_stat_with_online( # "dunk", sum_stat, item["stats"], online # ), # "TKPI": 0, # "TPlayedTime": sum_stat["playedTime"] if sum_stat else "0:00", # "TGameCount": ( # safe_int(sum_stat["games"]) # if sum_stat and sum_stat.get("games") != "" # else 0 # ) # + (1 if online else 0), # "TStartCount": ( # safe_int(sum_stat["isStarts"]) # if sum_stat and sum_stat.get("isStarts", 0) != "" # else 0 # ), # "CareerTShots1": calc_total_shots_str( # row_player_career_sum, item["stats"], online, 1 # ), # "CareerTShots2": calc_total_shots_str( # row_player_career_sum, item["stats"], online, 2 # ), # "CareerTShots3": calc_total_shots_str( # row_player_career_sum, item["stats"], online, 3 # ), # "CareerTShots23": calc_total_shots_str( # row_player_career_sum, item["stats"], online, [2, 3] # ), # "CareerTShot1Percent": calc_shot_percent_by_type( # row_player_career_sum, item["stats"], online, 1 # ), # "CareerTShot2Percent": calc_shot_percent_by_type( # row_player_career_sum, item["stats"], online, 2 # ), # "CareerTShot3Percent": calc_shot_percent_by_type( # row_player_career_sum, item["stats"], online, 3 # ), # "CareerTShot23Percent": calc_shot_percent_by_type( # row_player_career_sum, item["stats"], online, [2, 3] # ), # "CareerTPoints": sum_stat_with_online( # "points", row_player_career_sum, item["stats"], online # ), # "CareerTAssist": sum_stat_with_online( # "assist", row_player_career_sum, item["stats"], online # ), # "CareerTBlocks": sum_stat_with_online( # "blockShot", row_player_career_sum, item["stats"], online # ), # "CareerTDefRebound": sum_stat_with_online( # "defRebound", row_player_career_sum, item["stats"], online # ), # "CareerTOffRebound": sum_stat_with_online( # "offRebound", row_player_career_sum, item["stats"], online # ), # "CareerTRebound": ( # sum_stat_with_online( # "defRebound", # row_player_career_sum, # item["stats"], # online, # ) # + sum_stat_with_online( # "offRebound", # row_player_career_sum, # item["stats"], # online, # ) # ), # "CareerTSteal": sum_stat_with_online( # "steal", row_player_career_sum, item["stats"], online # ), # "CareerTTurnover": sum_stat_with_online( # "turnover", row_player_career_sum, item["stats"], online # ), # "CareerTFoul": sum_stat_with_online( # "foul", row_player_career_sum, item["stats"], online # ), # "CareerTOpponentFoul": sum_stat_with_online( # "foulsOnPlayer", # row_player_career_sum, # item["stats"], # online, # ), # "CareerTPlusMinus": 0, # оставить как есть # "CareerTDunk": sum_stat_with_online( # "dunk", row_player_career_sum, item["stats"], online # ), # "CareerTPlayedTime": ( # row_player_career_sum["playedTime"] # if row_player_career_sum # else "0:00" # ), # "CareerTGameCount": sum_stat_with_online( # "games", row_player_career_sum, item["stats"], online # ) # + (1 if online else 0), # "CareerTStartCount": sum_stat_with_online( # "isStarts", row_player_career_sum, item["stats"], online # ), # если нужно, можно +1 при старте # "AvgCarPoints": ( # row_player_career_avg["points"] # if row_player_career_avg # and row_player_career_avg["points"] != "" # else "0.0" # ), # "AvgCarAssist": ( # row_player_career_avg["assist"] # if row_player_career_avg # and row_player_career_avg["assist"] != "" # else "0.0" # ), # "AvgCarBlocks": ( # row_player_career_avg["blockShot"] # if row_player_career_avg # and row_player_career_avg["blockShot"] != "" # else "0.0" # ), # "AvgCarDefRebound": ( # row_player_career_avg["defRebound"] # if row_player_career_avg # and row_player_career_avg["defRebound"] != "" # else "0.0" # ), # "AvgCarOffRebound": ( # row_player_career_avg["offRebound"] # if row_player_career_avg # and row_player_career_avg["offRebound"] != "" # else "0.0" # ), # "AvgCarRebound": ( # row_player_career_avg["rebound"] # if row_player_career_avg # and row_player_career_avg["rebound"] != "" # else "0.0" # ), # "AvgCarSteal": ( # row_player_career_avg["steal"] # if row_player_career_avg # and row_player_career_avg["steal"] != "" # else "0.0" # ), # "AvgCarTurnover": ( # row_player_career_avg["turnover"] # if row_player_career_avg # and row_player_career_avg["turnover"] != "" # else "0.0" # ), # "AvgCarFoul": ( # row_player_career_avg["foul"] # if row_player_career_avg # and row_player_career_avg["foul"] != "" # else "0.0" # ), # "AvgCarOpponentFoul": ( # row_player_career_avg["foulsOnPlayer"] # if row_player_career_avg # and row_player_career_avg["foulsOnPlayer"] != "" # else "0.0" # ), # "AvgCarPlusMinus": ( # row_player_career_avg["plusMinus"] # if row_player_career_avg # and row_player_career_avg["plusMinus"] != "" # else "0.0" # ), # "AvgCarDunk": ( # row_player_career_avg["dunk"] # if row_player_career_avg # and row_player_career_avg["dunk"] != "" # else "0.0" # ), # "AvgCarKPI": "0.0", # "AvgCarPlayedTime": ( # row_player_career_avg["playedTime"] # if row_player_career_avg # and row_player_career_avg["playedTime"] != "" # else "0:00" # ), # "HeadCoachStatsCareer": HeadCoachStatsCareer, # "HeadCoachStatsTeam": HeadCoachStatsTeam, # # "PTS_Career_High": get_carrer_high(item["personId"], "points"), # # "AST_Career_High": get_carrer_high(item["personId"], "assist"), # # "REB_Career_High": get_carrer_high(item["personId"], "rebound"), # # "STL_Career_High": get_carrer_high(item["personId"], "steal"), # # "BLK_Career_High": get_carrer_high(item["personId"], "blockShot"), } team.append(player) count_player = sum(1 for x in team if x["startRole"] == "Player") # print(count_player) if count_player < 12: if team: # Check if team is not empty empty_rows = [ { key: ( False if key in ["captain", "isStart", "isOnCourt"] else ( 0 if key in [ "id", "pts", "weight", "height", "age", "ast", "stl", "blk", "blkVic", "dreb", "oreb", "reb", "to", "foul", "foulT", "foulD", "foulC", "foulB", "fouled", "plusMinus", "dunk", "kpi", ] else "" ) ) for key in team[0].keys() } for _ in range((4 if count_player <= 4 else 12) - count_player) ] team.extend(empty_rows) role_priority = { "Player": 0, "": 1, "Coach": 2, "Team": 3, None: 4, "Other": 5, # на случай неизвестных } # print(team) sorted_team = sorted( team, key=lambda x: role_priority.get( x.get("startRole", 99), 99 ), # 99 — по умолчанию ) out_path = f"{who}" atomic_write_json(sorted_team, out_path) logging.info("Сохранил payload: {out_path}") top_sorted_team = sorted( filter(lambda x: x["startRole"] in ["Player", ""], sorted_team), key=lambda x: ( x["pts"], x["dreb"] + x["oreb"], x["ast"], x["stl"], x["blk"], x["time"], ), reverse=True, ) for item in top_sorted_team: item["pts"] = "" if item["num"] == "" else item["pts"] item["foul"] = "" if item["num"] == "" else item["foul"] out_path = f"top{who.replace('t','T')}" atomic_write_json(top_sorted_team, out_path) logging.info("Сохранил payload: {out_path}") started_team = sorted( filter( lambda x: x["startRole"] == "Player" and x["isOnCourt"] is True, sorted_team, ), key=lambda x: int(x["num"]), reverse=False, ) out_path = f"started_{who}" atomic_write_json(started_team, out_path) logging.info("Сохранил payload: {out_path}") def time_outs_func(data_pbp: list[dict]) -> tuple[str, int, str, int]: """ Вычисляет количество оставшихся таймаутов для обеих команд и формирует строку состояния. Args: data_pbp: Список игровых событий (play-by-play). Returns: Кортеж: (строка команды 1, остаток, строка команды 2, остаток) """ timeout1 = [] timeout2 = [] for event in data_pbp: if event.get("play") == 23: if event.get("startNum") == 1: timeout1.append(event) elif event.get("startNum") == 2: timeout2.append(event) def timeout_status(timeout_list: list[dict], last_event: dict) -> tuple[str, int]: period = last_event.get("period", 0) sec = last_event.get("sec", 0) if period < 3: timeout_max = 2 count = sum(1 for t in timeout_list if t.get("period", 0) <= period) quarter = "1st half" elif period < 5: count = sum(1 for t in timeout_list if 3 <= t.get("period", 0) <= period) quarter = "2nd half" if period == 4 and sec >= 4800 and count in (0, 1): timeout_max = 2 else: timeout_max = 3 else: timeout_max = 1 count = sum(1 for t in timeout_list if t.get("period", 0) == period) quarter = f"OverTime {period - 4}" left = max(0, timeout_max - count) word = "Time-outs" if left != 1 else "Time-out" text = f"{left if left != 0 else 'No'} {word} left in {quarter}" return text, left if not data_pbp: return "", 0, "", 0 last_event = data_pbp[-1] t1_str, t1_left = timeout_status(timeout1, last_event) t2_str, t2_left = timeout_status(timeout2, last_event) return t1_str, t1_left, t2_str, t2_left def add_data_for_teams(new_data: list[dict]) -> tuple[float, list, float]: """ Возвращает усреднённые статистики команды: - средний возраст - очки со старта и скамейки + их доли - средний рост Args: new_data (list[dict]): Список игроков с полями "startRole", "stats", "age", "height" Returns: tuple: (avg_age: float, points: list, avg_height: float) """ players = [item for item in new_data if item.get("startRole") == "Player"] points_start = 0 points_bench = 0 total_age = 0 total_height = 0 player_count = len(players) for player in players: stats = player.get("stats") if stats: is_start = stats.get("isStart") # Очки if is_start is True: points_start += stats.get("points", 0) elif is_start is False: points_bench += stats.get("points", 0) # Возраст и рост total_age += player.get("age", 0) or 0 total_height += player.get("height", 0) or 0 total_points = points_start + points_bench points_start_pro = ( f"{round(points_start * 100 / total_points)}%" if total_points else "0%" ) points_bench_pro = ( f"{round(points_bench * 100 / total_points)}%" if total_points else "0%" ) avg_age = round(total_age / player_count, 1) if player_count else 0 avg_height = round(total_height / player_count, 1) if player_count else 0 points = [points_start, points_start_pro, points_bench, points_bench_pro] return avg_age, points, avg_height def add_new_team_stat( data: dict, avg_age: float, points: float, avg_height: float, timeout_str: str, timeout_left: str, ) -> dict: """ Добавляет в словарь команды форматированную статистику. Все значения приводятся к строкам. Args: data: Исходная статистика команды. avg_age: Средний возраст команды (строка). points: Кортеж из 4 строк: ptsStart, ptsStart_pro, ptsBench, ptsBench_pro. avg_height: Средний рост (в см). timeout_str: Строка отображения таймаутов. timeout_left: Остаток таймаутов. Returns: Обновлённый словарь `data` с новыми ключами. """ def safe_int(v): # Локальная защита от ValueError/TypeError try: return int(v) except (ValueError, TypeError): return 0 def format_percent(goal, shot): goal, shot = safe_int(goal), safe_int(shot) return f"{round(goal * 100 / shot)}%" if shot else "0%" goal1, shot1 = safe_int(data.get("goal1")), safe_int(data.get("shot1")) goal2, shot2 = safe_int(data.get("goal2")), safe_int(data.get("shot2")) goal3, shot3 = safe_int(data.get("goal3")), safe_int(data.get("shot3")) def_reb = safe_int(data.get("defReb")) off_reb = safe_int(data.get("offReb")) data.update( { "pt-1": f"{goal1}/{shot1}", "pt-2": f"{goal2}/{shot2}", "pt-3": f"{goal3}/{shot3}", "fg": f"{goal2 + goal3}/{shot2 + shot3}", "pt-1_pro": format_percent(goal1, shot1), "pt-2_pro": format_percent(goal2, shot2), "pt-3_pro": format_percent(goal3, shot3), "fg_pro": format_percent(goal2 + goal3, shot2 + shot3), "Reb": str(def_reb + off_reb), "avgAge": str(avg_age), "ptsStart": str(points[0]), "ptsStart_pro": str(points[1]), "ptsBench": str(points[2]), "ptsBench_pro": str(points[3]), "avgHeight": f"{avg_height} cm", "timeout_left": str(timeout_left), "timeout_str": str(timeout_str), } ) # Приводим все значения к строкам, если нужно строго для сериализации for k in data: data[k] = str(data[k]) return data stat_name_list = [ ("points", "Очки", "points"), ("pt-1", "Штрафные", "free throws"), ("pt-1_pro", "штрафные, процент", "free throws pro"), ("pt-2", "2-очковые", "2-points"), ("pt-2_pro", "2-очковые, процент", "2-points pro"), ("pt-3", "3-очковые", "3-points"), ("pt-3_pro", "3-очковые, процент", "3-points pro"), ("fg", "очки с игры", "field goals"), ("fg_pro", "Очки с игры, процент", "field goals pro"), ("assist", "Передачи", "assists"), ("pass", "", ""), ("defReb", "подборы в защите", ""), ("offReb", "подборы в нападении", ""), ("Reb", "Подборы", "rebounds"), ("steal", "Перехваты", "steals"), ("block", "Блокшоты", "blocks"), ("blocked", "", ""), ("turnover", "Потери", "turnovers"), ("foul", "Фолы", "fouls"), ("foulsOn", "", ""), ("foulT", "", ""), ("foulD", "", ""), ("foulC", "", ""), ("foulB", "", ""), ("second", "секунды", "seconds"), ("dunk", "данки", "dunks"), ("fastBreak", "", "fast breaks"), ("plusMinus", "+/-", "+/-"), ("avgAge", "", "avg Age"), ("ptsBench", "", "Bench PTS"), ("ptsBench_pro", "", "Bench PTS, %"), ("ptsStart", "", "Start PTS"), ("ptsStart_pro", "", "Start PTS, %"), ("avgHeight", "", "avg height"), ("timeout_left", "", "timeout left"), ("timeout_str", "", "timeout str"), ] def Team_Both_Stat(merged: dict, *, out_dir: str = "static") -> None: """ Обновляет файл team_stats.json, содержащий сравнение двух команд. Аргументы: stop_event (threading.Event): Событие для остановки цикла. """ logger.info("START making json for team statistics") try: teams = merged["result"]["teams"] plays = merged["result"].get("plays", []) # Разделение команд team_1 = next((t for t in teams if t["teamNumber"] == 1), None) team_2 = next((t for t in teams if t["teamNumber"] == 2), None) if not team_1 or not team_2: logger.warning("Не найдены обе команды в данных") # time.sleep() # Таймауты timeout_str1, timeout_left1, timeout_str2, timeout_left2 = time_outs_func(plays) # Возраст, очки, рост avg_age_1, points_1, avg_height_1 = add_data_for_teams(team_1.get("starts", [])) avg_age_2, points_2, avg_height_2 = add_data_for_teams(team_2.get("starts", [])) if not team_1.get("total") or not team_2.get("total"): logger.debug("Нет total у команд — пропускаю перезапись team_stats.json") # Форматирование общей статистики (как и было) total_1 = add_new_team_stat( team_1["total"], avg_age_1, points_1, avg_height_1, timeout_str1, timeout_left1, ) total_2 = add_new_team_stat( team_2["total"], avg_age_2, points_2, avg_height_2, timeout_str2, timeout_left2, ) # Финальный JSON result_json = [] for key in total_1: val1 = ( int(total_1[key]) if isinstance(total_1[key], float) else total_1[key] ) val2 = ( int(total_2[key]) if isinstance(total_2[key], float) else total_2[key] ) stat_rus, stat_eng = "", "" for s in stat_name_list: if s[0] == key: stat_rus, stat_eng = s[1], s[2] break result_json.append( { "name": key, "nameGFX_rus": stat_rus, "nameGFX_eng": stat_eng, "val1": val1, "val2": val2, } ) out_path = "team_stats" atomic_write_json(result_json, out_path) logging.info("Сохранил payload: {out_path}") logger.debug("Успешно записаны данные в team_stats.json") except Exception as e: logger.error(f"Ошибка при обработке командной статистики: {e}", exc_info=True) def render_loop(stop_event: threading.Event, out_name: str = "game"): """ Крутится в отдельном потоке. Читает api_*.json, собирает финальный state и сохраняет в static/.json. Работает, пока stop_event не установлен. """ logger.info("[RENDER_THREAD] start render loop") while not stop_event.is_set(): try: state = build_render_state() Team_Both_Stat(state) Json_Team_Generation(state, who="team1") Json_Team_Generation(state, who="team2") atomic_write_json([state["result"]["live_status"]], "live_status") atomic_write_json(state["result"], out_name) except Exception as ex: logger.exception(f"[RENDER_THREAD] error while building render state: {ex}") time.sleep(0.2) logger.info("[RENDER_THREAD] stop render loop") def main(): parser = argparse.ArgumentParser() parser.add_argument("--league", default="vtb") parser.add_argument("--team", required=True) parser.add_argument("--lang", default="en") args = parser.parse_args() session = create_session() # единый флаг остановки на ВСЮ программу stop_event = threading.Event() try: get_data_API(session, args.league, args.team, args.lang, stop_event) except KeyboardInterrupt: # ручное прерывание: просим все рабочие циклы сворачиваться logger.info("KeyboardInterrupt: stopping...") stop_event.set() except Exception as e: logger.exception(f"Fatal in main(): {e}") stop_event.set() if __name__ == "__main__": main()