""" Клиентский Telegram-бот chessCalc. Принимает ссылку на турнир chess-results.com, рассчитывает жеребьёвку следующего тура и выводит пары в таблице (MarkdownV2). """ import os import re import sys import asyncio from telegram import Update from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes from telegram.constants import ParseMode sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from swiss_calc.parser import fetch_tournament from swiss_calc.swiss import calculate_next_round from bots.stats import record_user, record_request TOKEN = os.environ.get('CLIENT_BOT_TOKEN', '') URL_PATTERN = re.compile(r'https?://(?:[\w-]+\.)?chess-results\.com/\S+') START_MSG = ( r'*♟️ chessCalc — жеребьёвка следующего тура*\n\n' r'Я рассчитываю пары следующего тура шахматного турнира ' r'_до официальной публикации_ на chess\-results\.com\.\n\n' r'Просто пришли мне ссылку на турнир — и я покажу, ' r'кто с кем играет в следующем туре\.\n\n' r'Пример ссылки:\n' r'`https://chess\-results\.com/tnr1393124\.aspx?lan=11`\n\n' r'Работает для швейцарских турниров\. Движок: FIDE Dutch System \(bbpPairings\)\.' ) def _md_escape(text: str) -> str: escape_chars = r'_*[]()~`>#+-=|{}.!' result = [] for ch in text: if ch in escape_chars: result.append('\\' + ch) else: result.append(ch) return ''.join(result) def _display_width(s: str) -> int: w = 0 for ch in s: cp = ord(ch) if (0x0400 <= cp <= 0x04FF or 0x0500 <= cp <= 0x052F or 0x2E80 <= cp <= 0x9FFF or 0xF900 <= cp <= 0xFAFF or 0xFE30 <= cp <= 0xFE4F): w += 2 else: w += 1 return w def _pad_col(s: str, width: int) -> str: cur = _display_width(s) if cur >= width: return s return s + ' ' * (width - cur) def format_pairings(pairings_data: dict, tournament_name: str) -> str: pairings = pairings_data['pairings'] rnd = pairings_data['round'] w_name, w_pts = 28, 5 border = '─┼─'.join('─' * w for w in [4, w_name, w_pts]) lines = [ f'*{_md_escape(tournament_name)}*', f'📋 *Тур {rnd}* — пары', '', '```', f'{_pad_col("#", 4)} │ {_pad_col("Игрок", w_name)} │ {_pad_col("О", w_pts)}', border, ] board = 1 for pairing in pairings: if len(pairing) != 3: continue p1, p2, color = pairing if color == 'bye': lines.append( f'{_pad_col(str(board), 4)} │ {_pad_col(p1.name, w_name)} │ ' f'{_pad_col(str(p1.points), w_pts)}' ) lines.append( f'{_pad_col("", 4)} │ {_pad_col("— BYE —", w_name)} │ ' f'{_pad_col("", w_pts)}' ) board += 1 continue if color == 'w': white, black = p1, p2 else: white, black = p2, p1 lines.append( f'{_pad_col(str(board), 4)} │ {_pad_col(white.name, w_name)} │ ' f'{_pad_col(f"{white.points:.1f}", w_pts)}' ) lines.append( f'{_pad_col("", 4)} │ {_pad_col(black.name, w_name)} │ ' f'{_pad_col(f"{black.points:.1f}", w_pts)}' ) board += 1 lines.append('```') source = pairings_data.get('source', 'algorithm') if source == 'bbp_pairings_fide_2025': lines.append('\n_FIDE 2025 \(bbpPairings\)_') elif source == 'swiss_algorithm_simplified': lines.append('\n_Упрощённый Swiss \(bbpPairings недоступен\)_') return '\n'.join(lines) async def 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(START_MSG, parse_mode=ParseMode.MARKDOWN_V2) async def handle_url(update: Update, context: ContextTypes.DEFAULT_TYPE): user = update.effective_user text = update.message.text.strip() record_user(user.id, user.username or '', user.first_name or '', user.last_name or '') match = URL_PATTERN.search(text) if not match: await update.message.reply_text( '❌ Не нашёл ссылку на chess\\-results\\.com в сообщении\\.\n\n' 'Пришли ссылку вида:\n' '`https://chess\\-results\\.com/tnr1393124\\.aspx?lan=11`', parse_mode=ParseMode.MARKDOWN_V2) return url = match.group(0) msg = await update.message.reply_text('⏳ Загружаю турнир\.\.\.', parse_mode=ParseMode.MARKDOWN_V2) try: tournament = fetch_tournament(url) except Exception as e: await msg.edit_text(f'❌ Ошибка загрузки: {_md_escape(str(e))}', parse_mode=ParseMode.MARKDOWN_V2) record_request(user.id, url, '', False, str(e)) return if tournament['current_round'] >= tournament['num_rounds']: await msg.edit_text('🏁 Турнир уже завершён\\. Жеребьёвки не будет\\.', parse_mode=ParseMode.MARKDOWN_V2) record_request(user.id, url, tournament.get('name', ''), True) return status_parts = [ f'✅ *{_md_escape(tournament["name"])}*', f'📅 Сыграно туров: {tournament["current_round"]} из {tournament["num_rounds"]}', f'👥 Участников: {len(tournament["standings"])}', ] await msg.edit_text('\n'.join(status_parts), parse_mode=ParseMode.MARKDOWN_V2) await asyncio.sleep(0.5) # Check if next round pairings already published on site base_url = re.sub(r'[&?]art=\d+', '', url) base_url = re.sub(r'[&?]rd=\d+', '', base_url) base_url = re.sub(r'[&?]turdet=\w+', '', base_url) base_url = re.sub(r'[&?]SNode=\w+', '', base_url) next_rd = tournament['current_round'] + 1 from swiss_calc.parser import fetch_url as _fetch_url, parse_round_pairings try: rd_html = _fetch_url(f'{base_url}&art=2&rd={next_rd}&turdet=YES') rd_games = parse_round_pairings(rd_html, next_rd) played = any(g.get('white_score') is not None for g in rd_games if g) if played: await update.message.reply_text( f'⚠️ Тур {next_rd} уже сыгран — на сайте есть результаты\\.', parse_mode=ParseMode.MARKDOWN_V2) record_request(user.id, url, tournament.get('name', ''), True) return if rd_games: await update.message.reply_text( f'⚠️ Жеребьёвка тура {next_rd} уже опубликована на chess\\-results\\.com\\.\n' 'Я всё равно посчитаю, но пары могут незначительно отличаться ' '\\(разные редакции правил FIDE\\)\\.', parse_mode=ParseMode.MARKDOWN_V2) except Exception: pass # Calculate next round calc_msg = await update.message.reply_text( f'🧮 Считаю жеребьёвку тура {next_rd}\\.\\.\\.', parse_mode=ParseMode.MARKDOWN_V2) try: result = calculate_next_round(tournament) except Exception as e: await calc_msg.edit_text(f'❌ Ошибка расчёта: {_md_escape(str(e))}', parse_mode=ParseMode.MARKDOWN_V2) record_request(user.id, url, tournament.get('name', ''), False, str(e)) return output = format_pairings(result, tournament['name']) # Telegram has 4096 char limit try: if len(output) > 4000: chunks = _chunk_output(output, 3800) await calc_msg.edit_text(chunks[0], parse_mode=ParseMode.MARKDOWN_V2, disable_web_page_preview=True) for chunk in chunks[1:]: await update.message.reply_text(chunk, parse_mode=ParseMode.MARKDOWN_V2, disable_web_page_preview=True) else: await calc_msg.edit_text(output, parse_mode=ParseMode.MARKDOWN_V2, disable_web_page_preview=True) except Exception: # MarkdownV2 parse failed — fall back to plain text plain = output.replace('*', '').replace('_', '').replace('\\', '') if len(plain) > 4000: chunks = _chunk_output(plain, 3800) await calc_msg.edit_text(chunks[0]) for chunk in chunks[1:]: await update.message.reply_text(chunk) else: await calc_msg.edit_text(plain) record_request(user.id, url, tournament.get('name', ''), True) def _chunk_output(text: str, size: int) -> list: """Split output at ``` boundaries: keep code blocks intact.""" chunks = [] in_code = False current = [] current_len = 0 for line in text.split('\n'): is_fence = line.strip().startswith('```') new_len = current_len + len(line) + 1 if new_len > size and current and not in_code: chunks.append('\n'.join(current)) current = [line] current_len = len(line) else: current.append(line) current_len += len(line) + 1 if is_fence: in_code = not in_code if current: chunks.append('\n'.join(current)) return chunks def main(): if not TOKEN: print('CLIENT_BOT_TOKEN not set', file=sys.stderr) sys.exit(1) app = Application.builder().token(TOKEN).build() app.add_handler(CommandHandler('start', start)) app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_url)) print('Client bot started', file=sys.stderr) app.run_polling() if __name__ == '__main__': main()