208 lines
6.7 KiB
Python
208 lines
6.7 KiB
Python
"""
|
||
PGN-партии игрока с Lichess-трансляций по FIDE ID.
|
||
|
||
Источник: https://lichess.org/api/broadcast
|
||
Формат: в PGN-тегах есть WhiteFideId / BlackFideId / TimeControl.
|
||
"""
|
||
|
||
import re
|
||
import time
|
||
import requests
|
||
from bs4 import BeautifulSoup
|
||
from typing import Optional, List, Dict
|
||
|
||
HEADERS = {
|
||
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36'
|
||
}
|
||
|
||
RATE_DELAY = 1.5
|
||
MAX_BROADCASTS = 200
|
||
MIN_YEAR = 2025
|
||
|
||
|
||
def get_player_name(fide_id: str) -> str:
|
||
"""Получить имя игрока с ratings.fide.com."""
|
||
url = f'https://ratings.fide.com/profile/{fide_id}'
|
||
resp = _fetch_with_retry(url)
|
||
if not resp:
|
||
return ''
|
||
|
||
soup = BeautifulSoup(resp.text, 'html.parser')
|
||
|
||
for tag in soup.find_all(['h1', 'h2', 'span']):
|
||
text = tag.get_text(strip=True)
|
||
if text and 3 < len(text) < 80 and re.search(r'[A-ZА-ЯЁ]', text):
|
||
if not re.match(r'^(MAIN|NEWS|RATINGS|CHAMPIONSHIP|CALENDAR|FIDE|DIRECTORY|PARTNERS|CONTACTS|SHOP|HOME|ARCHIVE)', text, re.IGNORECASE):
|
||
return text
|
||
|
||
full_text = soup.get_text()
|
||
m = re.search(r'FIDE\s*ID\s*\d+', full_text)
|
||
if m:
|
||
before = full_text[:m.start()].strip().split('\n')
|
||
for line in reversed(before):
|
||
line = line.strip()
|
||
if line and 3 < len(line) < 80 and re.search(r'[A-ZА-ЯЁ]', line):
|
||
return line
|
||
|
||
return ''
|
||
|
||
|
||
def _fetch_with_retry(url: str, max_retries: int = 2):
|
||
"""GET-запрос с retry при таймауте."""
|
||
for attempt in range(max_retries + 1):
|
||
try:
|
||
return requests.get(url, headers=HEADERS, timeout=15 + attempt * 10)
|
||
except requests.exceptions.Timeout:
|
||
if attempt < max_retries:
|
||
time.sleep(1)
|
||
else:
|
||
return None
|
||
except requests.exceptions.RequestException:
|
||
return None
|
||
return None
|
||
|
||
|
||
def search_broadcasts(name_part: str) -> List[Dict]:
|
||
"""Поиск Lichess-трансляций по имени игрока."""
|
||
url = f'https://lichess.org/api/broadcast?q={name_part}'
|
||
resp = requests.get(url, headers=HEADERS, timeout=30)
|
||
resp.raise_for_status()
|
||
|
||
broadcasts = []
|
||
for line in resp.text.strip().split('\n'):
|
||
line = line.strip()
|
||
if not line:
|
||
continue
|
||
try:
|
||
import json
|
||
data = json.loads(line)
|
||
if 'tour' in data:
|
||
broadcasts.append(data)
|
||
except json.JSONDecodeError:
|
||
continue
|
||
if len(broadcasts) >= MAX_BROADCASTS:
|
||
break
|
||
|
||
return broadcasts
|
||
|
||
|
||
def fetch_round_pgn(round_id: str, delay: float = RATE_DELAY) -> Optional[str]:
|
||
"""Скачать PGN одного раунда Lichess-трансляции."""
|
||
time.sleep(delay)
|
||
url = f'https://lichess.org/api/broadcast/round/{round_id}.pgn'
|
||
resp = requests.get(url, headers=HEADERS, timeout=30)
|
||
if resp.status_code == 429:
|
||
time.sleep(5)
|
||
resp = requests.get(url, headers=HEADERS, timeout=30)
|
||
if resp.status_code != 200:
|
||
return None
|
||
text = resp.text.strip()
|
||
return text if text else None
|
||
|
||
|
||
def filter_by_fide_id(pgn_text: str, fide_id: str) -> List[str]:
|
||
"""Выкусить из PGN только партии игрока с данным FIDE ID."""
|
||
games = []
|
||
raw_games = re.split(r'\n(?=\[Event )', pgn_text)
|
||
for game in raw_games:
|
||
game = game.strip()
|
||
if not game:
|
||
continue
|
||
if f'WhiteFideId "{fide_id}"' in game or f'BlackFideId "{fide_id}"' in game:
|
||
games.append(game)
|
||
return games
|
||
|
||
|
||
def classify_time_control(game_pgn: str) -> str:
|
||
"""Определить категорию: standard / rapid / blitz."""
|
||
m = re.search(r'\[TimeControl\s+"([^"]+)"\]', game_pgn)
|
||
if not m:
|
||
return 'unknown'
|
||
tc_str = m.group(1)
|
||
parts = tc_str.split('+')
|
||
try:
|
||
initial = int(parts[0])
|
||
except ValueError:
|
||
return 'unknown'
|
||
# initial time in seconds
|
||
if initial >= 3600:
|
||
return 'standard'
|
||
elif initial >= 600:
|
||
return 'rapid'
|
||
else:
|
||
return 'blitz'
|
||
|
||
|
||
def get_player_games(fide_id: str) -> Dict[str, str]:
|
||
"""Оркестратор: найти все PGN-партии игрока, сгруппировать по контролю."""
|
||
name = get_player_name(fide_id)
|
||
if not name:
|
||
return {}
|
||
|
||
parts = name.lower().split()
|
||
last_name = parts[0].rstrip(',') if parts else name.lower()
|
||
|
||
broadcasts = search_broadcasts(last_name)
|
||
|
||
# Приоритезируем трансляции, где имя игрока в названии
|
||
relevant_bcs = []
|
||
others = []
|
||
for bc in broadcasts:
|
||
bc_name = bc.get('tour', {}).get('name', '').lower()
|
||
if any(part in bc_name for part in parts if len(part) > 2):
|
||
relevant_bcs.append(bc)
|
||
else:
|
||
others.append(bc)
|
||
|
||
# Собираем finished-раунды: сначала из релевантных, потом остальные
|
||
rounds_to_fetch = []
|
||
MAX_ROUNDS = 30
|
||
|
||
for bc_list in (relevant_bcs, others):
|
||
for bc in bc_list:
|
||
for rd in bc.get('rounds', []):
|
||
if rd.get('finished') and len(rounds_to_fetch) < MAX_ROUNDS:
|
||
rounds_to_fetch.append(rd['id'])
|
||
if len(rounds_to_fetch) >= MAX_ROUNDS:
|
||
break
|
||
if len(rounds_to_fetch) >= MAX_ROUNDS:
|
||
break
|
||
|
||
result: Dict[str, list] = {'standard': [], 'rapid': [], 'blitz': []}
|
||
seen_games = set()
|
||
total_found = 0
|
||
|
||
for round_id in rounds_to_fetch:
|
||
pgn = fetch_round_pgn(round_id)
|
||
if not pgn:
|
||
continue
|
||
for game in filter_by_fide_id(pgn, fide_id):
|
||
key = game[:200]
|
||
if key in seen_games:
|
||
continue
|
||
seen_games.add(key)
|
||
tc = classify_time_control(game)
|
||
if tc in result:
|
||
result[tc].append(game)
|
||
else:
|
||
result['standard'].append(game)
|
||
total_found += 1
|
||
|
||
return {
|
||
'standard': '\n\n'.join(result['standard']),
|
||
'rapid': '\n\n'.join(result['rapid']),
|
||
'blitz': '\n\n'.join(result['blitz']),
|
||
}
|
||
|
||
|
||
def count_games(pgn_text: str) -> int:
|
||
"""Количество партий в PGN-строке."""
|
||
if not pgn_text.strip():
|
||
return 0
|
||
return len([g for g in pgn_text.split('\n\n[Event ') if '[Event ' in g or g.startswith('[Event ')])
|
||
|
||
|
||
def extract_fide_id(text: str) -> Optional[str]:
|
||
"""Вытащить FIDE ID из текста (ссылка ratings.fide.com или просто ID)."""
|
||
m = re.search(r'(\d{5,8})', text)
|
||
return m.group(1) if m else None
|