1. добавил scheduler для онлайн матчей и турнирки

2. добавил в расписание тура маски для счета и времени

Co-authored-by: Copilot <copilot@github.com>
This commit is contained in:
2026-04-23 17:39:44 +03:00
parent 44ec5cc69a
commit 32d1f8afa6
4 changed files with 603 additions and 0 deletions

View File

@@ -5,6 +5,9 @@ import socket
import uuid
from urllib.parse import urlencode
from xml.etree import ElementTree as ET
import os
import sys
import ctypes
import requests
import websockets
@@ -18,6 +21,71 @@ POLL_INTERVAL = 3
REQUEST_TIMEOUT = 3
RECONNECT_DELAY = 5
def resource_path(relative_path: str) -> str:
try:
base_path = sys._MEIPASS
except Exception:
base_path = os.path.abspath(".")
return os.path.join(base_path, relative_path)
def setup_green_console():
if os.name != "nt":
return
kernel32 = ctypes.windll.kernel32
h_out = kernel32.GetStdHandle(-11) # STD_OUTPUT_HANDLE
mode = ctypes.c_uint()
if kernel32.GetConsoleMode(h_out, ctypes.byref(mode)):
kernel32.SetConsoleMode(h_out, mode.value | 0x0004) # ENABLE_VIRTUAL_TERMINAL_PROCESSING
# зелёный текст по умолчанию
print("\033[92m", end="")
def set_console_icon():
if os.name != "nt":
return
ico_path = resource_path("app.ico")
if not os.path.exists(ico_path):
return
user32 = ctypes.windll.user32
# AppUserModelID помогает Windows корректнее показывать иконку приложения
ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID("my.agent.console")
hwnd = kernel32_get_console_window()
if not hwnd:
return
IMAGE_ICON = 1
LR_LOADFROMFILE = 0x00000010
WM_SETICON = 0x0080
ICON_SMALL = 0
ICON_BIG = 1
hicon = user32.LoadImageW(
None,
ico_path,
IMAGE_ICON,
0,
0,
LR_LOADFROMFILE
)
if hicon:
user32.SendMessageW(hwnd, WM_SETICON, ICON_SMALL, hicon)
user32.SendMessageW(hwnd, WM_SETICON, ICON_BIG, hicon)
def kernel32_get_console_window():
try:
return ctypes.windll.kernel32.GetConsoleWindow()
except Exception:
return 0
def read_vmix_dynamic_values():
try:
@@ -185,4 +253,6 @@ async def run_agent():
if __name__ == "__main__":
setup_green_console()
set_console_icon()
asyncio.run(run_agent())

5
app.py
View File

@@ -110,10 +110,13 @@ from services.vmix_json_service import (
get_vmix_players_goal,
)
from scheduler import run_scheduler
BASE_DIR = Path(__file__).resolve().parent
app = FastAPI()
templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
app.mount("/static", StaticFiles(directory=str(BASE_DIR / "static")), name="static")
run_scheduler()
class PublishVmixCommandPayload(BaseModel):
match_id: int
@@ -1426,6 +1429,8 @@ def vmix_schedule(session_token: str):
"color": (
"#37F193" if row[5] not in ["scheduled", "finished"] else "#FFFFFF00"
),
"mask_time": "#FFFFFF00" if row[2] is not None and row[3] is not None else "#FFFFFF",
"mask_score": "#FFFFFF00" if row[2] is None and row[3] is None else "#FFFFFF",
}
for row in rows
]

BIN
app.zip

Binary file not shown.

528
scheduler.py Normal file
View File

