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

версия с чтением и передачей бинарного файла

Browse Source
This commit is contained in:
Юрий Черненко
2025-11-11 19:46:41 +03:00
parent 71f1e62630
commit 0d06097181
2 changed files with 211 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@@ -1,5 +1,6 @@
.env
/.venv
logs/
/logs/*
__pycache__/
*.pyc

268
get_data.py
View File

@@ -1,5 +1,5 @@
from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import Response, JSONResponse, HTMLResponse
from fastapi import FastAPI, HTTPException, Query, Request
from fastapi.responses import Response, JSONResponse, HTMLResponse, StreamingResponse
from fastapi.encoders import jsonable_encoder
import pandas as pd
import requests, io, os
@@ -12,6 +12,11 @@ import time
from contextlib import asynccontextmanager
import numpy as np
import nasio
import logging
import logging.config
import platform
import json
from pprint import pprint
# --- Глобальные переменные ---
@@ -42,7 +47,6 @@ PASSWORD = os.getenv("SYNO_PASSWORD")
PATH = "/team-folders/GFX/Hockey/KHL/Soft/MATCH.xlsm"
def load_today_schedule():
"""Возвращает DataFrame матчей на сегодня с нужными колонками (или пустой DF)."""
url_tournaments = "http://stat2tv.khl.ru/tournaments.xml"
@@ -74,7 +78,17 @@ def load_today_schedule():
schedule_df = pd.read_xml(io.StringIO(r.text))
# Нужные колонки (скорректируй под реальные имена из XML)
needed_columns = ["id", "date", "time", "homeName_en", "visitorName_en", "arena_en", "arena_city_en", "homeCity_en", "visitorCity_en"]
needed_columns = [
"id",
"date",
"time",
"homeName_en",
"visitorName_en",
"arena_en",
"arena_city_en",
"homeCity_en",
"visitorCity_en",
]
exist = [c for c in needed_columns if c in schedule_df.columns]
schedule_df = schedule_df[exist].copy()
@@ -124,10 +138,7 @@ def _fetch_game_once(tournament_id: int, game_id: int) -> dict:
"""Один запрос к API матча -> чистый JSON из API."""
url = _build_game_url(tournament_id, game_id)
r = requests.get(
url,
auth=HTTPBasicAuth(api_user, api_pass),
verify=False,
timeout=10
url, auth=HTTPBasicAuth(api_user, api_pass), verify=False, timeout=10
)
r.raise_for_status()
@@ -183,11 +194,12 @@ async def lifespan(app: FastAPI):
_worker_thread.join(timeout=2)
print("🛑 Background thread stopped")
app = FastAPI(
lifespan=lifespan,
docs_url=None, # ❌ отключает /docs
redoc_url=None, # ❌ отключает /redoc
openapi_url=None # ❌ отключает /openapi.json
openapi_url=None, # ❌ отключает /openapi.json
)
@@ -202,8 +214,6 @@ async def games():
return Response(content=json_schedule, media_type="application/json")
@app.get("/select")
async def select():
df = load_today_schedule()
@@ -336,11 +346,13 @@ async def get_selected_game():
async def game_url():
if not (selected_game_id and current_tournament_id):
return JSONResponse({"message": "game_id или tournament_id не задан"})
return JSONResponse({
return JSONResponse(
{
"url": _build_game_url(current_tournament_id, selected_game_id),
"game_id": selected_game_id,
"tournament_id": current_tournament_id
})
"tournament_id": current_tournament_id,
}
)
# @app.get("/info")
@@ -356,7 +368,9 @@ async def game_data():
return JSONResponse(latest_game_data)
if latest_game_error:
return JSONResponse({"error": latest_game_error}, status_code=502)
return JSONResponse({"message": "Ещё нет данных. Выберите матч и подождите первое обновление."})
return JSONResponse(
{"message": "Ещё нет данных. Выберите матч и подождите первое обновление."}
)
@app.get("/referee")
@@ -383,7 +397,6 @@ async def referee():
return data_referees
async def team(who: str):
""" "who: A - домашняя команда, B - гостевая"""
with _latest_lock:
@@ -489,10 +502,12 @@ async def team(who:str):
async def team1():
return await team("A")
@app.get("/team2")
async def team2():
return await team("B")
# 👉 метка для первой строки (period row)
def _period_label(period: str | int) -> str:
s = str(period).strip().upper()
@@ -516,6 +531,7 @@ def _period_label(period: str | int) -> str:
return "FINAL"
return ""
# 👉 сортировка периодов: 1,2,3, затем OT/OT2/...
def _period_sort_key(k: str) -> tuple[int, int]:
s = str(k).strip().upper()
@@ -582,15 +598,21 @@ def format_team_stat(team1: dict, team2: dict, period: str | None = None) -> lis
formatted = []
if period is not None:
formatted.append({
formatted.append(
{
"name0": "period",
"name1": str(period),
"name2": "",
"StatParameterGFX": _period_label(period) or "Period"
})
"StatParameterGFX": _period_label(period) or "Period",
}
)
for key in keys:
row = {"name0": key, "name1": teams[0].get(key, ""), "name2": teams[1].get(key, "")}
row = {
"name0": key,
"name1": teams[0].get(key, ""),
"name2": teams[1].get(key, ""),
}
# подписи
for code, eng, _ru in stat_list:
if key == code:
@@ -608,6 +630,7 @@ def format_team_stat(team1: dict, team2: dict, period: str | None = None) -> lis
def flip(s: str) -> str:
parts = s.split()
return f"{parts[1]} {parts[0]}" if len(parts) >= 2 else s
r["name1"] = flip(r["name1"])
r["name2"] = flip(r["name2"])
@@ -622,11 +645,13 @@ def _iter_period_pairs(teams_periods: dict):
"""
a = teams_periods.get("A", {})
b = teams_periods.get("B", {})
def _key(k):
try:
return int(k)
except (TypeError, ValueError):
return k
for k in sorted(a.keys(), key=_key):
if k in b:
yield k, a[k], b[k]
@@ -638,22 +663,20 @@ def _build_all_stats(payload: dict) -> dict:
"""
total_a = payload["teams"]["A"]
total_b = payload["teams"]["B"]
result = {
"total": format_team_stat(total_a, total_b),
"periods": []
}
result = {"total": format_team_stat(total_a, total_b), "periods": []}
for period_key, a_stat, b_stat in _iter_period_pairs(payload["teams_periods"]):
result["periods"].append({
"period": period_key,
"stats": format_team_stat(a_stat, b_stat)
})
result["periods"].append(
{"period": period_key, "stats": format_team_stat(a_stat, b_stat)}
)
return result
@app.get("/teams/stats")
async def teams_stats(
scope: str = Query("all", pattern="^(all|total|period)$"),
n: str | None = Query(None, description="Номер периода (строка или число) при scope=period"),
n: str | None = Query(
None, description="Номер периода (строка или число) при scope=period"
),
):
"""
Все-в-одном: GET /teams/stats?scope=all
@@ -671,7 +694,9 @@ async def teams_stats(
payload = lgd["data"]
if scope == "total":
data = format_team_stat(payload["teams"]["A"], payload["teams"]["B"], period="Total")
data = format_team_stat(
payload["teams"]["A"], payload["teams"]["B"], period="Total"
)
return JSONResponse({"scope": "total", "data": data})
if scope == "period":
@@ -688,21 +713,23 @@ async def teams_stats(
a = payload["teams_periods"]["A"].get(period_key)
b = payload["teams_periods"]["B"].get(period_key)
if a is None or b is None:
raise HTTPException(status_code=404, detail=f"Период {period_key} не найден.")
return JSONResponse({
raise HTTPException(
status_code=404, detail=f"Период {period_key} не найден."
)
return JSONResponse(
{
"scope": "period",
"period": period_key,
"is_current": period_key == _current_period_key(payload),
"data": format_team_stat(a, b, period=period_key)
})
"data": format_team_stat(a, b, period=period_key),
}
)
# scope == "all"
cur = _current_period_key(payload)
return JSONResponse({
"scope": "all",
"current_period": cur,
"data": _build_all_stats(payload)
})
return JSONResponse(
{"scope": "all", "current_period": cur, "data": _build_all_stats(payload)}
)
def _norm_name(s: str | None) -> str:
@@ -711,8 +738,16 @@ def _norm_name(s: str | None) -> str:
return ""
return str(s).strip().casefold()
def _load_buf():
buf = nasio.load_bio(user=USER, password=PASSWORD,
nas_ip=SERVER_NAME, nas_port="443", path=PATH)
if isinstance(buf, (bytes, bytearray, memoryview)):
buf = io.BytesIO(buf)
buf.seek(0)
return buf
@app.get("/info")
async def info():
async def info(format: str = "xlsx", sheet: str = "TEAMS"):
# 1) Проверяем, выбран ли матч
global current_season
if not selected_game_id:
@@ -721,73 +756,128 @@ async def info():
# 2) Берём расписание и ищем строку по выбранному ID
df = load_today_schedule()
if df.empty:
return JSONResponse({"message": "Сегодня матчей нет", "selected_id": selected_game_id})
return JSONResponse(
{"message": "Сегодня матчей нет", "selected_id": selected_game_id}
)
# безопасно приводим id к int и ищем
try:
row = df.loc[df["id"].astype(int) == int(selected_game_id)].iloc[0]
except Exception:
return JSONResponse({"message": "Выбранный матч не найден в расписании на сегодня",
"selected_id": selected_game_id}, status_code=404)
return JSONResponse(
{
"message": "Выбранный матч не найден в расписании на сегодня",
"selected_id": selected_game_id,
},
status_code=404,
)
home_name = str(row.get("homeName_en", "")).strip()
away_name = str(row.get("visitorName_en", "")).strip()
# 3) Подтягиваем справочник команд из Excel (лист TEAMS)
teams_df = nasio.load_formatted(
user=USER,
password=PASSWORD,
nas_ip=SERVER_NAME,
nas_port="443",
path=PATH,
sheet="TEAMS"
src = _load_buf()
if format == "xlsx":
# читаем нужный лист из исходного XLSM
df = pd.read_excel(src, sheet_name=sheet, engine="openpyxl")
# пишем НОВЫЙ XLSX (без макросов) — это то, что понимает vMix
out = io.BytesIO()
with pd.ExcelWriter(out, engine="openpyxl") as writer:
df.to_excel(writer, sheet_name=sheet, index=False)
out.seek(0)
return StreamingResponse(
out,
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
headers={
# стабильное имя файла, чтобы vMix не путался
"Content-Disposition": "inline; filename=vmix.xlsx",
# отключаем кэш браузера/прокси, vMix сам опрашивает по интервалу
"Cache-Control": "no-cache, no-store, must-revalidate",
"Pragma": "no-cache",
},
)
# Оставляем только полезные поля (подгони под свой файл)
keep = [ "Team", "Logo", "Short", "HexPodl", "HexBase", "HexText" ]
keep = [c for c in keep if c in teams_df.columns]
teams_df = teams_df.loc[:, keep].copy()
elif format == "csv":
df = pd.read_excel(src, sheet_name=sheet, engine="openpyxl")
csv_bytes = df.to_csv(index=False, encoding="utf-8-sig").encode("utf-8")
return StreamingResponse(
io.BytesIO(csv_bytes),
media_type="text/csv; charset=utf-8",
headers={
"Content-Disposition": "inline; filename=vmix.csv",
"Cache-Control": "no-cache, no-store, must-revalidate",
"Pragma": "no-cache",
},
)
# 4) Нормализованные ключи для джоина по имени
teams_df["__key"] = teams_df["Team"].apply(_norm_name)
elif format == "json":
df = pd.read_excel(src, sheet_name=sheet, engine="openpyxl")
payload = json.dumps(df.to_dict(orient="records"), ensure_ascii=False)
return Response(
content=payload,
media_type="application/json; charset=utf-8",
headers={
"Cache-Control": "no-cache, no-store, must-revalidate",
"Pragma": "no-cache",
},
)
def _pick_team_info(name: str) -> dict:
key = _norm_name(name)
hit = teams_df.loc[teams_df["__key"] == key]
if hit.empty:
# не нашли точное совпадение — вернём только название
return {"Team": name}
rec = hit.iloc[0].to_dict()
rec.pop("__key", None)
# заменим NaN/inf на None, чтобы JSON не падал
for k, v in list(rec.items()):
if pd.isna(v) or v in (np.inf, -np.inf):
rec[k] = None
return rec
return Response("Unsupported format", status_code=400)
home_info = _pick_team_info(home_name)
away_info = _pick_team_info(away_name)
date_obj = datetime.strptime(row.get("datetime_str", ""), "%d.%m.%Y %H:%M")
try:
full_format = date_obj.strftime("%B %-d, %Y")
except ValueError:
full_format = date_obj.strftime("%B %#d, %Y")
# # Оставляем только полезные поля (подгони под свой файл)
# keep = ["Team", "Logo", "Short", "HexPodl", "HexBase", "HexText"]
# keep = [c for c in keep if c in teams_df.columns]
# teams_df = teams_df.loc[:, keep].copy()
# # 4) Нормализованные ключи для джоина по имени
# teams_df["__key"] = teams_df["Team"].apply(_norm_name)
payload = [{
"selected_id": int(selected_game_id),
"tournament_id": int(current_tournament_id) if current_tournament_id else None,
"datetime": str(full_format),
"arena": str(row.get("arena_en", "")),
"arena_city": str(row.get("arena_city_en", "")),
"home": home_info,
"home_city": str(row.get("homeCity_en", "")),
"away": away_info,
"away_city": str(row.get("visitorCity_en", "")),
"season": current_season,
}]
# def _pick_team_info(name: str) -> dict:
# key = _norm_name(name)
# hit = teams_df.loc[teams_df["__key"] == key]
# if hit.empty:
# # не нашли точное совпадение — вернём только название
# return {"Team": name}
# rec = hit.iloc[0].to_dict()
# rec.pop("__key", None)
# # заменим NaN/inf на None, чтобы JSON не падал
# for k, v in list(rec.items()):
# if pd.isna(v) or v in (np.inf, -np.inf):
# rec[k] = None
# return rec
# home_info = _pick_team_info(home_name)
# away_info = _pick_team_info(away_name)
# date_obj = datetime.strptime(row.get("datetime_str", ""), "%d.%m.%Y %H:%M")
# try:
# full_format = date_obj.strftime("%B %-d, %Y")
# except ValueError:
# full_format = date_obj.strftime("%B %#d, %Y")
# payload = [
# {
# "selected_id": int(selected_game_id),
# "tournament_id": (
# int(current_tournament_id) if current_tournament_id else None
# ),
# "datetime": str(full_format),
# "arena": str(row.get("arena_en", "")),
# "arena_city": str(row.get("arena_city_en", "")),
# "home": home_info,
# "home_city": str(row.get("homeCity_en", "")),
# "away": away_info,
# "away_city": str(row.get("visitorCity_en", "")),
# "season": current_season,
# }
# ]
# return JSONResponse(content=payload)
# except Exception as ex:
# pprint(ex)
return JSONResponse(content=payload)
if __name__ == "__main__":
uvicorn.run(