обновление для Кубка России

переделаны все парсеры на ссылки из базы
2026-07-02 12:46:29 +03:00
parent b0fe3ad5a4
commit 57907fd86b
29 changed files with 1838 additions and 240 deletions
--- a/parsers/parser_game.py
+++ b/parsers/parser_game.py
@@ -2,8 +2,9 @@ import requests
 from bs4 import BeautifulSoup

 from services.game_service import sync_match_page
+from source_config import DATA_MATCH_BASE_URL, match_url

-BASE_MATCH_URL = "https://wfl.rfs.ru/match/"
+BASE_MATCH_URL = DATA_MATCH_BASE_URL


 def fetch_html(url: str) -> str:
@@ -197,7 +198,7 @@ def parse_game_page(html: str) -> dict:


 def run_parser_game(match_external_id: str) -> None:
-    url = f"{BASE_MATCH_URL}{str(match_external_id).strip()}"
+    url = match_url(match_external_id)
    html = fetch_html(url)
    data = parse_game_page(html)
    
--- a/parsers/parser_players.py
+++ b/parsers/parser_players.py
@@ -3,10 +3,9 @@ from bs4 import BeautifulSoup
 from concurrent.futures import ThreadPoolExecutor

 from services.players_service import sync_team_roster
+from parsers.parser_sources import get_parser_source, source_absolute_url