@@ -0,0 +1,528 @@
import signal
import threading
import time
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
from psycopg2.extras import RealDictCursor
from db import get_connection
from parsers.parser_standings import run_parser_standings
import requests
from bs4 import BeautifulSoup
TZ = ZoneInfo("Europe/Moscow")
# За сколько до матча запускать watcher
MATCH_START_LEAD_MINUTES = 1
# Как часто обновлять live-матч
LIVE_MATCH_POLL_SECONDS = 60
# Как часто проверять матчи дня
MATCHES_LOOP_SECONDS = 60
# Как часто обновлять турнирную таблицу
STANDINGS_LOOP_SECONDS = 60
# Через сколько секунд между попытками после ошибки в worker
WORKER_ERROR_RETRY_SECONDS = 30
# Если подряд слишком много ошибок, worker остановится
WORKER_MAX_ERRORS_IN_ROW = 20
ACTIVE_MATCH_WORKERS: dict[int, threading.Event] = {}
ACTIVE_MATCH_THREADS: dict[int, threading.Thread] = {}
REGISTRY_LOCK = threading.Lock()
STANDINGS_LOCK = threading.Lock()
APP_STOP_EVENT = threading.Event()
# =========================================================
# ВСПОМОГАТЕЛЬНОЕ
# =========================================================
def log(message: str) -> None:
now = datetime.now(TZ).strftime("%Y-%m-%d %H:%M:%S")
print(f"[{now}] {message}")
def local_now_naive() -> datetime:
"""
В БД поле match_date у тебя TIMESTAMP без timezone.
Поэтому для корректного сравнения берем локальное время как naive.
"""
return datetime.now(TZ).replace(tzinfo=None)
def normalize_status(value: str | None) -> str:
status = (value or "").strip().lower()
if status in {"scheduled", "live", "finished", "postponed", "cancelled"}:
return status
return "scheduled"
# =========================================================
# БАЗА ДАННЫХ
# =========================================================
def get_today_matches_from_db() -> list[dict]:
"""
Возвращает матчи на сегодня, которые еще потенциально актуальны
для мониторинга. finished / cancelled можно не брать.
"""
now_local = datetime.now(TZ)
day_start = now_local.replace(hour=0, minute=0, second=0, microsecond=0).replace(tzinfo=None)
day_end = day_start + timedelta(days=1)
query = """
SELECT
m.id,
m.external_id,
m.match_date,
m.status,
m.home_score,
m.away_score,
m.tour,
m.season,
m.place,
m.date_raw,
m.score_add,
m.home_team_id,
m.away_team_id
FROM matches m
WHERE m.match_date >= %s
AND m.match_date < %s
AND COALESCE(m.status, 'scheduled') NOT IN ('finished', 'cancelled')
ORDER BY m.match_date ASC, m.id ASC
"""
conn = get_connection()
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(query, (day_start, day_end))
rows = cur.fetchall()
return [dict(row) for row in rows]
finally:
conn.close()
def update_match_live_state_in_db(
match_id: int,
status: str,
home_score: int | None,
away_score: int | None,
) -> None:
"""
Обновляет только live-состояние матча.
"""
query = """
UPDATE matches
SET
status = %s,
home_score = %s,
away_score = %s,
updated_at = NOW()
WHERE id = %s
"""
conn = get_connection()
try:
with conn.cursor() as cur:
cur.execute(query, (status, home_score, away_score, match_id))
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.close()
def get_match_snapshot_by_id(match_id: int) -> dict | None:
query = """
SELECT
m.id,
m.external_id,
m.match_date,
m.status,
m.home_score,
m.away_score,
m.tour,
m.season,
m.place,
m.date_raw,
m.score_add,
m.home_team_id,
m.away_team_id
FROM matches m
WHERE m.id = %s
LIMIT 1
"""
conn = get_connection()
try:
with conn.cursor(cursor_factory=RealDictCursor) as cur:
cur.execute(query, (match_id,))
row = cur.fetchone()
return dict(row) if row else None
finally:
conn.close()
# =========================================================
# ЭТУ ФУНКЦИЮ ТЫ ЗАПОЛНИШЬ САМ
# =========================================================
def fetch_html(url: str) -> str:
headers = {"User-Agent": "Mozilla/5.0", "Accept": "*/*"}
r = requests.get(url, headers=headers, timeout=20)
r.raise_for_status()
return r.text
def fetch_match_live_data(match_id: int) -> dict:
"""
Получение live-данных матча с сайта.
На вход получаешь ID матча, например:
{
"id": 123,
"external_id": "1069999",
"match_date": datetime(...),
"status": "scheduled",
"home_score": None,
"away_score": None,
"tour": "1 тур",
"season": "2025/2026",
"place": "...",
"date_raw": "...",
"score_add": None,
"home_team_id": 10,
"away_team_id": 11,
}
Должна вернуть dict минимум такого вида:
{
"status": "scheduled" | "live" | "finished",
"home_score": int | None,
"away_score": int | None,
}
Можно вернуть и больше полей, но scheduler использует только эти.
"""
html = fetch_html(f"https://wfl.rfs.ru/match/{match_id}")
soup = BeautifulSoup(html, "html.parser")
live = soup.find("section", class_="game game--future game--live js-game-live-label game--shadow game--progress")
if live:
scores = soup.find("div", class_="score__container").find_all("div", class_="score__item")
score1 = (scores[0].text).replace("-", 0) if len(scores) > 0 else None
score2 = (scores[2].text).replace("-", 0) if len(scores) > 1 else None
return [{"status": "live", "home_score": score1, "away_score": score2}]
raise NotImplementedError("Implement fetch_match_live_data(match_id) by yourself")
# =========================================================
# ЛОГИКА ЗАПУСКА WATCHER'ОВ
# =========================================================
def should_start_worker(match_row: dict, now_naive: datetime) -> bool:
"""
Нужно стартовать worker, если:
- матч еще не finished/cancelled
- и до старта осталось <= 1 минуты
- либо старт уже был раньше
"""
status = normalize_status(match_row.get("status"))
match_dt = match_row.get("match_date")
if match_dt is None:
return False
if status in {"finished", "cancelled"}:
return False
start_at = match_dt - timedelta(minutes=MATCH_START_LEAD_MINUTES)
return now_naive >= start_at
def is_match_terminal_status(status: str | None) -> bool:
return normalize_status(status) in {"finished", "cancelled"}
def register_worker(match_id: int, stop_event: threading.Event, thread: threading.Thread) -> bool:
with REGISTRY_LOCK:
if match_id in ACTIVE_MATCH_WORKERS:
return False
ACTIVE_MATCH_WORKERS[match_id] = stop_event
ACTIVE_MATCH_THREADS[match_id] = thread
return True
def unregister_worker(match_id: int) -> None:
with REGISTRY_LOCK:
ACTIVE_MATCH_WORKERS.pop(match_id, None)
ACTIVE_MATCH_THREADS.pop(match_id, None)
def stop_all_workers() -> None:
with REGISTRY_LOCK:
items = list(ACTIVE_MATCH_WORKERS.items())
for match_id, stop_event in items:
log(f"[scheduler] stopping worker for match_id={match_id}")
stop_event.set()
with REGISTRY_LOCK:
threads = list(ACTIVE_MATCH_THREADS.items())
for match_id, thread in threads:
if thread.is_alive():
thread.join(timeout=5)
log(f"[scheduler] worker joined for match_id={match_id}")
# =========================================================
# WORKER ОДНОГО МАТЧА
# =========================================================
def live_match_worker(match_row: dict, stop_event: threading.Event) -> None:
match_id = match_row["id"]
external_id = match_row.get("external_id")
errors_in_row = 0
log(f"[worker] started match_id={match_id} external_id={external_id}")
try:
while not APP_STOP_EVENT.is_set() and not stop_event.is_set():
# На каждой итерации перечитываем матч из БД:
# это полезно, если кто-то руками поменял статус или счет
current_match = get_match_snapshot_by_id(match_id)
if not current_match:
log(f"[worker] match_id={match_id} not found anymore, stopping")
break
current_status = normalize_status(current_match.get("status"))
if is_match_terminal_status(current_status):
log(f"[worker] match_id={match_id} already terminal status={current_status}, stopping")
break
try:
data = fetch_match_live_data(match_id)
errors_in_row = 0
except NotImplementedError:
log(f"[worker] fetch_match_live_data() is not implemented for match_id={match_id}")
break
except Exception as exc:
errors_in_row += 1
log(
f"[worker][error] match_id={match_id} "
f"fetch failed ({errors_in_row}/{WORKER_MAX_ERRORS_IN_ROW}): {exc}"
)
if errors_in_row >= WORKER_MAX_ERRORS_IN_ROW:
log(f"[worker] match_id={match_id} too many errors, stopping")
break
stop_event.wait(WORKER_ERROR_RETRY_SECONDS)
continue
new_status = normalize_status(data.get("status"))
new_home_score = data.get("home_score")
new_away_score = data.get("away_score")
changed = (
current_status != new_status
or current_match.get("home_score") != new_home_score
or current_match.get("away_score") != new_away_score
)
if changed:
update_match_live_state_in_db(
match_id=match_id,
status=new_status,
home_score=new_home_score,
away_score=new_away_score,
)
log(
f"[worker] match_id={match_id} updated "
f"status={new_status} score={new_home_score}:{new_away_score}"
)
else:
log(
f"[worker] match_id={match_id} no changes "
f"status={new_status} score={new_home_score}:{new_away_score}"
)
if is_match_terminal_status(new_status):
log(f"[worker] match_id={match_id} reached terminal status={new_status}, stopping")
break
stop_event.wait(LIVE_MATCH_POLL_SECONDS)
finally:
unregister_worker(match_id)
log(f"[worker] stopped match_id={match_id} external_id={external_id}")
# =========================================================
# LOOP: ПРОВЕРКА МАТЧЕЙ СЕГОДНЯ
# =========================================================
def watch_today_matches_once() -> None:
now_naive = local_now_naive()
try:
matches = get_today_matches_from_db()
except Exception as exc:
log(f"[matches_loop][error] failed to load today matches: {exc}")
return
if not matches:
log("[matches_loop] no matches today")
return
log(f"[matches_loop] today matches found: {len(matches)}")
for match_row in matches:
match_id = match_row["id"]
if not should_start_worker(match_row, now_naive):
continue
stop_event = threading.Event()
thread = threading.Thread(
target=live_match_worker,
args=(match_row, stop_event),
daemon=True,
name=f"live_match_worker_{match_id}",
)
registered = register_worker(match_id, stop_event, thread)
if not registered:
continue
thread.start()
log(
f"[matches_loop] worker launched "
f"match_id={match_id} external_id={match_row.get('external_id')}"
)
def matches_loop() -> None:
log("[matches_loop] started")
while not APP_STOP_EVENT.is_set():
started_at = time.time()
try:
watch_today_matches_once()
except Exception as exc:
log(f"[matches_loop][fatal] {exc}")
elapsed = time.time() - started_at
sleep_seconds = max(1, MATCHES_LOOP_SECONDS - int(elapsed))
APP_STOP_EVENT.wait(sleep_seconds)
log("[matches_loop] stopped")
# =========================================================
# LOOP: ТУРНИРНАЯ ТАБЛИЦА
# =========================================================
def standings_job() -> None:
# Не даем двум запускам standings идти одновременно
acquired = STANDINGS_LOCK.acquire(blocking=False)
if not acquired:
log("[standings_loop] skipped: previous standings job still running")
return
try:
run_parser_standings()
log("[standings_loop] standings updated")
except Exception as exc:
log(f"[standings_loop][error] {exc}")
finally:
STANDINGS_LOCK.release()
def standings_loop() -> None:
log("[standings_loop] started")
while not APP_STOP_EVENT.is_set():
started_at = time.time()
try:
standings_job()
except Exception as exc:
log(f"[standings_loop][fatal] {exc}")
elapsed = time.time() - started_at
sleep_seconds = max(1, STANDINGS_LOOP_SECONDS - int(elapsed))
APP_STOP_EVENT.wait(sleep_seconds)
log("[standings_loop] stopped")
# =========================================================
# ОСТАНОВКА ПРОЦЕССА
# =========================================================
def shutdown(signum=None, frame=None) -> None:
if APP_STOP_EVENT.is_set():
return
log(f"[scheduler] shutdown requested signum={signum}")
APP_STOP_EVENT.set()
stop_all_workers()
def install_signal_handlers() -> None:
signal.signal(signal.SIGINT, shutdown)
signal.signal(signal.SIGTERM, shutdown)
# =========================================================
# СТАРТ
# =========================================================
def run_scheduler() -> None:
install_signal_handlers()
log("[scheduler] started")
matches_thread = threading.Thread(
target=matches_loop,
daemon=True,
name="matches_loop_thread",
)
standings_thread = threading.Thread(
target=standings_loop,
daemon=True,
name="standings_loop_thread",
)
matches_thread.start()
standings_thread.start()
try:
while not APP_STOP_EVENT.is_set():
time.sleep(1)
finally:
shutdown()
matches_thread.join(timeout=5)
standings_thread.join(timeout=5)
log("[scheduler] fully stopped")
if __name__ == "__main__":
run_scheduler()