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)