TEST_WIN GIT
This commit is contained in:
250
get_data_GPT.py
Normal file
250
get_data_GPT.py
Normal file
@@ -0,0 +1,250 @@
|
||||
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)
|
||||
Reference in New Issue
Block a user