| import os |
| import re |
| import shutil |
| import asyncio |
| import logging |
| from pyrogram import Client, filters |
| from pyrogram.enums import ParseMode |
| from pyrogram.types import Message |
| from pyrogram.errors import MessageNotModified |
| from helper.telegraph_utils import get_raw_mediainfo_json, get_mediainfo_text, upload_mediainfo, build_mediainfo_cli_text |
|
|
| logger = logging.getLogger(__name__) |
|
|
| @Client.on_message(filters.private & filters.command("mediainfo")) |
| async def mediainfo_command(client, message: Message): |
| """Generate and return MediaInfo for the replied file.""" |
| if not message.reply_to_message: |
| return await message.reply_text( |
| "⚠️ **Reply to a video, audio, or document file** with /mediainfo to extract details.") |
|
|
| reply = message.reply_to_message |
| if not (reply.video or reply.document or reply.audio): |
| return await message.reply_text("⚠️ Replied message must contain a video, audio, or document file.") |
|
|
| msg = await message.reply_text("⚡ **Analyzing media streams... Please wait.**") |
| |
| |
| if reply.video: |
| ext = ".mp4" |
| filename = reply.video.file_name or "video.mp4" |
| elif reply.audio: |
| ext = ".mp3" |
| filename = reply.audio.file_name or "audio.mp3" |
| elif reply.document: |
| filename = reply.document.file_name or "document.mkv" |
| ext = os.path.splitext(filename)[1] or ".mkv" |
| else: |
| filename = "file.mp4" |
| ext = ".mp4" |
|
|
| if reply.video: |
| size = reply.video.file_size |
| elif reply.audio: |
| size = reply.audio.file_size |
| elif reply.document: |
| size = reply.document.file_size |
| else: |
| size = None |
|
|
| temp_path = f"downloads/mediainfo_{reply.id}{ext}" |
| os.makedirs("downloads", exist_ok=True) |
| |
| try: |
| |
| try: |
| await msg.edit("📋🐄 **Downloading media headers...**") |
| except MessageNotModified: |
| pass |
| file_path = await client.download_media(reply, file_name=temp_path) |
| if not file_path or not os.path.exists(file_path): |
| return await msg.edit("❌ Failed to download replied media.") |
|
|
| try: |
| await msg.edit("⚙️ **Extracting audio/video streams...**") |
| except MessageNotModified: |
| pass |
|
|
| |
| is_raw_flag = bool(re.search(r"(-|--)r", message.text or "")) |
|
|
| if is_raw_flag: |
| |
| cli_text = await build_mediainfo_cli_text(file_path, filename=filename, provided_size=size) |
| if not cli_text: |
| return await msg.edit("❌ Failed to generate raw Mediainfo output.") |
| txt_path = f"{temp_path}.txt" |
| with open(txt_path, 'w', encoding='utf-8') as f: |
| f.write(cli_text) |
| try: |
| await msg.edit("📄 Sending raw Mediainfo...") |
| except MessageNotModified: |
| pass |
| await client.send_document(chat_id=message.chat.id, document=txt_path, caption=f"Mediainfo: {filename}") |
| try: |
| os.remove(txt_path) |
| except: |
| pass |
| else: |
| raw_data = await get_raw_mediainfo_json(file_path) |
| if raw_data: |
| info_text = get_mediainfo_text(raw_data, filename) |
| if len(info_text) > 3900: |
| telegraph_url = await upload_mediainfo(file_path, filename) |
| if telegraph_url: |
| try: |
| await msg.edit( |
| f"📄 MediaInfo is too long for a chat message — full report here:\n{telegraph_url}", |
| disable_web_page_preview=False) |
| except MessageNotModified: |
| pass |
| else: |
| |
| txt_path = f"{temp_path}.txt" |
| with open(txt_path, 'w', encoding='utf-8') as f: |
| f.write(re.sub(r'<[^>]+>', '', info_text)) |
| await client.send_document(chat_id=message.chat.id, document=txt_path, caption=f"Mediainfo: {filename}") |
| try: |
| os.remove(txt_path) |
| except Exception: |
| pass |
| else: |
| try: |
| await msg.edit(info_text, parse_mode=ParseMode.HTML, disable_web_page_preview=False) |
| except MessageNotModified: |
| pass |
| else: |
| await msg.edit("❌ Failed to parse media stream metadata.") |
|
|
| except Exception as e: |
| logger.error(f"Error in mediainfo handler: {e}", exc_info=True) |
| try: |
| await msg.edit(f"❌ Error: {str(e)}") |
| except MessageNotModified: |
| pass |
|
|
| finally: |
| if os.path.exists(temp_path): |
| try: |
| os.remove(temp_path) |
| except: |
| pass |
|
|
