diff --git a/agent.py b/agent.py index c1fb435..34c87ec 100644 --- a/agent.py +++ b/agent.py @@ -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()) diff --git a/app.py b/app.py index 2099086..6d9468a 100644 --- a/app.py +++ b/app.py @@ -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 ] diff --git a/app.zip b/app.zip deleted file mode 100644 index 9a091af..0000000 Binary files a/app.zip and /dev/null differ diff --git a/scheduler.py b/scheduler.py new file mode 100644 index 0000000..97d95fa --- /dev/null +++ b/scheduler.py @@ -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() \ No newline at end of file