forked from Lolz/Auto-Pump
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| f23944d4c5 | |||
| 134fab488b |
@@ -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` — ошибки от форума (например, «нужно подождать») видны по каждой теме
|
||||
|
||||
## 👤 Автор
|
||||
|
||||
|
||||
+439
-421
@@ -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,472 @@ 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")
|
||||
|
||||
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"<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 "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)
|
||||
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 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)
|
||||
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 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
|
||||
|
||||
@@ -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
@@ -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
@@ -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:
|
||||
|
||||
Reference in New Issue
Block a user