""" Клиентский 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 from bots import tracker TOKEN = os.environ.get('CLIENT_BOT_TOKEN', '') URL_PATTERN = re.compile(r'https?://(?:[\w-]+\.)?chess-results\.com/\S+') START_MSG = ( '*♟️ chessCalc — жеребьёвка следующего тура*\n\n' 'Я рассчитываю пары следующего тура шахматного турнира ' '_до официальной публикации_ на chess\-results\.com\.\n\n' 'Просто пришли мне ссылку на турнир — и я покажу, ' 'кто с кем играет в следующем туре\.\n\n' 'Пример ссылки:\n' '`https://chess\-results\.com/tnr1393124\.aspx?lan=11`\n\n' 'Работает для швейцарских турниров\. Движок: FIDE Dutch System \(bbpPairings\)\.' ) LOCALE = { 'ru': { 'player': 'Игрок', 'rating': 'Рейт', 'points': 'Оч', 'tb1': 'Доп1', 'title': '📋 *Тур {rnd}* — пары', 'bye': '— BYE —', 'footer_bbp': '\n_FIDE 2025 \(bbpPairings\)_', 'footer_fallback': '\n_Упрощённый Swiss \(bbpPairings недоступен\)_', }, '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\)_', }, } 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 = '') -> 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_name 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), '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 = '') -> 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 (normalize names: strip commas, collapse spaces) 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_norm and (player_norm == _n(p1.name) or player_norm == _n(p2.name)): player_board = board_idx # Store the actual name from pairings for the ✅ marker if player_norm == _n(p1.name): player_name = p1.name else: player_name = 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 '') await update.message.reply_text(START_MSG, 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) had_pending = context.user_data.pop('pending_fide', None) if had_fide or had_pending: await update.message.reply_text('❎ Подписка отменена\.') else: await update.message.reply_text('Нет активного процесса подписки\.') async def my_players(update: Update, context: ContextTypes.DEFAULT_TYPE): user = update.effective_user subs = tracker.get_user_subs(user.id) if not subs: await update.message.reply_text('У вас нет активных подписок\.\n' '/addplayer — добавить игрока') return lines = ['*Ваши подписки:*'] for s in subs: safe_name = _md_escape(s['player_name']) safe_tour = _md_escape(s['tournament_name'] or '—') last_pts_str = _md_escape(f'{s["last_points"]:.1f}') lines.append( f' {s["id"]}\\. {safe_name} \({s["rating"]}, {s["fed"]}\) ' f'— {safe_tour} — {last_pts_str} очк\\.') lines.append('\n/removeplayer <номер> — удалить подписку') await update.message.reply_text('\n'.join(lines), parse_mode=ParseMode.MARKDOWN_V2) async def remove_player(update: Update, context: ContextTypes.DEFAULT_TYPE): user = update.effective_user parts = update.message.text.strip().split() if len(parts) < 2 or not parts[1].isdigit(): await update.message.reply_text( 'Использование: `/removeplayer <номер>`\n' '/myplayers — посмотреть список с номерами', parse_mode=ParseMode.MARKDOWN_V2) return sub_id = int(parts[1]) ok = tracker.remove_subscription(sub_id, user.id) if ok: await update.message.reply_text('✅ Подписка удалена\.') else: await update.message.reply_text('Подписка не найдена\. /myplayers — список') 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_str = f', {player["title"]}' if player.get('title') else '' await update.message.reply_text( f'✅ *{safe_name}*{title_str}\n' f'{"🇫 " if player["fed"] else ""}{player["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' \({t["start_date"]} — {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 tournaments found — fallback to manual URL context.user_data['pending_fide'] = { 'player': player, 'lang': lang_code, } await update.message.reply_text( '❌ Турниров не найдено автоматически\\.\n\n' 'Отправь ссылку на турнир chess\\-results\\.com, ' 'где играет этот игрок\\.\n' '/cancel — отмена', 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 _subscribe_to_tournament(update: Update, context, url: str, pending: dict): user = update.effective_user player = pending['player'] lang_code = pending['lang'] 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) context.user_data['pending_fide'] = pending return p = tracker.find_player(tournament, player['name']) if p is None: safe_name = _md_escape(player['name']) await msg.edit_text( f'❌ *{safe_name}* не найден в этом турнире\\.\n' 'Отправь другой URL или /cancel', parse_mode=ParseMode.MARKDOWN_V2) context.user_data['pending_fide'] = pending return p_sno = p.get('starting_sno', 0) results_count = len(p.get('results', [])) points = p.get('points', 0.0) tracker.add_subscription( user.id, player, url, tournament.get('name', ''), p_sno, results_count, points, lang_code) safe_name = _md_escape(player['name']) safe_tour = _md_escape(tournament.get('name', '')) await msg.edit_text( f'✅ *{safe_name}* найден в турнире\\.\n' f'📅 *{safe_tour}*\n' f'Ст\\.№ {p_sno}, {_md_escape(f"{points:.1f}")} очк\\. — отслеживаю\!\n\n' 'Я буду присылать результаты игрока и жеребьёвку следующих туров\.\n' '/myplayers — список подписок', 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 '') # State machine: waiting for FIDE ID → tournament URL → regular pairings # 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: player confirmed — waiting for tournament URL pending = context.user_data.get('pending_fide') if pending: 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`\n\n' '/cancel — отмена', parse_mode=ParseMode.MARKDOWN_V2) return url = match.group(0) context.user_data.pop('pending_fide', None) await _subscribe_to_tournament(update, context, url, pending) return # Step 3: 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(CommandHandler('removeplayer', remove_player)) app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_url)) app.job_queue.run_repeating( tracker.check_all_subscriptions, interval=300, first=10) app.job_queue.run_repeating( tracker.rescan_new_tournaments, interval=3600, first=60) 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()