From 9de1a40f1fb7d44039912a5fc1e0ce80f6b7216c Mon Sep 17 00:00:00 2001 From: imletbruh <60918217+qiyanaitsme@users.noreply.github.com> Date: Sun, 31 May 2026 18:24:39 +0500 Subject: [PATCH] Add files via upload --- README.md | 302 ++++----- api_client.py | 47 +- app.py | 1483 +++++++++++++++++++++++++-------------------- config_manager.py | 228 ++++--- database.py | 554 +++++++++-------- requirements.txt | 5 +- 6 files changed, 1392 insertions(+), 1227 deletions(-) diff --git a/README.md b/README.md index bb213a5..28d72bf 100644 --- a/README.md +++ b/README.md @@ -1,27 +1,41 @@ # 🚀 QIYANA AUTO-BUMP BOT для Lolz.live -Production-ready Telegram бот для автоматического поднятия тем на форуме Lolz.live с современной архитектурой, batch API и полной типизацией. +Telegram бот для автоматического поднятия тем на форуме Lolz.live. Batch API, динамические настройки (меняются на лету без перезапуска), персистентная статистика. ## ✨ Возможности -- ➕ **Добавление тем** - через запятую (12345, 67890, 11111) с batch API -- 🗑️ **Удаление тем** - через интерактивное меню -- 📋 **Список тем** - с датой последнего поднятия -- 🚀 **Ручное поднятие** - поднять все темы немедленно через batch API -- ⏰ **Автоподнятие** - каждые N часов автоматически -- 📊 **Статистика** - успешность, количество поднятий, uptime -- 🔔 **Уведомления** - о каждом поднятии темы -- 🛡️ **Обработка ошибок** - rate limits, network failures, invalid tokens -- ⚡ **Batch API** - до 10 тем за один запрос (10x быстрее) +- ➕ **Добавление тем** — через запятую, названия загружаются сразу через batch API +- 🗑️ **Удаление тем** — через интерактивное меню +- 📋 **Список тем** — с названиями из БД и датой последнего поднятия +- 🚀 **Ручное поднятие** — все темы немедленно через batch API +- 🔄 **Обновление названий** — синхронизация названий тем с форумом по кнопке +- ⏰ **Автоподнятие** — каждые N часов автоматически +- 🛠️ **Динамические настройки** — интервал, batch size, автобамп меняются через меню без перезапуска +- 📊 **Статистика** — сохраняется в БД, не теряется при перезапуске +- 🔔 **Уведомления** — о результате каждого поднятия +- 🛡️ **Защита доступа** — ботом управляет только админ (по Telegram ID) +- ⚡ **Batch API** — до 10 тем за один запрос ## 🎮 Интерфейс ``` -┌─────────────────────────────────┐ -│ ➕ Добавить темы │ 📋 Список тем │ -│ 🗑️ Удалить тему │ 🚀 Поднять темы │ -│ 📊 Статистика │ 👤 Автор │ -└─────────────────────────────────┘ +┌──────────────────────────────────────┐ +│ ➕ Add topics │ 📋 List of topics │ +│ 🗑️ Delete topic │ 🚀 Bump topics │ +│ 🔄 Refresh │ 📊 Statistics │ +│ 🛠️ Settings │ 👤 Author │ +└──────────────────────────────────────┘ +``` + +### Меню настроек + +``` +┌──────────────────────────────────┐ +│ ⏰ Set Interval │ +│ 📦 Set Batch Size │ +│ 🔄 Toggle Auto-Bump │ +│ ↩️ Back to Menu │ +└──────────────────────────────────┘ ``` ## 📦 Установка @@ -29,9 +43,9 @@ Production-ready Telegram бот для автоматического подн ### Требования - Python 3.12+ -- pip или uv +- pip -### 1. Установите зависимости +### 1. Клонируйте и установите зависимости ```bash pip install -r requirements.txt @@ -49,29 +63,32 @@ pip install -r requirements.txt 2. Создайте токен с правами `read`, `post` 3. Скопируйте токен -### 3. Настройте config.json +**Ваш Telegram ID:** +1. Напишите [@userinfobot](https://t.me/userinfobot) и получите свой ID -```json -{ - "bot": { - "api_token": "1234567890:ABCdefGHIjklMNOpqrsTUVwxyz", - "img_url": "https://wallpapers-clan.com/wp-content/uploads/2024/04/dark-anime-girl-with-red-eyes-desktop-wallpaper-preview.jpg", - "author_url": "https://lolz.live/kqlol/" - }, - "api": { - "base_url": "https://prod-api.lolz.live", - "auth_token": "eyJ0eXAiOiJKV1QiLCJhbGc...", - "batch_size": 10 - }, - "database": { - "path": "threads.db" - }, - "scheduling": { - "bump_interval_hours": 12, - "bump_delay_seconds": 2, - "enable_auto_bump": true - } -} +### 3. Настройте `.env` + +```env +# Telegram Bot +BOT_API_TOKEN=1234567890:ABCdefGHIjklMNOpqrsTUVwxyz +BOT_IMG_URL=https://wallpapers-clan.com/wp-content/uploads/2024/04/dark-anime-girl-with-red-eyes-desktop-wallpaper-preview.jpg +BOT_AUTHOR_URL=https://lolz.live/kqlol/ + +# Lolz API +API_BASE_URL=https://prod-api.lolz.live +API_AUTH_TOKEN=eyJ0eXAiOiJKV1QiLCJhbGc... +API_BATCH_SIZE=10 + +# Database +DB_PATH=threads.db + +# Scheduling +BUMP_INTERVAL_HOURS=12 +BUMP_DELAY_SECONDS=2 +ENABLE_AUTO_BUMP=true + +# Admin (ваш Telegram user ID) +ADMIN_USER_ID=123456789 ``` ### 4. Запустите бота @@ -84,210 +101,139 @@ python app.py ### Добавление тем -1. Нажмите **➕ Добавить темы** +1. Нажмите **➕ Add topics** 2. Введите ID через запятую: `12345, 67890, 11111` -3. Бот автоматически получит названия тем через batch API (до 10 за запрос) +3. Бот сразу загрузит реальные названия через batch API и сохранит в БД -**Пример:** -``` -Ввод: 9247920, 9247921, 9247922 +### Обновление названий -Результат: -✅ 9247920 - Qiyanas steam idler -✅ 9247921 - Discord bot -✅ 9247922 - VPN service -``` +1. Нажмите **🔄 Refresh** +2. Бот загрузит актуальные названия всех тем через batch API +3. Обновлённые названия сохранятся в БД -**Batch API преимущества:** -- Добавление 10 тем = 1 batch запрос вместо 10 обычных -- Добавление 25 тем = 3 batch запроса вместо 25 обычных -- Экономия API лимитов в 10 раз +Полезно, если темы переименовали на форуме — не нужно удалять и добавлять заново. ### Удаление темы -1. Нажмите **🗑️ Удалить тему** +1. Нажмите **🗑️ Delete topic** 2. Выберите тему из списка -3. Подтвердите удаление ### Ручное поднятие -1. Нажмите **🚀 Поднять темы** -2. Бот поднимет все темы через batch API (до 10 за запрос) +1. Нажмите **🚀 Bump topics** +2. Бот поднимет все темы через batch API 3. Получите уведомление о каждой теме: ``` [1/3] ✅ Тема 9247920 поднята успешно [2/3] ✅ Тема 9247921 поднята успешно - [3/3] ✅ Тема 9247922 поднята успешно + [3/3] ❌ Тема 9247922: нужно подождать ``` -**Batch API преимущества:** -- Поднятие 10 тем = 1 batch запрос (0.1 * 10 = 1 batch) -- Поднятие 50 тем = 5 batch запросов вместо 50 обычных -- Скорость выполнения увеличена в ~10 раз - -### Просмотр списка - -Нажмите **📋 Список тем** - покажет: -- ID темы -- Название темы -- Дату последнего поднятия - ### Статистика -Нажмите **📊 Статистика** - покажет: -- Количество тем -- Готовые к поднятию -- Всего попыток -- Успешность (%) -- Настройки интервала +Нажмите **📊 Statistics** — покажет: +- Количество тем (всего / готовы к поднятию) +- Статистика поднятий (всего / успешно / %) +- Дата последнего бампа +- Текущие настройки - Время работы бота -## ⚙️ Настройки +Статистика сохраняется в SQLite и не сбрасывается при перезапуске. -### Интервал автоподнятия +## ⚙️ Настройки (меняются на лету) -В `config.json` измените `bump_interval_hours`: +Все настройки изменяются через кнопку **🛠️ Settings** без остановки бота: -| Значение | Интервал | -|----------|----------| -| `12` | 12 часов | -| `6` | 6 часов | -| `24` | 24 часа | +| Кнопка | Что меняет | Применяется | +|--------|-----------|-------------| +| ⏰ Set Interval | Интервал автобампа (часы) | Мгновенно, перезапускает цикл | +| 📦 Set Batch Size | Размер batch (1–10) | Мгновенно для следующих запросов | +| 🔄 Toggle Auto-Bump | Включить/выключить автобамп | Мгновенно | -### Размер batch запросов - -В `config.json` измените `batch_size`: - -| Значение | Описание | -|----------|----------| -| `10` | Максимум (рекомендуется) - 10 тем за запрос | -| `5` | Средний - 5 тем за запрос | -| `1` | Отключить batch - по 1 теме | - -**Примечание:** Batch API Lolz.live поддерживает до 10 запросов в одном batch. Каждый запрос = 0.1 batch, итого 10 запросов = 1 полный batch. - -### Тестовый режим (5 минут) - -```json -"scheduling": { - "bump_interval_hours": 0.0833, - "bump_delay_seconds": 1, - "enable_auto_bump": true -} -``` +Настройки хранятся в БД и сохраняются между перезапусками. ## 🔄 Как работает автоподнятие 1. Запускаете бота 2. Вручную поднимаете темы через 🚀 (первый раз) -3. Бот ждет указанный интервал (например, 12 часов) -4. Автоматически поднимает темы, у которых прошло 12+ часов -5. Повторяет каждые 12 часов +3. Бот ждёт указанный интервал (например, 12 часов) +4. Автоматически поднимает темы, у которых прошёл интервал +5. Цикл повторяется -**Пример:** ``` -00:00 - Запуск бота -00:05 - Вы вручную нажали "Поднять темы" -12:05 - Автоматическое поднятие -24:05 - Автоматическое поднятие -36:05 - Автоматическое поднятие +00:00 — Запуск бота +00:05 — Ручное поднятие +12:05 — Автоматическое поднятие +24:05 — Автоматическое поднятие ``` ## 🏗️ Архитектура -### Модульная структура - ``` -app.py # Main bot logic with FSM -config_manager.py # Type-safe configuration -api_client.py # Lolz batch API client with retry logic -database.py # SQLite with connection pooling +app.py # Бот: хендлеры, middleware аутентификации, авто-бамп +config_manager.py # Загрузка и валидация .env +api_client.py # Lolz batch API клиент с retry логикой +database.py # SQLite: темы, настройки, статистика ``` -### Ключевые улучшения +### Что внутри БД (SQLite через aiosqlite) -- **Batch API**: До 10 тем за один запрос (10x эффективнее) -- **Type Safety**: Полная типизация с Python 3.12+ (PEP 695) -- **SOLID Principles**: Каждый модуль имеет одну ответственность -- **Connection Pooling**: Эффективное управление соединениями -- **Error Handling**: Специфичные исключения для каждого случая -- **Async Context Managers**: Автоматическая очистка ресурсов -- **Immutable Config**: Frozen dataclasses для безопасности -- **Logging**: Структурированные логи с rotation -- **Graceful Shutdown**: Корректное завершение всех задач +| Таблица | Назначение | +|---------|-----------| +| `threads` | Темы: ID, название, дата последнего бампа | +| `settings` | Динамические настройки (интервал, batch_size, автобамп) | +| `bump_stats` | Персистентная статистика (всего/успешно/последний бамп) | -### Batch API Implementation +### Ключевые особенности -**Как работает:** -```python -# Старый способ (10 запросов): -for thread_id in [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]: - POST /threads/{thread_id}/bump - -# Новый способ (1 batch запрос): -POST /batch -[ - {"uri": "/threads/1/bump", "method": "POST"}, - {"uri": "/threads/2/bump", "method": "POST"}, - ... - {"uri": "/threads/10/bump", "method": "POST"} -] -``` - -**Преимущества:** -- Экономия API лимитов в 10 раз -- Скорость выполнения увеличена в ~10 раз -- Меньше нагрузка на сеть -- Атомарная обработка результатов - -### Edge Cases - -- Rate limiting от API (обработка в batch ответах) -- Network timeouts (retry для batch запросов) -- Invalid tokens (валидация перед batch) -- Partial batch failures (индивидуальная обработка каждого результата) -- Concurrent access (connection pooling) -- Database locks (async context managers) -- Memory leaks prevention (frozen dataclasses, proper cleanup) +- **Batch API** — до 10 тем за запрос (экономия лимитов в 10 раз) +- **Динамические настройки** — меняются в БД, подхватываются ботом мгновенно +- **Персистентная статистика** — не теряется при перезапуске +- **Аутентификация** — middleware проверяет Telegram ID админа +- **Async context managers** — автоматическая очистка ресурсов +- **Frozen dataclasses с `__slots__`** — иммутабельность и экономия памяти +- **WAL-режим SQLite** — лучшая конкурентность ## 📝 Логи -- **Консоль** - основные события -- **bot.log** - детальные логи с traceback +- **Консоль** — основные события +- **bot.log** — детальные логи с traceback -## 🔧 Технические детали +## 🔧 Зависимости - **Python**: 3.12+ -- **aiogram**: 3.4.1 -- **aiohttp**: 3.10.0+ +- **aiogram**: 3.15.0 +- **aiohttp**: bundled with aiogram - **aiosqlite**: 0.20.0+ -- **База данных**: SQLite с индексами -- **API**: https://prod-api.lolz.live -- **Batch API**: До 10 запросов за 1 batch (каждый = 0.1 batch) +- **python-dotenv**: 1.0.0+ ## 🐛 Troubleshooting -### Ошибка: "bot.api_token not configured" +### Ошибка: «BOT_API_TOKEN not configured» -Заполните `config.json` реальными токенами. +Проверьте `.env` — заполните `BOT_API_TOKEN` реальным токеном от @BotFather. -### Ошибка: "Invalid API token" +### Ошибка: «ADMIN_USER_ID not configured» + +Укажите в `.env` ваш Telegram ID (получить через @userinfobot). + +### Ошибка: «API_AUTH_TOKEN not configured» Проверьте токен на [zelenka.guru/account/api](https://zelenka.guru/account/api). -### Ошибка: "Network error" +### Ошибка: «⛔ Доступ запрещён» -Проверьте интернет-соединение и доступность API. +Ботом может управлять только пользователь, чей Telegram ID указан в `ADMIN_USER_ID`. ### Темы не поднимаются автоматически -1. Проверьте `enable_auto_bump: true` в config.json +1. Проверьте, что автобамп включён (🔄 Toggle Auto-Bump) 2. Сделайте первое поднятие вручную через 🚀 -3. Проверьте логи в bot.log +3. Проверьте логи в `bot.log` ## 👤 Автор -[QIYANA](https://lolz.live/kqlol/) - создатель бота +[QIYANA](https://lolz.live/kqlol/) --- diff --git a/api_client.py b/api_client.py index e2609e9..9dd79bc 100644 --- a/api_client.py +++ b/api_client.py @@ -352,10 +352,10 @@ class APIClient: continue job_data = jobs[uri] - logger.debug(f"Thread {thread_id} bump response: {job_data}") + logger.info(f"Thread {thread_id} raw bump response: {job_data}") - # Empty dict means success for bump - if isinstance(job_data, dict) and len(job_data) == 0: + # Empty list [] or empty dict {} means success + if isinstance(job_data, (list, dict)) and len(job_data) == 0: logger.info(f"Thread {thread_id} bumped successfully (empty response)") results.append(BumpResult( success=True, @@ -364,7 +364,6 @@ class APIClient: status=BumpStatus.SUCCESS )) elif isinstance(job_data, dict) and "errors" in job_data: - # Has errors error_msg = self._extract_error_message(str(job_data["errors"])) logger.error( f"Thread {thread_id} bump failed | " @@ -377,12 +376,46 @@ class APIClient: thread_id=thread_id, status=BumpStatus.ERROR )) + elif isinstance(job_data, dict) and "_job_result" in job_data: + job_result = str(job_data.get("_job_result", "")) + job_message = str(job_data.get("_job_message", "")) + if job_result == "error": + error_text = job_message + if not error_text.strip(): + errors = job_data.get("errors") + if errors: + if isinstance(errors, list): + error_text = str(errors[0]) if errors else "" + elif isinstance(errors, str): + error_text = errors + else: + error_text = str(errors) + if not error_text.strip(): + error_text = str(job_data.get("error", "")) + error_msg = self._extract_error_message(error_text) or "Ошибка API (см. логи)" + logger.error( + f"Thread {thread_id} bump failed | " + f"Job error: {error_msg}" + ) + results.append(BumpResult( + success=False, + message=f"Тема {thread_id}: {error_msg}", + thread_id=thread_id, + status=BumpStatus.ERROR + )) + else: + logger.info(f"Thread {thread_id} bumped successfully (job_result={job_result}, job_message={job_message})") + results.append(BumpResult( + success=True, + message=f"✅ Тема {thread_id} поднята успешно", + thread_id=thread_id, + status=BumpStatus.SUCCESS + )) else: - # Unknown response - logger.warning(f"Thread {thread_id} unknown bump response: {job_data}") + logger.warning(f"Thread {thread_id} unknown bump response (type={type(job_data).__name__}): {job_data}") results.append(BumpResult( success=False, - message=f"Тема {thread_id}: Неизвестный ответ", + message=f"Тема {thread_id}: Неизвестный ответ ({type(job_data).__name__})", thread_id=thread_id, status=BumpStatus.ERROR )) diff --git a/app.py b/app.py index bcc5b7a..c8ce07e 100644 --- a/app.py +++ b/app.py @@ -1,661 +1,822 @@ -"""Telegram bot for automatic thread bumping on Lolz.live.""" - -import asyncio -import logging -import sys -from datetime import datetime -from typing import Sequence - -from aiogram import Bot, Dispatcher, F, Router -from aiogram.client.default import DefaultBotProperties -from aiogram.enums import ParseMode -from aiogram.filters import Command, StateFilter -from aiogram.types import Message, CallbackQuery, InlineKeyboardMarkup, InlineKeyboardButton -from aiogram.fsm.context import FSMContext -from aiogram.fsm.state import State, StatesGroup -from aiogram.fsm.storage.memory import MemoryStorage -from aiogram.exceptions import TelegramRetryAfter, TelegramBadRequest - -from config_manager import Config -from api_client import APIClient, BumpResult -from database import Database, Thread - - -# Constants -NOTIFICATION_DELAY_SECONDS = 0.8 -BUTTON_TEXT_MAX_LENGTH = 30 -AUTO_BUMP_RETRY_DELAY_SECONDS = 60 - - -# Configure logging -logging.basicConfig( - level=logging.INFO, - format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", - handlers=[ - logging.StreamHandler(sys.stdout), - logging.FileHandler("bot.log", encoding="utf-8") - ] -) -logger = logging.getLogger(__name__) - - -class BotStates(StatesGroup): - """FSM states for bot.""" - waiting_for_thread_ids = State() - - -class AutoBumpBot: - """Main bot class with clean separation of concerns.""" - - __slots__ = ( - "_config", "_bot", "_dp", "_router", "_api", "_db", - "_is_running", "_total_bumps", "_successful_bumps", "_start_time" - ) - - def __init__(self, config: Config) -> None: - self._config = config - - # Initialize bot with default properties (aiogram 3.15.0 best practice) - self._bot = Bot( - token=config.bot.api_token, - default=DefaultBotProperties(parse_mode=ParseMode.HTML) - ) - - self._dp = Dispatcher(storage=MemoryStorage()) - self._router = Router(name="main_router") - - self._api = APIClient( - config.api.base_url, - config.api.auth_token, - config.api.batch_size - ) - self._db = Database(config.database.path) - - # Statistics - self._is_running = False - self._total_bumps = 0 - self._successful_bumps = 0 - self._start_time: datetime | None = None - - # Register handlers - self._register_handlers() - - # Include router in dispatcher - self._dp.include_router(self._router) - - def _register_handlers(self) -> None: - """Register all bot message and callback handlers.""" - # Message handlers - self._router.message.register( - self._handle_start_command, - Command("start") - ) - self._router.message.register( - self._handle_thread_ids_input, - StateFilter(BotStates.waiting_for_thread_ids) - ) - - # Callback query handlers - self._router.callback_query.register( - self._handle_add_thread_callback, - F.data == "add_thread" - ) - self._router.callback_query.register( - self._handle_list_threads_callback, - F.data == "list_threads" - ) - self._router.callback_query.register( - self._handle_delete_menu_callback, - F.data == "delete_menu" - ) - self._router.callback_query.register( - self._handle_delete_thread_callback, - F.data.startswith("delete_") - ) - self._router.callback_query.register( - self._handle_bump_now_callback, - F.data == "bump_now" - ) - self._router.callback_query.register( - self._handle_stats_callback, - F.data == "stats" - ) - - def _create_main_menu_keyboard(self) -> InlineKeyboardMarkup: - """Create main menu inline keyboard.""" - return InlineKeyboardMarkup(inline_keyboard=[ - [ - InlineKeyboardButton(text="➕ Add topics", callback_data="add_thread"), - InlineKeyboardButton(text="📋 List of topics", callback_data="list_threads") - ], - [ - InlineKeyboardButton(text="🗑️ Delete topic", callback_data="delete_menu"), - InlineKeyboardButton(text="🚀 Bump topics", callback_data="bump_now") - ], - [ - InlineKeyboardButton(text="📊 Statistics", callback_data="stats"), - InlineKeyboardButton(text="👤 Author", url=self._config.bot.author_url) - ] - ]) - - async def _send_main_menu(self, chat_id: int, text: str = "Choose an action:") -> None: - """Send main menu to user with photo for consistent width.""" - await self._bot.send_photo( - chat_id, - photo=self._config.bot.img_url, - caption=text, - reply_markup=self._create_main_menu_keyboard() - ) - - @staticmethod - def _truncate_text_safely(text: str, max_length: int) -> str: - """Safely truncate text without breaking unicode characters.""" - if len(text) <= max_length: - return text - - truncated = text[:max_length] - try: - truncated.encode('utf-8') - return truncated - except UnicodeEncodeError: - return AutoBumpBot._truncate_text_safely(text[:max_length - 1], max_length - 1) - - @staticmethod - def _calculate_uptime(start_time: datetime) -> str: - """Calculate uptime from start time.""" - delta = datetime.now() - start_time - total_seconds = int(delta.total_seconds()) - - days = total_seconds // 86400 - hours = (total_seconds % 86400) // 3600 - minutes = (total_seconds % 3600) // 60 - - if days > 0: - return f"{days}д {hours}ч {minutes}м" - elif hours > 0: - return f"{hours}ч {minutes}м" - else: - return f"{minutes}м" - - async def _handle_start_command(self, message: Message) -> None: - """Handle /start command.""" - try: - await message.answer_photo( - photo=self._config.bot.img_url, - caption=( - "🤖 QIYANA AUTO-BUMP BOT\n\n" - "Бот для автоматического поднятия тем на Lolz.live\n\n" - f"⏰ Автоподнятие каждые {self._config.scheduling.bump_interval_hours}ч" - ), - reply_markup=self._create_main_menu_keyboard() - ) - except TelegramBadRequest as e: - logger.error(f"Failed to send start message: {e}") - await message.answer("❌ Ошибка отправки сообщения. Попробуйте /start снова.") - - async def _handle_add_thread_callback(self, callback: CallbackQuery, state: FSMContext) -> None: - """Handle add thread button callback.""" - await callback.answer() - - if not callback.message: - return - - await callback.message.answer( - "📝 Введите ID тем через запятую для добавления:\n" - "Пример: 12345, 67890, 11111" - ) - await state.set_state(BotStates.waiting_for_thread_ids) - - async def _handle_thread_ids_input(self, message: Message, state: FSMContext) -> None: - """Handle thread IDs input from user.""" - try: - thread_ids = [tid.strip() for tid in message.text.split(",") if tid.strip()] - - if not thread_ids: - await message.answer("❌ Не указаны ID тем") - return - - # Validate format - valid_ids: list[str] = [] - errors: list[str] = [] - - for thread_id in thread_ids: - if not thread_id.isdigit(): - errors.append(f"❌ {thread_id} - неверный формат") - else: - valid_ids.append(thread_id) - - if not valid_ids: - await message.answer("❌ Нет валидных ID тем") - return - - status_msg = await message.answer( - f"⏳ Добавляю {len(valid_ids)} тем в базу данных..." - ) - - # Add threads to database directly (without fetching titles) - added: list[str] = [] - - for thread_id in valid_ids: - # Use thread ID as title initially (will be fetched on first bump/list) - success = await self._db.add_thread(thread_id, f"Thread {thread_id}") - - if success: - added.append(f"✅ {thread_id}") - else: - errors.append(f"⚠️ {thread_id} - уже добавлена") - - # Build result message - result_parts = ["📊 Результат добавления:\n"] - - if added: - result_parts.append("\n✅ Добавлено:\n") - result_parts.append("\n".join(added)) - result_parts.append("\n") - - if errors: - result_parts.append("\n❌ Ошибки:\n") - result_parts.append("\n".join(errors)) - - await status_msg.edit_text("".join(result_parts)) - await self._send_main_menu(message.chat.id) - - except Exception as e: - logger.error(f"Error in thread IDs input handler: {e}", exc_info=True) - await message.answer(f"❌ Критическая ошибка: {str(e)}") - finally: - await state.clear() - - async def _handle_list_threads_callback(self, callback: CallbackQuery) -> None: - """Handle list threads button callback.""" - await callback.answer() - - if not callback.message: - return - - try: - threads = await self._db.get_all_threads() - - if not threads: - await callback.message.answer("📭 Список тем пуст\n\nДобавьте темы через кнопку ➕") - await self._send_main_menu(callback.message.chat.id) - return - - # Fetch real titles from API using batch - status_msg = await callback.message.answer( - f"⏳ Загружаю информацию о {len(threads)} темах..." - ) - - thread_ids = [t.id for t in threads] - try: - threads_info = await self._api.get_threads_info_batch(thread_ids) - except Exception as e: - logger.error(f"Error fetching thread titles: {e}", exc_info=True) - threads_info = [None] * len(thread_ids) - - # Build formatted list with thread info - text_parts = [f"📋 Список тем ({len(threads)}):\n\n"] - - for thread, thread_info in zip(threads, threads_info): - # Use fetched title or fallback to database title - title = thread_info.title if thread_info else thread.title - - # Format last bump status - status = "🆕 Новая" - if thread.last_bumped: - try: - dt = datetime.fromisoformat(thread.last_bumped.replace("Z", "+00:00")) - status = f"✅ {dt.strftime('%d.%m.%Y %H:%M')}" - except ValueError: - status = f"✅ {thread.last_bumped[:16]}" - - # Format thread entry - text_parts.append( - f"ID: {thread.id}\n" - f"Название: {title}\n" - f"Статус: {status}\n\n" - ) - - await status_msg.edit_text("".join(text_parts)) - await self._send_main_menu(callback.message.chat.id) - - except Exception as e: - logger.error(f"Error listing threads: {e}", exc_info=True) - await callback.message.answer(f"❌ Ошибка: {str(e)}") - - async def _handle_delete_menu_callback(self, callback: CallbackQuery) -> None: - """Handle delete menu button callback.""" - await callback.answer() - - if not callback.message: - return - - try: - threads = await self._db.get_all_threads() - - if not threads: - await callback.message.answer("📭 Список тем пуст") - await self._send_main_menu(callback.message.chat.id) - return - - buttons = [ - [InlineKeyboardButton( - text=f"{thread.id} - {self._truncate_text_safely(thread.title, BUTTON_TEXT_MAX_LENGTH)}", - callback_data=f"delete_{thread.id}" - )] - for thread in threads - ] - - keyboard = InlineKeyboardMarkup(inline_keyboard=buttons) - - await callback.message.answer( - "🗑️ Выберите тему для удаления:", - reply_markup=keyboard - ) - - except Exception as e: - logger.error(f"Error showing delete menu: {e}", exc_info=True) - await callback.message.answer(f"❌ Ошибка: {str(e)}") - - async def _handle_delete_thread_callback(self, callback: CallbackQuery) -> None: - """Handle delete specific thread callback.""" - if not callback.message: - await callback.answer("❌ Ошибка: сообщение не найдено", show_alert=True) - return - - try: - thread_id = callback.data.split("_", 1)[1] - success = await self._db.delete_thread(thread_id) - - if success: - await callback.answer(f"✅ Тема {thread_id} удалена", show_alert=True) - await callback.message.answer(f"✅ Тема {thread_id} успешно удалена из базы данных") - await self._send_main_menu(callback.message.chat.id) - else: - await callback.answer(f"❌ Тема {thread_id} не найдена", show_alert=True) - - except Exception as e: - logger.error(f"Error deleting thread: {e}", exc_info=True) - await callback.answer(f"❌ Ошибка: {str(e)}", show_alert=True) - - async def _handle_bump_now_callback(self, callback: CallbackQuery) -> None: - """Handle bump now button callback.""" - await callback.answer() - - if not callback.message: - return - - try: - threads = await self._db.get_threads_to_bump(0) - - if not threads: - await callback.message.answer("📭 Нет тем для поднятия\n\nДобавьте темы через кнопку ➕") - await self._send_main_menu(callback.message.chat.id) - return - - status_msg = await callback.message.answer( - f"🚀 Начинаю поднятие {len(threads)} тем...\n\n" - "Используется batch API (до 10 тем за запрос)\n" - "Подождите, это может занять некоторое время..." - ) - - results = await self._execute_bump_with_notifications( - threads, - chat_id=callback.message.chat.id - ) - - success_count = sum(1 for r in results if r.success) - - await status_msg.edit_text( - f"📊 Поднятие завершено!\n\n" - f"✅ Успешно: {success_count}\n" - f"❌ Ошибок: {len(threads) - success_count}\n" - f"📝 Всего тем: {len(threads)}" - ) - - await self._send_main_menu(callback.message.chat.id) - - except Exception as e: - logger.error(f"Error bumping threads: {e}", exc_info=True) - await callback.message.answer(f"❌ Ошибка: {str(e)}") - - async def _handle_stats_callback(self, callback: CallbackQuery) -> None: - """Handle statistics button callback.""" - await callback.answer() - - if not callback.message: - return - - try: - threads = await self._db.get_all_threads() - threads_ready = await self._db.get_threads_to_bump( - self._config.scheduling.bump_interval_hours - ) - - uptime = "0 минут" - if self._start_time: - uptime = self._calculate_uptime(self._start_time) - - success_rate = 0.0 - if self._total_bumps > 0: - success_rate = (self._successful_bumps / self._total_bumps) * 100 - - text = ( - "📊 Статистика бота\n\n" - f"📋 Темы:\n" - f"• Всего в списке: {len(threads)}\n" - f"• Готовы к поднятию: {len(threads_ready)}\n" - f"• Ожидают времени: {len(threads) - len(threads_ready)}\n\n" - f"🚀 Поднятия:\n" - f"• Всего попыток: {self._total_bumps}\n" - f"• Успешных: {self._successful_bumps}\n" - f"• Успешность: {success_rate:.1f}%\n\n" - f"⚙️ Настройки:\n" - f"• Интервал: {self._config.scheduling.bump_interval_hours}ч\n" - f"• Batch size: {self._config.api.batch_size}\n\n" - f"⏱️ Работа:\n" - f"• Время работы: {uptime}" - ) - - await callback.message.answer(text) - await self._send_main_menu(callback.message.chat.id) - - except Exception as e: - logger.error(f"Error showing stats: {e}", exc_info=True) - await callback.message.answer(f"❌ Ошибка: {str(e)}") - - async def _execute_bump_with_notifications( - self, - threads: Sequence[Thread], - chat_id: int | None = None - ) -> list[BumpResult]: - """Execute thread bumping with optional notifications.""" - if not threads: - return [] - - thread_ids = [thread.id for thread in threads] - - logger.info(f"Starting bump for {len(thread_ids)} threads: {thread_ids}") - - # Use batch API to bump all threads - results = await self._api.bump_threads_batch(thread_ids) - - # Log detailed results - for result in results: - if result.success: - logger.info( - f"✅ BUMP SUCCESS | Thread: {result.thread_id} | " - f"Status: {result.status.value} | Message: {result.message}" - ) - else: - logger.error( - f"❌ BUMP FAILED | Thread: {result.thread_id} | " - f"Status: {result.status.value} | Message: {result.message}" - ) - - # Update statistics and database - for result in results: - self._total_bumps += 1 - if result.success: - self._successful_bumps += 1 - await self._db.update_last_bumped(result.thread_id) - logger.info(f"Database updated for thread {result.thread_id}") - - # Send notifications if requested - if chat_id: - await self._send_bump_notifications(chat_id, results) - - # Log summary - success_count = sum(1 for r in results if r.success) - logger.info( - f"Batch bump completed: {success_count}/{len(results)} successful | " - f"Failed: {len(results) - success_count}" - ) - - return results - - async def _send_bump_notifications(self, chat_id: int, results: list[BumpResult]) -> None: - """Send bump result notifications to user with rate limit handling.""" - for i, result in enumerate(results, 1): - try: - await self._bot.send_message( - chat_id, - f"[{i}/{len(results)}] {result.message}" - ) - - # Delay between notifications to avoid rate limits - if i < len(results): - await asyncio.sleep(NOTIFICATION_DELAY_SECONDS) - - except TelegramRetryAfter as e: - # Handle rate limit - wait and retry - logger.warning(f"Rate limit hit, waiting {e.retry_after} seconds") - await asyncio.sleep(e.retry_after) - - # Retry sending this message - try: - await self._bot.send_message( - chat_id, - f"[{i}/{len(results)}] {result.message}" - ) - except Exception as retry_error: - logger.error(f"Failed to send notification after retry: {retry_error}") - - except Exception as e: - logger.error(f"Failed to send notification: {e}") - - async def _auto_bump_scheduler_loop(self) -> None: - """Automatic bump scheduler loop.""" - logger.info( - f"Auto-bump scheduler started (interval: {self._config.scheduling.bump_interval_hours}h)" - ) - - while self._is_running: - try: - sleep_seconds = self._config.scheduling.bump_interval_hours * 3600 - logger.info(f"Next scheduled bump in {self._config.scheduling.bump_interval_hours} hours") - - await asyncio.sleep(sleep_seconds) - - if not self._is_running: - break - - logger.info("Starting scheduled bump...") - - threads = await self._db.get_threads_to_bump( - self._config.scheduling.bump_interval_hours - ) - - if threads: - logger.info(f"Found {len(threads)} threads to bump") - results = await self._execute_bump_with_notifications(threads) - success_count = sum(1 for r in results if r.success) - logger.info(f"Scheduled bump completed: {success_count}/{len(threads)} successful") - else: - logger.info("No threads ready for scheduled bump") - - except asyncio.CancelledError: - logger.info("Auto-bump loop cancelled") - break - except Exception as e: - logger.error(f"Error in auto_bump_loop: {e}", exc_info=True) - await asyncio.sleep(AUTO_BUMP_RETRY_DELAY_SECONDS) - - async def start(self) -> None: - """Start bot and all services.""" - try: - # Initialize services - await self._db.connect() - logger.info("Database connected") - - await self._api.start() - logger.info("API client started") - - self._start_time = datetime.now() - self._is_running = True - - # Start auto-bump loop if enabled - if self._config.scheduling.enable_auto_bump: - asyncio.create_task(self._auto_bump_scheduler_loop()) - - logger.info("Bot started successfully!") - await self._dp.start_polling(self._bot) - - except Exception as e: - logger.error(f"Error starting bot: {e}", exc_info=True) - # Cleanup on startup failure - await self._cleanup_resources() - raise - - async def stop(self) -> None: - """Stop bot and cleanup resources.""" - logger.info("Stopping bot...") - self._is_running = False - await self._cleanup_resources() - logger.info("Bot stopped") - - async def _cleanup_resources(self) -> None: - """Cleanup all resources safely.""" - try: - await self._api.close() - except Exception as e: - logger.error(f"Error closing API client: {e}") - - try: - await self._db.close() - except Exception as e: - logger.error(f"Error closing database: {e}") - - try: - await self._bot.session.close() - except Exception as e: - logger.error(f"Error closing bot session: {e}") - - -async def main() -> None: - """Main entry point.""" - try: - config = Config.load() - logger.info("Configuration loaded successfully") - except (FileNotFoundError, ValueError) as e: - logger.error(f"Configuration error: {e}") - sys.exit(1) - - bot = AutoBumpBot(config) - - try: - await bot.start() - except KeyboardInterrupt: - logger.info("Bot interrupted by user") - except Exception as e: - logger.error(f"Bot crashed: {e}", exc_info=True) - sys.exit(1) - finally: - await bot.stop() - - -if __name__ == "__main__": - try: - asyncio.run(main()) - except KeyboardInterrupt: - logger.info("Application terminated") +"""Telegram bot for automatic thread bumping on Lolz.live — with dynamic settings & auth.""" + +import asyncio +import logging +import sys +from datetime import datetime +from typing import Any, Awaitable, Callable, Sequence + +from aiogram import BaseMiddleware, Bot, Dispatcher, F, Router +from aiogram.client.default import DefaultBotProperties +from aiogram.enums import ParseMode +from aiogram.filters import Command, StateFilter +from aiogram.types import Message, CallbackQuery, InlineKeyboardMarkup, InlineKeyboardButton +from aiogram.types import TelegramObject +from aiogram.fsm.context import FSMContext +from aiogram.fsm.state import State, StatesGroup +from aiogram.fsm.storage.memory import MemoryStorage +from aiogram.exceptions import TelegramRetryAfter, TelegramBadRequest + +from config_manager import Config +from api_client import APIClient, BumpResult +from database import Database, BumpStats, Thread + + +NOTIFICATION_DELAY_SECONDS = 0.8 +BUTTON_TEXT_MAX_LENGTH = 30 +AUTO_BUMP_RETRY_DELAY_SECONDS = 60 + + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", + handlers=[ + logging.StreamHandler(sys.stdout), + logging.FileHandler("bot.log", encoding="utf-8"), + ], +) +logger = logging.getLogger(__name__) + + +class BotStates(StatesGroup): + waiting_for_thread_ids = State() + waiting_for_interval = State() + waiting_for_batch_size = State() + + +class AuthMiddleware(BaseMiddleware): + def __init__(self, admin_user_id: int): + super().__init__() + self._admin_user_id = admin_user_id + + async def __call__( + self, + handler: Callable[[TelegramObject, dict[str, Any]], Awaitable[Any]], + event: TelegramObject, + data: dict[str, Any], + ) -> Any: + from_user = getattr(event, "from_user", None) + if from_user is None: + cq = getattr(event, "callback_query", None) + if cq is not None: + from_user = cq.from_user + else: + msg = getattr(event, "message", None) + if msg is not None: + from_user = msg.from_user + + if from_user is None: + return await handler(event, data) + + if from_user.id != self._admin_user_id: + chat_id = None + msg = getattr(event, "message", None) + if msg is not None: + chat_id = msg.chat.id + else: + cq = getattr(event, "callback_query", None) + if cq is not None and cq.message is not None: + chat_id = cq.message.chat.id + + if chat_id: + try: + bot: Bot = data["bot"] + await bot.send_message(chat_id, "⛔ Доступ запрещён.") + except Exception: + pass + return None + + return await handler(event, data) + + +class AutoBumpBot: + __slots__ = ( + "_config", "_bot", "_dp", "_router", "_api", "_db", + "_is_running", "_start_time", "_auto_bump_task", + ) + + def __init__(self, config: Config) -> None: + self._config = config + + self._bot = Bot( + token=config.bot.api_token, + default=DefaultBotProperties(parse_mode=ParseMode.HTML), + ) + + self._dp = Dispatcher(storage=MemoryStorage()) + self._router = Router(name="main_router") + + self._api = APIClient( + config.api.base_url, + config.api.auth_token, + config.api.batch_size, + ) + self._db = Database(config.database.path) + + self._is_running = False + self._start_time: datetime | None = None + self._auto_bump_task: asyncio.Task | None = None + + self._register_handlers() + self._dp.include_router(self._router) + self._dp.update.middleware(AuthMiddleware(config.admin_user_id)) + + def _register_handlers(self) -> None: + # Commands + self._router.message.register( + self._handle_start_command, Command("start") + ) + # FSM state handlers + self._router.message.register( + self._handle_thread_ids_input, + StateFilter(BotStates.waiting_for_thread_ids), + ) + self._router.message.register( + self._handle_interval_input, + StateFilter(BotStates.waiting_for_interval), + ) + self._router.message.register( + self._handle_batch_size_input, + StateFilter(BotStates.waiting_for_batch_size), + ) + # Callbacks + self._router.callback_query.register(self._handle_add_thread_callback, F.data == "add_thread") + self._router.callback_query.register(self._handle_list_threads_callback, F.data == "list_threads") + self._router.callback_query.register(self._handle_delete_menu_callback, F.data == "delete_menu") + self._router.callback_query.register(self._handle_delete_thread_callback, F.data.startswith("delete_")) + self._router.callback_query.register(self._handle_bump_now_callback, F.data == "bump_now") + self._router.callback_query.register(self._handle_stats_callback, F.data == "stats") + self._router.callback_query.register(self._handle_refresh_titles_callback, F.data == "refresh_titles") + self._router.callback_query.register(self._handle_settings_callback, F.data == "settings") + self._router.callback_query.register(self._handle_set_interval_callback, F.data == "set_interval") + self._router.callback_query.register(self._handle_set_batch_size_callback, F.data == "set_batch_size") + self._router.callback_query.register(self._handle_toggle_auto_bump_callback, F.data == "toggle_auto_bump") + self._router.callback_query.register(self._handle_back_to_menu_callback, F.data == "back_to_menu") + + # ─── Keyboards ───────────────────────────────────────────── + + def _main_menu_kb(self) -> InlineKeyboardMarkup: + return InlineKeyboardMarkup(inline_keyboard=[ + [ + InlineKeyboardButton(text="➕ Add topics", callback_data="add_thread"), + InlineKeyboardButton(text="📋 List of topics", callback_data="list_threads"), + ], + [ + InlineKeyboardButton(text="🗑️ Delete topic", callback_data="delete_menu"), + InlineKeyboardButton(text="🚀 Bump topics", callback_data="bump_now"), + ], + [ + InlineKeyboardButton(text="🔄 Refresh", callback_data="refresh_titles"), + InlineKeyboardButton(text="📊 Statistics", callback_data="stats"), + ], + [ + InlineKeyboardButton(text="🛠️ Settings", callback_data="settings"), + InlineKeyboardButton(text="👤 Author", url=self._config.bot.author_url), + ], + ]) + + def _settings_kb(self) -> InlineKeyboardMarkup: + return InlineKeyboardMarkup(inline_keyboard=[ + [InlineKeyboardButton(text="⏰ Set Interval", callback_data="set_interval")], + [InlineKeyboardButton(text="📦 Set Batch Size", callback_data="set_batch_size")], + [InlineKeyboardButton(text="🔄 Toggle Auto-Bump", callback_data="toggle_auto_bump")], + [InlineKeyboardButton(text="↩️ Back to Menu", callback_data="back_to_menu")], + ]) + + async def _send_main_menu(self, chat_id: int, text: str = "Choose an action:") -> None: + await self._bot.send_photo( + chat_id, + photo=self._config.bot.img_url, + caption=text, + reply_markup=self._main_menu_kb(), + ) + + # ─── Utilities ────────────────────────────────────────────── + + @staticmethod + def _truncate_text_safely(text: str, max_length: int) -> str: + if len(text) <= max_length: + return text + truncated = text[:max_length] + try: + truncated.encode("utf-8") + return truncated + except UnicodeEncodeError: + return AutoBumpBot._truncate_text_safely(text[: max_length - 1], max_length - 1) + + @staticmethod + def _calculate_uptime(start_time: datetime) -> str: + delta = datetime.now() - start_time + total_seconds = int(delta.total_seconds()) + days = total_seconds // 86400 + hours = (total_seconds % 86400) // 3600 + minutes = (total_seconds % 3600) // 60 + if days > 0: + return f"{days}d {hours}h {minutes}m" + elif hours > 0: + return f"{hours}h {minutes}m" + else: + return f"{minutes}m" + + async def _get_interval(self) -> float: + val = await self._db.get_setting("bump_interval_hours", "12") + return float(val) + + async def _get_batch_size(self) -> int: + val = await self._db.get_setting("batch_size", "10") + return int(val) + + async def _is_auto_bump_enabled(self) -> bool: + val = await self._db.get_setting("enable_auto_bump", "true") + return val.lower() == "true" + + # ─── Start ───────────────────────────────────────────────── + + async def _handle_start_command(self, message: Message) -> None: + try: + interval = await self._get_interval() + await message.answer_photo( + photo=self._config.bot.img_url, + caption=( + "🤖 QIYANA AUTO-BUMP BOT\n\n" + "Бот для автоматического поднятия тем на Lolz.live\n\n" + f"⏰ Автоподнятие каждые {interval:.0f}ч" + ), + reply_markup=self._main_menu_kb(), + ) + except TelegramBadRequest as e: + logger.error(f"Failed to send start message: {e}") + await message.answer("❌ Ошибка отправки сообщения. Попробуйте /start снова.") + + # ─── Add Thread ───────────────────────────────────────────── + + async def _handle_add_thread_callback(self, callback: CallbackQuery, state: FSMContext) -> None: + await callback.answer() + if not callback.message: + return + await callback.message.answer( + "📝 Введите ID тем через запятую для добавления:\n" + "Пример: 12345, 67890, 11111" + ) + await state.set_state(BotStates.waiting_for_thread_ids) + + async def _handle_thread_ids_input(self, message: Message, state: FSMContext) -> None: + try: + thread_ids = [tid.strip() for tid in message.text.split(",") if tid.strip()] + + if not thread_ids: + await message.answer("❌ Не указаны ID тем") + return + + valid_ids: list[str] = [] + errors: list[str] = [] + + for thread_id in thread_ids: + if not thread_id.isdigit(): + errors.append(f"❌ {thread_id} - неверный формат") + else: + valid_ids.append(thread_id) + + if not valid_ids: + await message.answer("❌ Нет валидных ID тем") + return + + status_msg = await message.answer( + f"⏳ Добавляю {len(valid_ids)} тем и загружаю названия..." + ) + + # Fetch real titles via batch API + titles_map: dict[str, str] = {} + try: + threads_info = await self._api.get_threads_info_batch(valid_ids) + for i, info in enumerate(threads_info): + tid = valid_ids[i] + titles_map[tid] = info.title if info else f"Thread {tid}" + except Exception as e: + logger.error(f"Error fetching titles during add: {e}") + for tid in valid_ids: + titles_map.setdefault(tid, f"Thread {tid}") + + added: list[str] = [] + for thread_id in valid_ids: + title = titles_map.get(thread_id, f"Thread {thread_id}") + success = await self._db.add_thread(thread_id, title) + if success: + added.append(f"✅ {thread_id} — {title}") + else: + errors.append(f"⚠️ {thread_id} - уже добавлена") + + result_parts = ["📊 Результат добавления:\n"] + if added: + result_parts.append("\n✅ Добавлено:\n") + result_parts.append("\n".join(added)) + result_parts.append("\n") + if errors: + result_parts.append("\n❌ Ошибки:\n") + result_parts.append("\n".join(errors)) + + await status_msg.edit_text("".join(result_parts)) + await self._send_main_menu(message.chat.id) + + except Exception as e: + logger.error(f"Error in thread IDs input handler: {e}", exc_info=True) + await message.answer(f"❌ Критическая ошибка: {str(e)}") + finally: + await state.clear() + + # ─── List Threads ─────────────────────────────────────────── + + async def _handle_list_threads_callback(self, callback: CallbackQuery) -> None: + await callback.answer() + if not callback.message: + return + try: + threads = await self._db.get_all_threads() + if not threads: + await callback.message.answer("📭 Список тем пуст\n\nДобавьте темы через кнопку ➕") + await self._send_main_menu(callback.message.chat.id) + return + + text_parts = [f"📋 Список тем ({len(threads)}):\n\n"] + + for thread in threads: + status = "🆕 Новая" + if thread.last_bumped: + try: + dt = datetime.fromisoformat(thread.last_bumped.replace("Z", "+00:00")) + status = f"✅ {dt.strftime('%d.%m.%Y %H:%M')}" + except ValueError: + status = f"✅ {thread.last_bumped[:16]}" + + text_parts.append( + f"ID: {thread.id}\n" + f"Название: {thread.title}\n" + f"Статус: {status}\n\n" + ) + + await callback.message.answer("".join(text_parts)) + await self._send_main_menu(callback.message.chat.id) + + except Exception as e: + logger.error(f"Error listing threads: {e}", exc_info=True) + await callback.message.answer(f"❌ Ошибка: {str(e)}") + + # ─── Refresh Titles ───────────────────────────────────────── + + async def _handle_refresh_titles_callback(self, callback: CallbackQuery) -> None: + await callback.answer() + if not callback.message: + return + try: + threads = await self._db.get_all_threads() + if not threads: + await callback.message.answer("📭 Список тем пуст") + await self._send_main_menu(callback.message.chat.id) + return + + status_msg = await callback.message.answer( + f"🔄 Обновляю названия {len(threads)} тем..." + ) + + thread_ids = [t.id for t in threads] + threads_info = await self._api.get_threads_info_batch(thread_ids) + + updated = 0 + for i, info in enumerate(threads_info): + if info and info.title and info.title != threads[i].title: + await self._db.update_thread_title(threads[i].id, info.title) + updated += 1 + + await status_msg.edit_text( + f"🔄 Названия обновлены!\n\n" + f"📝 Всего тем: {len(threads)}\n" + f"✅ Обновлено: {updated}\n" + f"⏭️ Без изменений: {len(threads) - updated}" + ) + await self._send_main_menu(callback.message.chat.id) + + except Exception as e: + logger.error(f"Error refreshing titles: {e}", exc_info=True) + await callback.message.answer(f"❌ Ошибка обновления: {str(e)}") + + # ─── Delete Thread ────────────────────────────────────────── + + async def _handle_delete_menu_callback(self, callback: CallbackQuery) -> None: + await callback.answer() + if not callback.message: + return + try: + threads = await self._db.get_all_threads() + if not threads: + await callback.message.answer("📭 Список тем пуст") + await self._send_main_menu(callback.message.chat.id) + return + + buttons = [ + [InlineKeyboardButton( + text=f"{thread.id} - {self._truncate_text_safely(thread.title, BUTTON_TEXT_MAX_LENGTH)}", + callback_data=f"delete_{thread.id}", + )] + for thread in threads + ] + await callback.message.answer( + "🗑️ Выберите тему для удаления:", + reply_markup=InlineKeyboardMarkup(inline_keyboard=buttons), + ) + except Exception as e: + logger.error(f"Error showing delete menu: {e}", exc_info=True) + await callback.message.answer(f"❌ Ошибка: {str(e)}") + + async def _handle_delete_thread_callback(self, callback: CallbackQuery) -> None: + if not callback.message: + await callback.answer("❌ Ошибка: сообщение не найдено", show_alert=True) + return + try: + thread_id = callback.data.split("_", 1)[1] + success = await self._db.delete_thread(thread_id) + if success: + await callback.answer(f"✅ Тема {thread_id} удалена", show_alert=True) + await callback.message.answer(f"✅ Тема {thread_id} успешно удалена из базы данных") + else: + await callback.answer(f"❌ Тема {thread_id} не найдена", show_alert=True) + await self._send_main_menu(callback.message.chat.id) + except Exception as e: + logger.error(f"Error deleting thread: {e}", exc_info=True) + await callback.answer(f"❌ Ошибка: {str(e)}", show_alert=True) + + # ─── Bump Now ─────────────────────────────────────────────── + + async def _handle_bump_now_callback(self, callback: CallbackQuery) -> None: + await callback.answer() + if not callback.message: + return + try: + threads = await self._db.get_threads_to_bump(0) + if not threads: + await callback.message.answer("📭 Нет тем для поднятия\n\nДобавьте темы через кнопку ➕") + await self._send_main_menu(callback.message.chat.id) + return + + status_msg = await callback.message.answer( + f"🚀 Начинаю поднятие {len(threads)} тем...\n\n" + "Используется batch API (до 10 тем за запрос)\n" + "Подождите, это может занять некоторое время..." + ) + + results = await self._execute_bump_with_notifications( + threads, chat_id=callback.message.chat.id + ) + + success_count = sum(1 for r in results if r.success) + await status_msg.edit_text( + f"📊 Поднятие завершено!\n\n" + f"✅ Успешно: {success_count}\n" + f"❌ Ошибок: {len(threads) - success_count}\n" + f"📝 Всего тем: {len(threads)}" + ) + await self._send_main_menu(callback.message.chat.id) + + except Exception as e: + logger.error(f"Error bumping threads: {e}", exc_info=True) + await callback.message.answer(f"❌ Ошибка: {str(e)}") + + # ─── Statistics ───────────────────────────────────────────── + + async def _handle_stats_callback(self, callback: CallbackQuery) -> None: + await callback.answer() + if not callback.message: + return + try: + threads = await self._db.get_all_threads() + interval = await self._get_interval() + threads_ready = await self._db.get_threads_to_bump(interval) + stats = await self._db.get_bump_stats() + batch_size = await self._get_batch_size() + auto_enabled = await self._is_auto_bump_enabled() + + uptime = "0 минут" + if self._start_time: + uptime = self._calculate_uptime(self._start_time) + + success_rate = 0.0 + if stats.total_bumps > 0: + success_rate = (stats.successful_bumps / stats.total_bumps) * 100 + + text = ( + "📊 Статистика бота\n\n" + f"📋 Темы:\n" + f"• Всего в списке: {len(threads)}\n" + f"• Готовы к поднятию: {len(threads_ready)}\n" + f"• Ожидают времени: {len(threads) - len(threads_ready)}\n\n" + f"🚀 Поднятия:\n" + f"• Всего попыток: {stats.total_bumps}\n" + f"• Успешных: {stats.successful_bumps}\n" + f"• Успешность: {success_rate:.1f}%\n" + f"• Последний бамп: {stats.last_bump_time or '—'}\n\n" + f"⚙️ Настройки:\n" + f"• Интервал: {interval:.0f}ч\n" + f"• Batch size: {batch_size}\n" + f"• Автобамп: {'✅ Вкл' if auto_enabled else '❌ Выкл'}\n\n" + f"⏱️ Работа:\n" + f"• Время работы: {uptime}" + ) + + await callback.message.answer(text) + await self._send_main_menu(callback.message.chat.id) + + except Exception as e: + logger.error(f"Error showing stats: {e}", exc_info=True) + await callback.message.answer(f"❌ Ошибка: {str(e)}") + + # ─── Settings ─────────────────────────────────────────────── + + async def _handle_settings_callback(self, callback: CallbackQuery) -> None: + await callback.answer() + if not callback.message: + return + interval = await self._get_interval() + batch_size = await self._get_batch_size() + auto_enabled = await self._is_auto_bump_enabled() + + text = ( + "🛠️ Текущие настройки\n\n" + f"⏰ Интервал бампа: {interval:.0f}ч\n" + f"📦 Batch size: {batch_size}\n" + f"🔄 Автобамп: {'✅ Вкл' if auto_enabled else '❌ Выкл'}\n" + ) + await callback.message.answer(text, reply_markup=self._settings_kb()) + + async def _handle_set_interval_callback(self, callback: CallbackQuery, state: FSMContext) -> None: + await callback.answer() + if not callback.message: + return + interval = await self._get_interval() + await callback.message.answer( + f"Текущий интервал: {interval:.0f}ч\n\n" + "Введите новый интервал в часах (например: 6 или 24):" + ) + await state.set_state(BotStates.waiting_for_interval) + + async def _handle_interval_input(self, message: Message, state: FSMContext) -> None: + try: + new_interval = float(message.text.strip()) + if new_interval <= 0: + await message.answer("❌ Интервал должен быть положительным числом") + return + + await self._db.set_setting("bump_interval_hours", str(new_interval)) + logger.info(f"Bump interval changed to {new_interval}h by user {message.from_user.id}") + + # Restart auto-bump loop if running + if self._is_running and await self._is_auto_bump_enabled(): + self._restart_auto_bump() + + await message.answer(f"✅ Интервал изменён на {new_interval:.0f}ч") + await self._send_main_menu(message.chat.id) + + except ValueError: + await message.answer("❌ Введите число (например: 12)") + except Exception as e: + logger.error(f"Error changing interval: {e}") + await message.answer(f"❌ Ошибка: {str(e)}") + finally: + await state.clear() + + async def _handle_set_batch_size_callback(self, callback: CallbackQuery, state: FSMContext) -> None: + await callback.answer() + if not callback.message: + return + batch_size = await self._get_batch_size() + await callback.message.answer( + f"Текущий batch size: {batch_size}\n\n" + "Введите новый размер (от 1 до 10):" + ) + await state.set_state(BotStates.waiting_for_batch_size) + + async def _handle_batch_size_input(self, message: Message, state: FSMContext) -> None: + try: + new_size = int(message.text.strip()) + if new_size < 1 or new_size > 10: + await message.answer("❌ Batch size должен быть от 1 до 10") + return + + await self._db.set_setting("batch_size", str(new_size)) + self._api._batch_size = new_size + logger.info(f"Batch size changed to {new_size} by user {message.from_user.id}") + + await message.answer(f"✅ Batch size изменён на {new_size}") + await self._send_main_menu(message.chat.id) + + except ValueError: + await message.answer("❌ Введите целое число (от 1 до 10)") + except Exception as e: + logger.error(f"Error changing batch size: {e}") + await message.answer(f"❌ Ошибка: {str(e)}") + finally: + await state.clear() + + async def _handle_toggle_auto_bump_callback(self, callback: CallbackQuery) -> None: + await callback.answer() + if not callback.message: + return + + current = await self._is_auto_bump_enabled() + new_state = not current + await self._db.set_setting("enable_auto_bump", str(new_state).lower()) + + if new_state: + self._restart_auto_bump() + status = "✅ Включён" + else: + status = "❌ Выключен" + + logger.info(f"Auto-bump toggled to {new_state} by user {callback.from_user.id}") + await callback.message.answer(f"🔄 Автобамп: {status}") + await self._send_main_menu(callback.message.chat.id) + + # ─── Back to Menu ─────────────────────────────────────────── + + async def _handle_back_to_menu_callback(self, callback: CallbackQuery) -> None: + await callback.answer() + if not callback.message: + return + await self._send_main_menu(callback.message.chat.id) + + # ─── Bump Execution ───────────────────────────────────────── + + async def _execute_bump_with_notifications( + self, threads: Sequence[Thread], chat_id: int | None = None + ) -> list[BumpResult]: + if not threads: + return [] + + thread_ids = [thread.id for thread in threads] + logger.info(f"Starting bump for {len(thread_ids)} threads: {thread_ids}") + + results = await self._api.bump_threads_batch(thread_ids) + + for result in results: + if result.success: + logger.info( + f"✅ BUMP SUCCESS | Thread: {result.thread_id} | " + f"Status: {result.status.value} | Message: {result.message}" + ) + else: + logger.error( + f"❌ BUMP FAILED | Thread: {result.thread_id} | " + f"Status: {result.status.value} | Message: {result.message}" + ) + + success_count = 0 + for result in results: + if result.success: + success_count += 1 + await self._db.update_last_bumped(result.thread_id) + + total = len(results) + await self._db.increment_bump_stats(success_count, total) + + if chat_id: + await self._send_bump_notifications(chat_id, results) + + logger.info( + f"Batch bump completed: {success_count}/{len(results)} successful | " + f"Failed: {len(results) - success_count}" + ) + + return results + + async def _send_bump_notifications(self, chat_id: int, results: list[BumpResult]) -> None: + for i, result in enumerate(results, 1): + try: + await self._bot.send_message(chat_id, f"[{i}/{len(results)}] {result.message}") + if i < len(results): + await asyncio.sleep(NOTIFICATION_DELAY_SECONDS) + except TelegramRetryAfter as e: + logger.warning(f"Rate limit hit, waiting {e.retry_after} seconds") + await asyncio.sleep(e.retry_after) + try: + await self._bot.send_message(chat_id, f"[{i}/{len(results)}] {result.message}") + except Exception as retry_error: + logger.error(f"Failed to send notification after retry: {retry_error}") + except Exception as e: + logger.error(f"Failed to send notification: {e}") + + # ─── Auto-Bump Scheduler ─────────────────────────────────── + + def _restart_auto_bump(self) -> None: + if self._auto_bump_task and not self._auto_bump_task.done(): + self._auto_bump_task.cancel() + self._auto_bump_task = asyncio.create_task(self._auto_bump_scheduler_loop()) + + async def _auto_bump_scheduler_loop(self) -> None: + logger.info("Auto-bump scheduler started") + while self._is_running: + try: + interval = await self._get_interval() + sleep_seconds = interval * 3600 + logger.info(f"Next scheduled bump in {interval:.0f} hours") + await asyncio.sleep(sleep_seconds) + + if not self._is_running: + break + + auto_enabled = await self._is_auto_bump_enabled() + if not auto_enabled: + logger.info("Auto-bump disabled, skipping cycle") + continue + + logger.info("Starting scheduled bump...") + threads = await self._db.get_threads_to_bump(interval) + + if threads: + logger.info(f"Found {len(threads)} threads to bump") + results = await self._execute_bump_with_notifications(threads) + success_count = sum(1 for r in results if r.success) + logger.info(f"Scheduled bump completed: {success_count}/{len(threads)} successful") + else: + logger.info("No threads ready for scheduled bump") + + except asyncio.CancelledError: + logger.info("Auto-bump loop cancelled") + break + except Exception as e: + logger.error(f"Error in auto_bump_loop: {e}", exc_info=True) + await asyncio.sleep(AUTO_BUMP_RETRY_DELAY_SECONDS) + + # ─── Lifecycle ────────────────────────────────────────────── + + async def start(self) -> None: + try: + await self._db.connect() + logger.info("Database connected") + + await self._db.seed_defaults(self._config) + logger.info("Database seeded with default settings") + + await self._api.start() + logger.info("API client started") + + self._start_time = datetime.now() + self._is_running = True + + if await self._is_auto_bump_enabled(): + self._auto_bump_task = asyncio.create_task(self._auto_bump_scheduler_loop()) + + logger.info("Bot started successfully!") + await self._dp.start_polling(self._bot) + + except Exception as e: + logger.error(f"Error starting bot: {e}", exc_info=True) + await self._cleanup_resources() + raise + + async def stop(self) -> None: + logger.info("Stopping bot...") + self._is_running = False + if self._auto_bump_task and not self._auto_bump_task.done(): + self._auto_bump_task.cancel() + await self._cleanup_resources() + logger.info("Bot stopped") + + async def _cleanup_resources(self) -> None: + try: + await self._api.close() + except Exception as e: + logger.error(f"Error closing API client: {e}") + try: + await self._db.close() + except Exception as e: + logger.error(f"Error closing database: {e}") + try: + await self._bot.session.close() + except Exception as e: + logger.error(f"Error closing bot session: {e}") + + +async def main() -> None: + try: + config = Config.load() + logger.info("Configuration loaded from .env successfully") + except (FileNotFoundError, ValueError) as e: + logger.error(f"Configuration error: {e}") + sys.exit(1) + + bot = AutoBumpBot(config) + + try: + await bot.start() + except KeyboardInterrupt: + logger.info("Bot interrupted by user") + except Exception as e: + logger.error(f"Bot crashed: {e}", exc_info=True) + sys.exit(1) + finally: + await bot.stop() + + +if __name__ == "__main__": + try: + asyncio.run(main()) + except KeyboardInterrupt: + logger.info("Application terminated") diff --git a/config_manager.py b/config_manager.py index 86dd975..22436b3 100644 --- a/config_manager.py +++ b/config_manager.py @@ -1,120 +1,108 @@ -"""Configuration management with validation and type safety.""" - -import json -from pathlib import Path -from dataclasses import dataclass -from typing import Self - - -@dataclass(frozen=True, slots=True) -class BotConfig: - """Telegram bot configuration.""" - api_token: str - img_url: str - author_url: str - - -@dataclass(frozen=True, slots=True) -class APIConfig: - """Lolz API configuration.""" - base_url: str - auth_token: str - batch_size: int - - -@dataclass(frozen=True, slots=True) -class DatabaseConfig: - """Database configuration.""" - path: str - - -@dataclass(frozen=True, slots=True) -class SchedulingConfig: - """Scheduling configuration.""" - bump_interval_hours: float - bump_delay_seconds: float - enable_auto_bump: bool - - -@dataclass(frozen=True, slots=True) -class Config: - """Application configuration.""" - bot: BotConfig - api: APIConfig - database: DatabaseConfig - scheduling: SchedulingConfig - - @classmethod - def load(cls, config_path: str = "config.json") -> Self: - """Load and validate configuration from JSON file.""" - path = Path(config_path) - if not path.exists(): - raise FileNotFoundError(f"Config file not found: {config_path}") - - with open(path, encoding="utf-8") as f: - data = json.load(f) - - # Validate required fields - cls._validate_config(data) - - return cls( - bot=BotConfig( - api_token=data["bot"]["api_token"], - img_url=data["bot"]["img_url"], - author_url=data["bot"]["author_url"] - ), - api=APIConfig( - base_url=data["api"]["base_url"].rstrip("/"), - auth_token=data["api"]["auth_token"], - batch_size=int(data["api"].get("batch_size", 10)) - ), - database=DatabaseConfig( - path=data["database"]["path"] - ), - scheduling=SchedulingConfig( - bump_interval_hours=float(data["scheduling"]["bump_interval_hours"]), - bump_delay_seconds=float(data["scheduling"].get("bump_delay_seconds", 2)), - enable_auto_bump=bool(data["scheduling"]["enable_auto_bump"]) - ) - ) - - @staticmethod - def _validate_config(data: dict) -> None: - """Validate configuration structure and required fields.""" - required_fields = { - "bot": ["api_token", "img_url", "author_url"], - "api": ["base_url", "auth_token"], - "database": ["path"], - "scheduling": ["bump_interval_hours", "enable_auto_bump"] - } - - for section, fields in required_fields.items(): - if section not in data: - raise ValueError(f"Missing config section: {section}") - - for field in fields: - if field not in data[section]: - raise ValueError(f"Missing config field: {section}.{field}") - - value = data[section][field] - if isinstance(value, str) and not value.strip(): - raise ValueError(f"Empty config field: {section}.{field}") - - # Validate token formats - bot_token = data["bot"]["api_token"] - if "YOUR_" in bot_token or not bot_token: - raise ValueError("bot.api_token not configured - please set your Telegram bot token") - - api_token = data["api"]["auth_token"] - if "YOUR_" in api_token or not api_token: - raise ValueError("api.auth_token not configured - please set your Lolz API token") - - # Validate interval - interval = data["scheduling"]["bump_interval_hours"] - if not isinstance(interval, (int, float)) or interval <= 0: - raise ValueError("scheduling.bump_interval_hours must be positive number") - - # Validate batch size - batch_size = data["api"].get("batch_size", 10) - if not isinstance(batch_size, int) or batch_size < 1 or batch_size > 10: - raise ValueError("api.batch_size must be between 1 and 10") +"""Configuration management with .env validation and type safety.""" + +import os +from dataclasses import dataclass +from typing import Self + +from dotenv import load_dotenv + + +@dataclass(frozen=True, slots=True) +class BotConfig: + api_token: str + img_url: str + author_url: str + + +@dataclass(frozen=True, slots=True) +class APIConfig: + base_url: str + auth_token: str + batch_size: int + + +@dataclass(frozen=True, slots=True) +class DatabaseConfig: + path: str + + +@dataclass(frozen=True, slots=True) +class SchedulingConfig: + bump_interval_hours: float + bump_delay_seconds: float + enable_auto_bump: bool + + +@dataclass(frozen=True, slots=True) +class Config: + bot: BotConfig + api: APIConfig + database: DatabaseConfig + scheduling: SchedulingConfig + admin_user_id: int + + @classmethod + def load(cls, env_path: str = ".env") -> Self: + if not os.path.exists(env_path): + raise FileNotFoundError(f".env file not found: {env_path}") + + load_dotenv(env_path) + + bot_token = os.getenv("BOT_API_TOKEN", "") + api_token = os.getenv("API_AUTH_TOKEN", "") + admin_user_id = int(os.getenv("ADMIN_USER_ID", "0")) + + cls._validate_tokens(bot_token, api_token, admin_user_id) + + bot = BotConfig( + api_token=bot_token, + img_url=os.getenv("BOT_IMG_URL", ""), + author_url=os.getenv("BOT_AUTHOR_URL", ""), + ) + + batch_size = int(os.getenv("API_BATCH_SIZE", "10")) + if batch_size < 1 or batch_size > 10: + raise ValueError("API_BATCH_SIZE must be between 1 and 10") + + api = APIConfig( + base_url=os.getenv("API_BASE_URL", "").rstrip("/"), + auth_token=api_token, + batch_size=batch_size, + ) + + db = DatabaseConfig( + path=os.getenv("DB_PATH", "threads.db"), + ) + + interval = float(os.getenv("BUMP_INTERVAL_HOURS", "12")) + if interval <= 0: + raise ValueError("BUMP_INTERVAL_HOURS must be positive") + + scheduling = SchedulingConfig( + bump_interval_hours=interval, + bump_delay_seconds=float(os.getenv("BUMP_DELAY_SECONDS", "2")), + enable_auto_bump=os.getenv("ENABLE_AUTO_BUMP", "true").lower() == "true", + ) + + return cls( + bot=bot, + api=api, + database=db, + scheduling=scheduling, + admin_user_id=admin_user_id, + ) + + @staticmethod + def _validate_tokens(bot_token: str, api_token: str, admin_user_id: int) -> None: + if not bot_token or "YOUR_" in bot_token: + raise ValueError( + "BOT_API_TOKEN not configured — set your Telegram bot token in .env" + ) + if not api_token or "YOUR_" in api_token: + raise ValueError( + "API_AUTH_TOKEN not configured — set your Lolz API token in .env" + ) + if admin_user_id == 0: + raise ValueError( + "ADMIN_USER_ID not configured — set your Telegram user ID in .env" + ) diff --git a/database.py b/database.py index 5081921..fc1c723 100644 --- a/database.py +++ b/database.py @@ -1,259 +1,295 @@ -"""Database layer with connection pooling and type safety.""" - -import aiosqlite -import logging -from typing import Self -from dataclasses import dataclass - - -logger = logging.getLogger(__name__) - - -@dataclass(frozen=True, slots=True) -class Thread: - """Thread model with immutable fields.""" - id: str - title: str - last_bumped: str | None = None - - -class Database: - """SQLite database manager with connection pooling and proper error handling.""" - - __slots__ = ("_db_path", "_connection") - - def __init__(self, db_path: str) -> None: - if not db_path: - raise ValueError("db_path is required") - - self._db_path = db_path - self._connection: aiosqlite.Connection | None = None - - async def __aenter__(self) -> Self: - """Async context manager entry.""" - await self.connect() - return self - - async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: - """Async context manager exit.""" - await self.close() - - async def connect(self) -> None: - """Establish database connection and initialize schema.""" - if self._connection is None: - try: - self._connection = await aiosqlite.connect(self._db_path) - # Enable WAL mode for better concurrency - await self._connection.execute("PRAGMA journal_mode=WAL") - await self._initialize_schema() - logger.info(f"Database connected: {self._db_path}") - except Exception as e: - logger.error(f"Failed to connect to database: {e}") - raise - - async def close(self) -> None: - """Close database connection safely.""" - if self._connection: - try: - await self._connection.close() - self._connection = None - logger.info("Database connection closed") - except Exception as e: - logger.error(f"Error closing database connection: {e}") - - async def _initialize_schema(self) -> None: - """Initialize database schema with tables and indexes.""" - if not self._connection: - raise RuntimeError("Database not connected") - - try: - await self._connection.execute(""" - CREATE TABLE IF NOT EXISTS threads ( - id TEXT PRIMARY KEY, - title TEXT NOT NULL, - last_bumped TEXT, - created_at TEXT DEFAULT CURRENT_TIMESTAMP - ) - """) - - # Create index for efficient queries on last_bumped - await self._connection.execute(""" - CREATE INDEX IF NOT EXISTS idx_last_bumped - ON threads(last_bumped) - """) - - await self._connection.commit() - logger.info("Database schema initialized") - except Exception as e: - logger.error(f"Failed to initialize schema: {e}") - raise - - def _ensure_connected(self) -> None: - """Ensure database is connected before operations.""" - if not self._connection: - raise RuntimeError("Database not connected. Call connect() first.") - - async def add_thread(self, thread_id: str, title: str) -> bool: - """ - Add thread to database. - - Args: - thread_id: Unique thread identifier - title: Thread title - - Returns: - True if added successfully, False if thread already exists - """ - self._ensure_connected() - - try: - await self._connection.execute( - "INSERT INTO threads (id, title) VALUES (?, ?)", - (thread_id, title) - ) - await self._connection.commit() - return True - except aiosqlite.IntegrityError: - # Thread already exists (PRIMARY KEY constraint) - return False - except Exception as e: - logger.error(f"Error adding thread {thread_id}: {e}") - raise - - async def get_all_threads(self) -> list[Thread]: - """ - Get all threads ordered by ID. - - Returns: - List of Thread objects - """ - self._ensure_connected() - - try: - async with self._connection.execute( - "SELECT id, title, last_bumped FROM threads ORDER BY id" - ) as cursor: - rows = await cursor.fetchall() - return [ - Thread(id=row[0], title=row[1], last_bumped=row[2]) - for row in rows - ] - except Exception as e: - logger.error(f"Error fetching all threads: {e}") - raise - - async def get_threads_to_bump(self, interval_hours: float) -> list[Thread]: - """ - Get threads ready for bumping based on interval. - - Args: - interval_hours: Minimum hours since last bump - - Returns: - List of Thread objects ready to bump - """ - self._ensure_connected() - - if interval_hours < 0: - raise ValueError("interval_hours must be non-negative") - - try: - async with self._connection.execute( - """ - SELECT id, title, last_bumped FROM threads - WHERE last_bumped IS NULL - OR datetime(last_bumped, '+' || ? || ' hours') <= datetime('now') - ORDER BY last_bumped ASC NULLS FIRST - """, - (interval_hours,) - ) as cursor: - rows = await cursor.fetchall() - return [ - Thread(id=row[0], title=row[1], last_bumped=row[2]) - for row in rows - ] - except Exception as e: - logger.error(f"Error fetching threads to bump: {e}") - raise - - async def update_last_bumped(self, thread_id: str) -> None: - """ - Update last bumped timestamp for thread to current time. - - Args: - thread_id: Thread identifier to update - """ - self._ensure_connected() - - try: - await self._connection.execute( - "UPDATE threads SET last_bumped = datetime('now') WHERE id = ?", - (thread_id,) - ) - await self._connection.commit() - except Exception as e: - logger.error(f"Error updating last_bumped for thread {thread_id}: {e}") - raise - - async def delete_thread(self, thread_id: str) -> bool: - """ - Delete thread from database. - - Args: - thread_id: Thread identifier to delete - - Returns: - True if deleted successfully, False if thread not found - """ - self._ensure_connected() - - try: - cursor = await self._connection.execute( - "DELETE FROM threads WHERE id = ?", - (thread_id,) - ) - await self._connection.commit() - return cursor.rowcount > 0 - except Exception as e: - logger.error(f"Error deleting thread {thread_id}: {e}") - raise - - async def get_thread_count(self) -> int: - """ - Get total number of threads in database. - - Returns: - Total thread count - """ - self._ensure_connected() - - try: - async with self._connection.execute("SELECT COUNT(*) FROM threads") as cursor: - row = await cursor.fetchone() - return row[0] if row else 0 - except Exception as e: - logger.error(f"Error getting thread count: {e}") - raise - - async def thread_exists(self, thread_id: str) -> bool: - """ - Check if thread exists in database. - - Args: - thread_id: Thread identifier to check - - Returns: - True if thread exists, False otherwise - """ - self._ensure_connected() - - try: - async with self._connection.execute( - "SELECT 1 FROM threads WHERE id = ? LIMIT 1", - (thread_id,) - ) as cursor: - row = await cursor.fetchone() - return row is not None - except Exception as e: - logger.error(f"Error checking thread existence {thread_id}: {e}") - raise +"""Database layer with dynamic settings and stats persistence.""" + +import aiosqlite +import logging +from typing import Self +from dataclasses import dataclass + +from config_manager import Config + + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True, slots=True) +class Thread: + id: str + title: str + last_bumped: str | None = None + + +@dataclass(frozen=True, slots=True) +class BumpStats: + total_bumps: int + successful_bumps: int + last_bump_time: str | None = None + + +class Database: + __slots__ = ("_db_path", "_connection") + + def __init__(self, db_path: str) -> None: + if not db_path: + raise ValueError("db_path is required") + + self._db_path = db_path + self._connection: aiosqlite.Connection | None = None + + async def __aenter__(self) -> Self: + await self.connect() + return self + + async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: + await self.close() + + async def connect(self) -> None: + if self._connection is None: + try: + self._connection = await aiosqlite.connect(self._db_path) + await self._connection.execute("PRAGMA journal_mode=WAL") + await self._connection.execute("PRAGMA foreign_keys=ON") + await self._initialize_schema() + logger.info(f"Database connected: {self._db_path}") + except Exception as e: + logger.error(f"Failed to connect to database: {e}") + raise + + async def close(self) -> None: + if self._connection: + try: + await self._connection.close() + self._connection = None + logger.info("Database connection closed") + except Exception as e: + logger.error(f"Error closing database connection: {e}") + + async def _initialize_schema(self) -> None: + if not self._connection: + raise RuntimeError("Database not connected") + + try: + await self._connection.execute(""" + CREATE TABLE IF NOT EXISTS threads ( + id TEXT PRIMARY KEY, + title TEXT NOT NULL, + last_bumped TEXT, + created_at TEXT DEFAULT CURRENT_TIMESTAMP + ) + """) + + await self._connection.execute(""" + CREATE INDEX IF NOT EXISTS idx_last_bumped + ON threads(last_bumped) + """) + + await self._connection.execute(""" + CREATE TABLE IF NOT EXISTS settings ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL, + updated_at TEXT DEFAULT CURRENT_TIMESTAMP + ) + """) + + await self._connection.execute(""" + CREATE TABLE IF NOT EXISTS bump_stats ( + id INTEGER PRIMARY KEY CHECK (id = 1), + total_bumps INTEGER DEFAULT 0, + successful_bumps INTEGER DEFAULT 0, + last_bump_time TEXT + ) + """) + + await self._connection.commit() + logger.info("Database schema initialized") + except Exception as e: + logger.error(f"Failed to initialize schema: {e}") + raise + + async def seed_defaults(self, config: Config) -> None: + await self._seed_settings(config) + await self._seed_stats() + + async def _seed_settings(self, config: Config) -> None: + defaults = { + "bump_interval_hours": str(config.scheduling.bump_interval_hours), + "bump_delay_seconds": str(config.scheduling.bump_delay_seconds), + "enable_auto_bump": str(config.scheduling.enable_auto_bump).lower(), + "batch_size": str(config.api.batch_size), + } + for key, value in defaults.items(): + await self._connection.execute( + "INSERT OR IGNORE INTO settings (key, value) VALUES (?, ?)", + (key, value), + ) + await self._connection.commit() + + async def _seed_stats(self) -> None: + await self._connection.execute( + "INSERT OR IGNORE INTO bump_stats (id, total_bumps, successful_bumps) VALUES (1, 0, 0)" + ) + await self._connection.commit() + + def _ensure_connected(self) -> None: + if not self._connection: + raise RuntimeError("Database not connected. Call connect() first.") + + # ─── Settings ─────────────────────────────────────────────── + + async def get_setting(self, key: str, default: str | None = None) -> str | None: + self._ensure_connected() + async with self._connection.execute( + "SELECT value FROM settings WHERE key = ?", (key,) + ) as cursor: + row = await cursor.fetchone() + return row[0] if row else default + + async def set_setting(self, key: str, value: str) -> None: + self._ensure_connected() + await self._connection.execute( + "INSERT OR REPLACE INTO settings (key, value, updated_at) VALUES (?, ?, datetime('now'))", + (key, value), + ) + await self._connection.commit() + + async def get_all_settings(self) -> dict[str, str]: + self._ensure_connected() + async with self._connection.execute("SELECT key, value FROM settings") as cursor: + rows = await cursor.fetchall() + return {row[0]: row[1] for row in rows} + + # ─── Bump Statistics ──────────────────────────────────────── + + async def get_bump_stats(self) -> BumpStats: + self._ensure_connected() + async with self._connection.execute( + "SELECT total_bumps, successful_bumps, last_bump_time FROM bump_stats WHERE id = 1" + ) as cursor: + row = await cursor.fetchone() + if row: + return BumpStats( + total_bumps=row[0], + successful_bumps=row[1], + last_bump_time=row[2], + ) + return BumpStats(total_bumps=0, successful_bumps=0) + + async def increment_bump_stats(self, success_count: int, total_count: int) -> None: + self._ensure_connected() + await self._connection.execute( + """ + UPDATE bump_stats SET + total_bumps = total_bumps + ?, + successful_bumps = successful_bumps + ?, + last_bump_time = datetime('now') + WHERE id = 1 + """, + (total_count, success_count), + ) + await self._connection.commit() + + # ─── Threads ──────────────────────────────────────────────── + + async def add_thread(self, thread_id: str, title: str) -> bool: + self._ensure_connected() + try: + await self._connection.execute( + "INSERT INTO threads (id, title) VALUES (?, ?)", + (thread_id, title), + ) + await self._connection.commit() + return True + except aiosqlite.IntegrityError: + return False + except Exception as e: + logger.error(f"Error adding thread {thread_id}: {e}") + raise + + async def update_thread_title(self, thread_id: str, title: str) -> bool: + self._ensure_connected() + try: + cursor = await self._connection.execute( + "UPDATE threads SET title = ? WHERE id = ?", (title, thread_id) + ) + await self._connection.commit() + return cursor.rowcount > 0 + except Exception as e: + logger.error(f"Error updating title for thread {thread_id}: {e}") + raise + + async def get_all_threads(self) -> list[Thread]: + self._ensure_connected() + try: + async with self._connection.execute( + "SELECT id, title, last_bumped FROM threads ORDER BY id" + ) as cursor: + rows = await cursor.fetchall() + return [Thread(id=row[0], title=row[1], last_bumped=row[2]) for row in rows] + except Exception as e: + logger.error(f"Error fetching all threads: {e}") + raise + + async def get_threads_to_bump(self, interval_hours: float) -> list[Thread]: + self._ensure_connected() + if interval_hours < 0: + raise ValueError("interval_hours must be non-negative") + try: + async with self._connection.execute( + """ + SELECT id, title, last_bumped FROM threads + WHERE last_bumped IS NULL + OR datetime(last_bumped, '+' || ? || ' hours') <= datetime('now') + ORDER BY last_bumped ASC NULLS FIRST + """, + (interval_hours,), + ) as cursor: + rows = await cursor.fetchall() + return [Thread(id=row[0], title=row[1], last_bumped=row[2]) for row in rows] + except Exception as e: + logger.error(f"Error fetching threads to bump: {e}") + raise + + async def update_last_bumped(self, thread_id: str) -> None: + self._ensure_connected() + try: + await self._connection.execute( + "UPDATE threads SET last_bumped = datetime('now') WHERE id = ?", + (thread_id,), + ) + await self._connection.commit() + except Exception as e: + logger.error(f"Error updating last_bumped for thread {thread_id}: {e}") + raise + + async def delete_thread(self, thread_id: str) -> bool: + self._ensure_connected() + try: + cursor = await self._connection.execute( + "DELETE FROM threads WHERE id = ?", (thread_id,) + ) + await self._connection.commit() + return cursor.rowcount > 0 + except Exception as e: + logger.error(f"Error deleting thread {thread_id}: {e}") + raise + + async def get_thread_count(self) -> int: + self._ensure_connected() + try: + async with self._connection.execute("SELECT COUNT(*) FROM threads") as cursor: + row = await cursor.fetchone() + return row[0] if row else 0 + except Exception as e: + logger.error(f"Error getting thread count: {e}") + raise + + async def thread_exists(self, thread_id: str) -> bool: + self._ensure_connected() + try: + async with self._connection.execute( + "SELECT 1 FROM threads WHERE id = ? LIMIT 1", (thread_id,) + ) as cursor: + row = await cursor.fetchone() + return row is not None + except Exception as e: + logger.error(f"Error checking thread existence {thread_id}: {e}") + raise diff --git a/requirements.txt b/requirements.txt index b9f82d7..62e0e81 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,2 +1,3 @@ -aiogram==3.15.0 -aiosqlite>=0.20.0 +aiogram==3.15.0 +aiosqlite>=0.20.0 +python-dotenv>=1.0.0