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( league: str, season: str, game_id: int, lang: str, game_meta: dict, stop_event: threading.Event, ): """ Запускает два рабочих цикла: - poll_game_live (опрашивает API матча) - render_loop (собирает ui_state.json) Управляет их остановкой. """ logger.info( f"[LIVE_THREAD] start live loop for game_id={game_id} (league={league}, season={season})" ) # отдельная сессия только для лайва session = create_session() # поток рендера render_thread = threading.Thread( target=render_loop, args=(stop_event,), # дефолт out_name="game" в твоём render_loop можно оставить daemon=False, ) render_thread.start() logger.info("[LIVE_THREAD] render thread spawned") try: poll_game_live( session=session, league=league, season=season, game_id=game_id, lang=lang, game_meta=game_meta, stop_event=stop_event, ) 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] stopping render thread for game_id={game_id}") render_thread.join() logger.info(f"[LIVE_THREAD] stop live loop for game_id={game_id}") def Scores_Quarter(merged: dict, *, out_dir: str = "static") -> None: """ Поток, обновляющий JSON со счётом по четвертям. """ logger.info("START making json for scores quarter") quarters = ["Q1", "Q2", "Q3", "Q4", "OT1", "OT2", "OT3", "OT4"] score_by_quarter = [{"Q": q, "score1": "", "score2": ""} for q in quarters] try: # Сначала пробуем fullScore full_score_str = merged.get("result", {}).get("game", {}).get("fullScore", "") if full_score_str: full_score_list = full_score_str.split(",") for i, score_str in enumerate(full_score_list[: len(score_by_quarter)]): parts = score_str.split(":") if len(parts) == 2: score_by_quarter[i]["score1"] = parts[0] score_by_quarter[i]["score2"] = parts[1] logger.info("Счёт по четвертям получен из fullScore.") # Если нет fullScore, пробуем scoreByPeriods elif "scoreByPeriods" in merged.get("result", {}): periods = merged["result"]["scoreByPeriods"] for i, score in enumerate(periods[: len(score_by_quarter)]): score_by_quarter[i]["score1"] = str(score.get("score1", "")) score_by_quarter[i]["score2"] = str(score.get("score2", "")) logger.info("Счёт по четвертям получен из scoreByPeriods.") else: logger.debug("Нет данных по счёту, сохраняем пустые значения.") out_path = "scores" atomic_write_json(score_by_quarter, out_path) logging.info("Сохранил payload: {out_path}") except Exception as e: logger.error(f"Ошибка в Scores_Quarter: {e}", exc_info=True) def poll_game_live( session, league: str, season: str, game_id: int, lang: str, game_meta: dict, stop_event: threading.Event, ): slow_endpoints = ["game"] # "pregame-fullstats" можно добавить обратно fast_endpoints = ["live-status", "box-score", "play-by-play"] last_call = {name: 0 for name in slow_endpoints + fast_endpoints} with ThreadPoolExecutor(max_workers=5) as executor: while True: # внешний стоп с клавиатуры / по команде if stop_event.is_set(): logger.info(f"[POLL] 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", "result"): 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}") 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) # ещё одна точка выхода даже если статус не изменился if stop_event.is_set(): logger.info(f"[POLL] stop_event set after sleep -> break live poll for game {game_id}") break def get_data_API(session, league: str, team: str, lang: str, stop_event: threading.Event): 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"] 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) if is_game_live(today_game["game"]): t = threading.Thread( target=run_live_loop, args=(league, season, game_id, lang, today_game["game"], stop_event), daemon=False, ) t.start() logger.info("live thread spawned, waiting for it to finish...") try: t.join() except KeyboardInterrupt: logger.info("KeyboardInterrupt while waiting live thread -> stop_event") stop_event.set() t.join() logger.info("live 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") Scores_Quarter(state) 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()