commit 41722c0ffce7ea813449d1cd33e3f6a23029d949 Author: exostring Date: Fri Aug 14 19:10:27 2026 +0500 Initial commit for RedAirsoft VK to TG and MAX poster diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..fa682e7 --- /dev/null +++ b/.env.example @@ -0,0 +1,88 @@ +# ========================================== +# VKontakte Settings +# ========================================== +# VK User/Service access token +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") +VK_SOURCE=redairsoft +# VK API Version +VK_API_VERSION=5.199 +# Number of wall posts to fetch each cycle (default 10) +VK_CHECK_COUNT=10 + +# ========================================== +# Telegram Settings +# ========================================== +# Telegram Bot Token (from @BotFather) +TG_BOT_TOKEN=123456789:ABCdefGHIjklMNOpqrSTUvwxYZ +# Destination Chat/Channel ID (e.g., "-1001234567890" or with topic thread "-1001234567890:42") +TG_CHAT_ID=-1001234567890 +# Optional TG Media Storage Channel (e.g. for generating permanent file_ids) +TG_MEDIA_CHANNEL_ID= +# Admin Telegram ID(s) for notifications & reports (single ID or comma-separated "123456,789012") +TG_ADMIN_IDS=123456789 + +# ========================================== +# Local Telegram Bot API (Optional) +# If TELEGRAM_API_ID & TELEGRAM_API_HASH are provided, local API server runs inside container automatically +# ========================================== +LOCAL_BOT_API_URL= +TELEGRAM_API_ID= +TELEGRAM_API_HASH= + +# ========================================== +# MAX Messenger Settings +# ========================================== +# MAX Bot Token +MAX_BOT_TOKEN=your_max_bot_token_here +# MAX Destination Chat ID +MAX_CHAT_ID=123456 +# MAX API Base URL +MAX_API_BASE_URL=https://platform-api2.max.ru +# MAX Auto Reaction emoji on published post +MAX_AUTO_REACTION=👍 +MAX_AUTO_REACTION_ENABLED=true + +# ========================================== +# Polling, Reporting & Maintenance +# ========================================== +# Check interval in minutes (default: 15) +CHECK_INTERVAL_MINUTES=15 +# Initial start behavior on fresh DB: +# "skip_existing" (default) - ignores old history, posts only new posts published after start +# "publish_latest_one" - publishes only the 1 latest post and skips older ones +# "publish_all" - publishes all wall posts fetched +BOOTSTRAP_MODE=skip_existing +# Send report to admin even if 0 new posts were found +REPORT_EMPTY_RUNS=false +# Send report to admin if an error occurred during checking/posting +REPORT_ON_ERROR=true +# Clean stale temp files older than X minutes (default: 30) +CACHE_MAX_AGE_MINUTES=30 + +# ========================================== +# Media & Video Limits +# ========================================== +# Max video size for Cloud TG API in MB (limit: 49-50 MB) +VIDEO_MAX_SIZE_MB_CLOUD=49 +# Max video size when using Local TG API in MB (supports up to 2000 MB) +VIDEO_MAX_SIZE_MB_LOCAL=500 +# Max video duration in seconds (default: 600s / 10 min) +VIDEO_MAX_DURATION_SEC=600 +# Max video resolution height (default: 720p) +VIDEO_MAX_HEIGHT=720 + +# ========================================== +# Post Formatting & Customization +# ========================================== +# Text to prepend before every post +HEADER_TEXT= +# Text to append after every post +FOOTER_TEXT= +# Common hashtags to attach (e.g., "#redairsoft #страйкбол") +COMMON_TAGS=#redairsoft #страйкбол +# Make the first line of the post bold +FORMAT_FIRST_LINE_BOLD=true + +# Logging Level (DEBUG, INFO, WARNING, ERROR) +LOG_LEVEL=INFO diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..84dc95a --- /dev/null +++ b/.gitignore @@ -0,0 +1,11 @@ +.env +data/ +cache/ +__pycache__/ +*.py[cod] +*$py.class +.venv/ +env/ +venv/ +*.log +.DS_Store diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..0331a5a --- /dev/null +++ b/Dockerfile @@ -0,0 +1,49 @@ +FROM aiogram/telegram-bot-api:latest AS telegram-bot-api + +FROM python:3.12-slim + +WORKDIR /app + +# Install system dependencies & ffmpeg +RUN apt-get update && apt-get install -y --no-install-recommends \ + ca-certificates \ + curl \ + ffmpeg \ + git \ + && rm -rf /var/lib/apt/lists/* + +# Install Russian Trusted CA certificates +RUN mkdir -p /usr/local/share/ca-certificates/russian-trusted \ + && curl -fsSLk https://gu-st.ru/content/lending/russian_trusted_root_ca_pem.crt \ + -o /usr/local/share/ca-certificates/russian-trusted/russian_trusted_root_ca_pem.crt \ + && curl -fsSLk https://gu-st.ru/content/lending/russian_trusted_sub_ca_pem.crt \ + -o /usr/local/share/ca-certificates/russian-trusted/russian_trusted_sub_ca_pem.crt \ + && update-ca-certificates + +# Copy local Telegram Bot API binary and dependencies +COPY --from=telegram-bot-api /usr/local/bin/telegram-bot-api /usr/local/bin/telegram-bot-api +COPY --from=telegram-bot-api /lib/ld-musl-x86_64.so.1 /lib/ld-musl-x86_64.so.1 +COPY --from=telegram-bot-api /usr/lib/libssl.so.3 /usr/lib/libssl.so.3 +COPY --from=telegram-bot-api /usr/lib/libcrypto.so.3 /usr/lib/libcrypto.so.3 +COPY --from=telegram-bot-api /usr/lib/libz.so.1 /usr/lib/libz.so.1 +COPY --from=telegram-bot-api /usr/lib/libstdc++.so.6 /usr/lib/libstdc++.so.6 +COPY --from=telegram-bot-api /usr/lib/libgcc_s.so.1 /usr/lib/libgcc_s.so.1 + +# Install Python requirements +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +# Copy source code and entrypoint +COPY src/ src/ +COPY docker-entrypoint.sh /usr/local/bin/poster-entrypoint +RUN chmod +x /usr/local/bin/poster-entrypoint \ + && mkdir -p /app/data /tmp/poster_cache /var/lib/telegram-bot-api /tmp/telegram-bot-api + +ENV PYTHONPATH=/app/src +ENV PYTHONUNBUFFERED=1 +ENV TELEGRAM_WORK_DIR=/var/lib/telegram-bot-api +ENV TELEGRAM_TEMP_DIR=/tmp/telegram-bot-api +ENV TELEGRAM_HTTP_PORT=8081 + +ENTRYPOINT ["poster-entrypoint"] +CMD ["python", "src/main.py"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..e15d322 --- /dev/null +++ b/README.md @@ -0,0 +1,83 @@ +# RedAirsoft VK to Telegram & MAX Messenger Poster + +Автоматизированный Docker-сервис для периодического парсинга новых постов из сообщества ВКонтакте и их одновременной публикации в **Telegram** и мессенджер **МАКС** (MAX Messenger) с отправкой отчётов администраторам в личные сообщения. + +--- + +## ⚡ Особенности и возможности + +1. **Telegram Rich Messages (современный формат) & Legacy fallback**: + - Автоматическая сборка Rich Messages с ``, инлайн-фото/видео и форматированным текстом. + - Поддержка облачного Telegram Bot API и локального `telegram-bot-api` (для видео больше 50 МБ). + - При недоступности формата Rich Message — автоматический переход на классические медиагруппы (`sendMediaGroup` / `sendPhoto` / `sendVideo`). +2. **Надёжная интеграция с Мессенджером МАКС**: + - Загрузка изображений и видео через `/uploads?type=...`. + - Полный цикл поллинга готовности видео (`/videos/{token}`) перед отправкой сообщения. + - Обработка `attachment.not.ready` и рейтов (429 Retry-After). + - Автоматическая простановка реакций (например, 👍) на первый опубликованный пост. + - Чанкинг длинных текстов (до 4000 символов). +3. **Безопасная очистка кэша и загрузок**: + - Немедленное удаление временных файлов после обработки каждого поста. + - Фоновый процесс очистки старых «хвостов» (`cleaner.py`), который **никогда** не удаляет файлы, находящиеся в процессе скачивания или отправки (благодаря реестру активных локов). +4. **Отчёты администраторам в ЛС Telegram**: + - Поддержка одного или списка Telegram ID через запятую (`TG_ADMIN_IDS=123456,789012`). + - Отправка ссылок на оригинал в ВК, опубликованный пост в Telegram и МАКС. +5. **Сохранение состояния (SQLite)**: + - База данных в папке `data/poster.db` (монтируется в Docker volume). + - Исключает повторную публикацию уже обработанных постов при перезапуске контейнера. + +--- + +## 🛠 Быстрый старт + +### 1. Клонирование и настройка окружения + +Скопируйте пример файла конфигурации: +```bash +cp .env.example .env +``` + +Заполните переменные в файле `.env`: +- `VK_ACCESS_TOKEN`: ваш токен ВКонтакте. +- `VK_SOURCE`: ссылка, короткое имя или ID группы ВК (например, `redairsoft` или `https://vk.com/redairsoft`). +- `TG_BOT_TOKEN`: токен бота Telegram (от `@BotFather`). +- `TG_CHAT_ID`: ID канала/группы Telegram (например, `-1001234567890`). +- `TG_ADMIN_IDS`: ID администраторов через запятую для отчётов. +- `MAX_BOT_TOKEN`: токен бота в мессенджере МАКС. +- `MAX_CHAT_ID`: ID чата в МАКС. + +### 2. Запуск через Docker Compose + +Запустите контейнер в фоновом режиме: +```bash +docker compose up -d --build +``` + +Просмотр логов: +```bash +docker compose logs -f +``` + +--- + +## 📁 Структура проекта + +``` +├── src/ +│ ├── config.py # Загрузка и валидация настроек из .env +│ ├── database.py # SQLite хранилище опубликованных постов +│ ├── vk_client.py # Клиент VK API (wall.get, resolve, media extraction) +│ ├── media_processor.py # Загрузчик фото/видео (yt-dlp, ffmpeg) с защитой активных файлов +│ ├── text_formatter.py # Форматирование текста (HTML, заголовки, хештеги) +│ ├── tg_poster.py # Публикация в Telegram (Rich Message + Legacy) +│ ├── max_poster.py # Публикация в MAX Messenger (Uploads + Messages + Reactions) +│ ├── cleaner.py # Фоновая очистка зависшего кэша +│ ├── admin_notifier.py # Отправка отчетов администраторам +│ └── main.py # Главный цикл оркестрации +├── Dockerfile # Мультистейдж образ с ffmpeg и сертификатами +├── docker-compose.yml # Конфигурация Docker Compose +├── docker-entrypoint.sh # Скрипт запуска и опционального поднятия local bot api +├── requirements.txt # Зависимости Python +├── .env.example # Шаблон переменных окружения +└── README.md +``` diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..8900fbc --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,15 @@ +services: + vk-poster: + build: + context: . + dockerfile: Dockerfile + container_name: redairsoft-vk-poster + restart: unless-stopped + env_file: + - .env + volumes: + - ./data:/app/data + - ./cache:/tmp/poster_cache + # Port 8081 can be exposed if external access to local Telegram Bot API is desired + # ports: + # - "8081:8081" diff --git a/docker-entrypoint.sh b/docker-entrypoint.sh new file mode 100644 index 0000000..90585a5 --- /dev/null +++ b/docker-entrypoint.sh @@ -0,0 +1,36 @@ +#!/usr/bin/env sh +set -eu + +tg_pid="" + +stop_children() { + if [ -n "$tg_pid" ] && kill -0 "$tg_pid" 2>/dev/null; then + echo "Stopping local telegram-bot-api (PID $tg_pid)..." + kill "$tg_pid" + wait "$tg_pid" 2>/dev/null || true + fi +} + +trap stop_children INT TERM EXIT + +if [ -n "${TELEGRAM_API_ID:-}" ] && [ -n "${TELEGRAM_API_HASH:-}" ]; then + mkdir -p "${TELEGRAM_WORK_DIR:-/var/lib/telegram-bot-api}" "${TELEGRAM_TEMP_DIR:-/tmp/telegram-bot-api}" + telegram-bot-api \ + --api-id="${TELEGRAM_API_ID}" \ + --api-hash="${TELEGRAM_API_HASH}" \ + --dir="${TELEGRAM_WORK_DIR:-/var/lib/telegram-bot-api}" \ + --temp-dir="${TELEGRAM_TEMP_DIR:-/tmp/telegram-bot-api}" \ + --http-port="${TELEGRAM_HTTP_PORT:-8081}" \ + --local & + tg_pid="$!" + echo "Started local Telegram Bot API server on 127.0.0.1:${TELEGRAM_HTTP_PORT:-8081}" + + # Auto configure local bot api url if not explicitly defined + if [ -z "${LOCAL_BOT_API_URL:-}" ]; then + export LOCAL_BOT_API_URL="http://127.0.0.1:${TELEGRAM_HTTP_PORT:-8081}" + fi +else + echo "TELEGRAM_API_ID / TELEGRAM_API_HASH are not set; local Telegram Bot API will not be started." +fi + +exec "$@" diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..1b45bc2 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,7 @@ +aiogram==3.21.0 +aiohttp==3.12.13 +aiosqlite==0.21.0 +loguru==0.7.3 +pydantic-settings==2.10.1 +Pillow==11.3.0 +yt-dlp>=2026.1.1 diff --git a/src/admin_notifier.py b/src/admin_notifier.py new file mode 100644 index 0000000..5e137f8 --- /dev/null +++ b/src/admin_notifier.py @@ -0,0 +1,102 @@ +from __future__ import annotations + +import asyncio +from datetime import datetime +from typing import Any, Optional +from aiogram import Bot +from loguru import logger +from .config import settings + + +class AdminNotifier: + def __init__(self, bot: Bot) -> None: + self.bot = bot + self.admin_ids = settings.admin_id_list + + async def send_to_all(self, text: str) -> None: + 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") + + if error: + text = ( + f"⚠️ [VK Poster Alert] Ошибка при проверке группы\n\n" + f"🏷 Группа: {group_name} (VK)\n" + f"⏱ Время: {now_str}\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", + "Список публикаций:", + ] + + for idx, p in enumerate(found_posts, 1): + post_id = p.get("vk_post_id") + vk_post_url = p.get("vk_post_url") + tg_url = p.get("tg_url") + max_url = p.get("max_url") + tg_status = p.get("tg_status") + max_status = p.get("max_status") + tg_err = p.get("tg_error") + max_err = p.get("max_error") + + lines.append(f"\n{idx}. Пост #{post_id}") + lines.append(f"• Оригинал в VK") + + if tg_status == "published": + tg_link = f"Ссылка" if tg_url else "Опубликовано" + lines.append(f"• Telegram: ✅ {tg_link}") + elif tg_err: + lines.append(f"• Telegram: ❌ Ошибка: {tg_err[:100]}") + + if max_status == "published": + max_link = f"Ссылка" if max_url else "Опубликовано" + lines.append(f"• MAX: ✅ {max_link}") + elif max_err: + lines.append(f"• MAX: ❌ Ошибка: {max_err[:100]}") + + await self.send_to_all("\n".join(lines)) diff --git a/src/cleaner.py b/src/cleaner.py new file mode 100644 index 0000000..0346884 --- /dev/null +++ b/src/cleaner.py @@ -0,0 +1,57 @@ +from __future__ import annotations + +import asyncio +import os +import time +from pathlib import Path +from loguru import logger +from .config import settings +from .media_processor import is_path_active + + +async def cleanup_stale_cache(max_age_minutes: Optional[int] = None) -> int: + """ + Deletes files from the cache directory that are older than max_age_minutes, + strictly skipping any files that are currently active in downloads/processing. + """ + if max_age_minutes is None: + max_age_minutes = settings.cache_max_age_minutes + + cache_dir = settings.cache_path + if not cache_dir.exists(): + return 0 + + threshold = time.time() - (max_age_minutes * 60) + removed_count = 0 + + for entry in cache_dir.iterdir(): + if not entry.is_file(): + continue + + try: + # Strictly check if file is currently registered as actively used + if await is_path_active(entry): + continue + + stat = entry.stat() + if stat.st_mtime < threshold: + entry.unlink(missing_ok=True) + removed_count += 1 + logger.debug("Removed stale cache file: {}", entry.name) + except Exception as exc: + logger.warning("Error during cache file inspection/cleanup for {}: {}", entry.name, exc) + + if removed_count > 0: + logger.info("Cleaned up {} stale cache files older than {}m", removed_count, max_age_minutes) + return removed_count + + +async def run_cleaner_loop(interval_minutes: int = 15) -> None: + """Background task running periodic cache cleanup.""" + logger.info("Cache cleanup worker started (interval: {}m)", interval_minutes) + while True: + try: + await cleanup_stale_cache() + except Exception as exc: + logger.warning("Error in periodic cache cleaner: {}", exc) + await asyncio.sleep(interval_minutes * 60) diff --git a/src/config.py b/src/config.py new file mode 100644 index 0000000..31efbcc --- /dev/null +++ b/src/config.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +import os +from pathlib import Path +from typing import Any +from pydantic_settings import BaseSettings, SettingsConfigDict + + +class Settings(BaseSettings): + model_config = SettingsConfigDict( + env_file=".env", + env_file_encoding="utf-8", + extra="ignore", + ) + + # VK Settings + vk_access_token: str = "" + vk_source: str = "" # 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 + + # Telegram Poster Settings + tg_bot_token: str = "" + tg_chat_id: str = "" # Destination chat/channel, 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" + telegram_api_id: str = "" + telegram_api_hash: str = "" + + # MAX Messenger Settings + max_bot_token: str = "" + max_chat_id: str = "" # Destination chat ID in MAX + max_api_base_url: str = "https://platform-api2.max.ru" + max_auto_reaction: str = "👍" + max_auto_reaction_enabled: bool = True + max_reaction_path_template: str = "/messages/{message_id}/reactions" + max_video_ready_attempts: int = 6 + max_video_ready_delay_sec: float = 8.0 + + # Polling, Worker & Bootstrap Settings + check_interval_minutes: int = 15 + report_empty_runs: bool = False + report_on_error: bool = True + # bootstrap_mode: "skip_existing" (default: ignores old historical posts on first start, only tracks new posts), + # "publish_latest_one" (publishes only the single latest post and skips older ones), + # "publish_all" (publishes all found posts) + bootstrap_mode: str = "skip_existing" + + # Database & Storage + database_path: str = "data/poster.db" + cache_dir: str = "/tmp/poster_cache" + cache_max_age_minutes: int = 30 + + # Media & Video Limits + video_max_size_mb_cloud: int = 49 + video_max_size_mb_local: int = 500 + video_max_duration_sec: int = 600 + video_max_height: int = 720 + media_download_timeout_sec: int = 60 + yt_dlp_timeout_sec: int = 300 + + # Text Styling & Decoration + header_text: str = "" + footer_text: str = "" + common_tags: str = "" + format_first_line_bold: bool = True + + # Logging + log_level: str = "INFO" + + @property + def admin_id_list(self) -> list[int]: + ids: list[int] = [] + for raw in (self.tg_admin_ids or "").split(","): + raw = raw.strip() + if raw.lstrip("-").isdigit(): + ids.append(int(raw)) + return ids + + @property + def cache_path(self) -> Path: + p = Path(self.cache_dir).resolve() + p.mkdir(parents=True, exist_ok=True) + return p + + @property + def db_path(self) -> Path: + p = Path(self.database_path).resolve() + p.parent.mkdir(parents=True, exist_ok=True) + return p + + +settings = Settings() diff --git a/src/database.py b/src/database.py new file mode 100644 index 0000000..0ea7ca4 --- /dev/null +++ b/src/database.py @@ -0,0 +1,204 @@ +from __future__ import annotations + +import json +from datetime import datetime +from typing import Any, Optional +import aiosqlite +from loguru import logger +from .config import settings + + +class Database: + def __init__(self, db_path: Optional[str] = None) -> None: + self.db_path = str(settings.db_path if db_path is None else db_path) + + async def init(self) -> None: + logger.info("Initializing database at {}", self.db_path) + async with aiosqlite.connect(self.db_path) as db: + await db.execute("PRAGMA journal_mode=WAL;") + await db.execute( + """ + CREATE TABLE IF NOT EXISTS posts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + 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(vk_owner_id, vk_post_id) + ); + """ + ) + await db.execute( + """ + CREATE TABLE IF NOT EXISTS publication_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + checked_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + found_count INTEGER DEFAULT 0, + published_tg_count INTEGER DEFAULT 0, + published_max_count INTEGER DEFAULT 0, + status TEXT DEFAULT 'ok', + error TEXT + ); + """ + ) + await db.commit() + + async def has_any_posts(self, owner_id: int) -> bool: + async with aiosqlite.connect(self.db_path) as db: + cursor = await db.execute( + "SELECT 1 FROM posts WHERE vk_owner_id = ? LIMIT 1", + (owner_id,), + ) + row = await cursor.fetchone() + return bool(row) + + async def mark_post_skipped( + self, + owner_id: int, + post_id: int, + posted_at: int, + text: str, + raw_data: dict[str, Any], + reason: str = "bootstrap_initial_skip", + ) -> None: + raw_json = json.dumps(raw_data, ensure_ascii=False) + async with aiosqlite.connect(self.db_path) 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 + tg_status = 'skipped', + max_status = 'skipped'; + """, + (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 with aiosqlite.connect(self.db_path) as db: + db.row_factory = aiosqlite.Row + cursor = await db.execute( + "SELECT id, tg_status, max_status FROM posts WHERE vk_owner_id = ? AND vk_post_id = ?", + (owner_id, post_id), + ) + row = await cursor.fetchone() + if not row: + return False + # If already published on both or marked skipped, it's processed + return bool( + row["tg_status"] in ("published", "skipped") + and row["max_status"] in ("published", "skipped") + ) + + async def get_post(self, owner_id: int, post_id: int) -> Optional[dict[str, Any]]: + async with aiosqlite.connect(self.db_path) 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), + ) + row = await cursor.fetchone() + return dict(row) if row else None + + async def save_or_update_post( + self, + owner_id: int, + post_id: int, + posted_at: int, + text: str, + raw_data: dict[str, Any], + ) -> int: + raw_json = json.dumps(raw_data, ensure_ascii=False) + async with aiosqlite.connect(self.db_path) 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 + text = excluded.text, + raw_json = excluded.raw_json + RETURNING id; + """, + (owner_id, post_id, posted_at, text, raw_json), + ) + row = await cursor.fetchone() + await db.commit() + return int(row[0]) if row else 0 + + async def update_tg_result( + self, + post_db_id: int, + status: str, + message_ids: Optional[list[int]] = None, + url: Optional[str] = None, + error: Optional[str] = None, + ) -> None: + msg_str = ",".join(str(m) for m in message_ids) if message_ids else None + async with aiosqlite.connect(self.db_path) as db: + await db.execute( + """ + UPDATE posts + SET tg_status = ?, + tg_message_ids = COALESCE(?, tg_message_ids), + tg_url = COALESCE(?, tg_url), + tg_error = ?, + published_at = CASE WHEN ? = 'published' THEN CURRENT_TIMESTAMP ELSE published_at END + WHERE id = ?; + """, + (status, msg_str, url, error, status, post_db_id), + ) + await db.commit() + + async def update_max_result( + self, + post_db_id: int, + status: str, + message_ids: Optional[list[str]] = None, + url: Optional[str] = None, + error: Optional[str] = None, + ) -> None: + msg_str = ",".join(message_ids) if message_ids else None + async with aiosqlite.connect(self.db_path) as db: + await db.execute( + """ + UPDATE posts + SET max_status = ?, + max_message_ids = COALESCE(?, max_message_ids), + max_url = COALESCE(?, max_url), + max_error = ?, + published_at = CASE WHEN ? = 'published' THEN CURRENT_TIMESTAMP ELSE published_at END + WHERE id = ?; + """, + (status, msg_str, url, error, status, post_db_id), + ) + await db.commit() + + async def record_run( + self, + found_count: int, + tg_count: int, + max_count: int, + status: str = "ok", + error: Optional[str] = None, + ) -> None: + async with aiosqlite.connect(self.db_path) as db: + await db.execute( + """ + INSERT INTO publication_runs (found_count, published_tg_count, published_max_count, status, error) + VALUES (?, ?, ?, ?, ?); + """, + (found_count, tg_count, max_count, status, error), + ) + await db.commit() diff --git a/src/main.py b/src/main.py new file mode 100644 index 0000000..4ce6d16 --- /dev/null +++ b/src/main.py @@ -0,0 +1,306 @@ +from __future__ import annotations + +import asyncio +import os +import signal +import sys +from typing import Any, Optional +from loguru import logger +from .admin_notifier import AdminNotifier +from .cleaner import cleanup_stale_cache, run_cleaner_loop +from .config import settings +from .database import Database +from .max_poster import MAXPoster +from .media_processor import MediaProcessor +from .tg_poster import TelegramPoster +from .vk_client import VKClient, VKPost + + +class ServiceApp: + def __init__(self) -> None: + self.db = Database() + 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.running = False + self.cleaner_task: Optional[asyncio.Task] = None + + async def init(self) -> None: + logger.remove() + logger.add( + sys.stdout, + level=settings.log_level, + format="{time:YYYY-MM-DD HH:mm:ss} | {level: <8} | {name}:{line} - {message}", + ) + logger.info("Initializing VK to TG & MAX Poster Service...") + + # Initialize SQLite DB + await self.db.init() + + # Initialize Telegram Poster + await self.tg_poster.init() + if self.tg_poster.bot: + self.admin_notifier = AdminNotifier(self.tg_poster.bot) + + # Resolve VK Group + if not settings.vk_source: + raise ValueError("VK_SOURCE is not set in configuration") + + async with VKClient() 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) + + # Start periodic background cache cleaner + self.cleaner_task = asyncio.create_task(run_cleaner_loop(interval_minutes=15)) + + async def process_new_post(self, 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) + + # Save to database + post_db_id = await self.db.save_or_update_post( + owner_id=post.owner_id, + post_id=post.post_id, + posted_at=post.date, + text=post.text, + raw_data=post.raw, + ) + + media_processor = MediaProcessor(is_local_tg_api=self.tg_poster.is_local_api) + processed_media = [] + + result_summary: dict[str, Any] = { + "vk_post_id": post.post_id, + "vk_post_url": vk_url, + "tg_status": "pending", + "tg_url": None, + "tg_error": None, + "max_status": "pending", + "max_url": None, + "max_error": None, + } + + try: + # 1. Download/extract media + if post.media: + logger.info("Downloading {} media items for post #{}...", len(post.media), post.post_id) + processed_media = await media_processor.process_media_items(post.media) + + # 2. Publish to Telegram + try: + tg_mids, tg_url = await self.tg_poster.post_to_telegram( + raw_text=post.text, + media_items=processed_media, + vk_url=vk_url, + ) + await self.db.update_tg_result( + post_db_id=post_db_id, + status="published", + message_ids=tg_mids, + url=tg_url, + ) + result_summary["tg_status"] = "published" + result_summary["tg_url"] = tg_url + except Exception as exc: + err = str(exc) + logger.exception("Telegram post error for #{}: {}", 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 + + # 3. Publish to MAX Messenger + if settings.max_bot_token and settings.max_chat_id: + try: + max_mids, max_url = await self.max_poster.post_to_max( + raw_text=post.text, + media_items=processed_media, + vk_url=vk_url, + ) + await self.db.update_max_result( + post_db_id=post_db_id, + status="published", + message_ids=max_mids, + url=max_url, + ) + result_summary["max_status"] = "published" + result_summary["max_url"] = max_url + except Exception as exc: + err = str(exc) + logger.exception("MAX post error for #{}: {}", 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 + else: + await self.db.update_max_result(post_db_id=post_db_id, status="skipped") + result_summary["max_status"] = "skipped" + + finally: + # Immediate cleanup of temporary media files + if processed_media: + await media_processor.cleanup(processed_media) + + 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 + published_reports: list[dict[str, Any]] = [] + + try: + async with VKClient() as vk: + latest_posts = await vk.get_latest_posts( + owner_id=self.vk_group_owner_id, + count=settings.vk_check_count, + ) + + # Check if this is the very first run on an empty database + is_initial_start = not await self.db.has_any_posts(self.vk_group_owner_id) + if is_initial_start and latest_posts: + logger.info( + "Initial start detected on fresh database. Applying bootstrap mode: '{}'", + settings.bootstrap_mode, + ) + if settings.bootstrap_mode == "skip_existing": + for p in latest_posts: + await self.db.mark_post_skipped( + owner_id=p.owner_id, + post_id=p.post_id, + posted_at=p.date, + text=p.text, + 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)) + latest_posts = [] + elif settings.bootstrap_mode == "publish_latest_one": + # Mark all except the single latest post as skipped + 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( + owner_id=p.owner_id, + post_id=p.post_id, + posted_at=p.date, + text=p.text, + raw_data=p.raw, + reason="initial_bootstrap_skip", + ) + latest_posts = [newest] + logger.info("Bootstrap mode: keeping only the single newest post #{}", newest.post_id) + + # Filter out processed posts and sort oldest -> newest + for p in latest_posts: + if p.is_repost: + logger.debug("Skipping repost #{}", p.post_id) + continue + is_done = await self.db.is_post_processed(p.owner_id, p.post_id) + if not is_done: + posts_to_process.append(p) + + # Sort chronological (oldest to newest) + posts_to_process.sort(key=lambda p: p.date) + + if posts_to_process: + logger.info("Found {} new posts to publish", len(posts_to_process)) + for post in posts_to_process: + report = await self.process_new_post(post) + published_reports.append(report) + await asyncio.sleep(2.0) # Pause between posts + else: + logger.info("No new posts found.") + + 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( + found_count=len(posts_to_process), + tg_count=tg_success, + max_count=max_success, + status="ok", + ) + + except Exception as exc: + cycle_error = str(exc) + logger.exception("Error during parse cycle: {}", exc) + await self.db.record_run( + found_count=0, + tg_count=0, + max_count=0, + status="error", + error=cycle_error, + ) + + # Notify admins + 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, + ) + + async def run(self) -> None: + await self.init() + self.running = True + logger.info("Service started. Checking every {} minutes.", settings.check_interval_minutes) + + while self.running: + try: + await self.run_cycle() + except Exception as exc: + logger.exception("Unexpected error in main loop: {}", exc) + + # Sleep between cycles + sleep_seconds = settings.check_interval_minutes * 60 + logger.debug("Sleeping for {} seconds until next check...", sleep_seconds) + for _ in range(sleep_seconds): + if not self.running: + break + await asyncio.sleep(1.0) + + async def close(self) -> None: + logger.info("Stopping service...") + self.running = False + if self.cleaner_task: + self.cleaner_task.cancel() + await self.tg_poster.close() + # Clean any remaining stale files on shutdown + await cleanup_stale_cache(max_age_minutes=0) + logger.info("Service stopped cleanly.") + + +async def main() -> None: + app = ServiceApp() + loop = asyncio.get_running_loop() + + def handle_signal(): + logger.info("Signal received, stopping...") + asyncio.create_task(app.close()) + + for sig in (signal.SIGINT, signal.SIGTERM): + try: + loop.add_signal_handler(sig, handle_signal) + except NotImplementedError: + # Signal handlers not implemented on Windows event loop for non-main threads + pass + + try: + await app.run() + finally: + await app.close() + + +if __name__ == "__main__": + try: + asyncio.run(main()) + except (KeyboardInterrupt, SystemExit): + logger.info("Application exited.") diff --git a/src/max_poster.py b/src/max_poster.py new file mode 100644 index 0000000..1bde739 --- /dev/null +++ b/src/max_poster.py @@ -0,0 +1,302 @@ +from __future__ import annotations + +import asyncio +import json +import mimetypes +from pathlib import Path +from typing import Any, Optional +from urllib.parse import quote +import aiohttp +from loguru import logger +from .config import settings +from .media_processor import ProcessedMedia +from .text_formatter import format_post_text, split_message_chunks + +MAX_MESSAGE_LIMIT = 4000 +MAX_MEDIA_ITEMS = 10 + + +class MAXAPIError(RuntimeError): + def __init__(self, message: str, status: Optional[int] = None, code: str = "") -> None: + super().__init__(message) + self.status = status + self.code = code + + +class MAXAPIClient: + def __init__( + self, + token: str, + base_url: str = "https://platform-api2.max.ru", + timeout_sec: int = 120, + max_attempts: int = 4, + retry_backoff_max_sec: int = 30, + ) -> None: + self.token = token + self.base_url = base_url.rstrip("/") + self.timeout_sec = timeout_sec + self.max_attempts = max_attempts + self.retry_backoff_max_sec = retry_backoff_max_sec + self.session: Optional[aiohttp.ClientSession] = None + + async def __aenter__(self) -> "MAXAPIClient": + timeout = aiohttp.ClientTimeout(total=self.timeout_sec) + self.session = aiohttp.ClientSession( + timeout=timeout, headers={"Authorization": self.token} + ) + return self + + async def __aexit__(self, exc_type, exc, tb) -> None: + if self.session: + await self.session.close() + + async def request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]: + if not self.session: + raise RuntimeError("MAX session is not initialized") + url = f"{self.base_url}{path}" + last_error = "" + for attempt in range(1, self.max_attempts + 1): + try: + async with self.session.request(method, url, **kwargs) as resp: + text = await resp.text() + try: + data = json.loads(text) if text else {} + except json.JSONDecodeError: + data = {"raw": text} + if 200 <= resp.status < 300: + return data + + retry_after = resp.headers.get("Retry-After") + message = data.get("message") if isinstance(data, dict) else text + code = data.get("code") if isinstance(data, dict) else "" + last_error = f"MAX API {resp.status} {code}: {message or text}" + + if resp.status == 429 and retry_after: + await asyncio.sleep(float(retry_after) + 0.5) + continue + if code == "attachment.not.ready" and attempt < self.max_attempts: + await asyncio.sleep(min(2 ** attempt, self.retry_backoff_max_sec)) + continue + if resp.status < 500: + raise MAXAPIError(last_error, resp.status, str(code or "")) + except Exception as exc: + last_error = str(exc) + if attempt >= self.max_attempts: + raise + await asyncio.sleep(min(2 ** attempt, self.retry_backoff_max_sec)) + raise RuntimeError(last_error or "MAX API request failed") + + async def send_message( + self, + chat_id: str, + text: str, + attachments: Optional[list[dict[str, Any]]] = None, + ) -> dict[str, Any]: + payload: dict[str, Any] = {"text": text, "notify": True, "format": "html"} + if attachments: + payload["attachments"] = attachments + return await self.request("POST", f"/messages?chat_id={quote(str(chat_id))}", json=payload) + + async def get_message(self, message_id: str) -> dict[str, Any]: + return await self.request("GET", f"/messages/{quote(message_id, safe='')}") + + async def get_video_info(self, token: str) -> dict[str, Any]: + return await self.request("GET", f"/videos/{quote(token, safe='')}") + + async def get_upload_url(self, media_type: str) -> dict[str, Any]: + upload_type = "video" if media_type == "video" else "image" + return await self.request("POST", f"/uploads?type={upload_type}") + + async def upload_file(self, upload_url: str, path: Path) -> dict[str, Any]: + if not self.session: + raise RuntimeError("MAX session is not initialized") + form = aiohttp.FormData() + content_type = mimetypes.guess_type(str(path))[0] or "application/octet-stream" + with path.open("rb") as fh: + form.add_field("data", fh, filename=path.name, content_type=content_type) + async with self.session.post(upload_url, data=form) as resp: + text = await resp.text() + try: + data = json.loads(text) if text else {} + except json.JSONDecodeError: + data = {"raw": text} + if resp.status < 200 or resp.status >= 300: + raise RuntimeError(f"MAX upload {resp.status}: {data}") + return data + + async def react(self, path_template: str, message_id: str, reaction: str) -> None: + path = path_template.format(message_id=message_id) + payload = {"reaction": reaction} + await self.request("POST", path, json=payload) + + +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 + self.video_ready_delay_sec = settings.max_video_ready_delay_sec + + def message_id_from_response(self, data: dict[str, Any]) -> str: + candidates = [ + data.get("id"), + data.get("mid"), + (data.get("message") or {}).get("id") if isinstance(data.get("message"), dict) else None, + (data.get("message") or {}).get("mid") if isinstance(data.get("message"), dict) else None, + ] + message = data.get("message") if isinstance(data.get("message"), dict) else {} + body = message.get("body") if isinstance(message.get("body"), dict) else {} + candidates.extend([body.get("id"), body.get("mid"), body.get("message_id")]) + return next((str(v) for v in candidates if v), "") + + def message_url_from_response(self, data: dict[str, Any]) -> Optional[str]: + candidates = [data.get("url")] + message = data.get("message") if isinstance(data.get("message"), dict) else {} + body = message.get("body") if isinstance(message.get("body"), dict) else {} + candidates.extend([message.get("url"), body.get("url")]) + return next((str(v).strip() for v in candidates if str(v or "").strip()), None) + + def video_tokens(self, attachments: list[dict[str, Any]]) -> list[str]: + tokens: list[str] = [] + for item in attachments: + if item.get("type") != "video": + continue + payload = item.get("payload") if isinstance(item.get("payload"), dict) else {} + token = str(payload.get("token") or "").strip() + if token: + tokens.append(token) + return tokens + + async def wait_for_videos(self, client: MAXAPIClient, attachments: list[dict[str, Any]]) -> None: + tokens = self.video_tokens(attachments) + if not tokens: + return + delay = self.video_ready_delay_sec + for attempt in range(1, self.video_ready_attempts + 1): + pending: list[str] = [] + for token in tokens: + info = await client.get_video_info(token) + if not info.get("urls"): + pending.append(token) + if not pending: + return + if attempt >= self.video_ready_attempts: + logger.warning("MAX video still processing after {} attempts, proceeding...", attempt) + return + logger.info("MAX video processing... waiting {}s (attempt {}/{})", delay, attempt, self.video_ready_attempts) + await asyncio.sleep(delay) + delay = min(delay * 1.5, 30.0) + + async def send_message_waiting_for_media( + self, client: MAXAPIClient, 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) + except MAXAPIError as exc: + if exc.code != "attachment.not.ready" or attempt >= attempts: + raise + logger.warning("MAX attachment not ready, retrying send in {}s ({}/{})", delay, attempt + 1, attempts) + await asyncio.sleep(delay) + delay = min(delay * 1.5, 30.0) + raise RuntimeError("MAX attachment readiness retry exhausted") + + async def upload_media_item( + self, client: MAXAPIClient, item: ProcessedMedia + ) -> Optional[dict[str, Any]]: + if not item.local_path or not item.local_path.exists(): + return None + + media_type = "video" if item.media_type == "video" else "image" + upload_info = await client.get_upload_url(media_type) + upload_url = str(upload_info.get("url") or "").strip() + if not upload_url: + raise RuntimeError(f"MAX upload URL is empty: {upload_info}") + + payload = await client.upload_file(upload_url, item.local_path) + if media_type == "video" and upload_info.get("token") and not payload.get("token"): + payload["token"] = upload_info["token"] + + has_image = bool(payload.get("photos")) if isinstance(payload.get("photos"), dict) else False + if not payload.get("token") and not has_image: + raise RuntimeError(f"MAX upload did not return valid token: {payload}") + + return {"type": media_type, "payload": payload} + + async def post_to_max( + self, + raw_text: str, + media_items: list[ProcessedMedia], + vk_url: Optional[str] = None, + ) -> tuple[list[str], Optional[str]]: + if not self.token or not self.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] + link_text = "" + if link_only: + lines = [f"- {m.media_type}: {m.original_url}" for m in link_only] + link_text = "\n\nМедиа по ссылке:\n" + "\n".join(lines) + + full_raw = (raw_text + link_text).strip() + formatted_text = format_post_text( + full_raw, + parse_mode="html", + bold_first_line=settings.format_first_line_bold, + vk_url=vk_url, + ) + + chunks = split_message_chunks(formatted_text, self.message_limit) + valid_media = [m for m in media_items if not m.is_link_only and m.local_path][:MAX_MEDIA_ITEMS] + + message_ids: list[str] = [] + first_url: Optional[str] = None + + async with MAXAPIClient(self.token, self.api_base_url) as client: + attachments: list[dict[str, Any]] = [] + for item in valid_media: + try: + att = await self.upload_media_item(client, item) + if att: + attachments.append(att) + except Exception as exc: + logger.warning("MAX media upload failed for {}: {}", item.attachment_id, exc) + + if attachments: + await self.wait_for_videos(client, attachments) + + first_text = chunks[0] if chunks else "" + res = await self.send_message_waiting_for_media(client, first_text, attachments) + + first_mid = self.message_id_from_response(res) + if first_mid: + message_ids.append(first_mid) + msg_obj = res.get("message") if isinstance(res.get("message"), dict) else res + first_url = self.message_url_from_response(msg_obj) + + # Try auto reaction if enabled + if settings.max_auto_reaction_enabled and settings.max_auto_reaction: + try: + await client.react( + settings.max_reaction_path_template, + first_mid, + settings.max_auto_reaction, + ) + except Exception as exc: + logger.warning("MAX auto reaction failed: {}", exc) + + for chunk in chunks[1:]: + await asyncio.sleep(1.0) + sub_res = await client.send_message(self.chat_id, chunk) + sub_mid = self.message_id_from_response(sub_res) + if sub_mid: + message_ids.append(sub_mid) + + logger.info("Sent MAX Message: {}", message_ids) + return message_ids, first_url diff --git a/src/media_processor.py b/src/media_processor.py new file mode 100644 index 0000000..dca50cb --- /dev/null +++ b/src/media_processor.py @@ -0,0 +1,274 @@ +from __future__ import annotations + +import asyncio +import os +import sys +import tempfile +import time +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Optional, Set +import aiohttp +from loguru import logger +from .config import settings + + +# Global registry of paths currently being written / downloaded / processed +_ACTIVE_LOCK = asyncio.Lock() +_ACTIVE_PATHS: Set[str] = set() + + +async def register_active_path(path: Path) -> None: + async with _ACTIVE_LOCK: + _ACTIVE_PATHS.add(str(path.resolve())) + + +async def unregister_active_path(path: Path) -> None: + async with _ACTIVE_LOCK: + _ACTIVE_PATHS.discard(str(path.resolve())) + + +async def is_path_active(path: Path) -> bool: + async with _ACTIVE_LOCK: + return str(path.resolve()) in _ACTIVE_PATHS + + +@dataclass +class ProcessedMedia: + media_type: str # "photo" or "video" + local_path: Optional[Path] + original_url: str + attachment_id: str + size_bytes: int = 0 + duration_sec: Optional[int] = None + width: Optional[int] = None + height: Optional[int] = None + error: Optional[str] = None + is_link_only: bool = False + + +class MediaProcessor: + def __init__(self, is_local_tg_api: bool = False) -> None: + self.is_local_tg_api = is_local_tg_api + self.cache_dir = settings.cache_path + + @property + def max_video_bytes(self) -> int: + max_mb = ( + settings.video_max_size_mb_local + if self.is_local_tg_api + else settings.video_max_size_mb_cloud + ) + return max_mb * 1024 * 1024 + + async def download_photo( + self, session: aiohttp.ClientSession, url: str, prefix: str = "photo_" + ) -> Optional[Path]: + if not url.startswith(("http://", "https://")): + return None + + suffix = Path(url.split("?", 1)[0]).suffix[:6] or ".jpg" + tmp = tempfile.NamedTemporaryFile( + dir=self.cache_dir, prefix=prefix, suffix=suffix, delete=False + ) + tmp_path = Path(tmp.name) + tmp.close() + + await register_active_path(tmp_path) + try: + async with session.get( + url, timeout=aiohttp.ClientTimeout(total=settings.media_download_timeout_sec) + ) as resp: + if resp.status != 200: + logger.warning("Photo download failed with status {}: {}", resp.status, url) + tmp_path.unlink(missing_ok=True) + await unregister_active_path(tmp_path) + return None + with open(tmp_path, "wb") as f: + async for chunk in resp.content.iter_chunked(64 * 1024): + f.write(chunk) + return tmp_path + except Exception as exc: + logger.warning("Error downloading photo {}: {}", url, exc) + tmp_path.unlink(missing_ok=True) + await unregister_active_path(tmp_path) + return None + + async def download_video_ytdlp( + self, url: str, prefix: str = "video_" + ) -> tuple[Optional[Path], Optional[str], bool]: + """Returns (path, error_message, is_permanent_error)""" + tmp = tempfile.NamedTemporaryFile( + dir=self.cache_dir, prefix=prefix, suffix=".mp4", delete=False + ) + output_path = Path(tmp.name) + tmp.close() + + await register_active_path(output_path) + max_size = self.max_video_bytes + netrc_path = None + + cmd = [sys.executable, "-m", "yt_dlp"] + if "vk.com" in url and settings.vk_access_token: + netrc_path = f"{output_path}.netrc" + try: + with open(netrc_path, "w", encoding="utf-8") as fh: + fh.write(f"machine vk.com login vk_token password {settings.vk_access_token}\n") + cmd.extend(["--netrc-location", netrc_path]) + except Exception as exc: + logger.warning("Failed to write netrc for yt-dlp: {}", exc) + + cmd.extend([ + url, + "-o", str(output_path), + "--no-playlist", + "--match-filter", f"duration <= {settings.video_max_duration_sec}", + "--merge-output-format", "mp4", + "-f", ( + f"best[height<={settings.video_max_height}][filesize<{max_size}]" + f"/best[height<={settings.video_max_height}]" + f"/bestvideo[height<={settings.video_max_height}][filesize<{max_size}]+bestaudio/best" + f"/best[filesize<{max_size}]" + f"/best" + ), + "--quiet", "--no-warnings", + ]) + + proc = None + stderr = b"" + try: + proc = await asyncio.create_subprocess_exec( + *cmd, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + _, stderr = await asyncio.wait_for( + proc.communicate(), timeout=settings.yt_dlp_timeout_sec + ) + except asyncio.TimeoutError: + if proc: + try: + proc.kill() + await proc.communicate() + except Exception: + pass + output_path.unlink(missing_ok=True) + await unregister_active_path(output_path) + return None, "video download timeout", False + finally: + if netrc_path: + Path(netrc_path).unlink(missing_ok=True) + + if proc.returncode != 0 or not output_path.exists() or output_path.stat().st_size == 0: + err_msg = (stderr or b"").decode("utf-8", errors="ignore").lower() + is_permanent = any( + m in err_msg for m in ( + "removed", "unavailable", "private", "access denied", "does not pass filter", "sign in" + ) + ) + output_path.unlink(missing_ok=True) + await unregister_active_path(output_path) + return None, f"yt-dlp failed: {err_msg[:200]}", is_permanent + + size = output_path.stat().st_size + if size > max_size: + logger.warning( + "Downloaded video size {} MB exceeds limit {} MB", + round(size / (1024 * 1024), 2), + round(max_size / (1024 * 1024), 2), + ) + output_path.unlink(missing_ok=True) + await unregister_active_path(output_path) + return None, "video exceeds size limit", True + + return output_path, None, False + + async def process_media_items( + self, items: list[Any] + ) -> list[ProcessedMedia]: + results: list[ProcessedMedia] = [] + async with aiohttp.ClientSession() as session: + for item in items: + media_type = item.media_type + url = item.url + att_id = item.attachment_id + + if media_type == "photo": + p = await self.download_photo(session, url, prefix=f"p_{att_id}_") + if p: + results.append( + ProcessedMedia( + media_type="photo", + local_path=p, + original_url=url, + attachment_id=att_id, + size_bytes=p.stat().st_size, + width=item.width, + height=item.height, + ) + ) + else: + results.append( + ProcessedMedia( + media_type="photo", + local_path=None, + original_url=url, + attachment_id=att_id, + error="failed to download photo", + is_link_only=True, + ) + ) + elif media_type == "video": + # Check duration + if item.duration_sec and item.duration_sec > settings.video_max_duration_sec: + results.append( + ProcessedMedia( + media_type="video", + local_path=None, + original_url=url, + attachment_id=att_id, + duration_sec=item.duration_sec, + error="video duration exceeds maximum", + is_link_only=True, + ) + ) + continue + + p, err, perm = await self.download_video_ytdlp(url, prefix=f"v_{att_id}_") + if p: + results.append( + ProcessedMedia( + media_type="video", + local_path=p, + original_url=url, + attachment_id=att_id, + size_bytes=p.stat().st_size, + duration_sec=item.duration_sec, + width=item.width, + height=item.height, + ) + ) + else: + results.append( + ProcessedMedia( + media_type="video", + local_path=None, + original_url=url, + attachment_id=att_id, + duration_sec=item.duration_sec, + error=err or "failed to download video", + is_link_only=True, + ) + ) + return results + + async def cleanup(self, items: list[ProcessedMedia]) -> None: + """Immediately unregisters and removes downloaded local files.""" + for item in items: + if item.local_path: + try: + await unregister_active_path(item.local_path) + item.local_path.unlink(missing_ok=True) + except Exception as exc: + logger.warning("Error cleaning up {}: {}", item.local_path, exc) + item.local_path = None diff --git a/src/text_formatter.py b/src/text_formatter.py new file mode 100644 index 0000000..dcda956 --- /dev/null +++ b/src/text_formatter.py @@ -0,0 +1,96 @@ +from __future__ import annotations + +import html +import re +from typing import Optional +from .config import settings + + +def split_message_chunks(text: str, limit: int = 4000) -> list[str]: + text = str(text or "").strip() + if not text: + return [] + chunks: list[str] = [] + while len(text) > limit: + split_at = text.rfind("\n", 0, limit) + if split_at < limit // 2: + split_at = text.rfind(" ", 0, limit) + if split_at < limit // 2: + split_at = limit + chunks.append(text[:split_at].strip()) + text = text[split_at:].strip() + if text: + chunks.append(text) + return chunks + + +def clean_dividers(lines: list[str]) -> list[str]: + sep_pattern = re.compile(r"^\s*[━—─\-=\*\#_]{2,}\s*$") + 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() + + +def format_post_text( + raw_text: str, + *, + parse_mode: str = "html", # "html" or "plain" + header: Optional[str] = None, + footer: Optional[str] = None, + tags: Optional[str] = None, + bold_first_line: bool = True, + vk_url: Optional[str] = None, +) -> str: + header = (settings.header_text if header is None else header).strip() + footer = (settings.footer_text if footer is None else footer).strip() + tags = (settings.common_tags if tags is None else tags).strip() + + body = strip_trailing_hashtags(raw_text) + lines = clean_dividers(body.splitlines()) + + title_idx: Optional[int] = None + for idx, line in enumerate(lines): + if line.strip(): + title_idx = idx + break + + formatted_lines: list[str] = [] + for idx, line in enumerate(lines): + if idx == title_idx and bold_first_line and line.strip(): + if parse_mode == "html": + escaped = html.escape(line.strip()) + formatted_lines.append(f"{escaped}") + else: + formatted_lines.append(f"**{line.strip()}**") + # Blank line after header if next line is not empty + if idx + 1 < len(lines) and lines[idx + 1].strip() != "": + formatted_lines.append("") + else: + if parse_mode == "html": + formatted_lines.append(html.escape(line)) + else: + formatted_lines.append(line) + + normalized_body = re.sub(r"\n{3,}", "\n\n", "\n".join(formatted_lines)).strip() + + parts: list[str] = [] + if header: + parts.append(html.escape(header) if parse_mode == "html" else header) + if normalized_body: + parts.append(normalized_body) + if footer: + parts.append(html.escape(footer) if parse_mode == "html" else footer) + if tags: + parts.append(html.escape(tags) if parse_mode == "html" else tags) + + return "\n\n".join(part for part in parts if part).strip() diff --git a/src/tg_poster.py b/src/tg_poster.py new file mode 100644 index 0000000..346d06d --- /dev/null +++ b/src/tg_poster.py @@ -0,0 +1,407 @@ +from __future__ import annotations + +import asyncio +import os +import re +from pathlib import Path +from typing import Any, Optional +import aiohttp +from aiogram import Bot +from aiogram.client.session.aiohttp import AiohttpSession +from aiogram.client.telegram import TelegramAPIServer +from aiogram.exceptions import TelegramRetryAfter +from aiogram.types import FSInputFile, InputMediaPhoto, InputMediaVideo +from loguru import logger +from .config import settings +from .media_processor import ProcessedMedia +from .text_formatter import format_post_text, split_message_chunks + +MAX_MEDIA_GROUP = 10 +MAX_RICH_MEDIA = 50 +MAX_RICH_TEXT = 32768 + + +def parse_topic(value: str) -> tuple[int, Optional[int]]: + raw = str(value or "").strip() + if not raw: + raise ValueError("Telegram chat ID is empty") + if ":" in raw: + chat_id, thread_id = raw.split(":", 1) + return int(chat_id), int(thread_id) + return int(raw), None + + +def tg_message_url(chat_id: int, message_id: int) -> str: + chat = str(abs(int(chat_id))) + if chat.startswith("100"): + chat = chat[3:] + return f"https://t.me/c/{chat}/{message_id}" + + +class RichMessageUnavailable(RuntimeError): + pass + + +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 + self.caption_limit: int = 1024 + self.message_limit: int = 4096 + + async def init(self) -> None: + 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) + except Exception as exc: + logger.warning("Could not parse TG_MEDIA_CHANNEL_ID: {}", exc) + + local_url = (settings.local_bot_api_url or "").strip().rstrip("/") + if local_url: + try: + session = AiohttpSession( + api=TelegramAPIServer.from_base(local_url, is_local=True) + ) + test_bot = Bot(token=settings.tg_bot_token, session=session) + me = await test_bot.get_me() + self.bot = test_bot + self.is_local_api = True + logger.info("Connected to local Telegram Bot API at {}: @{}", local_url, me.username) + except Exception as exc: + logger.warning( + "Local Telegram Bot API unavailable at {}: {}. Falling back to standard API.", + local_url, + exc, + ) + self.bot = Bot(token=settings.tg_bot_token) + self.is_local_api = False + else: + self.bot = Bot(token=settings.tg_bot_token) + self.is_local_api = False + + async def close(self) -> None: + 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 + kwargs: dict[str, Any] = {"chat_id": int(c_id)} + if t_id: + kwargs["message_thread_id"] = int(t_id) + return kwargs + + async def tg_retry(self, fn): + for attempt in range(1, 4): + try: + return await fn() + except TelegramRetryAfter as exc: + delay = float(exc.retry_after) + 0.5 + logger.warning("Telegram flood control: retry after {}s", delay) + await asyncio.sleep(delay) + except Exception as exc: + if attempt >= 3: + raise + logger.warning("Telegram request error attempt {}: {}", attempt, exc) + await asyncio.sleep(2 ** attempt) + raise RuntimeError("Telegram retries exhausted") + + async def upload_media_for_file_ids( + self, media_items: list[ProcessedMedia] + ) -> dict[str, str]: + """Uploads files to destination/storage to obtain tg_file_ids""" + file_ids: dict[str, str] = {} + valid_items = [m for m in media_items if m.local_path and m.local_path.exists()] + if not valid_items or not self.bot: + return file_ids + + # If storage channel is configured, we send there; otherwise we send directly + use_storage = bool(self.storage_chat_id) + group = [] + for item in valid_items[:MAX_MEDIA_GROUP]: + fs = FSInputFile(str(item.local_path)) + if item.media_type == "photo": + group.append(InputMediaPhoto(media=fs)) + else: + group.append(InputMediaVideo(media=fs)) + + if not group: + return file_ids + + try: + if len(group) == 1: + item = valid_items[0] + fs = FSInputFile(str(item.local_path)) + if item.media_type == "photo": + msg = await self.tg_retry( + lambda: self.bot.send_photo(photo=fs, **self.chat_kwargs(use_storage)) + ) + if msg.photo: + file_ids[item.attachment_id] = msg.photo[-1].file_id + else: + msg = await self.tg_retry( + lambda: self.bot.send_video(video=fs, **self.chat_kwargs(use_storage)) + ) + if msg.video: + file_ids[item.attachment_id] = msg.video.file_id + else: + msgs = await self.tg_retry( + lambda: self.bot.send_media_group(media=group, **self.chat_kwargs(use_storage)) + ) + for item, msg in zip(valid_items, msgs): + if item.media_type == "photo" and msg.photo: + file_ids[item.attachment_id] = msg.photo[-1].file_id + elif item.media_type == "video" and msg.video: + file_ids[item.attachment_id] = msg.video.file_id + except Exception as exc: + logger.warning("Error uploading media items for file_ids: {}", exc) + + return file_ids + + def build_rich_text_html(self, text: str) -> str: + paragraphs = [] + for p in text.strip().split("\n\n"): + body = "
".join(line for line in p.splitlines() if line.strip()) + if body: + paragraphs.append(f"

