""" plugins/compress.py -------------------- /compress (used as a REPLY to a video/document file) Flow: 1. User replies to a media message with /compress 2. Bot shows quality buttons (High / Medium / Low / HEVC) 3. Bot shows resolution buttons (480p / 720p / 1080p / Original) 4. download -> ffmpeg re-encode (scale + codec) with a live progress bar -> upload with the user's saved thumbnail/caption -> show size-before vs size-after -> cleanup temp files Shortcuts (skip the buttons): /compress hevc 480p /compress high 1080p /compress medium (resolution stays original) Uses the same safe_edit/safe_delete helpers already added to file_rename.py so a deleted/replaced status message never crashes the task. """ import os import re import time import uuid import shutil import asyncio import logging from pyrogram import Client, filters from pyrogram.types import Message, CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup from helper.database import codeflixbots from helper.utils import humanbytes, TimeFormatter, progress_for_pyrogram from plugins.file_rename import ( safe_edit, safe_delete, safe_answer, _build_caption, detect_media_stream_kind, get_audio_languages, _leech_style_progress, ) from pyrogram.enums import ParseMode logger = logging.getLogger(__name__) # quality preset -> codec/crf/preset settings QUALITY_PRESETS = { "high": {"codec": "libx264", "crf": "20", "preset": "medium", "label": "🟢 High (x264)"}, "medium": {"codec": "libx264", "crf": "24", "preset": "fast", "label": "🟡 Medium (x264)"}, "low": {"codec": "libx264", "crf": "28", "preset": "faster", "label": "🔴 Low (x264)"}, "hevc": {"codec": "libx265", "crf": "26", "preset": "medium", "label": "🟣 HEVC (smallest size)"}, } # resolution preset -> target height (None = keep original resolution) RESOLUTION_PRESETS = { "480p": {"height": 480, "label": "480p"}, "720p": {"height": 720, "label": "720p"}, "1080p": {"height": 1080, "label": "1080p"}, "original": {"height": None, "label": "Original res"}, } # in-memory registry: task_id -> task dict (mirrors the leech task pattern) compress_tasks: dict[str, dict] = {} MAX_CONCURRENT_COMPRESS = 3 def _progress_bar(pct: float, length: int = 20) -> str: pct = max(0, min(100, pct)) filled = int(length * pct / 100) return "▓" * filled + "░" * (length - filled) async def _get_duration_seconds(file_path: str) -> float: ffprobe = shutil.which("ffprobe") if not ffprobe: return 0.0 try: proc = await asyncio.create_subprocess_exec( ffprobe, "-v", "quiet", "-print_format", "json", "-show_format", file_path, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) stdout, _ = await proc.communicate() import json data = json.loads(stdout.decode("utf-8", errors="ignore")) return float(data.get("format", {}).get("duration", 0) or 0) except Exception: return 0.0 def _resolution_buttons(task_id: str) -> InlineKeyboardMarkup: return InlineKeyboardMarkup([ [ InlineKeyboardButton("480p", callback_data=f"cmp_r_480p_{task_id}"), InlineKeyboardButton("720p", callback_data=f"cmp_r_720p_{task_id}"), ], [ InlineKeyboardButton("1080p", callback_data=f"cmp_r_1080p_{task_id}"), InlineKeyboardButton("🚫 Keep Original", callback_data=f"cmp_r_original_{task_id}"), ], [InlineKeyboardButton("❌ Cancel", callback_data=f"cmp_cancel_{task_id}")], ]) # --------------------------------------------------------------------------- # /compress command - must be used as a reply to a video/document # --------------------------------------------------------------------------- @Client.on_message(filters.command("compress") & filters.private) async def compress_command(client: Client, message: Message): replied = message.reply_to_message if not replied or not (replied.video or replied.document): await message.reply_text( "❌ Video/file message ah **reply** pannitu `/compress` type pannunga.\n" "Illa direct ah: `/compress hevc 480p` madhiri kudukalam." ) return active_for_user = sum(1 for t in compress_tasks.values() if t["user_id"] == message.from_user.id) if active_for_user >= MAX_CONCURRENT_COMPRESS: await message.reply_text(f"❌ Max {MAX_CONCURRENT_COMPRESS} compress tasks already running.") return task_id = uuid.uuid4().hex[:6] # allow "/compress hevc 480p" to skip button steps entirely args = message.text.split()[1:] direct_quality = None direct_resolution = None for arg in args: a = arg.strip().lower() if a in QUALITY_PRESETS: direct_quality = a elif a in RESOLUTION_PRESETS: direct_resolution = a compress_tasks[task_id] = { "task_id": task_id, "user_id": message.from_user.id, "chat_id": message.chat.id, "replied_message": replied, "prompt_message": None, "cancelled": False, "proc": None, "status_message": None, "temp_paths": [], } if direct_quality and direct_resolution: status_msg = await message.reply_text(f"⏬ **Downloading...**\n**Task ID:** `{task_id}`") compress_tasks[task_id]["prompt_message"] = status_msg compress_tasks[task_id]["status_message"] = status_msg compress_tasks[task_id]["quality"] = direct_quality compress_tasks[task_id]["resolution"] = direct_resolution asyncio_task = asyncio.create_task(_run_compress_task(client, compress_tasks[task_id])) compress_tasks[task_id]["asyncio_task"] = asyncio_task return if direct_quality: # quality given but no resolution - jump straight to resolution buttons compress_tasks[task_id]["quality"] = direct_quality prompt = await message.reply_text( "📐 **Resolution select pannunga:**", reply_markup=_resolution_buttons(task_id), ) compress_tasks[task_id]["prompt_message"] = prompt return buttons = InlineKeyboardMarkup([ [ InlineKeyboardButton(QUALITY_PRESETS["high"]["label"], callback_data=f"cmp_q_high_{task_id}"), InlineKeyboardButton(QUALITY_PRESETS["medium"]["label"], callback_data=f"cmp_q_medium_{task_id}"), ], [ InlineKeyboardButton(QUALITY_PRESETS["low"]["label"], callback_data=f"cmp_q_low_{task_id}"), InlineKeyboardButton(QUALITY_PRESETS["hevc"]["label"], callback_data=f"cmp_q_hevc_{task_id}"), ], [InlineKeyboardButton("❌ Cancel", callback_data=f"cmp_cancel_{task_id}")], ]) prompt = await message.reply_text( "🎛 **Compress quality select pannunga:**\n\n" "🟢 High (x264) — best quality, slower, bigger size\n" "🟡 Medium (x264) — balanced (recommended)\n" "🔴 Low (x264) — smallest x264 size, fastest\n" "🟣 HEVC — smallest overall size (same quality), but slower to encode\n\n" "_Tip: next time, `/compress hevc 480p` (reply to file) skips these menus._", reply_markup=buttons, ) compress_tasks[task_id]["prompt_message"] = prompt @Client.on_callback_query(filters.regex(r"^cmp_cancel_")) async def compress_cancel(client: Client, query: CallbackQuery): task_id = query.data.replace("cmp_cancel_", "") task = compress_tasks.get(task_id) if not task: await safe_answer(query, "Task already finished / not found.", show_alert=True) return task["cancelled"] = True proc = task.get("proc") if proc and proc.returncode is None: try: proc.kill() except Exception: pass await safe_answer(query, "Cancelling...") target_msg = task.get("status_message") or task.get("prompt_message") await safe_edit(target_msg, f"❌ Cancelled.\n**Task ID:** `{task_id}`") _cleanup_compress_task(task) @Client.on_callback_query(filters.regex(r"^cmp_q_")) async def compress_quality_chosen(client: Client, query: CallbackQuery): match = re.match(r"^cmp_q_(high|medium|low|hevc)_(.+)$", query.data) if not match: return quality, task_id = match.group(1), match.group(2) task = compress_tasks.get(task_id) if not task: await safe_answer(query, "Task expired, please /compress again.", show_alert=True) return await safe_answer(query) task["quality"] = quality await safe_edit( task["prompt_message"], "📐 **Resolution select pannunga:**", _resolution_buttons(task_id), ) @Client.on_callback_query(filters.regex(r"^cmp_r_")) async def compress_resolution_chosen(client: Client, query: CallbackQuery): match = re.match(r"^cmp_r_(480p|720p|1080p|original)_(.+)$", query.data) if not match: return resolution, task_id = match.group(1), match.group(2) task = compress_tasks.get(task_id) if not task or "quality" not in task: await safe_answer(query, "Task expired, please /compress again.", show_alert=True) return await safe_answer(query) task["resolution"] = resolution status_msg = await safe_edit( task["prompt_message"], f"⏬ **Downloading...**\n**Task ID:** `{task_id}`", InlineKeyboardMarkup([[InlineKeyboardButton("❌ Cancel", callback_data=f"cmp_cancel_{task_id}")]]), ) task["status_message"] = status_msg or task["prompt_message"] asyncio_task = asyncio.create_task(_run_compress_task(client, task)) task["asyncio_task"] = asyncio_task def _cleanup_compress_task(task: dict): for path in list(task.get("temp_paths", [])): try: if path and os.path.exists(path): os.remove(path) except Exception as e: logger.warning(f"compress cleanup failed for {path}: {e}") compress_tasks.pop(task["task_id"], None) # --------------------------------------------------------------------------- # Main compress runner # --------------------------------------------------------------------------- async def _run_compress_task(client: Client, task: dict): task_id = task["task_id"] replied = task["replied_message"] user_id = task["user_id"] quality = task["quality"] resolution = task.get("resolution", "original") preset = QUALITY_PRESETS[quality] res_preset = RESOLUTION_PRESETS[resolution] os.makedirs("downloads", exist_ok=True) os.makedirs("compressed", exist_ok=True) input_path = None output_path = None try: # ── Download ── start = time.time() input_path = await client.download_media( replied, file_name=f"downloads/compress_{task_id}_{replied.video.file_name if replied.video else replied.document.file_name}", progress=progress_for_pyrogram, progress_args=("Downloading...", task["status_message"], start), ) task["temp_paths"].append(input_path) if task["cancelled"]: return original_size = os.path.getsize(input_path) base_name, input_ext = os.path.splitext(input_path) input_ext = input_ext if input_ext else ".mkv" # Force mkv if input is MKV to safely retain multiple audio and ASS subs if "mp4" not in input_ext.lower(): input_ext = ".mkv" output_path = f"{base_name}_compressed{input_ext}" task["temp_paths"].append(output_path) duration = await _get_duration_seconds(input_path) # ── ffmpeg re-encode ── def _build_cmd(include_subs: bool): c = ["ffmpeg", "-y", "-i", input_path] if include_subs: c += ["-map", "0"] # keep every stream: all audio languages + subs else: c += ["-map", "0:v", "-map", "0:a?"] # subs dropped (in case of incompatible container) c += ["-c:v", preset["codec"], "-crf", preset["crf"], "-preset", preset["preset"]] if res_preset["height"]: c += ["-vf", f"scale=-2:{res_preset['height']}"] # Use 'copy' if MKV, else explicitly convert audio to AAC for MP4 compatibility if input_ext.lower() == ".mkv": c += ["-c:a", "copy"] else: c += ["-c:a", "aac", "-b:a", "128k"] if include_subs: c += ["-c:s", "copy"] if preset["codec"] == "libx265" and input_ext.lower() == ".mp4": c += ["-tag:v", "hvc1"] # proper HEVC-in-mp4 tag c += ["-progress", "pipe:1", "-nostats", output_path] return c async def _run_ffmpeg(cmd): proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE ) task["proc"] = proc last_update = 0 start_encode = time.time() while True: line = await proc.stdout.readline() if not line: break if task["cancelled"]: proc.kill() return None, b"" line = line.decode("utf-8", errors="ignore").strip() if line.startswith("out_time_ms=") or line.startswith("out_time="): current_us = _parse_ffmpeg_time(line) now = time.time() if now - last_update >= 2 and duration: last_update = now pct = min(100, (current_us / duration) * 100) if duration else 0 elapsed = now - start_encode await safe_edit( task["status_message"], f"⚙️ **Compressing ({preset['label']}, {res_preset['label']})...**\n\n" f"`[{_progress_bar(pct)}]` {pct:.1f}%\n" f"**Elapsed:** {TimeFormatter(int(elapsed * 1000))}\n" f"**Task ID:** `{task_id}`", InlineKeyboardMarkup([[InlineKeyboardButton("❌ Cancel", callback_data=f"cmp_cancel_{task_id}")]]), ) await proc.wait() stderr_data = await proc.stderr.read() return proc.returncode, stderr_data returncode, stderr_data = await _run_ffmpeg(_build_cmd(include_subs=True)) if task["cancelled"]: return if returncode != 0: err_text = stderr_data.decode("utf-8", errors="ignore") if stderr_data else "" if "subtitle" in err_text.lower() or "currently supported in container" in err_text.lower() or "could not find tag" in err_text.lower(): # Common case: source has ASS/SSA subtitles, which mp4 can't hold - retry without subs instead of failing outright logger.warning(f"Compress: subtitle copy failed ({err_text[-200:]}), retrying without subtitles") await safe_edit( task["status_message"], f"⚠️ Subtitle format mp4-ku compatible illa, subs illama retry pannuren...\n**Task ID:** `{task_id}`", ) returncode, stderr_data = await _run_ffmpeg(_build_cmd(include_subs=False)) if task["cancelled"]: return if returncode != 0 or not os.path.exists(output_path): await safe_edit(task["status_message"], f"❌ Compression failed (ffmpeg error).\n**Task ID:** `{task_id}`") return compressed_size = os.path.getsize(output_path) saved_pct = (1 - (compressed_size / original_size)) * 100 if original_size else 0 # ── Upload ── await safe_edit( task["status_message"], f"📤 **Uploading...**\n**Task ID:** `{task_id}`", ) thumb_path = None try: thumb_file_id = await codeflixbots.get_thumbnail(user_id) if thumb_file_id: thumb_path = await client.download_media(thumb_file_id) task["temp_paths"].append(thumb_path) except Exception as e: logger.warning(f"Compress: thumbnail resolve failed, uploading without thumb: {e}") thumb_path = None output_filename = os.path.basename(output_path) languages = await get_audio_languages(output_path, output_filename) custom_caption = await codeflixbots.get_caption(user_id) extra_note = ( f"🗜 Compressed ({preset['label']}, {res_preset['label']})\n" f"Original: {humanbytes(original_size)} → New: {humanbytes(compressed_size)} " f"({saved_pct:.1f}% smaller)" ) base_caption = _build_caption( custom_caption=custom_caption, new_filename=output_filename, file_name=output_filename, original_file_name=os.path.basename(input_path), file_size=compressed_size, duration=int(duration), quality=res_preset["label"], season="", episode="", languages=languages, format_template=output_filename, telegraph_url=None, ) final_caption = f"{base_caption}\n\n{extra_note}" if base_caption else extra_note # respect the user's video/document upload-format setting effective_media_hint = await detect_media_stream_kind(output_path, fallback_hint="video") upload_format = await codeflixbots.get_upload_format(user_id) if upload_format == "auto": effective_type = effective_media_hint or "document" elif upload_format == "video": effective_type = "video" if effective_media_hint in ("video", "document") else "audio" else: effective_type = "document" upload_params = { "chat_id": task["chat_id"], "caption": final_caption, "parse_mode": ParseMode.HTML, "thumb": thumb_path if thumb_path else None, "progress": _leech_style_progress, "progress_args": (task, "Uploading"), } task["_upload_start"] = time.time() if effective_type == "video": await client.send_video(video=output_path, file_name=output_filename, duration=int(duration), **upload_params) elif effective_type == "audio": await client.send_audio(audio=output_path, file_name=output_filename, duration=int(duration), **upload_params) else: await client.send_document(document=output_path, file_name=output_filename, **upload_params) await safe_delete(task["status_message"]) except asyncio.CancelledError: await safe_edit(task["status_message"], f"❌ Cancelled.\n**Task ID:** `{task_id}`") except Exception as e: logger.error(f"Compress task {task_id} failed: {e}") await safe_edit(task["status_message"], f"❌ **Compress failed:** `{e}`\n**Task ID:** `{task_id}`") finally: _cleanup_compress_task(task) def _parse_ffmpeg_time(line: str) -> float: """Parses ffmpeg -progress output lines like 'out_time_ms=12345678' into seconds.""" try: key, val = line.split("=", 1) if key == "out_time_ms": return float(val) / 1_000_000 if key == "out_time": h, m, s = val.split(":") return int(h) * 3600 + int(m) * 60 + float(s) except Exception: pass return 0.0