Commit 79f24f6

@@ -1,5 +1,7 @@
 #!/usr/bin/env python
 # -*- coding: utf-8 -*-
+import asyncio
+import json
 from datetime import timedelta
 from decimal import ROUND_DOWN, Decimal, getcontext
 from pathlib import Path
@@ -183,3 +185,30 @@ async def ffmpeg_h264(client: Client, message: Message, **kwargs):
     h264_path = await convert_to_h264(fpath, force_re_encoding=True)
     await send2tg(client, media_message, media=[{"video": h264_path}], **kwargs)
     await modify_progress(del_status=True, **kwargs)
+
+
+async def ffprobe(client: Client, message: Message, **kwargs):
+    """Get media info."""
+    if not equal_prefix(message.content, prefix=PREFIX.FFPROBE):
+        return
+    media_message = message.reply_to_message if message.reply_to_message else message
+    info = parse_msg(media_message)
+    if info["mtype"] not in ["video", "audio", "photo"]:
+        await send2tg(client, media_message, texts="❌ffprobe不支持该媒体类型", **kwargs)
+        return
+    kwargs["progress"] = kwargs.get("progress", await message.reply_text("⏬正在下载媒体文件...", quote=True))
+    fpath = Path(DOWNLOAD_DIR).joinpath(f"{info['mid']}-{info['cid']}-{info['file_name']}").as_posix()
+    if not Path(fpath).is_file():
+        fpath: str = await client.download_media(media_message, file_name=fpath)  # type: ignore
+    if not Path(fpath).is_file():
+        await modify_progress(text="❌文件下载失败", force_update=True, **kwargs)
+        return
+    try:
+        cmd = ["ffprobe", "-show_streams", "-print_format", "json", fpath]
+        process = await asyncio.create_subprocess_exec(*cmd, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE)
+        stdout_data, _ = await process.communicate()
+        metadata = json.loads(stdout_data)
+        await send2tg(client, media_message, texts=json.dumps(metadata, ensure_ascii=False, indent=2), **kwargs)
+    except Exception as e:
+        await send2tg(client, media_message, texts=f"❌获取媒体信息失败: {e}", **kwargs)
+    await modify_progress(del_status=True, **kwargs)

@@ -107,6 +107,7 @@ class PREFIX:
     TMDB = os.getenv("PREFIX_TMDB", "/tmdb").lower()
     FFMPEG_CUT = os.getenv("PREFIX_FFMPEG_CUT", "/cut").lower()
     FFMPEG_H264 = os.getenv("PREFIX_FFMPEG_H264", "/h264").lower()
+    FFPROBE = os.getenv("PREFIX_FFPROBE", "/ffprobe").lower()
     WATERMARK = os.getenv("PREFIX_WATERMARK", "/wm, /watermark").lower()
     VERSION = os.getenv("PREFIX_VERSION", "/version").lower()
 

@@ -23,7 +23,7 @@ from others.convert_chinese import chinese_conversion
 from others.download_external import download_url_in_message
 from others.extract_audio import extract_audio_file
 from others.favorite import save_favorite, send_favorite
-from others.ffmpeg import ffmpeg_cut, ffmpeg_h264
+from others.ffmpeg import ffmpeg_cut, ffmpeg_h264, ffprobe
 from others.raw_img_file import convert_raw_img_file
 from others.search_google import search_google
 from others.search_ytb import search_youtube
@@ -158,6 +158,7 @@ async def handle_utilities(
     if ffmpeg:
         await ffmpeg_cut(client, message, **kwargs)
         await ffmpeg_h264(client, message, **kwargs)
+        await ffprobe(client, message, **kwargs)
     if watermark:
         await add_watermark(client, message, **kwargs)
     if version:
@@ -448,6 +449,7 @@ def get_social_media_help(chat_id: int | str, ctype: str, prefix: str):
     if permission["ffmpeg"]:
         msg += f"\n✂️**视频切片**: `{PREFIX.FFMPEG_CUT}` 回复视频消息"
         msg += f"\n🎬**视频转码**: `{PREFIX.FFMPEG_H264}` 回复视频消息"
+        msg += f"\nℹ️**媒体信息**: `{PREFIX.FFPROBE}` 获取媒体信息"  # noqa: RUF001
     if permission["watermark"]:
         msg += f"\n💧**添加水印**: `{PREFIX.WATERMARK}` 回复媒体消息"