diff --git a/.env.example b/.env.example index adf4b0b..d6fff5b 100644 --- a/.env.example +++ b/.env.example @@ -1,9 +1,22 @@ +# ========================================== +# Donor(VK) -> Recipient(TG/MAX) routes +# ========================================== +# Multiple VK donor groups, each with its own Telegram/MAX destination(s) and its own +# TG/MAX on-off switches, are configured in a separate JSON file (default data/routes.json), +# NOT here. See data/routes.json.example for the format. +# +# VK_SOURCE / TG_CHAT_ID / MAX_CHAT_ID below are ONLY used once, as a fallback: if +# data/routes.json doesn't exist yet, it gets auto-generated from these three values +# as a single route (id "default") on first run. After that, edit routes.json directly - +# these three env vars are then ignored. +ROUTES_CONFIG_PATH=data/routes.json + # ========================================== # VKontakte Settings # ========================================== -# VK User/Service access token +# VK User/Service access token (shared by all routes) VK_ACCESS_TOKEN=vk1.a.your_vk_token_here -# Target VK source: screen name, URL or owner_id (e.g., "redairsoft", "club123456", "-123456", "https://vk.com/redairsoft") +# Legacy single-route fallback - see routes.json note above VK_SOURCE=redairsoft # VK API Version VK_API_VERSION=5.199 @@ -13,9 +26,10 @@ VK_CHECK_COUNT=10 # ========================================== # Telegram Settings # ========================================== -# Telegram Bot Token (from @BotFather) +# Telegram Bot Token (from @BotFather) - shared by all routes, posts into whichever +# chat_id each route configures TG_BOT_TOKEN=123456789:ABCdefGHIjklMNOpqrSTUvwxYZ -# Destination Chat/Channel ID (e.g., "-1001234567890" or with topic thread "-1001234567890:42") +# Legacy single-route fallback - see routes.json note above TG_CHAT_ID=-1001234567890 # Optional TG Media Storage Channel (e.g. for generating permanent file_ids) TG_MEDIA_CHANNEL_ID= @@ -36,9 +50,9 @@ TELEGRAM_API_HASH= # ========================================== # MAX Messenger Settings # ========================================== -# MAX Bot Token +# MAX Bot Token - shared by all routes, posts into whichever chat_id each route configures MAX_BOT_TOKEN=your_max_bot_token_here -# MAX Destination Chat ID +# Legacy single-route fallback - see routes.json note above MAX_CHAT_ID=123456 # MAX API Base URL MAX_API_BASE_URL=https://platform-api2.max.ru diff --git a/README.md b/README.md index 4b83780..72411ec 100644 --- a/README.md +++ b/README.md @@ -6,47 +6,42 @@ ## 📌 Инфраструктура и Деплой +**Актуальный способ деплоя — Coolify**, автодеплой по пушу в `gitea/main`. Старый ручной способ (checkout в `/opt/redairsoft_poster/` + `docker-compose` руками через Proxmox, описанный ниже архивно) больше не используется — тот контейнер (`redairsoft-vk-poster`) был удалён 2026-08-15 как заброшенный. + | Параметр | Значение | |---|---| -| **Хост Proxmox** | `192.168.1.222` (root SSH через ключ `~/.ssh/id_ed25519_proxmox`) | -| **LXC Контейнер** | **CT 107** (`redairsoft-poster`, IP: `192.168.1.106`, Debian 12) | -| **Ресурсы LXC** | 2 vCPU, 1.5 GB RAM, 512 MB Swap, 16 GB Disk, `onboot=1` | +| **Деплой** | Coolify, приложение "RedAirsoft Poster", сервер `redairsoft-poster-lxc` | +| **URL** | `http://hjbmqj1ohievdrebpltnzex0.192.168.1.106.sslip.io` | +| **Хост контейнера** | Тот же LXC 107 (`192.168.1.106`) — Coolify управляет им напрямую, не через `/opt/redairsoft_poster/` | +| **Имя Docker-контейнера** | Динамическое, вида `-`, тег образа = хэш закоммиченного коммита. Смотреть через `docker ps` — актуальный это `Up`, с тегом = последний `git log` хэш | | **Gitea Репозиторий** | `http://192.168.1.135:3000/exostring/redairsoft_vk_gt_max_sender` | -| **Gitea SSH Remote** | `ssh://git@192.168.1.135:2222/exostring/redairsoft_vk_gt_max_sender.git` | -| **Путь проекта на LXC 107** | `/opt/redairsoft_poster/` | -| **Имя Docker контейнера** | `redairsoft-vk-poster` | +| **Хост Proxmox** (для ручной диагностики) | `192.168.1.222` (root SSH через ключ `~/.ssh/id_ed25519_proxmox`) | | **Локальный Telegram Bot API** | Встроен в контейнер (`http://127.0.0.1:8081`), поддерживает загрузку видео до 2 ГБ | -| **База данных** | SQLite `/opt/redairsoft_poster/data/poster.db` (персистентный том Docker `./data`) | +| **База данных** | SQLite `/app/data/poster.db` внутри контейнера (персистентный том) | ---- - -## 🚀 Управление проектом (Команды из консоли / PowerShell) - -Все команды выполняются через Proxmox хост: +### Диагностика через SSH (read-only) ```bash -# 1. Посмотреть статус и логи бота: -ssh -i ~/.ssh/id_ed25519_proxmox root@192.168.1.222 'pct exec 107 -- docker logs --tail 50 redairsoft-vk-poster' +# Найти актуальный контейнер (смотреть на тег образа = свежий commit hash): +ssh -i ~/.ssh/id_ed25519_proxmox root@192.168.1.222 'pct exec 107 -- docker ps -a' -# 2. Логи в реальном времени (follow): -ssh -i ~/.ssh/id_ed25519_proxmox root@192.168.1.222 'pct exec 107 -- docker logs -f redairsoft-vk-poster' - -# 3. Запустить проект: -ssh -i ~/.ssh/id_ed25519_proxmox root@192.168.1.222 'pct exec 107 -- bash -c "cd /opt/redairsoft_poster && docker-compose up -d"' - -# 4. Остановить проект: -ssh -i ~/.ssh/id_ed25519_proxmox root@192.168.1.222 'pct exec 107 -- bash -c "cd /opt/redairsoft_poster && docker-compose stop"' - -# 5. Обновить из Gitea и пересобрать: -ssh -i ~/.ssh/id_ed25519_proxmox root@192.168.1.222 'pct exec 107 -- bash -c "cd /opt/redairsoft_poster && git pull origin main && docker-compose build && docker-compose up -d"' +# Логи актуального контейнера: +ssh -i ~/.ssh/id_ed25519_proxmox root@192.168.1.222 'pct exec 107 -- docker logs --tail 100 <имя-контейнера>' ``` +Деплой новой версии — просто `git push` в `gitea/main`, Coolify подхватывает автоматически. + --- ## ⚙️ Как работает сервис +0. **Донор→рецепиент маршруты (`data/routes.json`)**: + - Сервис может опрашивать сразу несколько групп ВК ("доноров"), и для каждой отдельно настроено, в какие Telegram и МАКС чаты публиковать ("рецепиенты"), плюс независимые вкл/выкл для TG и МАКС на каждом маршруте. + - Список маршрутов хранится в `data/routes.json` (персистентный том, не в git). Если файла нет — при первом запуске он создаётся автоматически из старых `.env`-переменных `VK_SOURCE`/`TG_CHAT_ID`/`MAX_CHAT_ID` (один маршрут с `id: "default"`), так что апгрейд с однo-группового режима ничего не ломает. + - Пример формата — [`routes.json.example`](routes.json.example) в корне репозитория. Правка `routes.json` требует перезапуска контейнера, чтобы изменения подхватились. + - Маршруты в рамках одного цикла опроса обрабатываются **строго последовательно** (без параллелизма) — это защищает от флуда на VK API, загрузке медиа в Telegram/локальный Bot API и отправке в МАКС, поскольку токены ботов общие на все маршруты. 1. **Опрос стены ВКонтакте**: - - Каждые 15 минут делает запрос к методу `wall.get` для группы `public36860851` (`owner_id: -36860851`). + - Каждые 15 минут (`CHECK_INTERVAL_MINUTES`) делает запрос к методу `wall.get` для каждой группы из `routes.json`. - Фильтрует посты автора (без репостов и рекламы сторонних сообществ). 2. **Защита от спама и режим запуска (`BOOTSTRAP_MODE`)**: - Установлен режим `BOOTSTRAP_MODE=skip_existing`. @@ -70,19 +65,72 @@ ssh -i ~/.ssh/id_ed25519_proxmox root@192.168.1.222 'pct exec 107 -- bash -c "cd --- +## 🔀 Маршруты донор→рецепиент (`data/routes.json`) + +Формат одного маршрута: + +```json +{ + "id": "redairsoft_main", + "name": "Red Airsoft (основная группа)", + "vk_source": "public36860851", + "tg_chat_id": "-1001303630155", + "tg_enabled": true, + "max_chat_id": "123456", + "max_enabled": true +} +``` + +| Поле | Описание | +|---|---| +| `id` | Уникальный технический идентификатор маршрута (используется в БД и логах) | +| `name` | Человекочитаемое имя для отчётов администраторам | +| `vk_source` | Группа-донор ВК: screen name, URL или owner_id | +| `tg_chat_id` / `max_chat_id` | Куда публиковать (chat/channel ID) | +| `tg_enabled` / `max_enabled` | Переключатель — публиковать ли в эту платформу для этого маршрута (только TG, только МАКС, или оба) | + +Можно завести несколько маршрутов с разными `vk_source`, каждый — в свои TG/МАКС чаты. Токены ботов (`TG_BOT_TOKEN`, `MAX_BOT_TOKEN`, `VK_ACCESS_TOKEN`) общие на все маршруты — один бот пишет в разные чаты. + +`vk_source` принимает screen name, полный URL (`vk.com` и `vk.ru`) или `owner_id` — можно указывать как есть, без ручной нормализации. + +Файл — обычный JSON, но допускает построчные комментарии `// текст`, чтобы подписывать, где какой маршрут (полноценных JSON-комментариев не существует, здесь это добавлено отдельно — строка, у которой после пробелов идёт `//`, вырезается перед парсингом): + +```jsonc +[ + +// redairsoft +{ + "id": "redairsoft", + "vk_source": "public36860851", + "tg_chat_id": "-1001303630155", + "max_chat_id": "-69722432869632" +}, + +// strike_expo +{ + "id": "strike_expo", + "vk_source": "https://vk.ru/strike_expo", + "tg_chat_id": "-1003099077190", + "max_chat_id": "-77898705339648" +} + +] +``` + +--- + ## 🔑 Конфигурация (`.env`) | Переменная | Описание | |---|---| -| `VK_ACCESS_TOKEN` | Сервисный токен приложения ВК (`d0a64...`) | -| `VK_SOURCE` | `public36860851` | -| `TG_BOT_TOKEN` | Токен Telegram бота (от `@BotFather`) | -| `TG_CHAT_ID` | ID канала Telegram (например `-1001303630155`) | +| `VK_ACCESS_TOKEN` | Сервисный токен приложения ВК (`d0a64...`), общий на все маршруты | +| `ROUTES_CONFIG_PATH` | Путь к файлу маршрутов, по умолчанию `data/routes.json` | +| `VK_SOURCE` / `TG_CHAT_ID` / `MAX_CHAT_ID` | Легаси fallback: используются только для авто-генерации первого маршрута, если `routes.json` ещё не существует | +| `TG_BOT_TOKEN` | Токен Telegram бота (от `@BotFather`), общий на все маршруты | | `TG_MEDIA_CHANNEL_ID` | Скрытый канал-хранилище (опционально) | | `TG_ADMIN_IDS` | Telegram ID админов через запятую (например `442509142`) | | `LOCAL_BOT_API_URL` | `http://127.0.0.1:8081` (встроен в контейнер) | -| `MAX_BOT_TOKEN` | Токен бота в мессенджере МАКС | -| `MAX_CHAT_ID` | ID чата в МАКС | +| `MAX_BOT_TOKEN` | Токен бота в мессенджере МАКС, общий на все маршруты | | `BOOTSTRAP_MODE` | `skip_existing` | | `CHECK_INTERVAL_MINUTES` | `15` | | `VIDEO_MAX_DURATION_SEC` | `7200` (2 часа) | @@ -95,18 +143,20 @@ ssh -i ~/.ssh/id_ed25519_proxmox root@192.168.1.222 'pct exec 107 -- bash -c "cd ``` ├── src/ │ ├── config.py # Настройки Pydantic Settings -│ ├── database.py # Хранилище SQLite (посты, статусы, даты) +│ ├── routes.py # Загрузка/валидация data/routes.json (донор->рецепиент маршруты) +│ ├── database.py # Хранилище SQLite (посты, статусы по route_id, даты) │ ├── vk_client.py # Клиент VK API (wall.get, извлечение медиа) │ ├── media_processor.py # Загрузка фото/видео (yt-dlp, aiohttp) │ ├── text_formatter.py # Конвертер ссылок ВК, очистка текста, HTML │ ├── tg_poster.py # Отправка в Telegram (Rich Message + Local API) │ ├── max_poster.py # Отправка в MAX Messenger (Uploads + Polling) │ ├── cleaner.py # Фоновая очистка временных файлов -│ ├── admin_notifier.py # Отправка отчётов администраторам -│ └── main.py # Точка входа и главный цикл опроса +│ ├── admin_notifier.py # Агрегированные отчёты администраторам по всем маршрутам +│ └── main.py # Точка входа, цикл опроса по всем маршрутам ├── Dockerfile # Multi-stage образ: aiogram/telegram-bot-api + python:3.12-slim + ffmpeg ├── docker-compose.yml # Конфигурация запуска сервиса ├── docker-entrypoint.sh # Автозапуск локального Telegram Bot API + постера ├── requirements.txt # Python зависимости (aiogram, yt-dlp, loguru, aiohttp, pydantic) +├── routes.json.example # Пример формата data/routes.json └── .env # Переменные окружения ``` diff --git a/routes.json.example b/routes.json.example new file mode 100644 index 0000000..e33482c --- /dev/null +++ b/routes.json.example @@ -0,0 +1,25 @@ +[ + +// redairsoft_main +{ + "id": "redairsoft_main", + "name": "Red Airsoft (основная группа)", + "vk_source": "public36860851", + "tg_chat_id": "-1001303630155", + "tg_enabled": true, + "max_chat_id": "123456", + "max_enabled": true +}, + +// second_donor - только в TG, MAX выключен +{ + "id": "second_donor", + "name": "Вторая донор-группа", + "vk_source": "someothergroup", + "tg_chat_id": "-1009999999999", + "tg_enabled": true, + "max_chat_id": "", + "max_enabled": false +} + +] diff --git a/src/admin_notifier.py b/src/admin_notifier.py index e5039be..85752e4 100644 --- a/src/admin_notifier.py +++ b/src/admin_notifier.py @@ -1,14 +1,15 @@ from __future__ import annotations from loguru import logger -import asyncio from datetime import datetime from typing import Any, Optional from aiogram import Bot try: from .config import settings + from .text_formatter import split_message_chunks except (ImportError, ValueError): from config import settings + from text_formatter import split_message_chunks class AdminNotifier: @@ -20,63 +21,31 @@ class AdminNotifier: if not self.admin_ids: return - for admin_id in self.admin_ids: - try: - await self.bot.send_message( - chat_id=admin_id, - text=text, - parse_mode="HTML", - disable_web_page_preview=True, - ) - except Exception as exc: - logger.warning("Failed to send report to admin {}: {}", admin_id, exc) - - async def notify_cycle_result( - self, - group_name: str, - vk_url: str, - found_posts: list[dict[str, Any]], - error: Optional[str] = None, - ) -> None: - if not self.admin_ids: - return - - # If 0 posts found and no error, check settings - if not found_posts and not error: - if not settings.report_empty_runs: - return - - now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + for chunk in split_message_chunks(text, 4000): + for admin_id in self.admin_ids: + try: + await self.bot.send_message( + chat_id=admin_id, + text=chunk, + parse_mode="HTML", + disable_web_page_preview=True, + ) + except Exception as exc: + logger.warning("Failed to send report to admin {}: {}", admin_id, exc) + def _route_section(self, route_name: str, vk_url: str, found_posts: list[dict[str, Any]], error: Optional[str]) -> Optional[str]: if error: - text = ( - f"⚠️ [VK Poster Alert] Ошибка при проверке группы\n\n" - f"🏷 Группа: {group_name} (VK)\n" - f"⏱ Время: {now_str}\n" - f"❌ Ошибка: {error}" + return ( + f"⚠️ {route_name} (VK) — ошибка проверки:\n" + f"{error}" ) - await self.send_to_all(text) - return if not found_posts: - text = ( - f"ℹ️ [VK Poster] Проверка завершена\n\n" - f"🏷 Группа: {group_name} (VK)\n" - f"⏱ Время: {now_str}\n" - f"📥 Новых постов не обнаружено." - ) - await self.send_to_all(text) - return - - # We have published posts - lines = [ - f"🚀 [VK Poster] Опубликованы новые посты!\n", - f"🏷 Группа: {group_name} (VK)", - f"⏱ Время: {now_str}", - f"📊 Количество: {len(found_posts)}\n", - "Список публикаций:", - ] + if not settings.report_empty_runs: + return None + return f"ℹ️ {route_name} (VK) — новых постов не обнаружено." + lines = [f"🚀 {route_name} (VK) — опубликовано: {len(found_posts)}"] for idx, p in enumerate(found_posts, 1): post_id = p.get("vk_post_id") vk_post_url = p.get("vk_post_url") @@ -102,4 +71,32 @@ class AdminNotifier: elif max_err: lines.append(f"• MAX: ❌ Ошибка: {max_err[:100]}") - await self.send_to_all("\n".join(lines)) + return "\n".join(lines) + + async def notify_cycle_result(self, route_results: list[dict[str, Any]]) -> None: + """route_results: one entry per route this cycle - + {"route_name": str, "vk_url": str, "reports": list[dict], "error": Optional[str]} + Sent as a single aggregated report per cycle (chunked if too long) rather + than one message per route, to avoid flooding admins when there are several + donor->recipient routes. + """ + if not self.admin_ids: + return + + sections: list[str] = [] + for rr in route_results: + section = self._route_section( + rr.get("route_name", rr.get("route_id", "?")), + rr.get("vk_url", ""), + rr.get("reports", []), + rr.get("error"), + ) + if section: + sections.append(section) + + if not sections: + return + + now_str = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + text = f"[VK Poster] Итоги проверки — {now_str}\n\n" + "\n\n".join(sections) + await self.send_to_all(text) diff --git a/src/config.py b/src/config.py index 879196f..2e3981e 100644 --- a/src/config.py +++ b/src/config.py @@ -15,14 +15,19 @@ class Settings(BaseSettings): # VK Settings vk_access_token: str = "" - vk_source: str = "" # URL, screen name or owner_id (e.g., "redairsoft", "club12345", "-12345") + vk_source: str = "" # Legacy single-route fallback: URL, screen name or owner_id (e.g., "redairsoft", "club12345", "-12345") vk_api_version: str = "5.199" vk_check_count: int = 10 vk_rate_limit_rps: int = 3 + # Donor(VK)->Recipient(TG/MAX) routes. When this file doesn't exist yet, it's + # generated from the legacy vk_source/tg_chat_id/max_chat_id below (single route, + # id "default") so existing single-group deployments keep working unchanged. + routes_config_path: str = "data/routes.json" + # Telegram Poster Settings tg_bot_token: str = "" - tg_chat_id: str = "" # Destination chat/channel, e.g. "-1001234567890" or "-1001234567890:42" + tg_chat_id: str = "" # Legacy single-route fallback, e.g. "-1001234567890" or "-1001234567890:42" tg_media_channel_id: str = "" # Optional storage channel tg_admin_ids: str = "" # Comma-separated admin IDs for reports, e.g. "123456,789012" local_bot_api_url: str = "" # e.g., "http://127.0.0.1:8081" @@ -36,7 +41,7 @@ class Settings(BaseSettings): # MAX Messenger Settings max_bot_token: str = "" - max_chat_id: str = "" # Destination chat ID in MAX + max_chat_id: str = "" # Legacy single-route fallback destination chat ID in MAX max_api_base_url: str = "https://platform-api2.max.ru" # MAX's documented hard cap for a single video attachment (dev.max.ru/docs-api). # Videos over this are sent as a text link instead of failing the whole post. @@ -104,5 +109,11 @@ class Settings(BaseSettings): p.parent.mkdir(parents=True, exist_ok=True) return p + @property + def routes_path(self) -> Path: + p = Path(self.routes_config_path).resolve() + p.parent.mkdir(parents=True, exist_ok=True) + return p + settings = Settings() diff --git a/src/database.py b/src/database.py index 682ad20..0b80ba6 100644 --- a/src/database.py +++ b/src/database.py @@ -17,6 +17,12 @@ except (ImportError, ValueError): # failing with "database is locked". _BUSY_TIMEOUT_MS = 5000 +# route_id used to backfill rows written before multi-route support existed, +# and the id load_routes() assigns to the auto-generated legacy single route - +# keeping these in sync means an upgrade from the old single-group deployment +# doesn't lose "already published" history for that group. +_LEGACY_ROUTE_ID = "default" + class Database: def __init__(self, db_path: Optional[str] = None) -> None: @@ -28,6 +34,73 @@ class Database: await db.execute(f"PRAGMA busy_timeout = {_BUSY_TIMEOUT_MS};") yield db + async def _table_columns(self, db: aiosqlite.Connection, table: str) -> set[str]: + cursor = await db.execute(f"PRAGMA table_info({table})") + rows = await cursor.fetchall() + return {row[1] for row in rows} + + async def _ensure_column(self, db: aiosqlite.Connection, table: str, column: str, ddl: str) -> None: + cols = await self._table_columns(db, table) + if column not in cols: + await db.execute(f"ALTER TABLE {table} ADD COLUMN {ddl}") + + async def _migrate_posts_table(self, db: aiosqlite.Connection) -> None: + """Adds route_id to a pre-multi-route posts table and rebuilds the UNIQUE + constraint to (route_id, vk_owner_id, vk_post_id). A plain ALTER TABLE ADD + COLUMN can't change a table-level UNIQUE constraint in SQLite, so this does + the standard rename/recreate/copy/drop dance. Existing rows are backfilled + with _LEGACY_ROUTE_ID, matching the route id routes.load_routes() assigns + to the auto-generated single route - so "already published" history for + the pre-existing group survives the upgrade. + """ + cols = await self._table_columns(db, "posts") + if "route_id" in cols: + return + + logger.info("Migrating 'posts' table: adding route_id (backfilled as '{}')", _LEGACY_ROUTE_ID) + await db.execute("ALTER TABLE posts RENAME TO posts_old;") + await db.execute( + """ + CREATE TABLE posts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + route_id TEXT NOT NULL DEFAULT 'default', + vk_post_id INTEGER NOT NULL, + vk_owner_id INTEGER NOT NULL, + posted_at INTEGER, + text TEXT, + raw_json TEXT, + tg_status TEXT DEFAULT 'pending', + tg_message_ids TEXT, + tg_url TEXT, + tg_error TEXT, + max_status TEXT DEFAULT 'pending', + max_message_ids TEXT, + max_url TEXT, + max_error TEXT, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + published_at TIMESTAMP, + UNIQUE(route_id, vk_owner_id, vk_post_id) + ); + """ + ) + await db.execute( + f""" + INSERT INTO posts ( + id, route_id, vk_post_id, vk_owner_id, posted_at, text, raw_json, + tg_status, tg_message_ids, tg_url, tg_error, + max_status, max_message_ids, max_url, max_error, + created_at, published_at + ) + SELECT id, '{_LEGACY_ROUTE_ID}', vk_post_id, vk_owner_id, posted_at, text, raw_json, + tg_status, tg_message_ids, tg_url, tg_error, + max_status, max_message_ids, max_url, max_error, + created_at, published_at + FROM posts_old; + """ + ) + await db.execute("DROP TABLE posts_old;") + logger.info("Migration of 'posts' table complete.") + async def init(self) -> None: logger.info("Initializing database at {}", self.db_path) async with self._connect() as db: @@ -36,6 +109,7 @@ class Database: """ CREATE TABLE IF NOT EXISTS posts ( id INTEGER PRIMARY KEY AUTOINCREMENT, + route_id TEXT NOT NULL DEFAULT 'default', vk_post_id INTEGER NOT NULL, vk_owner_id INTEGER NOT NULL, posted_at INTEGER, @@ -51,10 +125,12 @@ class Database: max_error TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, published_at TIMESTAMP, - UNIQUE(vk_owner_id, vk_post_id) + UNIQUE(route_id, vk_owner_id, vk_post_id) ); """ ) + await self._migrate_posts_table(db) + await db.execute( """ CREATE TABLE IF NOT EXISTS publication_runs ( @@ -68,19 +144,22 @@ class Database: ); """ ) + await self._ensure_column(db, "publication_runs", "route_id", "route_id TEXT") + await self._ensure_column(db, "publication_runs", "route_name", "route_name TEXT") await db.commit() - async def has_any_posts(self, owner_id: int) -> bool: + async def has_any_posts(self, route_id: str, owner_id: int) -> bool: async with self._connect() as db: cursor = await db.execute( - "SELECT 1 FROM posts WHERE vk_owner_id = ? LIMIT 1", - (owner_id,), + "SELECT 1 FROM posts WHERE route_id = ? AND vk_owner_id = ? LIMIT 1", + (route_id, owner_id), ) row = await cursor.fetchone() return bool(row) async def mark_post_skipped( self, + route_id: str, owner_id: int, post_id: int, posted_at: int, @@ -92,22 +171,22 @@ class Database: async with self._connect() as db: await db.execute( """ - INSERT INTO posts (vk_owner_id, vk_post_id, posted_at, text, raw_json, tg_status, max_status, tg_error, max_error) - VALUES (?, ?, ?, ?, ?, 'skipped', 'skipped', ?, ?) - ON CONFLICT(vk_owner_id, vk_post_id) DO UPDATE SET + INSERT INTO posts (route_id, vk_owner_id, vk_post_id, posted_at, text, raw_json, tg_status, max_status, tg_error, max_error) + VALUES (?, ?, ?, ?, ?, ?, 'skipped', 'skipped', ?, ?) + ON CONFLICT(route_id, vk_owner_id, vk_post_id) DO UPDATE SET tg_status = 'skipped', max_status = 'skipped'; """, - (owner_id, post_id, posted_at, text, raw_json, reason, reason), + (route_id, owner_id, post_id, posted_at, text, raw_json, reason, reason), ) await db.commit() - async def is_post_processed(self, owner_id: int, post_id: int) -> bool: + async def is_post_processed(self, route_id: str, owner_id: int, post_id: int) -> bool: async with self._connect() as db: db.row_factory = aiosqlite.Row cursor = await db.execute( - "SELECT tg_status, max_status FROM posts WHERE vk_owner_id = ? AND vk_post_id = ?", - (owner_id, post_id), + "SELECT tg_status, max_status FROM posts WHERE route_id = ? AND vk_owner_id = ? AND vk_post_id = ?", + (route_id, owner_id, post_id), ) row = await cursor.fetchone() if not row: @@ -117,18 +196,19 @@ class Database: and row["max_status"] in ("published", "skipped") ) - async def get_post(self, owner_id: int, post_id: int) -> Optional[dict[str, Any]]: + async def get_post(self, route_id: str, owner_id: int, post_id: int) -> Optional[dict[str, Any]]: async with self._connect() as db: db.row_factory = aiosqlite.Row cursor = await db.execute( - "SELECT * FROM posts WHERE vk_owner_id = ? AND vk_post_id = ?", - (owner_id, post_id), + "SELECT * FROM posts WHERE route_id = ? AND vk_owner_id = ? AND vk_post_id = ?", + (route_id, owner_id, post_id), ) row = await cursor.fetchone() return dict(row) if row else None async def save_or_update_post( self, + route_id: str, owner_id: int, post_id: int, posted_at: int, @@ -139,14 +219,14 @@ class Database: async with self._connect() as db: cursor = await db.execute( """ - INSERT INTO posts (vk_owner_id, vk_post_id, posted_at, text, raw_json) - VALUES (?, ?, ?, ?, ?) - ON CONFLICT(vk_owner_id, vk_post_id) DO UPDATE SET + INSERT INTO posts (route_id, vk_owner_id, vk_post_id, posted_at, text, raw_json) + VALUES (?, ?, ?, ?, ?, ?) + ON CONFLICT(route_id, vk_owner_id, vk_post_id) DO UPDATE SET text = excluded.text, raw_json = excluded.raw_json RETURNING id; """, - (owner_id, post_id, posted_at, text, raw_json), + (route_id, owner_id, post_id, posted_at, text, raw_json), ) row = await cursor.fetchone() await db.commit() @@ -202,6 +282,8 @@ class Database: async def record_run( self, + route_id: str, + route_name: str, found_count: int, tg_count: int, max_count: int, @@ -211,9 +293,9 @@ class Database: async with self._connect() as db: await db.execute( """ - INSERT INTO publication_runs (found_count, published_tg_count, published_max_count, status, error) - VALUES (?, ?, ?, ?, ?); + INSERT INTO publication_runs (route_id, route_name, found_count, published_tg_count, published_max_count, status, error) + VALUES (?, ?, ?, ?, ?, ?, ?); """, - (found_count, tg_count, max_count, status, error), + (route_id, route_name, found_count, tg_count, max_count, status, error), ) await db.commit() diff --git a/src/main.py b/src/main.py index e42fa99..5c9d55b 100644 --- a/src/main.py +++ b/src/main.py @@ -12,7 +12,8 @@ try: from .database import Database from .max_poster import MAXPoster from .media_processor import MediaProcessor - from .tg_poster import TelegramPoster + from .routes import Route, load_routes + from .tg_poster import TelegramPoster, parse_topic from .vk_client import VKClient, VKPost except (ImportError, ValueError): from admin_notifier import AdminNotifier @@ -21,7 +22,8 @@ except (ImportError, ValueError): from database import Database from max_poster import MAXPoster from media_processor import MediaProcessor - from tg_poster import TelegramPoster + from routes import Route, load_routes + from tg_poster import TelegramPoster, parse_topic from vk_client import VKClient, VKPost @@ -31,9 +33,9 @@ class ServiceApp: self.tg_poster = TelegramPoster() self.max_poster = MAXPoster() self.admin_notifier: Optional[AdminNotifier] = None - self.vk_group_owner_id: Optional[int] = None - self.vk_group_name: str = "" - self.vk_group_url: str = "" + self.routes: list[Route] = [] + # route.id -> {owner_id, name, url, tg_chat_id, tg_thread_id, max_chat_id} + self.route_info: dict[str, dict[str, Any]] = {} self.running = False self.cleaner_task: Optional[asyncio.Task] = None @@ -52,30 +54,49 @@ class ServiceApp: if self.tg_poster.bot: self.admin_notifier = AdminNotifier(self.tg_poster.bot) - if not settings.vk_source: - raise ValueError("VK_SOURCE is not set in configuration") + self.routes = load_routes() + if not self.routes: + raise ValueError("No routes configured - nothing to do") async with VKClient(rps=settings.vk_rate_limit_rps) as vk: - screen_name, owner_id, name = await vk.resolve_group(settings.vk_source) - self.vk_group_owner_id = owner_id - self.vk_group_name = name - self.vk_group_url = f"https://vk.com/{screen_name}" - logger.info("Resolved VK Group: '{}' (owner_id: {}, url: {})", name, owner_id, self.vk_group_url) + for route in self.routes: + screen_name, owner_id, name = await vk.resolve_group(route.vk_source) + tg_chat_id: Optional[int] = None + tg_thread_id: Optional[int] = None + if route.tg_enabled: + tg_chat_id, tg_thread_id = parse_topic(route.tg_chat_id) + self.route_info[route.id] = { + "owner_id": owner_id, + "name": name, + "url": f"https://vk.com/{screen_name}", + "tg_chat_id": tg_chat_id, + "tg_thread_id": tg_thread_id, + "max_chat_id": route.max_chat_id if route.max_enabled else None, + } + logger.info( + "Route '{}': VK '{}' (owner_id {}) -> TG {} / MAX {}", + route.id, + name, + owner_id, + "on" if route.tg_enabled else "off", + "on" if route.max_enabled else "off", + ) self.cleaner_task = asyncio.create_task(run_cleaner_loop(interval_minutes=15)) - async def process_new_post(self, post: VKPost) -> dict[str, Any]: + async def process_new_post(self, route: Route, info: dict[str, Any], post: VKPost) -> dict[str, Any]: vk_url = f"https://vk.com/wall{post.owner_id}_{post.post_id}" - logger.info("Processing post #{} from {}", post.post_id, vk_url) + logger.info("[{}] Processing post #{} from {}", route.id, post.post_id, vk_url) post_db_id = await self.db.save_or_update_post( + route_id=route.id, owner_id=post.owner_id, post_id=post.post_id, posted_at=post.date, text=post.text, raw_data=post.raw, ) - existing = await self.db.get_post(post.owner_id, post.post_id) or {} + existing = await self.db.get_post(route.id, post.owner_id, post.post_id) or {} tg_done = existing.get("tg_status") in ("published", "skipped") max_done = existing.get("max_status") in ("published", "skipped") @@ -97,17 +118,23 @@ class ServiceApp: # Download media once, shared by both platforms (skip entirely if # both are already done - nothing left to attach). if post.media and not (tg_done and max_done): - logger.info("Downloading {} media items for post #{}...", len(post.media), post.post_id) + logger.info("[{}] Downloading {} media items for post #{}...", route.id, len(post.media), post.post_id) processed_media = await media_processor.process_media_items(post.media) # Telegram - independent of MAX, so a MAX failure never blocks/retries this. - if tg_done: - logger.debug("Post #{} already resolved for Telegram ({}), skipping resend.", post.post_id, existing.get("tg_status")) + if not route.tg_enabled: + if not tg_done: + await self.db.update_tg_result(post_db_id=post_db_id, status="skipped") + result_summary["tg_status"] = "skipped" + elif tg_done: + logger.debug("[{}] Post #{} already resolved for Telegram ({}), skipping resend.", route.id, post.post_id, existing.get("tg_status")) else: try: tg_mids, tg_url = await self.tg_poster.post_to_telegram( raw_text=post.text, media_items=processed_media, + chat_id=info["tg_chat_id"], + thread_id=info["tg_thread_id"], vk_url=vk_url, ) await self.db.update_tg_result( @@ -120,23 +147,24 @@ class ServiceApp: result_summary["tg_url"] = tg_url except Exception as exc: err = str(exc) - logger.exception("Telegram post error for #{}: {}", post.post_id, exc) + logger.exception("[{}] Telegram post error for #{}: {}", route.id, post.post_id, exc) await self.db.update_tg_result(post_db_id=post_db_id, status="failed", error=err) result_summary["tg_status"] = "failed" result_summary["tg_error"] = err # MAX - independent of Telegram. - if not (settings.max_bot_token and settings.max_chat_id): + if not route.max_enabled or not (settings.max_bot_token and info["max_chat_id"]): if not max_done: await self.db.update_max_result(post_db_id=post_db_id, status="skipped") result_summary["max_status"] = "skipped" elif max_done: - logger.debug("Post #{} already resolved for MAX ({}), skipping resend.", post.post_id, existing.get("max_status")) + logger.debug("[{}] Post #{} already resolved for MAX ({}), skipping resend.", route.id, post.post_id, existing.get("max_status")) else: try: max_mids, max_url = await self.max_poster.post_to_max( raw_text=post.text, media_items=processed_media, + chat_id=info["max_chat_id"], vk_url=vk_url, ) await self.db.update_max_result( @@ -149,7 +177,7 @@ class ServiceApp: result_summary["max_url"] = max_url except Exception as exc: err = str(exc) - logger.exception("MAX post error for #{}: {}", post.post_id, exc) + logger.exception("[{}] MAX post error for #{}: {}", route.id, post.post_id, exc) await self.db.update_max_result(post_db_id=post_db_id, status="failed", error=err) result_summary["max_status"] = "failed" result_summary["max_error"] = err @@ -162,31 +190,27 @@ class ServiceApp: return result_summary - async def run_cycle(self) -> None: - if self.vk_group_owner_id is None: - return - - logger.info("Checking VK group '{}' for new posts...", self.vk_group_name) - posts_to_process: list[VKPost] = [] - cycle_error: Optional[str] = None + async def run_route_cycle(self, vk: VKClient, route: Route) -> dict[str, Any]: + info = self.route_info[route.id] + route_error: Optional[str] = None published_reports: list[dict[str, Any]] = [] + posts_to_process: list[VKPost] = [] try: - async with VKClient(rps=settings.vk_rate_limit_rps) as vk: - latest_posts = await vk.get_latest_posts( - owner_id=self.vk_group_owner_id, - count=settings.vk_check_count, - ) + logger.info("[{}] Checking VK group '{}' for new posts...", route.id, info["name"]) + latest_posts = await vk.get_latest_posts(owner_id=info["owner_id"], count=settings.vk_check_count) - is_initial_start = not await self.db.has_any_posts(self.vk_group_owner_id) + is_initial_start = not await self.db.has_any_posts(route.id, info["owner_id"]) if is_initial_start and latest_posts: logger.info( - "Initial start detected on fresh database. Applying bootstrap mode: '{}'", + "[{}] Initial start detected for this route. Applying bootstrap mode: '{}'", + route.id, settings.bootstrap_mode, ) if settings.bootstrap_mode == "skip_existing": for p in latest_posts: await self.db.mark_post_skipped( + route_id=route.id, owner_id=p.owner_id, post_id=p.post_id, posted_at=p.date, @@ -194,13 +218,14 @@ class ServiceApp: raw_data=p.raw, reason="initial_bootstrap_skip", ) - logger.info("Marked {} existing posts as already known. Only new future posts will be published.", len(latest_posts)) + logger.info("[{}] Marked {} existing posts as already known. Only new future posts will be published.", route.id, len(latest_posts)) latest_posts = [] elif settings.bootstrap_mode == "publish_latest_one": newest = max(latest_posts, key=lambda p: p.date) for p in latest_posts: if p.post_id != newest.post_id: await self.db.mark_post_skipped( + route_id=route.id, owner_id=p.owner_id, post_id=p.post_id, posted_at=p.date, @@ -209,30 +234,32 @@ class ServiceApp: reason="initial_bootstrap_skip", ) latest_posts = [newest] - logger.info("Bootstrap mode: keeping only the single newest post #{}", newest.post_id) + logger.info("[{}] Bootstrap mode: keeping only the single newest post #{}", route.id, newest.post_id) for p in latest_posts: if p.is_repost: - logger.debug("Skipping repost #{}", p.post_id) + logger.debug("[{}] Skipping repost #{}", route.id, p.post_id) continue - is_done = await self.db.is_post_processed(p.owner_id, p.post_id) + is_done = await self.db.is_post_processed(route.id, p.owner_id, p.post_id) if not is_done: posts_to_process.append(p) posts_to_process.sort(key=lambda p: p.date) if posts_to_process: - logger.info("Found {} new posts to publish", len(posts_to_process)) + logger.info("[{}] Found {} new posts to publish", route.id, len(posts_to_process)) for post in posts_to_process: - report = await self.process_new_post(post) + report = await self.process_new_post(route, info, post) published_reports.append(report) await asyncio.sleep(2.0) # gentle pacing between posts else: - logger.info("No new posts found.") + logger.info("[{}] No new posts found.", route.id) tg_success = sum(1 for r in published_reports if r.get("tg_status") == "published") max_success = sum(1 for r in published_reports if r.get("max_status") == "published") await self.db.record_run( + route_id=route.id, + route_name=info["name"], found_count=len(posts_to_process), tg_count=tg_success, max_count=max_success, @@ -240,31 +267,50 @@ class ServiceApp: ) except Exception as exc: - cycle_error = str(exc) - logger.exception("Error during parse cycle: {}", exc) + route_error = str(exc) + logger.exception("[{}] Error during route cycle: {}", route.id, exc) await self.db.record_run( + route_id=route.id, + route_name=info.get("name", route.id), found_count=0, tg_count=0, max_count=0, status="error", - error=cycle_error, + error=route_error, ) - # Report straight from this cycle's own results/error - no separate - # task reading the DB back, so there's no way for the two to drift - # out of sync (that was the actual bug the worker-split version had). + return { + "route_id": route.id, + "route_name": info["name"], + "vk_url": info["url"], + "reports": published_reports, + "error": route_error, + } + + async def run_cycle(self) -> None: + if not self.routes: + return + + all_results: list[dict[str, Any]] = [] + # Routes are processed strictly one after another (shared VK client, + # shared TG/MAX bot tokens) - no gather/create_task across routes, so + # flood control on any single upstream API stays global and correct. + async with VKClient(rps=settings.vk_rate_limit_rps) as vk: + for idx, route in enumerate(self.routes): + result = await self.run_route_cycle(vk, route) + all_results.append(result) + if idx < len(self.routes) - 1: + await asyncio.sleep(1.5) + + # Single aggregated report for the whole cycle, covering every route - + # avoids flooding admins with one message per route per cycle. if self.admin_notifier: - await self.admin_notifier.notify_cycle_result( - group_name=self.vk_group_name, - vk_url=self.vk_group_url, - found_posts=published_reports, - error=cycle_error, - ) + await self.admin_notifier.notify_cycle_result(all_results) async def run(self) -> None: await self.init() self.running = True - logger.info("Service started. Checking every {} minutes.", settings.check_interval_minutes) + logger.info("Service started. {} route(s) configured. Checking every {} minutes.", len(self.routes), settings.check_interval_minutes) while self.running: try: diff --git a/src/max_poster.py b/src/max_poster.py index 12fb689..2f05296 100644 --- a/src/max_poster.py +++ b/src/max_poster.py @@ -145,7 +145,6 @@ class MAXAPIClient: class MAXPoster: def __init__(self) -> None: self.token = settings.max_bot_token - self.chat_id = settings.max_chat_id self.api_base_url = settings.max_api_base_url self.message_limit = MAX_MESSAGE_LIMIT self.video_ready_attempts = settings.max_video_ready_attempts @@ -202,14 +201,14 @@ class MAXPoster: delay = min(delay * 1.5, 30.0) async def send_message_waiting_for_media( - self, client: MAXAPIClient, text: str, attachments: list[dict[str, Any]] + self, client: MAXAPIClient, chat_id: str, text: str, attachments: list[dict[str, Any]] ) -> dict[str, Any]: has_video = any(item.get("type") == "video" for item in attachments) attempts = self.video_ready_attempts if has_video else 1 delay = self.video_ready_delay_sec for attempt in range(1, attempts + 1): try: - return await client.send_message(self.chat_id, text, attachments=attachments or None) + return await client.send_message(chat_id, text, attachments=attachments or None) except MAXAPIError as exc: if exc.code != "attachment.not.ready" or attempt >= attempts: raise @@ -251,7 +250,7 @@ class MAXPoster: returned separately so send_post() can add a "watch via link" note instead of silently dropping them. """ - if not self.token or not self.chat_id: + if not self.token: return [], [] valid_media = [m for m in media_items if not m.is_link_only and m.local_path] if not valid_media: @@ -287,13 +286,14 @@ class MAXPoster: self, raw_text: str, attachments: list[dict[str, Any]], + chat_id: str, vk_url: Optional[str] = None, link_only_media: Optional[list[ProcessedMedia]] = None, oversized_videos: Optional[list[ProcessedMedia]] = None, ) -> tuple[list[str], Optional[str]]: """Post stage - assumes upload_media() already ran and every video attachment is confirmed ready.""" - if not self.token or not self.chat_id: + if not self.token or not chat_id: logger.warning("MAX bot token or chat ID is not set; skipping MAX post.") return [], None @@ -324,7 +324,7 @@ class MAXPoster: async with MAXAPIClient(self.token, self.api_base_url) as client: first_text = chunks[0] if chunks else "" - res = await self.send_message_waiting_for_media(client, first_text, media_groups[0]) + res = await self.send_message_waiting_for_media(client, chat_id, first_text, media_groups[0]) first_mid = self.message_id_from_response(res) if first_mid: @@ -336,14 +336,14 @@ class MAXPoster: if not group: continue await asyncio.sleep(1.0) - sub_res = await self.send_message_waiting_for_media(client, "", group) + sub_res = await self.send_message_waiting_for_media(client, chat_id, "", group) sub_mid = self.message_id_from_response(sub_res) if sub_mid: message_ids.append(sub_mid) for chunk in chunks[1:]: await asyncio.sleep(1.0) - sub_res = await client.send_message(self.chat_id, chunk) + sub_res = await client.send_message(chat_id, chunk) sub_mid = self.message_id_from_response(sub_res) if sub_mid: message_ids.append(sub_mid) @@ -355,14 +355,15 @@ class MAXPoster: self, raw_text: str, media_items: list[ProcessedMedia], + chat_id: str, vk_url: Optional[str] = None, ) -> tuple[list[str], Optional[str]]: """Back-compat convenience wrapper: upload_media() + send_post() in one call.""" - if not self.token or not self.chat_id: + if not self.token or not chat_id: logger.warning("MAX bot token or chat ID is not set; skipping MAX post.") return [], None link_only = [m for m in media_items if m.is_link_only] attachments, oversized = await self.upload_media(media_items) return await self.send_post( - raw_text, attachments, vk_url=vk_url, link_only_media=link_only, oversized_videos=oversized + raw_text, attachments, chat_id, vk_url=vk_url, link_only_media=link_only, oversized_videos=oversized ) diff --git a/src/routes.py b/src/routes.py new file mode 100644 index 0000000..ab10d49 --- /dev/null +++ b/src/routes.py @@ -0,0 +1,140 @@ +from __future__ import annotations +from loguru import logger + +import json +import re +from dataclasses import dataclass +from pathlib import Path +try: + from .config import settings +except (ImportError, ValueError): + from config import settings + + +_COMMENT_LINE_RE = re.compile(r"^\s*//.*$") + + +def _strip_comment_lines(raw_text: str) -> str: + """Allows '// like this' full-line comments in routes.json (plain JSON has no + comment syntax) so each donor->recipient route can be labeled inline. Only + strips lines whose trimmed content starts with '//' - never touches partial + lines, so a value containing '//' (e.g. a vk.ru URL) is always safe since it + only ever appears inside a quoted string, never at the start of a line.""" + return "\n".join( + line for line in raw_text.splitlines() if not _COMMENT_LINE_RE.match(line) + ) + + +@dataclass +class Route: + id: str + name: str + vk_source: str + tg_chat_id: str = "" + tg_enabled: bool = True + max_chat_id: str = "" + max_enabled: bool = True + + +def _legacy_route() -> Route: + return Route( + id="default", + name=settings.vk_source or "default", + vk_source=settings.vk_source, + tg_chat_id=settings.tg_chat_id, + tg_enabled=bool(settings.tg_chat_id), + max_chat_id=settings.max_chat_id, + max_enabled=bool(settings.max_chat_id), + ) + + +def _write_routes(path: Path, routes: list[Route]) -> None: + payload = [ + { + "id": r.id, + "name": r.name, + "vk_source": r.vk_source, + "tg_chat_id": r.tg_chat_id, + "tg_enabled": r.tg_enabled, + "max_chat_id": r.max_chat_id, + "max_enabled": r.max_enabled, + } + for r in routes + ] + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") + + +def load_routes() -> list[Route]: + """Loads donor->recipient routes from ROUTES_CONFIG_PATH (default data/routes.json). + + If the file doesn't exist yet, synthesizes a single route from the legacy + VK_SOURCE/TG_CHAT_ID/MAX_CHAT_ID env vars (route id 'default') and writes it + to disk, so routes.json becomes the single source of truth going forward + without breaking existing single-group deployments. + """ + path = settings.routes_path + if not path.exists(): + if not settings.vk_source: + raise ValueError( + f"No routes config found at {path} and no legacy VK_SOURCE configured - " + "set VK_SOURCE (single group) or create routes.json (multiple groups)." + ) + route = _legacy_route() + logger.info( + "No routes config found at {} - generating one from legacy .env settings " + "(route id='default'). Edit this file to add/change donor->recipient routes.", + path, + ) + _write_routes(path, [route]) + return [route] + + try: + raw = json.loads(_strip_comment_lines(path.read_text(encoding="utf-8"))) + except Exception as exc: + raise ValueError(f"Failed to parse routes config at {path}: {exc}") from exc + + if not isinstance(raw, list) or not raw: + raise ValueError(f"Routes config at {path} must be a non-empty JSON array") + + routes: list[Route] = [] + seen_ids: set[str] = set() + for idx, entry in enumerate(raw): + if not isinstance(entry, dict): + raise ValueError(f"Route #{idx} in {path} is not an object") + + route_id = str(entry.get("id") or "").strip() + vk_source = str(entry.get("vk_source") or "").strip() + if not route_id: + raise ValueError(f"Route #{idx} in {path} is missing required 'id'") + if not vk_source: + raise ValueError(f"Route '{route_id}' in {path} is missing required 'vk_source'") + if route_id in seen_ids: + raise ValueError(f"Duplicate route id '{route_id}' in {path}") + seen_ids.add(route_id) + + tg_chat_id = str(entry.get("tg_chat_id") or "").strip() + max_chat_id = str(entry.get("max_chat_id") or "").strip() + tg_enabled = bool(entry.get("tg_enabled", True)) and bool(tg_chat_id) + max_enabled = bool(entry.get("max_enabled", True)) and bool(max_chat_id) + + if not tg_enabled and not max_enabled: + logger.warning( + "Route '{}' has neither Telegram nor MAX enabled/configured - " + "it will be polled but nothing will be published.", + route_id, + ) + + routes.append( + Route( + id=route_id, + name=str(entry.get("name") or route_id), + vk_source=vk_source, + tg_chat_id=tg_chat_id, + tg_enabled=tg_enabled, + max_chat_id=max_chat_id, + max_enabled=max_enabled, + ) + ) + + return routes diff --git a/src/text_formatter.py b/src/text_formatter.py index 8692e96..25af74b 100644 --- a/src/text_formatter.py +++ b/src/text_formatter.py @@ -34,18 +34,6 @@ def clean_dividers(lines: list[str]) -> list[str]: return ["" if sep_pattern.match(line) else line for line in lines] -def strip_trailing_hashtags(text: str) -> str: - lines = text.rstrip().splitlines() - while lines and not lines[-1].strip(): - lines.pop() - if not lines: - return "" - parts = lines[-1].split() - if parts and all(part.startswith("#") for part in parts): - lines.pop() - return "\n".join(lines).rstrip() - - # Matches VK bracket markup: [club123|Name], [id123|Name], [public123|Name], # [event123|Name] or a raw-URL bracket link [https://example.com|Title] _VK_LINK_RE = re.compile( @@ -213,8 +201,7 @@ def format_post_text( footer = normalize_wrapper_text(settings.footer_text if footer is None else footer, parse_mode) tags = normalize_wrapper_text(settings.common_tags if tags is None else tags, parse_mode) - body = strip_trailing_hashtags(raw_text) - lines = clean_dividers(body.splitlines()) + lines = clean_dividers(raw_text.splitlines()) title_idx: Optional[int] = None for idx, line in enumerate(lines): diff --git a/src/tg_poster.py b/src/tg_poster.py index 5867844..b8a0469 100644 --- a/src/tg_poster.py +++ b/src/tg_poster.py @@ -48,8 +48,6 @@ class RichMessageUnavailable(RuntimeError): class TelegramPoster: def __init__(self) -> None: self.bot: Optional[Bot] = None - self.chat_id: int = 0 - self.thread_id: Optional[int] = None self.storage_chat_id: Optional[int] = None self.storage_thread_id: Optional[int] = None self.is_local_api: bool = False @@ -57,12 +55,11 @@ class TelegramPoster: self.message_limit: int = 4096 async def init(self) -> None: + """Chat destinations are no longer fixed here - each route supplies its own + chat_id/thread_id per call, since one bot posts into many chats/routes.""" if not settings.tg_bot_token: raise ValueError("TG_BOT_TOKEN is not configured") - if not settings.tg_chat_id: - raise ValueError("TG_CHAT_ID is not configured") - self.chat_id, self.thread_id = parse_topic(settings.tg_chat_id) if settings.tg_media_channel_id: try: self.storage_chat_id, self.storage_thread_id = parse_topic(settings.tg_media_channel_id) @@ -97,9 +94,9 @@ class TelegramPoster: if self.bot and self.bot.session: await self.bot.session.close() - def chat_kwargs(self, is_storage: bool = False) -> dict[str, Any]: - c_id = self.storage_chat_id if is_storage and self.storage_chat_id else self.chat_id - t_id = self.storage_thread_id if is_storage and self.storage_chat_id else self.thread_id + def chat_kwargs(self, chat_id: int, thread_id: Optional[int] = None, is_storage: bool = False) -> dict[str, Any]: + c_id = self.storage_chat_id if is_storage and self.storage_chat_id else chat_id + t_id = self.storage_thread_id if is_storage and self.storage_chat_id else thread_id kwargs: dict[str, Any] = {"chat_id": int(c_id)} if t_id: kwargs["message_thread_id"] = int(t_id) @@ -135,12 +132,12 @@ class TelegramPoster: await asyncio.sleep(delay) return await fn() - async def set_reaction(self, message_id: int) -> None: + async def set_reaction(self, chat_id: int, message_id: int) -> None: if not (settings.tg_auto_reaction_enabled and settings.tg_auto_reaction and self.bot): return try: await self.bot.set_message_reaction( - chat_id=self.chat_id, + chat_id=chat_id, message_id=message_id, reaction=[ReactionTypeEmoji(emoji=settings.tg_auto_reaction)], ) @@ -148,7 +145,7 @@ class TelegramPoster: logger.warning("Telegram auto reaction failed: {}", exc) async def upload_media_for_file_ids( - self, media_items: list[ProcessedMedia] + self, media_items: list[ProcessedMedia], chat_id: int, thread_id: Optional[int] = None ) -> dict[str, str]: """Uploads files to destination/storage to obtain tg_file_ids""" file_ids: dict[str, str] = {} @@ -175,19 +172,19 @@ class TelegramPoster: fs = FSInputFile(str(item.local_path)) if item.media_type == "photo": msg = await self.tg_retry_media( - lambda: self.bot.send_photo(photo=fs, **self.chat_kwargs(use_storage)) + lambda: self.bot.send_photo(photo=fs, **self.chat_kwargs(chat_id, thread_id, use_storage)) ) if msg.photo: file_ids[item.attachment_id] = msg.photo[-1].file_id else: msg = await self.tg_retry_media( - lambda: self.bot.send_video(video=fs, **self.chat_kwargs(use_storage)) + lambda: self.bot.send_video(video=fs, **self.chat_kwargs(chat_id, thread_id, use_storage)) ) if msg.video: file_ids[item.attachment_id] = msg.video.file_id else: msgs = await self.tg_retry_media( - lambda: self.bot.send_media_group(media=group, **self.chat_kwargs(use_storage)) + lambda: self.bot.send_media_group(media=group, **self.chat_kwargs(chat_id, thread_id, use_storage)) ) for item, msg in zip(valid_items, msgs): if item.media_type == "photo" and msg.photo: @@ -245,13 +242,15 @@ class TelegramPoster: "media": rich_media, } - async def send_rich_message(self, rich_message: dict[str, Any]) -> list[int]: + async def send_rich_message( + self, rich_message: dict[str, Any], chat_id: int, thread_id: Optional[int] = None + ) -> list[int]: data: dict[str, Any] = { - "chat_id": int(self.chat_id), + "chat_id": int(chat_id), "rich_message": rich_message, } - if self.thread_id: - data["message_thread_id"] = int(self.thread_id) + if thread_id: + data["message_thread_id"] = int(thread_id) # Always the cloud endpoint: this call only references already-uploaded # file_ids, no raw bytes cross the wire, so routing via the local Bot API @@ -285,7 +284,12 @@ class TelegramPoster: raise RichMessageUnavailable(description) async def send_media_post( - self, text: str, media_items: list[ProcessedMedia], file_ids: dict[str, str] + self, + text: str, + media_items: list[ProcessedMedia], + file_ids: dict[str, str], + chat_id: int, + thread_id: Optional[int] = None, ) -> list[int]: if not self.bot: raise RuntimeError("Bot not initialized") @@ -307,7 +311,7 @@ class TelegramPoster: if not input_media: # Text only post - return await self.send_text_post(text) + return await self.send_text_post(text, chat_id, thread_id) # Handle caption vs overflow text_chunks: list[str] = [] @@ -328,7 +332,7 @@ class TelegramPoster: photo=item["media"], caption=first_caption or None, parse_mode="HTML", - **self.chat_kwargs(), + **self.chat_kwargs(chat_id, thread_id), ) ) else: @@ -337,7 +341,7 @@ class TelegramPoster: video=item["media"], caption=first_caption or None, parse_mode="HTML", - **self.chat_kwargs(), + **self.chat_kwargs(chat_id, thread_id), ) ) message_ids.append(int(msg.message_id)) @@ -376,14 +380,14 @@ class TelegramPoster: text=c, parse_mode="HTML", disable_web_page_preview=True, - **self.chat_kwargs(), + **self.chat_kwargs(chat_id, thread_id), ) ) message_ids.append(int(msg.message_id)) return message_ids - async def send_text_post(self, text: str) -> list[int]: + async def send_text_post(self, text: str, chat_id: int, thread_id: Optional[int] = None) -> list[int]: if not self.bot: raise RuntimeError("Bot not initialized") chunks = split_message_chunks(text, self.message_limit) @@ -394,13 +398,15 @@ class TelegramPoster: text=c, parse_mode="HTML", disable_web_page_preview=True, - **self.chat_kwargs(), + **self.chat_kwargs(chat_id, thread_id), ) ) mids.append(int(msg.message_id)) return mids - async def upload_media(self, media_items: list[ProcessedMedia]) -> dict[str, str]: + async def upload_media( + self, media_items: list[ProcessedMedia], chat_id: int, thread_id: Optional[int] = None + ) -> dict[str, str]: """Upload stage: mint reusable file_ids from the storage channel, if configured. Returns {} when no storage channel is set up - send_post() then falls back to @@ -409,13 +415,15 @@ class TelegramPoster: valid_media = [m for m in media_items if not m.is_link_only and m.local_path] if not valid_media or not self.storage_chat_id: return {} - return await self.upload_media_for_file_ids(valid_media) + return await self.upload_media_for_file_ids(valid_media, chat_id, thread_id) async def send_post( self, raw_text: str, media_items: list[ProcessedMedia], file_ids: dict[str, str], + chat_id: int, + thread_id: Optional[int] = None, vk_url: Optional[str] = None, ) -> tuple[list[int], Optional[str]]: """ @@ -451,28 +459,30 @@ class TelegramPoster: rich_msg = self.build_rich_message(formatted_text, valid_media, file_ids) if rich_msg: try: - mids = await self.send_rich_message(rich_msg) - url = tg_message_url(self.chat_id, mids[0]) if mids else None + mids = await self.send_rich_message(rich_msg, chat_id, thread_id) + url = tg_message_url(chat_id, mids[0]) if mids else None logger.info("Sent Telegram rich message: {}", mids) if mids: - await self.set_reaction(mids[0]) + await self.set_reaction(chat_id, mids[0]) return mids, url except RichMessageUnavailable as exc: logger.warning("Telegram sendRichMessage failed: {}. Falling back to standard send.", exc) - mids = await self.send_media_post(formatted_text, valid_media, file_ids) - url = tg_message_url(self.chat_id, mids[0]) if mids else None + mids = await self.send_media_post(formatted_text, valid_media, file_ids, chat_id, thread_id) + url = tg_message_url(chat_id, mids[0]) if mids else None logger.info("Sent Telegram message: {}", mids) if mids: - await self.set_reaction(mids[0]) + await self.set_reaction(chat_id, mids[0]) return mids, url async def post_to_telegram( self, raw_text: str, media_items: list[ProcessedMedia], + chat_id: int, + thread_id: Optional[int] = None, vk_url: Optional[str] = None, ) -> tuple[list[int], Optional[str]]: """Back-compat convenience wrapper: upload_media() + send_post() in one call.""" - file_ids = await self.upload_media(media_items) - return await self.send_post(raw_text, media_items, file_ids, vk_url=vk_url) + file_ids = await self.upload_media(media_items, chat_id, thread_id) + return await self.send_post(raw_text, media_items, file_ids, chat_id, thread_id, vk_url=vk_url)