import os import telebot from telebot import types import requests from bs4 import BeautifulSoup import sqlite3 import pandas as pd from apscheduler.schedulers.background import BackgroundScheduler API_TOKEN = os.environ.get("TG_TOKEN") bot = telebot.TeleBot(API_TOKEN, parse_mode='Markdown') # Установка команд в меню бота bot.set_my_commands([ telebot.types.BotCommand("/start", "Начать работу с ботом"), telebot.types.BotCommand("/delete_fshr_id", "Удалить ФШР ID"), telebot.types.BotCommand("/delete_lichess_username", "Удалить ник Lichess"), telebot.types.BotCommand("/bot_info", "Информация о боте"), ]) # Загрузка норм norms_df = pd.read_csv("normy.csv") # Инициализация базы данных conn = sqlite3.connect("data/users.db", check_same_thread=False) cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS users ( chat_id INTEGER PRIMARY KEY, fshr_id TEXT, lichess_nick TEXT, last_rating_text TEXT ) ''') conn.commit() user_states = {} user_data = {} def get_main_menu(): markup = types.ReplyKeyboardMarkup(resize_keyboard=True) markup.add(types.KeyboardButton("МОЙ РЕЙТИНГ ФШР")) markup.add(types.KeyboardButton("МОЙ РЕЙТИНГ LICHESS")) markup.add(types.KeyboardButton("ПРОВЕРИТЬ РАЗРЯДНУЮ НОРМУ")) return markup # --- РЕЙТИНГИ ФШР и ФИДЕ --- def parse_rating_blocks(soup): fshr_ratings = {} fide_ratings = {} panels = soup.find_all("div", class_="panel panel-default") for panel in panels: items = panel.find_all("li", class_="list-group-item") for item in items: text = item.get_text(strip=True, separator=" ") if "рейтинг" in text.lower() and "место" in text.lower(): parts = text.split("рейтинг:") if len(parts) > 1: rating_part = parts[1].split("место")[0].strip().rstrip(',') if "Классические" in text: fshr_ratings["Классика"] = rating_part elif "Быстрые" in text: fshr_ratings["Рапид"] = rating_part elif "Блиц" in text: fshr_ratings["Блиц"] = rating_part fide_panel = soup.find("div", class_="panel panel-info") if fide_panel: fide_items = fide_panel.find_all("li", class_="list-group-item") for item in fide_items: if "РЕЙТИНГ" in item.get_text(strip=True).upper(): spans = item.find_all("span") for span in spans: text = span.get_text(strip=True) if text.startswith("std"): fide_ratings["Классика"] = text[4:] elif text.startswith("rpd"): fide_ratings["Рапид"] = text[4:] elif text.startswith("blz"): fide_ratings["Блиц"] = text[4:] return fshr_ratings, fide_ratings def format_ratings(fshr, fide): lines = [] if fshr: lines.append("*Твой рейтинг ФШР:*") for k in ["Классика", "Рапид", "Блиц"]: if k in fshr: lines.append(f"{k}: {fshr[k]}") lines.append("") if fide: lines.append("*Твой рейтинг ФИДЕ:*") for k in ["Классика", "Рапид", "Блиц"]: if k in fide: lines.append(f"{k}: {fide[k]}") return "\n".join(lines) def get_fshr_rating(fshr_id): url = f"https://ratings.ruchess.ru/people/{fshr_id}" headers = {"User-Agent": "Mozilla/5.0"} try: response = requests.get(url, headers=headers) response.raise_for_status() soup = BeautifulSoup(response.text, 'html.parser') return parse_rating_blocks(soup) except: return {}, {} def save_user(chat_id, fshr_id=None, lichess_nick=None, rating_text=None): cursor.execute(''' INSERT INTO users (chat_id, fshr_id, lichess_nick, last_rating_text) VALUES (?, ?, ?, ?) ON CONFLICT(chat_id) DO UPDATE SET fshr_id=COALESCE(excluded.fshr_id, fshr_id), lichess_nick=COALESCE(excluded.lichess_nick, lichess_nick), last_rating_text=COALESCE(excluded.last_rating_text, last_rating_text) ''', (chat_id, fshr_id, lichess_nick, rating_text)) conn.commit() def get_user(chat_id): cursor.execute('SELECT fshr_id, lichess_nick, last_rating_text FROM users WHERE chat_id = ?', (chat_id,)) return cursor.fetchone() def compare_ratings(old_text, new_text): changes = [] if not old_text: return changes old_lines = {line.split(":")[0].strip(): line.split(":")[-1].strip() for line in old_text.splitlines() if ":" in line and "рейтинг" in line} new_lines = {line.split(":")[0].strip(): line.split(":")[-1].strip() for line in new_text.splitlines() if ":" in line and "рейтинг" in line} for k in new_lines: if k in old_lines and old_lines[k] != new_lines[k]: changes.append(f"Изменился рейтинг {k}. Был {old_lines[k]}, стал {new_lines[k]}.") return changes @bot.message_handler(commands=['start']) def send_welcome(message): markup = get_main_menu() bot.send_message(message.chat.id, "Привет! Нажми кнопку, чтобы получить рейтинг или проверить норму.", reply_markup=markup) @bot.message_handler(func=lambda m: m.text == "МОЙ РЕЙТИНГ ФШР") def send_my_rating(message): user = get_user(message.chat.id) markup = get_main_menu() if user and user[0]: fshr_id = user[0] fshr, fide = get_fshr_rating(fshr_id) text = format_ratings(fshr, fide) save_user(message.chat.id, fshr_id=fshr_id, rating_text=text) bot.send_message(message.chat.id, text, reply_markup=markup) else: user_states[message.chat.id] = "waiting_for_id" bot.send_message(message.chat.id, "Введи свой ID ФШР", reply_markup=markup) @bot.message_handler(func=lambda m: user_states.get(m.chat.id) == "waiting_for_id") def handle_id_input(message): user_states[message.chat.id] = None fshr_id = message.text.strip() fshr, fide = get_fshr_rating(fshr_id) markup = get_main_menu() if fshr or fide: text = format_ratings(fshr, fide) save_user(message.chat.id, fshr_id=fshr_id, rating_text=text) bot.send_message(message.chat.id, text, reply_markup=markup) else: bot.send_message(message.chat.id, "Не удалось получить рейтинг. Проверь ID.", reply_markup=markup) # --- LICHESS --- def get_lichess_ratings(username): url = f"https://lichess.org/api/user/{username}" headers = {"User-Agent": "Mozilla/5.0"} try: response = requests.get(url, headers=headers) response.raise_for_status() data = response.json() perfs = data.get("perfs", {}) ratings = {} for mode, label in [("classical", "Классика"), ("rapid", "Рапид"), ("blitz", "Блиц"), ("bullet", "Пуля")]: if mode in perfs: ratings[label] = perfs[mode].get("rating") return ratings except: return {} def format_lichess_ratings(username, ratings): lines = [f"*Твой рейтинг Lichess (@{username}):*"] for k in ratings: lines.append(f"{k}: {ratings[k]}") return "\n".join(lines) @bot.message_handler(func=lambda m: m.text == "МОЙ РЕЙТИНГ LICHESS") def send_lichess_rating(message): user = get_user(message.chat.id) markup = get_main_menu() if user and user[1]: ratings = get_lichess_ratings(user[1]) if ratings: bot.send_message(message.chat.id, format_lichess_ratings(user[1], ratings), reply_markup=markup) else: bot.send_message(message.chat.id, "Не удалось получить рейтинг.", reply_markup=markup) else: user_states[message.chat.id] = "waiting_for_lichess_nick" bot.send_message(message.chat.id, "Какой у тебя ник на Lichess?", reply_markup=markup) @bot.message_handler(func=lambda m: user_states.get(m.chat.id) == "waiting_for_lichess_nick") def handle_lichess_nick(message): user_states[message.chat.id] = None username = message.text.strip() ratings = get_lichess_ratings(username) markup = get_main_menu() if ratings: save_user(message.chat.id, lichess_nick=username) bot.send_message(message.chat.id, format_lichess_ratings(username, ratings), reply_markup=markup) else: bot.send_message(message.chat.id, "Не удалось получить рейтинг. Возможно, ник неверен.", reply_markup=markup) # --- ПРОВЕРКА НОРМЫ --- @bot.message_handler(func=lambda m: m.text == "ПРОВЕРИТЬ РАЗРЯДНУЮ НОРМУ") def start_norm_check(message): user_states[message.chat.id] = "waiting_for_chess_type" markup = types.ReplyKeyboardMarkup(resize_keyboard=True) markup.add("Классика", "Рапид", "Блиц") bot.send_message(message.chat.id, "Выбери вид шахмат:", reply_markup=markup) @bot.message_handler(func=lambda m: user_states.get(m.chat.id) == "waiting_for_chess_type") def handle_chess_type(message): user_data[message.chat.id] = {"Вид шахмат": message.text} user_states[message.chat.id] = "waiting_for_games" remove_markup = types.ReplyKeyboardRemove() bot.send_message(message.chat.id, "Сколько было сыграно партий?", reply_markup=remove_markup) @bot.message_handler(func=lambda m: user_states.get(m.chat.id) == "waiting_for_games") def handle_games(message): try: games = int(message.text) user_data[message.chat.id]["Количество партий"] = games # Проверка: есть ли вообще такая норма с таким количеством партий и выбранным видом шахмат chess_type = user_data[message.chat.id]["Вид шахмат"] min_required_df = norms_df[ (norms_df["Вид шахмат"] == chess_type) & (norms_df["Количество партий"] == games) ] if min_required_df.empty: user_states[message.chat.id] = None user_data.pop(message.chat.id, None) bot.send_message( message.chat.id, "❌ Норма не выполнена — не хватает партий.", reply_markup=get_main_menu() ) return user_states[message.chat.id] = "waiting_for_points" bot.send_message(message.chat.id, "Сколько очков набрано?") except: bot.send_message(message.chat.id, "Введите число.") @bot.message_handler(func=lambda m: user_states.get(m.chat.id) == "waiting_for_points") def handle_points(message): try: points = float(message.text.replace(",", ".")) user_data[message.chat.id]["Количество очков"] = points user_states[message.chat.id] = "waiting_for_avg" bot.send_message( message.chat.id, "Какой средний рейтинг соперников?\n\nЕсли не знаешь средний рейтинг, напиши через пробел рейтинги всех соперников, и я посчитаю сам 🙂" ) except: bot.send_message(message.chat.id, "Введите число.") @bot.message_handler(func=lambda m: user_states.get(m.chat.id) == "waiting_for_avg") def handle_avg(message): try: text = message.text.replace(",", ".") numbers = [float(x) for x in text.split() if x.replace('.', '', 1).isdigit()] if not numbers: raise ValueError("Нет чисел") avg = int(numbers[0]) if len(numbers) == 1 else int(sum(numbers) / len(numbers)) user_data[message.chat.id]["Средний рейтинг"] = avg user_states[message.chat.id] = "waiting_for_norm_gender" markup = types.ReplyKeyboardMarkup(resize_keyboard=True) markup.add("Мужская", "Женская") bot.send_message( message.chat.id, "Теперь выбери норму (мужская или женская):", reply_markup=markup ) except Exception as e: print("Ошибка при обработке среднего рейтинга:", e) bot.send_message(message.chat.id, "Введите число или список рейтингов через пробел (например: 1250 1300 1400).") @bot.message_handler(func=lambda m: user_states.get(m.chat.id) == "waiting_for_norm_gender") def handle_norm_gender(message): try: data = user_data.pop(message.chat.id) user_states[message.chat.id] = None data["Норма"] = message.text + " норма" avg = data["Средний рейтинг"] df = norms_df[ (norms_df["Норма"] == data["Норма"]) & (norms_df["Вид шахмат"] == data["Вид шахмат"]) & (norms_df["Количество партий"] == data["Количество партий"]) & (norms_df["Количество очков"] <= data["Количество очков"]) & (norms_df["Минимальный рейтинг"] <= avg) & (norms_df["Максимальный рейтинг"] >= avg) ] if df.empty: bot.send_message( message.chat.id, "❌ Норма не выполнена — не хватает партий или очков, или рейтинг соперников слишком низкий.", reply_markup=get_main_menu() ) else: row = df.sort_values(by=["Количество очков", "Минимальный рейтинг"], ascending=[False, False]).iloc[0] bot.send_message( message.chat.id, f"✅ Выполненная норма: *{row['Выполненная норма']}* (при {data['Количество партий']} партиях, {data['Количество очков']} очках и среднем рейтинге соперников {avg}).", reply_markup=get_main_menu() ) except Exception as e: print("Ошибка при финальной проверке нормы:", e) bot.send_message(message.chat.id, "Произошла ошибка. Попробуй ещё раз.") # --- ПЕРИОДИЧЕСКАЯ ПРОВЕРКА --- scheduler = BackgroundScheduler() @scheduler.scheduled_job('cron', hour='9,18') def check_for_updates(): cursor.execute('SELECT chat_id, fshr_id, last_rating_text FROM users') for chat_id, fshr_id, old_rating in cursor.fetchall(): fshr, fide = get_fshr_rating(fshr_id) new_text = format_ratings(fshr, fide) changes = compare_ratings(old_rating, new_text) if changes: msg = "\n".join(changes) + "\n\n" + new_text bot.send_message(chat_id, msg) save_user(chat_id, fshr_id=fshr_id, rating_text=new_text) scheduler.start() print("Бот запущен...") @bot.message_handler(commands=['delete_fshr_id']) def delete_fshr_id(message): cursor.execute('UPDATE users SET fshr_id = NULL, last_rating_text = NULL WHERE chat_id = ?', (message.chat.id,)) conn.commit() bot.send_message(message.chat.id, "✅ Твой ФШР ID удалён.") @bot.message_handler(commands=['delete_lichess_username']) def delete_lichess_username(message): cursor.execute('UPDATE users SET lichess_nick = NULL WHERE chat_id = ?', (message.chat.id,)) conn.commit() bot.send_message(message.chat.id, "✅ Твой ник на Lichess удалён.") @bot.message_handler(commands=['bot_info']) def bot_info(message): bot.send_message( message.chat.id, "🤖 *Вот, что я умею:*\n" "— Показываю твой рейтинг ФШР и ФИДЕ\n" "— Автоматически уведомляю об изменении рейтинга ФШР и ФИДЕ\n" "— Показываю рейтинг на Lichess\n" "— Проверяю выполнение нормы (разряд, КМС)\n" "Ты можешь использовать кнопки внизу или команды из меню 😉" ) bot.polling(none_stop=True)