1493 lines
69 KiB
Python
1493 lines
69 KiB
Python
#!/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"(?<!\d)(\d{2})\s*[/–—-]\s*(\d{2})(?!\d)", text)
|
||
if short:
|
||
left, right = map(int, short.groups())
|
||
# Сезоны КХЛ начинаются с 2008 года; двухзначные значения трактуем как 20xx.
|
||
return f"20{left:02d}/20{right:02d}"
|
||
return ""
|
||
|
||
|
||
def atomic_json(path: Path, payload: Any) -> 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>, только если 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())
|