From 74dd696783e2d4a06ac33ab5f13730abe42bd48c Mon Sep 17 00:00:00 2001 From: vrubel Date: Mon, 15 Jun 2026 19:34:24 +0000 Subject: [PATCH] =?UTF-8?q?=D0=94=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=D0=B0=20=D0=BA=D0=BE=D0=BC=D0=B0=D0=BD=D0=B4=D0=B0=20/ge?= =?UTF-8?q?tGamesPlayer:=20=D0=BF=D0=BE=D0=B8=D1=81=D0=BA=20PGN-=D0=BF?= =?UTF-8?q?=D0=B0=D1=80=D1=82=D0=B8=D0=B9=20=D0=BF=D0=BE=20FIDE=20ID=20?= =?UTF-8?q?=D1=87=D0=B5=D1=80=D0=B5=D0=B7=20Lichess=20broadcast=20API?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- bots/client_bot.py | 83 ++++++++++++++- swiss_calc/pgn_fetcher.py | 208 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 290 insertions(+), 1 deletion(-) create mode 100644 swiss_calc/pgn_fetcher.py diff --git a/bots/client_bot.py b/bots/client_bot.py index f9f1c31..1604845 100644 --- a/bots/client_bot.py +++ b/bots/client_bot.py @@ -8,7 +8,7 @@ import re import sys import asyncio from telegram import Update -from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes +from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes, ConversationHandler from telegram.constants import ParseMode sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) @@ -386,6 +386,80 @@ def _chunk_output(text: str, size: int) -> list: return chunks +ASK_FIDE_ID = 1 + + +async def get_games_start(update: Update, context: ContextTypes.DEFAULT_TYPE): + user = update.effective_user + record_user(user.id, user.username or '', user.first_name or '', + user.last_name or '') + await update.message.reply_text( + '🔍 Пришли FIDE ID или ссылку на профиль ratings.fide.com\n\n' + 'Пример: `46616543`\n' + 'Или: `https://ratings.fide.com/profile/46616543`', + parse_mode=ParseMode.MARKDOWN_V2) + return ASK_FIDE_ID + + +async def get_games_receive_id(update: Update, context: ContextTypes.DEFAULT_TYPE): + user = update.effective_user + text = update.message.text.strip() + + from swiss_calc.pgn_fetcher import extract_fide_id, get_player_games, count_games + + fide_id = extract_fide_id(text) + if not fide_id: + await update.message.reply_text( + '❌ Не нашёл FIDE ID в сообщении\\. Пришли 5\\-8 значный номер или ссылку на ratings\\.fide\\.com', + parse_mode=ParseMode.MARKDOWN_V2) + return ASK_FIDE_ID + + msg = await update.message.reply_text( + f'⏳ Ищу партии для FIDE ID {fide_id}\\.\\.\\.', + parse_mode=ParseMode.MARKDOWN_V2) + + try: + games = get_player_games(fide_id) + except Exception as e: + await msg.edit_text(f'❌ Ошибка: {str(e)[:200]}') + return ConversationHandler.END + + sent = 0 + for tc in ('standard', 'rapid', 'blitz'): + pgn = games.get(tc, '') + n = count_games(pgn) + if n == 0: + continue + filename = f'fide_{fide_id}_{tc}.pgn' + try: + await update.message.reply_document( + document=bytes(pgn, 'utf-8'), + filename=filename, + caption=f'📋 {tc.upper()} — {n} партий' + ) + sent += 1 + except Exception as e: + await update.message.reply_text( + f'⚠️ Не удалось отправить {tc}: {str(e)[:100]}') + + if sent == 0: + await update.message.reply_text( + '🔍 Партий не найдено в Lichess\\-трансляциях за последнее время\\.', + parse_mode=ParseMode.MARKDOWN_V2) + + try: + await msg.delete() + except Exception: + pass + + return ConversationHandler.END + + +async def cancel_get_games(update: Update, context: ContextTypes.DEFAULT_TYPE): + await update.message.reply_text('Отменено\\.', parse_mode=ParseMode.MARKDOWN_V2) + return ConversationHandler.END + + def main(): if not TOKEN: print('CLIENT_BOT_TOKEN not set', file=sys.stderr) @@ -394,6 +468,13 @@ def main(): app = Application.builder().token(TOKEN).build() app.add_handler(CommandHandler('start', start)) app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_url)) + app.add_handler(ConversationHandler( + entry_points=[CommandHandler('getGamesPlayer', get_games_start)], + states={ + ASK_FIDE_ID: [MessageHandler(filters.TEXT & ~filters.COMMAND, get_games_receive_id)], + }, + fallbacks=[CommandHandler('cancel', cancel_get_games)], + )) print('Client bot started', file=sys.stderr) app.run_polling() diff --git a/swiss_calc/pgn_fetcher.py b/swiss_calc/pgn_fetcher.py new file mode 100644 index 0000000..11b5b80 --- /dev/null +++ b/swiss_calc/pgn_fetcher.py @@ -0,0 +1,208 @@ +""" +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