Compare commits

2 Commits

Author SHA1 Message Date
imletbruh f23944d4c5 Add files via upload 2026-09-08 18:46:51 +05:00
imletbruh 134fab488b Add files via upload 2026-09-06 09:42:56 +05:00
5 changed files with 812 additions and 585 deletions
+56 -57
View File
@@ -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` — ошибки от форума (например, «нужно подождать») видны по каждой теме
## 👤 Автор
+397 -379
View File
@@ -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,12 +64,128 @@ 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"<br\s*/?>", "\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")
__slots__ = ("_base_url", "_auth_token", "_session", "_batch_size", "_batch_delay_seconds")
def __init__(self, base_url: str, auth_token: str, batch_size: int = 10) -> None:
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")
@@ -57,14 +196,13 @@ class APIClient:
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:
@@ -73,164 +211,271 @@ class APIClient:
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"
async def _safe_json(resp: aiohttp.ClientResponse) -> object:
try:
return await resp.json(content_type=None)
except Exception:
return None
# Remove HTML tags
error_msg = re.sub(r"<br\s*/?>", "\n", error_msg)
error_msg = re.sub(r"<[^>]+>", "", error_msg)
@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] = []
# Split by newlines and filter empty parts
parts = [p.strip() for p in error_msg.split("\n") if p.strip()]
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 not parts:
return "Unknown error"
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)
# Look for rate limit message first
for part in parts:
if "должны подождать" in part.lower() or "должен подождать" in part.lower():
return part
positive = [c for c in candidates if c > 0]
return min(positive) if positive else None
# Return last meaningful part
return parts[-1]
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()
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=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 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)
last_error = f"rate limited (429), waited {wait:.0f}s"
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 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)
last_error = f"rate limited (429), waited {wait:.0f}s"
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 get_threads_info_batch)."""
"""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 information using batch API (up to 10 per request)."""
if not self._session:
await self.start()
"""Get multiple thread titles 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[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)
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]:
"""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}"
}
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)
batch_url = f"{self._base_url}/batch"
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
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}")
# ─── My threads listing ────────────────────────────────────
if resp.status == 401:
raise ValueError("Invalid API token - check your configuration")
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")
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)
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 ""),
))
# Parse batch response
batch_response = await resp.json()
logger.debug(f"Batch response: {batch_response}")
try:
total = int(body.get("threads_total") or len(items))
except (TypeError, ValueError):
total = len(items)
return ThreadsPage(threads=items, total=total)
# 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)
# ─── Token validation ──────────────────────────────────────
jobs = batch_response["jobs"]
if not isinstance(jobs, dict):
logger.error(f"Jobs is not a dict: {type(jobs)}")
return [None] * len(thread_ids)
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
# 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
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)
# ─── Bump ──────────────────────────────────────────────────
async def bump_thread(self, thread_id: str) -> BumpResult:
"""Bump single thread via API (legacy method, prefer bump_threads_batch)."""
@@ -238,280 +483,53 @@ class APIClient:
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
]
logger.info(f"🚀 Executing bump batch request for {len(thread_ids)} threads: {list(thread_ids)}")
batch_url = f"{self._base_url}/batch"
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
]
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
)
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
+224 -87
View File
@@ -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()
@@ -55,18 +68,22 @@ class AuthMiddleware(BaseMiddleware):
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
# `event` is the raw Update; `.event` resolves it to whichever concrete
# sub-object is actually set (message, callback_query, chat_member, ...).
try:
actual_event = event.event
except Exception:
actual_event = event
from_user = getattr(actual_event, "from_user", None)
if from_user is None:
return await handler(event, data)
# Unknown identity (e.g. channel_post, poll, message_reaction) — fail
# closed instead of letting it through unauthenticated.
logger.warning(
f"AuthMiddleware: update without from_user "
f"(type={getattr(event, 'event_type', '?')}), denying"
)
return None
if from_user.id != self._admin_user_id:
chat_id = None
@@ -110,6 +127,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 +165,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 +181,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 +209,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 +246,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,18 +262,17 @@ 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=(
"🤖 <b>QIYANA AUTO-BUMP BOT</b>\n\n"
"Бот для автоматического поднятия тем на Lolz.live\n\n"
f"⏰ Автоподнятие каждые <b>{interval:.0f}ч</b>"
f"⏰ Автоподнятие каждые <b>{_format_interval(interval)}</b>"
),
reply_markup=self._main_menu_kb(),
)
except TelegramBadRequest as e:
logger.error(f"Failed to send start message: {e}")
except Exception as e:
logger.error(f"Failed to send start message: {e}", exc_info=True)
await message.answer("❌ Ошибка отправки сообщения. Попробуйте /start снова.")
# ─── Add Thread ─────────────────────────────────────────────
@@ -256,7 +283,8 @@ class AutoBumpBot:
return
await callback.message.answer(
"📝 Введите ID тем через запятую для добавления:\n"
"Пример: <code>12345, 67890, 11111</code>"
"Пример: <code>12345, 67890, 11111</code>\n\n"
"Или добавьте свои темы с форума через кнопку 📥 <b>My topics</b>"
)
await state.set_state(BotStates.waiting_for_thread_ids)
@@ -289,9 +317,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 +361,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 +389,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 +410,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 +427,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"📥 <b>Мои темы</b> — страница {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 +577,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 +614,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 +640,7 @@ class AutoBumpBot:
f"• Успешность: {success_rate:.1f}%\n"
f"• Последний бамп: {stats.last_bump_time or ''}\n\n"
f"⚙️ <b>Настройки:</b>\n"
f"• Интервал: {interval:.0f}ч\n"
f"• Интервал: {_format_interval(interval)}\n"
f"• Batch size: {batch_size}\n"
f"• Автобамп: {'✅ Вкл' if auto_enabled else '❌ Выкл'}\n\n"
f"⏱️ <b>Работа:</b>\n"
@@ -534,13 +660,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 = (
"🛠️ <b>Текущие настройки</b>\n\n"
f"⏰ Интервал бампа: <b>{interval:.0f}ч</b>\n"
f"⏰ Интервал бампа: <b>{_format_interval(interval)}</b>\n"
f"📦 Batch size: <b>{batch_size}</b>\n"
f"🔄 Автобамп: <b>{'✅ Вкл' if auto_enabled else '❌ Выкл'}</b>\n"
)
@@ -550,10 +676,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"Текущий интервал: <b>{interval:.0f}ч</b>\n\n"
"Введите новый интервал в часах (например: <code>6</code> или <code>24</code>):"
f"Текущий интервал: <b>{_format_interval(interval)}</b>\n\n"
"Введите новый интервал в минутах (например: <code>5</code> или <code>720</code>):"
)
await state.set_state(BotStates.waiting_for_interval)
@@ -564,18 +690,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"✅ Интервал изменён на <b>{new_interval:.0f}ч</b>")
# Tick-based scheduler picks up the new interval on the next tick — no restart needed
await message.answer(f"✅ Интервал изменён на <b>{_format_interval(new_interval)}</b>")
await self._send_main_menu(message.chat.id)
except ValueError:
await message.answer("❌ Введите число (например: <code>12</code>)")
await message.answer("❌ Введите число минут (например: <code>5</code>)")
except Exception as e:
logger.error(f"Error changing interval: {e}")
await message.answer(f"❌ Ошибка: {str(e)}")
@@ -601,7 +724,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 изменён на <b>{new_size}</b>")
@@ -628,6 +751,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,23 +779,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.record_bump_success(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)
await self._db.record_bump_failure(result.thread_id)
total = len(results)
await self._db.increment_bump_stats(success_count, total)
@@ -702,40 +824,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 +884,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 +912,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")
+23 -8
View File
@@ -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_delay_seconds=float(os.getenv("BUMP_DELAY_SECONDS", "2")),
bump_interval_minutes=interval_minutes,
bump_delay_seconds=float(os.getenv("BUMP_DELAY_SECONDS", "1")),
enable_auto_bump=os.getenv("ENABLE_AUTO_BUMP", "true").lower() == "true",
scheduler_tick_seconds=tick,
)
return cls(
+70 -12
View File
@@ -73,15 +73,32 @@ class Database:
id TEXT PRIMARY KEY,
title TEXT NOT NULL,
last_bumped TEXT,
last_attempt TEXT,
fail_streak INTEGER NOT NULL DEFAULT 0,
created_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
# Migration for DBs created before last_attempt/fail_streak existed.
async with self._connection.execute("PRAGMA table_info(threads)") as cursor:
existing_columns = {row[1] for row in await cursor.fetchall()}
if "last_attempt" not in existing_columns:
await self._connection.execute("ALTER TABLE threads ADD COLUMN last_attempt TEXT")
if "fail_streak" not in existing_columns:
await self._connection.execute(
"ALTER TABLE threads ADD COLUMN fail_streak INTEGER NOT NULL DEFAULT 0"
)
await self._connection.execute("""
CREATE INDEX IF NOT EXISTS idx_last_bumped
ON threads(last_bumped)
""")
await self._connection.execute("""
CREATE INDEX IF NOT EXISTS idx_last_attempt
ON threads(last_attempt)
""")
await self._connection.execute("""
CREATE TABLE IF NOT EXISTS settings (
key TEXT PRIMARY KEY,
@@ -111,8 +128,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 +244,36 @@ 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 attempt.
Threads that keep failing back off exponentially (interval * 2**fail_streak,
capped at 60x) so a permanently broken thread doesn't get retried on every
single tick and burn through the batch rate limit.
"""
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')
ORDER BY last_bumped ASC NULLS FIRST
WHERE last_attempt IS NULL
OR datetime(
last_attempt,
'+' || (? * CASE
WHEN fail_streak <= 0 THEN 1
WHEN fail_streak = 1 THEN 2
WHEN fail_streak = 2 THEN 4
WHEN fail_streak = 3 THEN 8
WHEN fail_streak = 4 THEN 16
WHEN fail_streak = 5 THEN 32
ELSE 60
END) || ' minutes'
) <= datetime('now')
ORDER BY last_attempt 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]
@@ -248,16 +281,41 @@ class Database:
logger.error(f"Error fetching threads to bump: {e}")
raise
async def update_last_bumped(self, thread_id: str) -> None:
async def record_bump_success(self, thread_id: str) -> None:
"""Mark a thread as successfully bumped and reset its failure backoff."""
self._ensure_connected()
try:
await self._connection.execute(
"UPDATE threads SET last_bumped = datetime('now') WHERE id = ?",
"""
UPDATE threads SET
last_bumped = datetime('now'),
last_attempt = datetime('now'),
fail_streak = 0
WHERE id = ?
""",
(thread_id,),
)
await self._connection.commit()
except Exception as e:
logger.error(f"Error updating last_bumped for thread {thread_id}: {e}")
logger.error(f"Error recording bump success for thread {thread_id}: {e}")
raise
async def record_bump_failure(self, thread_id: str) -> None:
"""Record a failed bump attempt and increase its retry backoff."""
self._ensure_connected()
try:
await self._connection.execute(
"""
UPDATE threads SET
last_attempt = datetime('now'),
fail_streak = fail_streak + 1
WHERE id = ?
""",
(thread_id,),
)
await self._connection.commit()
except Exception as e:
logger.error(f"Error recording bump failure for thread {thread_id}: {e}")
raise
async def delete_thread(self, thread_id: str) -> bool: