""" Клиентский Telegram-бот chessCalc. Принимает ссылку на турнир chess-results.com, рассчитывает жеребьёвку следующего тура и выводит пары в таблице (MarkdownV2). """ import os import re import sys import asyncio from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup from telegram.ext import Application, CommandHandler, MessageHandler, CallbackQueryHandler, 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 from bots import tracker TOKEN = os.environ.get('CLIENT_BOT_TOKEN', '') URL_PATTERN = re.compile(r'https?://(?:[\w-]+\.)?chess-results\.com/\S+') START_MSG = { 'ru': ( '♟️ *chessCalc* — жеребьёвка следующего тура\n\n' 'Рассчитываю пары _до официальной публикации_ на chess\-results\.com\.\n' 'Слежу за игроками: как только тур завершён — присылаю результат и жеребьёвку следующего\.\n\n' '*Команды:*\n' '/addplayer — подписаться на игрока по FIDE ID\n' '/myplayers — список отслеживаемых игроков\n' '/removeplayer — отписаться от игрока\n\n' '*Разовый расчёт:*\n' 'Пришли ссылку на турнир — получи жеребьёвку прямо сейчас\n' '`https://chess\-results\.com/tnr1393124\.aspx?lan=11`\n\n' '_Движок: FIDE Dutch System \(bbpPairings\), точность 100% на большинстве турниров_' ), 'en': ( '♟️ *chessCalc* — next round pairings\n\n' 'I calculate pairings _before official publication_ on chess\-results\.com\.\n' 'I also track players: when a round finishes — I send the result and next round pairings\.\n\n' '*Commands:*\n' '/addplayer — subscribe to a player by FIDE ID\n' '/myplayers — list of tracked players\n' '/removeplayer — unsubscribe from a player\n\n' '*One\-time calculation:*\n' 'Send a tournament link — get pairings right now\n' '`https://chess\-results\.com/tnr1393124\.aspx?lan=11`\n\n' '_Engine: FIDE Dutch System \(bbpPairings\), 100% accuracy on most tournaments_' ), } LOCALE = { 'ru': { 'player': 'Игрок', 'rating': 'Рейт', 'points': 'Оч', 'tb1': 'Доп1', 'title': '📋 *Тур {rnd}* — пары', 'bye': '— BYE —', 'footer_bbp': '\n_FIDE 2025 \(bbpPairings\)_', 'footer_fallback': '\n_Упрощённый Swiss \(bbpPairings недоступен\)_', 'no_players': 'У вас нет отслеживаемых игроков\.\n/addplayer — добавить игрока', 'my_players_title': 'Ваши игроки', 'waiting': '⏳ _Ожидаю турнир_', 'round_label': 'тур', 'points_label': 'очк\\.', }, 'en': { 'player': 'Player', 'rating': 'Rat', 'points': 'Pts', 'tb1': 'TB1', 'title': '📋 *Round {rnd}* — pairings', 'bye': '— BYE —', 'footer_bbp': '\n_FIDE 2025 \(bbpPairings\)_', 'footer_fallback': '\n_Simplified Swiss \(bbpPairings unavailable\)_', 'no_players': 'You have no tracked players\.\n/addplayer — add a player', 'my_players_title': 'Your players', 'waiting': '⏳ _Waiting for a tournament_', 'round_label': 'round', 'points_label': 'pts', }, } def _get_locale(lang_code: str) -> dict: if lang_code and lang_code.startswith('ru'): return LOCALE['ru'] return LOCALE['en'] 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 (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: # Truncate to fit, then pad to exactly width result = [] w = 0 for ch in s: cw = 2 if _display_width(ch) == 2 else 1 if w + cw + 3 > width: # +3 for "..." break result.append(ch) w += cw truncated = ''.join(result) + '...' pad_to = width - _display_width(truncated) return truncated + ' ' * max(0, pad_to) return s + ' ' * (width - cur) def format_pairings(pairings_data: dict, tournament_name: str, lang: str = '', player_name: str = '', player_sno: int = 0) -> str: locale = _get_locale(lang) pairings = pairings_data['pairings'] rnd = pairings_data['round'] # If highlighting a player, reduce name width to compensate for ✅ emoji (visually 2 chars wide) w_name = 15 if player_sno else 17 w_board, w_rating, w_pts, w_tb1 = 2, 4, 4, 5 border = '+'.join('-' * w for w in [w_board, w_name, w_rating, w_pts, w_tb1]) header = (f'{_pad_col("#", w_board)}|{_pad_col(locale["player"], w_name)}|' f'{_pad_col(locale["rating"], w_rating)}|{_pad_col(locale["points"], w_pts)}|' f'{_pad_col(locale["tb1"], w_tb1)}') table_header = f'```\n{header}\n{border}\n' source = pairings_data.get('source', 'algorithm') footer = locale['footer_bbp'] if source == 'bbp_pairings_fide_2025' else \ locale['footer_fallback'] if source == 'swiss_algorithm_simplified' else '' safe_tour = _md_escape(tournament_name) title = f'📋 *{safe_tour}*\n*{locale["title"].format(rnd=rnd)}*' return { 'title': title, 'rows': _build_table_rows(pairings, w_board, w_name, w_rating, w_pts, w_tb1, border, locale['bye'], player_name, player_sno), 'table_header': table_header, 'footer': footer, } def _build_table_rows(pairings: list, w_board: int, w_name: int, w_rating: int, w_pts: int, w_tb1: int, border: str, bye_label: str = '— BYE —', player_name: str = '', player_sno: int = 0) -> list: def _rating_str(p): return str(p.rating) if p.rating else '' def _tb1_str(p): return f"{p.tb[0]:.1f}" if p.tb else '' # Find the board where the player plays. # Primary: match by SNo (reliable across Latin/Cyrillic name variations). # Fallback: match by normalized name. def _n(s: str) -> str: return s.replace(',', '').replace(' ', ' ').lower() player_board = 0 board_idx = 1 player_norm = _n(player_name) if player_name else '' for pairing in pairings: if len(pairing) != 3: continue p1, p2, _color = pairing if player_sno and (p1.sno == player_sno or p2.sno == player_sno): player_board = board_idx player_name = p1.name if p1.sno == player_sno else p2.name break if player_norm and (player_norm == _n(p1.name) or player_norm == _n(p2.name)): player_board = board_idx player_name = p1.name if player_norm == _n(p1.name) else p2.name break board_idx += 1 # If we have a player, show ±5 boards around them if player_board > 0: lo = max(1, player_board - 5) hi = player_board + 5 else: lo, hi = 1, 999999 def _name_cell(name: str) -> str: if player_name and name == player_name: return f'✅ {name}' return name rows = [] board = 1 for pi, pairing in enumerate(pairings): if len(pairing) != 3: continue p1, p2, color = pairing if board < lo: board += 1 continue if board > hi: break if color == 'bye': rows.append( f'{_pad_col(str(board), w_board)}|' f'{_pad_col(_name_cell(p1.name), w_name)}|' f'{_pad_col(_rating_str(p1), w_rating)}|' f'{_pad_col(f"{p1.points:.1f}", w_pts)}|' f'{_pad_col(_tb1_str(p1), w_tb1)}' ) rows.append( f'{_pad_col("", w_board)}|' f'{_pad_col(bye_label, w_name)}|' f'{_pad_col("", w_rating)}|' f'{_pad_col("", w_pts)}|' f'{_pad_col("", w_tb1)}' ) board += 1 if pi + 1 < len(pairings) and board <= hi: rows.append(border) continue if color == 'w': white, black = p1, p2 else: white, black = p2, p1 rows.append( f'{_pad_col(str(board), w_board)}|' f'{_pad_col(_name_cell(white.name), w_name)}|' f'{_pad_col(_rating_str(white), w_rating)}|' f'{_pad_col(f"{white.points:.1f}", w_pts)}|' f'{_pad_col(_tb1_str(white), w_tb1)}' ) rows.append( f'{_pad_col("", w_board)}|' f'{_pad_col(_name_cell(black.name), w_name)}|' f'{_pad_col(_rating_str(black), w_rating)}|' f'{_pad_col(f"{black.points:.1f}", w_pts)}|' f'{_pad_col(_tb1_str(black), w_tb1)}' ) board += 1 if pi + 1 < len(pairings) and board <= hi: rows.append(border) return rows def _render_chunks(fmt: dict) -> list: header = fmt['table_header'] footer = fmt['footer'] rows = fmt['rows'] chunks = [] title_line = fmt['title'] + '\n' # Estimate overhead per chunk overhead = len(title_line) + len(header) + len(footer) + 10 available = 3800 - overhead rows_per_chunk = max(15, available // 60) rows_per_chunk = (rows_per_chunk // 3) * 3 # 3 rows/pair (white, black, border) for i in range(0, len(rows), rows_per_chunk): batch = rows[i:i + rows_per_chunk] parts = [] if i == 0: parts.append(title_line) parts.append(header) parts.extend(batch) parts.append('```') if i + rows_per_chunk >= len(rows): parts.append(footer) chunks.append('\n'.join(parts)) return chunks 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 '') lang = 'ru' if (user.language_code or '').startswith('ru') else 'en' await update.message.reply_text(START_MSG[lang], parse_mode=ParseMode.MARKDOWN_V2) async def add_player(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 '') context.user_data['awaiting_fide_id'] = True await update.message.reply_text( 'Пришли мне FIDE ID игрока \(число\) или ссылку на профиль FIDE:\n' r'`https://ratings\.fide\.com/profile/1503014`\n' '\n' '/cancel — отмена', parse_mode=ParseMode.MARKDOWN_V2) async def cancel_sub(update: Update, context: ContextTypes.DEFAULT_TYPE): had_fide = context.user_data.pop('awaiting_fide_id', None) if had_fide: await update.message.reply_text('❎ Подписка отменена\.') else: await update.message.reply_text('Нет активного процесса подписки\.') def _player_card(p: dict, loc: dict) -> tuple[str, InlineKeyboardMarkup]: safe_name = _md_escape(p['player_name']) fed = p['fed'] or '' rating = p['rating'] or '—' fide_id = p['fide_id'] lines = [f'👤 *{safe_name}*'] lines.append(f'🏳 {_md_escape(fed)} · {rating}') tours = p['tournaments'] waiting = all(t['name'] is None for t in tours) if waiting: lines.append(loc['waiting']) else: for t in tours: if t['name'] is None: continue safe_tour = _md_escape(t['name']) pts = _md_escape(f'{t["points"]:.1f}') rd = t['round_done'] lines.append(f'📌 {safe_tour}') lines.append(f' {pts} {loc["points_label"]}, {loc["round_label"]} {rd}') keyboard = InlineKeyboardMarkup([[ InlineKeyboardButton('🗑 Удалить', callback_data=f'remove:{fide_id}') ]]) return '\n'.join(lines), keyboard async def my_players(update: Update, context: ContextTypes.DEFAULT_TYPE): user = update.effective_user loc = _get_locale(user.language_code or '') players = tracker.get_user_subs(user.id) if not players: await update.message.reply_text(loc['no_players'], parse_mode=ParseMode.MARKDOWN_V2) return header = f'*{_md_escape(loc["my_players_title"])} \\({len(players)}\\):*' await update.message.reply_text(header, parse_mode=ParseMode.MARKDOWN_V2) for p in players: text, keyboard = _player_card(p, loc) await update.message.reply_text(text, parse_mode=ParseMode.MARKDOWN_V2, reply_markup=keyboard) async def remove_player_callback(update: Update, context: ContextTypes.DEFAULT_TYPE): query = update.callback_query await query.answer() user = query.from_user loc = _get_locale(user.language_code or '') fide_id = int(query.data.split(':')[1]) ok = tracker.remove_player_subscriptions(fide_id, user.id) if ok: # Strike through the card text and remove the button struck = query.message.text.replace('👤', '❌') await query.edit_message_text(struck, reply_markup=None) else: await query.edit_message_text(query.message.text + '\n\n_уже удалён_', parse_mode=ParseMode.MARKDOWN_V2, reply_markup=None) def _extract_fide_id(text: str) -> int | None: """Extract FIDE ID from plain number or profile URL.""" m = re.search(r'profile/(\d+)', text) if m: return int(m.group(1)) # Try plain number (between 6-8 digits) for token in text.split(): if token.isdigit() and 100000 <= int(token) <= 9999999999: return int(token) return None async def _lookup_and_confirm(update: Update, context, fide_id: int): user = update.effective_user lang_code = user.language_code or '' try: player = tracker.fetch_fide_player(fide_id) except Exception as e: await update.message.reply_text( f'❌ Не удалось найти игрока: {_md_escape(str(e))}\n' 'Попробуй другой ID или /cancel', parse_mode=ParseMode.MARKDOWN_V2) context.user_data['awaiting_fide_id'] = True return safe_name = _md_escape(player['name']) title = player.get('title') or '' title_str = f', {_md_escape(title)}' if title and title != 'None' else '' fed = player.get('fed') or '' await update.message.reply_text( f'✅ *{safe_name}*{title_str}\n' f'{"🇫 " if fed else ""}{_md_escape(fed)} \\| Рейтинг: {player["rating"]}\n\n' '🔍 Ищу турниры на chess\\-results\\.com\\.\\.', parse_mode=ParseMode.MARKDOWN_V2) # Run scan asynchronously (avoid blocking the event loop) # Use canonical FIDE ID from profile page (may differ from URL-extracted ID) from concurrent.futures import ThreadPoolExecutor loop = asyncio.get_event_loop() tournaments = await loop.run_in_executor( None, tracker.scan_for_player, player['fide_id']) if tournaments: # Auto-subscribe to all found tournaments lines = [f'✅ *Найдено турниров: {len(tournaments)}*'] for t in tournaments: tracker.add_subscription( user.id, player, t['url'], t['name'], t['sno'], 0, 0.0, lang_code, t['start_date'], t['end_date']) dates = '' if t['start_date'] and t['end_date']: dates = f' \\({_md_escape(t["start_date"])} — {_md_escape(t["end_date"])}\\)' safe_tour = _md_escape(t['name']) lines.append(f' • {safe_tour}{dates}') lines.append('\nЯ буду присылать результаты и жеребьёвку\.\n' '/myplayers \\- список подписок') await update.message.reply_text('\n'.join(lines), parse_mode=ParseMode.MARKDOWN_V2) else: # No active tournament found — watch for future ones tracker.add_player_watch(user.id, player, lang_code) safe_name = _md_escape(player['name']) await update.message.reply_text( f'👁 *{safe_name}* поставлен на отслеживание\.\n\n' 'Активного турнира сейчас не найдено\\. ' 'Как только игрок появится в новом турнире — сразу подпишу и пришлю жеребьёвку\\.\n\n' '/myplayers \\- список подписок', parse_mode=ParseMode.MARKDOWN_V2) def _apply_round_results(tournament, games, rd): standings = tournament['standings'] sno_to_standing = {s['starting_sno']: s for s in standings if s.get('starting_sno')} name_to_standing = { re.sub(r'\s+', '', s['name'].lower()): s for s in standings } name_parts_to_standing: dict[str, list] = {} for s in standings: parts = re.sub(r'\s+', '', s['name'].lower()).split() for p in parts: name_parts_to_standing.setdefault(p, []).append(s) def _find_player(sno, name_str): p = sno_to_standing.get(sno) if p: return p if name_str: norm = re.sub(r'\s+', '', name_str.lower()) p = name_to_standing.get(norm) if p: return p parts = norm.split() for part in parts: candidates = name_parts_to_standing.get(part, []) if len(candidates) == 1: return candidates[0] return None for g in games: if g is None: continue ws, bs = g['white_sno'], g['black_sno'] ws_result = g.get('white_score') bs_result = g.get('black_score') if ws_result is None: continue ws_val = float(ws_result) bs_val = float(bs_result) wp = _find_player(ws, g.get('white_name', '')) bp = _find_player(bs, g.get('black_name', '')) if wp: opp_sno = bs if bs != 0 else (bp.get('starting_sno') if bp else 0) existing = [r for r in wp.get('results', []) if r.get('round') != rd] existing.append({'opponent': opp_sno, 'color': 'w', 'score': ws_val, 'round': rd}) wp['results'] = existing if bp: opp_sno = ws if ws != 0 else (wp.get('starting_sno') if wp else 0) existing = [r for r in bp.get('results', []) if r.get('round') != rd] existing.append({'opponent': opp_sno, 'color': 'b', 'score': bs_val, 'round': rd}) bp['results'] = existing for s in standings: sno = s.get('starting_sno') if sno and len(s.get('results', [])) < rd: orig_results = s.get('_orig_results', []) if len(orig_results) >= rd: res = orig_results[rd - 1] res['round'] = rd s['results'] = list(s.get('results', [])) s['results'].append(res) for s in standings: s['points'] = sum(r.get('score', 0) for r in s.get('results', [])) tournament['current_round'] = rd 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 '') # Step 1: user just called /addplayer — waiting for FIDE ID if context.user_data.pop('awaiting_fide_id', None): fide_id = _extract_fide_id(text) if fide_id is None: await update.message.reply_text( 'Не удалось распознать FIDE ID\. Отправь число или ссылку вида\n' r'`https://ratings\.fide\.com/profile/1503014`' '\n' '/cancel — отмена', parse_mode=ParseMode.MARKDOWN_V2) context.user_data['awaiting_fide_id'] = True return await _lookup_and_confirm(update, context, fide_id) return # Step 2: regular pairings request 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 initial_rd = next_rd from swiss_calc.parser import fetch_url as _fetch_url, parse_round_pairings try: while next_rd <= tournament['num_rounds']: 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: _apply_round_results(tournament, rd_games, next_rd) next_rd += 1 continue if rd_games: await update.message.reply_text( f'⚠️ Жеребьёвка тура {next_rd} уже опубликована на chess\\-results\\.com\\.\n' 'Я всё равно посчитаю, но пары могут незначительно отличаться ' '\\(разные редакции правил FIDE\\)\\.', parse_mode=ParseMode.MARKDOWN_V2) break else: await msg.edit_text('🏁 Турнир уже завершён\\. Жеребьёвки не будет\\.', parse_mode=ParseMode.MARKDOWN_V2) record_request(user.id, url, tournament.get('name', ''), True) return if next_rd > initial_rd: await msg.edit_text( f'✅ *{_md_escape(tournament["name"])}*\n' f'📅 Сыграно туров: {tournament["current_round"]} из {tournament["num_rounds"]}\n' f'👥 Участников: {len(tournament["standings"])}', 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 lang_code = user.language_code or '' # If user has a subscription for this tournament, highlight their player player_name = '' try: subs = tracker.get_user_subs(user.id) for s in subs: sub_tour = s.get('tournament_name', '') if sub_tour and sub_tour == tournament.get('name', ''): player_name = s.get('player_name', '') break except Exception: pass fmt = format_pairings(result, tournament['name'], lang_code, player_name) chunks = _render_chunks(fmt) try: 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) except Exception as e: # Fallback: plain text without MarkdownV2 plain_chunks = [c.replace('*', '').replace('_', '').replace('\\', '') for c in chunks] await calc_msg.edit_text(plain_chunks[0]) for chunk in plain_chunks[1:]: await update.message.reply_text(chunk) record_request(user.id, url, tournament.get('name', ''), True) def _chunk_output(text: str, size: int) -> list: lines = text.split('\n') chunks = [] current = [] current_len = 0 for line in lines: line_len = len(line) + 1 # +1 for newline if current_len + line_len > size and current: chunks.append('\n'.join(current)) current = [] current_len = 0 current.append(line) current_len += line_len 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(CommandHandler('addplayer', add_player)) app.add_handler(CommandHandler('cancel', cancel_sub)) app.add_handler(CommandHandler('myplayers', my_players)) app.add_handler(CallbackQueryHandler(remove_player_callback, pattern=r'^remove:\d+$')) app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_url)) app.job_queue.run_repeating( tracker.check_all_subscriptions, interval=60, first=10) app.job_queue.run_repeating( tracker.rescan_new_tournaments, interval=3600, first=60) app.job_queue.run_repeating( tracker.rescan_existing_tournaments, interval=7200, first=300) app.job_queue.run_repeating( tracker.warmup_cache, interval=21600, first=10) print('Client bot started', file=sys.stderr) print('Client bot started', file=sys.stderr) app.run_polling() if __name__ == '__main__': main()