ychernenko/RFB
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test1

Browse Source
This commit is contained in:
Юрий Черненко
2025-10-27 16:55:15 +03:00
parent 000b304ed9
commit c4de3c84fe
1 changed files with 561 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

562
get_data_new.py
View File

@@ -1 +1,561 @@
test=4
import time
from datetime import datetime
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 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/<name>.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 poll_game_live(session, league: str, season: str, game_id: int, lang: str, game_meta: dict):
"""
Онлайн-цикл:
- "game" и "pregame-fullstats" раз в 600 сек
- "live-status", "box-score", "play-by-play" раз в 1 сек
Всё, что надо сейчас дернуть, дергаем параллельно.
Цикл выходит, когда матч перестаёт быть live.
"""
slow_endpoints = ["game",] #"pregame-fullstats"]
fast_endpoints = ["live-status", "box-score", "play-by-play"]
last_call = {}
now = time.time()
for name in slow_endpoints + fast_endpoints:
last_call[name] = 0 # форсим первый вызов сразу
# пул потоков: 5 нам хватит (у нас максимум 5 ручек одновременно)
with ThreadPoolExecutor(max_workers=5) as executor:
while True:
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,
)
)
# будем смотреть на ответы, особенно live-status
game_finished = False
for fut in as_completed(futures):
try:
ep_name, data = fut.result()
last_call[ep_name] = now # помечаем как обновлённый
if ep_name == "live-status":
# проверяем статус, не закончилась ли игра
if isinstance(data, dict):
st = (
data.get("status")
or data.get("gameStatus")
or ""
).lower()
if st in ("resultconfirmed", "finished", "final"):
logger.info(
f"Game {game_id} finished by live-status -> stop loop"
)
game_finished = True
except Exception as e:
logger.exception(f"poll endpoint error: {e}")
# вторая страховка (инфо из календаря)
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
# чуть притормозим, чтобы не жарить CPU
time.sleep(0.2)
logger.debug("live poll tick ok")
def get_data_API2(session, league: str, team: str, lang: str):
json_seasons = fetch_api_data(
session, "seasons", host=HOST, league=league, lang=lang
)
if not json_seasons:
print("Не удалось получить список сезонов")
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:
print("Не удалось получить список матчей")
return
today_game, last_played = get_game_id(json_calendar, team)
if last_played:
game_id = last_played["game"]["id"]
fetch_api_data(session, "game", host=HOST, game_id=game_id, lang=lang)
elif today_game:
print("ОНЛАЙН")
game_id = today_game["game"]["id"]
fetch_api_data(session, "game", host=HOST, game_id=game_id, lang=lang)
fetch_api_data(
session,
"pregame",
host=HOST,
league=league,
season=season,
game_id=game_id,
lang=lang,
)
def get_data_API(session, league: str, team: str, lang: str):
json_seasons = fetch_api_data(
session, "seasons", host=HOST, league=league, lang=lang
)
if not json_seasons:
logger.error("Не удалось получить список сезонов")
return
season = json_seasons[0]["season"]
# standings и calendar просто кешируем в файлы
fetch_api_data(
session, "standings", host=HOST, league=league, season=season, lang=lang
)
json_calendar = fetch_api_data(
session, "calendar", host=HOST, league=league, season=season, lang=lang
)
if not json_calendar:
logger.error("Не удалось получить список матчей")
return
today_game, last_played = get_game_id(json_calendar, team)
# если есть завершённая последняя игра — просто сохраним её статические данные
if last_played and not today_game:
game_id = last_played["game"]["id"]
logger.info(f"Последний завершённый матч id={game_id}")
fetch_api_data(session, "game", host=HOST, game_id=game_id, lang=lang)
return
# если есть матч сегодня
if today_game:
game_id = today_game["game"]["id"]
logger.info(f"Онлайн матч id={game_id}")
# сразу получить стартовые данные перед циклом
fetch_api_data(session, "game", host=HOST, game_id=game_id, lang=lang)
# fetch_api_data(
# session,
# "pregame-fullstats",
# host=HOST,
# league=league,
# season=season,
# game_id=game_id,
# lang=lang,
# )
# если матч реально идёт -> запускаем быстрый опрос
if is_game_live(today_game["game"]):
poll_game_live(
session=session,
league=league,
season=season,
game_id=game_id,
lang=lang,
game_meta=today_game["game"],
)
else:
logger.info(
"Матч ещё не стартовал, но сегодня. Просто сохранили стартовые данные."
)
return
logger.info("Для этой команды игр сегодня нет и нет завершённой последней игры.")
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()
get_data_API(session, args.league, args.team, args.lang)
if __name__ == "__main__":
main()