| """ |
| 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_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_PRESETS = { |
| "480p": {"height": 480, "label": "480p"}, |
| "720p": {"height": 720, "label": "720p"}, |
| "1080p": {"height": 1080, "label": "1080p"}, |
| "original": {"height": None, "label": "Original res"}, |
| } |
|
|
| |
| 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}")], |
| ]) |
|
|
|
|
| |
| |
| |
| @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] |
|
|
| |
| 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: |
| |
| 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) |
|
|
|
|
| |
| |
| |
| 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: |
| |
| 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" |
| |
| 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) |
|
|
| |
| def _build_cmd(include_subs: bool): |
| c = ["ffmpeg", "-y", "-i", input_path] |
| if include_subs: |
| c += ["-map", "0"] |
| else: |
| c += ["-map", "0:v", "-map", "0:a?"] |
| c += ["-c:v", preset["codec"], "-crf", preset["crf"], "-preset", preset["preset"]] |
| if res_preset["height"]: |
| c += ["-vf", f"scale=-2:{res_preset['height']}"] |
| |
| |
| 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"] |
| 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(): |
| |
| 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 |
|
|
| |
| 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 |
|
|
| |
| 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 |