-URL_TEAMS = "https://wfl.rfs.ru/tournament/1061879/teams"
-
 AMPLUA_FULL = {
    "Пз.": "Полузащитник",
    "Вр.": "Вратарь",
@@ -23,19 +22,26 @@ def fetch_html(url: str) -> str:
    return r.text


-def get_links(html: str) -> list[dict]:
+def get_links(html: str, source: dict) -> list[dict]:
    soup = BeautifulSoup(html, "html.parser")
    links: list[dict] = []

-    items = soup.find("ul", class_="teams__list").find_all("li")
-    for i in items:
-        href = i.find("a", class_="teams__link").get("href")
-        team_external_id = href.split("team_id=")[-1].strip()
+    teams_list = soup.find("ul", class_="teams__list")
+    if not teams_list:
+        return links

+    items = teams_list.find_all("li")
+    for item in items:
+        link_el = item.find("a", class_="teams__link")
+        href = link_el.get("href") if link_el else ""
+        if not href:
+            continue
+
+        team_external_id = href.split("team_id=")[-1].strip() if "team_id=" in href else ""
        links.append(
            {
                "team_external_id": team_external_id,
-                "url": "https://wfl.rfs.ru" + href,
+                "url": source_absolute_url(source, href),
            }
        )

@@ -92,13 +98,14 @@ def parse_team(html: str) -> dict:

        full_player = name_p.get_text(strip=True) if name_p else ""
        parts = full_player.split()
+        pos_short = pos_td.get_text(strip=True) if pos_td else ""

        players.append(
            {
                "player_id": player_id or "",
                "number": number_td.get_text(strip=True) if number_td else "",
-                "pos": pos_td.get_text(strip=True) if pos_td else "",
-                "amplua": AMPLUA_FULL[pos_td.get_text(strip=True) if pos_td else ""],
+                "pos": pos_short,
+                "amplua": AMPLUA_FULL.get(pos_short, pos_short),
                "player": full_player,
                "lastname": parts[0] if len(parts) >= 1 else "",
                "name": parts[-1] if len(parts) >= 2 else "",
@@ -144,9 +151,13 @@ def parse_team(html: str) -> dict:
    }


-def run_parser_players() -> None:
-    html = fetch_html(URL_TEAMS)
-    links = get_links(html)
+def run_parser_players(source_key: str | None = None) -> None:
+    source = get_parser_source(source_key)
+    print(f"[parser_players] Источник: {source['title']}")
+    print(f"[parser_players] URL: {source['teams_url']}")
+
+    html = fetch_html(source["teams_url"])
+    links = get_links(html, source)

    with ThreadPoolExecutor(max_workers=8) as pool:
        futures = {pool.submit(fetch_html, item["url"]): item for item in links}
@@ -169,6 +180,9 @@ def run_parser_players() -> None:
            except Exception as e:
                print(f"[parser_players] error team={item['team_external_id']}: {e}")

+    if not links:
+        print("[parser_players] Команды для парсинга игроков не найдены")
+

 if __name__ == "__main__":
    run_parser_players()
--- a/parsers/parser_schedule.py
+++ b/parsers/parser_schedule.py
@@ -5,6 +5,7 @@ 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
+from parsers.parser_sources import get_parser_source


 MONTHS_RU = {
@@ -24,11 +25,6 @@ MONTHS_RU = {

 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:
@@ -76,7 +72,7 @@ def safe_int(value: str | None) -> int | None:
    return int(value) if value.isdigit() else None


-def parse_schedule(html: str) -> list[dict]:
+def parse_schedule(html: str, season: str, source_key: str | None = None) -> list[dict]:
    soup = BeautifulSoup(html, "html.parser")
    matches_data: list[dict] = []

@@ -151,10 +147,11 @@ def parse_schedule(html: str) -> list[dict]:
                    "home_score": home_score,
                    "away_score": away_score,
                    "tour": tour,
-                    "season": SEASON,
+                    "season": season,
                    "place": place,
                    "date_raw": time_site,
                    "score_add": score_add,
+                    "source_key": source_key,
                }
            )
            
@@ -162,12 +159,18 @@ def parse_schedule(html: str) -> list[dict]:
    return matches_data


-def run_parser_schedule() -> None:
-    html = fetch_html(URL_SCHEDULE)
-    matches_data = parse_schedule(html)
+def run_parser_schedule(source_key: str | None = None) -> None:
+    source = get_parser_source(source_key)
+    print(f"[parser_schedule] Источник: {source['title']}")
+    print(f"[parser_schedule] URL: {source['schedule_url']}")
+
+    html = fetch_html(source["schedule_url"])
+    matches_data = parse_schedule(html, source["season"], source["key"])
    if matches_data:
        sync_matches(matches_data)
        print(f"[parser_schedule] Matches synced: {len(matches_data)}")
+    else:
+        print("[parser_schedule] Матчи не найдены")


 if __name__ == "__main__":
--- a/parsers/parser_sources.py
+++ b/parsers/parser_sources.py
@@ -0,0 +1,230 @@
+from __future__ import annotations
+
+from urllib.parse import urljoin
+
+from repositories.project_settings_repository import (
+    DEFAULT_APP_SETTINGS,
+    DEFAULT_PARSER_SOURCES,
+    ensure_project_settings_tables,
+    get_app_setting,
+    get_parser_source_from_db,
+    list_parser_sources_from_db,
+)
+
+
+def _fallback_sources() -> list[dict]:
+    return [dict(item) for item in DEFAULT_PARSER_SOURCES]
+
+
+def _fallback_sources_dict() -> dict[str, dict]:
+    return {item["key"]: dict(item) for item in DEFAULT_PARSER_SOURCES}
+
+
+def _fallback_default_source_key() -> str:
+    return DEFAULT_APP_SETTINGS["default_parser_source_key"]
+
+
+def _fallback_base_url() -> str:
+    return DEFAULT_APP_SETTINGS["rfs_base_url"].rstrip("/")
+
+
+def _get_base_url() -> str:
+    try:
+        ensure_project_settings_tables()
+        return get_app_setting("rfs_base_url", _fallback_base_url()).rstrip("/")
+    except Exception:
+        return _fallback_base_url()
+
+
+RFS_BASE_URL = _get_base_url()
+
+
+def get_rfs_base_url() -> str:
+    return _get_base_url()
+
+
+def get_default_source_key() -> str:
+    try:
+        ensure_project_settings_tables()
+        value = get_app_setting("default_parser_source_key", _fallback_default_source_key())
+        value = (value or _fallback_default_source_key()).upper().strip()
+        known_keys = {source["key"] for source in list_parser_sources()}
+        return value if value in known_keys else _fallback_default_source_key()
+    except Exception:
+        return _fallback_default_source_key()
+
+
+def get_parser_sources() -> dict[str, dict]:
+    try:
+        ensure_project_settings_tables()
+        sources = list_parser_sources_from_db(active_only=True)
+        if not sources:
+            return _fallback_sources_dict()
+        return {source["key"]: source for source in sources}
+    except Exception:
+        return _fallback_sources_dict()
+
+
+def list_parser_sources() -> list[dict]:
+    sources = get_parser_sources()
+    return list(sources.values())
+
+
+def get_parser_source(source_key: str | None = None) -> dict:
+    """Возвращает настройки источника из БД. Если БД недоступна, используется fallback."""
+    source_key = (source_key or get_default_source_key() or "SUPERLEAGUE").upper().strip()
+    try:
+        ensure_project_settings_tables()
+        source = get_parser_source_from_db(source_key)
+        if source:
+            return source
+    except Exception:
+        pass
+
+    fallback = _fallback_sources_dict()
+    if source_key in fallback:
+        return fallback[source_key]
+    raise ValueError(f"Неизвестный источник парсинга: {source_key}")
+
+
+def source_absolute_url(source: dict, path_or_url: str) -> str:
+    value = (path_or_url or "").strip()
+    if value.startswith(("http://", "https://")):
+        return value
+    base_url = (source.get("base_url") or get_rfs_base_url()).rstrip("/")
+    return urljoin(base_url + "/", value.lstrip("/"))
+
+
+def source_match_url(source: dict, match_external_id: str | int) -> str:
+    match_base_url = (source.get("match_base_url") or f"{get_rfs_base_url()}/match/").rstrip("/") + "/"
+    return f"{match_base_url}{str(match_external_id).strip()}"
+
+
+def extract_logo_filename(value: str | None) -> str:
+    """Возвращает только имя файла логотипа из старого полного пути или нового значения."""
+    value = str(value or "").strip().strip('"').strip("'")
+    if not value:
+        return ""
+
+    normalized = value.replace("/", "\\")
+    filename = normalized.split("\\")[-1].strip()
+    return filename
+
+
+def ensure_logo_extension(filename: str) -> str:
+    filename = extract_logo_filename(filename)
+    if not filename:
+        return ""
+    if "." not in filename.rsplit("\\", 1)[-1]:
+        return f"{filename}.png"
+    return filename
+
+
+def get_logo_base_path(source_key: str | None = None) -> str:
+    source = get_parser_source(source_key)
+    return str(source.get("logo_base_path") or "").rstrip("\\/")
+
+
+def build_logo_path(source_key: str | None, filename: str | None) -> str:
+    """Собирает полный путь к логотипу для vMix по источнику турнира."""
+    logo_file = ensure_logo_extension(filename)
+    if not logo_file:
+        return ""
+
+    base_path = get_logo_base_path(source_key)
+    if not base_path:
+        return logo_file
+
+    return f"{base_path}\\{logo_file}"
+
+
+def build_logo_variant_path(
+    source_key: str | None,
+    filename: str | None,
+    variant: str | None = None,
+) -> str:
+    """
+    Собирает путь к варианту логотипа.
+    variant="white" -> Динамо_Белый / Зенит_Белый
+    variant="blue"  -> Динамо_Синий / Зенит_Синий
+    Остальные команды остаются без изменения.
+    """
+    logo_file = ensure_logo_extension(filename)
+    if not logo_file:
+        return ""
+
+    suffix = ""
+    if variant == "white":
+        suffix = "_Белый"
+    elif variant == "blue":
+        suffix = "_Синий"
+
+    if suffix:
+        lower = logo_file.lower()
+        if "динамо" in lower or "зенит" in lower:
+            if "." in logo_file:
+                stem, ext = logo_file.rsplit(".", 1)
+                ext = "." + ext
+            else:
+                stem, ext = logo_file, ".png"
+
+            for old_suffix in ("_Белый", "_Синий"):
+                if stem.endswith(old_suffix):
+                    stem = stem[: -len(old_suffix)]
+            logo_file = f"{stem}{suffix}{ext}"
+
+    return build_logo_path(source_key, logo_file)
+
+
+
+def extract_photo_filename(value: str | None) -> str:
+    """Возвращает имя файла/относительный путь фото из старого полного пути или нового значения."""
+    value = str(value or "").strip().strip('"').strip("'")
+    if not value:
+        return ""
+
+    normalized = value.replace("/", "\\")
+    lower = normalized.lower()
+
+    # Для старых полных путей пытаемся сохранить относительную часть от папки Photo.
+    marker = "\\photo\\"
+    if marker in lower:
+        idx = lower.rfind(marker)
+        return normalized[idx + len(marker):].strip("\\")
+
+    parts = [part for part in normalized.split("\\") if part]
+    if len(parts) >= 2 and ":" in parts[0]:
+        return parts[-1].strip()
+    return normalized.strip("\\")
+
+
+def ensure_photo_extension(filename: str) -> str:
+    filename = extract_photo_filename(filename)
+    if not filename:
+        return ""
+    if "." not in filename.rsplit("\\", 1)[-1]:
+        return f"{filename}.png"
+    return filename
+
+
+def get_photo_base_path(source_key: str | None = None) -> str:
+    source = get_parser_source(source_key)
+    return str(source.get("photo_base_path") or "").rstrip("\\/")
+
+
+def build_photo_path(source_key: str | None, filename: str | None) -> str:
+    """Собирает полный путь к фото игрока для vMix по источнику турнира."""
+    photo_file = ensure_photo_extension(filename)
+    if not photo_file:
+        return ""
+
+    base_path = get_photo_base_path(source_key)
+    if not base_path:
+        return photo_file
+
+    return f"{base_path}\\{photo_file}"
+
+
+def build_empty_photo_path(source_key: str | None = None) -> str:
+    """Путь к пустой картинке фото для выбранного источника."""
+    return build_photo_path(source_key, "EMPTY.png")
--- a/parsers/parser_standings.py
+++ b/parsers/parser_standings.py
@@ -2,9 +2,8 @@ import requests
 from bs4 import BeautifulSoup

 from services.standings_service import sync_standings
+from parsers.parser_sources import get_parser_source

-URL_STANDINGS = "https://wfl.rfs.ru/tournament/1061879/tables"
-SEASON = "2025/2026"


 def fetch_html(url: str) -> str:
@@ -65,16 +64,22 @@ def parse_standings(html: str) -> list[dict]:
    return standings


-def run_parser_standings() -> None:
-    html = fetch_html(URL_STANDINGS)
+def run_parser_standings(source_key: str | None = None) -> None:
+    source = get_parser_source(source_key)
+    print(f"[parser_standings] Источник: {source['title']}")
+    print(f"[parser_standings] URL: {source['standings_url']}")
+
+    html = fetch_html(source["standings_url"])
    standings_rows = parse_standings(html)
    if standings_rows:
        sync_standings(
-            season=SEASON,
+            season=source["season"],
            standings_rows=standings_rows,
        )

        print(f"[parser_standings] Synced rows: {len(standings_rows)}")
+    else:
+        print("[parser_standings] Строки турнирной таблицы не найдены")


 if __name__ == "__main__":
--- a/parsers/parser_teams.py
+++ b/parsers/parser_teams.py
@@ -3,9 +3,7 @@ from bs4 import BeautifulSoup
 from concurrent.futures import ThreadPoolExecutor

 from services.teams_service import sync_teams
-
-
-TEAMS_URL = "https://wfl.rfs.ru/tournament/1061879/teams"
+from parsers.parser_sources import get_parser_source, source_absolute_url


 def fetch_html(url: str) -> str:
@@ -15,45 +13,51 @@ def fetch_html(url: str) -> str:
    return response.text


-def get_links(html) -> list:
+def get_links(html: str, source: dict) -> list[str]:
    soup = BeautifulSoup(html, "html.parser")
-    links: list[dict] = []
-    items = soup.find("ul", class_="teams__list").find_all("li")
-    for i in items:
-        links.append(
-            "https://wfl.rfs.ru/team/"
-            + i.find("a", class_="teams__link").get("href").split("team_id=")[-1]
-        )
+    links: list[str] = []
+
+    teams_list = soup.find("ul", class_="teams__list")
+    if not teams_list:
+        return links
+
+    items = teams_list.find_all("li")
+    for item in items:
+        link_el = item.find("a", class_="teams__link")
+        href = link_el.get("href") if link_el else ""
+        if not href:
+            continue
+
+        if "team_id=" in href:
+            team_external_id = href.split("team_id=")[-1].strip()
+            links.append(source_absolute_url(source, "/team/" + team_external_id))
+        else:
+            links.append(source_absolute_url(source, href))
+
    return links


-def get_url_teams() -> list[dict]:
-    html = fetch_html(TEAMS_URL)
-    links = get_links(html)
+def get_url_teams(source_key: str | None = None) -> list[dict]:
+    source = get_parser_source(source_key)
+    html = fetch_html(source["teams_url"])
+    links = get_links(html, source)
    teams_data: list[dict] = []

    with ThreadPoolExecutor() as pool:
-        responses = [
-            pool.submit(
-                fetch_html,
-                link,
-            )
-            for link in links
-        ]
+        responses = [pool.submit(fetch_html, link) for link in links]
        for result in responses:
            try:
                html = result.result()
                team_data = parse_teams_html(html)
                teams_data.append(team_data)
            except Exception as e:
-                print(f"Error fetching team data: {e}")
+                print(f"[parser_teams] Error fetching team data: {e}")

    return teams_data


 def parse_teams_html(html: str) -> dict:
    soup = BeautifulSoup(html, "html.parser")
-    teams_data: dict = {}
    name = soup.find("a", class_="team-promo__team-name").text.strip()
    external_id = soup.find("a", class_="team-promo__logo").get("href").split("/")[-1]
    stat_info = soup.find("ul", class_="stats-info").find_all(
@@ -65,7 +69,7 @@ def parse_teams_html(html: str) -> dict:
    goals = stat_info[2].text.strip()
    tournaments = stat_info[3].text.strip()

-    teams_data = {
+    return {
        "external_id": str(external_id),
        "name": name,
        "logo_url": logo_url,
@@ -75,14 +79,18 @@ def parse_teams_html(html: str) -> dict:
        "tournaments": tournaments,
    }

-    return teams_data

+def run_parser_teams(source_key: str | None = None) -> None:
+    source = get_parser_source(source_key)
+    print(f"[parser_teams] Источник: {source['title']}")
+    print(f"[parser_teams] URL: {source['teams_url']}")

-def run_parser_teams() -> None:
-    teams_data = get_url_teams()
+    teams_data = get_url_teams(source_key)
    if teams_data:
        sync_teams(teams_data)
-        print(f"Teams synced: {len(teams_data)}")
+        print(f"[parser_teams] Teams synced: {len(teams_data)}")
+    else:
+        print("[parser_teams] Команды не найдены")


 if __name__ == "__main__":