250 lines
8.3 KiB
Python
250 lines
8.3 KiB
Python
import os
|
||
import logging
|
||
import logging.config
|
||
import platform
|
||
import requests
|
||
import urllib3
|
||
import json
|
||
|
||
|
||
class Config:
|
||
VERSION = "v.2.0"
|
||
MYHOST = platform.node()
|
||
LOG_DIR = "logs"
|
||
JSON_DIR = "JSON"
|
||
LEAGUE = "vtb"
|
||
API_URL = os.getenv("API_URL", (
|
||
"https://basket.sportoteka.org/"
|
||
if "uba" in LEAGUE.lower()
|
||
# else "https://deti.russiabasket.org/"
|
||
else "https://pro.russiabasket.org/"
|
||
))
|
||
|
||
LANG = "en"
|
||
TOKEN = os.getenv("BOT_TOKEN", "7639240596:AAH0YtdQoWZSC-_R_EW4wKAHHNLIA0F_ARY")
|
||
GROUP_CHAT = os.getenv("GROUP_CHAT_ID", 228977654)
|
||
TIMEOUT_ONLINE = 1
|
||
FETCH_INTERVAL = 2
|
||
TIMEOUT_DATA_OFF = 60
|
||
|
||
@staticmethod
|
||
def init_dirs():
|
||
os.makedirs(Config.LOG_DIR, exist_ok=True)
|
||
os.makedirs(Config.JSON_DIR, exist_ok=True)
|
||
|
||
|
||
class LoggerFactory:
|
||
@staticmethod
|
||
def setup_logger(name: str = __name__):
|
||
logging_config = {
|
||
"version": 1,
|
||
"handlers": {
|
||
"console": {
|
||
"class": "logging.StreamHandler",
|
||
"level": "DEBUG",
|
||
"formatter": "detailed",
|
||
"stream": "ext://sys.stdout",
|
||
},
|
||
"file": {
|
||
"class": "logging.FileHandler",
|
||
"level": "DEBUG",
|
||
"formatter": "detailed",
|
||
"filename": f"{Config.LOG_DIR}/GFX_{Config.MYHOST}.log",
|
||
"encoding": "utf-8",
|
||
},
|
||
},
|
||
"loggers": {
|
||
name: {
|
||
"handlers": ["console", "file"],
|
||
"level": "DEBUG",
|
||
}
|
||
},
|
||
"formatters": {
|
||
"detailed": {
|
||
"class": "logging.Formatter",
|
||
"format": "%(asctime)s %(levelname)-8s [%(funcName)s] - %(message)s",
|
||
"datefmt": "%d.%m.%Y %H:%M:%S",
|
||
}
|
||
},
|
||
}
|
||
logging.config.dictConfig(logging_config)
|
||
return logging.getLogger(name)
|
||
|
||
|
||
Config.init_dirs()
|
||
logger = LoggerFactory.setup_logger()
|
||
urllib3.disable_warnings()
|
||
|
||
|
||
class ApiClient:
|
||
def __init__(self, base_url: str, league: str, lang: str, logger):
|
||
self.base_url = base_url
|
||
self.league = league
|
||
self.lang = lang
|
||
self.logger = logger
|
||
|
||
def get_json(self, endpoint: str, params: dict = None):
|
||
url = self.base_url + endpoint
|
||
try:
|
||
self.logger.debug(f"GET: {url} | params: {params}")
|
||
response = requests.get(url, params=params, verify=False, timeout=10)
|
||
response.raise_for_status()
|
||
return response.json()
|
||
except requests.RequestException as e:
|
||
self.logger.warning(f"Ошибка при запросе {url}: {e}")
|
||
return None
|
||
|
||
def get_box_score(self, game_id):
|
||
return self.get_json("api/abc/games/box-score", {"Id": game_id})
|
||
|
||
def get_game_info(self, game_id):
|
||
return self.get_json("api/abc/games/game", {"Id": game_id, "Lang": self.lang})
|
||
|
||
def get_play_by_play(self, game_id):
|
||
return self.get_json("api/abc/games/play-by-play", {"Id": game_id})
|
||
|
||
def get_live_status(self, game_id):
|
||
return self.get_json("api/abc/games/live-status", {"id": game_id})
|
||
|
||
def get_player_stat_season(self, player_id, season):
|
||
return self.get_json("api/abc/players/stats", {"teamId": 0, "Tag": self.league, "season": season, "Id": player_id})
|
||
|
||
def get_player_stat_career(self, player_id):
|
||
return self.get_json("api/abc/players/career", {"teamId": 0, "Tag": self.league, "Id": player_id})
|
||
|
||
def get_coach_stat(self, coach_id, season, team_id):
|
||
return self.get_json("api/abc/coaches/career", {"teamId": team_id, "tag": self.league, "season": season, "Id": coach_id})
|
||
|
||
def get_combined_game_data(self, game_id):
|
||
box_score = self.get_box_score(game_id)
|
||
if not box_score:
|
||
return None
|
||
|
||
if box_score.get("status") != "Ok":
|
||
return self.get_game_info(game_id)
|
||
|
||
game_info = self.get_game_info(game_id)
|
||
play_by_play = self.get_play_by_play(game_id)
|
||
|
||
if not game_info or not play_by_play:
|
||
return None
|
||
|
||
for index_team, team_data in enumerate(game_info["result"].get("teams", [])[1:]):
|
||
for s in team_data.get("starts", []):
|
||
matching_stats = next(
|
||
(stat for stat in box_score["result"]["teams"][index_team]["starts"] if stat["startNum"] == s["startNum"]),
|
||
None
|
||
)
|
||
if matching_stats:
|
||
s["stats"] = matching_stats
|
||
team_data["total"] = box_score["result"]["teams"][index_team].get("total", {})
|
||
|
||
game_info["result"]["plays"] = play_by_play.get("result", [])
|
||
game_info["result"]["scoreByPeriods"] = box_score["result"].get("scoreByPeriods", [])
|
||
game_info["result"]["fullScore"] = box_score["result"].get("fullScore", {})
|
||
return game_info
|
||
|
||
|
||
def rewrite_file(filename, data):
|
||
filepath = os.path.join(Config.JSON_DIR, f"{filename}.json")
|
||
try:
|
||
with open(filepath, "w", encoding="utf-8") as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||
logger.debug(f"Файл {filepath} перезаписан.")
|
||
except Exception as e:
|
||
logger.error(f"Не удалось записать {filepath}: {e}")
|
||
|
||
|
||
def safe_int(value, default=0):
|
||
try:
|
||
return int(value)
|
||
except (ValueError, TypeError):
|
||
return default
|
||
|
||
|
||
def format_time(seconds):
|
||
try:
|
||
minutes = int(seconds // 60)
|
||
seconds = int(seconds % 60)
|
||
return f"{minutes}:{seconds:02}"
|
||
except Exception:
|
||
return "0:00"
|
||
|
||
|
||
def safe_percent(numerator, denominator):
|
||
try:
|
||
n = safe_int(numerator)
|
||
d = safe_int(denominator)
|
||
return f"{round(n * 100 / d, 1)}%" if d != 0 else "0.0%"
|
||
except Exception:
|
||
return "0.0%"
|
||
|
||
|
||
class TeamStatBuilder:
|
||
def __init__(self, api_client: ApiClient):
|
||
self.api = api_client
|
||
|
||
def build_team_json(self, team_data, season, team_id, game_id, team_key):
|
||
players = []
|
||
for player in team_data.get("starts", []):
|
||
person_id = player.get("personId")
|
||
if not person_id:
|
||
continue
|
||
stat_season = self.api.get_player_stat_season(person_id, season)
|
||
stat_career = self.api.get_player_stat_career(person_id)
|
||
player_info = {
|
||
"id": person_id,
|
||
"name": player.get("name"),
|
||
"position": player.get("position"),
|
||
"stats": player.get("stats", {}),
|
||
"seasonStats": stat_season.get("result") if stat_season else {},
|
||
"careerStats": stat_career.get("result") if stat_career else {}
|
||
}
|
||
players.append(player_info)
|
||
|
||
coach_id = team_data.get("coachId")
|
||
coach_stats = self.api.get_coach_stat(coach_id, season, team_id) if coach_id else {}
|
||
|
||
return {
|
||
"teamId": team_id,
|
||
"teamKey": team_key,
|
||
"players": players,
|
||
"coachStats": coach_stats.get("result") if coach_stats else {},
|
||
"total": team_data.get("total", {})
|
||
}
|
||
|
||
|
||
def process_game(game_id):
|
||
api = ApiClient(Config.API_URL, Config.LEAGUE, Config.LANG, logger)
|
||
team_builder = TeamStatBuilder(api)
|
||
|
||
logger.info(f"Обработка игры {game_id}...")
|
||
data = api.get_combined_game_data(game_id)
|
||
if not data or "result" not in data:
|
||
logger.error("Нет данных для обработки игры.")
|
||
return
|
||
|
||
result = data["result"]
|
||
season = result.get("season")
|
||
teams = result.get("teams", [])
|
||
|
||
if len(teams) < 3:
|
||
logger.error("Недостаточно данных о командах.")
|
||
return
|
||
|
||
team1 = team_builder.build_team_json(teams[1], season, result.get("team1Id"), game_id, "team1")
|
||
team2 = team_builder.build_team_json(teams[2], season, result.get("team2Id"), game_id, "team2")
|
||
|
||
rewrite_file("team1", team1)
|
||
rewrite_file("team2", team2)
|
||
|
||
top1 = sorted(team1["players"], key=lambda x: x["stats"].get("efficiency", 0), reverse=True)[:5]
|
||
top2 = sorted(team2["players"], key=lambda x: x["stats"].get("efficiency", 0), reverse=True)[:5]
|
||
|
||
rewrite_file("topTeam1", top1)
|
||
rewrite_file("topTeam2", top2)
|
||
|
||
logger.info("Готово!")
|
||
|
||
|
||
process_game(916116) |