From ec65aaf57dd9e5dde22cb1788befb36249b86972 Mon Sep 17 00:00:00 2001 From: exostring Date: Sat, 15 Aug 2026 12:17:47 +0500 Subject: [PATCH] Fix duplicate-post and timeout bugs in TG/MAX posting pipeline - Stop blindly retrying send_photo/send_video/send_media_group and MAX send_message on ambiguous network timeouts - a timeout doesn't prove the message wasn't delivered, and retrying risked posting duplicates (observed live: a video posted 3-5x after repeated timeout retries). - Raise/rework timeouts that were too short for real large-file transfer speeds: TG media upload timeout, and yt-dlp download now uses stall detection (killed only on true silence) instead of a flat ceiling that was cutting off legitimately slow-but-successful video downloads. - Fix sendRichMessage: ok=true with an unparseable message_id no longer triggers a fallback send (Telegram already created the message). - MAX: videos over the documented 250MB cap are now sent as a "watch via link" note instead of silently failing the upload. - Add PRAGMA busy_timeout to all DB connections. - Remove unused MAX_MEDIA_CHANNEL_ID (MAX's /uploads returns a portable token directly, no staging channel needed, unlike Telegram). Co-Authored-By: Claude Sonnet 5 --- .env.example | 6 ++ src/cleaner.py | 5 +- src/config.py | 15 ++++- src/database.py | 38 ++++++++----- src/main.py | 29 ++++------ src/max_poster.py | 125 ++++++++++++++++++++++++++++++----------- src/media_processor.py | 50 +++++++++++------ src/text_formatter.py | 23 ++++++++ src/tg_poster.py | 93 ++++++++++++++++++++++-------- 9 files changed, 276 insertions(+), 108 deletions(-) diff --git a/.env.example b/.env.example index 8f86999..adf4b0b 100644 --- a/.env.example +++ b/.env.example @@ -21,6 +21,9 @@ TG_CHAT_ID=-1001234567890 TG_MEDIA_CHANNEL_ID= # Admin Telegram ID(s) for notifications & reports (single ID or comma-separated "123456,789012") TG_ADMIN_IDS=123456789 +# Timeout (seconds) for send_photo/send_video/send_media_group. Must comfortably +# cover your largest expected video at real upload speed (default 1800 = 30min). +TG_MEDIA_UPLOAD_TIMEOUT_SEC=1800 # ========================================== # Local Telegram Bot API (Optional) @@ -39,6 +42,9 @@ MAX_BOT_TOKEN=your_max_bot_token_here MAX_CHAT_ID=123456 # MAX API Base URL MAX_API_BASE_URL=https://platform-api2.max.ru +# MAX's documented hard cap for a single video attachment. Videos over this are +# sent as a "watch via link" text note instead of failing the upload. +MAX_VIDEO_LIMIT_MB=250 # ========================================== # Polling, Reporting & Maintenance diff --git a/src/cleaner.py b/src/cleaner.py index fcf785f..68a5c2c 100644 --- a/src/cleaner.py +++ b/src/cleaner.py @@ -2,9 +2,8 @@ from __future__ import annotations from loguru import logger import asyncio -import os import time -from pathlib import Path +from typing import Optional try: from .config import settings from .media_processor import is_path_active @@ -17,6 +16,8 @@ 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. + Safety net for the immediate per-post cleanup in main.py's process_new_post: + catches anything left behind if the process died mid-post. """ if max_age_minutes is None: max_age_minutes = settings.cache_max_age_minutes diff --git a/src/config.py b/src/config.py index 236572e..879196f 100644 --- a/src/config.py +++ b/src/config.py @@ -26,6 +26,10 @@ class Settings(BaseSettings): 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" + # Ceiling for send_photo/send_video/send_media_group requests. Must comfortably + # cover a 2GB upload at realistic throughput, not just the observed happy path - + # measured ~3.5MB/s for a 207MB video means 2GB alone can take ~10min. + tg_media_upload_timeout_sec: int = 1800 # NOTE: TELEGRAM_API_ID / TELEGRAM_API_HASH are intentionally not modeled here - # they're only consumed by docker-entrypoint.sh (raw env) to start the local # telegram-bot-api binary, never read from Python. @@ -34,6 +38,9 @@ class Settings(BaseSettings): max_bot_token: str = "" max_chat_id: str = "" # 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. + max_video_limit_mb: int = 250 max_video_ready_attempts: int = 6 max_video_ready_delay_sec: float = 8.0 @@ -57,7 +64,13 @@ class Settings(BaseSettings): video_max_duration_sec: int = 7200 video_max_height: int = 720 media_download_timeout_sec: int = 120 - yt_dlp_timeout_sec: int = 600 + # Absolute backstop for the whole download, regardless of progress (guards + # against a pathological slow-trickle that never actually stalls). + yt_dlp_timeout_sec: int = 3600 + # Killed only if yt-dlp produces NO progress output for this long - a real + # stall, not just a big/slow file. This is the timeout that actually matters + # day to day; yt_dlp_timeout_sec above is just the outer safety net. + yt_dlp_stall_timeout_sec: int = 120 # Text Styling & Decoration header_text: str = "" diff --git a/src/database.py b/src/database.py index 97be1bb..682ad20 100644 --- a/src/database.py +++ b/src/database.py @@ -2,8 +2,8 @@ from __future__ import annotations from loguru import logger import json -from datetime import datetime -from typing import Any, Optional +from contextlib import asynccontextmanager +from typing import Any, AsyncIterator, Optional import aiosqlite try: from .config import settings @@ -11,13 +11,26 @@ except (ImportError, ValueError): from config import settings +# Each method opens its own short-lived connection (no pooling). busy_timeout +# means a second connection opened concurrently (e.g. inspecting the DB by +# hand with sqlite3 while the service runs) waits instead of immediately +# failing with "database is locked". +_BUSY_TIMEOUT_MS = 5000 + + 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) + @asynccontextmanager + async def _connect(self) -> AsyncIterator[aiosqlite.Connection]: + async with aiosqlite.connect(self.db_path) as db: + await db.execute(f"PRAGMA busy_timeout = {_BUSY_TIMEOUT_MS};") + yield db + async def init(self) -> None: logger.info("Initializing database at {}", self.db_path) - async with aiosqlite.connect(self.db_path) as db: + async with self._connect() as db: await db.execute("PRAGMA journal_mode=WAL;") await db.execute( """ @@ -58,7 +71,7 @@ class Database: await db.commit() async def has_any_posts(self, owner_id: int) -> bool: - async with aiosqlite.connect(self.db_path) as db: + async with self._connect() as db: cursor = await db.execute( "SELECT 1 FROM posts WHERE vk_owner_id = ? LIMIT 1", (owner_id,), @@ -76,7 +89,7 @@ class Database: reason: str = "bootstrap_initial_skip", ) -> None: raw_json = json.dumps(raw_data, ensure_ascii=False) - async with aiosqlite.connect(self.db_path) as db: + 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) @@ -90,23 +103,22 @@ class Database: 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: + async with self._connect() 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 = ?", + "SELECT 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: + 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 = ?", @@ -124,7 +136,7 @@ class Database: raw_data: dict[str, Any], ) -> int: raw_json = json.dumps(raw_data, ensure_ascii=False) - async with aiosqlite.connect(self.db_path) as db: + async with self._connect() as db: cursor = await db.execute( """ INSERT INTO posts (vk_owner_id, vk_post_id, posted_at, text, raw_json) @@ -149,7 +161,7 @@ class Database: 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: + async with self._connect() as db: await db.execute( """ UPDATE posts @@ -173,7 +185,7 @@ class Database: error: Optional[str] = None, ) -> None: msg_str = ",".join(message_ids) if message_ids else None - async with aiosqlite.connect(self.db_path) as db: + async with self._connect() as db: await db.execute( """ UPDATE posts @@ -196,7 +208,7 @@ class Database: status: str = "ok", error: Optional[str] = None, ) -> None: - async with aiosqlite.connect(self.db_path) as db: + async with self._connect() as db: await db.execute( """ INSERT INTO publication_runs (found_count, published_tg_count, published_max_count, status, error) diff --git a/src/main.py b/src/main.py index f840341..e42fa99 100644 --- a/src/main.py +++ b/src/main.py @@ -1,7 +1,6 @@ from __future__ import annotations import asyncio -import os import signal import sys from typing import Any, Optional @@ -47,15 +46,12 @@ class ServiceApp: ) 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") @@ -66,14 +62,12 @@ class ServiceApp: 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, @@ -100,12 +94,13 @@ class ServiceApp: } try: - # 1. Download/extract media (skip entirely if both platforms are already done) + # 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) processed_media = await media_processor.process_media_items(post.media) - # 2. Publish to Telegram (only if not already published/skipped for this post) + # 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")) else: @@ -130,7 +125,7 @@ class ServiceApp: result_summary["tg_status"] = "failed" result_summary["tg_error"] = err - # 3. Publish to MAX Messenger (only if not already published/skipped for this post) + # MAX - independent of Telegram. if not (settings.max_bot_token and settings.max_chat_id): if not max_done: await self.db.update_max_result(post_db_id=post_db_id, status="skipped") @@ -160,7 +155,8 @@ class ServiceApp: result_summary["max_error"] = err finally: - # Immediate cleanup of temporary media files + # Immediate cleanup of temporary media files. If the process dies + # before this runs, cleaner.py's age-based sweep catches it later. if processed_media: await media_processor.cleanup(processed_media) @@ -182,7 +178,6 @@ class ServiceApp: 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( @@ -202,7 +197,6 @@ class ServiceApp: 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: @@ -217,7 +211,6 @@ class ServiceApp: 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) @@ -226,7 +219,6 @@ class ServiceApp: 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: @@ -234,7 +226,7 @@ class ServiceApp: 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 + await asyncio.sleep(2.0) # gentle pacing between posts else: logger.info("No new posts found.") @@ -258,7 +250,9 @@ class ServiceApp: error=cycle_error, ) - # Notify admins + # 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). if self.admin_notifier: await self.admin_notifier.notify_cycle_result( group_name=self.vk_group_name, @@ -278,7 +272,6 @@ class ServiceApp: 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): @@ -292,7 +285,6 @@ class ServiceApp: 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.") @@ -309,7 +301,6 @@ async def main() -> None: try: loop.add_signal_handler(sig, handle_signal) except NotImplementedError: - # Signal handlers not implemented on Windows event loop for non-main threads pass try: diff --git a/src/max_poster.py b/src/max_poster.py index 81d26ad..12fb689 100644 --- a/src/max_poster.py +++ b/src/max_poster.py @@ -11,11 +11,11 @@ import aiohttp try: from .config import settings from .media_processor import ProcessedMedia - from .text_formatter import build_media_unavailable_note, format_post_text, split_message_chunks + from .text_formatter import build_media_unavailable_note, build_oversized_video_note, format_post_text, split_message_chunks except (ImportError, ValueError): from config import settings from media_processor import ProcessedMedia - from text_formatter import build_media_unavailable_note, format_post_text, split_message_chunks + from text_formatter import build_media_unavailable_note, build_oversized_video_note, format_post_text, split_message_chunks MAX_MESSAGE_LIMIT = 4000 MAX_MEDIA_ITEMS = 10 @@ -55,7 +55,15 @@ class MAXAPIClient: if self.session: await self.session.close() - async def request(self, method: str, path: str, **kwargs: Any) -> dict[str, Any]: + async def request( + self, method: str, path: str, retry_on_exception: bool = True, **kwargs: Any + ) -> dict[str, Any]: + """retry_on_exception=False for calls where a timeout/connection error is + ambiguous about whether the server actually processed it (e.g. send_message): + a network exception there does NOT prove nothing was sent, so blindly + retrying risks posting duplicates. HTTP-response-based retries (429, + attachment.not.ready) stay safe regardless - those only fire once we + know the server explicitly rejected the request.""" if not self.session: raise RuntimeError("MAX session is not initialized") url = f"{self.base_url}{path}" @@ -84,9 +92,11 @@ class MAXAPIClient: continue if resp.status < 500: raise MAXAPIError(last_error, resp.status, str(code or "")) + except MAXAPIError: + raise except Exception as exc: last_error = str(exc) - if attempt >= self.max_attempts: + if not retry_on_exception or 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") @@ -100,7 +110,9 @@ class MAXAPIClient: 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) + return await self.request( + "POST", f"/messages?chat_id={quote(str(chat_id))}", retry_on_exception=False, json=payload + ) async def get_message(self, message_id: str) -> dict[str, Any]: return await self.request("GET", f"/messages/{quote(message_id, safe='')}") @@ -228,27 +240,59 @@ class MAXPoster: return {"type": media_type, "payload": payload} - async def upload_media_group( - self, client: MAXAPIClient, items: list[ProcessedMedia] - ) -> list[dict[str, Any]]: - attachments: list[dict[str, Any]] = [] - for item in items: - 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) - return attachments + async def upload_media( + self, media_items: list[ProcessedMedia] + ) -> tuple[list[dict[str, Any]], list[ProcessedMedia]]: + """Upload stage: push every media item to MAX and poll until all video + attachments report ready, so send_post() never blocks on readiness. - async def post_to_max( + Returns (attachments, oversized_videos): videos over MAX's documented + 250MB cap are never even attempted (MAX would just reject them) - they're + 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: + return [], [] + valid_media = [m for m in media_items if not m.is_link_only and m.local_path] + if not valid_media: + return [], [] + + max_video_bytes = settings.max_video_limit_mb * 1024 * 1024 + oversized: list[ProcessedMedia] = [] + uploadable: list[ProcessedMedia] = [] + for item in valid_media: + if item.media_type == "video" and item.size_bytes > max_video_bytes: + logger.info( + "Video {} ({} MB) exceeds MAX's {}MB limit, sending as a link instead", + item.attachment_id, round(item.size_bytes / (1024 * 1024), 1), settings.max_video_limit_mb, + ) + oversized.append(item) + else: + uploadable.append(item) + + attachments: list[dict[str, Any]] = [] + async with MAXAPIClient(self.token, self.api_base_url) as client: + for item in uploadable: + 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) + return attachments, oversized + + async def send_post( self, raw_text: str, - media_items: list[ProcessedMedia], + attachments: list[dict[str, Any]], 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: logger.warning("MAX bot token or chat ID is not set; skipping MAX post.") return [], None @@ -260,18 +304,18 @@ class MAXPoster: vk_url=vk_url, ) - link_only = [m for m in media_items if m.is_link_only] - note = build_media_unavailable_note(link_only, parse_mode="html") - if note: - formatted_text = f"{formatted_text}\n\n{note}" if formatted_text else note + note = build_media_unavailable_note(link_only_media or [], parse_mode="html") + video_note = build_oversized_video_note(oversized_videos or [], parse_mode="html") + for extra in (note, video_note): + if extra: + formatted_text = f"{formatted_text}\n\n{extra}" if formatted_text else extra 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] # Chunk into groups of MAX_MEDIA_ITEMS instead of silently dropping the excess: # the first group rides with the text message, extra groups go out as follow-ups. media_groups = ( - [valid_media[i : i + MAX_MEDIA_ITEMS] for i in range(0, len(valid_media), MAX_MEDIA_ITEMS)] - if valid_media + [attachments[i : i + MAX_MEDIA_ITEMS] for i in range(0, len(attachments), MAX_MEDIA_ITEMS)] + if attachments else [[]] ) @@ -279,10 +323,8 @@ class MAXPoster: first_url: Optional[str] = None async with MAXAPIClient(self.token, self.api_base_url) as client: - first_attachments = await self.upload_media_group(client, media_groups[0]) - first_text = chunks[0] if chunks else "" - res = await self.send_message_waiting_for_media(client, first_text, first_attachments) + res = await self.send_message_waiting_for_media(client, first_text, media_groups[0]) first_mid = self.message_id_from_response(res) if first_mid: @@ -291,11 +333,10 @@ class MAXPoster: first_url = self.message_url_from_response(msg_obj) for group in media_groups[1:]: - attachments = await self.upload_media_group(client, group) - if not attachments: + if not group: continue await asyncio.sleep(1.0) - sub_res = await self.send_message_waiting_for_media(client, "", attachments) + sub_res = await self.send_message_waiting_for_media(client, "", group) sub_mid = self.message_id_from_response(sub_res) if sub_mid: message_ids.append(sub_mid) @@ -309,3 +350,19 @@ class MAXPoster: logger.info("Sent MAX Message: {}", message_ids) return message_ids, first_url + + async def post_to_max( + self, + raw_text: str, + media_items: list[ProcessedMedia], + 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: + 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 + ) diff --git a/src/media_processor.py b/src/media_processor.py index e36acfe..cde62cb 100644 --- a/src/media_processor.py +++ b/src/media_processor.py @@ -132,36 +132,54 @@ class MediaProcessor: f"/bestvideo[height<={settings.video_max_height}][filesize<{max_size}]+bestaudio" f"/bestvideo[height<={settings.video_max_height}]+bestaudio" ), - "--quiet", "--no-warnings", + # No --quiet: we need yt-dlp's own progress lines to tell a slow-but-alive + # download (fine, however long it takes) apart from a genuinely hung one - + # a flat wall-clock timeout can't tell those apart and was killing large + # videos that just needed more time (same class of bug as the TG upload + # timeout). --newline makes each progress update its own line instead of + # overwriting via \r, so we can read it with readline(). + "--newline", "--no-warnings", ]) proc = None - stderr = b"" + output_lines: list[str] = [] + deadline = asyncio.get_running_loop().time() + settings.yt_dlp_timeout_sec + stall_reason: Optional[str] = None try: proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, ) - _, stderr = await asyncio.wait_for( - proc.communicate(), timeout=settings.yt_dlp_timeout_sec - ) - except asyncio.TimeoutError: - if proc: + while True: + remaining_total = deadline - asyncio.get_running_loop().time() + if remaining_total <= 0: + stall_reason = "video download exceeded overall timeout" + break + wait_for = min(settings.yt_dlp_stall_timeout_sec, remaining_total) 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 + line = await asyncio.wait_for(proc.stdout.readline(), timeout=wait_for) + except asyncio.TimeoutError: + stall_reason = "video download stalled (no progress from yt-dlp)" + break + if not line: + break # stdout closed - process is finishing up + output_lines.append(line.decode("utf-8", errors="ignore")) + + if stall_reason: + proc.kill() + await proc.communicate() + output_path.unlink(missing_ok=True) + await unregister_active_path(output_path) + return None, stall_reason, False + + await proc.wait() 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() + err_msg = "".join(output_lines).lower() is_permanent = any( m in err_msg for m in ( "removed", "unavailable", "private", "access denied", "does not pass filter", "sign in" diff --git a/src/text_formatter.py b/src/text_formatter.py index c07ce63..8692e96 100644 --- a/src/text_formatter.py +++ b/src/text_formatter.py @@ -176,6 +176,29 @@ def build_media_unavailable_note(link_only_items: list, parse_mode: str = "html" return header + "\n" + "\n".join(lines) +def build_oversized_video_note(items: list, parse_mode: str = "html") -> str: + """Like build_media_unavailable_note, but for videos that were deliberately + skipped for exceeding a platform's size limit (e.g. MAX's 250MB video cap) - + friendlier tone since this isn't a failure, just a known platform limit.""" + if not items: + return "" + lines = [] + for item in items: + url = str(getattr(item, "original_url", "") or "").strip() + if not url: + continue + if parse_mode == "html": + lines.append(f'- видео по ссылке') + else: + lines.append(f"- видео по ссылке: {url}") + if not lines: + return "" + header = "Видео слишком большое для платформы, посмотреть можно здесь:" + if parse_mode == "html": + header = f"{html.escape(header)}" + return header + "\n" + "\n".join(lines) + + def format_post_text( raw_text: str, *, diff --git a/src/tg_poster.py b/src/tg_poster.py index 522a674..5867844 100644 --- a/src/tg_poster.py +++ b/src/tg_poster.py @@ -73,7 +73,8 @@ class TelegramPoster: if local_url: try: session = AiohttpSession( - api=TelegramAPIServer.from_base(local_url, is_local=True) + api=TelegramAPIServer.from_base(local_url, is_local=True), + timeout=settings.tg_media_upload_timeout_sec, ) test_bot = Bot(token=settings.tg_bot_token, session=session) me = await test_bot.get_me() @@ -86,10 +87,10 @@ class TelegramPoster: local_url, exc, ) - self.bot = Bot(token=settings.tg_bot_token) + self.bot = Bot(token=settings.tg_bot_token, session=AiohttpSession(timeout=settings.tg_media_upload_timeout_sec)) self.is_local_api = False else: - self.bot = Bot(token=settings.tg_bot_token) + self.bot = Bot(token=settings.tg_bot_token, session=AiohttpSession(timeout=settings.tg_media_upload_timeout_sec)) self.is_local_api = False async def close(self) -> None: @@ -119,6 +120,21 @@ class TelegramPoster: await asyncio.sleep(2 ** attempt) raise RuntimeError("Telegram retries exhausted") + async def tg_retry_media(self, fn): + """Like tg_retry, but for calls that upload actual file bytes (send_photo/ + send_video/send_media_group). A network timeout there is NOT proof the + message wasn't delivered - large videos can finish server-side after the + client gives up waiting - so blindly resending risks posting duplicates. + Only flood control (an explicit, safe-to-retry signal) gets a retry; + anything else propagates immediately.""" + 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) + return await fn() + async def set_reaction(self, message_id: int) -> None: if not (settings.tg_auto_reaction_enabled and settings.tg_auto_reaction and self.bot): return @@ -158,19 +174,19 @@ class TelegramPoster: item = valid_items[0] fs = FSInputFile(str(item.local_path)) if item.media_type == "photo": - msg = await self.tg_retry( + msg = await self.tg_retry_media( 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( + msg = await self.tg_retry_media( 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( + msgs = await self.tg_retry_media( lambda: self.bot.send_media_group(media=group, **self.chat_kwargs(use_storage)) ) for item, msg in zip(valid_items, msgs): @@ -251,7 +267,14 @@ class TelegramPoster: mid = res.get("message_id") if mid: return [int(mid)] - raise RichMessageUnavailable("sendRichMessage returned no message_id") + # ok=true means Telegram DID create the message, even though we + # couldn't parse its id from this response. Must NOT raise + # RichMessageUnavailable here - that would trigger send_post's + # fallback and post the same content a second time via the + # standard path, landing one "broken" (unparsed-id) message next + # to a normal one. + logger.warning("sendRichMessage returned ok=true but no parseable message_id: {}", payload) + return [] description = str(payload.get("description") or f"HTTP {resp.status}") if "Too Many Requests" in description and isinstance(payload.get("parameters"), dict): @@ -300,7 +323,7 @@ class TelegramPoster: if len(first_group) == 1: item = first_group[0] if item["type"] == "photo": - msg = await self.tg_retry( + msg = await self.tg_retry_media( lambda: self.bot.send_photo( photo=item["media"], caption=first_caption or None, @@ -309,7 +332,7 @@ class TelegramPoster: ) ) else: - msg = await self.tg_retry( + msg = await self.tg_retry_media( lambda: self.bot.send_video( video=item["media"], caption=first_caption or None, @@ -326,7 +349,7 @@ class TelegramPoster: 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( + msgs = await self.tg_retry_media( lambda: self.bot.send_media_group(media=group, **self.chat_kwargs()) ) message_ids.extend(int(m.message_id) for m in msgs) @@ -341,7 +364,7 @@ class TelegramPoster: else InputMediaVideo(media=item["media"]) for item in chunk ] - msgs = await self.tg_retry( + msgs = await self.tg_retry_media( lambda: self.bot.send_media_group(media=g, **self.chat_kwargs()) ) message_ids.extend(int(m.message_id) for m in msgs) @@ -377,19 +400,36 @@ class TelegramPoster: mids.append(int(msg.message_id)) return mids - async def post_to_telegram( + async def upload_media(self, media_items: list[ProcessedMedia]) -> 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 + sending local files directly, same as the pre-split behavior. + """ + 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) + + async def send_post( self, raw_text: str, media_items: list[ProcessedMedia], + file_ids: dict[str, str], vk_url: Optional[str] = None, ) -> tuple[list[int], Optional[str]]: """ - Main Telegram posting routine: + Post stage - assumes upload_media() already ran (file_ids may be empty if no + storage channel is configured, in which case local files are sent directly): 1. Formats text for HTML parse mode and notes any media that couldn't be attached. - 2. Uploads media to the storage channel (if configured) to obtain reusable file_ids. - 3. Tries sendRichMessage (Bot API 10.1+) for a proper collage + rich text. - 4. Falls back to standard aiogram calls (send_photo/send_video/send_media_group) - if rich message is unavailable (no storage channel, old Bot API server, etc). + 2. If the text fits the platform's normal limit (1024 caption chars with media, + 4096 message chars without), sends it via the standard, well-tested + send_photo/send_video/send_media_group/send_message calls directly. + 3. Only reaches for sendRichMessage - a non-standard endpoint - when the text is + too long to fit as a single caption, since that's the one thing the standard + path can't do in a single message (it would otherwise split into a media + message + separate follow-up text messages). + 4. Falls back to the standard path if sendRichMessage is unavailable/fails. """ formatted_text = format_post_text( raw_text, @@ -404,13 +444,10 @@ class TelegramPoster: if note: formatted_text = f"{formatted_text}\n\n{note}" if formatted_text else note - # Obtain file_ids from the storage channel, if configured, to avoid re-uploading - # (also a prerequisite for rich messages, which reference media by file_id). - file_ids: dict[str, str] = {} - if valid_media and self.storage_chat_id: - file_ids = await self.upload_media_for_file_ids(valid_media) + normal_limit = self.caption_limit if valid_media else self.message_limit + fits_normal_limit = len(formatted_text) <= normal_limit - if valid_media and file_ids: + if valid_media and file_ids and not fits_normal_limit: rich_msg = self.build_rich_message(formatted_text, valid_media, file_ids) if rich_msg: try: @@ -429,3 +466,13 @@ class TelegramPoster: if mids: await self.set_reaction(mids[0]) return mids, url + + async def post_to_telegram( + self, + raw_text: str, + media_items: list[ProcessedMedia], + 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)