#!/usr/bin/env python3 # -*- coding: utf-8 -*- """Единый HTML-парсер KHL.ru: игроки, тренеры и судьи. Единственный источник данных: https://www.khl.ru/ Примеры: python khl_site_collector.py --kind players python khl_site_collector.py --kind coaches python khl_site_collector.py --kind officials python khl_site_collector.py --kind all Парсер не использует khl.api.webcaster.pro и не требует Selenium. Для устойчивости к антибот-защите по возможности применяется curl_cffi с impersonate="chrome", с requests как резервным HTTP-транспортом. """ from __future__ import annotations import argparse import hashlib import json import logging import math import os import re import threading import time from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from typing import Any, Iterable from urllib.parse import parse_qsl, urlencode, urljoin, urlparse, urlunparse import requests as std_requests from bs4 import BeautifulSoup, Tag from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry try: from curl_cffi import requests as curl_requests except ImportError: # pragma: no cover curl_requests = None LOG = logging.getLogger("khl_site_collector") BASE = "https://www.khl.ru" KINDS = ("players", "coaches", "officials") LIST_URLS = { "players": (f"{BASE}/players/", f"{BASE}/players/season/all/"), "coaches": (f"{BASE}/coaches/", f"{BASE}/coaches/season/all/"), "officials": (f"{BASE}/officials/", f"{BASE}/officials/season/all/", f"{BASE}/officials/stat/"), } DETAIL_RE = { "players": re.compile(r"(?:https?://www\.khl\.ru)?/players/(\d+)/?$", re.I), "coaches": re.compile(r"(?:https?://www\.khl\.ru)?/coaches/(\d+)/?$", re.I), "officials": re.compile(r"(?:https?://www\.khl\.ru)?/officials/(\d+)/?$", re.I), } BAD_MARKERS = ("access denied", "forbidden", "captcha", "проверка браузера", "доступ запрещен") SEASON_RE = re.compile(r"((?:19|20)\d{2})\s*[/–—-]\s*((?:19|20)?\d{2})") STAT_ID_MAP = { # Полевые игроки "и": "gp", "игры": "gp", "ш": "g", "г": "g", "голы": "g", "а": "a", "передачи": "a", "о": "pts", "очки": "pts", "+/-": "pm", "±": "pm", "+": "plus", "-": "minus", "штр": "pim", "штраф": "pim", "штрафы": "pim", "шр": "g_even", "шб": "g_pp", "шм": "g_sh", "шо": "g_ot", "шп": "gwg", "рб": "sds", "бв": "shots", "%бв": "shot_pct", "бв/и": "shots_avg", "вбр": "faceoffs", "ввбр": "faceoff_wins", "%вбр": "faceoff_pct", "вп/и": "toi_avg", "см/и": "sft_avg", "впр/и": "toi_even_avg", "смр/и": "sft_even_avg", "впб/и": "toi_pp_avg", "смб/и": "sft_pp_avg", "впм/и": "toi_sh_avg", "смм/и": "sft_sh_avg", "спр": "hits", "блб": "blocks", "фоп": "fouls_against", "отб": "takeaways", "пхт": "interceptions", # Вратари "в": "w", "п": "l", "иб": "shootout_games", "бр": "shots_against", "пш": "ga", "об": "sv", "%об": "sv_pct", "кн": "gaa", 'и"0"': "so", "и“0”": "so", "и«0»": "so", "вп": "toi", } STAT_HINTS = { "И": "Количество проведённых игр", "Ш": "Заброшенные шайбы", "А": "Передачи", "О": "Очки", "+/-": "Плюс/Минус", "+": "Плюс", "-": "Минус", "Штр": "Штрафное время", "ШР": "Шайбы в равенстве", "ШБ": "Шайбы в большинстве", "ШМ": "Шайбы в меньшинстве", "ШО": "Шайбы в овертайме", "ШП": "Победные шайбы", "РБ": "Решающие буллиты", "БВ": "Броски по воротам", "%БВ": "Процент реализованных бросков", "БВ/И": "Среднее количество бросков по воротам за игру", "Вбр": "Вбрасывания", "ВВбр": "Выигранные вбрасывания", "%Вбр": "Процент выигранных вбрасываний", "ВП/И": "Среднее время на площадке за игру", "См/И": "Среднее количество смен за игру", "ВПР/И": "Среднее время на площадке при игре в равных составах за игру", "СмР/И": "Среднее количество смен при игре в равных составах за игру", "ВПБ/И": "Среднее время на площадке при игре в большинстве за игру", "СмБ/И": "Среднее количество смен при игре в большинстве за игру", "ВПМ/И": "Среднее время на площадке при игре в меньшинстве за игру", "СмМ/И": "Среднее количество смен при игре в меньшинстве за игру", "СПр": "Силовые приёмы", "БлБ": "Блокированные броски", "ФоП": "Фолы против", "ОТБ": "Отборы шайбы", "ПХТ": "Перехваты передач", "В": "Выигрыши", "П": "Проигрыши", "ИБ": "Игры с буллитными сериями", "Бр": "Броски", "ПШ": "Пропущено шайб", "ОБ": "Отражённые броски", "%ОБ": "Процент отражённых бросков", "КН": "Коэффициент надёжности = 60 мин × ПШ / ВП", 'И"0"': "Сухие игры", "ВП": "Время на площадке", } PROFILE_LABELS = { "дата рождения": "birthday", "возраст": "age", "гражданство": "country", "страна": "country", "сборная": "national_team", "контракт до": "contract_until", "место рождения": "birth_place", "рост": "height", "вес": "weight", "хват": "stick", "амплуа": "role", "позиция": "role", "номер": "shirt_number", } def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def clean(value: Any) -> str: return re.sub(r"\s+", " ", str(value or "")).strip() def normalize_season(value: str) -> str: text = clean(value) match = SEASON_RE.search(text) if match: left, right = match.groups() if len(right) == 2: right = left[:2] + right return f"{left}/{right}" short = re.search(r"(? None: path.parent.mkdir(parents=True, exist_ok=True) tmp = path.with_name(f".{path.name}.{os.getpid()}.{time.time_ns()}.tmp") try: tmp.write_text(json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n", encoding="utf-8") os.replace(tmp, path) finally: try: tmp.unlink(missing_ok=True) except OSError: pass def url_with_page(url: str, page: int) -> str: if page <= 1: return url parsed = urlparse(url) query = dict(parse_qsl(parsed.query, keep_blank_values=True)) query["PAGEN_1"] = str(page) return urlunparse((parsed.scheme, parsed.netloc, parsed.path, parsed.params, urlencode(query), "")) def stat_id(label: str, index: int) -> str: key = clean(label).casefold().replace("ё", "е") if key in STAT_ID_MAP: return STAT_ID_MAP[key] slug = re.sub(r"[^a-zа-я0-9%+/-]+", "_", key, flags=re.I).strip("_") return f"khl_{slug or index}" def numeric(value: str) -> Any: text = clean(value).replace(" ", " ") if not text or text in {"—", "-"}: return None # Временные значения оставляем строкой. if re.fullmatch(r"\d{1,3}:\d{2}", text): mm, ss = map(int, text.split(":")) return mm + ss / 60.0 normalized = text.replace("%", "").replace(",", ".") if re.fullmatch(r"[+-]?\d+(?:\.\d+)?", normalized): num = float(normalized) return int(num) if num.is_integer() else num return text def _image_url(node: Tag) -> str: src = clean(node.get("data-src") or node.get("data-original") or node.get("src")) return urljoin(BASE, src) if src and not src.startswith("data:") else "" def _generic_image(url: str, alt: str = "") -> bool: text = f"{url} {alt}".casefold() bad = ( "logo", "logotype", "sprite", "icon", "favicon", "banner", "advert", "sponsor", "placeholder", "default-avatar", "default_avatar", "khl-logo", "logo_khl", "/flags/", "flag_", "/clubs/", "/teams/", ) return any(token in text for token in bad) def first_image(soup: BeautifulSoup, kind: str, person_id: int, person_name: str = "") -> str: """Возвращает именно портрет человека, не логотип КХЛ/клуба. На KHL.ru изображения могут приезжать с CDN и без ID в URL, поэтому используем несколько сигналов: alt, близость к имени, размеры, путь и og:image. """ name_tokens = [x.casefold() for x in re.findall(r"[A-Za-zА-Яа-яЁё-]{3,}", person_name)[:3]] candidates: list[tuple[int, str]] = [] def add(url: str, score: int, alt: str = "") -> None: if not url or _generic_image(url, alt): return low = f"{url} {alt}".casefold() if str(person_id) in low: score += 90 if f"/{kind}/" in low or "/persons/" in low or "/people/" in low: score += 35 if any(token in low for token in ("portrait", "person", "player", "coach", "official", "referee", "photo")): score += 20 if name_tokens and sum(1 for token in name_tokens if token in low) >= min(2, len(name_tokens)): score += 80 candidates.append((score, url)) # Соц.мета часто содержит крупное фото карточки, но логотипы отсеиваются выше. for prop in ("og:image", "twitter:image"): meta = soup.find("meta", attrs={"property": prop}) or soup.find("meta", attrs={"name": prop}) if meta: add(urljoin(BASE, clean(meta.get("content"))), 25) for img in soup.find_all("img"): url = _image_url(img) alt = clean(img.get("alt") or img.get("title")) if not url: continue score = 0 try: width = int(re.sub(r"\D", "", str(img.get("width") or "0")) or 0) height = int(re.sub(r"\D", "", str(img.get("height") or "0")) or 0) if width >= 120 and height >= 120: score += 25 if width and height and 0.6 <= width / height <= 1.4: score += 8 except Exception: pass parent_text = clean(img.parent.get_text(" ", strip=True)) if isinstance(img.parent, Tag) else "" if person_name and person_name.casefold() in parent_text.casefold(): score += 60 add(url, score, alt) if not candidates: return "" candidates.sort(key=lambda x: x[0], reverse=True) # Для персонала не подставляем случайное изображение при слабой уверенности. if kind in {"coaches", "officials"} and candidates[0][0] < 30: return "" return candidates[0][1] def _find_profile_anchor(values: list[str], person_name: str) -> int | None: """Находит начало персонального блока, не путая его с меню/фильтрами KHL.""" wanted = clean(person_name).casefold() if not wanted: return None for i, value in enumerate(values[:900]): if clean(value).casefold() == wanted: return i # Иногда инициалы/пробелы в DOM слегка отличаются — допускаем совпадение всех слов. tokens = [x.casefold().strip(".,") for x in clean(person_name).split() if len(x.strip(".,")) > 1] if tokens: for i, value in enumerate(values[:900]): low = clean(value).casefold() if all(token in low for token in tokens): return i return None def extract_profile_fields(soup: BeautifulSoup, kind: str = "", person_name: str = "") -> dict[str, Any]: all_values = [clean(x) for x in soup.stripped_strings if clean(x)] fields: dict[str, Any] = {} labels = set(PROFILE_LABELS) # Критично: профиль ищем только рядом с ФИО. На общей странице выше есть фильтры # «Амплуа», «Гражданство» и пункты меню «Игрок», «Конференция», которые раньше # ошибочно попадали в карточку. anchor = _find_profile_anchor(all_values, person_name) if anchor is not None: end = min(len(all_values), anchor + 90) for i in range(anchor + 1, min(len(all_values), anchor + 90)): if all_values[i].casefold() in {"статистика", "матчи", "новости"}: end = i break values = all_values[anchor:end] else: values = all_values for i, value in enumerate(values): label = value.casefold().strip(" :") target = PROFILE_LABELS.get(label) if not target or target in fields: continue for candidate in values[i + 1:i + 7]: c = clean(candidate) if not c or c.casefold().strip(" :") in labels: continue if len(c) > 120: break low = c.casefold().strip() if target == "role" and low in {"игрок", "игроки", "player", "players"}: continue if target in {"age", "height", "weight", "shirt_number"}: match = re.search(r"\b(\d{1,3})\b", c) if not match: continue fields[target] = int(match.group(1)) else: fields[target] = c break if kind == "players": allowed_roles = {"вратарь", "защитник", "нападающий"} role = clean(fields.get("role")).casefold() if role not in allowed_roles: fields.pop("role", None) if not fields.get("role"): # На текущей карточке KHL амплуа идёт сразу после RU/EN ФИО. search_values = values[:28] if anchor is not None else values[:160] for value in search_values: low = clean(value).casefold().strip() if low in allowed_roles: fields["role"] = low break elif not fields.get("role"): allowed = {"главный", "линейный", "главный тренер", "тренер", "судья", "арбитр"} for value in values[:120]: low = clean(value).casefold().strip() if low in allowed: fields["role"] = value break if not fields.get("shirt_number"): for value in values[:80]: match = re.search(r"№\s*(\d{1,3})", value) if match: fields["shirt_number"] = int(match.group(1)) break stick = clean(fields.get("stick")).casefold() if stick in {"l", "left", "лев", "левый"}: fields["stick"] = "левый" elif stick in {"r", "right", "прав", "правый"}: fields["stick"] = "правый" return fields def _title_person_name(soup: BeautifulSoup, kind: str) -> str: """ФИО из , только если title действительно относится к одной персоне.""" if not soup.title: return "" title = clean(soup.title.get_text(" ", strip=True)) markers = { "players": r"(?:хоккеист|игрок)", "coaches": r"тренер", "officials": r"судья", } match = re.search(rf"^(.+?),\s*{markers[kind]}\b", title, flags=re.I) if not match: return "" candidate = clean(match.group(1)) return candidate if 1 < len(candidate) < 120 else "" def _reorder_khl_title_name(raw: str) -> str: """Title KHL: «Фамилия Имя [инициал]» -> «Имя [инициал] Фамилия».""" parts = clean(raw).split() if len(parts) < 2: return clean(raw) return " ".join(parts[1:] + parts[:1]) def _looks_like_ru_person_name(value: str) -> bool: value = clean(value) if not (3 <= len(value) <= 90): return False low = value.casefold() blocked_fragments = ( "континентальная", "хоккейная лига", "конференц", "дивизион", "игроки", "тренеры", "судьи", "статистика", "матчи", "новости", "фильтры", "дата рождения", "гражданство", "сборная", "контракт", "амплуа", "рост", "вес", "хват", "сезон", "турнир", "команда", "клуб", ) if any(x in low for x in blocked_fragments): return False # 2–4 слова на кириллице; допускаем инициалы «В.». return bool(re.fullmatch(r"[А-ЯЁ][А-Яа-яЁё'’-]+(?:\s+(?:[А-ЯЁ][А-Яа-яЁё'’-]+|[А-ЯЁ][А-Яа-яЁё]{0,3}\.)){1,3}", value)) def _looks_like_en_person_name(value: str) -> bool: value = clean(value) if not (3 <= len(value) <= 90): return False blocked = { "khl", "fonbet khl", "conference", "statistics", "stats", "matches", "news", "subscribe", "image", "players", "coaches", "officials", "kontinental hockey league", } if value.casefold() in blocked: return False return bool(re.fullmatch(r"[A-Za-z][A-Za-z'’-]*(?:\s+(?:[A-Za-z][A-Za-z'’-]*|[A-Za-z]\.)){1,4}", value)) def _metadata_person_name(soup: BeautifulSoup, kind: str) -> str: """Try structured metadata before visible global navigation text.""" candidates: list[str] = [] for attrs in ( {"property": "og:title"}, {"name": "twitter:title"}, {"name": "title"}, ): node = soup.find("meta", attrs=attrs) if node and node.get("content"): candidates.append(clean(node.get("content"))) for script in soup.find_all("script", attrs={"type": "application/ld+json"}, limit=20): raw = script.string or script.get_text(" ", strip=True) if not raw: continue try: payload = json.loads(raw) except Exception: continue stack = payload if isinstance(payload, list) else [payload] while stack: obj = stack.pop() if isinstance(obj, list): stack.extend(obj) continue if not isinstance(obj, dict): continue typ = str(obj.get("@type") or "").casefold() if typ in {"person", "athlete"} and obj.get("name"): candidates.insert(0, clean(obj.get("name"))) for value in obj.values(): if isinstance(value, (dict, list)): stack.append(value) marker_re = {"players": r"(?:хоккеист|игрок)", "coaches": r"тренер", "officials": r"(?:судья|арбитр)"}[kind] for raw in candidates: value = re.split(r"\s*[|—–]\s*", raw, maxsplit=1)[0].strip() value = re.split(rf",\s*{marker_re}\b", value, maxsplit=1, flags=re.I)[0].strip() if _looks_like_ru_person_name(value): return value return "" def _profile_role_index(strings: list[str], kind: str) -> int | None: role_patterns = { "players": ("вратарь", "защитник", "нападающий"), "coaches": ("тренер",), "officials": ("судья", "арбитр"), } for i, value in enumerate(strings[:360]): low = value.casefold() if any(role in low for role in role_patterns[kind]): return i for i, value in enumerate(strings[:360]): if value.casefold() in {"дата рождения", "гражданство", "возраст", "рост", "вес", "хват"}: return i return None def extract_names(soup: BeautifulSoup, kind: str) -> tuple[str, str]: strings = [clean(x) for x in soup.stripped_strings if clean(x)] raw_title_name = _title_person_name(soup, kind) metadata_name = _metadata_person_name(soup, kind) expected_source = raw_title_name or metadata_name expected_tokens = [x.casefold().strip(".,") for x in expected_source.split()[:2] if len(x) > 1] name = "" # Для игроков персональный <title> сейчас стабилен: «Фамилия Имя И., хоккеист…». # Используем его как источник истины и переставляем фамилию в конец. Это исключает # попадание «Конференция «Запад»» из глобального меню страницы. if raw_title_name and kind == "players": name = _reorder_khl_title_name(raw_title_name) elif metadata_name and kind == "players": name = _reorder_khl_title_name(metadata_name) elif metadata_name: name = metadata_name # Для остальных случаев ищем точное ФИО внутри профильного блока. if not name and expected_tokens: for candidate in strings[:320]: low = candidate.casefold() if all(token in low for token in expected_tokens) and _looks_like_ru_person_name(candidate): name = candidate break # 2. Не зависим от title: на карточке ФИО находится перед амплуа/персональными полями. marker = _profile_role_index(strings, kind) if not name and marker is not None: candidates = [strings[idx] for idx in range(max(0, marker - 18), marker) if _looks_like_ru_person_name(strings[idx])] if candidates: name = candidates[-1] # 3. Резерв по типовым контейнерам имени. if not name: for node in soup.find_all(["h1", "h2", "h3", "div", "span"], limit=900): classes = " ".join(node.get("class") or []).casefold() if not any(k in classes for k in ("name", "title", "person", "player", "coach", "official")): continue candidate = clean(node.get_text(" ", strip=True)) if _looks_like_ru_person_name(candidate): name = candidate break # 4. Безопасный резерв из валидного персонального title. if not name and raw_title_name: name = _reorder_khl_title_name(raw_title_name) if kind == "players" else raw_title_name name = re.sub(r"\s+№\s*\d+.*$", "", clean(name)).strip() # EN ФИО ищем в том же профильном блоке, рядом с RU ФИО/амплуа. name_en = "" anchors: list[int] = [] if name: for i, value in enumerate(strings[:360]): if value == name: anchors.append(i) break if marker is not None: anchors.append(marker) pool: list[str] = [] for anchor in anchors: pool.extend(strings[max(0, anchor - 8):min(len(strings), anchor + 12)]) pool.extend(strings[:260]) seen: set[str] = set() for candidate in pool: if candidate in seen: continue seen.add(candidate) if _looks_like_en_person_name(candidate): name_en = candidate break return name, name_en def parse_table(table: Tag) -> dict[str, Any] | None: headers: list[str] = [] for tr in table.select("thead tr"): cells = tr.find_all(["th", "td"], recursive=False) if cells: candidate = [clean(x.get_text(" ", strip=True)) for x in cells] if len(candidate) >= len(headers): headers = candidate if not headers: first = table.find("tr") if first: cells = first.find_all(["th", "td"], recursive=False) if cells and any(c.name == "th" for c in cells): headers = [clean(x.get_text(" ", strip=True)) for x in cells] rows: list[dict[str, Any]] = [] section = "" for tr in table.find_all("tr"): cells = tr.find_all(["th", "td"], recursive=False) if not cells: continue values = [clean(x.get_text(" ", strip=True)) for x in cells] if not any(values): continue if headers and values == headers: continue if len(cells) == 1 or (len(cells) <= 2 and cells[0].get("colspan")): candidate = clean(values[0]) if candidate: section = candidate continue links = [urljoin(BASE, str(a.get("href"))) for a in tr.find_all("a", href=True)] images = [] for img in tr.find_all("img"): src = _image_url(img) low = f"{src} {clean(img.get('alt') or '')}".casefold() if src and not any(token in low for token in ("sprite", "icon", "banner", "advert", "sponsor", "/flags/", "flag_")): images.append(src) link_items = [] for a in tr.find_all("a", href=True): href = urljoin(BASE, str(a.get("href"))) title = clean(a.get_text(" ", strip=True)) image = "" img = a.find("img") if img: image = _image_url(img) low = f"{image} {clean(img.get('alt') or '')}".casefold() if image and any(token in low for token in ("sprite", "icon", "banner", "advert", "sponsor", "/flags/", "flag_")): image = "" link_items.append({"title": title, "url": href, "image": image}) rows.append({"values": values, "section": section, "links": list(dict.fromkeys(links)), "images": list(dict.fromkeys(images)), "link_items": link_items}) if not rows: return None caption = clean(table.find("caption").get_text(" ", strip=True)) if table.find("caption") else "" return {"headers": headers, "rows": rows, "caption": caption} def parse_all_tables(soup: BeautifulSoup) -> list[dict[str, Any]]: out: list[dict[str, Any]] = [] for table in soup.find_all("table"): parsed = parse_table(table) if parsed: out.append(parsed) return out def table_is_matches(table: dict[str, Any]) -> bool: heads = [clean(x).casefold().replace("ё", "е") for x in table.get("headers") or []] return any(x == "дата" or x.startswith("дата ") for x in heads) and any("команд" in x for x in heads) and any("счет" in x for x in heads) def table_is_player_stats(table: dict[str, Any]) -> bool: heads = [clean(x).casefold() for x in table.get("headers") or []] compact = set(heads) return ("и" in compact or any("количество проведенных игр" in x for x in heads)) and ("ш" in compact or "а" in compact or any(x in compact for x in ("в", "п", "пш", "%об"))) def split_tournament_team(value: str, section: str) -> tuple[str, str, str]: text = clean(value) section_text = clean(section) season = normalize_season(section_text) or normalize_season(text) combined = f"{section_text} {text}".casefold().replace("ё", "е") if "плей" in combined: stage = "Плей-офф" elif "надежд" in combined: stage = "Кубок Надежды" elif "рег." in combined or "регуляр" in combined: stage = "Регулярный чемпионат" else: stage = re.sub(r"(?:19|20)?\d{2}\s*[/–—-]\s*(?:19|20)?\d{2}", "", section_text).strip(" |·—-") or "КХЛ" team = text # Если сезон/этап и команда оказались в одной ячейке, вырезаем служебную часть. if season: cleaned = SEASON_RE.sub("", text) cleaned = re.sub(r"(?<!\d)\d{2}\s*[/–—-]\s*\d{2}(?!\d)", "", cleaned) for token in ("рег.чемпионат", "регулярный чемпионат", "плей-офф", "кубок надежды"): cleaned = re.sub(re.escape(token), "", cleaned, flags=re.I) candidate = cleaned.strip(" |·—-") if candidate: team = candidate return season, clean(stage), clean(team) def stable_stage_id(season: str, stage: str, team: str, row_index: int) -> int: raw = f"{season}|{stage}|{team}|{row_index}".encode("utf-8") return 1_000_000 + int(hashlib.sha1(raw).hexdigest()[:8], 16) % 800_000_000 def _stat_description(label: str) -> str: return STAT_HINTS.get(clean(label), clean(label)) def _row_team_media(row: dict[str, Any], team_name: str) -> tuple[str, str]: team_url = "" logo = "" for item in row.get("link_items") or []: path = urlparse(str(item.get("url") or "")).path.casefold() title = clean(item.get("title")) if any(token in path for token in ("/clubs/", "/club/", "/teams/", "/team/")): if not team_url: team_url = str(item.get("url") or "") if item.get("image") and not logo: logo = str(item["image"]) if title and team_name and team_name.casefold().split(" (")[0] in title.casefold() and item.get("image"): logo = str(item["image"]) team_url = str(item.get("url") or team_url) break if not logo: for image in row.get("images") or []: if image and not _generic_image(image): logo = image break return team_url, logo def _stats_from_row(headers: list[str], vals: list[str]) -> tuple[int | None, list[dict[str, Any]]]: offset = max(0, len(headers) - len(vals)) aligned_headers = ([""] * offset + headers)[-len(vals):] if headers else [f"Колонка {i+1}" for i in range(len(vals))] stats: list[dict[str, Any]] = [] shirt_number = None for ci, (header, value) in enumerate(zip(aligned_headers, vals)): label = clean(header) if ci == 0: continue if label in {"№", "N", "#"}: parsed = numeric(value) if isinstance(parsed, (int, float)): shirt_number = int(parsed) continue sid = stat_id(label or f"c{ci}", ci) val = numeric(value) if val is None: continue stats.append({"id": sid, "title": label or f"Колонка {ci+1}", "description": _stat_description(label), "val": val}) return shirt_number, stats def _summary_key(label: str) -> str: low = clean(label).casefold().replace("ё", "е").strip(" :") if low.startswith("регуляр"): return "regular" if "плей" in low: return "playoff" if "надежд" in low: return "hope" if low.startswith("всего в кхл") or low == "всего": return "all" return "" def player_from_profile(player_id: int, url: str, html_text: str) -> tuple[dict[str, Any], list[dict[str, Any]]]: soup = BeautifulSoup(html_text, "html.parser") name, name_en = extract_names(soup, "players") fields = extract_profile_fields(soup, "players", name) tables = parse_all_tables(soup) photo = first_image(soup, "players", player_id, name) profile: dict[str, Any] = { "id": player_id, "khl_id": player_id, "name": name, "name_en": name_en, "image": photo, "source_url": url, } profile.update(fields) top_strings = [clean(x) for x in soup.stripped_strings][:220] role_text = clean(profile.get("role")) if role_text: for idx, value in enumerate(top_strings): if value.casefold() == role_text.casefold() and idx > 0: for candidate in reversed(top_strings[max(0, idx - 5):idx]): low = candidate.casefold() if candidate in {name, name_en, "Image"} or "находится в" in low or "игрок" in low: continue if len(candidate) <= 80 and not re.search(r"\d", candidate) and candidate.casefold() != "конференция": profile["team"] = {"name": candidate} break break stages: dict[str, Any] = {} stage_meta: list[dict[str, Any]] = [] teams: dict[str, dict[str, Any]] = {} site_aggregates: dict[str, Any] = {} stats_tables = [t for t in tables if table_is_player_stats(t)] for table_idx, table in enumerate(stats_tables): headers = list(table.get("headers") or []) rows = table.get("rows") or [] for row_idx, row in enumerate(rows): vals = list(row.get("values") or []) if len(vals) < 2: continue first = clean(vals[0]) low_first = first.casefold().replace("ё", "е") # Строки отдельных матчей находятся на KHL в той же широкой таблице, # но не являются сезонными строками статистики. if re.fullmatch(r"(?:\d{4}-\d{2}-\d{2}|\d{1,2}\s+[А-Яа-яЁё]{3,}\s+\d{4}|\d{1,2}\.\d{1,2}\.\d{4})", first): if len(vals) >= 3 and re.search(r"\b\d{1,2}\s*:\s*\d{1,2}\b", clean(vals[2])): continue # Суммарная статистика KHL — сохраняем как есть, со всеми колонками сайта. summary_key = _summary_key(first) if summary_key and ("суммар" in clean(row.get("section")).casefold() or not normalize_season(first)): _shirt, summary_stats = _stats_from_row(headers, vals) if summary_stats: site_aggregates[summary_key] = {"stage_count": 0, "stats": summary_stats, "source": "khl.ru"} continue if low_first.startswith("всего") or "суммарная статистика" in low_first: continue season, stage_title, team_name = split_tournament_team(first, row.get("section") or "") if not season: continue if not team_name: team_name = "—" shirt_number, stats = _stats_from_row(headers, vals) if not stats: continue team_url, team_logo = _row_team_media(row, team_name) sid = stable_stage_id(season, stage_title, team_name, table_idx * 1000 + row_idx) role = clean(fields.get("role")) team_obj = {"name": team_name} if team_logo: team_obj["image"] = team_logo if team_url: team_obj["url"] = team_url data = { "role": role, "shirt_number": shirt_number, "team": team_obj, "teams": [{**team_obj, "seasons": season}], "stats": stats, } stages[str(sid)] = {"data": data} stage_meta.append({"id": sid, "season": season, "title": stage_title or "КХЛ", "source_url": url}) team_key = team_name.casefold() team_item = teams.setdefault(team_key, {"name": team_name, "seasons": set(), "image": "", "url": ""}) team_item["seasons"].add(season) if team_logo and not team_item["image"]: team_item["image"] = team_logo if team_url and not team_item["url"]: team_item["url"] = team_url if teams: profile["teams"] = [ {k: v for k, v in { "name": item["name"], "seasons": ", ".join(sorted(item["seasons"], reverse=True)), "image": item.get("image") or "", "url": item.get("url") or "", }.items() if v} for item in sorted(teams.values(), key=lambda x: x["name"].casefold()) ] profile["seasons_count"] = {"khl": len({m["season"] for m in stage_meta if m.get("season")})} if stage_meta: latest = sorted(stage_meta, key=lambda x: (x.get("season") or "", x["id"]), reverse=True)[0] latest_record = stages.get(str(latest["id"]), {}).get("data") or {} if latest_record.get("team"): profile["team"] = latest_record["team"] if latest_record.get("shirt_number") is not None: profile["shirt_number"] = latest_record["shirt_number"] # Полное имя — источник истины. Разбивка нужна только для поиска/совместимости. parts = name.split() identity = {"id": player_id, "khl_id": player_id, "full_name": name, "first_name": parts[0] if parts else "", "last_name": parts[-1] if len(parts) > 1 else ""} player: dict[str, Any] = {"identity": identity, "profile": profile, "stages": stages, "site_aggregates": site_aggregates} if name_en: en_parts = name_en.split() player["identity_en"] = {"id": player_id, "khl_id": player_id, "full_name": name_en, "first_name": en_parts[0] if en_parts else "", "last_name": en_parts[-1] if len(en_parts) > 1 else ""} return player, stage_meta def staff_from_profile(kind: str, person_id: int, url: str, html_text: str) -> dict[str, Any]: soup = BeautifulSoup(html_text, "html.parser") name, name_en = extract_names(soup, kind) # Title карточек персонала KHL стабильно начинается с «Фамилия Имя [Отчество], тренер/судья». # Храним именно этот исходный порядок, а интерфейс уже показывает Имя + Фамилия. if soup.title: title = clean(soup.title.get_text(" ", strip=True)) marker = "тренер" if kind == "coaches" else "судья" candidate = re.split(rf",\s*{marker}\b", title, maxsplit=1, flags=re.I)[0].strip() if candidate and candidate.casefold() not in {"тренеры", "судьи"} and len(candidate) < 120: name = candidate fields = extract_profile_fields(soup, kind, name) photo = first_image(soup, kind, person_id, name) tables = parse_all_tables(soup) seasons = sorted({s for t in soup.stripped_strings if (s := normalize_season(clean(t)))}, reverse=True) role = clean(fields.get("role")) appearances: list[dict[str, Any]] = [] for season in seasons: appearances.append({ "id": person_id, "language": "ru", "url": url, "name": name, "text": name, "image": photo, "team_links": [], "context": {"season": season, "role": role, "language": "ru", "page_url": url}, }) if not appearances: appearances.append({ "id": person_id, "language": "ru", "url": url, "name": name, "text": name, "image": photo, "team_links": [], "context": {"role": role, "language": "ru", "page_url": url}, }) parts = name.split() ru_last = parts[0] if parts else "" ru_first = parts[1] if len(parts) > 1 else "" ru_middle = " ".join(parts[2:]) if len(parts) > 2 else "" en_parts = name_en.split() en_first = en_parts[0] if en_parts else "" en_last = en_parts[-1] if len(en_parts) > 1 else "" field_labels = {"birthday": "Дата рождения", "age": "Возраст", "country": "Гражданство", "height": "Рост", "weight": "Вес", "stick": "Хват", "role": "Амплуа", "shirt_number": "Номер"} display_fields = {field_labels.get(k, k): v for k, v in fields.items()} profile = { "id": person_id, "language": "ru", "url": url, "name": name, "name_en": name_en, "name_parts": {"full": name, "first_name": ru_first, "last_name": ru_last, "middle_name": ru_middle, "parse_quality": "khl_title"}, "page_title": clean(soup.title.get_text(" ", strip=True)) if soup.title else "", "photos": [photo] if photo else [], "fields": display_fields, "seasons": seasons, "teams": [], "events": [], "news": [], "links": [], "tables": tables, "sections": [], "html_bytes": len(html_text.encode("utf-8")), } for anchor in soup.find_all("a", href=True): href = urljoin(BASE, str(anchor.get("href"))) title = clean(anchor.get_text(" ", strip=True)) item = {"title": title, "url": href} path = urlparse(href).path.casefold() if "/clubs/" in path or "/teams/" in path: profile["teams"].append(item) if any(x in path for x in ("/game/", "/calendar/", "/text/")): profile["events"].append(item) if "/news/" in path: profile["news"].append(item) profile["links"].append(item) for key in ("teams", "events", "news", "links"): seen = set(); unique = [] for item in profile[key]: if item["url"] in seen: continue seen.add(item["url"]); unique.append(item) profile[key] = unique for appearance in appearances: appearance["team_links"] = list(profile["teams"]) return { "id": person_id, "kind": kind, "url": url, "url_ru": url, "name": name, "name_ru": name, "name_en": name_en, "last_name_ru": ru_last, "first_name_ru": ru_first, "middle_name_ru": ru_middle, "last_name_en": en_last, "first_name_en": en_first, "middle_name_en": "", "photos": [photo] if photo else [], "appearances": appearances, "profile": profile, "languages": {"ru": profile}, } @dataclass(slots=True) class Config: output_dir: Path cache_dir: Path workers: int timeout: float retries: int delay: float max_pages: int max_people: int force: bool official_scan_min: int official_scan_max: int current_only: bool class HttpClient: def __init__(self, config: Config) -> None: self.config = config self.local = threading.local() self.cache_lock = threading.Lock() self.config.cache_dir.mkdir(parents=True, exist_ok=True) @staticmethod def headers() -> dict[str, str]: return { "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.6", "Cache-Control": "no-cache", "Pragma": "no-cache", "Referer": f"{BASE}/", "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/150.0.0.0 Safari/537.36", } def session(self): session = getattr(self.local, "session", None) if session is not None: return session if curl_requests is not None: session = curl_requests.Session(impersonate="chrome") session.headers.update(self.headers()) else: session = std_requests.Session() retry = Retry(total=self.config.retries, connect=self.config.retries, read=self.config.retries, status=self.config.retries, backoff_factor=0.8, status_forcelist=(408, 425, 429, 500, 502, 503, 504), allowed_methods=frozenset({"GET"}), raise_on_status=False) adapter = HTTPAdapter(max_retries=retry, pool_connections=max(8, self.config.workers * 2), pool_maxsize=max(8, self.config.workers * 2)) session.mount("https://", adapter); session.mount("http://", adapter) session.headers.update(self.headers()) self.local.session = session return session def cache_path(self, url: str) -> Path: return self.config.cache_dir / f"{hashlib.sha256(url.encode()).hexdigest()}.html" @staticmethod def valid_html(text: str, kind: str | None = None) -> bool: low = text[:12000].casefold() if len(text) < 600 or "<html" not in low or any(x in low for x in BAD_MARKERS): return False if kind and kind not in low and "khl" not in low: return False return True def fetch(self, url: str, *, use_cache: bool = True) -> str: cache = self.cache_path(url) if use_cache and not self.config.force and cache.exists(): text = cache.read_text(encoding="utf-8", errors="ignore") if self.valid_html(text): return text if self.config.delay: time.sleep(self.config.delay) response = self.session().get(url, timeout=self.config.timeout, allow_redirects=True) status = int(getattr(response, "status_code", 0)) if status >= 400: raise RuntimeError(f"HTTP {status}: {url}") text = response.text if not self.valid_html(text): raise RuntimeError(f"KHL.ru вернул невалидную HTML-страницу: {url}") cache.parent.mkdir(parents=True, exist_ok=True) with self.cache_lock: cache.write_text(text, encoding="utf-8") return text def detail_links(kind: str, html_text: str) -> dict[int, str]: soup = BeautifulSoup(html_text, "html.parser") found: dict[int, str] = {} pattern = DETAIL_RE[kind] for a in soup.find_all("a", href=True): url = urljoin(BASE, str(a.get("href"))) parsed = urlparse(url) canonical = urlunparse((parsed.scheme, parsed.netloc, parsed.path, "", "", "")) match = pattern.match(canonical) if match: found[int(match.group(1))] = canonical if canonical.endswith("/") else canonical + "/" return found def _strip_list_label(value: str, *labels: str) -> str: text = clean(value) for label in labels: text = re.sub(rf"^\s*{re.escape(label)}\s*[:\-]?\s*", "", text, flags=re.I) return clean(text) def _normalize_country(value: str) -> tuple[str, str]: """Возвращает (название страны, двухбуквенный код если он присутствует).""" text = _strip_list_label(value, "Гражданство", "Страна") match = re.match(r"^([A-Z]{2})\s+(.+)$", text) if match: return clean(match.group(2)), match.group(1) country_codes = { "россия": "RU", "беларусь": "BY", "казахстан": "KZ", "финляндия": "FI", "швеция": "SE", "чехия": "CZ", "словакия": "SK", "латвия": "LV", "литва": "LT", "эстония": "EE", "сша": "US", "канада": "CA", "германия": "DE", "швейцария": "CH", "австрия": "AT", "дания": "DK", "норвегия": "NO", "франция": "FR", "словения": "SI", "хорватия": "HR", "польша": "PL", "венгрия": "HU", "китай": "CN", "япония": "JP", "южная корея": "KR", "великобритания": "GB", "италия": "IT", "украина": "UA", } return text, country_codes.get(text.casefold(), "") def player_list_hints(html_text: str) -> dict[int, dict[str, Any]]: """Структурированные данные из /players/. Список игроков KHL гораздо надёжнее для ФИО и амплуа, чем глобальный текст персональной страницы: в нём есть отдельные колонки «Фамилия, имя», «Клуб», «Амплуа», «Дата рождения», «Возраст», «Гражданство» и т.д. Эти значения используем как источник истины для верхней карточки, а персональную страницу — для расширенной статистики и EN-имени. """ soup = BeautifulSoup(html_text, "html.parser") hints: dict[int, dict[str, Any]] = {} def player_id_from_row(row: dict[str, Any]) -> int | None: for raw_url in row.get("links") or []: parsed = urlparse(str(raw_url)) canonical = urlunparse((parsed.scheme, parsed.netloc, parsed.path, "", "", "")) match = DETAIL_RE["players"].match(canonical) if match: return int(match.group(1)) return None for table in parse_all_tables(soup): headers = [clean(x) for x in table.get("headers") or []] header_keys = [x.casefold().replace("ё", "е") for x in headers] for row in table.get("rows") or []: pid = player_id_from_row(row) if pid is None: continue values = [clean(x) for x in row.get("values") or []] if not values: continue # Responsive-таблицы иногда имеют больше заголовков, чем видимых td. offset = max(0, len(headers) - len(values)) aligned = ([""] * offset + headers)[-len(values):] if headers else [""] * len(values) data: dict[str, Any] = {"source": "players_list"} for header, value in zip(aligned, values): hk = clean(header).casefold().replace("ё", "е") if not value: continue if "фамил" in hk or ("имя" in hk and "команд" not in hk): candidate = _strip_list_label(value, "Фамилия, имя", "Игрок") if _looks_like_ru_person_name(candidate): data["name_raw"] = candidate elif "клуб" in hk or "команд" in hk: candidate = _strip_list_label(value, "Клуб", "Команда") if candidate and not any(x in candidate.casefold() for x in ("конференц", "дивизион")): data["team"] = candidate elif "амплуа" in hk or "позици" in hk: low = _strip_list_label(value, "Амплуа", "Позиция").casefold() for role in ("вратарь", "защитник", "нападающий"): if role in low: data["role"] = role break elif "дата рож" in hk: candidate = _strip_list_label(value, "Дата рождения") if candidate: data["birthday"] = candidate elif hk == "возраст" or "возраст" in hk: m = re.search(r"\b(\d{1,3})\b", _strip_list_label(value, "Возраст")) if m: data["age"] = int(m.group(1)) elif "граждан" in hk or hk == "страна": country, code = _normalize_country(value) if country: data["country"] = country if code: data["country_code"] = code elif "сборн" in hk: candidate = _strip_list_label(value, "Сборная") if candidate: data["national_team"] = candidate elif "контракт" in hk: candidate = _strip_list_label(value, "Контракт до") if candidate: data["contract_until"] = candidate # Fallback по содержимому строки — работает и при нестандартных thead. if not data.get("name_raw"): for value in values: candidate = _strip_list_label(value, "Фамилия, имя", "Игрок") if _looks_like_ru_person_name(candidate): data["name_raw"] = candidate break if not data.get("role"): joined = " | ".join(values).casefold() for role in ("вратарь", "защитник", "нападающий"): if role in joined: data["role"] = role break if not data.get("birthday"): for value in values: if "дата рождения" in value.casefold(): candidate = _strip_list_label(value, "Дата рождения") if candidate: data["birthday"] = candidate break if not data.get("country"): for value in values: if "гражданство" in value.casefold(): country, code = _normalize_country(value) if country: data["country"] = country if code: data["country_code"] = code break # Логотип/клуб — если таблица содержит ссылку на клуб. team_name = clean(data.get("team")) team_url, team_logo = _row_team_media(row, team_name) if team_url: data["team_url"] = team_url if team_logo: data["team_logo"] = team_logo if data.get("name_raw") or data.get("role"): hints[pid] = {**hints.get(pid, {}), **data} # Резерв: иногда ФИО есть прямо в ссылке, даже если таблица распарсилась иначе. for a in soup.find_all("a", href=True): url = urljoin(BASE, str(a.get("href"))) parsed = urlparse(url) canonical = urlunparse((parsed.scheme, parsed.netloc, parsed.path, "", "", "")) match = DETAIL_RE["players"].match(canonical) if not match: continue pid = int(match.group(1)) text = clean(a.get_text(" ", strip=True)) if _looks_like_ru_person_name(text): hints.setdefault(pid, {})["name_raw"] = text return hints def _merge_player_hints(target: dict[int, dict[str, Any]], incoming: dict[int, dict[str, Any]]) -> None: """Первый увиденный (обычно текущий) список имеет приоритет; исторические страницы только дополняют пробелы.""" for pid, hint in incoming.items(): dst = target.setdefault(pid, {}) for key, value in hint.items(): if key not in dst or dst.get(key) in (None, ""): dst[key] = value def discover_by_lists(kind: str, client: HttpClient, config: Config) -> tuple[dict[int, str], list[dict[str, Any]], dict[int, dict[str, Any]]]: """Читает основной раздел KHL.ru и его представление «все сезоны». Это всё один источник (www.khl.ru), но второй URL нужен, чтобы основной список текущего состава не ограничивал историческую базу. """ found: dict[int, str] = {} pages_meta: list[dict[str, Any]] = [] player_hints: dict[int, dict[str, Any]] = {} max_pages = config.max_pages or {"players": 400, "coaches": 80, "officials": 80}[kind] for base_url in LIST_URLS[kind]: try: first_html = client.fetch(base_url) except Exception as exc: LOG.warning("%s список %s недоступен: %s", kind, base_url, exc) continue first_links = detail_links(kind, first_html) found.update(first_links) if kind == "players": _merge_player_hints(player_hints, player_list_hints(first_html)) pages_meta.append({"url": base_url, "entry_count": len(first_links), "new_count": len(first_links)}) if not first_links: continue empty_streak = 0 previous_signature: tuple[int, ...] | None = tuple(sorted(first_links)) for page in range(2, max_pages + 1): url = url_with_page(base_url, page) try: html = client.fetch(url) except Exception as exc: LOG.warning("%s %s page %d: %s", kind, base_url, page, exc) empty_streak += 1 if empty_streak >= 2: break continue links = detail_links(kind, html) if kind == "players": _merge_player_hints(player_hints, player_list_hints(html)) signature = tuple(sorted(links)) new_ids = set(links) - set(found) pages_meta.append({"url": url, "entry_count": len(links), "new_count": len(new_ids)}) if signature and signature == previous_signature: break previous_signature = signature if new_ids: found.update(links); empty_streak = 0 else: empty_streak += 1 if empty_streak >= 2: break if config.max_people and len(found) >= config.max_people: break if page % 10 == 0: LOG.info("%s: %s, страница %d, всего %d карточек", kind, base_url, page, len(found)) if config.max_people and len(found) >= config.max_people: break return found, pages_meta, player_hints def discover_from_sitemap(kind: str, client: HttpClient) -> dict[int, str]: found: dict[int, str] = {} queue = [f"{BASE}/sitemap.xml", f"{BASE}/sitemap_index.xml"] visited: set[str] = set() while queue and len(visited) < 100: url = queue.pop(0) if url in visited: continue visited.add(url) try: text = client.fetch(url, use_cache=True) except Exception: continue for loc in re.findall(r"<loc>\s*(.*?)\s*</loc>", text, flags=re.I | re.S): item = clean(loc).replace("&", "&") match = DETAIL_RE[kind].match(item) if match: pid = int(match.group(1)); found[pid] = item if item.endswith("/") else item + "/" elif "sitemap" in item.casefold() and item not in visited: queue.append(item) return found def scan_official_ids(client: HttpClient, config: Config, existing: dict[int, str]) -> dict[int, str]: out = dict(existing) ids = [i for i in range(config.official_scan_min, config.official_scan_max + 1) if i not in out] if not ids: return out LOG.warning("Список судей недостаточен; резервно проверяю карточки ID %d..%d на KHL.ru", config.official_scan_min, config.official_scan_max) def worker(pid: int) -> tuple[int, bool]: url = f"{BASE}/officials/{pid}/" try: html = client.fetch(url) soup = BeautifulSoup(html, "html.parser") title = clean(soup.title.get_text(" ", strip=True)) if soup.title else "" return pid, "суд" in title.casefold() or bool(extract_profile_fields(soup).get("role")) except Exception: return pid, False with ThreadPoolExecutor(max_workers=min(config.workers, 10)) as pool: futures = {pool.submit(worker, pid): pid for pid in ids} for n, future in enumerate(as_completed(futures), 1): pid, ok = future.result() if ok: out[pid] = f"{BASE}/officials/{pid}/" if n % 100 == 0: LOG.info("Судьи scan: %d/%d, найдено %d", n, len(ids), len(out)) return out def collect_kind(kind: str, client: HttpClient, config: Config) -> Path: urls, pages, player_hints = discover_by_lists(kind, client, config) if kind == "officials" and len(urls) < 20: sitemap = discover_from_sitemap(kind, client) urls.update(sitemap) if len(urls) < 20: urls = scan_official_ids(client, config, urls) if not urls: raise RuntimeError(f"Не удалось найти карточки {kind} на KHL.ru") if config.max_people: urls = dict(list(sorted(urls.items()))[:config.max_people]) LOG.info("%s: найдено %d карточек", kind, len(urls)) errors: list[dict[str, str]] = [] people: list[Any] = [] stage_map: dict[int, dict[str, Any]] = {} def worker(item: tuple[int, str]): pid, url = item html = client.fetch(url) if kind == "players": player, stages = player_from_profile(pid, url, html) identity = player.get("identity") or {} profile = player.get("profile") or {} hint = dict(player_hints.get(pid) or {}) # ФИО и амплуа в списке /players/ лежат в структурированных колонках и # являются надёжнее глобального текста персональной страницы. current_name = clean(identity.get("full_name") or profile.get("name")) raw_hint = clean(hint.get("name_raw")) if raw_hint and _looks_like_ru_person_name(raw_hint): # Keep KHL list order (Фамилия Имя) specifically for the left menu. profile["list_name_ru"] = raw_hint if raw_hint and (not _looks_like_ru_person_name(current_name)): display_hint = _reorder_khl_title_name(raw_hint) profile["name"] = display_hint identity["full_name"] = display_hint parts = display_hint.split() identity["first_name"] = parts[0] if parts else "" identity["last_name"] = parts[-1] if len(parts) > 1 else "" identity["middle_name"] = " ".join(parts[1:-1]) if len(parts) > 2 else "" role_hint = clean(hint.get("role")).casefold() if role_hint in {"вратарь", "защитник", "нападающий"}: profile["role"] = role_hint # Этапы тоже должны знать амплуа, иначе фильтр слева остаётся пустым. for record in (player.get("stages") or {}).values(): (record.get("data") or {})["role"] = role_hint for key in ("birthday", "age", "country", "national_team", "contract_until"): if hint.get(key) not in (None, ""): profile[key] = hint[key] if hint.get("country_code"): profile["country_code"] = hint["country_code"] if hint.get("team"): team_obj = {"name": hint["team"]} if hint.get("team_url"): team_obj["url"] = hint["team_url"] if hint.get("team_logo"): team_obj["image"] = hint["team_logo"] profile["team"] = team_obj player["identity"] = identity player["profile"] = profile return pid, (player, stages) return pid, staff_from_profile(kind, pid, url, html) items = list(sorted(urls.items())) with ThreadPoolExecutor(max_workers=config.workers) as pool: future_map = {pool.submit(worker, item): item for item in items} for done, future in enumerate(as_completed(future_map), 1): pid, url = future_map[future] try: _, payload = future.result() if kind == "players": player, stages = payload people.append(player) for stage in stages: stage_map[int(stage["id"])] = stage else: people.append(payload) except Exception as exc: errors.append({"id": str(pid), "url": url, "error": str(exc)}) LOG.warning("%s %s: %s", kind, pid, exc) if done % 25 == 0 or done == len(items): LOG.info("%s карточки: %d/%d (ошибок %d)", kind, done, len(items), len(errors)) if kind == "officials" and config.current_only and people: all_seasons = sorted({season for item in people for season in ((item.get("profile") or {}).get("seasons") or [])}) if all_seasons: latest_season = all_seasons[-1] people = [item for item in people if latest_season in ((item.get("profile") or {}).get("seasons") or [])] LOG.info("Судьи: current-only = %s, осталось %d", latest_season, len(people)) if kind == "players": people.sort(key=lambda x: clean((x.get("identity") or {}).get("full_name")).casefold()) payload = { "meta": {"source": f"{BASE}/players/", "source_type": "khl.ru html", "generated_at": utc_now(), "player_count": len(people), "error_count": len(errors), "webcaster_used": False}, "errors": errors, "stages": sorted(stage_map.values(), key=lambda x: (x.get("season") or "", x.get("title") or "", x["id"])), "players": people, } out = config.output_dir / "khl_players_all.json" else: people.sort(key=lambda x: clean(x.get("name_ru") or x.get("name")).casefold()) payload = { "meta": {"kind": kind, "source": LIST_URLS[kind][0], "source_type": "khl.ru html", "generated_at": utc_now(), "people_count": len(people), "page_count": len(pages), "error_count": len(errors), "webcaster_used": False}, "errors": errors, "pages": pages, "people": people, } out = config.output_dir / ("khl_coaches_all.json" if kind == "coaches" else "khl_officials_all.json") if not people: raise RuntimeError(f"KHL.ru не отдал ни одной корректной карточки {kind}; старый JSON не перезаписан") atomic_json(out, payload) LOG.info("Записано: %s", out.resolve()) return out def parse_args() -> tuple[Config, list[str]]: parser = argparse.ArgumentParser(description="Единый парсер KHL.ru без Webcaster API") parser.add_argument("--kind", action="append", choices=(*KINDS, "all"), default=[]) parser.add_argument("--output-dir", type=Path, default=Path("data")) parser.add_argument("--cache-dir", type=Path, default=Path("cache/khl_site_html")) parser.add_argument("--workers", type=int, default=10) parser.add_argument("--timeout", type=float, default=25.0) parser.add_argument("--retries", type=int, default=4) parser.add_argument("--delay", type=float, default=0.03) parser.add_argument("--max-pages", type=int, default=0, help="0 = авто до конца") parser.add_argument("--max-people", type=int, default=0, help="тестовый лимит; 0 = все") parser.add_argument("--force", action="store_true", help="не использовать HTML-кэш") parser.add_argument("--official-scan-min", type=int, default=500) parser.add_argument("--official-scan-max", type=int, default=1600) parser.add_argument("--current-only", action="store_true", help="для судей оставить только последний сезон, найденный на карточках") parser.add_argument("--verbose", action="store_true") args = parser.parse_args() logging.basicConfig(level=logging.DEBUG if args.verbose else logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s", datefmt="%H:%M:%S") kinds = args.kind or ["all"] if "all" in kinds: kinds = list(KINDS) kinds = list(dict.fromkeys(kinds)) config = Config(output_dir=args.output_dir, cache_dir=args.cache_dir, workers=max(1, min(16, args.workers)), timeout=max(5.0, args.timeout), retries=max(0, args.retries), delay=max(0.0, args.delay), max_pages=max(0, args.max_pages), max_people=max(0, args.max_people), force=args.force, official_scan_min=max(1, args.official_scan_min), official_scan_max=max(args.official_scan_min, args.official_scan_max), current_only=bool(args.current_only)) return config, kinds def main() -> int: config, kinds = parse_args() client = HttpClient(config) try: for kind in kinds: collect_kind(kind, client, config) except KeyboardInterrupt: LOG.warning("Остановлено пользователем") return 130 except Exception as exc: LOG.exception("Ошибка: %s", exc) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())