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-06 20:49:24 +03:00
parent d95b5f10d1
commit c50c97bc82
4 changed files with 609 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

467
get_data.py
View File

@@ -1,5 +1,6 @@
from fastapi import FastAPI
from fastapi import FastAPI, HTTPException, Query
from fastapi.responses import Response, JSONResponse, HTMLResponse
from fastapi.encoders import jsonable_encoder
import pandas as pd
import requests, io, os
from requests.auth import HTTPBasicAuth
@@ -9,11 +10,14 @@ import uvicorn
from threading import Thread, Event, Lock
import time
from contextlib import asynccontextmanager
from smb_excel import read_excel_from_smb, fetch_smb_file
import numpy as np
# --- Глобальные переменные ---
selected_game_id: int | None = None
current_tournament_id: int | None = None # будем обновлять при загрузке расписания
current_season: str | None = None # будем обновлять при загрузке расписания
latest_game_data: dict | None = None # сюда кладём последние данные по матчу
latest_game_error: str | None = None
_latest_lock = Lock()
@@ -26,6 +30,13 @@ api_user = os.getenv("API_USER")
api_pass = os.getenv("API_PASS")
league = os.getenv("LEAGUE")
POLL_SEC = int(os.getenv("GAME_POLL_SECONDS"))
SERVER_NAME = os.getenv("SERVER_NAME")
SERVER_IP = os.getenv("SERVER_IP")
SHARE = os.getenv("SHARE")
PATH_IN_SHARE = os.getenv("PATH_IN_SHARE")
USER = os.getenv("USER")
PASSWORD = os.getenv("PASSWORD")
DOMAIN = os.getenv("DOMAIN")
@@ -51,14 +62,16 @@ def load_today_schedule():
return pd.DataFrame()
tournament_id = int(filtered.iloc[0]["id"])
global current_tournament_id
season = str(filtered.iloc[0]["season"])
global current_tournament_id, current_season
current_tournament_id = tournament_id
current_season = season
url_schedule = f"http://stat2tv.khl.ru/{tournament_id}/schedule-{tournament_id}.xml"
r = requests.get(url_schedule, auth=HTTPBasicAuth(api_user, api_pass), verify=False)
schedule_df = pd.read_xml(io.StringIO(r.text))
# Нужные колонки (скорректируй под реальные имена из XML)
needed_columns = ["id", "date", "time", "homeName_en", "visitorName_en", "arena"]
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()
@@ -103,6 +116,7 @@ def _build_game_url(tournament_id: int, game_id: int) -> str:
# Если у тебя другой шаблон — просто поменяй строку ниже.
return f"http://stat2tv.khl.ru/{tournament_id}/json_en/{game_id}.json"
def _fetch_game_once(tournament_id: int, game_id: int) -> dict:
"""Один запрос к API матча -> чистый JSON из API."""
url = _build_game_url(tournament_id, game_id)
@@ -176,6 +190,7 @@ app = FastAPI(
@app.get("/games")
async def games():
df = load_today_schedule()
if df.empty:
return JSONResponse({"message": "Сегодня матчей нет"})
@@ -184,6 +199,8 @@ async def games():
return Response(content=json_schedule, media_type="application/json")
@app.get("/select")
async def select():
df = load_today_schedule()
@@ -199,7 +216,8 @@ async def select():
home = row.get("homeName_en", "")
away = row.get("visitorName_en", "")
when = row.get("datetime_str", "")
arena = row.get("arena", "")
arena = row.get("arena_en", "")
city = row.get("arena_city_en", "")
rows_html.append(
f"""
<tr data-id="{gid}">
@@ -208,6 +226,7 @@ async def select():
<td>{home}</td>
<td>{away}</td>
<td>{arena}</td>
<td>{city}</td>
<td><button class="pick">Выбрать</button></td>
</tr>
"""
@@ -238,7 +257,7 @@ async def select():
<table id="tbl">
<thead>
<tr>
<th>ID</th><th>Дата/время</th><th>Хозяева</th><th>Гости</th><th>Арена</th><th></th>
<th>ID</th><th>Дата/время</th><th>Хозяева</th><th>Гости</th><th>Арена</th><th>Город</th><th></th>
</tr>
</thead>
<tbody>
@@ -319,6 +338,13 @@ async def game_url():
"game_id": selected_game_id,
"tournament_id": current_tournament_id
})
# @app.get("/info")
# async def info():
# if selected_game_id:
# df = load_today_schedule()
@app.get("/data")
async def game_data():
@@ -333,12 +359,6 @@ async def game_data():
@app.get("/referee")
async def referee():
json_data = latest_game_data["data"]
referees_id = [
json_data["game"]["mref1_id"],
json_data["game"]["mref2_id"],
json_data["game"]["lref1_id"],
json_data["game"]["lref2_id"],
]
data_referees = [
{
"number": json_data["game"]["mref1_num"],
@@ -361,9 +381,432 @@ async def referee():
# def team(who:str):
async def team(who:str):
""""who: A - домашняя команда, B - гостевая"""
with _latest_lock:
lgd = latest_game_data
# print(lgd)
if not lgd or "data" not in lgd:
return [{"details":"Нет данных по матчу!"}]
players1_temp = lgd["data"]["players"][who]
players1 = []
players1_f = []
players1_d = []
players1_g = []
goaltenders1 = []
for player in players1_temp:
players1_temp[player]["mask"] = "#FF0000"
# for s in seasonA_json:
# if int(players1_temp[player]["id"]) == int(s["id"]):
# for key, value in s.items():
# # Создаем новый ключ с префиксом
# new_key = f"season_{key}"
# players1_temp[player][new_key] = value
if players1_temp[player]["ps"] not in ["в", "g"]:
data1 = players1_temp[player]
if "." in data1["name"]:
lastname1, *names1 = data1["name"].split()
names_new = " ".join(names1)
elif len(data1["name"].split()) == 3:
*lastname1, names1 = data1["name"].split()
names_new = names1
lastname1 = " ".join(lastname1)
else:
lastname1, *names1 = data1["name"].split()
names_new = " ".join(names1)
if players1_temp[player]["ps"] == "н":
position = "нападающий"
elif players1_temp[player]["ps"] == "з":
position = "защитник"
elif players1_temp[player]["ps"] == "f":
position = "forward"
elif players1_temp[player]["ps"] == "d":
position = "defenseman"
data_with_number = {
"number": player,
"NameSurnameGFX": names_new + " " + lastname1,
"NameGFX": names_new,
"SurnameGFX": lastname1,
"PositionGFX": position,
**data1,
}
if players1_temp[player]["ps"] == "н":
players1_f.append(data_with_number)
elif players1_temp[player]["ps"] == "з":
players1_d.append(data_with_number)
elif players1_temp[player]["ps"] == "f":
players1_f.append(data_with_number)
elif players1_temp[player]["ps"] == "d":
players1_d.append(data_with_number)
else:
data2 = players1_temp[player]
position = ""
if players1_temp[player]["ps"] == "в":
position = "вратарь"
elif players1_temp[player]["ps"] == "g":
position = "Goaltender"
lastname1, *names1 = data2["name"].split()
names_new = " ".join(names1)
data_with_number2 = {
"number": player,
"NameSurnameGFX": names_new + " " + lastname1,
"NameGFX": names_new,
"SurnameGFX": lastname1,
"PositionGFX": position,
**data2,
}
players1_g.append(data_with_number2)
goaltenders1.append(data_with_number2)
def make_empty(example_list):
if not example_list:
return {}
return {key: "" for key in example_list[0].keys()}
empty_d = make_empty(players1_d)
empty_f = make_empty(players1_f)
empty_g = make_empty(players1_g)
# добивка пустыми слотами
while len(players1_d) < 9:
players1_d.append(empty_d.copy())
while len(players1_f) < 13:
players1_f.append(empty_f.copy())
while len(players1_g) < 3:
players1_g.append(empty_g.copy())
players1 = players1_d + players1_f + players1_g
# print(len(players1))
return players1
@app.get("/team1")
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()
# овертаймы: OT, OT1, OT2... или числовые >= 4
if s == "OT" or (s.startswith("OT") and s[2:].isdigit()):
return "AFTER OVERTIME"
if s.isdigit():
n = int(s)
if n >= 4:
return "AFTER OVERTIME"
# 1–3 — с порядковым суффиксом
if n % 100 in (11, 12, 13):
suffix = "TH"
else:
suffix = {1: "ST", 2: "ND", 3: "RD"}.get(n % 10, "TH")
return f"AFTER {n}{suffix} PERIOD"
# например, Total
if s == "TOTAL":
return "FINAL"
return ""
# 👉 сортировка периодов: 1,2,3, затем OT/OT2/...
def _period_sort_key(k: str) -> tuple[int, int]:
s = str(k).strip().upper()
# 1–3 — обычные периоды
if s.isdigit():
n = int(s)
if 1 <= n <= 3:
return (n, 0)
# 4-й и далее — трактуем как овертаймы (4 -> OT1, 5 -> OT2 ...)
return (100, n - 3)
# явные овертаймы: OT, OT1, OT2...
if s == "OT":
return (100, 1)
if s.startswith("OT") and s[2:].isdigit():
return (100, int(s[2:]))
# неизвестные — в конец
return (200, 0)
def _sorted_period_keys(teams_periods: dict) -> list[str]:
a = teams_periods.get("A", {})
b = teams_periods.get("B", {})
keys = [k for k in a.keys() if k in b]
return sorted(keys, key=_period_sort_key)
def _current_period_key(payload: dict) -> str | None:
keys = _sorted_period_keys(payload.get("teams_periods", {}))
return keys[-1] if keys else None
# >>> 1) Замените вашу async-функцию get_team_stat на обычную sync:
def format_team_stat(team1: dict, team2: dict, period: str | None = None) -> list[dict]:
"""Форматирует статы двух команд в список записей для GFX (общие ключи + подписи)."""
stat_list = [
("coach_id", "coach id", "ID тренеров"),
("coach_fullname", "coach fullname", "Тренеры"),
("name", "name", "Команды"),
("outs", "Outs", ""),
("pim", "Penalty Minutes", "Минуты штрафа"),
("shots", "Shots on goal", "Броски в створ"),
("goals", "Goals", "Голы"),
("fo", "Face-offs", "Вбрасывания всего"),
("fow", "Face-offs won", "Вбрасывания"),
("fow_pct", "Face-offs won, %", "Выигранные вбрасывания, %"),
("hits", "Hits", "Силовые приемы"),
("bls", "Blocked shots", "Блокированные броски"),
("tka", "Takeaways", "Отборы"),
("toa", "Time on attack", "Время в атаке"),
("tie", "Even Strength Time on Ice", "Время в равных составах"),
("tipp", "Powerplay Time On Ice", "Время при большинстве"),
("tish", "Shorthanded Time On Ice", "Время при меньшинстве"),
("tien", "Emptry Net Time On Ice", "Время с пустыми воротами"),
("gva", "Giveaways", "Потери шайбы"),
("p_intc", "Pass interceptions", "Перехваты передачи"),
]
teams = [{k: str(v) for k, v in t.items()} for t in [team1, team2]]
keys = list(teams[0].keys())
formatted = []
if period is not None:
formatted.append({
"name0": "period",
"name1": str(period),
"name2": "",
"StatParameterGFX": _period_label(period) or "Period"
})
for key in keys:
row = {"name0": key, "name1": teams[0].get(key, ""), "name2": teams[1].get(key, "")}
# подписи
for code, eng, _ru in stat_list:
if key == code:
row["StatParameterGFX"] = eng
break
formatted.append(row)
# постобработка отдельных полей
for r in formatted:
if r["name0"] == "fow_pct":
r["name1"] = str(round(float(r["name1"]))) if r["name1"] else r["name1"]
r["name2"] = str(round(float(r["name2"]))) if r["name2"] else r["name2"]
if r["name0"] == "coach_fullname":
# "Фамилия Имя" -> "Имя Фамилия" если есть пробел
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"])
return formatted
# >>> 2) Вспомогалки для обхода периодов
def _iter_period_pairs(teams_periods: dict):
"""
Итерируем по периодам, отдаём (period_key, A_stats, B_stats).
Ключи сортируем по числу, если это цифры.
"""
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]
def _build_all_stats(payload: dict) -> dict:
"""
Собирает общий блок ('total') и список по периодам ('periods') из json матча.
"""
total_a = payload["teams"]["A"]
total_b = payload["teams"]["B"]
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)
})
return result
def xl():
df = read_excel_from_smb(
server_name=SERVER_NAME,
server_ip=SERVER_IP,
share_name=SHARE,
file_path_in_share=PATH_IN_SHARE,
username=USER,
password=PASSWORD,
domain=DOMAIN,
client_machine_name="KHL_SOFT",
sheet_name="TEAMS", # или None, или список листов
)
df = df.replace([float("inf"), float("-inf")], pd.NA)
columns_to_keep = [
"Team", "Logo", "Short", "HexPodl", # ← укажи свои
]
df = df.loc[:, [c for c in columns_to_keep if c in df.columns]]
json_text = df.to_json(orient="records", force_ascii=False) # NaN/NA -> null
return Response(content=json_text, media_type="application/json; charset=utf-8")
@app.get("/teams/stats")
async def teams_stats(
scope: str = Query("all", pattern="^(all|total|period)$"),
n: str | None = Query(None, description="Номер периода (строка или число) при scope=period"),
):
"""
Все-в-одном: GET /teams/stats?scope=all
вернёт { total: [...], periods: [{period: "1", stats:[...]}, ...] }
Только общий: GET /teams/stats?scope=total
Конкретный период: GET /teams/stats?scope=period&n=2
"""
# читаем атомарно снапшот
with _latest_lock:
lgd = latest_game_data
if not lgd or "data" not in lgd or not isinstance(lgd["data"], dict):
raise HTTPException(status_code=400, detail="Нет данных по матчу.")
payload = lgd["data"]
if scope == "total":
data = format_team_stat(payload["teams"]["A"], payload["teams"]["B"], period="Total")
return JSONResponse({"scope": "total", "data": data})
if scope == "period":
# 👉 если n не задан/или просит актуальный — берём последний период
wants_current = (n is None) or (str(n).strip().lower() in {"current", "last"})
if wants_current:
key = _current_period_key(payload)
if key is None:
raise HTTPException(status_code=404, detail="Периоды не найдены.")
n = key # используем актуальный ключ периода
# ключи в исходном json строковые
period_key = str(n)
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({
"scope": "period",
"period": period_key,
"is_current": period_key == _current_period_key(payload),
"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)
})
def _norm_name(s: str | None) -> str:
"""Нормализует название команды для сравнения."""
if not s:
return ""
return str(s).strip().casefold()
@app.get("/info")
async def info():
# 1) Проверяем, выбран ли матч
global current_season
if not selected_game_id:
return JSONResponse({"message": "Матч не выбран", "selected_id": None})
# 2) Берём расписание и ищем строку по выбранному ID
df = load_today_schedule()
if df.empty:
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)
home_name = str(row.get("homeName_en", "")).strip()
away_name = str(row.get("visitorName_en", "")).strip()
# 3) Подтягиваем справочник команд из Excel (лист TEAMS)
teams_df = read_excel_from_smb(
server_name=SERVER_NAME,
server_ip=SERVER_IP,
share_name=SHARE,
file_path_in_share=PATH_IN_SHARE,
username=USER,
password=PASSWORD,
domain=DOMAIN,
client_machine_name="KHL_SOFT",
sheet_name="TEAMS",
)
# Оставляем только полезные поля (подгони под свой файл)
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)
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)
if __name__ == "__main__":
uvicorn.run(