import requests from bs4 import BeautifulSoup from datetime import datetime from zoneinfo import ZoneInfo from services.schedule_service import sync_matches from repositories.team_repository import get_team_external_id_by_name, get_team_id_by_external_id MONTHS_RU = { "янв.": 1, "февр.": 2, "мар.": 3, "апр.": 4, "мая": 5, "июн.": 6, "июл.": 7, "авг.": 8, "сент.": 9, "окт.": 10, "нояб.": 11, "дек.": 12, } TZ = ZoneInfo("Europe/Moscow") URL_SCHEDULE = ( "https://wfl.rfs.ru/tournament/1061879/calendar?round_id=1117550&type=tours" ) SEASON = "2025/2026" def parse_russian_date(date_str: str, year: int | None = None) -> datetime: # пример: "07 мар., пт, 13:00" if year is None: year = datetime.now(TZ).year parts = [p.strip() for p in date_str.split(",")] if len(parts) < 3: raise ValueError(f"Некорректный формат даты: {date_str}") day_month = parts[0] time_part = parts[2] day, month_str = day_month.split() month = MONTHS_RU[month_str] hh, mm = time_part.split(":") return datetime(year, month, int(day), int(hh), int(mm), tzinfo=TZ) def fetch_html(url: str) -> str: headers = {"User-Agent": "Mozilla/5.0", "Accept": "*/*"} response = requests.get(url, headers=headers, timeout=20) response.raise_for_status() return response.text def normalize_status(score1: str | None, score2: str | None, dt: datetime) -> str: now = datetime.now(TZ) if score1 is not None and score2 is not None: return "finished" if dt <= now: return "live" return "scheduled" def safe_int(value: str | None) -> int | None: if value is None: return None value = value.strip() return int(value) if value.isdigit() else None def parse_schedule(html: str) -> list[dict]: soup = BeautifulSoup(html, "html.parser") matches_data: list[dict] = [] table_div = soup.find("div", class_="timetable__main") if not table_div: return matches_data rows = table_div.find_all("div", class_="timetable__unit js-schedule-games-cont") for row in rows: tour_el = row.find("span", class_="timetable__head-text") tour = tour_el.get_text(strip=True) if tour_el else None matches = row.find_all("li", class_="timetable__item") for match in matches: score_link = match.find("a", class_="timetable__score") href = score_link.get("href") if score_link else "" match_id = href.split("/")[-1] if href else None time_el = match.find("span", class_="timetable__time") time_site = time_el.get_text(strip=True) if time_el else None if not time_site or not match_id: continue try: dt = parse_russian_date(time_site) except Exception as exc: print(f"[parser_schedule] Ошибка даты для матча {match_id}: {exc}") continue teams = match.find_all("div", class_="timetable__team-name") team1_name = teams[0].get_text(strip=True) if len(teams) > 0 else None team2_name = teams[1].get_text(strip=True) if len(teams) > 1 else None if not team1_name or not team2_name: print(f"[parser_schedule] Пропуск матча {match_id}: нет названий команд") continue home_team_external_id = get_team_external_id_by_name(team1_name) away_team_external_id = get_team_external_id_by_name(team2_name) if not home_team_external_id or not away_team_external_id: print( f"[parser_schedule] Пропуск матча {match_id}: " f"не найдены команды в БД ({team1_name} vs {team2_name})" ) continue score1 = score2 = None score_main_el = match.find("div", class_="timetable__score-main") if score_main_el: score_main = score_main_el.get_text(strip=True) if "-" in score_main and "- : -" not in score_main: parts = [s.strip() for s in score_main.split("-")] if len(parts) == 2: score1, score2 = parts[0], parts[1] home_score = safe_int(score1) away_score = safe_int(score2) status = normalize_status(home_score, away_score, dt) place_el = match.find("span", class_="timetable__place-name") place = place_el.get_text(strip=True) if place_el else None score_add_el = match.find("div", class_="timetable__score-additional") score_add = score_add_el.get_text(strip=True) if score_add_el else None matches_data.append( { "external_id": str(match_id), "home_team_external_id": str(home_team_external_id), "away_team_external_id": str(away_team_external_id), "match_date": dt.strftime("%Y-%m-%d %H:%M:%S"), "status": status, "home_score": home_score, "away_score": away_score, "tour": tour, "season": SEASON, "place": place, "date_raw": time_site, "score_add": score_add, } ) matches_data.sort(key=lambda x: x["match_date"] or "") return matches_data def run_parser_schedule() -> None: html = fetch_html(URL_SCHEDULE) matches_data = parse_schedule(html) sync_matches(matches_data) print(f"[parser_schedule] Matches synced: {len(matches_data)}") if __name__ == "__main__": run_parser_schedule()