from __future__ import annotations
import os
import sys
import json
import time
import socket
import urllib3
import logging
import logging.config
import argparse
import platform
import requests
import threading
import numpy as np
import pandas as pd
from threading import Event, Lock
from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict, Any
from datetime import datetime, timedelta, timezone
from ipaddress import ip_address as _ip_parse
import errno
urllib3.disable_warnings()
# === Настройки и логгер ===
myhost = platform.node()
VERSION = "v.2.1 от 08.10.2025"
TAGS = [
{"tag": "vtb", "name": "Единая Лига ВТБ", "lang": "en"},
{"tag": "vtbyouth", "name": "Молодежка ВТБ", "lang": ""},
{"tag": "rfb-deti", "name": "Дети", "lang": ""},
{"tag": "rfb-silent", "name": "Тихий!баскетбол", "lang": ""},
{"tag": "orgRoot", "name": "Все соревнования", "lang": ""},
{"tag": "vtb-supercup", "name": "Супер-Кубок ЕЛ ВТБ", "lang": ""},
{"tag": "uba-leto", "name": "UBA лето", "lang": "ru"},
{"tag": "phygital-russia-cup-m", "name": "Фиджитал кубок", "lang": "ru"},
{"tag": "phygital", "name": "Фиджитал", "lang": "ru"},
{"tag": "3x3Root", "name": "3х3", "lang": "ru"},
{"tag": "LS3x3", "name": "Лига Сильных 3х3", "lang": "ru"},
{"tag": "uba", "name": "ЮБА", "lang": "ru"},
{"tag": "uba-main", "name": "ЮБА-маин", "lang": "ru"},
{"tag": "msl", "name": "Суперлига. Мужчины", "lang": ""},
{"tag": "mhl", "name": "Высшая лига. Мужчины", "lang": ""},
{"tag": "mcup", "name": "Кубок России. Мужчины", "lang": ""},
{"tag": "wpremier", "name": "Премьер-Лига. Женщины", "lang": ""},
{"tag": "wsl", "name": "Суперлига. Женщины", "lang": ""},
{"tag": "whl", "name": "Высшая лига. Женщины", "lang": ""},
{"tag": "wcup", "name": "Кубок России. Женщины", "lang": ""},
{"tag": "uba-summer", "name": "UBA Leto", "lang": "ru"},
{"tag": "unics", "name": "UNICS", "lang": "ru"},
]
TOKEN = "7639240596:AAH0YtdQoWZSC-_R_EW4wKAHHNLIA0F_ARY"
GROUP_CHAT = 228977654
if not os.path.exists("logs"):
os.makedirs("logs")
if not os.path.exists("JSON"):
os.makedirs("JSON")
LOG_CONFIG = {
"version": 1,
"handlers": {
"telegram": {
"class": "telegram_handler.TelegramHandler",
"level": "INFO",
"token": TOKEN,
"chat_id": GROUP_CHAT,
"formatter": "telegram",
},
"console": {
"class": "logging.StreamHandler",
"level": "WARNING",
"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": "%(levelname)s %(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
# Глобальный кэш и лок для потокобезопасности
_GAME_CACHE: dict[tuple[int, str], dict] = {}
_GAME_CACHE_LOCK = Lock()
THROTTLE_OLD_MINUTES = 30
FOLDER_JSON = "JSON" if sys.platform.startswith("win") else "static"
def get_ip_address():
try:
# Попытка получить IP-адрес с использованием внешнего сервиса
# Может потребоваться подключение к интернету
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
s.connect(("8.8.8.8", 80))
ip_address = s.getsockname()[0]
except socket.error:
# Если не удалось получить IP-адрес через внешний сервис,
# используем метод для локального получения IP
ip_address = socket.gethostbyname(socket.gethostname())
return ip_address
def read_match_id_json(path="match_id.json", attempts=10, delay=0.2):
"""Надёжное чтение match_id.json с ретраями при EBUSY/битом JSON."""
d = delay
for i in range(attempts):
try:
if not os.path.isfile(path):
return {}
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except json.JSONDecodeError:
# файл переписывают — подождём и попробуем снова
time.sleep(d); d = min(d*1.6, 2.0)
except OSError as e:
# EBUSY (errno=16) — подождём и ещё раз
if getattr(e, "errno", None) == errno.EBUSY:
time.sleep(d); d = min(d*1.6, 2.0)
continue
# иные ошибки — пробрасываем дальше
raise
logger.error("Не удалось прочитать match_id.json после нескольких попыток; возвращаю {}")
return {}
# === Аргументы командной строки ===
parser = argparse.ArgumentParser(description="VTB Data Fetcher")
parser.add_argument("--league", type=str, default="vtb", help="League tag")
parser.add_argument("--lang", type=str, default="en", help="Language")
# parser.add_argument("--team", type=str, required=True, help="Team name")
# parser.add_argument("--region", type=int, default=0, help="for tvstart.ru")
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument("--team", type=str, help="Team name")
group.add_argument("--region", action="store_true", help="for tvstart.ru")
args = parser.parse_args()
LEAGUE = args.league
LANG = args.lang
if args.team:
TEAM = args.team
else: # значит указан --region
ip_check = read_match_id_json("match_id.json") or {}
ip_address = get_ip_address()
TEAM = ip_check.get(ip_address, {}).get("team")
if TEAM is None:
parser.error("Не удалось определить команду по IP. Укажите --team явно.")
# === Глобальные настройки ===
TIMEOUT_ONLINE = 1
FETCH_INTERVAL = 2
TIMEOUT_DATA_OFF = 60
game_online_data = None
game_online_lock = threading.Lock()
game_status_data = None
game_status_lock = threading.Lock()
def _is_local_ip(ip: str) -> bool:
"""True, если IP локальный/loopback/link-local или не распарсился."""
try:
ipobj = _ip_parse(ip)
return ipobj.is_private or ipobj.is_loopback or ipobj.is_link_local
except Exception:
# Если get_ip_address() вернул что-то странное — считаем локальным, чтобы не префиксовать.
return True
def _ipcheck() -> str:
"""Возвращает префикс для имени файла по IP.
Если IP локальный или host не найден — возвращает пустую строку.
"""
try:
ip_str = get_ip_address()
except Exception:
ip_str = None
ip_map = globals().get("ip_check") or {}
if ip_str and isinstance(ip_map, dict):
host = (ip_map.get(ip_str) or {}).get("host")
if host:
return f"{host}_"
return ""
def rewrite_file(filename: str, data: dict, directory: str = "JSON") -> None:
"""
Перезаписывает JSON-файл с заданными данными.
Если запуск локальный (локальный IP), префикс host в имени файла не используется.
Если IP не локальный и есть словарь ip_check с хостами — добавим префикс host_.
"""
# Если глобальная константа задана — используем её
try:
directory = FOLDER_JSON # type: ignore[name-defined]
except NameError:
# иначе используем аргумент по умолчанию/переданный
pass
os.makedirs(directory, exist_ok=True)
host_prefix = _ipcheck()
filepath = os.path.join(directory, f"{host_prefix}{filename}.json")
# print(filepath) # оставил как у тебя; можно заменить на logger.debug при желании
try:
with open(filepath, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
logger.info(f"Файл {filepath} перезаписан.")
except Exception as e:
logger.error(f"Ошибка при записи файла {filepath}: {e}")
def get_json(url: str, timeout: int = 10, verify_ssl: bool = True) -> dict | None:
"""
Получает JSON-ответ по указанному URL.
Args:
url (str): URL для запроса.
timeout (int): Таймаут запроса в секундах.
verify_ssl (bool): Проверять ли SSL-сертификат.
Returns:
dict | None: Распарсенный JSON или None при ошибке.
"""
try:
logger.info(f"Пытаюсь получить данные с {url}")
response = requests.get(url, timeout=timeout, verify=verify_ssl)
response.raise_for_status() # выбросит исключение, если статус != 2xx
return response.json()
except requests.RequestException as e:
logger.warning(f"Ошибка при запросе: {url}\n{e}")
return None
except ValueError as e:
logger.warning(f"Некорректный JSON-ответ от {url}:\n{e}")
return None
def Game_Online2(game_id: int) -> dict | None:
"""
Получает и объединяет данные об игре, включая онлайн-информацию.
Использует глобальные переменные: URL и LANG.
Args:
game_id (int): ID матча.
Returns:
dict | None: Объект игры или None при ошибке.
"""
global URL, LANG
def build_url(endpoint: str) -> str:
return f"{URL}api/abc/games/{endpoint}?Id={game_id}&Lang={LANG}"
# 1. Получаем box score
box_score = get_json(build_url("box-score"))
if not box_score or box_score.get("status") != "Ok":
# Получаем данные старого матча
game = get_json(build_url("game"))
if game:
logger.info("У нас получилось получить данные со старого матча")
else:
logger.warning(
f"Не удалось получить данные старого матча: game_id={game_id}"
)
return game
# 2. Получаем онлайн-данные
game = get_json(build_url("game"))
play_by_play = get_json(build_url("play-by-play"))
if not game or not play_by_play:
logger.warning(f"Ошибка при получении онлайн-данных для матча {game_id}")
return None
try:
# 3. Совмещаем статистику
for index_team, team in enumerate(game["result"]["teams"][1:]):
box_team = box_score["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", {})
# 4. Добавляем дополнительные данные
game["result"]["plays"] = play_by_play.get("result", [])
game["result"]["scoreByPeriods"] = box_score["result"].get("scoreByPeriods", [])
game["result"]["fullScore"] = box_score["result"].get("fullScore", {})
logger.info("Склеил данные по онлайн матчу")
return game
except Exception as e:
logger.error(f"Ошибка при обработке данных игры {game_id}: {e}")
return None
def Game_Online(game_id: int) -> dict | None:
"""
Получает и объединяет данные об игре, включая онлайн-информацию,
с троттлингом обновления "старых" матчей (не чаще, чем раз в 30 минут).
Использует глобальные переменные: URL и LANG.
Args:
game_id (int): ID матча.
Returns:
dict | None: Объект игры или None при ошибке.
"""
global URL, LANG, _GAME_CACHE, _GAME_CACHE_LOCK
now = datetime.now(timezone.utc)
cache_key = (game_id, str(LANG))
# Время троттлинга для старых матчей
OLD_GAME_THROTTLE_MINUTES = 10
# 0. Предварительная проверка: если недавно получали "старый" матч — отдаем кэш
with _GAME_CACHE_LOCK:
cached = _GAME_CACHE.get(cache_key)
if (
cached
and cached.get("mode") == "old"
and (now - cached.get("ts", now)) < timedelta(minutes=OLD_GAME_THROTTLE_MINUTES)
):
return cached.get("data")
def build_url(endpoint: str) -> str:
return f"{URL}api/abc/games/{endpoint}?Id={game_id}&Lang={LANG}"
box_score = get_json(build_url("box-score"))
print(box_score)
if not box_score or box_score.get("status") != "Ok":
# Проверим — матч сейчас online?
live = get_json(f"{URL}api/abc/games/live-status?id={game_id}")
is_online = bool(
live and live.get("status") == "Ok"
and live.get("result", {}).get("gameStatus") == "Online"
)
game = get_json(build_url("game"))
if game:
logger.info("У нас получилось получить данные со старого матча")
# Только если матч не в онлайн-режиме — кладём в кэш как 'old'
if not is_online:
with _GAME_CACHE_LOCK:
_GAME_CACHE[cache_key] = {"mode": "old", "ts": now, "data": game}
else:
logger.warning(f"Не удалось получить данные старого матча: game_id={game_id}")
return game
# 1. Получаем box score
# box_score = get_json(build_url("box-score"))
# if not box_score or box_score.get("status") != "Ok":
# # Получаем данные старого матча (и кэшируем их с режимом 'old')
# live = get_json(f"{URL}api/abc/games/live-status?id={game_id}")
# is_online = bool(live and live.get("status") == "Ok" and live.get("result", {}).get("gameStatus") == "Online")
# game = get_json(build_url("game"))
# if game:
# logger.debug("У нас получилось получить данные со старого матча")
# if not is_online:
# with _GAME_CACHE_LOCK:
# _GAME_CACHE[cache_key] = {"mode": "old", "ts": now, "data": game}
# else:
# logger.warning(f"Не удалось получить данные старого матча: game_id={game_id}")
# return game
# game = get_json(build_url("game"))
# if game:
# logger.debug("У нас получилось получить данные со старого матча")
# with _GAME_CACHE_LOCK:
# _GAME_CACHE[cache_key] = {"mode": "old", "ts": now, "data": game}
# else:
# logger.warning(
# f"Не удалось получить данные старого матча: game_id={game_id}"
# )
# # Даже неудачный ответ имеет смысл закэшировать как 'old' на короткое время,
# # но чтобы не скрыть будущий успех, кэшируем только при наличии данных.
# return game
# 2. Получаем онлайн-данные
game = get_json(build_url("game"))
play_by_play = get_json(build_url("play-by-play"))
if not game or not play_by_play:
logger.warning(f"Ошибка при получении онлайн-данных для матча {game_id}")
return None
try:
# 3. Совмещаем статистику
for index_team, team in enumerate(game["result"]["teams"][1:]):
box_team = box_score["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", {})
# 4. Добавляем дополнительные данные
game["result"]["plays"] = play_by_play.get("result", [])
game["result"]["scoreByPeriods"] = box_score["result"].get("scoreByPeriods", [])
game["result"]["fullScore"] = box_score["result"].get("fullScore", {})
logger.info("Склеил данные по онлайн матчу")
# Обновляем кэш и снимаем режим 'old'
with _GAME_CACHE_LOCK:
_GAME_CACHE[cache_key] = {"mode": "online", "ts": now, "data": game}
return game
except Exception as e:
logger.error(f"Ошибка при обработке данных игры {game_id}: {e}")
return None
def game_online_loop(game_id: int, stop_event: threading.Event) -> None:
"""
Цикл обновления онлайн-данных игры с заданным интервалом.
Args:
game_id (int): ID игры.
stop_event (threading.Event): Событие для остановки цикла.
"""
global game_online_data
while not stop_event.is_set():
try:
data = Game_Online(game_id)
if data:
with game_online_lock:
game_online_data = data
else:
logger.warning(f"Game_Online вернул None для game_id={game_id}")
except Exception as e:
logger.error(f"Ошибка в game_online_loop: {e}", exc_info=True)
stop_event.wait(TIMEOUT_ONLINE) # Лучше чем time.sleep — можно остановить сразу
def coach_team_stat(data: list[dict], team_id: int) -> dict:
"""
Считает статистику тренера по матчам и сезонам для конкретной команды.
Args:
data (list[dict]): Список сезонов с данными.
team_id (int): ID команды.
Returns:
dict: Статистика тренера.
"""
total_games = total_wins = total_loses = seasons = 0
for d in data or []:
team = d.get("team")
if team and team.get("id") == team_id:
seasons += 1
total_games += d.get("games", 0)
total_wins += d.get("wins", 0)
total_loses += d.get("loses", 0)
return {
"games": total_games,
"wins": total_wins,
"loses": total_loses,
"gamesAsCoach": total_games,
"winsAsCoach": total_wins,
"losesAsCoach": total_loses,
"season": seasons,
}
def safe_int(value: any, default: int = 0) -> int:
"""
Безопасное преобразование значения в int. Возвращает default при ошибке.
Args:
value: Значение для преобразования (строка, число и т.п.).
default: Значение по умолчанию при ошибке.
Returns:
Целое число или default.
"""
try:
if isinstance(value, str):
value = value.strip()
return int(float(value))
except (ValueError, TypeError):
return default
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 safe_percent(numerator: float | int | str, denominator: float | int | str) -> str:
"""
Безопасно вычисляет процент (numerator / denominator * 100), округляя до 1 знака.
Args:
numerator: Числитель (может быть числом или строкой).
denominator: Знаменатель (может быть числом или строкой).
Returns:
Строка с процентом, например "75.0%" или "0.0%" при ошибке.
"""
try:
num = float(str(numerator).strip())
den = float(str(denominator).strip())
if den == 0:
return "0.0%"
return f"{round(num * 100 / den, 1)}%"
except (ValueError, TypeError):
return "0.0%"
def calc_shot_percent_by_type(
sum_stat: dict | None,
item_stats: dict,
online: bool,
shot_types: int | list[int] = 1,
digits: int = 1,
empty: str = "0.0%",
) -> str:
"""
Возвращает процент реализации бросков по одному или нескольким типам (goalX / shotX).
Args:
sum_stat: Сезонная статистика игрока или None.
item_stats: Онлайн-статистика (например, item["stats"]).
online: Учитывать ли онлайн-данные.
shot_types: Целое число или список номеров бросков (1, 2, 3, или [2, 3]).
digits: Кол-во знаков после запятой.
empty: Что возвращать при отсутствии данных или делении на 0.
Returns:
Строка процента, например '68.5%' или '0.0%'.
"""
if isinstance(shot_types, int):
shot_types = [shot_types]
if not sum_stat or item_stats is None:
return empty
total_goal = 0
total_shot = 0
# print(item_stats)
for t in shot_types:
# print(t)
goal_key = f"goal{t}"
shot_key = f"shot{t}"
sum_goal_raw = sum_stat.get(goal_key)
sum_shot_raw = sum_stat.get(shot_key)
item_goal_raw = item_stats.get(goal_key)
item_shot_raw = item_stats.get(shot_key)
# Если какие-либо данные отсутствуют, или являются "", считаем 0
sum_goal = safe_int(sum_goal_raw if sum_goal_raw != "" else 0)
sum_shot = safe_int(sum_shot_raw if sum_shot_raw != "" else 0)
item_goal = safe_int(item_goal_raw if item_goal_raw != "" else 0)
item_shot = safe_int(item_shot_raw if item_shot_raw != "" else 0)
total_goal += sum_goal + (item_goal if online else 0)
total_shot += sum_shot + (item_shot if online else 0)
if total_shot == 0:
return empty
percent = round(total_goal * 100 / total_shot, digits)
return f"{percent}%"
def calc_total_shots_str(
sum_stat: dict | None,
item_stats: dict,
online: bool,
shot_types: int | list[int],
empty: str = "0/0",
) -> str:
"""
Формирует строку вида 'goalX/shotX' или сумму по нескольким типам бросков.
Args:
sum_stat: Сезонная статистика игрока (может быть None).
item_stats: Онлайн-статистика из item["stats"].
online: Добавлять ли онлайн-значения.
shot_types: Один номер броска (int) или список (например, [2, 3] для TShots23).
empty: Что возвращать при отсутствии данных.
Returns:
Строка вида '5/8' или '0/0' при ошибке/отсутствии данных.
"""
if isinstance(shot_types, int):
shot_types = [shot_types]
if not sum_stat or item_stats is None:
return empty
total_goal = 0
total_shot = 0
for t in shot_types:
goal_key = f"goal{t}"
shot_key = f"shot{t}"
if sum_stat.get(shot_key) == "" or item_stats.get(shot_key) == "":
return empty
goal = safe_int(sum_stat.get(goal_key))
shot = safe_int(sum_stat.get(shot_key))
if online:
goal += safe_int(item_stats.get(goal_key) or 0)
shot += safe_int(item_stats.get(shot_key) or 0)
total_goal += goal
total_shot += shot
return f"{total_goal}/{total_shot}"
def sum_stat_with_online(
stat_name: str, base_stat: dict | None, online_stat: dict | None, online: bool
) -> int:
base = safe_int((base_stat or {}).get(stat_name))
online_val = safe_int((online_stat or {}).get(stat_name)) if online else 0
return base + online_val
def Json_Team_Generation(who, data, stop_event):
logger.info(f"START making json for {data[who]}, {data[f'{who}_id']}")
global game_online_data
initialized = False
while not stop_event.is_set():
try:
with game_online_lock:
game_online_data_copy = game_online_data
if game_online_data_copy is not None:
if who == "team1":
logger.debug(
f"send {game_online_data_copy['result']['team1']['name']}"
)
for i in game_online_data_copy["result"]["teams"]:
if i["teamNumber"] == 1:
payload = i
elif who == "team2":
logger.debug(
f"send {game_online_data_copy['result']['team2']['name']}"
)
for i in game_online_data_copy["result"]["teams"]:
if i["teamNumber"] == 2:
payload = i
# print(payload)
# получаю ID игроков
url = f"{URL}api/abc/games/live-status?id={data['game_id']}"
json_live_status = get_json(url)
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 = True
# получаю статистику на игроков по сезону и карьере
if not initialized:
player_ids = [
i["personId"]
for i in payload["starts"]
if i["startRole"] == "Player"
]
print(player_ids)
coach_ids = [
i["personId"]
for i in payload["starts"]
if i["startRole"] == "Coach" and i["personId"] is not None
]
print(coach_ids)
player_season_stat = []
player_career_stat = []
coach_stat = []
with ThreadPoolExecutor() as pool:
player_season_stat_temp = [
pool.submit(Player_Stat_Season, player_id, data["season"])
for player_id in player_ids
]
player_career_stat_temp = [
pool.submit(Player_Stat_Career, player_id)
for player_id in player_ids
]
coach_stat_temp = [
pool.submit(
Coach_Stat, coach_id, data["season"], data[f"{who}_id"]
)
for coach_id in coach_ids
]
player_futures = [pool.submit(Player_all_game, pid) for pid in player_ids]
all_players_games = []
for fut in as_completed(player_futures):
try:
all_players_games.append(fut.result())
except Exception as e:
logger.exception(f"Ошибка при обработке игрока: {e}")
player_season_stat += [
res.result() for res in player_season_stat_temp
]
player_career_stat += [
res.result() for res in player_career_stat_temp
]
coach_stat += [res.result() for res in coach_stat_temp]
initialized = True
# print(coach_stat)
# while not stop_event.is_set():
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"]
HeadCoachStatsCareer, HeadCoachStatsTeam = "", ""
team = []
# print(starts)
for item in starts:
# print(player_season_stat)
row_player_season = next(
(
v
for row in player_season_stat if row
for k, v in row.items()
if k == item["personId"]
),
None,
)
row_player_career = next(
(
v
for row in player_career_stat if row
for k, v in row.items()
if k == item["personId"]
),
None,
)
# print(item)
row_coach_stat = next(
(
v
for row in coach_stat if row
for k, v in row.items()
if k == item["personId"]
),
None,
)
row_player_season_avg, sum_stat = (
(
next(
(
r["stats"]
for r in row_player_season
if r["class"] == "Avg"
),
None,
),
next(
(
r["stats"]
for r in row_player_season
if r["class"] == "Sum"
),
None,
),
)
if row_player_season
else (None, None)
)
row_player_career_avg, row_player_career_sum = (
(
next(
(
r["stats"]
for r in row_player_career
if r["class"] == "Avg"
),
None,
),
next(
(
r["stats"]
for r in row_player_career
if r["class"] == "Sum"
),
None,
),
)
if row_player_career
else (None, None)
)
if row_coach_stat:
games_word = (
"game" if row_coach_stat[-1]["games"] in [0, 1] else "games"
)
total_season = len(row_coach_stat) - 1
season_word = "season" if total_season == 1 else "seasons"
procent = (
round(
(row_coach_stat[-1]["wins"] * 100)
/ row_coach_stat[-1]["games"],
1,
)
if row_coach_stat[-1]["games"] != 0
else ""
)
# HeadCoachStatsCareer = (
# f'{row_s["total_seasons"]} {season_word} in The VTB United League career'
# )
HeadCoachStatsCareer = "in The VTB United League career"
if total_season == 0:
HeadCoachStatsCareer = f'{row_coach_stat[-1]["games"] if row_coach_stat[-1]["games"] != 0 else "first"} {games_word} as {data[who]} head coach'
HeadCoachStatsCareer += (
f'\n{row_coach_stat[-1]["games"] if row_coach_stat[-1]["games"] != 0 else "first"} {games_word}: {row_coach_stat[-1]["wins"]}-{row_coach_stat[-1]["loses"]} ({procent}% wins)'
if row_coach_stat[-1]["games"] != 0
else ""
)
coach_team_stat_temp = coach_team_stat(
row_coach_stat, data[f"{who}_id"]
)
games_word_team = (
"game"
if coach_team_stat_temp["games"] in [0, 1]
else "games"
)
season_word_team = (
"season"
if coach_team_stat_temp["season"] == 1
else "seasons"
)
procent_team = (
round(
(coach_team_stat_temp["wins"] * 100)
/ coach_team_stat_temp["games"],
1,
)
if coach_team_stat_temp["games"] != 0
else ""
)
# HeadCoachStatsTeam = (
# f'{coach_team_stat_temp["season"]} {season_word_team} as {data[f"{who}"]} head coach'
# )
HeadCoachStatsTeam = f"{data[f'{who}']} head coach"
if coach_team_stat_temp["season"] == 0:
HeadCoachStatsTeam = f'{coach_team_stat_temp["games"] if coach_team_stat_temp["games"] != 0 else "first"} {games_word} as {data[f"{who}"]} head coach'
HeadCoachStatsTeam += (
f'\n{coach_team_stat_temp["games"] if coach_team_stat_temp["games"] != 0 else "first"} {games_word}: {coach_team_stat_temp["wins"]}-{coach_team_stat_temp["loses"]} ({procent_team}% wins)'
if coach_team_stat_temp["games"] != 0
else ""
)
text = ""
if row_player_season_avg:
if LANG == "en":
text = f"GAMES: {row_player_season_avg['games']} MINUTES: {row_player_season_avg['playedTime']} "
text += (
f"PTS: {row_player_season_avg['points']} "
if row_player_season_avg["points"] != ""
else ""
)
text += (
f"REB: {row_player_season_avg['rebound']} "
if row_player_season_avg["rebound"] != ""
and float(row_player_season_avg["rebound"]) >= 1.0
else ""
)
text += (
f"AST: {row_player_season_avg['assist']} "
if row_player_season_avg["assist"] != ""
and float(row_player_season_avg["assist"]) >= 1.0
else ""
)
text += (
f"STL: {row_player_season_avg['steal']} "
if row_player_season_avg["steal"] != ""
and float(row_player_season_avg["steal"]) >= 1.0
else ""
)
text += (
f"BLK: {row_player_season_avg['blockShot']} "
if row_player_season_avg["blockShot"] != ""
and float(row_player_season_avg["blockShot"]) >= 1.0
else ""
)
else:
text = f"ИГРЫ: {row_player_season_avg['games']} ВРЕМЯ: {row_player_season_avg['playedTime']} "
text += (
f"ОЧКИ: {row_player_season_avg['points']} "
if row_player_season_avg["points"] != ""
else ""
)
text += (
f"ПОДБОРЫ: {row_player_season_avg['rebound']} "
if row_player_season_avg["rebound"] != ""
and float(row_player_season_avg["rebound"]) >= 1.0
else ""
)
text += (
f"ПЕРЕДАЧИ: {row_player_season_avg['assist']} "
if row_player_season_avg["assist"] != ""
and float(row_player_season_avg["assist"]) >= 1.0
else ""
)
text += (
f"ПЕРЕХВАТЫ: {row_player_season_avg['steal']} "
if row_player_season_avg["steal"] != ""
and float(row_player_season_avg["steal"]) >= 1.0
else ""
)
text += (
f"БЛОКШОТЫ: {row_player_season_avg['blockShot']} "
if row_player_season_avg["blockShot"] != ""
and float(row_player_season_avg["blockShot"]) >= 1.0
else ""
)
text = text.strip()
# print(item["personId"] if item["startRole"] != "Team" else , item["startRole"])
# if not item["personId"]:
# item["personId"] = data[f'{who}_id']
# print(item)
# item["personId"] = item["personId"] if item["personId"] else data[f'{who}_id']
# print(item)
# print(data[f'{who}_id'], type(data[f'{who}_id']), data)
# print(item)
player = {
"id": (
item["personId"]
if item["personId"]
else int(data[f"{who}_id"])
),
"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", 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 при старте
"HeadCoachStatsCareer": HeadCoachStatsCareer,
"HeadCoachStatsTeam": HeadCoachStatsTeam,
}
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 — по умолчанию
)
rewrite_file(who, sorted_team)
rewrite_file(f"{who}_copy", sorted_team)
# print(sorted_team)
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"]
rewrite_file(f"top{who.replace('t','T')}", top_sorted_team)
started_team = sorted(
filter(
lambda x: x["startRole"] == "Player" and x["isOnCourt"] is True,
sorted_team,
),
key=lambda x: int(x["num"]),
reverse=False,
)
rewrite_file(f"started_{who}", started_team)
time.sleep(TIMEOUT_ONLINE)
else:
print(f"{who} НЕ ПОЛУЧАЕТСЯ ПРОЧИТАТЬ")
except Exception as e:
print(f"[{who}] Ошибка: {e}, {e.with_traceback()}")
time.sleep(TIMEOUT_ONLINE)
def Player_Stat_Season(player_id: str, season: str) -> dict:
url = f"{URL}api/abc/players/stats?teamId=0&Tag={LEAGUE}&season={season}&Id={player_id}"
player_stat_season = get_json(url)
if not player_stat_season:
logger.error(f"Пустой ответ от API для игрока {player_id} за сезон {season}")
return {player_id: default_player_stats_season()}
items = player_stat_season.get("items")
if items:
logger.info(
f"Данные за сезон {season} для игрока {player_id} успешно получены."
)
return {player_id: items[-2:]} # последние две записи: Sum и Avg
logger.warning(
f"Нет данных на игрока {player_id} за сезон {season}. Вероятно, еще не играл."
)
return {player_id: default_player_stats_season()}
def Player_all_game_in_season2(player_id: str, season:str) -> dict:
url = f"{URL}api/abc/players/stats?tag={LEAGUE}&season={season}&id={player_id}"
player_games = get_json(url)
# games = {}
if not player_games:
logger.error(f"Пустой ответ от API для игрока {player_id}")
return {player_id: default_player_stats_season()}
for i in player_games.get("items"):
i["season"] = season
return {player_games.get("items")}
def Player_all_game2(player_id: str) -> dict:
url = f"{URL}api/abc/players/info?tag={LEAGUE}&id={player_id}"
player_seasons = get_json(url)
if not player_seasons:
logger.error(f"Пустой ответ от API для игрока {player_id}")
return {player_id: default_player_stats_season()}
seasons = player_seasons.get("result").get("seasons")
player_game = []
with ThreadPoolExecutor() as pool:
player_season_stat_temp = [
pool.submit(Player_all_game_in_season, player_id, season["id"])
for season in seasons
]
for i in player_season_stat_temp:
print(i.result())
player_game += [
res.result() for res in player_season_stat_temp
]
rewrite_file(player_id, player_game)
def Player_all_game_in_season(player_id: str, season: str) -> List[Dict[str, Any]]:
url = f"{URL}api/abc/players/stats?tag={LEAGUE}&season={season}&id={player_id}"
player_games = get_json(url)
if not player_games:
logger.error(f"Пустой ответ от API для игрока {player_id}, сезон {season}")
return [
] # возвращаем пустой список, чтобы тип был стабилен
items = player_games.get("items") or []
# гарантируем список словарей
if not isinstance(items, list):
logger.warning(f"Неверный формат 'items' для {player_id}, сезон {season}: {type(items)}")
return []
for it in items:
if isinstance(it, dict):
it["season"] = season
return items
def Player_all_game(player_id: str) -> List[Dict[str, Any]]:
# url = f"{URL}api/abc/players/info?tag={LEAGUE}&id={player_id}"
# player_seasons = get_json(url)
# if not player_seasons:
# logger.debug(f"Пустой ответ от API для игрока {player_id}")
# return [] # последовательный тип
# result = player_seasons.get("result") or {}
# seasons = result.get("seasons") or []
seasons = [
{"id": 2026},
{"id": 2025},
{"id": 2024},
{"id": 2023},
{"id": 2022},
{"id": 2021},
{"id": 2020},
{"id": 2019},
{"id": 2018},
{"id": 2017},
{"id": 2016},
{"id": 2015},
{"id": 2014},
{"id": 2013},
{"id": 2012},
{"id": 2011},
{"id": 2010},
]
if not isinstance(seasons, list) or not seasons:
logger.error(f"Нет сезонов для игрока {player_id}")
return []
all_games: List[Dict[str, Any]] = []
with ThreadPoolExecutor() as pool:
futures = [pool.submit(Player_all_game_in_season, player_id, s.get("id")) for s in seasons if s.get("id")]
for fut in as_completed(futures):
try:
items = fut.result() # это уже список словарей
all_games.extend(items)
except Exception as e:
logger.exception(f"Ошибка при сборе игр игрока {player_id}: {e}")
# если нужно писать на диск — пишем уже готовый JSON-совместимый список
rewrite_file(player_id, all_games)
return all_games
def default_player_stats_season() -> list:
empty_stats = {
"games": 0,
"isStarts": "",
"points": "",
"goal2": "",
"shot2": "",
"goal3": "",
"shot3": "",
"goal1": "",
"shot1": "",
"goal23": "",
"shot23": "",
"shot2Percent": "",
"shot3Percent": "",
"shot23Percent": "",
"shot1Percent": "",
"assist": "",
"pass": "",
"steal": "",
"blockShot": "",
"blockedOwnShot": "",
"defRebound": "",
"offRebound": "",
"rebound": "",
"foulsOnPlayer": "",
"turnover": "",
"foul": "",
"second": 0,
"playedTime": "",
"dunk": "",
"fastBreak": "",
"plusMinus": None,
}
return [
{"team": None, "game": None, "stats": empty_stats.copy(), "class": "Sum"},
{"team": None, "game": None, "stats": empty_stats.copy(), "class": "Avg"},
]
def default_player_stats() -> list:
empty_stats = {
"games": 0,
"isStarts": "",
"points": "",
"goal2": "",
"shot2": "",
"goal3": "",
"shot3": "",
"goal1": "",
"shot1": "",
"goal23": "",
"shot23": "",
"shot2Percent": "",
"shot3Percent": "",
"shot23Percent": "",
"shot1Percent": "",
"assist": "",
"pass": "",
"steal": "",
"blockShot": "",
"blockedOwnShot": "",
"defRebound": "",
"offRebound": "",
"rebound": "",
"foulsOnPlayer": "",
"turnover": "",
"foul": "",
"second": 0,
"playedTime": "",
"dunk": "",
"fastBreak": "",
"plusMinus": None,
}
return [
{"season": None, "team": None, "stats": empty_stats.copy(), "class": "Sum"},
{"season": None, "team": None, "stats": empty_stats.copy(), "class": "Avg"},
]
def Player_Stat_Career(player_id: str) -> dict:
url = f"{URL}api/abc/players/career?teamId=0&Tag={LEAGUE}&Id={player_id}"
player_stat_career = get_json(url)
if not player_stat_career:
logger.error(f"Пустой ответ от API для игрока {player_id}")
return {player_id: default_player_stats()}
items = player_stat_career.get("items")
if items:
logger.info(f"Данные за карьеру игрока {player_id} успешно получены.")
return {player_id: items[-2:]} # последние два сезона (Sum и Avg)
logger.warning(f"Данные на игрока {player_id} не найдены. Вероятно, новичок.")
return {player_id: default_player_stats()}
def Coach_Stat(coach_id: str, season: str, team_id: str) -> dict | None:
url = f"{URL}api/abc/coaches/career?teamId={team_id}&tag={LEAGUE}&season={season}&Id={coach_id}"
coach_stat = get_json(url)
if not coach_stat:
logger.error(f"Пустой ответ от API для тренера {coach_id}")
return None
items = coach_stat.get("items")
if items:
logger.info(f"Данные за карьеру тренера {coach_id} успешно получены.")
return {coach_id: items}
logger.warning(f"Данные для тренера {coach_id} не найдены. Возможно, он новичок.")
return None
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
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 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
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(stop_event: threading.Event) -> None:
"""
Обновляет файл team_stats.json, содержащий сравнение двух команд.
Аргументы:
stop_event (threading.Event): Событие для остановки цикла.
"""
logger.info("START making json for team statistics")
global game_online_data
while not stop_event.is_set():
with game_online_lock:
game_data = game_online_data
if not game_data:
time.sleep(1)
continue
try:
teams = game_data["result"]["teams"]
plays = game_data["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_ONLINE)
continue
# Таймауты
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.warning("Нет total у команд — пропускаю перезапись team_stats.json")
stop_event.wait(TIMEOUT_ONLINE)
continue
# Форматирование общей статистики (как и было)
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,
)
# # Форматирование общей статистики
# 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,
}
)
rewrite_file("team_stats", result_json)
logger.info("Успешно записаны данные в team_stats.json")
except Exception as e:
logger.error(
f"Ошибка при обработке командной статистики: {e}", exc_info=True
)
stop_event.wait(TIMEOUT_ONLINE)
def Referee(stop_event: threading.Event) -> None:
"""
Поток, создающий JSON-файл с информацией о судьях матча.
"""
logger.info("START making json for referee")
global game_online_data
desired_order = [
"Crew chief",
"Referee 1",
"Referee 2",
"Commissioner",
"Ст.судья",
"Судья 1",
"Судья 2",
"Комиссар",
]
while not stop_event.is_set():
with game_online_lock:
game_data = game_online_data
if not game_data:
stop_event.wait(TIMEOUT_ONLINE)
continue
try:
# Найти судей (teamNumber == 0)
team_ref = next(
(t for t in game_data["result"]["teams"] if t["teamNumber"] == 0), None
)
if not team_ref:
logger.warning("Не найдена судейская бригада в данных.")
stop_event.wait(TIMEOUT_DATA_OFF)
continue
referees_raw = team_ref.get("starts", [])
# print(referees_raw)
referees = []
for r in referees_raw:
flag_code = (
r.get("countryId", "").lower() if r.get("countryName") else ""
)
referees.append(
{
"displayNumber": r.get("displayNumber", ""),
"positionName": r.get("positionName", ""),
"lastNameGFX": f"{r.get('firstName', '')} {r.get('lastName', '')}".strip(),
"secondName": r.get("secondName", ""),
"birthday": r.get("birthday", ""),
"age": r.get("age", 0),
"flag": f"https://flagicons.lipis.dev/flags/4x3/{flag_code}.svg",
}
)
# Сортировка по позиции
referees = sorted(
referees,
key=lambda x: (
desired_order.index(x["positionName"])
if x["positionName"] in desired_order
else len(desired_order)
),
)
rewrite_file("referee", referees)
logger.info("Успешно записаны судьи в файл")
except Exception as e:
logger.error(f"Ошибка в Referee потоке: {e}", exc_info=True)
stop_event.wait(TIMEOUT_DATA_OFF)
def Scores_Quarter(stop_event: threading.Event) -> None:
"""
Поток, обновляющий JSON со счётом по четвертям.
"""
logger.info("START making json for scores quarter")
global game_online_data
quarters = ["Q1", "Q2", "Q3", "Q4", "OT1", "OT2", "OT3", "OT4"]
while not stop_event.is_set():
try:
with game_online_lock:
game_data = game_online_data
if not game_data:
stop_event.wait(FETCH_INTERVAL)
continue
rewrite_file("game_online", game_data)
score_by_quarter = [{"Q": q, "score1": "", "score2": ""} for q in quarters]
# Сначала пробуем fullScore
full_score_str = (
game_data.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 game_data.get("result", {}):
periods = game_data["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.warning("Нет данных по счёту, сохраняем пустые значения.")
rewrite_file("scores", score_by_quarter)
except Exception as e:
logger.error(f"Ошибка в Scores_Quarter: {e}", exc_info=True)
stop_event.wait(TIMEOUT_ONLINE)
def status_online_func(data: dict) -> dict | None:
"""
Получает онлайн-статус игры и возвращает данные + путь к PNG-фолам.
"""
global URL
try:
game_id = data["game_id"]
url = f"{URL}api/abc/games/live-status?id={game_id}"
json_live_status = get_json(url)
if json_live_status.get("status") != "Ok":
logger.warning(f"Live status API вернул не 'Ok': {json_live_status}")
return None
status_data = json_live_status["result"]
path_to_png = (
r"D:\ГРАФИКА\БАСКЕТБОЛ\ЕДИНАЯ ЛИГА ВТБ 2022-2023\Scorebug Indicators"
)
fouls_a = min(status_data.get("foulsA", 0), 5)
fouls_b = min(status_data.get("foulsB", 0), 5)
status_data["foulsA_png"] = f"{path_to_png}\\Home_{fouls_a}.png"
status_data["foulsB_png"] = f"{path_to_png}\\Away_{fouls_b}.png"
return status_data
except Exception as e:
logger.error(f"Ошибка в status_online_func: {e}", exc_info=True)
return None
def Status_Online(data: dict, stop_event: threading.Event) -> None:
"""
Поток, обновляющий JSON-файл с онлайн-статусом матча.
"""
logger.info("START making json for status online")
global game_status_data
while not stop_event.is_set():
try:
result = status_online_func(data)
if result:
with game_status_lock:
game_status_data = result
rewrite_file("live_status", [game_status_data])
logger.info("Успешно записан онлайн-статус в файл.")
else:
logger.warning("status_online_func вернула None — пропуск записи.")
except Exception as e:
logger.error(f"Ошибка в Status_Online: {e}", exc_info=True)
stop_event.wait(TIMEOUT_ONLINE)
def Play_By_Play(data: dict, stop_event: threading.Event) -> None:
"""
Поток, обновляющий JSON-файл с последовательностью бросков в матче.
"""
logger.info("START making json for play-by-play")
global game_online_data, LEAGUE
while not stop_event.is_set():
try:
with game_online_lock:
game_data = game_online_data
if not game_data:
logger.error("game_online_data отсутствует")
stop_event.wait(TIMEOUT_DATA_OFF)
continue
teams = game_data.get("result", {}).get("teams", [])
team1_data = next((i for i in teams if i.get("teamNumber") == 1), None)
team2_data = next((i for i in teams if i.get("teamNumber") == 2), None)
if not team1_data or not team2_data:
logger.warning("Не удалось получить команды из game_online_data")
stop_event.wait(TIMEOUT_DATA_OFF)
continue
team1_name = data["team1"]
team2_name = data["team2"]
team1_startnum = [
p["startNum"]
for p in team1_data.get("starts", [])
if p.get("startRole") == "Player"
]
team2_startnum = [
p["startNum"]
for p in team2_data.get("starts", [])
if p.get("startRole") == "Player"
]
plays = game_data.get("result", {}).get("plays", [])
if not plays:
logger.warning("нет данных в play-by-play")
stop_event.wait(TIMEOUT_DATA_OFF)
continue
# Получение текущего времени игры
url = f"{URL}api/abc/games/live-status?id={data['game_id']}"
json_live_status = get_json(url)
last_event = plays[-1]
if not json_live_status or json_live_status.get("status") == "Not Found":
period = last_event.get("period", 1)
second = 0
else:
period = (json_live_status or {}).get("result", {}).get("period", 1)
second = (json_live_status or {}).get("result", {}).get("second", 0)
# Создание DataFrame из событий
df = pd.DataFrame(plays[::-1])
# Преобразование для лиги 3x3
if "3x3" in LEAGUE:
df["play"].replace({2: 1, 3: 2}, inplace=True)
df_goals = df[df["play"].isin([1, 2, 3])].copy()
if df_goals.empty:
logger.warning("нет данных о голах в play-by-play")
stop_event.wait(TIMEOUT_DATA_OFF)
continue
# Расчёты по очкам и времени
df_goals["score1"] = (
df_goals["startNum"].isin(team1_startnum) * df_goals["play"]
)
df_goals["score2"] = (
df_goals["startNum"].isin(team2_startnum) * df_goals["play"]
)
df_goals["score_sum1"] = df_goals["score1"].fillna(0).cumsum()
df_goals["score_sum2"] = df_goals["score2"].fillna(0).cumsum()
df_goals["new_sec"] = (
pd.to_numeric(df_goals["sec"], errors="coerce").fillna(0).astype(int)
// 10
)
df_goals["time_now"] = (600 if period < 5 else 300) - second
df_goals["quar"] = period - df_goals["period"]
df_goals["diff_time"] = np.where(
df_goals["quar"] == 0,
df_goals["time_now"] - df_goals["new_sec"],
(600 * df_goals["quar"] - df_goals["new_sec"]) + df_goals["time_now"],
)
df_goals["diff_time_str"] = df_goals["diff_time"].apply(
lambda x: (
f"{x // 60}:{str(x % 60).zfill(2)}" if isinstance(x, int) else x
)
)
# Текстовые поля
def generate_text(row, with_time=False, is_rus=False):
s1, s2 = int(row["score_sum1"]), int(row["score_sum2"])
team = (
team1_name
if not pd.isna(row["score1"]) and row["score1"] != 0
else team2_name
)
# ✅ Правильный порядок счёта в зависимости от команды
if team == team1_name:
score = f"{s1}-{s2}"
else:
score = f"{s2}-{s1}"
time_str = (
f" за {row['diff_time_str']}"
if is_rus
else f" in last {row['diff_time_str']}"
)
prefix = "рывок" if is_rus else "run"
return f"{team} {score} {prefix}{time_str if with_time else ''}"
df_goals["text_rus"] = df_goals.apply(
lambda r: generate_text(r, is_rus=True, with_time=False), axis=1
)
df_goals["text_time_rus"] = df_goals.apply(
lambda r: generate_text(r, is_rus=True, with_time=True), axis=1
)
df_goals["text"] = df_goals.apply(
lambda r: generate_text(r, is_rus=False, with_time=False), axis=1
)
df_goals["text_time"] = df_goals.apply(
lambda r: generate_text(r, is_rus=False, with_time=True), axis=1
)
df_goals["team"] = df_goals["score1"].apply(
lambda x: team1_name if not pd.isna(x) and x != 0 else team2_name
)
# Удаление лишнего
drop_cols = [
"children",
"start",
"stop",
"hl",
"sort",
"startNum",
"zone",
"x",
"y",
]
df_goals.drop(columns=drop_cols, inplace=True, errors="ignore")
# Порядок колонок
main_cols = ["text", "text_time"]
all_cols = main_cols + [
col for col in df_goals.columns if col not in main_cols
]
df_goals = df_goals[all_cols]
# Сохранение JSON
directory = FOLDER_JSON
os.makedirs(directory, exist_ok=True)
# ip_address = get_ip_address()
# host = ip_check.get(ip_address, {}).get("host")
host_prefix = _ipcheck()
filepath = os.path.join(directory, f"{host_prefix}play_by_play.json")
df_goals.to_json(filepath, orient="records", force_ascii=False, indent=4)
logger.info("Успешно положил данные об play-by-play в файл")
except Exception as e:
logger.error(f"Ошибка в Play_By_Play: {e}", exc_info=True)
stop_event.wait(TIMEOUT_ONLINE)
def schedule_daily_restart():
"""Перезапуск get_season_and_schedule каждый день в 00:05 (только не на Windows)."""
if not sys.platform.startswith("win"):
while True:
now = datetime.now()
next_run = (now + timedelta(days=1)).replace(hour=0, minute=5, second=0, microsecond=0)
sleep_time = (next_run - now).total_seconds()
logger.info(f"Следующий перезапуск get_season_and_schedule запланирован на {next_run.strftime('%Y-%m-%d %H:%M')}")
time.sleep(sleep_time)
try:
logger.info("⏰ Автоматический перезапуск get_season_and_schedule()")
get_season_and_schedule()
except Exception as e:
logger.error(f"Ошибка при автоматическом перезапуске: {e}", exc_info=True)
def clean_np_ints(obj):
if isinstance(obj, dict):
return {k: clean_np_ints(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [clean_np_ints(v) for v in obj]
elif isinstance(obj, np.integer):
return int(obj)
else:
return obj
def get_season_and_schedule() -> dict | None:
"""
Получает текущий сезон и ближайший сыгранный матч для команды.
Использует глобальные переменные: URL, LEAGUE, LANG, TEAM.
Returns:
dict | None: Словарь с данными матча или None при ошибке.
"""
global URL, LEAGUE, LANG, TEAM
try:
# Получение активного сезона
season_url = f"{URL}api/abc/comps/seasons?Tag={LEAGUE}&Lang={LANG}"
season_data = get_json(season_url)
season = (
season_data.get("result", [{}])[0].get("season") if season_data else None
)
print(season)
if not season:
logger.warning("Сезон не найден в данных.")
return None
# season = 2025
# Получение расписания
schedule_url = f"{URL}api/abc/comps/calendar?Tag={LEAGUE}&Season={season}&Lang={LANG}&MaxResultCount=1000"
schedule_data = get_json(schedule_url)
rewrite_file("schedule", schedule_data)
items = schedule_data.get("items") if schedule_data else None
if not items:
logger.warning("Расписание не содержит игр.")
return None
df = pd.json_normalize(items)
# Преобразование и фильтрация
df["game.localDate"] = pd.to_datetime(
df["game.localDate"], format="%d.%m.%Y", errors="coerce"
)
df["game.DateStr"] = df["game.localDate"].dt.strftime("%d.%m.%Y")
df["team1.name"] = df["team1.name"].str.strip()
current_date = pd.to_datetime(datetime.now().date())
df_filtered = df[
(df["game.localDate"] <= current_date)
& (df["team1.name"].str.lower() == TEAM.lower())
]
if df_filtered.empty:
logger.warning("Нет подходящих матчей для команды на сегодня.")
return None
last_game = df_filtered.iloc[-1]
return clean_np_ints({
"season": season,
"game_id": last_game["game.id"],
"team1_id": last_game["team1.teamId"],
"team2_id": last_game["team2.teamId"],
"team1": last_game["team1.name"],
"team2": last_game["team2.name"],
"when": last_game["game.DateStr"],
"time": last_game["game.localTime"],
})
except Exception as e:
logger.error(f"Ошибка при получении сезона или расписания: {e}", exc_info=True)
return None
def Standing_func(data: dict, stop_event: threading.Event) -> None:
logger.info("START making json for standings")
global URL, LEAGUE, LANG
while not stop_event.is_set():
try:
season = data["season"]
url = (
f"{URL}api/abc/comps/actual-standings?tag={LEAGUE}&season={season}&lang={LANG}"
)
data_standings = get_json(url)
if data_standings and "items" in data_standings and data_standings["items"]:
standings_temp = data_standings["items"]
for item in standings_temp:
if "standings" in item and item["standings"] != []:
standings_temp = item["standings"]
df = pd.json_normalize(standings_temp)
del df["scores"]
if not df["totalWin"].isna().all():
df["w_l"] = (
df["totalWin"].astype(str)
+ " / "
+ df["totalDefeat"].astype(str)
)
df["procent"] = df.apply(
lambda row: (
0
if row["w_l"] == "0 / 0"
or row["totalGames"] == 0
or pd.isna(row["totalWin"])
else round(
row["totalWin"] * 100 / row["totalGames"]
+ 0.000005
)
),
axis=1,
)
df["plus_minus"] = (
df["totalGoalPlus"] - df["totalGoalMinus"]
)
df["name"] = df["name"].replace("PBC Uralmash", "Uralmash")
directory = FOLDER_JSON
os.makedirs(directory, exist_ok=True)
host = _ipcheck()
filepath = os.path.join(
directory,
f"{host}standings_{LEAGUE}_{item['comp']['name'].replace(' ', '_')}.json",
)
df.to_json(
filepath,
orient="records",
force_ascii=False,
indent=4,
)
logger.info("Standings data saved successfully.")
elif "playoffPairs" in item and item["playoffPairs"] != []:
standings_temp = item["playoffPairs"]
df = pd.json_normalize(standings_temp)
directory = FOLDER_JSON
os.makedirs(directory, exist_ok=True)
host = _ipcheck()
filepath = os.path.join(
directory,
f"{host}standings_{LEAGUE}_{item['comp']['name'].replace(' ', '_')}.json",
)
df.to_json(
filepath,
orient="records",
force_ascii=False,
indent=4,
)
logger.info("Standings data saved successfully.")
except Exception as e:
logger.warning(f"Ошибка в турнирном положении: {e}")
stop_event.wait(TIMEOUT_DATA_OFF)
def convert_numpy(obj):
if isinstance(obj, dict):
return {k: convert_numpy(v) for k, v in obj.items()}
elif isinstance(obj, list):
return [convert_numpy(v) for v in obj]
elif isinstance(obj, (np.integer, np.int64, np.int32)):
return int(obj)
elif isinstance(obj, (np.floating, np.float64, np.float32)):
return float(obj)
elif isinstance(obj, (np.bool_)):
return bool(obj)
elif isinstance(obj, (np.ndarray,)):
return obj.tolist()
return obj
def How_To_Play_Quarter(data: dict) -> None:
logger.info("START making json for How_To_Play_Quarter")
global LEAGUE, LANG
game_id = data["game_id"]
team1_id, team2_id = data["team1_id"], data["team2_id"]
team1_name, team2_name = data["team1"], data["team2"]
season = data["season"]
url = f"{URL}api/abc/comps/calendar?Tag={LEAGUE}&Season={season}&Lang={LANG}&MaxResultCount=1000"
schedule_data = get_json(url)
df_schedule = pd.json_normalize(schedule_data["items"])
df_schedule = df_schedule[
[
"game.id",
"ot",
"game.gameStatus",
"game.score1",
"game.score2",
"game.fullScore",
"team1.teamId",
"team1.name",
"team2.teamId",
"team2.name",
]
]
df_schedule = df_schedule[
(df_schedule["game.id"] < game_id)
& (
df_schedule["team1.teamId"].isin([team1_id, team2_id])
| df_schedule["team2.teamId"].isin([team1_id, team2_id])
)
]
QUARTERS = ["Q1", "Q2", "Q3", "Q4", "OT1", "OT2", "OT3", "OT4"]
def parse_quarters(df):
temp_score_quarter = df["game.fullScore"].str.split(",")
df = df.copy()
for i in range(1, 9):
col = f"OT{i - 4}" if i > 4 else f"Q{i}"
df[col] = temp_score_quarter.apply(
lambda x: x[i - 1] if x and len(x) >= i else None
)
return df
def compute_results(df, team_id):
df = parse_quarters(df)
for q in QUARTERS:
def result(row):
if pd.notna(row[q]) and ":" in row[q]:
score1, score2 = map(int, row[q].split(":"))
if row["team1.teamId"] == team_id:
return (
"win"
if score1 > score2
else "lose" if score1 < score2 else "draw"
)
elif row["team2.teamId"] == team_id:
return (
"win"
if score2 > score1
else "lose" if score2 < score1 else "draw"
)
return ""
df[f"wld{q}"] = df.apply(result, axis=1)
df[f"win{q}"] = (df[f"wld{q}"] == "win").astype(int)
df[f"lose{q}"] = (df[f"wld{q}"] == "lose").astype(int)
df[f"draw{q}"] = (df[f"wld{q}"] == "draw").astype(int)
return df
def compute_scores(df, team_id):
df = parse_quarters(df)
for q in QUARTERS:
def score(row):
if pd.notna(row[q]) and ":" in row[q]:
score1, score2 = map(int, row[q].split(":"))
if row["team1.teamId"] == team_id:
return score1
elif row["team2.teamId"] == team_id:
return score2
return None
df[f"score{q}"] = df.apply(score, axis=1)
return df
df_team1 = compute_results(df_schedule.copy(), team1_id)
df_team2 = compute_results(df_schedule.copy(), team2_id)
df_scores1 = compute_scores(df_schedule.copy(), team1_id)
df_scores2 = compute_scores(df_schedule.copy(), team2_id)
def aggregate_data(team_name, df_result, df_score):
team_stats = {"team": team_name}
for q in QUARTERS:
team_stats[f"win{q}"] = df_result[f"win{q}"].sum()
team_stats[f"lose{q}"] = df_result[f"lose{q}"].sum()
team_stats[f"draw{q}"] = df_result[f"draw{q}"].sum()
team_stats[f"score{q}"] = int(df_score[f"score{q}"].sum())
team_stats[f"score_avg{q}"] = (
round(df_score[f"score{q}"].mean(), 1)
if not df_score[f"score{q}"].isna().all()
else None
)
return team_stats
schedule_json_quarter = [
aggregate_data(team1_name, df_team1, df_scores1),
aggregate_data(team2_name, df_team2, df_scores2),
]
json_ready_data = convert_numpy(schedule_json_quarter)
rewrite_file("scores_quarter", json_ready_data)
def pregame_data(data: dict) -> None:
logger.info("START making json for pregame_data")
global LEAGUE, LANG
game_id = data["game_id"]
team1_id, team2_id = data["team1_id"], data["team2_id"]
team1_name, team2_name = data["team1"], data["team2"]
season = data["season"]
teams = []
for team_id in (team1_id, team2_id):
url = f"{URL}api/abc/teams/stats?Tag={LEAGUE}&Season={season}&Id={team_id}"
team_stat = get_json(url)
data_team = team_stat["result"]["totalStats"]
data_team["team"] = team_stat["result"]["team"]["name"]
data_team["games"] = team_stat["result"]["games"]
temp_team = {
"team": data_team["team"],
"games": data_team["games"],
"points": round((data_team["points"] / data_team["games"]), 1),
"points_2": round((data_team["goal2"] * 100 / data_team["shot2"]), 1),
"points_3": round((data_team["goal3"] * 100 / data_team["shot3"]), 1),
"points_23": round((data_team["goal23"] * 100 / data_team["shot23"]), 1),
"points_1": round((data_team["goal1"] * 100 / data_team["shot1"]), 1),
"assists": round((data_team["assist"] / data_team["games"]), 1),
"rebounds": round(((data_team["defRebound"] + data_team["offRebound"]) / data_team["games"]), 1),
"steals": round((data_team["steal"] / data_team["games"]), 1),
"turnovers": round((data_team["turnover"] / data_team["games"]), 1),
"blocks": round((data_team["blockShot"] / data_team["games"]), 1),
"fouls": round((data_team["foul"] / data_team["games"]), 1),
}
teams.append(temp_team)
rewrite_file("team_comparison", teams)
def main():
global TEAM, LANG, URL
if TEAM is None:
logger.critical(f"{myhost}\nКоманда не указана")
time.sleep(1)
sys.exit(1)
logger.info(
f"{myhost}\n!!!!!!!!СТАРТ!!!!!!!!! \nHOST: {TEAM}\nTAG: {LEAGUE} \nВерсия кода: {VERSION}"
)
if LANG is None:
LANG = next(
(tag["lang"] for tag in TAGS if tag["tag"].lower() == LEAGUE.lower()), ""
)
URL = (
"https://basket.sportoteka.org/"
if "uba" in LEAGUE.lower()
# else "https://pro.russiabasket.org/"
else "https://vtb-league.org/"
)
stop_event = Event()
data = get_season_and_schedule()
if not data:
logger.critical("Не удалось получить данные сезона/матча.")
sys.exit(1)
# === Ежедневный перезапуск функции на Linux/Mac ===
if not sys.platform.startswith("win"):
threading.Thread(target=schedule_daily_restart, daemon=True, name="ScheduleRestart").start()
logger.info(
f"{myhost}\n{data['team1']} VS {data['team2']}\n{data['when']} {data['time']}"
)
# logger.debug(data)
# data = {
# "season": 2026,
# "game_id": 921412,
# "team1_id":3059,
# "team2_id":682,
# "team1":"BETCITY PARMA",
# "team2":"Avtodor",
# }
# Список потоков и их задач
threads = [
threading.Thread(
target=game_online_loop,
args=(data["game_id"], stop_event),
name="OnlineLoop",
),
threading.Thread(
target=Json_Team_Generation,
args=("team1", data, stop_event),
name="Team1JSON",
),
threading.Thread(
target=Json_Team_Generation,
args=("team2", data, stop_event),
name="Team2JSON",
),
threading.Thread(
target=Team_Both_Stat, args=(stop_event,), name="BothTeamsStat"
),
threading.Thread(target=Referee, args=(stop_event,), name="Referee"),
threading.Thread(
target=Scores_Quarter, args=(stop_event,), name="QuarterScore"
),
threading.Thread(
target=Status_Online, args=(data, stop_event), name="StatusOnline"
),
threading.Thread(
target=Play_By_Play, args=(data, stop_event), name="PlayByPlay"
),
threading.Thread(
target=Standing_func, args=(data, stop_event), name="Standings"
),
]
# Запуск всех потоков
for t in threads:
t.start()
logger.debug(f"Поток {t.name} запущен.")
How_To_Play_Quarter(data)
pregame_data(data)
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
logger.info("Остановка по Ctrl+C... Завершение потоков.")
stop_event.set()
for t in threads:
t.join()
logger.debug(f"Поток {t.name} завершён.")
logger.info("Все потоки завершены.")
if __name__ == "__main__":
main()