import asyncio import logging import time import gspread from google.oauth2.service_account import Credentials from aiogram import Bot, Dispatcher, types from aiogram.filters import Command # ============================================================================== # 1. НАСТРОЙКА ЛОГИРОВАНИЯ (Обязательно в самом верху файла!) # ============================================================================== logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s") # ============================================================================== # --- НАСТРОЙКИ --- # ВСТАВЬТЕ СЮДА НОВЫЙ ТОКЕН ПОСЛЕ /revoke В BOTFATHER! BOT_TOKEN = "8962252858:AAEkZSgyWKaRHB_wlISvhcRVsbV91F32oi0" # Вставьте ID вашей Google Таблицы (длинная строка из адресной строки между /d/ и /edit) SPREADSHEET_ID = "1JU8RuJNGgDetNpSaAoUhbUwAiuPDNOSdFG2SfaWjmTI" # ВАЖНО: Все ключи должны быть в нижнем регистре для корректного поиска! GROUP_ALIASES = { "космос": "Космос", "группа космос": "Космос", "юз": "Юго-запад", "юго запад": "Юго-запад", "юго-запад": "Юго-запад", "группа юз": "Юго-запад", "семья": "Семья", "группа семья": "Семья", "арбат": "Старый Арбат", "старый арбат": "Старый Арбат", "надежда": "Надежда", "движение надежда": "Надежда", "группа надежда": "Надежда", "сказка": "Сказка", "группа сказка": "Сказка", "мп": "Московский проспект", "московский проспект": "Московский проспект", "литком": "ЛитКом", "псо": "ПСО", "бу": "БУ", "пм": "ПМ", "рс": "РС", "группа рс": "РС", "нововоронеж": "Движение Нововоронеж", "движение нововоронеж": "Движение Нововоронеж", "калач": "Только сегодня Калач", "только сегодня калач": "Только сегодня Калач", "только сегодня": "Только сегодня Калач", "лиски": "Виктория Лиски", "виктория лиски": "Виктория Лиски", "виктория": "Виктория Лиски", "группа виктория": "Виктория Лиски", "щепка": "Щепка Курск", "щепка курск": "Щепка Курск", "группа щепка": "Щепка Курск", "кузня": "Кузня Курск", "кузня курск": "Кузня Курск", "высота": "Высота Белгород", "высота белгород": "Высота Белгород", "наркоз": "Наркоз Липецк", "липецк": "Наркоз Липецк", "волна": "Волна Липецк", "группа волна": "Волна Липецк", "тамбов": "Утренняя звезда Тамбов", "утренняя звезда": "Утренняя звезда Тамбов", "детектор": "Детектор Старый Оскол", "старый оскол": "Детектор Старый Оскол", "среда": "Среда Белгород", "группа среда": "Среда Белгород" } # ============================================================================== # 3. ПОДКЛЮЧЕНИЕ К GOOGLE ТАБЛИЦЕ # ============================================================================== SCOPES = ['https://www.googleapis.com/auth/spreadsheets'] try: creds = Credentials.from_service_account_file('credentials.json', scopes=SCOPES) gc = gspread.authorize(creds) sheet = gc.open_by_key(SPREADSHEET_ID).sheet1 logging.info("✅ Google Таблица успешно подключена!") except FileNotFoundError: logging.error("❌ Файл 'credentials.json' не найден. Положите его в папку с ботом.") exit(1) except Exception as e: logging.error(f"❌ Ошибка подключения к Google Таблице: {e}") exit(1) # ============================================================================== # 4. НАСТРОЙКА БОТА # ============================================================================== bot = Bot(token=BOT_TOKEN) dp = Dispatcher() # ============================================================================== # 5. ВСПОМОГАТЕЛЬНЫЕ ФУНКЦИИ (Синхронные, будут вызываться через asyncio.to_thread) # ============================================================================== def get_col_index(sheet_obj, target_name): """Находит номер колонки по названию группы (игнорирует переносы строк и регистр)""" headers = sheet_obj.row_values(1) target_clean = target_name.lower().replace('\n', ' ').replace(' ', ' ').strip() for i, h in enumerate(headers): h_clean = h.lower().replace('\n', ' ').replace(' ', ' ').strip() if target_clean == h_clean: return i + 1 # gspread использует индексацию с 1 return None def get_message_text(message: types.Message) -> str: """Безопасно извлекает текст: либо из сообщения, либо из подписи (caption) к фото/медиа.""" return message.text or message.caption or "" def parse_flag_event(text: str): """Анализирует текст. Первая найденная группа = забирает, вторая = отдает.""" text_lower = text.lower() if 'флаг' not in text_lower: return None, None, False is_own = 'свой' in text_lower found_groups = [] # Сортируем по длине ключа, чтобы "юго-запад" матчился раньше, чем "юз" sorted_aliases = sorted(GROUP_ALIASES.keys(), key=len, reverse=True) for alias in sorted_aliases: if alias in text_lower: canonical_name = GROUP_ALIASES[alias] if canonical_name not in found_groups: found_groups.append(canonical_name) if len(found_groups) >= 2: return found_groups[0], found_groups[1], is_own return None, None, False def update_horizontal_sheet(sheet_obj, winner: str, loser: str, is_own: bool): """Обновляет данные, минимизируя количество запросов к API (Batch Read)""" col_winner = get_col_index(sheet_obj, winner) col_loser = get_col_index(sheet_obj, loser) if not col_winner or not col_loser: raise ValueError(f"Не найдены колонки для '{winner}' или '{loser}' в таблице") # Читаем 4-ю и 5-ю строки ОДНИМ запросом для получения актуальных данных data = sheet_obj.get('4:5') row_4 = data[0] if len(data) > 0 else [] row_5 = data[1] if len(data) > 1 else [] idx_winner = col_winner - 1 idx_loser = col_loser - 1 if is_own: sheet_obj.update_cell(4, col_winner, "на месте") cell_val = row_5[idx_loser] if idx_loser < len(row_5) else "" captured_list = [x.strip() for x in str(cell_val).replace('\n', ',').split(',') if x.strip()] if winner in captured_list: captured_list.remove(winner) sheet_obj.update_cell(5, col_loser, ", ".join(captured_list)) else: sheet_obj.update_cell(4, col_loser, f"у {winner}") cell_val = row_5[idx_winner] if idx_winner < len(row_5) else "" captured_list = [x.strip() for x in str(cell_val).replace('\n', ',').split(',') if x.strip()] if loser not in captured_list: captured_list.append(loser) sheet_obj.update_cell(5, col_winner, ", ".join(captured_list)) # Сбрасываем кэш сводки после любого изменения, чтобы следующий запрос показал свежие данные global _summary_cache _summary_cache["timestamp"] = 0 # ============================================================================== # 6. КЭШИРОВАНИЕ И ГЕНЕРАЦИЯ СВОДКИ (Оптимизация против ошибки 429) # ============================================================================== _summary_cache = {"data": None, "timestamp": 0} CACHE_TTL_SECONDS = 30 # Храним кэш 30 секунд для защиты от спама командой /флаги def generate_summary(sheet_obj) -> str: """Формирует текстовую сводку, делая ВСЕГО ОДИН запрос к API Google""" try: # Читаем строки с 1 по 5 включительно ЗА ОДИН запрос! data = sheet_obj.get('1:5') except Exception as e: raise e if not data or len(data) < 5: return "📊 Таблица пуста или не заполнена." headers = data[0] # 1-я строка (заголовки) row_4 = data[3] # 4-я строка (статус своего флага) row_5 = data[4] # 5-я строка (захваченные флаги) summary_lines = [] # Начинаем с индекса 1 (вторая колонка), так как 0-я это "Группы и комитеты" for col in range(1, len(headers)): group = headers[col] if not group or not str(group).strip(): continue own_flag_status = row_4[col] if col < len(row_4) else "" captured = row_5[col] if col < len(row_5) else "" flags_text = [] if own_flag_status: clean_status = str(own_flag_status).strip() if clean_status.lower() == "на месте": flags_text.append("свой флаг на месте") else: flags_text.append(f"свой флаг {clean_status}") else: flags_text.append("статус своего флага не указан") if captured: captured_list = [x.strip() for x in str(captured).replace('\n', ',').split(',') if x.strip()] for cap in captured_list: flags_text.append(f"захвачен флаг: {cap}") summary_lines.append(f"🚩 **{group}**: {', '.join(flags_text)}") return "\n".join(summary_lines) if summary_lines else "📊 Таблица пуста или не заполнена." def get_cached_summary(sheet_obj) -> str: """Возвращает кэшированную сводку или генерирует новую, если кэш устарел""" current_time = time.time() if _summary_cache["data"] and (current_time - _summary_cache["timestamp"]) < CACHE_TTL_SECONDS: return _summary_cache["data"] new_summary = generate_summary(sheet_obj) _summary_cache["data"] = new_summary _summary_cache["timestamp"] = current_time return new_summary # ============================================================================== # 7. ФИЛЬТРЫ СООБЩЕНИЙ # ============================================================================== def is_summary_filter(message: types.Message) -> bool: """Проверяет, является ли сообщение запросом сводки""" text = get_message_text(message).lower().strip() if not text: return False clean_text = text.replace('@googleanvrn_bot', '').strip() clean_text = clean_text.lstrip('/').strip() return clean_text in ['флаги', 'summary'] def is_flag_event_filter(message: types.Message) -> bool: """Проверяет, является ли сообщение событием захвата флага""" text = get_message_text(message) if not text: return False text_lower = text.lower().strip() # ВАЖНО: Сначала исключаем запросы сводки, чтобы слово "флаги" не триггерило этот фильтр if is_summary_filter(message): return False return bool('@googleanvrn_bot' in text_lower and 'флаг' in text_lower) # ============================================================================== # 8. ОБРАБОТЧИКИ СООБЩЕНИЙ (Асинхронные) # ============================================================================== @dp.message(Command("start")) async def cmd_start(message: types.Message): await message.answer( "🤖 Бот для учета флагов запущен!\n\n" "Вы можете отправить обычное сообщение или фото с подписью:\n" "@googleAnVrn_bot Космос забрал свой флаг у ЮЗ\n" "@googleAnVrn_bot Литком забрали флаг у РС\n\n" "Для получения сводки используйте: /флаги или @googleAnVrn_bot флаги" ) @dp.message(is_summary_filter) async def cmd_summary(message: types.Message): await message.answer("⏳ Собираю данные из таблицы...") try: # Используем кэшированную версию функции в отдельном потоке summary = await asyncio.to_thread(get_cached_summary, sheet) await message.answer(summary, parse_mode="Markdown") except gspread.exceptions.APIError as api_err: # Специальная обработка ошибки 429 (Quota Exceeded) if "429" in str(api_err): logging.warning("⚠️ Превышен лимит запросов к Google Sheets (429).") await message.answer( "⚠️ Слишком много запросов к таблице прямо сейчас. " "Google временно ограничил чтение. Пожалуйста, подождите 1 минуту и попробуйте снова." ) else: logging.error(f"Ошибка Google API: {api_err}", exc_info=True) await message.answer("❌ Произошла ошибка при обращении к Google Таблице.") except Exception as e: logging.error(f"Критическая ошибка генерации сводки: {e}", exc_info=True) await message.answer("❌ Не удалось прочитать таблицу.") @dp.message(is_flag_event_filter) async def handle_flag_event(message: types.Message): text_to_parse = get_message_text(message) winner, loser, is_own = parse_flag_event(text_to_parse) if not winner or not loser: await message.reply( "⚠️ Не удалось распознать две группы в сообщении или подписи к фото. " "Используйте понятные названия (например: Космос, ЮЗ, ЛитКом, РС, Надежда, Сказка, Семья, МП, Щепка, Кузня, Высота, Только сегодня, Виктория)." ) return try: # Оборачиваем синхронную работу с gspread в asyncio.to_thread await asyncio.to_thread(update_horizontal_sheet, sheet, winner, loser, is_own) if is_own: text_response = f"✅ Зафиксировано: **{winner}** вернул(а) свой флаг (теперь он на месте)." else: text_response = f"✅ Зафиксировано: **{winner}** забрал(а) флаг у **{loser}**." await message.reply(text_response, parse_mode="Markdown") except ValueError as ve: logging.warning(f"Ошибка валидации данных: {ve}") await message.reply(f"❌ Ошибка: {ve}. Проверьте заголовки в таблице.") except Exception as e: logging.error(f"Критическая ошибка записи в таблицу: {e}", exc_info=True) await message.reply("❌ Произошла непредвиденная ошибка при записи в таблицу.") # ============================================================================== # 9. ЗАПУСК # ============================================================================== async def main(): logging.info("🚀 Бот запущен и начинает слушать сообщения...") await dp.start_polling(bot) if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: logging.info("🛑 Бот остановлен пользователем.")