From 134fab488b564308c8c1091bd532fdeefe7bb279 Mon Sep 17 00:00:00 2001 From: imletbruh <60918217+qiyanaitsme@users.noreply.github.com> Date: Sun, 6 Sep 2026 09:42:56 +0500 Subject: [PATCH] Add files via upload --- README.md | 113 +++--- api_client.py | 858 +++++++++++++++++++++++----------------------- app.py | 282 +++++++++++---- config_manager.py | 29 +- database.py | 14 +- 5 files changed, 729 insertions(+), 567 deletions(-) diff --git a/README.md b/README.md index 28d72bf..bd2e67f 100644 --- a/README.md +++ b/README.md @@ -1,43 +1,36 @@ # 🚀 QIYANA AUTO-BUMP BOT для Lolz.live -Telegram бот для автоматического поднятия тем на форуме Lolz.live. Batch API, динамические настройки (меняются на лету без перезапуска), персистентная статистика. +Telegram бот для автоматического поднятия тем на форуме Lolz.live. Batch API, тикающий планировщик, динамические настройки (меняются на лету без перезапуска), персистентная статистика. + +Клиент сверен с официальной спецификацией API (`forum.json` — Lolzteam Public API v1.1.44a). ## ✨ Возможности - ➕ **Добавление тем** — через запятую, названия загружаются сразу через batch API +- 📥 **Мои темы** — выбор своих тем прямо с форума (пагинация, тап — добавить), без ручного ввода ID - 🗑️ **Удаление тем** — через интерактивное меню -- 📋 **Список тем** — с названиями из БД и датой последнего поднятия +- 📋 **Список тем** — с названиями и датой последнего поднятия - 🚀 **Ручное поднятие** — все темы немедленно через batch API - 🔄 **Обновление названий** — синхронизация названий тем с форумом по кнопке -- ⏰ **Автоподнятие** — каждые N часов автоматически -- 🛠️ **Динамические настройки** — интервал, batch size, автобамп меняются через меню без перезапуска +- ⏰ **Автоподнятие** — тикающий планировщик (каждые 60 секунд проверяет, какие темы «созрели»), настройки применяются без перезапуска +- 🛠️ **Динамические настройки** — интервал, batch size, автобамп меняются через меню - 📊 **Статистика** — сохраняется в БД, не теряется при перезапуске - 🔔 **Уведомления** — о результате каждого поднятия - 🛡️ **Защита доступа** — ботом управляет только админ (по Telegram ID) -- ⚡ **Batch API** — до 10 тем за один запрос +- ⚡ **Batch API** — до 10 тем за один запрос, обработка 429 и rate-limit заголовков ## 🎮 Интерфейс ``` ┌──────────────────────────────────────┐ -│ ➕ Add topics │ 📋 List of topics │ -│ 🗑️ Delete topic │ 🚀 Bump topics │ -│ 🔄 Refresh │ 📊 Statistics │ -│ 🛠️ Settings │ 👤 Author │ +│ ➕ Add topics │ 📥 My topics │ +│ 📋 List of topics│ 🗑️ Delete topic │ +│ 🚀 Bump topics │ 🔄 Refresh │ +│ 📊 Statistics │ 🛠️ Settings │ +│ 👤 Author │ └──────────────────────────────────────┘ ``` -### Меню настроек - -``` -┌──────────────────────────────────┐ -│ ⏰ Set Interval │ -│ 📦 Set Batch Size │ -│ 🔄 Toggle Auto-Bump │ -│ ↩️ Back to Menu │ -└──────────────────────────────────┘ -``` - ## 📦 Установка ### Требования @@ -59,8 +52,8 @@ pip install -r requirements.txt 3. Скопируйте токен **Lolz API Token:** -1. Перейдите на [zelenka.guru/account/api](https://zelenka.guru/account/api) -2. Создайте токен с правами `read`, `post` +1. Перейдите на [lolz.live/account/api](https://lolz.live/account/api) +2. Создайте токен с правами (scopes) `read`, `post` 3. Скопируйте токен **Ваш Telegram ID:** @@ -75,7 +68,8 @@ BOT_IMG_URL=https://wallpapers-clan.com/wp-content/uploads/2024/04/dark-anime-gi BOT_AUTHOR_URL=https://lolz.live/kqlol/ # Lolz API -API_BASE_URL=https://prod-api.lolz.live +# Официальные серверы: https://api.lolz.team, https://api.lolz.live, https://api.zelenka.guru +API_BASE_URL=https://api.lolz.live API_AUTH_TOKEN=eyJ0eXAiOiJKV1QiLCJhbGc... API_BATCH_SIZE=10 @@ -83,40 +77,34 @@ API_BATCH_SIZE=10 DB_PATH=threads.db # Scheduling -BUMP_INTERVAL_HOURS=12 -BUMP_DELAY_SECONDS=2 +BUMP_INTERVAL_MINUTES=5 +BUMP_DELAY_SECONDS=1 ENABLE_AUTO_BUMP=true +SCHEDULER_TICK_SECONDS=60 # Admin (ваш Telegram user ID) ADMIN_USER_ID=123456789 ``` +> `BOT_IMG_URL` можно оставить пустым — бот отправит меню текстом. + ### 4. Запустите бота ```bash python app.py ``` +На старте бот проверяет токен через `GET /users/me` и пишет в лог, от какого аккаунта работает. + ## 📖 Использование ### Добавление тем 1. Нажмите **➕ Add topics** 2. Введите ID через запятую: `12345, 67890, 11111` -3. Бот сразу загрузит реальные названия через batch API и сохранит в БД +3. Бот загрузит названия через batch API и сохранит в БД -### Обновление названий - -1. Нажмите **🔄 Refresh** -2. Бот загрузит актуальные названия всех тем через batch API -3. Обновлённые названия сохранятся в БД - -Полезно, если темы переименовали на форуме — не нужно удалять и добавлять заново. - -### Удаление темы - -1. Нажмите **🗑️ Delete topic** -2. Выберите тему из списка +Либо нажмите **📥 My topics** — бот покажет ваши темы с форума (по `GET /threads?tab=mythreads`) с пагинацией; нажмите на тему, чтобы добавить её. ### Ручное поднятие @@ -129,6 +117,10 @@ python app.py [3/3] ❌ Тема 9247922: нужно подождать ``` +### Обновление названий (🔄 Refresh) + +Бот загружает актуальные названия тем с форума. Полезно, если темы переименовали. + ### Статистика Нажмите **📊 Statistics** — покажет: @@ -142,11 +134,9 @@ python app.py ## ⚙️ Настройки (меняются на лету) -Все настройки изменяются через кнопку **🛠️ Settings** без остановки бота: - | Кнопка | Что меняет | Применяется | |--------|-----------|-------------| -| ⏰ Set Interval | Интервал автобампа (часы) | Мгновенно, перезапускает цикл | +| ⏰ Set Interval | Интервал автобампа в минутах (`5`, `720`, …) | На следующем тике планировщика (до 60 с) | | 📦 Set Batch Size | Размер batch (1–10) | Мгновенно для следующих запросов | | 🔄 Toggle Auto-Bump | Включить/выключить автобамп | Мгновенно | @@ -154,25 +144,34 @@ python app.py ## 🔄 Как работает автоподнятие -1. Запускаете бота -2. Вручную поднимаете темы через 🚀 (первый раз) -3. Бот ждёт указанный интервал (например, 12 часов) -4. Автоматически поднимает темы, у которых прошёл интервал -5. Цикл повторяется +Планировщик **тикающий**: каждые `SCHEDULER_TICK_SECONDS` (по умолчанию 60 с) бот читает настройки из БД, находит темы, у которых прошёл интервал, и поднимает их. Плюсы: + +- темы, добавленные позже, подхватываются в ближайший тик; +- смена интервала применяется без перезапуска и без «потери фазы»; +- бот бампает ровно по таймеру, без обращений к форуму перед бампом. ``` -00:00 — Запуск бота +00:00 — Запуск бота (тик каждые 60с) 00:05 — Ручное поднятие -12:05 — Автоматическое поднятие -24:05 — Автоматическое поднятие +00:10 — Первый тик, где тема «созрела» → автоподнятие ``` +## 🌐 Лимиты API (по спецификации) + +| Эндпоинт | Лимит | +|----------|-------| +| GET-запросы | 300 / минуту | +| Не-GET (в т.ч. bump) | 30 / минуту | +| `/batch` | 20 / минуту (независимый бакет) | + +При превышении API возвращает `429` и заголовки `X-RateLimit-*`. Бот читает их и ждёт сброса окна, а между batch-вызовами соблюдает паузу `BUMP_DELAY_SECONDS`. + ## 🏗️ Архитектура ``` -app.py # Бот: хендлеры, middleware аутентификации, авто-бамп +app.py # Бот: хендлеры, middleware аутентификации, тикающий авто-бамп config_manager.py # Загрузка и валидация .env -api_client.py # Lolz batch API клиент с retry логикой +api_client.py # Lolz batch API клиент: union-парсер ответов, rate-limit, retry database.py # SQLite: темы, настройки, статистика ``` @@ -186,12 +185,12 @@ database.py # SQLite: темы, настройки, статистика ### Ключевые особенности -- **Batch API** — до 10 тем за запрос (экономия лимитов в 10 раз) -- **Динамические настройки** — меняются в БД, подхватываются ботом мгновенно +- **Batch API** — до 10 тем за запрос; job'ы ключуются явным `id` (по спецификации) +- **Union-парсер bump-ответов** — понимает и документированную форму `{status, message, system_info}`, и «пустой ответ = успех», и legacy `_job_result` +- **Тикающий планировщик** — настройки применяются без перезапуска +- **Ретраи только там, где есть смысл** — сеть/5xx/429; 4xx не ретраятся - **Персистентная статистика** — не теряется при перезапуске - **Аутентификация** — middleware проверяет Telegram ID админа -- **Async context managers** — автоматическая очистка ресурсов -- **Frozen dataclasses с `__slots__`** — иммутабельность и экономия памяти - **WAL-режим SQLite** — лучшая конкурентность ## 📝 Логи @@ -219,7 +218,7 @@ database.py # SQLite: темы, настройки, статистика ### Ошибка: «API_AUTH_TOKEN not configured» -Проверьте токен на [zelenka.guru/account/api](https://zelenka.guru/account/api). +Проверьте токен на [lolz.live/account/api](https://lolz.live/account/api). В логе при старте будет `API token validation failed`, если токен невалиден. ### Ошибка: «⛔ Доступ запрещён» @@ -229,7 +228,7 @@ database.py # SQLite: темы, настройки, статистика 1. Проверьте, что автобамп включён (🔄 Toggle Auto-Bump) 2. Сделайте первое поднятие вручную через 🚀 -3. Проверьте логи в `bot.log` +3. Проверьте логи в `bot.log` — ошибки от форума (например, «нужно подождать») видны по каждой теме ## 👤 Автор diff --git a/api_client.py b/api_client.py index 9dd79bc..893bfca 100644 --- a/api_client.py +++ b/api_client.py @@ -1,17 +1,28 @@ -"""Lolz API client with batch request support and proper error handling.""" +"""Lolz API client (spec-compliant: forum.json / Lolzteam Public API v1.1.44a). + +Covers: POST /batch (id-keyed jobs), GET /threads/{id}, POST /threads/{id}/bump, +GET /threads?tab=mythreads, GET /users/me. +Rate limits per spec: GET 300/min, non-GET 30/min, /batch 20/min (429 + X-RateLimit-*). +""" -import re -import logging -import aiohttp import asyncio -from typing import Self, Sequence +import logging +import re +import time from dataclasses import dataclass from enum import Enum +from typing import Self, Sequence + +import aiohttp # Constants MAX_RETRY_ATTEMPTS = 3 RETRY_DELAY_SECONDS = 2 +MAX_RATE_LIMIT_WAIT_SECONDS = 180 +DEFAULT_429_WAIT_SECONDS = RETRY_DELAY_SECONDS * 5 +# "status" values that mean success in the documented bump response {status, message} +SUCCESSFUL_STATUSES = {"ok", "success", "true"} logger = logging.getLogger(__name__) @@ -25,6 +36,18 @@ class BumpStatus(Enum): ERROR = "error" +class LolzAPIError(Exception): + """Base error for Lolz API failures.""" + + +class InvalidTokenError(LolzAPIError): + """API token is invalid or missing scopes.""" + + +class BatchRequestError(LolzAPIError): + """Batch/API request failed in a way that won't be fixed by retrying.""" + + @dataclass(frozen=True, slots=True) class BumpResult: """Result of bump operation.""" @@ -41,477 +64,470 @@ class ThreadInfo: title: str +@dataclass(frozen=True, slots=True) +class ThreadsPage: + """Page of GET /threads listing.""" + threads: list[ThreadInfo] + total: int + + +def extract_error_message(error_msg: str) -> str: + """Extract and clean error message from API response.""" + if not error_msg: + return "" + + # Remove HTML tags + error_msg = re.sub(r"", "\n", error_msg) + error_msg = re.sub(r"<[^>]+>", "", error_msg) + + # Split by newlines and filter empty parts + parts = [p.strip() for p in error_msg.split("\n") if p.strip()] + + if not parts: + return "" + + # Look for rate limit message first + for part in parts: + if "должны подождать" in part.lower() or "должен подождать" in part.lower(): + return part + + # Return last meaningful part + return parts[-1] + + +def _errors_to_text(errors: object) -> str: + """Normalize an API 'errors' payload (list/dict/str/other) into plain text.""" + if isinstance(errors, list): + return "; ".join(str(e) for e in errors) + if isinstance(errors, dict): + return "; ".join(str(v) for v in errors.values()) + return str(errors) + + +def _error_status(message: str) -> BumpStatus: + if "подождать" in message.lower(): + return BumpStatus.RATE_LIMITED + return BumpStatus.ERROR + + +def parse_bump_job_result(thread_id: str, job_data: object) -> BumpResult: + """Parse a single job result of a batch bump request. + + Handles every documented/observed shape: + - ``[]`` / ``{}`` — empty response means success (observed on live API) + - ``{"_job_result": "error", "_job_message": ...}`` — legacy error wrapper + - ``{"errors": [...]}`` — error response + - ``{"status": ..., "message": ..., "system_info": ...}`` — documented + POST /threads/{id}/bump 200 response shape + """ + if job_data is None: + return BumpResult(False, f"Тема {thread_id}: Нет ответа от сервера", thread_id, BumpStatus.ERROR) + + if isinstance(job_data, (list, dict)) and len(job_data) == 0: + return BumpResult(True, f"✅ Тема {thread_id} поднята успешно", thread_id, BumpStatus.SUCCESS) + + if not isinstance(job_data, dict): + logger.warning(f"Thread {thread_id} unknown bump response (type={type(job_data).__name__}): {job_data}") + return BumpResult( + False, + f"Тема {thread_id}: Неизвестный ответ ({type(job_data).__name__})", + thread_id, + BumpStatus.ERROR, + ) + + # Legacy wrapper observed on live API + if "_job_result" in job_data: + job_result = str(job_data.get("_job_result", "")).lower() + if job_result == "error": + error_text = str(job_data.get("_job_message", "") or "") + if not error_text.strip(): + errors = job_data.get("errors") + if errors: + error_text = _errors_to_text(errors) + if not error_text.strip(): + error_text = str(job_data.get("error", "")) + error_msg = extract_error_message(error_text) or "Ошибка API без текста (см. raw response в логах)" + logger.error( + f"Thread {thread_id} bump failed (legacy wrapper) | " + f"Raw: {job_data} | Extracted: {error_msg}" + ) + return BumpResult(False, f"Тема {thread_id}: {error_msg}", thread_id, _error_status(error_msg)) + logger.info(f"Thread {thread_id} bumped successfully (job_result={job_result})") + return BumpResult(True, f"✅ Тема {thread_id} поднята успешно", thread_id, BumpStatus.SUCCESS) + + if job_data.get("errors"): + error_msg = extract_error_message(_errors_to_text(job_data["errors"])) or "Ошибка API" + logger.error(f"Thread {thread_id} bump failed | Errors: {job_data['errors']} | Extracted: {error_msg}") + return BumpResult(False, f"Тема {thread_id}: {error_msg}", thread_id, _error_status(error_msg)) + + # Documented endpoint response: {"status": ..., "message": ..., "system_info": ...} + status_value = job_data.get("status") + if status_value is not None: + if str(status_value).strip().lower() in SUCCESSFUL_STATUSES: + logger.info(f"Thread {thread_id} bumped successfully (status={status_value})") + return BumpResult(True, f"✅ Тема {thread_id} поднята успешно", thread_id, BumpStatus.SUCCESS) + error_msg = extract_error_message(str(job_data.get("message", ""))) or f"Ошибка API (status={status_value})" + logger.error(f"Thread {thread_id} bump failed | status={status_value} | message={error_msg}") + return BumpResult(False, f"Тема {thread_id}: {error_msg}", thread_id, _error_status(error_msg)) + + logger.warning(f"Thread {thread_id} unknown bump response (dict without recognized keys): {job_data}") + return BumpResult(False, f"Тема {thread_id}: Неизвестный ответ (dict)", thread_id, BumpStatus.ERROR) + + class APIClient: """Lolz API client with batch request support and connection pooling.""" - - __slots__ = ("_base_url", "_auth_token", "_session", "_batch_size") - - def __init__(self, base_url: str, auth_token: str, batch_size: int = 10) -> None: + + __slots__ = ("_base_url", "_auth_token", "_session", "_batch_size", "_batch_delay_seconds") + + def __init__( + self, + base_url: str, + auth_token: str, + batch_size: int = 10, + batch_delay_seconds: float = 1.0, + ) -> None: if not base_url or not auth_token: raise ValueError("base_url and auth_token are required") - + if batch_size < 1 or batch_size > 10: raise ValueError("batch_size must be between 1 and 10") - + self._base_url = base_url.rstrip("/") self._auth_token = auth_token self._session: aiohttp.ClientSession | None = None self._batch_size = batch_size - + self._batch_delay_seconds = max(0.0, float(batch_delay_seconds)) + async def __aenter__(self) -> Self: - """Async context manager entry.""" await self.start() return self - + async def __aexit__(self, exc_type, exc_val, exc_tb) -> None: - """Async context manager exit.""" await self.close() - + async def start(self) -> None: """Initialize HTTP session with connection pooling.""" if self._session is None or self._session.closed: headers = { "Accept": "application/json", "Authorization": f"Bearer {self._auth_token}", - "User-Agent": "AutoBumpBot/4.0", - "Content-Type": "application/json" + "User-Agent": "AutoBumpBot/5.0", + "Content-Type": "application/json", } timeout = aiohttp.ClientTimeout(total=30, connect=10) connector = aiohttp.TCPConnector(limit=10, limit_per_host=5) self._session = aiohttp.ClientSession( headers=headers, timeout=timeout, - connector=connector + connector=connector, ) - + async def close(self) -> None: """Close HTTP session and cleanup resources.""" if self._session and not self._session.closed: await self._session.close() - # Wait for connections to close properly await asyncio.sleep(0.25) self._session = None - + + def set_batch_size(self, batch_size: int) -> None: + if batch_size < 1 or batch_size > 10: + raise ValueError("batch_size must be between 1 and 10") + self._batch_size = batch_size + + # ─── Low-level batch executor ────────────────────────────── + @staticmethod - def _extract_error_message(error_msg: str) -> str: - """Extract and clean error message from API response.""" - if not error_msg: - return "Unknown error" - - # Remove HTML tags - error_msg = re.sub(r"", "\n", error_msg) - error_msg = re.sub(r"<[^>]+>", "", error_msg) - - # Split by newlines and filter empty parts - parts = [p.strip() for p in error_msg.split("\n") if p.strip()] - - if not parts: - return "Unknown error" - - # Look for rate limit message first - for part in parts: - if "должны подождать" in part.lower() or "должен подождать" in part.lower(): - return part - - # Return last meaningful part - return parts[-1] - - async def get_thread_info(self, thread_id: str) -> ThreadInfo | None: - """Get single thread information from API (legacy method, prefer get_threads_info_batch).""" - results = await self.get_threads_info_batch([thread_id]) - return results[0] if results else None - - async def get_threads_info_batch(self, thread_ids: Sequence[str]) -> list[ThreadInfo | None]: - """Get multiple thread information using batch API (up to 10 per request).""" + async def _safe_json(resp: aiohttp.ClientResponse) -> object: + try: + return await resp.json(content_type=None) + except Exception: + return None + + @staticmethod + def _rate_limit_wait_seconds(headers: object, body: object) -> float | None: + """Estimate wait time from X-RateLimit-Reset / Retry-After / system_info.rate_limit.""" + now = time.time() + candidates: list[float] = [] + + if headers is not None: + reset = headers.get("X-RateLimit-Reset") + if reset: + try: + candidates.append(float(reset) - now) + except (TypeError, ValueError): + pass + retry_after = headers.get("Retry-After") + if retry_after: + try: + candidates.append(float(retry_after)) + except (TypeError, ValueError): + pass + + if isinstance(body, dict): + system_info = body.get("system_info") + rate_limit = system_info.get("rate_limit") if isinstance(system_info, dict) else None + reset = rate_limit.get("reset") if isinstance(rate_limit, dict) else None + if isinstance(reset, (int, float)) and reset > 0: + candidates.append(float(reset) - now) + + positive = [c for c in candidates if c > 0] + return min(positive) if positive else None + + async def _execute_batch(self, batch_payload: Sequence[dict]) -> dict: + """POST /batch with retry and rate-limit handling. Returns the 'jobs' mapping.""" if not self._session: await self.start() - - if not thread_ids: - return [] - - # Process in batches of up to batch_size - all_results: list[ThreadInfo | None] = [] - - for i in range(0, len(thread_ids), self._batch_size): - batch = thread_ids[i:i + self._batch_size] - batch_results = await self._fetch_threads_info_batch(batch) - all_results.extend(batch_results) - - return all_results - - async def _fetch_threads_info_batch(self, thread_ids: Sequence[str]) -> list[ThreadInfo | None]: - """Execute single batch request to get info for up to 10 threads with retry logic.""" - if not thread_ids: - return [] - - # Build batch request payload - must use full URLs for batch API - batch_payload = [ - { - "method": "GET", - "uri": f"{self._base_url}/threads/{thread_id}" - } - for thread_id in thread_ids - ] - batch_url = f"{self._base_url}/batch" - + last_error = "unknown error" + for attempt in range(MAX_RETRY_ATTEMPTS): try: - async with self._session.post(batch_url, json=batch_payload) as resp: - logger.info(f"Batch API request for {len(thread_ids)} threads, status: {resp.status}") - + async with self._session.post(batch_url, json=list(batch_payload)) as resp: + logger.info( + f"Batch API request ({len(batch_payload)} jobs), " + f"status: {resp.status}, attempt {attempt + 1}" + ) + + if resp.status == 200: + try: + body = await resp.json() + except Exception: + raise BatchRequestError("Неверный JSON в ответе batch API") + if isinstance(body, dict) and isinstance(body.get("jobs"), dict): + return body["jobs"] + raise BatchRequestError("Неверный формат ответа batch API (нет 'jobs')") + + body = await self._safe_json(resp) + if resp.status == 401: - raise ValueError("Invalid API token - check your configuration") - - if resp.status != 200: - response_text = await resp.text() - logger.error(f"Batch API failed with status {resp.status}: {response_text}") - if attempt < MAX_RETRY_ATTEMPTS - 1: - await asyncio.sleep(RETRY_DELAY_SECONDS * (attempt + 1)) - continue - # Final attempt failed - return [None] * len(thread_ids) - - # Parse batch response - batch_response = await resp.json() - logger.debug(f"Batch response: {batch_response}") - - # Validate response structure - API returns {"jobs": {...}, "system_info": {...}} - if not isinstance(batch_response, dict) or "jobs" not in batch_response: - logger.error(f"Batch response missing 'jobs' key: {batch_response}") - return [None] * len(thread_ids) - - jobs = batch_response["jobs"] - if not isinstance(jobs, dict): - logger.error(f"Jobs is not a dict: {type(jobs)}") - return [None] * len(thread_ids) - - # Process each thread - results: list[ThreadInfo | None] = [] - for thread_id in thread_ids: - uri = f"{self._base_url}/threads/{thread_id}" - - if uri not in jobs: - logger.warning(f"Thread {thread_id} not in jobs response") - results.append(None) - continue - - job_data = jobs[uri] - if not isinstance(job_data, dict): - logger.warning(f"Job data for {thread_id} is not a dict") - results.append(None) - continue - - # Extract thread info - thread_data = job_data.get("thread", {}) - if not isinstance(thread_data, dict): - logger.warning(f"Thread data for {thread_id} is not a dict") - results.append(None) - continue - - thread_info = ThreadInfo( - thread_id=thread_id, - title=thread_data.get("thread_title", "Unknown") - ) - results.append(thread_info) - - return results - - except aiohttp.ClientError as e: - if attempt < MAX_RETRY_ATTEMPTS - 1: - await asyncio.sleep(RETRY_DELAY_SECONDS * (attempt + 1)) - continue - raise ConnectionError(f"Network error after {MAX_RETRY_ATTEMPTS} attempts: {e}") from e - except ValueError: + raise InvalidTokenError("Invalid API token - check your configuration") + + if resp.status == 429: + wait = self._rate_limit_wait_seconds(resp.headers, body) + wait = wait if wait is not None else DEFAULT_429_WAIT_SECONDS + wait = min(wait, MAX_RATE_LIMIT_WAIT_SECONDS) + logger.warning(f"Batch rate-limited (429), waiting {wait:.0f}s (attempt {attempt + 1})") + await asyncio.sleep(wait) + continue + + if resp.status == 400: + errors = body.get("errors") if isinstance(body, dict) else None + message = extract_error_message(_errors_to_text(errors)) if errors else "" + raise BatchRequestError(message or "Batch request rejected (HTTP 400)") + + if 400 <= resp.status < 500: + raise BatchRequestError(f"HTTP {resp.status}: {str(body)[:200]}") + + # 5xx and others are retryable + last_error = f"HTTP {resp.status}" + + except (InvalidTokenError, BatchRequestError): raise - except Exception as e: - if attempt < MAX_RETRY_ATTEMPTS - 1: - await asyncio.sleep(RETRY_DELAY_SECONDS * (attempt + 1)) - continue - raise RuntimeError(f"Unexpected error getting thread info: {e}") from e - - return [None] * len(thread_ids) - + except aiohttp.ClientError as e: + last_error = f"network error: {e}" + + if attempt < MAX_RETRY_ATTEMPTS - 1: + backoff = RETRY_DELAY_SECONDS * (attempt + 1) + logger.warning(f"Batch attempt {attempt + 1} failed ({last_error}), retrying in {backoff}s") + await asyncio.sleep(backoff) + + raise BatchRequestError(f"Batch failed after {MAX_RETRY_ATTEMPTS} attempts: {last_error}") + + async def _get_json(self, path: str, params: dict[str, str] | None = None) -> object: + """GET request with retry and rate-limit handling. Returns parsed JSON body.""" + if not self._session: + await self.start() + url = f"{self._base_url}{path}" + last_error = "unknown error" + + for attempt in range(MAX_RETRY_ATTEMPTS): + try: + async with self._session.get(url, params=params) as resp: + logger.info(f"GET {path} -> HTTP {resp.status}, attempt {attempt + 1}") + + if resp.status == 200: + return await resp.json(content_type=None) + + body = await self._safe_json(resp) + + if resp.status == 401: + raise InvalidTokenError("Invalid API token - check your configuration") + + if resp.status == 429: + wait = self._rate_limit_wait_seconds(resp.headers, body) + wait = wait if wait is not None else DEFAULT_429_WAIT_SECONDS + wait = min(wait, MAX_RATE_LIMIT_WAIT_SECONDS) + logger.warning(f"GET {path} rate-limited (429), waiting {wait:.0f}s") + await asyncio.sleep(wait) + continue + + if 400 <= resp.status < 500: + raise BatchRequestError(f"GET {path} failed: HTTP {resp.status}") + + last_error = f"HTTP {resp.status}" + + except (InvalidTokenError, BatchRequestError): + raise + except aiohttp.ClientError as e: + last_error = f"network error: {e}" + + if attempt < MAX_RETRY_ATTEMPTS - 1: + backoff = RETRY_DELAY_SECONDS * (attempt + 1) + logger.warning(f"GET {path} attempt {attempt + 1} failed ({last_error}), retrying in {backoff}s") + await asyncio.sleep(backoff) + + raise BatchRequestError(f"GET {path} failed after {MAX_RETRY_ATTEMPTS} attempts: {last_error}") + + @staticmethod + def _job_for(thread_id: str, jobs: dict, uri: str) -> object: + """Look up a job result by explicit 'id' first, then fall back to URI key.""" + return jobs[thread_id] if thread_id in jobs else jobs.get(uri) + + # ─── Thread info (titles) ────────────────────────────────── + + async def get_thread_info(self, thread_id: str) -> ThreadInfo | None: + """Get single thread information from API (legacy method, prefer batch methods).""" + results = await self.get_threads_info_batch([thread_id]) + return results[0] if results else None + + async def get_threads_info_batch(self, thread_ids: Sequence[str]) -> list[ThreadInfo | None]: + """Get multiple thread titles using batch API (up to batch_size per request).""" + if not thread_ids: + return [] + + all_results: list[ThreadInfo | None] = [] + for i in range(0, len(thread_ids), self._batch_size): + batch = thread_ids[i:i + self._batch_size] + if all_results and self._batch_delay_seconds > 0: + await asyncio.sleep(self._batch_delay_seconds) + all_results.extend(await self._fetch_threads_info_batch(batch)) + return all_results + + async def _fetch_threads_info_batch(self, thread_ids: Sequence[str]) -> list[ThreadInfo | None]: + if not thread_ids: + return [] + + payload = [ + {"id": str(thread_id), "method": "GET", "uri": f"{self._base_url}/threads/{thread_id}"} + for thread_id in thread_ids + ] + try: + jobs = await self._execute_batch(payload) + except InvalidTokenError: + raise + except BatchRequestError as e: + logger.error(f"Thread info batch failed: {e}") + return [None] * len(thread_ids) + + results: list[ThreadInfo | None] = [] + for thread_id in thread_ids: + uri = f"{self._base_url}/threads/{thread_id}" + job = self._job_for(str(thread_id), jobs, uri) + thread = job.get("thread") if isinstance(job, dict) else None + if isinstance(thread, dict): + results.append(ThreadInfo(thread_id=str(thread_id), title=str(thread.get("thread_title", "Unknown")))) + else: + logger.warning(f"Thread {thread_id}: no thread data in jobs response") + results.append(None) + return results + + # ─── My threads listing ──────────────────────────────────── + + async def get_my_threads(self, page: int = 1, limit: int = 10) -> ThreadsPage: + """List the token owner's threads: GET /threads?tab=mythreads.""" + body = await self._get_json( + "/threads", + params={"tab": "mythreads", "page": str(page), "limit": str(limit)}, + ) + if not isinstance(body, dict): + raise BatchRequestError("Неверный формат ответа /threads") + + items: list[ThreadInfo] = [] + for thread in body.get("threads") or []: + if not isinstance(thread, dict): + continue + items.append(ThreadInfo( + thread_id=str(thread.get("thread_id", "")), + title=str(thread.get("thread_title") or ""), + )) + + try: + total = int(body.get("threads_total") or len(items)) + except (TypeError, ValueError): + total = len(items) + return ThreadsPage(threads=items, total=total) + + # ─── Token validation ────────────────────────────────────── + + async def get_me(self) -> dict | None: + """Validate the token via GET /users/me. Returns the user payload or None.""" + try: + body = await self._get_json("/users/me") + except (InvalidTokenError, BatchRequestError, ConnectionError) as e: + logger.warning(f"get_me failed: {e}") + return None + if isinstance(body, dict): + user = body.get("user") + if isinstance(user, dict): + return user + return body + return None + + # ─── Bump ────────────────────────────────────────────────── + async def bump_thread(self, thread_id: str) -> BumpResult: """Bump single thread via API (legacy method, prefer bump_threads_batch).""" results = await self.bump_threads_batch([thread_id]) return results[0] - + async def bump_threads_batch(self, thread_ids: Sequence[str]) -> list[BumpResult]: - """Bump multiple threads using batch API (up to 10 per request).""" - if not self._session: - await self.start() - + """Bump multiple threads using batch API (up to batch_size per request).""" if not thread_ids: return [] - - # Process in batches of up to batch_size + all_results: list[BumpResult] = [] - for i in range(0, len(thread_ids), self._batch_size): batch = thread_ids[i:i + self._batch_size] - batch_results = await self._execute_bump_batch(batch) - all_results.extend(batch_results) - + if all_results and self._batch_delay_seconds > 0: + await asyncio.sleep(self._batch_delay_seconds) + all_results.extend(await self._execute_bump_batch(batch)) return all_results - + async def _execute_bump_batch(self, thread_ids: Sequence[str]) -> list[BumpResult]: - """Execute single batch bump request for up to 10 threads with retry logic.""" + """Execute a single batch bump request with retry and rate-limit handling.""" if not thread_ids: return [] - - # Build batch request payload - must use full URLs for batch API - batch_payload = [ - { - "method": "POST", - "uri": f"{self._base_url}/threads/{thread_id}/bump" - } + + payload = [ + {"id": str(thread_id), "method": "POST", "uri": f"{self._base_url}/threads/{thread_id}/bump"} for thread_id in thread_ids ] - - batch_url = f"{self._base_url}/batch" - - logger.info(f"🚀 Executing bump batch request for {len(thread_ids)} threads: {thread_ids}") - logger.debug(f"Bump batch payload: {batch_payload}") - - for attempt in range(MAX_RETRY_ATTEMPTS): - try: - async with self._session.post(batch_url, json=batch_payload) as resp: - logger.info(f"Bump batch API response: HTTP {resp.status}") - - if resp.status == 401: - # Unauthorized - return error for all threads - return [ - BumpResult( - success=False, - message=f"Тема {tid}: Неверный токен API", - thread_id=tid, - status=BumpStatus.UNAUTHORIZED - ) - for tid in thread_ids - ] - - if resp.status != 200: - if attempt < MAX_RETRY_ATTEMPTS - 1: - await asyncio.sleep(RETRY_DELAY_SECONDS * (attempt + 1)) - continue - # Final attempt failed - return [ - BumpResult( - success=False, - message=f"Тема {tid}: Ошибка batch запроса (HTTP {resp.status})", - thread_id=tid, - status=BumpStatus.ERROR - ) - for tid in thread_ids - ] - - # Parse batch response - batch_response = await resp.json() - logger.debug(f"Bump batch response: {batch_response}") - - # Validate response structure - API returns {"jobs": {...}, "system_info": {...}} - if not isinstance(batch_response, dict) or "jobs" not in batch_response: - logger.error(f"Bump batch response missing 'jobs' key") - return [ - BumpResult( - success=False, - message=f"Тема {tid}: Неверный формат ответа API", - thread_id=tid, - status=BumpStatus.ERROR - ) - for tid in thread_ids - ] - - jobs = batch_response["jobs"] - if not isinstance(jobs, dict): - logger.error(f"Bump jobs is not a dict") - return [ - BumpResult( - success=False, - message=f"Тема {tid}: Неверный формат ответа API", - thread_id=tid, - status=BumpStatus.ERROR - ) - for tid in thread_ids - ] - - # Process each thread - results: list[BumpResult] = [] - for thread_id in thread_ids: - uri = f"{self._base_url}/threads/{thread_id}/bump" - - if uri not in jobs: - logger.error(f"Bump for thread {thread_id} not in jobs response | URI: {uri}") - results.append(BumpResult( - success=False, - message=f"Тема {thread_id}: Нет ответа от сервера", - thread_id=thread_id, - status=BumpStatus.ERROR - )) - continue - - job_data = jobs[uri] - logger.info(f"Thread {thread_id} raw bump response: {job_data}") - - # 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, - message=f"✅ Тема {thread_id} поднята успешно", - thread_id=thread_id, - status=BumpStatus.SUCCESS - )) - elif isinstance(job_data, dict) and "errors" in job_data: - error_msg = self._extract_error_message(str(job_data["errors"])) - logger.error( - f"Thread {thread_id} bump failed | " - f"Errors: {job_data['errors']} | " - f"Extracted: {error_msg}" - ) - results.append(BumpResult( - success=False, - message=f"Тема {thread_id}: {error_msg}", - 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: - 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}: Неизвестный ответ ({type(job_data).__name__})", - thread_id=thread_id, - status=BumpStatus.ERROR - )) - - logger.info(f"Bump batch processed: {len(results)} results") - return results - - except aiohttp.ClientError as e: - if attempt < MAX_RETRY_ATTEMPTS - 1: - await asyncio.sleep(RETRY_DELAY_SECONDS * (attempt + 1)) - continue - # Network error - return error for all threads - return [ - BumpResult( - success=False, - message=f"Тема {tid}: Ошибка сети - {str(e)}", - thread_id=tid, - status=BumpStatus.ERROR - ) - for tid in thread_ids - ] - except Exception as e: - if attempt < MAX_RETRY_ATTEMPTS - 1: - await asyncio.sleep(RETRY_DELAY_SECONDS * (attempt + 1)) - continue - # Unexpected error - return error for all threads - return [ - BumpResult( - success=False, - message=f"Тема {tid}: Неожиданная ошибка - {str(e)}", - thread_id=tid, - status=BumpStatus.ERROR - ) - for tid in thread_ids - ] - - # Should never reach here, but just in case - return [ - BumpResult( - success=False, - message=f"Тема {tid}: Превышено количество попыток", - thread_id=tid, - status=BumpStatus.ERROR - ) - for tid in thread_ids - ] - - def _parse_bump_response(self, thread_id: str, response_data: dict) -> BumpResult: - """Parse individual bump response from batch result.""" - # Check for errors in response - if "errors" in response_data and response_data["errors"]: - errors = response_data["errors"] - if isinstance(errors, list) and errors: - error_msg = str(errors[0]) - cleaned_msg = self._extract_error_message(error_msg) - - # Determine status - status = BumpStatus.ERROR - if "подождать" in cleaned_msg.lower(): - status = BumpStatus.RATE_LIMITED - - return BumpResult( - success=False, - message=f"Тема {thread_id}: {cleaned_msg}", - thread_id=thread_id, - status=status - ) - - # Check HTTP status code in batch response - status_code = response_data.get("_status_code", 200) - - if status_code == 200: - return BumpResult( - success=True, - message=f"✅ Тема {thread_id} поднята успешно", - thread_id=thread_id, - status=BumpStatus.SUCCESS - ) - elif status_code == 404: - return BumpResult( - success=False, - message=f"Тема {thread_id}: Не найдена", - thread_id=thread_id, - status=BumpStatus.NOT_FOUND - ) - elif status_code == 401: - return BumpResult( - success=False, - message=f"Тема {thread_id}: Неверный токен API", - thread_id=thread_id, - status=BumpStatus.UNAUTHORIZED - ) - - return BumpResult( - success=False, - message=f"Тема {thread_id}: HTTP {status_code}", - thread_id=thread_id, - status=BumpStatus.ERROR - ) + logger.info(f"🚀 Executing bump batch request for {len(thread_ids)} threads: {list(thread_ids)}") + + try: + jobs = await self._execute_batch(payload) + except InvalidTokenError: + return [ + BumpResult(False, f"Тема {tid}: Неверный токен API", tid, BumpStatus.UNAUTHORIZED) + for tid in thread_ids + ] + except BatchRequestError as e: + return [BumpResult(False, f"Тема {tid}: {e}", tid, BumpStatus.ERROR) for tid in thread_ids] + except Exception as e: + logger.error(f"Unexpected bump batch error: {e}", exc_info=True) + return [ + BumpResult(False, f"Тема {tid}: Неожиданная ошибка - {e}", tid, BumpStatus.ERROR) + for tid in thread_ids + ] + + results: list[BumpResult] = [] + for thread_id in thread_ids: + uri = f"{self._base_url}/threads/{thread_id}/bump" + job = self._job_for(str(thread_id), jobs, uri) + logger.info(f"Thread {thread_id} raw bump response: {job}") + result = parse_bump_job_result(str(thread_id), job) + log = logger.info if result.success else logger.error + log(f"BUMP {'SUCCESS' if result.success else 'FAILED'} | Thread: {thread_id} | " + f"Status: {result.status.value} | Message: {result.message}") + results.append(result) + return results diff --git a/app.py b/app.py index c8ce07e..dac2764 100644 --- a/app.py +++ b/app.py @@ -1,4 +1,5 @@ -"""Telegram bot for automatic thread bumping on Lolz.live — with dynamic settings & auth.""" +"""Telegram bot for automatic thread bumping on Lolz.live — tick-based scheduler, +batch API, forum-backed thread listing.""" import asyncio import logging @@ -24,6 +25,7 @@ from database import Database, BumpStats, Thread NOTIFICATION_DELAY_SECONDS = 0.8 BUTTON_TEXT_MAX_LENGTH = 30 +MY_THREADS_PAGE_SIZE = 10 AUTO_BUMP_RETRY_DELAY_SECONDS = 60 @@ -38,6 +40,17 @@ logging.basicConfig( logger = logging.getLogger(__name__) +def _format_interval(minutes: float) -> str: + """Human-friendly interval: minutes below an hour, trimmed decimals above.""" + if minutes < 60: + if minutes == int(minutes): + return f"{int(minutes)} мин" + return f"{minutes:.1f} мин" + hours = minutes / 60 + text = f"{hours:.2f}".rstrip("0").rstrip(".") + return f"{text}ч" + + class BotStates(StatesGroup): waiting_for_thread_ids = State() waiting_for_interval = State() @@ -110,6 +123,7 @@ class AutoBumpBot: config.api.base_url, config.api.auth_token, config.api.batch_size, + config.scheduling.bump_delay_seconds, ) self._db = Database(config.database.path) @@ -147,6 +161,10 @@ class AutoBumpBot: 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_my_threads_callback, F.data == "my_threads") + self._router.callback_query.register(self._handle_my_threads_page_callback, F.data.startswith("mypage_")) + self._router.callback_query.register(self._handle_my_add_callback, F.data.startswith("myadd_")) + self._router.callback_query.register(self._handle_noop_callback, F.data == "noop") 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") @@ -159,18 +177,21 @@ class AutoBumpBot: return InlineKeyboardMarkup(inline_keyboard=[ [ InlineKeyboardButton(text="➕ Add topics", callback_data="add_thread"), + InlineKeyboardButton(text="📥 My topics", callback_data="my_threads"), + ], + [ 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="📊 Statistics", callback_data="stats"), InlineKeyboardButton(text="🛠️ Settings", callback_data="settings"), + ], + [ InlineKeyboardButton(text="👤 Author", url=self._config.bot.author_url), ], ]) @@ -184,12 +205,15 @@ class AutoBumpBot: ]) 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(), - ) + markup = self._main_menu_kb() + img_url = self._config.bot.img_url + if img_url: + try: + await self._bot.send_photo(chat_id, photo=img_url, caption=text, reply_markup=markup) + return + except TelegramBadRequest as e: + logger.warning(f"send_photo failed, falling back to text message: {e}") + await self._bot.send_message(chat_id, text, reply_markup=markup) # ─── Utilities ────────────────────────────────────────────── @@ -218,8 +242,8 @@ class AutoBumpBot: else: return f"{minutes}m" - async def _get_interval(self) -> float: - val = await self._db.get_setting("bump_interval_hours", "12") + async def _get_interval_minutes(self) -> float: + val = await self._db.get_setting("bump_interval_minutes", "60") return float(val) async def _get_batch_size(self) -> int: @@ -234,15 +258,14 @@ class AutoBumpBot: 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=( + interval = await self._get_interval_minutes() + await self._send_main_menu( + message.chat.id, + text=( "🤖 QIYANA AUTO-BUMP BOT\n\n" "Бот для автоматического поднятия тем на Lolz.live\n\n" - f"⏰ Автоподнятие каждые {interval:.0f}ч" + f"⏰ Автоподнятие каждые {_format_interval(interval)}" ), - reply_markup=self._main_menu_kb(), ) except TelegramBadRequest as e: logger.error(f"Failed to send start message: {e}") @@ -256,7 +279,8 @@ class AutoBumpBot: return await callback.message.answer( "📝 Введите ID тем через запятую для добавления:\n" - "Пример: 12345, 67890, 11111" + "Пример: 12345, 67890, 11111\n\n" + "Или добавьте свои темы с форума через кнопку 📥 My topics" ) await state.set_state(BotStates.waiting_for_thread_ids) @@ -289,9 +313,9 @@ class AutoBumpBot: 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}" + for tid, info in zip(valid_ids, threads_info): + if info and info.title: + titles_map[tid] = info.title except Exception as e: logger.error(f"Error fetching titles during add: {e}") for tid in valid_ids: @@ -333,7 +357,7 @@ class AutoBumpBot: try: threads = await self._db.get_all_threads() if not threads: - await callback.message.answer("📭 Список тем пуст\n\nДобавьте темы через кнопку ➕") + await callback.message.answer("📭 Список тем пуст\n\nДобавьте темы через кнопку ➕ или 📥") await self._send_main_menu(callback.message.chat.id) return @@ -361,7 +385,7 @@ class AutoBumpBot: logger.error(f"Error listing threads: {e}", exc_info=True) await callback.message.answer(f"❌ Ошибка: {str(e)}") - # ─── Refresh Titles ───────────────────────────────────────── + # ─── Refresh (titles sync) ────────────────────────────────── async def _handle_refresh_titles_callback(self, callback: CallbackQuery) -> None: await callback.answer() @@ -382,9 +406,9 @@ class AutoBumpBot: 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) + for info, thread in zip(threads_info, threads): + if info and info.title and info.title != thread.title: + await self._db.update_thread_title(thread.id, info.title) updated += 1 await status_msg.edit_text( @@ -399,6 +423,104 @@ class AutoBumpBot: logger.error(f"Error refreshing titles: {e}", exc_info=True) await callback.message.answer(f"❌ Ошибка обновления: {str(e)}") + # ─── My Topics (forum listing) ───────────────────────────── + + async def _handle_my_threads_callback(self, callback: CallbackQuery) -> None: + await callback.answer() + if not callback.message: + return + await self._show_my_threads(callback.message.chat.id, page=1) + + async def _handle_my_threads_page_callback(self, callback: CallbackQuery) -> None: + await callback.answer() + if not callback.message: + return + try: + page = int(callback.data.split("_", 1)[1]) + except ValueError: + page = 1 + await self._show_my_threads(callback.message.chat.id, page=max(1, page), + message_to_edit=callback.message) + + async def _show_my_threads( + self, chat_id: int, page: int = 1, message_to_edit: Message | None = None + ) -> None: + try: + page_data = await self._api.get_my_threads(page=page, limit=MY_THREADS_PAGE_SIZE) + except Exception as e: + logger.error(f"Error fetching my threads: {e}", exc_info=True) + text = f"❌ Не удалось получить список тем с форума: {e}" + if message_to_edit: + try: + await message_to_edit.edit_text(text) + except TelegramBadRequest: + pass + else: + await self._bot.send_message(chat_id, text) + return + + total_pages = max(1, -(-page_data.total // MY_THREADS_PAGE_SIZE)) + + if not page_data.threads: + text = "📭 Ваши темы на форуме не найдены (или токен не даёт их видеть)" + markup = InlineKeyboardMarkup(inline_keyboard=[ + [InlineKeyboardButton(text="↩️ Back to Menu", callback_data="back_to_menu")], + ]) + else: + buttons: list[list[InlineKeyboardButton]] = [] + for t in page_data.threads: + title = self._truncate_text_safely(t.title or f"Thread {t.thread_id}", 60) + buttons.append([ + InlineKeyboardButton(text=title, callback_data=f"myadd_{t.thread_id}") + ]) + + nav: list[InlineKeyboardButton] = [] + if page > 1: + nav.append(InlineKeyboardButton(text="◀️", callback_data=f"mypage_{page - 1}")) + nav.append(InlineKeyboardButton(text=f"📄 {page}/{total_pages}", callback_data="noop")) + if page < total_pages: + nav.append(InlineKeyboardButton(text="▶️", callback_data=f"mypage_{page + 1}")) + buttons.append(nav) + buttons.append([InlineKeyboardButton(text="↩️ Back to Menu", callback_data="back_to_menu")]) + + text = ( + f"📥 Мои темы — страница {page}/{total_pages}" + f" (всего {page_data.total})\n\n" + "Нажмите на тему, чтобы добавить её в список бампа." + ) + markup = InlineKeyboardMarkup(inline_keyboard=buttons) + + if message_to_edit: + try: + await message_to_edit.edit_text(text, reply_markup=markup) + except TelegramBadRequest: + pass + else: + await self._bot.send_message(chat_id, text, reply_markup=markup) + + async def _handle_my_add_callback(self, callback: CallbackQuery) -> None: + if not callback.data: + return + thread_id = callback.data.split("_", 1)[1] + try: + info_list = await self._api.get_threads_info_batch([thread_id]) + info = info_list[0] if info_list else None + except Exception as e: + logger.error(f"Error fetching thread {thread_id} for add: {e}") + await callback.answer(f"❌ Ошибка получения темы: {e}", show_alert=True) + return + + title = (info.title if info and info.title else f"Thread {thread_id}") + added = await self._db.add_thread(thread_id, title) + if added: + logger.info(f"Thread {thread_id} added from my-topics listing") + await callback.answer(f"✅ Добавлена: {title}") + else: + await callback.answer("⚠️ Тема уже в списке", show_alert=True) + + async def _handle_noop_callback(self, callback: CallbackQuery) -> None: + await callback.answer() + # ─── Delete Thread ────────────────────────────────────────── async def _handle_delete_menu_callback(self, callback: CallbackQuery) -> None: @@ -451,9 +573,9 @@ class AutoBumpBot: if not callback.message: return try: - threads = await self._db.get_threads_to_bump(0) + threads = await self._db.get_all_threads() if not threads: - await callback.message.answer("📭 Нет тем для поднятия\n\nДобавьте темы через кнопку ➕") + await callback.message.answer("📭 Нет тем для поднятия\n\nДобавьте темы через кнопку ➕ или 📥") await self._send_main_menu(callback.message.chat.id) return @@ -488,7 +610,7 @@ class AutoBumpBot: return try: threads = await self._db.get_all_threads() - interval = await self._get_interval() + interval = await self._get_interval_minutes() threads_ready = await self._db.get_threads_to_bump(interval) stats = await self._db.get_bump_stats() batch_size = await self._get_batch_size() @@ -514,7 +636,7 @@ class AutoBumpBot: f"• Успешность: {success_rate:.1f}%\n" f"• Последний бамп: {stats.last_bump_time or '—'}\n\n" f"⚙️ Настройки:\n" - f"• Интервал: {interval:.0f}ч\n" + f"• Интервал: {_format_interval(interval)}\n" f"• Batch size: {batch_size}\n" f"• Автобамп: {'✅ Вкл' if auto_enabled else '❌ Выкл'}\n\n" f"⏱️ Работа:\n" @@ -534,13 +656,13 @@ class AutoBumpBot: await callback.answer() if not callback.message: return - interval = await self._get_interval() + interval = await self._get_interval_minutes() batch_size = await self._get_batch_size() auto_enabled = await self._is_auto_bump_enabled() text = ( "🛠️ Текущие настройки\n\n" - f"⏰ Интервал бампа: {interval:.0f}ч\n" + f"⏰ Интервал бампа: {_format_interval(interval)}\n" f"📦 Batch size: {batch_size}\n" f"🔄 Автобамп: {'✅ Вкл' if auto_enabled else '❌ Выкл'}\n" ) @@ -550,10 +672,10 @@ class AutoBumpBot: await callback.answer() if not callback.message: return - interval = await self._get_interval() + interval = await self._get_interval_minutes() await callback.message.answer( - f"Текущий интервал: {interval:.0f}ч\n\n" - "Введите новый интервал в часах (например: 6 или 24):" + f"Текущий интервал: {_format_interval(interval)}\n\n" + "Введите новый интервал в минутах (например: 5 или 720):" ) await state.set_state(BotStates.waiting_for_interval) @@ -564,18 +686,15 @@ class AutoBumpBot: 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}") + await self._db.set_setting("bump_interval_minutes", str(new_interval)) + logger.info(f"Bump interval changed to {new_interval} min 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}ч") + # Tick-based scheduler picks up the new interval on the next tick — no restart needed + await message.answer(f"✅ Интервал изменён на {_format_interval(new_interval)}") await self._send_main_menu(message.chat.id) except ValueError: - await message.answer("❌ Введите число (например: 12)") + await message.answer("❌ Введите число минут (например: 5)") except Exception as e: logger.error(f"Error changing interval: {e}") await message.answer(f"❌ Ошибка: {str(e)}") @@ -601,7 +720,7 @@ class AutoBumpBot: return await self._db.set_setting("batch_size", str(new_size)) - self._api._batch_size = new_size + self._api.set_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}") @@ -628,6 +747,7 @@ class AutoBumpBot: self._restart_auto_bump() status = "✅ Включён" else: + self._stop_auto_bump() status = "❌ Выключен" logger.info(f"Auto-bump toggled to {new_state} by user {callback.from_user.id}") @@ -655,24 +775,21 @@ class AutoBumpBot: results = await self._api.bump_threads_batch(thread_ids) + success_count = 0 for result in results: if result.success: + success_count += 1 logger.info( f"✅ BUMP SUCCESS | Thread: {result.thread_id} | " f"Status: {result.status.value} | Message: {result.message}" ) + await self._db.update_last_bumped(result.thread_id) 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) @@ -702,40 +819,45 @@ class AutoBumpBot: except Exception as e: logger.error(f"Failed to send notification: {e}") - # ─── Auto-Bump Scheduler ─────────────────────────────────── + # ─── Auto-Bump Scheduler (tick-based) ────────────────────── def _restart_auto_bump(self) -> None: - if self._auto_bump_task and not self._auto_bump_task.done(): - self._auto_bump_task.cancel() + self._stop_auto_bump() self._auto_bump_task = asyncio.create_task(self._auto_bump_scheduler_loop()) + def _stop_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 = None + async def _auto_bump_scheduler_loop(self) -> None: - logger.info("Auto-bump scheduler started") + """Tick every SCHEDULER_TICK_SECONDS: read settings from DB, bump due threads. + + Reads interval/auto-bump from the DB on every tick, so settings changes + apply immediately without restarting the loop and without losing phase. + """ + tick_seconds = self._config.scheduling.scheduler_tick_seconds + logger.info(f"Auto-bump scheduler started (tick every {tick_seconds:.0f}s)") 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) - + await asyncio.sleep(tick_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") + if not await self._is_auto_bump_enabled(): continue - logger.info("Starting scheduled bump...") + interval = await self._get_interval_minutes() threads = await self._db.get_threads_to_bump(interval) + if not threads: + continue - 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") + logger.info(f"Auto-bump: {len(threads)} threads are due") + results = await self._execute_bump_with_notifications(threads, chat_id=None) + success_count = sum(1 for r in results if r.success) + logger.info( + f"Scheduled bump completed: {success_count}/{len(results)} successful" + ) except asyncio.CancelledError: logger.info("Auto-bump loop cancelled") @@ -757,6 +879,17 @@ class AutoBumpBot: await self._api.start() logger.info("API client started") + # Validate the Lolz API token early so misconfiguration is visible immediately + me = await self._api.get_me() + if me: + logger.info( + f"API token OK | user_id={me.get('user_id', '?')} username={me.get('username', '?')}" + ) + else: + logger.warning( + "API token validation failed — bumps will likely fail until API_AUTH_TOKEN is fixed" + ) + self._start_time = datetime.now() self._is_running = True @@ -774,8 +907,7 @@ class AutoBumpBot: 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() + self._stop_auto_bump() await self._cleanup_resources() logger.info("Bot stopped") diff --git a/config_manager.py b/config_manager.py index 22436b3..727c376 100644 --- a/config_manager.py +++ b/config_manager.py @@ -6,6 +6,9 @@ from typing import Self from dotenv import load_dotenv +# Official API servers per forum.json (spec): production + documented alternates +DEFAULT_API_BASE_URL = "https://api.lolz.live" + @dataclass(frozen=True, slots=True) class BotConfig: @@ -28,9 +31,10 @@ class DatabaseConfig: @dataclass(frozen=True, slots=True) class SchedulingConfig: - bump_interval_hours: float + bump_interval_minutes: float bump_delay_seconds: float enable_auto_bump: bool + scheduler_tick_seconds: float @dataclass(frozen=True, slots=True) @@ -50,7 +54,13 @@ class Config: bot_token = os.getenv("BOT_API_TOKEN", "") api_token = os.getenv("API_AUTH_TOKEN", "") - admin_user_id = int(os.getenv("ADMIN_USER_ID", "0")) + + try: + admin_user_id = int(os.getenv("ADMIN_USER_ID", "0")) + except ValueError: + raise ValueError( + "ADMIN_USER_ID must be an integer — set your Telegram user ID in .env" + ) cls._validate_tokens(bot_token, api_token, admin_user_id) @@ -65,7 +75,7 @@ class Config: raise ValueError("API_BATCH_SIZE must be between 1 and 10") api = APIConfig( - base_url=os.getenv("API_BASE_URL", "").rstrip("/"), + base_url=os.getenv("API_BASE_URL", DEFAULT_API_BASE_URL).rstrip("/"), auth_token=api_token, batch_size=batch_size, ) @@ -74,14 +84,19 @@ class Config: 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") + interval_minutes = float(os.getenv("BUMP_INTERVAL_MINUTES", "60")) + if interval_minutes <= 0: + raise ValueError("BUMP_INTERVAL_MINUTES must be positive") + + tick = float(os.getenv("SCHEDULER_TICK_SECONDS", "60")) + if tick <= 0: + raise ValueError("SCHEDULER_TICK_SECONDS must be positive") scheduling = SchedulingConfig( - bump_interval_hours=interval, + bump_interval_minutes=interval_minutes, bump_delay_seconds=float(os.getenv("BUMP_DELAY_SECONDS", "2")), enable_auto_bump=os.getenv("ENABLE_AUTO_BUMP", "true").lower() == "true", + scheduler_tick_seconds=tick, ) return cls( diff --git a/database.py b/database.py index fc1c723..b1084fc 100644 --- a/database.py +++ b/database.py @@ -111,8 +111,7 @@ class Database: 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), + "bump_interval_minutes": str(config.scheduling.bump_interval_minutes), "enable_auto_bump": str(config.scheduling.enable_auto_bump).lower(), "batch_size": str(config.api.batch_size), } @@ -228,19 +227,20 @@ class Database: logger.error(f"Error fetching all threads: {e}") raise - async def get_threads_to_bump(self, interval_hours: float) -> list[Thread]: + async def get_threads_to_bump(self, interval_minutes: float) -> list[Thread]: + """Threads due for a bump: the interval has passed since the last one.""" self._ensure_connected() - if interval_hours < 0: - raise ValueError("interval_hours must be non-negative") + if interval_minutes < 0: + raise ValueError("interval_minutes 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') + OR datetime(last_bumped, '+' || ? || ' minutes') <= datetime('now') ORDER BY last_bumped ASC NULLS FIRST """, - (interval_hours,), + (interval_minutes,), ) as cursor: rows = await cursor.fetchall() return [Thread(id=row[0], title=row[1], last_bumped=row[2]) for row in rows]