import asyncio
import logging
import sqlite3
from datetime import datetime, timedelta
from typing import Dict, Any, Optional
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import (
Application, CommandHandler, CallbackQueryHandler,
MessageHandler, filters, ContextTypes, ConversationHandler
)
from config import (
TELEGRAM_BOT_TOKEN, PERIOD_OPTIONS, POLL_INTERVAL,
POLL_TIMEOUT, DROP_PENDING_UPDATES, ALLOWED_UPDATES,
LICHESS_STATS_API_BASE_URL, ADMINPANEL_TELEGRAM_BOT_TOKEN
)
from database import Database
from lichess_api import LichessAPI
from formatters import StatsFormatter
from i18n import t
from admin_bot import get_admin_bot, init_admin_bot
from message_counters import MessageCounters
# Configure logging
logging.basicConfig(
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level=logging.DEBUG
)
logger = logging.getLogger(__name__)
# Conversation states
WAITING_FOR_TOKEN, WAITING_FOR_USERNAME = range(2)
class LichessBot:
def __init__(self):
self.db = Database()
self.lichess_api = LichessAPI()
self.periodic_tasks = {} # Store periodic tasks
self.period_start_times = {} # Store start times for each gamer
self.application = None # Will be set when application is created
self.counters = MessageCounters() # Message counters
def get_user_language_from_update(self, update: Update) -> str:
"""Always return English language"""
# Update user info in database
user = update.effective_user
if user:
self.db.add_or_get_telegram_user(
user_id=user.id,
username=user.username,
first_name=user.first_name,
last_name=user.last_name,
language_code=None
)
return 'en'
async def start_existing_periodic_tasks(self):
"""Start periodic tasks for all user-gamer pairs that have periods set"""
try:
gamers_with_periods = self.db.get_all_gamers_with_periods()
logger.info(f"Found {len(gamers_with_periods)} user-gamer pairs with periodic notifications")
for gamer in gamers_with_periods:
if gamer['period_minutes'] > 0:
user_id = gamer['user_id']
# Start periodic task with user_id and gamer
await self.start_periodic_task(gamer, user_id, gamer['period_minutes'])
logger.info(f"Started periodic task for {gamer['username']} (user {user_id}) with period {gamer['period_minutes']} minutes")
# Start daily counter reset task
asyncio.create_task(self.daily_counter_reset_task())
logger.info("Started daily counter reset task")
except Exception as e:
logger.error(f"Error starting existing periodic tasks: {e}")
async def daily_counter_reset_task(self):
"""Background task to reset daily counters at midnight"""
while True:
try:
# Calculate seconds until next midnight
now = datetime.now()
next_midnight = (now + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
seconds_until_midnight = (next_midnight - now).total_seconds()
logger.info(f"Daily counter reset task: waiting {seconds_until_midnight} seconds until next midnight")
await asyncio.sleep(seconds_until_midnight)
# Reset daily counters
self.counters._reset_daily_counters_if_needed()
logger.info("Daily counters reset at midnight")
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"Error in daily counter reset task: {e}")
import traceback
logger.error(traceback.format_exc())
# Wait 1 hour before retrying
await asyncio.sleep(3600)
async def start(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Start command handler"""
logger.info(f"📝 start() method called for user {update.effective_user.id}")
# Register user in database
user = update.effective_user
lang_code = user.language_code if user else None
logger.info(f"User info: id={user.id}, username={user.username}, lang_code={lang_code}")
is_new_user = self.db.add_or_get_telegram_user(
user_id=user.id,
username=user.username,
first_name=user.first_name,
last_name=user.last_name,
language_code=lang_code
)
# Notify admin bot about new user
if is_new_user:
admin_bot = get_admin_bot()
if admin_bot:
await admin_bot.notify_new_user(
user_id=user.id,
username=user.username,
first_name=user.first_name
)
lang = self.get_user_language_from_update(update)
start_msg = t('start_message', lang)
await update.message.reply_text(start_msg)
self.counters.increment('start')
async def start_and_addgamer(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Start command that shows welcome message and starts addgamer conversation"""
try:
# Run the regular start command
await self.start(update, context)
# Start addgamer conversation and return state
return await self.addgamer_start(update, context)
except Exception as e:
logger.error(f"Error in start_and_addgamer: {e}")
import traceback
logger.error(traceback.format_exc())
try:
await update.message.reply_text(f"Error: {e}")
except:
pass
return ConversationHandler.END
async def addgamer_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Start addgamer command - simple username only"""
logger.info(f"addgamer_start called for user {update.effective_user.id}")
lang = self.get_user_language_from_update(update)
try:
await update.message.reply_text(t('addgamer_prompt', lang))
logger.info(f"Addgamer prompt sent to user {update.effective_user.id}")
self.counters.increment('addgamer')
except Exception as e:
logger.error(f"Error sending addgamer prompt: {e}")
import traceback
logger.error(traceback.format_exc())
return WAITING_FOR_USERNAME
async def addtoken_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Start addtoken command - token required"""
lang = self.get_user_language_from_update(update)
await update.message.reply_text(t('addtoken_prompt', lang))
self.counters.increment('addtoken')
return WAITING_FOR_TOKEN
async def handle_token(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle token input for /addtoken"""
token = update.message.text.strip()
user_id = update.effective_user.id
# Get username from token
profile = await self.lichess_api.get_user_profile(token)
if profile:
username = profile.get('username')
if username:
# Check if this gamer is already tracked by this user
user_gamers = self.db.get_user_gamers(user_id)
existing_gamer = next((g for g in user_gamers if g['username'] == username), None)
lang = self.get_user_language_from_update(update)
if existing_gamer:
# Update token for existing gamer
self.db.add_user_gamer(user_id, existing_gamer['id'], token)
await update.message.reply_text(
t('token_added', lang, username=username)
)
else:
# Add new gamer and link with token
# Check if gamer already exists
import sqlite3
with sqlite3.connect(self.db.db_path) as conn:
cursor = conn.cursor()
cursor.execute("SELECT id FROM gamers WHERE username = ?", (username,))
existing_gamer = cursor.fetchone()
is_new_gamer = existing_gamer is None
gamer_id = self.db.add_gamer(username)
self.db.add_user_gamer(user_id, gamer_id, token)
# Set default period to 1 hour (60 minutes) for new gamer
self.db.set_user_gamer_period(user_id, gamer_id, 60)
# If this is the first gamer for this user, make it active
user_gamers = self.db.get_user_gamers(user_id)
if len(user_gamers) == 1:
self.db.set_user_active_gamer(user_id, gamer_id)
# Notify admin bot about new player (only if it's a new gamer)
if is_new_gamer:
admin_bot = get_admin_bot()
if admin_bot:
user_obj = update.effective_user
await admin_bot.notify_new_player(
player_username=username,
added_by_user_id=user_id,
added_by_username=user_obj.username if user_obj else None
)
await update.message.reply_text(
t('gamer_added_with_token', lang, username=username)
)
return ConversationHandler.END
else:
lang = self.get_user_language_from_update(update)
await update.message.reply_text(
t('token_username_error', lang)
)
return WAITING_FOR_TOKEN
else:
lang = self.get_user_language_from_update(update)
await update.message.reply_text(
t('invalid_token', lang)
)
return WAITING_FOR_TOKEN
async def handle_username(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle username input for /addgamer"""
username = update.message.text.strip()
user_id = update.effective_user.id
lang = self.get_user_language_from_update(update)
if not username:
await update.message.reply_text(
t('empty_username', lang)
)
return WAITING_FOR_USERNAME
# Check if user exists on Lichess
user_exists = await self.lichess_api.check_user_exists(username)
if not user_exists:
await update.message.reply_text(
t('user_not_found', lang, username=username)
)
return WAITING_FOR_USERNAME
# Add gamer to database (without token)
# Check if gamer already exists
import sqlite3
with sqlite3.connect(self.db.db_path) as conn:
cursor = conn.cursor()
cursor.execute("SELECT id FROM gamers WHERE username = ?", (username,))
existing_gamer = cursor.fetchone()
is_new_gamer = existing_gamer is None
gamer_id = self.db.add_gamer(username)
# Link user to gamer (without token)
self.db.add_user_gamer(user_id, gamer_id, None)
# Set default period to 1 hour (60 minutes) for new gamer
self.db.set_user_gamer_period(user_id, gamer_id, 60)
# If this is the first gamer for this user, make it active
user_gamers = self.db.get_user_gamers(user_id)
if len(user_gamers) == 1:
self.db.set_user_active_gamer(user_id, gamer_id)
# Notify admin bot about new player (only if it's a new gamer)
if is_new_gamer:
logger.info(f"New gamer detected: {username}, notifying admin bot...")
admin_bot = get_admin_bot()
if admin_bot:
user_obj = update.effective_user
try:
await admin_bot.notify_new_player(
player_username=username,
added_by_user_id=user_id,
added_by_username=user_obj.username if user_obj else None
)
logger.info(f"Admin bot notification sent for player {username}")
except Exception as e:
logger.error(f"Failed to notify admin bot: {e}")
import traceback
logger.error(traceback.format_exc())
else:
logger.warning("Admin bot not available for notification")
else:
logger.info(f"Gamer {username} already exists, skipping admin notification")
lang = self.get_user_language_from_update(update)
await update.message.reply_text(
t('gamer_added', lang, username=username)
)
return ConversationHandler.END
async def getgamers(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Get all gamers for the current user and allow selection"""
user_id = update.effective_user.id
gamers = self.db.get_user_gamers(user_id)
logger.info(f"getgamers: user_id={user_id}, found {len(gamers)} gamers")
lang = self.get_user_language_from_update(update)
if not gamers:
await update.message.reply_text(t('no_gamers', lang))
self.counters.increment('getgamers')
return
self.counters.increment('getgamers')
# Show loading message
loading_msg = await update.message.reply_text(t('loading_ratings', lang))
# Prepare data for each gamer
gamers_data = []
for i, gamer in enumerate(gamers):
try:
logger.info(f"Processing gamer {i+1}/{len(gamers)}: {gamer['username']} (ID: {gamer['id']})")
status = "🟢" if gamer['is_active'] else "⚪"
username = gamer['username']
# Get user ratings from Lichess API
logger.debug(f"Requesting ratings for {username}")
ratings_data = await self.lichess_api.get_user_ratings(username)
logger.debug(f"Received ratings for {username}: {ratings_data is not None}")
# Add delay between requests to avoid rate limiting
if i < len(gamers) - 1: # Don't sleep after the last one
await asyncio.sleep(0.5) # 500ms delay between requests
if ratings_data and 'perfs' in ratings_data:
perfs = ratings_data['perfs']
bullet_rating = perfs.get('bullet', {}).get('rating', 'N/A')
blitz_rating = perfs.get('blitz', {}).get('rating', 'N/A')
rapid_rating = perfs.get('rapid', {}).get('rating', 'N/A')
else:
bullet_rating = blitz_rating = rapid_rating = 'N/A'
# Add period information if period > 0
period_minutes = gamer.get('period_minutes', 0)
period_suffix = t('period_minutes_suffix', lang)
period_text = f" · {period_minutes}{period_suffix}" if period_minutes > 0 else ""
gamers_data.append({
'id': gamer['id'],
'status': status,
'username': username,
'bullet': bullet_rating,
'blitz': blitz_rating,
'rapid': rapid_rating,
'period': period_text
})
logger.info(f"Successfully added gamer {username} to gamers_data (total: {len(gamers_data)})")
except Exception as e:
logger.error(f"Error processing gamer {gamer.get('username', 'unknown')}: {e}")
import traceback
logger.error(f"Traceback: {traceback.format_exc()}")
# Still add the gamer with N/A ratings
period_minutes = gamer.get('period_minutes', 0)
period_suffix = t('period_minutes_suffix', lang)
period_text = f" · {period_minutes}{period_suffix}" if period_minutes > 0 else ""
gamers_data.append({
'id': gamer['id'],
'status': "🟢" if gamer['is_active'] else "⚪",
'username': gamer['username'],
'bullet': 'N/A',
'blitz': 'N/A',
'rapid': 'N/A',
'period': period_text
})
logger.info(f"Added gamer {gamer['username']} with N/A ratings due to error")
# Create text message with stats
text_lines = []
for gamer in gamers_data:
text_lines.append(
f"{gamer['status']} {gamer['username']} "
f"⚡ {gamer['bullet']} 🔥 {gamer['blitz']} 🐇 {gamer['rapid']}{gamer['period']}"
)
logger.info(f"getgamers: prepared {len(gamers_data)} gamers for display")
gamers_text = t('select_active_gamer', lang) + "\n".join(text_lines)
logger.info(f"getgamers: message length: {len(gamers_text)} characters")
# Create simple keyboard with just usernames
keyboard = []
for gamer in gamers_data:
keyboard.append([InlineKeyboardButton(
text=f"{gamer['status']} {gamer['username']}",
callback_data=f"select_{gamer['id']}"
)])
logger.debug(f"Added button for {gamer['username']} (ID: {gamer['id']})")
logger.info(f"getgamers: created {len(keyboard)} buttons in keyboard")
reply_markup = InlineKeyboardMarkup(keyboard)
# Edit the loading message with the results
try:
await loading_msg.edit_text(
gamers_text,
parse_mode='HTML',
reply_markup=reply_markup
)
except Exception as e:
logger.error(f"Error editing message: {e}")
# If edit fails, delete the loading message and send a new one
try:
await loading_msg.delete()
except:
pass
await update.message.reply_text(
gamers_text,
parse_mode='HTML',
reply_markup=reply_markup
)
async def select_gamer(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle gamer selection"""
query = update.callback_query
await query.answer()
user_id = query.from_user.id
gamer_id = int(query.data.split('_')[1])
# Get user gamers to find the selected one
gamers = self.db.get_user_gamers(user_id)
selected_gamer = next((g for g in gamers if g['id'] == gamer_id), None)
# Для callback query обновляем язык пользователя если есть в update
if update.effective_user:
self.db.add_or_get_telegram_user(
user_id=update.effective_user.id,
username=update.effective_user.username,
first_name=update.effective_user.first_name,
last_name=update.effective_user.last_name,
language_code=update.effective_user.language_code
)
lang = self.db.get_user_language(user_id)
if selected_gamer:
# Set active gamer for this user
self.db.set_user_active_gamer(user_id, gamer_id)
await query.edit_message_text(
t('active_gamer_set', lang, username=selected_gamer['username'])
)
else:
await query.edit_message_text(t('gamer_not_found', lang))
async def delgamer(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Show gamers list for deletion"""
user_id = update.effective_user.id
gamers = self.db.get_user_gamers(user_id)
lang = self.get_user_language_from_update(update)
if not gamers:
await update.message.reply_text(t('no_gamers_to_delete', lang))
self.counters.increment('delgamer')
return
self.counters.increment('delgamer')
# Show loading message
loading_msg = await update.message.reply_text(t('loading_gamers', lang))
# Create text message with stats
text_lines = []
keyboard = []
for i, gamer in enumerate(gamers):
status = "🟢" if gamer['is_active'] else "⚪"
username = gamer['username']
# Get user ratings from Lichess API
ratings_data = await self.lichess_api.get_user_ratings(username)
# Add delay between requests to avoid rate limiting
if i < len(gamers) - 1: # Don't sleep after the last one
await asyncio.sleep(0.5) # 500ms delay between requests
if ratings_data and 'perfs' in ratings_data:
perfs = ratings_data['perfs']
bullet_rating = perfs.get('bullet', {}).get('rating', 'N/A')
blitz_rating = perfs.get('blitz', {}).get('rating', 'N/A')
rapid_rating = perfs.get('rapid', {}).get('rating', 'N/A')
else:
bullet_rating = blitz_rating = rapid_rating = 'N/A'
period_minutes = gamer.get('period_minutes', 0)
period_suffix = t('period_minutes_suffix', lang)
period_text = f" · {period_minutes}{period_suffix}" if period_minutes > 0 else ""
text_lines.append(
f"{status} {username} "
f"⚡ {bullet_rating} 🔥 {blitz_rating} 🐇 {rapid_rating}{period_text}"
)
# Add delete button
keyboard.append([InlineKeyboardButton(
text=f"🗑️ {username}",
callback_data=f"delete_{gamer['id']}"
)])
gamers_text = t('select_gamer_to_delete', lang) + "\n".join(text_lines)
reply_markup = InlineKeyboardMarkup(keyboard)
# Edit the loading message with the results
try:
await loading_msg.edit_text(
gamers_text,
parse_mode='HTML',
reply_markup=reply_markup
)
except Exception as e:
logger.error(f"Error editing message: {e}")
try:
await loading_msg.delete()
except:
pass
await update.message.reply_text(
gamers_text,
parse_mode='HTML',
reply_markup=reply_markup
)
async def handle_delete_gamer(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle gamer deletion"""
query = update.callback_query
await query.answer()
user_id = query.from_user.id
gamer_id = int(query.data.split('_')[1])
# Get gamer info before deletion
gamers = self.db.get_user_gamers(user_id)
gamer_to_delete = next((g for g in gamers if g['id'] == gamer_id), None)
# Для callback query получаем язык из БД
if update.effective_user:
self.db.add_or_get_telegram_user(
user_id=update.effective_user.id,
username=update.effective_user.username,
first_name=update.effective_user.first_name,
last_name=update.effective_user.last_name,
language_code=update.effective_user.language_code
)
lang = self.db.get_user_language(user_id)
if gamer_to_delete:
username = gamer_to_delete['username']
deleted = self.db.remove_user_gamer(user_id, gamer_id)
if deleted:
await query.edit_message_text(
t('gamer_deleted', lang, username=username),
parse_mode='HTML'
)
else:
await query.edit_message_text(t('delete_failed', lang))
else:
await query.edit_message_text(t('gamer_not_found', lang))
async def get_stats(self, update: Update, context: ContextTypes.DEFAULT_TYPE, period: str):
"""Get statistics for a period"""
user_id = update.effective_user.id
# Get active gamer for this user
active_gamer = self.db.get_user_active_gamer(user_id)
lang = self.get_user_language_from_update(update)
if not active_gamer:
await update.message.reply_text(
t('no_active_gamer', lang)
)
return
username = active_gamer['username']
# Get stats based on period
if period == "today":
data = await self.lichess_api.get_today_stats(username)
elif period == "yesterday":
data = await self.lichess_api.get_yesterday_stats(username)
elif period == "week":
data = await self.lichess_api.get_week_stats(username)
else:
await update.message.reply_text(t('unknown_period', lang))
return
# Format and send response
formatted_response = StatsFormatter.format_stats_response(data, username, period, lang)
await update.message.reply_text(formatted_response)
# Increment counter for the period command
if period == "today":
self.counters.increment('today')
elif period == "yesterday":
self.counters.increment('yesterday')
elif period == "week":
self.counters.increment('week')
async def today(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Today command"""
await self.get_stats(update, context, "today")
async def yesterday(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Yesterday command"""
await self.get_stats(update, context, "yesterday")
async def week(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Week command"""
await self.get_stats(update, context, "week")
async def setperiod(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Set period command"""
user_id = update.effective_user.id
# Get active gamer for this user
active_gamer = self.db.get_user_active_gamer(user_id)
lang = self.get_user_language_from_update(update)
if not active_gamer:
await update.message.reply_text(
t('no_active_gamer', lang)
)
return
keyboard = []
for period in PERIOD_OPTIONS:
if period == 0:
button_text = t('disable_notifications', lang)
else:
button_text = t('period_minutes', lang, period=period)
keyboard.append([InlineKeyboardButton(button_text, callback_data=f"period_{period}")])
reply_markup = InlineKeyboardMarkup(keyboard)
await update.message.reply_text(
t('select_period', lang, username=active_gamer['username']),
reply_markup=reply_markup
)
self.counters.increment('setperiod')
async def select_period(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Handle period selection"""
query = update.callback_query
await query.answer()
user_id = query.from_user.id
period = int(query.data.split('_')[1])
# Get active gamer for this user
active_gamer = self.db.get_user_active_gamer(user_id)
# Для callback query получаем язык из БД
if update.effective_user:
self.db.add_or_get_telegram_user(
user_id=update.effective_user.id,
username=update.effective_user.username,
first_name=update.effective_user.first_name,
last_name=update.effective_user.last_name,
language_code=update.effective_user.language_code
)
lang = self.db.get_user_language(user_id)
if active_gamer:
# Set period for this user-gamer pair
self.db.set_user_gamer_period(user_id, active_gamer['id'], period)
if period == 0:
await query.edit_message_text(
t('notifications_disabled', lang, username=active_gamer['username'])
)
else:
await query.edit_message_text(
t('period_set', lang, period=period, username=active_gamer['username'])
)
# Start periodic task for this gamer (send to user's personal messages)
await self.start_periodic_task(active_gamer, user_id, period)
async def check_language(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Check and display current language settings"""
user = update.effective_user
if user:
# Обновляем язык из update
lang = self.get_user_language_from_update(update)
# Получаем язык из БД для отображения
db_lang = self.db.get_user_language(user.id)
# Получаем language_code из БД напрямую
with sqlite3.connect(self.db.db_path) as conn:
cursor = conn.cursor()
cursor.execute("SELECT language_code FROM telegram_users WHERE user_id = ?", (user.id,))
row = cursor.fetchone()
db_language_code = row[0] if row and row[0] else None
message = (
f"🌐 Language Info:\n"
f"Current language: {lang}\n"
f"DB language: {db_lang}\n"
f"DB language_code: {db_language_code}\n"
f"Update language_code: {user.language_code}\n\n"
f"Language used: {lang}\n\n"
f"Bot uses English language only."
)
await update.message.reply_text(message)
self.counters.increment('lang')
else:
await update.message.reply_text("❌ Failed to get user information")
async def reset_language(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
"""Reset user language in database - will be detected from /start"""
user = update.effective_user
if user:
# Сбрасываем язык в БД (устанавливаем NULL)
with sqlite3.connect(self.db.db_path) as conn:
cursor = conn.cursor()
cursor.execute(
"UPDATE telegram_users SET language_code = NULL WHERE user_id = ?",
(user.id,)
)
conn.commit()
# Language is always English
lang = 'en'
await update.message.reply_text(
"✅ Language reset! Bot uses English language only."
)
self.counters.increment('resetlang')
else:
await update.message.reply_text("❌ Failed to get user information")
async def start_periodic_task(self, gamer: Dict[str, Any], user_id: int, period_minutes: int):
"""Start periodic task for a gamer"""
task_key = f"{gamer['id']}_{user_id}"
# Cancel existing task if any
if task_key in self.periodic_tasks:
self.periodic_tasks[task_key].cancel()
logger.info(f"Cancelled existing periodic task for {gamer['username']}")
# Remove old start time
if task_key in self.period_start_times:
del self.period_start_times[task_key]
# Create new periodic task
task = asyncio.create_task(
self.periodic_check(gamer, user_id, period_minutes)
)
self.periodic_tasks[task_key] = task
async def periodic_check(self, gamer: Dict[str, Any], user_id: int, period_minutes: int):
"""Periodic check for gamer activity"""
task_key = f"{gamer['id']}_{user_id}"
# Запоминаем время начала отслеживания
start_time = datetime.now()
self.period_start_times[task_key] = start_time
logger.info(f"Started periodic monitoring for {gamer['username']} with {period_minutes} minute intervals")
while True:
try:
# Ждем заданное количество минут
await asyncio.sleep(period_minutes * 60)
# Получаем время начала периода
period_start = self.period_start_times.get(task_key, start_time)
now = datetime.now()
# Рассчитываем timestamps в миллисекундах
since_timestamp = int(period_start.timestamp() * 1000)
until_timestamp = int(now.timestamp() * 1000)
logger.info(f"Checking period for {gamer['username']}: {period_start} to {now}")
logger.info(f"Unix timestamps: since={since_timestamp}, until={until_timestamp}")
# Делаем запросы к API
games_url = f"{LICHESS_STATS_API_BASE_URL}/games/{gamer['username']}/period?since={since_timestamp}&until={until_timestamp}"
logger.info(f"🎮 GAMES API REQUEST: {games_url}")
games_data = await self.lichess_api.get_games_period(
gamer['username'], since_timestamp, until_timestamp
)
logger.info(f"Games API response: {games_data}")
puzzles_data = None
if gamer['token']:
puzzles_url = f"{LICHESS_STATS_API_BASE_URL}/puzzle/period?since={since_timestamp}&until={until_timestamp}&max=150"
logger.info(f"🧩 PUZZLES API REQUEST: {puzzles_url}")
puzzles_data = await self.lichess_api.get_puzzles_period(
gamer['token'], since_timestamp, until_timestamp, max_puzzles=150
)
logger.info(f"Puzzles API response: {puzzles_data}")
else:
logger.info(f"No token for {gamer['username']}, skipping puzzles API call")
# Проверяем наличие реальной активности
has_games = False
total_games = 0
total_losses = 0
if games_data and games_data.get('data'):
total_games = games_data.get('data', {}).get('total', {}).get('games_played', 0)
total_losses = games_data.get('data', {}).get('total', {}).get('losses', 0)
has_games = total_games > 0
has_puzzles = False
if puzzles_data and puzzles_data.get('data'):
total_puzzles = puzzles_data.get('data', {}).get('total_attempts', 0)
has_puzzles = total_puzzles > 0
# Детальное логирование для отладки
logger.info(f"Activity check for {gamer['username']}: has_games={has_games}, has_puzzles={has_puzzles}")
if games_data and games_data.get('data'):
total_games = games_data.get('data', {}).get('total', {}).get('games_played', 0)
total_losses = games_data.get('data', {}).get('total', {}).get('losses', 0)
logger.info(f"Games data: total_games={total_games}, total_losses={total_losses}")
if puzzles_data and puzzles_data.get('data'):
total_puzzles = puzzles_data.get('data', {}).get('total_attempts', 0)
logger.info(f"Puzzles data: total_attempts={total_puzzles}")
# Отправляем уведомление только если есть реальная активность
if has_games or has_puzzles:
try:
# Get user language from database
user_lang = self.db.get_user_language(user_id)
notification = StatsFormatter.format_period_notification(
gamer['username'], games_data, puzzles_data, period_minutes, lang=user_lang
)
if self.application:
try:
await self.application.bot.send_message(
chat_id=user_id,
text=notification
)
logger.info(f"Sent periodic notification for {gamer['username']} to user {user_id}")
# Обновляем время начала только после успешной отправки уведомления
self.period_start_times[task_key] = now
# Increment periodic notification counter
self.counters.increment('periodic_notification')
except Exception as e:
logger.error(f"Failed to send notification to user {user_id}: {e}")
# Не обновляем время начала при ошибке отправки
except Exception as e:
logger.error(f"Error formatting notification for {gamer['username']}: {e}")
import traceback
logger.error(f"Traceback: {traceback.format_exc()}")
# Не обновляем время начала при ошибке форматирования
else:
logger.info(f"No activity found for {gamer['username']} in the last {period_minutes} minutes")
# Обновляем время начала даже если нет активности, чтобы не зацикливаться
self.period_start_times[task_key] = now
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"Error in periodic check: {e}")
import traceback
logger.error(f"Full traceback: {traceback.format_exc()}")
def setup_handlers(self, application: Application):
"""Setup all handlers"""
self.application = application # Store application reference
# Conversation handler for addgamer (simple username only)
addgamer_conv = ConversationHandler(
entry_points=[
CommandHandler("addgamer", self.addgamer_start),
CommandHandler("start", self.start_and_addgamer) # Also handle /start to start addgamer flow
],
states={
WAITING_FOR_USERNAME: [MessageHandler(filters.TEXT & ~filters.COMMAND, self.handle_username)],
},
fallbacks=[CommandHandler("cancel", lambda u, c: ConversationHandler.END)],
name="addgamer_conversation"
)
# Conversation handler for addtoken (token required)
addtoken_conv = ConversationHandler(
entry_points=[CommandHandler("addtoken", self.addtoken_start)],
states={
WAITING_FOR_TOKEN: [MessageHandler(filters.TEXT & ~filters.COMMAND, self.handle_token)],
},
fallbacks=[CommandHandler("cancel", lambda u, c: ConversationHandler.END)]
)
# Add conversation handlers
application.add_handler(addgamer_conv)
application.add_handler(addtoken_conv)
application.add_handler(CommandHandler("getgamers", self.getgamers))
application.add_handler(CommandHandler("delgamer", self.delgamer))
application.add_handler(CommandHandler("today", self.today))
application.add_handler(CommandHandler("yesterday", self.yesterday))
application.add_handler(CommandHandler("week", self.week))
application.add_handler(CommandHandler("setperiod", self.setperiod))
application.add_handler(CommandHandler("lang", self.check_language))
application.add_handler(CommandHandler("resetlang", self.reset_language))
# Callback handlers
application.add_handler(CallbackQueryHandler(self.select_gamer, pattern="^select_"))
application.add_handler(CallbackQueryHandler(self.handle_delete_gamer, pattern="^delete_"))
application.add_handler(CallbackQueryHandler(self.select_period, pattern="^period_"))
def main():
"""Main function"""
# Initialize admin bot for notifications (admin bot runs as separate service)
init_admin_bot()
bot = LichessBot()
# Create application with Long Polling configuration
application = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
# Setup handlers
bot.setup_handlers(application)
# Set application reference for periodic tasks
bot.application = application
# Start periodic tasks for existing gamers (will be called after application starts)
async def post_init(app):
await bot.start_existing_periodic_tasks()
application.post_init = post_init
# Add error handler
async def error_handler(update: object, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Log the error and send a telegram message to notify the developer."""
logger.error(f"Exception while handling an update: {context.error}")
import traceback
logger.error(traceback.format_exc())
application.add_error_handler(error_handler)
# Start the bot with Long Polling
logger.info("Starting Lichess Statistics Bot with Long Polling...")
try:
application.run_polling(
poll_interval=POLL_INTERVAL,
timeout=POLL_TIMEOUT,
drop_pending_updates=DROP_PENDING_UPDATES,
allowed_updates=ALLOWED_UPDATES
)
except Exception as e:
logger.error(f"Fatal error in run_polling: {e}")
import traceback
logger.error(traceback.format_exc())
raise
if __name__ == '__main__':
main()