@@ -0,0 +1,321 @@
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 В
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 ( " 🛑 Бот остановлен пользователем. " )
