528 lines
16 KiB
Python
528 lines
16 KiB
Python
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() |