from aiogram import Bot, Dispatcher, types, F from aiogram.filters import Command, CommandObject from aiogram.fsm.context import FSMContext from aiogram.fsm.state import State, StatesGroup from aiogram.fsm.storage.memory import MemoryStorage import asyncio import aiofiles import fcntl import os import json import datetime import hashlib import aiosqlite from contextlib import asynccontextmanager import logging from asyncio import Lock import time from collections import OrderedDict # Настройки API_TOKEN = '8114676396:AAFtLSyueO-HBXmFclSc3UIJYJYCqbdhv4A' CHANNEL_USERNAME = '@donballon_news' DATABASE_PATH = 'votes.db' # Настройка логирования logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ # logging.FileHandler('/root/telegram_bot/server_bot_client.log', encoding='utf-8'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) # Инициализация бота bot = Bot(token=API_TOKEN) storage = MemoryStorage() dp = Dispatcher(storage=storage) # Глобальные блокировки для предотвращения гонки данных vote_locks = Lock() async def is_db_locked(): """Проверяет, заблокирована ли база данных (серверный бот онлайн)""" try: lock_file = f"{DATABASE_PATH}.lock" if not os.path.exists(lock_file): return False test_fd = os.open(lock_file, os.O_RDONLY) try: fcntl.flock(test_fd, fcntl.LOCK_SH | fcntl.LOCK_NB) fcntl.flock(test_fd, fcntl.LOCK_UN) os.close(test_fd) return False except IOError: os.close(test_fd) return True except Exception as e: logger.error(f"Ошибка проверки блокировки БД: {e}") return False async def wait_for_db_unlock(timeout=5): """Ждет разблокировки БД с таймаутом""" start_time = time.time() while time.time() - start_time < timeout: if not await is_db_locked(): return True await asyncio.sleep(0.1) return False class TimedLockManager: def __init__(self, max_size=1000, ttl=3600): self.locks = OrderedDict() self.max_size = max_size self.ttl = ttl async def get_lock(self, key): now = time.time() self._cleanup(now) if key not in self.locks: if len(self.locks) >= self.max_size: self.locks.popitem(last=False) self.locks[key] = (Lock(), now) return self.locks[key][0] def _cleanup(self, current_time): expired_keys = [] for key, (lock, created_at) in self.locks.items(): if current_time - created_at > self.ttl: expired_keys.append(key) for key in expired_keys: del self.locks[key] media_group_locks = TimedLockManager() class Database: def __init__(self, db_path): self.db_path = db_path self._connection = None async def get_connection(self): if self._connection is None: self._connection = await aiosqlite.connect(self.db_path) await self._connection.execute("PRAGMA journal_mode=WAL") await self._connection.execute("PRAGMA synchronous=NORMAL") return self._connection @asynccontextmanager async def get_cursor(self): conn = await self.get_connection() async with conn.cursor() as cursor: try: yield cursor await conn.commit() except Exception: await conn.rollback() raise db = Database(DATABASE_PATH) class VoteStates(StatesGroup): waiting_for_vote = State() async def init_database(): """Инициализация базы данных""" try: async with db.get_cursor() as cursor: await cursor.execute(''' CREATE TABLE IF NOT EXISTS active_votes ( id INTEGER PRIMARY KEY AUTOINCREMENT, created_at TEXT NOT NULL, variables TEXT NOT NULL, file_ids_with_types TEXT, end_time TEXT, channel_message_id INTEGER, finish_button_message_id INTEGER, is_active INTEGER DEFAULT 1, winners_count INTEGER DEFAULT 1 ) ''') await cursor.execute(''' CREATE TABLE IF NOT EXISTS user_votes ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER NOT NULL, vote_id INTEGER NOT NULL, option_name TEXT NOT NULL, voted_at TEXT NOT NULL, UNIQUE ( user_id, vote_id ) ) ''') await cursor.execute(''' CREATE TABLE IF NOT EXISTS subscribers ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER UNIQUE NOT NULL, subscribed_at TEXT NOT NULL ) ''') await cursor.execute(''' CREATE TABLE IF NOT EXISTS pending_subscribers ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER UNIQUE NOT NULL, created_at TEXT NOT NULL, message_ids TEXT, vote_ids TEXT ) ''') await cursor.execute('CREATE INDEX IF NOT EXISTS idx_user_votes_user_id ON user_votes(user_id)') await cursor.execute('CREATE INDEX IF NOT EXISTS idx_user_votes_vote_id ON user_votes(vote_id)') await cursor.execute('CREATE INDEX IF NOT EXISTS idx_subscribers_user_id ON subscribers(user_id)') await cursor.execute( 'CREATE INDEX IF NOT EXISTS idx_pending_subscribers_user_id ON pending_subscribers(user_id)') await cursor.execute( 'CREATE INDEX IF NOT EXISTS idx_pending_subscribers_created_at ON pending_subscribers(created_at)') logger.info("База данных инициализирована") except Exception as e: logger.error(f"Ошибка инициализации БД: {e}") raise async def add_user_to_pending(user_id, message_id, vote_id=None): """Добавляет пользователя в таблицу ожидающих подписки""" try: async with db.get_cursor() as cursor: # Проверяем, есть ли уже пользователь в ожидании await cursor.execute('SELECT message_ids, vote_ids FROM pending_subscribers WHERE user_id = ?', (user_id,)) existing = await cursor.fetchone() if existing: # Обновляем message_ids и vote_ids - добавляем новые данные к существующим existing_message_ids = json.loads(existing[0]) if existing[0] else [] existing_vote_ids = json.loads(existing[1]) if existing[1] and existing[1] != 'null' else [] if message_id not in existing_message_ids: existing_message_ids.append(message_id) if vote_id and vote_id not in existing_vote_ids: existing_vote_ids.append(vote_id) await cursor.execute( 'UPDATE pending_subscribers SET message_ids = ?, vote_ids = ?, created_at = ? WHERE user_id = ?', (json.dumps(existing_message_ids), json.dumps(existing_vote_ids), datetime.datetime.now().isoformat(), user_id) ) logger.info(f"Добавлен message_id {message_id} для пользователя {user_id} в ожидании подписки") else: # Добавляем нового пользователя vote_ids = [vote_id] if vote_id else [] await cursor.execute( 'INSERT INTO pending_subscribers (user_id, created_at, message_ids, vote_ids) VALUES (?, ?, ?, ?)', (user_id, datetime.datetime.now().isoformat(), json.dumps([message_id]), json.dumps(vote_ids)) ) logger.info(f"Пользователь {user_id} добавлен в ожидание подписки с message_id {message_id}") except Exception as e: logger.error(f"Ошибка добавления пользователя {user_id} в ожидание подписки: {e}") async def remove_user_from_pending(user_id): """Удаляет пользователя из таблицы ожидающих подписки""" try: async with db.get_cursor() as cursor: await cursor.execute('DELETE FROM pending_subscribers WHERE user_id = ?', (user_id,)) logger.info(f"Пользователь {user_id} удален из ожидания подписки") except Exception as e: logger.error(f"Ошибка удаления пользователя {user_id} из ожидания подписки: {e}") async def get_pending_user_message_ids(user_id): """Получает все message_ids для пользователя из таблицы ожидания""" try: async with db.get_cursor() as cursor: await cursor.execute('SELECT message_ids FROM pending_subscribers WHERE user_id = ?', (user_id,)) result = await cursor.fetchone() if result and result[0]: return json.loads(result[0]) return [] except Exception as e: logger.error(f"Ошибка получения message_ids для пользователя {user_id}: {e}") return [] async def get_pending_user_vote_ids(user_id): """Получает все vote_ids для пользователя из таблицы ожидания""" try: async with db.get_cursor() as cursor: await cursor.execute('SELECT vote_ids FROM pending_subscribers WHERE user_id = ?', (user_id,)) result = await cursor.fetchone() if result and result[0] and result[0] != 'null': return json.loads(result[0]) return [] except Exception as e: logger.error(f"Ошибка получения vote_ids для пользователя {user_id}: {e}") return [] async def cleanup_old_pending_users(): """Удаляет старых пользователей из таблицы ожидания (старше 10 минут)""" try: ten_minutes_ago = (datetime.datetime.now() - datetime.timedelta(minutes=10)).isoformat() async with db.get_cursor() as cursor: await cursor.execute('DELETE FROM pending_subscribers WHERE created_at < ?', (ten_minutes_ago,)) deleted_count = cursor.rowcount if deleted_count > 0: logger.info(f"Удалено {deleted_count} старых пользователей из ожидания подписки") except Exception as e: logger.error(f"Ошибка очистки старых пользователей из ожидания: {e}") async def delete_all_subscription_messages(user_id): """Удаляет ВСЕ сообщения о подписке для пользователя""" try: message_ids = await get_pending_user_message_ids(user_id) deleted_count = 0 for message_id in message_ids: try: await bot.delete_message(chat_id=user_id, message_id=message_id) deleted_count += 1 logger.info(f"Удалено сообщение {message_id} для пользователя {user_id}") except Exception as e: logger.warning(f"Не удалось удалить сообщение {message_id} для пользователя {user_id}: {e}") continue # Удаляем пользователя из ожидания после удаления всех сообщений await remove_user_from_pending(user_id) logger.info(f"Удалено {deleted_count} сообщений о подписке для пользователя {user_id}") except Exception as e: logger.error(f"Ошибка удаления сообщений о подписке для пользователя {user_id}: {e}") async def check_pending_subscriptions(): """Проверяет ожидающих пользователей и отправляет голосование если подписались""" try: async with db.get_cursor() as cursor: await cursor.execute('SELECT user_id, message_ids, vote_ids FROM pending_subscribers') pending_users = await cursor.fetchall() processed_users = set() # Для отслеживания уже обработанных пользователей for user_id, message_ids_json, vote_ids_json in pending_users: try: # Пропускаем если пользователь уже обработан if user_id in processed_users: continue processed_users.add(user_id) # Проверяем подписку subscribed = await is_subscribed(user_id, CHANNEL_USERNAME) if subscribed: # Получаем vote_ids для этого пользователя vote_ids = json.loads(vote_ids_json) if vote_ids_json and vote_ids_json != 'null' else [] # Удаляем ВСЕ сообщения о неподписке await delete_all_subscription_messages(user_id) # Если есть конкретные vote_ids, отправляем соответствующие голосования if vote_ids: for vote_id in vote_ids: if await is_vote_active_by_id(vote_id): # Проверяем, не голосовал ли уже пользователь в этом голосовании if not await has_user_voted_in_vote(user_id, vote_id): vote_data, votes_data, vote_hash = await get_vote_data_by_id(vote_id) if vote_data and vote_hash: await send_vote_to_user(user_id, vote_data, votes_data, vote_hash, vote_id) logger.info( f"Автоматически отправлено голосование {vote_id} пользователю {user_id}") else: # Если голосование не найдено, отправляем сообщение try: await bot.send_message( user_id, "Голосование завершено или не активно." ) except Exception as e: logger.error(f"Не удалось отправить сообщение пользователю {user_id}: {e}") else: # Пользователь уже голосовал - отправляем сообщение try: await bot.send_message( user_id, "Вы уже проголосовали в данном голосовании!" ) logger.info( f"Пользователь {user_id} уже голосовал в голосовании {vote_id} - отправлено уведомление") except Exception as e: logger.error(f"Не удалось отправить сообщение пользователю {user_id}: {e}") else: # Отправляем все активные голосования, в которых пользователь еще не голосовал active_votes = await get_active_votes() sent_votes = 0 for vote in active_votes: if not await has_user_voted_in_vote(user_id, vote['id']): vote_data, votes_data, vote_hash = await get_vote_data_by_id(vote['id']) if vote_data and vote_hash: await send_vote_to_user(user_id, vote_data, votes_data, vote_hash, vote['id']) sent_votes += 1 logger.info( f"Автоматически отправлено голосование {vote['id']} пользователю {user_id}") if sent_votes == 0: # Пользователь уже проголосовал во всех активных голосованиях try: await bot.send_message( user_id, "Вы уже проголосовали во всех активных голосованиях!" ) logger.info( f"Пользователь {user_id} уже проголосовал во всех голосованиях - отправлено уведомление") except Exception as e: logger.error(f"Не удалось отправить сообщение пользователю {user_id}: {e}") except Exception as e: logger.error(f"Ошибка проверки пользователя {user_id}: {e}") continue except Exception as e: logger.error(f"Ошибка проверки ожидающих подписок: {e}") async def start_pending_checker(): """Запускает периодическую проверку ожидающих пользователей""" while True: try: await check_pending_subscriptions() await cleanup_old_pending_users() await asyncio.sleep(30) # Проверка каждые 30 секунд except Exception as e: logger.error(f"Ошибка в pending_checker: {e}") await asyncio.sleep(30) async def is_vote_active_by_id(vote_id): """Проверяет, активно ли конкретное голосование по ID""" try: async with db.get_cursor() as cursor: await cursor.execute('SELECT id FROM active_votes WHERE id = ? AND is_active = 1', (vote_id,)) result = await cursor.fetchone() return result is not None except Exception as e: logger.error(f"Ошибка проверки активного голосования {vote_id}: {e}") return False async def get_active_votes(): """Получает список всех активных голосований""" try: async with db.get_cursor() as cursor: await cursor.execute('SELECT id, created_at, variables FROM active_votes WHERE is_active = 1') results = await cursor.fetchall() votes = [] for result in results: votes.append({ 'id': result[0], 'created_at': result[1], 'variables': result[2] }) return votes except Exception as e: logger.error(f"Ошибка получения активных голосований: {e}") return [] async def get_vote_hash_by_id(vote_id): """Генерирует короткий хеш конкретного голосования для проверки актуальности""" try: if not await is_vote_active_by_id(vote_id): return None async with db.get_cursor() as cursor: await cursor.execute( 'SELECT created_at, variables FROM active_votes WHERE id = ? AND is_active = 1', (vote_id,) ) result = await cursor.fetchone() if not result: return None created_at, variables_json = result vote_string = created_at + variables_json return hashlib.md5(vote_string.encode()).hexdigest()[:8] except Exception as e: logger.error(f"Ошибка генерации хеша голосования {vote_id}: {e}") return None @dp.chat_member() async def track_chat_member_update(chat_member: types.ChatMemberUpdated): """Отслеживание обновлений статуса участников""" try: user_id = chat_member.new_chat_member.user.id old_status = chat_member.old_chat_member.status new_status = chat_member.new_chat_member.status if (old_status in ['left', 'kicked'] and new_status == 'member'): async with db.get_cursor() as cursor: await cursor.execute( 'INSERT OR REPLACE INTO subscribers (user_id, subscribed_at) VALUES (?, ?)', (user_id, datetime.datetime.now().isoformat()) ) logger.info(f"Пользователь {user_id} подписался на канал") # Немедленно проверяем и отправляем голосование если пользователь в ожидании message_ids = await get_pending_user_message_ids(user_id) if message_ids: await check_pending_subscriptions() except Exception as e: logger.error(f"Ошибка обработки обновления статуса пользователя: {e}") async def get_date_subscriber(user_id): """Получает дату подписки пользователя""" try: async with db.get_cursor() as cursor: await cursor.execute( 'SELECT subscribed_at FROM subscribers WHERE user_id = ?', (user_id,) ) result = await cursor.fetchone() return result[0] if result else None except Exception as e: logger.error(f"Ошибка получения даты подписки для пользователя {user_id}: {e}") return None async def is_subscribed(user_id, channel): """Проверяет подписку пользователя на канал""" try: member = await bot.get_chat_member(chat_id=channel, user_id=user_id) if member.status in ["creator", "owner", "administrator"]: return True if member.status == 'member': # Сначала проверяем есть ли пользователь в таблице подписчиков subscribe_date = await get_date_subscriber(user_id) if not subscribe_date: # Если пользователя нет в таблице, но он подписан - добавляем его # Это означает что он подписался до включения бота async with db.get_cursor() as cursor: await cursor.execute( 'INSERT OR REPLACE INTO subscribers (user_id, subscribed_at) VALUES (?, ?)', (user_id, datetime.datetime.now().isoformat()) ) logger.info(f"Пользователь {user_id} добавлен в подписчики (подписан до включения бота)") return True # Убрана проверка на 2 минуты - теперь сразу возвращаем True если подписан return True return False except Exception as e: logger.error(f"Ошибка проверки подписки пользователя {user_id}: {e}") return False def create_callback_data(option_name, vote_hash, vote_id): """Создает безопасный callback_data с хешированием и ID голосования""" short_hash = vote_hash[:8] if vote_hash else "0000" option_hash = hashlib.md5(option_name.encode()).hexdigest()[:6] return f"v_{option_hash}_{short_hash}_{vote_id}" def parse_callback_data(callback_data, vote_data): """Парсит callback_data и находит соответствующий вариант""" if not callback_data.startswith("v_"): return None, None parts = callback_data.split('_') if len(parts) < 4: return None, None option_hash = parts[1] vote_hash = parts[2] vote_id = int(parts[3]) for variable in vote_data.get('variables', []): var_hash = hashlib.md5(variable['name'].encode()).hexdigest()[:6] if var_hash == option_hash: return variable['name'], vote_id return None, None async def get_vote_data_by_id(vote_id): """Получает данные голосования из базы данных по ID""" try: if not await is_vote_active_by_id(vote_id): return None, None, None async with db.get_cursor() as cursor: await cursor.execute( 'SELECT created_at, variables, file_ids_with_types FROM active_votes WHERE id = ? AND is_active = 1', (vote_id,) ) result = await cursor.fetchone() if not result: return None, None, None created_at, variables_json, file_ids_json = result vote_data = { 'id': vote_id, 'created_at': created_at, 'variables': json.loads(variables_json), 'file_ids_with_types': json.loads(file_ids_json) if file_ids_json else [] } await cursor.execute(''' SELECT option_name, COUNT(*) as vote_count FROM user_votes WHERE vote_id = ? GROUP BY option_name ''', (vote_id,)) votes_results = await cursor.fetchall() votes_data = {} for option_name, vote_count in votes_results: votes_data[option_name] = vote_count current_hash = await get_vote_hash_by_id(vote_id) return vote_data, votes_data, current_hash except Exception as e: logger.error(f"Ошибка чтения данных голосования {vote_id}: {e}") return None, None, None async def save_vote(option_name, user_id, vote_hash, vote_id): """Сохраняет голос пользователя с проверкой актуальности голосования""" try: async with db.get_cursor() as cursor: await cursor.execute(''' SELECT av.id, av.created_at, av.variables FROM active_votes av WHERE av.id = ? AND av.is_active = 1 ''', (vote_id,)) result = await cursor.fetchone() if not result: return False vote_db_id, created_at, variables_json = result current_hash = hashlib.md5((created_at + variables_json).encode()).hexdigest()[:8] if current_hash != vote_hash: return False await cursor.execute(''' INSERT OR IGNORE INTO user_votes (user_id, vote_id, option_name, voted_at) VALUES (?, ?, ?, ?) ''', (user_id, vote_id, option_name, datetime.datetime.now().isoformat())) success = cursor.rowcount > 0 if success: logger.info(f"Пользователь {user_id} проголосовал за '{option_name}' в голосовании {vote_id}") return success except Exception as e: logger.error(f"Ошибка сохранения голоса пользователя {user_id} в голосовании {vote_id}: {e}") return False async def has_user_voted_in_vote(user_id, vote_id): """Проверяет, голосовал ли уже пользователь в конкретном голосовании""" try: if not await is_vote_active_by_id(vote_id): return False async with db.get_cursor() as cursor: await cursor.execute(''' SELECT uv.id FROM user_votes uv WHERE uv.user_id = ? AND uv.vote_id = ? ''', (user_id, vote_id)) return await cursor.fetchone() is not None except Exception as e: logger.error(f"Ошибка проверки голоса пользователя {user_id} в голосовании {vote_id}: {e}") return False async def send_vote_to_user(user_id, vote_data, votes_data, vote_hash, vote_id): """Отправляет голосование пользователю с учетом типов медиа""" try: if not await is_vote_active_by_id(vote_id): await bot.send_message(user_id, "Голосование завершено или не активно.") return if 'file_ids_with_types' in vote_data and vote_data['file_ids_with_types']: media_group = [] for i, media_item in enumerate(vote_data['file_ids_with_types']): file_id = media_item['file_id'] media_type = media_item['type'] if i == 0: if media_type == 'photo': media_group.append(types.InputMediaPhoto( media=file_id, caption="Голосование от Дон Баллон\n\nВыберите вариант ответа:" )) elif media_type == 'video': media_group.append(types.InputMediaVideo( media=file_id, caption="Голосование от Дон Баллон\n\nВыберите вариант ответа:" )) else: if media_type == 'photo': media_group.append(types.InputMediaPhoto(media=file_id)) elif media_type == 'video': media_group.append(types.InputMediaVideo(media=file_id)) if media_group: await bot.send_media_group(chat_id=user_id, media=media_group) elif 'file_ids' in vote_data and vote_data['file_ids']: first_file_id = vote_data['file_ids'][0] try: await bot.send_photo( chat_id=user_id, photo=first_file_id, caption="Голосование от Дон Баллон\n\nВыберите вариант ответа:" ) except Exception as photo_error: try: await bot.send_video( chat_id=user_id, video=first_file_id, caption="Голосование от Дон Баллон\n\nВыберите вариант ответа:" ) except Exception as video_error: logger.error(f"Не удалось отправить медиа пользователю {user_id}: {video_error}") await bot.send_message(user_id, "Голосование от Дон Баллон\n\nВыберите вариант ответа:") else: await bot.send_message(user_id, "Голосование от Дон Баллон\n\nВыберите вариант ответа:") keyboard = types.InlineKeyboardMarkup(inline_keyboard=[]) for i, variable in enumerate(vote_data.get('variables', []), 1): button_text = f"{i}. {variable['name']}" callback_data = create_callback_data(variable['name'], vote_hash, vote_id) keyboard.inline_keyboard.append([ types.InlineKeyboardButton( text=button_text, callback_data=callback_data ) ]) await bot.send_message( user_id, "Выберите вариант для голосования:", reply_markup=keyboard ) logger.info(f"Голосование {vote_id} отправлено пользователю {user_id}") except Exception as e: logger.error(f"Ошибка отправки голосования {vote_id} пользователю {user_id}: {e}") await bot.send_message(user_id, "Произошла ошибка при загрузке голосования") @dp.message(Command('start')) async def cmd_start(message: types.Message, command: CommandObject, state: FSMContext): """Обработчик команды /start""" user_id = message.from_user.id if command.args and command.args.startswith("vote_"): try: vote_id = int(command.args.split("_")[1]) except (ValueError, IndexError): await message.answer("Некорректная ссылка на голосование.") return if not await is_vote_active_by_id(vote_id): await message.answer("Данное голосование завершено или не активно.") return subscribed = await is_subscribed(user_id, CHANNEL_USERNAME) if not subscribed: keyboard = types.InlineKeyboardMarkup(inline_keyboard=[[ types.InlineKeyboardButton( text="Перейти в канал", url=f"https://t.me/{CHANNEL_USERNAME[1:]}" ) ]]) sent_message = await message.answer( "Для участия в голосовании необходимо быть подписанным на канал.\n\n" "Подпишитесь, пожалуйста, на канал и нажмите на кнопку ниже:", reply_markup=keyboard ) # Добавляем пользователя в ожидание с конкретным vote_id await add_user_to_pending(user_id, sent_message.message_id, vote_id) return if await has_user_voted_in_vote(user_id, vote_id): await message.answer("Вы уже проголосовали в этом голосовании!") return vote_data, votes_data, vote_hash = await get_vote_data_by_id(vote_id) if not vote_data or not vote_hash: await message.answer("В настоящее время нет активных голосований.") return await send_vote_to_user(user_id, vote_data, votes_data, vote_hash, vote_id) await state.set_state(VoteStates.waiting_for_vote) logger.info(f"Пользователь {user_id} начал голосование {vote_id}") elif command.args and command.args == "lastvote": # Отправляем все активные голосования, в которых пользователь еще не голосовал active_votes = await get_active_votes() if not active_votes: await message.answer("В настоящее время нет активных голосований.") return subscribed = await is_subscribed(user_id, CHANNEL_USERNAME) if not subscribed: keyboard = types.InlineKeyboardMarkup(inline_keyboard=[[ types.InlineKeyboardButton( text="Перейти в канал", url=f"https://t.me/{CHANNEL_USERNAME[1:]}" ) ]]) sent_message = await message.answer( "Для участия в голосовании необходимо быть подписанным на канал.\n\n" "Подпишитесь, пожалуйста, на канал и нажмите на кнопку ниже:", reply_markup=keyboard ) # Добавляем пользователя в ожидание без конкретного vote_id (все голосования) await add_user_to_pending(user_id, sent_message.message_id) return sent_votes = 0 for vote in active_votes: if not await has_user_voted_in_vote(user_id, vote['id']): vote_data, votes_data, vote_hash = await get_vote_data_by_id(vote['id']) if vote_data and vote_hash: await send_vote_to_user(user_id, vote_data, votes_data, vote_hash, vote['id']) sent_votes += 1 logger.info(f"Отправлено голосование {vote['id']} пользователю {user_id}") if sent_votes == 0: await message.answer("Вы уже проголосовали во всех активных голосованиях!") else: await message.answer("Здравствуйте! Вас приветствует бот голосований DonBallon") @dp.callback_query(F.data == "check_subscription") async def check_subscription(callback: types.CallbackQuery, state: FSMContext): """Проверка подписки пользователя""" user_id = callback.from_user.id if not await is_db_locked(): await callback.message.edit_text("Сервер временно недоступен. Попробуйте позже.") return # Получаем vote_id из состояния или данных сообщения vote_ids = await get_pending_user_vote_ids(user_id) if not vote_ids: await callback.message.edit_text("Голосование завершено или не активно.") return subscribed = await is_subscribed(user_id, CHANNEL_USERNAME) if subscribed: # Удаляем ВСЕ сообщения о подписке для этого пользователя await delete_all_subscription_messages(user_id) sent_votes = 0 # Отправляем все голосования из vote_ids for vote_id in vote_ids: if await is_vote_active_by_id(vote_id): # Проверяем, не голосовал ли уже пользователь в этом голосовании if not await has_user_voted_in_vote(user_id, vote_id): vote_data, votes_data, vote_hash = await get_vote_data_by_id(vote_id) if vote_data and vote_hash: await callback.answer("Отлично! Загружаем голосование...", show_alert=False) await send_vote_to_user(user_id, vote_data, votes_data, vote_hash, vote_id) await state.set_state(VoteStates.waiting_for_vote) sent_votes += 1 logger.info(f"Пользователь {user_id} прошел проверку подписки и начал голосование {vote_id}") else: # Пользователь уже голосовал - отправляем сообщение try: await bot.send_message( user_id, "Вы уже проголосовали в данном голосовании!" ) logger.info( f"Пользователь {user_id} уже голосовал в голосовании {vote_id} - отправлено уведомление") except Exception as e: logger.error(f"Не удалось отправить сообщение пользователю {user_id}: {e}") if sent_votes == 0: # Если не отправили ни одного голосования, но пользователь подписан await callback.answer("Вы уже проголосовали во всех активных голосованиях!", show_alert=True) else: keyboard = types.InlineKeyboardMarkup(inline_keyboard=[[ types.InlineKeyboardButton( text="Перейти в канал", url=f"https://t.me/{CHANNEL_USERNAME[1:]}" ) ]]) # Обновляем сообщение и добавляем в ожидание await callback.message.edit_text( "Вы еще не подписаны на канал.\n\n" "Подпишитесь, пожалуйста, на канал и нажмите на кнопку ниже:", reply_markup=keyboard ) # Обновляем message_id в ожидании await add_user_to_pending(user_id, callback.message.message_id) @dp.callback_query(F.data.startswith("v_")) async def process_vote(callback: types.CallbackQuery, state: FSMContext): """Обработка голоса пользователя""" user_id = callback.from_user.id start_time = asyncio.get_event_loop().time() if not await is_db_locked(): await callback.answer("Технические неполадки. Сервер временно недоступен. Попробуйте позже.", show_alert=True) return async with vote_locks: try: if not await is_db_locked(): await callback.answer("Сервер стал недоступен во время обработки запроса. Попробуйте позже.", show_alert=True) return # Парсим callback_data чтобы получить vote_id parts = callback.data.split('_') if len(parts) < 4: await callback.answer("Ошибка обработки голоса!", show_alert=True) return vote_id = int(parts[3]) vote_data, votes_data, current_hash = await get_vote_data_by_id(vote_id) if not vote_data: await callback.answer("Голосование не найдено!", show_alert=True) return option_name, parsed_vote_id = parse_callback_data(callback.data, vote_data) if not option_name: await callback.answer("Вариант голосования не найден!", show_alert=True) return parts = callback.data.split('_') if len(parts) < 4: await callback.answer("Ошибка обработки голоса!", show_alert=True) return vote_hash = parts[2] if not await is_vote_active_by_id(vote_id) or current_hash != vote_hash: await callback.answer("Голосование завершено или не активно!", show_alert=True) await callback.message.edit_text("Голосование завершено.") await state.clear() return subscribed = await is_subscribed(user_id, CHANNEL_USERNAME) if not subscribed: await callback.answer("Для голосования необходимо быть подписчиком канала!", show_alert=True) return if await has_user_voted_in_vote(user_id, vote_id): await callback.answer("Вы уже проголосовали в этом голосовании!", show_alert=True) return success = await save_vote(option_name, user_id, vote_hash, vote_id) if success: await callback.answer(f"Вы проголосовали за: {option_name}", show_alert=True) vote_data, votes_data, current_hash = await get_vote_data_by_id(vote_id) if vote_data: keyboard = types.InlineKeyboardMarkup(inline_keyboard=[]) for i, variable in enumerate(vote_data.get('variables', []), 1): button_text = f"{i}. {variable['name']}" callback_data = create_callback_data(variable['name'], current_hash, vote_id) keyboard.inline_keyboard.append([ types.InlineKeyboardButton( text=button_text, callback_data=callback_data ) ]) try: await callback.message.edit_reply_markup(reply_markup=keyboard) except Exception: pass await state.clear() else: await callback.answer("Ошибка при сохранении голоса!", show_alert=True) except Exception as e: logger.error(f"Ошибка обработки голоса от пользователя {user_id}: {e}") await callback.answer("Произошла ошибка при обработке голоса!", show_alert=True) finally: processing_time = asyncio.get_event_loop().time() - start_time if processing_time > 2.0: logger.warning(f"Медленная обработка голоса: {processing_time:.2f}с для пользователя {user_id}") async def main(): """Основная функция запуска бота""" try: await init_database() logger.info("Бот голосований запущен") # Запускаем задачу проверки ожидающих пользователей asyncio.create_task(start_pending_checker()) await dp.start_polling(bot) except Exception as e: logger.error(f"Критическая ошибка при запуске бота: {e}") finally: await bot.session.close() if __name__ == '__main__': asyncio.run(main())