376 lines
17 KiB
Python
376 lines
17 KiB
Python
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)
|
||
|