{body}

") + return "\n".join(paragraphs) + + def build_rich_message( + self, text: str, media_items: list[ProcessedMedia], file_ids: dict[str, str] + ) -> Optional[dict[str, Any]]: + rich_media = [] + media_tags = [] + + valid_items = [ + m for m in media_items if m.attachment_id in file_ids + ][:MAX_RICH_MEDIA] + + if not valid_items: + return None + + for idx, item in enumerate(valid_items): + f_id = file_ids[item.attachment_id] + media_id = f"m{idx}" + m_type = "photo" if item.media_type == "photo" else "video" + rich_media.append({"id": media_id, "media": {"type": m_type, "media": f_id}}) + if item.media_type == "photo": + media_tags.append(f'') + else: + media_tags.append(f'') + + rich_text = self.build_rich_text_html(text) + if len(rich_text) > MAX_RICH_TEXT: + return None + + media_html = ( + media_tags[0] if len(media_tags) == 1 else f"{''.join(media_tags)}" + ) + return { + "html": f"{media_html}\n{rich_text}" if rich_text else media_html, + "media": rich_media, + } + + async def send_rich_message(self, rich_message: dict[str, Any]) -> list[int]: + data: dict[str, Any] = { + "chat_id": int(self.chat_id), + "rich_message": rich_message, + } + if self.thread_id: + data["message_thread_id"] = int(self.thread_id) + + base_url = (settings.local_bot_api_url or "").strip().rstrip("/") or "https://api.telegram.org" + url = f"{base_url}/bot{settings.tg_bot_token}/sendRichMessage" + + timeout = aiohttp.ClientTimeout(total=90) + async with aiohttp.ClientSession(timeout=timeout) as session: + async with session.post(url, json=data) as resp: + payload = await resp.json(content_type=None) + + if payload.get("ok"): + res = payload.get("result", {}) + mid = res.get("message_id") + if mid: + return [int(mid)] + raise RichMessageUnavailable("sendRichMessage ok but no message_id") + + desc = str(payload.get("description") or f"HTTP {resp.status}") + raise RichMessageUnavailable(desc) + + async def send_legacy_media_post( + self, text: str, media_items: list[ProcessedMedia], file_ids: dict[str, str] + ) -> list[int]: + if not self.bot: + raise RuntimeError("Bot not initialized") + + input_media = [] + valid_items = [ + m for m in media_items if m.local_path or m.attachment_id in file_ids + ] + + for item in valid_items: + fid = file_ids.get(item.attachment_id) + val = fid if fid else (FSInputFile(str(item.local_path)) if item.local_path else None) + if not val: + continue + if item.media_type == "photo": + input_media.append({"type": "photo", "media": val}) + else: + input_media.append({"type": "video", "media": val}) + + if not input_media: + # Text only post + return await self.send_text_post(text) + + # Handle caption vs overflow + text_chunks: list[str] = [] + if len(text) <= self.caption_limit: + first_caption = text + else: + first_caption = "" + text_chunks = split_message_chunks(text, self.message_limit) + + first_group = input_media[:MAX_MEDIA_GROUP] + message_ids: list[int] = [] + + if len(first_group) == 1: + item = first_group[0] + if item["type"] == "photo": + msg = await self.tg_retry( + lambda: self.bot.send_photo( + photo=item["media"], + caption=first_caption or None, + parse_mode="HTML", + **self.chat_kwargs(), + ) + ) + else: + msg = await self.tg_retry( + lambda: self.bot.send_video( + video=item["media"], + caption=first_caption or None, + parse_mode="HTML", + **self.chat_kwargs(), + ) + ) + message_ids.append(int(msg.message_id)) + else: + group = [] + for idx, item in enumerate(first_group): + cap = first_caption if idx == 0 and first_caption else None + if item["type"] == "photo": + group.append(InputMediaPhoto(media=item["media"], caption=cap, parse_mode="HTML")) + else: + group.append(InputMediaVideo(media=item["media"], caption=cap, parse_mode="HTML")) + msgs = await self.tg_retry( + lambda: self.bot.send_media_group(media=group, **self.chat_kwargs()) + ) + message_ids.extend(int(m.message_id) for m in msgs) + + # Remaining media chunks if more than 10 + rest = input_media[MAX_MEDIA_GROUP:] + for start in range(0, len(rest), MAX_MEDIA_GROUP): + chunk = rest[start : start + MAX_MEDIA_GROUP] + g = [ + InputMediaPhoto(media=item["media"]) + if item["type"] == "photo" + else InputMediaVideo(media=item["media"]) + for item in chunk + ] + msgs = await self.tg_retry( + lambda: self.bot.send_media_group(media=g, **self.chat_kwargs()) + ) + message_ids.extend(int(m.message_id) for m in msgs) + + # Overflow text chunks if caption exceeded 1024 chars + for chunk in text_chunks: + msg = await self.tg_retry( + lambda c=chunk: self.bot.send_message( + text=c, + parse_mode="HTML", + disable_web_page_preview=True, + **self.chat_kwargs(), + ) + ) + message_ids.append(int(msg.message_id)) + + return message_ids + + async def send_text_post(self, text: str) -> list[int]: + if not self.bot: + raise RuntimeError("Bot not initialized") + chunks = split_message_chunks(text, self.message_limit) + mids: list[int] = [] + for chunk in chunks: + msg = await self.tg_retry( + lambda c=chunk: self.bot.send_message( + text=c, + parse_mode="HTML", + disable_web_page_preview=True, + **self.chat_kwargs(), + ) + ) + mids.append(int(msg.message_id)) + return mids + + async def post_to_telegram( + self, + raw_text: str, + media_items: list[ProcessedMedia], + vk_url: Optional[str] = None, + ) -> tuple[list[int], Optional[str]]: + """ + Main Telegram posting routine: + 1. Formats text for HTML parse mode. + 2. Uploads media (or storage channel if configured) to get file_ids. + 3. Tries sendRichMessage first. + 4. If unavailable, falls back to legacy media groups / single media / text. + """ + # Append link-only media notice if any videos/photos couldn't be downloaded + link_only = [m for m in media_items if m.is_link_only] + link_text = "" + if link_only: + lines = [f"- {m.media_type}: ссылка" for m in link_only] + link_text = "\n\nМедиа по ссылке:\n" + "\n".join(lines) + + full_raw = (raw_text + link_text).strip() + formatted_text = format_post_text( + full_raw, + parse_mode="html", + bold_first_line=settings.format_first_line_bold, + vk_url=vk_url, + ) + + valid_media = [m for m in media_items if not m.is_link_only and m.local_path] + + # 1. Obtain file_ids if we have storage channel or if we want rich message + file_ids: dict[str, str] = {} + if valid_media and self.storage_chat_id: + file_ids = await self.upload_media_for_file_ids(valid_media) + + # 2. Try Rich Message if file_ids are available + if file_ids: + 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 + logger.info("Sent Telegram Rich Message: {}", mids) + return mids, url + except RichMessageUnavailable as exc: + logger.warning("Telegram sendRichMessage failed: {}. Falling back to legacy.", exc) + + # 3. Fallback to legacy media group / text + mids = await self.send_legacy_media_post(formatted_text, valid_media, file_ids) + url = tg_message_url(self.chat_id, mids[0]) if mids else None + logger.info("Sent Telegram Legacy Message: {}", mids) + return mids, url diff --git a/src/vk_client.py b/src/vk_client.py new file mode 100644 index 0000000..329ff52 --- /dev/null +++ b/src/vk_client.py @@ -0,0 +1,224 @@ +from __future__ import annotations + +import asyncio +import re +import time +from dataclasses import dataclass +from typing import Any, Optional +import aiohttp +from loguru import logger +from .config import settings + + +class VKAPIError(RuntimeError): + def __init__(self, code: Optional[int], message: str) -> None: + self.code = code + super().__init__(f"VK API error {code}: {message}") + + +class VKRateLimiter: + def __init__(self, rps: int = 3) -> None: + self.rps = max(1, int(rps)) + self.interval = 1.0 / self.rps + self._last = 0.0 + self._lock = asyncio.Lock() + + async def acquire(self) -> None: + async with self._lock: + now = time.monotonic() + wait = self.interval - (now - self._last) + if wait > 0: + await asyncio.sleep(wait) + self._last = time.monotonic() + + +@dataclass +class VKMediaItem: + media_type: str # "photo" or "video" + url: str + attachment_id: str + width: Optional[int] = None + height: Optional[int] = None + duration_sec: Optional[int] = None + title: Optional[str] = None + + +@dataclass +class VKPost: + post_id: int + owner_id: int + date: int + text: str + media: list[VKMediaItem] + raw: dict[str, Any] + is_pinned: bool = False + is_repost: bool = False + + +def normalize_vk_source(value: str) -> str: + raw = str(value or "").strip() + if not raw: + return "" + for host in ("vk.com", "vk.ru", "m.vk.com"): + if host in raw: + raw = raw.split(host, 1)[1] + break + raw = raw.lstrip("/") + raw = raw.split("?", 1)[0].split("#", 1)[0].strip() + raw = re.sub(r"^(club|public)", "", raw, flags=re.IGNORECASE) + return raw.lower() + + +class VKClient: + base_url = "https://api.vk.com/method" + + def __init__( + self, + token: Optional[str] = None, + version: Optional[str] = None, + rps: int = 3, + ) -> None: + self.token = token or settings.vk_access_token + self.version = version or settings.vk_api_version + self.limiter = VKRateLimiter(rps) + self.session: Optional[aiohttp.ClientSession] = None + + async def __aenter__(self) -> "VKClient": + self.session = aiohttp.ClientSession( + timeout=aiohttp.ClientTimeout(total=45, connect=10) + ) + return self + + async def __aexit__(self, *args) -> None: + if self.session: + await self.session.close() + + async def call(self, method: str, **params: Any) -> Any: + if not self.session: + raise RuntimeError("VKClient session not initialized") + if not self.token: + raise RuntimeError("VK_ACCESS_TOKEN is empty") + + payload = dict(params) + payload["access_token"] = self.token + payload["v"] = self.version + + for attempt in range(1, 4): + try: + await self.limiter.acquire() + async with self.session.post(f"{self.base_url}/{method}", data=payload) as resp: + resp.raise_for_status() + data = await resp.json(content_type=None) + + if "error" not in data: + return data.get("response", {}) + + err = data["error"] + code = err.get("error_code") + msg = err.get("error_msg", "unknown") + if code == 6 and attempt < 3: + logger.warning("VK rate limit reached, sleeping 1.5s") + await asyncio.sleep(1.5) + continue + raise VKAPIError(code, msg) + except VKAPIError: + raise + except Exception as exc: + if attempt >= 3: + raise + logger.warning("VK call {} error attempt {}: {}", method, attempt, exc) + await asyncio.sleep(1.5 * attempt) + + raise VKAPIError(None, "VK retries exhausted") + + async def resolve_group(self, input_value: str) -> tuple[str, int, str]: + """Resolves screen name/link to (screen_name, owner_id (negative), title)""" + norm = normalize_vk_source(input_value) + if not norm: + raise ValueError(f"Invalid VK source identifier: {input_value}") + + if norm.lstrip("-").isdigit(): + group_id = abs(int(norm)) + res = await self.call("groups.getById", group_id=str(group_id)) + else: + res = await self.call("groups.getById", group_id=norm) + + items = res if isinstance(res, list) else res.get("groups", []) + if not items: + raise VKAPIError(None, f"Cannot resolve VK group for: {input_value}") + + g = items[0] + gid = int(g["id"]) + screen_name = str(g.get("screen_name") or norm) + name = str(g.get("name") or screen_name) + return screen_name, -gid, name + + def extract_media(self, post_raw: dict[str, Any]) -> list[VKMediaItem]: + items: list[VKMediaItem] = [] + for att in post_raw.get("attachments", []) or []: + att_type = att.get("type") + if att_type == "photo": + photo = att.get("photo") or {} + owner_id = photo.get("owner_id") + photo_id = photo.get("id") + sizes = sorted( + photo.get("sizes", []) or [], + key=lambda s: int(s.get("width") or 0) * int(s.get("height") or 0), + reverse=True, + ) + if sizes and owner_id is not None and photo_id is not None: + best = sizes[0] + items.append( + VKMediaItem( + media_type="photo", + url=str(best.get("url") or ""), + attachment_id=f"photo{owner_id}_{photo_id}", + width=best.get("width"), + height=best.get("height"), + ) + ) + elif att_type == "video": + video = att.get("video") or {} + owner_id = video.get("owner_id") + video_id = video.get("id") + access_key = video.get("access_key") + if owner_id is not None and video_id is not None: + video_url = f"https://vk.com/video{owner_id}_{video_id}" + if access_key: + video_url += f"_{access_key}" + items.append( + VKMediaItem( + media_type="video", + url=video_url, + attachment_id=f"video{owner_id}_{video_id}", + width=video.get("width"), + height=video.get("height"), + duration_sec=video.get("duration"), + title=video.get("title"), + ) + ) + return items + + async def get_latest_posts(self, owner_id: int, count: int = 10) -> list[VKPost]: + res = await self.call("wall.get", owner_id=owner_id, count=count, filter="owner") + items = res.get("items", []) if isinstance(res, dict) else [] + posts: list[VKPost] = [] + for raw in items: + if raw.get("is_deleted") or not raw.get("id") or not raw.get("date"): + continue + is_pinned = bool(raw.get("is_pinned")) + is_repost = bool(raw.get("copy_history")) + media = self.extract_media(raw) + posts.append( + VKPost( + post_id=int(raw["id"]), + owner_id=int(raw.get("owner_id") or owner_id), + date=int(raw["date"]), + text=str(raw.get("text") or ""), + media=media, + raw=raw, + is_pinned=is_pinned, + is_repost=is_repost, + ) + ) + return posts