Feat file info

examples/05_media_bot.py

@@ -8,16 +8,19 @@
 - Обработку входящих вложений: image, file, audio, video
 - Пересылку сообщений через message.forward()
 - SenderAction.SENDING_PHOTO / SENDING_VIDEO / SENDING_FILE
+- FileInspector — получение метаинформации о файле без полной загрузки
 
 Команды:
     /photo     — отправить тестовое изображение из файла
     /buffer    — отправить изображение из буфера (байты)
     /upload    — загрузить медиа заранее, затем отправить
+    /info      — метаинформация о replied-вложении (FileInspector)
 
 Любой файл/фото/аудио/видео от пользователя пересылается обратно
 с описанием типа вложения.
 
-Аналог Telegram: send_photo, send_document, send_audio, forward_message
+Аналог Telegram: send_photo, send_document, send_audio, forward_message,
+get_file
 
 Запуск:
     MAX_BOT_TOKEN=your_token python 05_media_bot.py
@@ -31,11 +34,14 @@
 from pathlib import Path
 from typing import TYPE_CHECKING
 
+import aiohttp
+
 # Опционально: загрузка .env, если установлен python-dotenv
 with contextlib.suppress(ImportError):
     from dotenv import load_dotenv
 
     load_dotenv()
+
 from maxapi import Bot, Dispatcher, F
 from maxapi.enums.sender_action import SenderAction
 from maxapi.filters.command import Command, CommandStart
@@ -50,13 +56,34 @@
     from maxapi.types.updates.message_created import MessageCreated
 
 logging.basicConfig(level=logging.INFO)
+log = logging.getLogger()
 
 bot = Bot()
 dp = Dispatcher()
 
 PHOTO_PATH = Path(__file__).resolve().parent.parent / "logo.png"
 
 
+# ============================================================================
+# Хелперы
+# ============================================================================
+
+
+def _get_first_attachment_url(attachments) -> str | None:
+    """Извлекает URL из первого вложения."""
+    if not attachments:
+        return None
+    first = attachments[0]
+    if hasattr(first, "url") and first.url:
+        return first.url
+    return None
+
+
+# ============================================================================
+# Команды
+# ============================================================================
+
+
 @dp.message_created(CommandStart())
 async def on_start(event: MessageCreated) -> None:
     """Приветствие с описанием команд."""
@@ -65,7 +92,8 @@ async def on_start(event: MessageCreated) -> None:
         "Команды:\n"
         "/photo  — фото из файла\n"
         "/buffer — фото из буфера\n"
-        "/upload — предзагрузка медиа\n\n"
+        "/upload — предзагрузка медиа\n"
+        "/info — метаинформация о replied-вложении\n\n"
         "Пришли мне любой файл, фото, аудио или видео — "
         "я расскажу, что получил, и перешлю обратно."
     )
@@ -98,6 +126,7 @@ async def cmd_buffer(event: MessageCreated) -> None:
     chat_id = event.message.recipient.chat_id
     if chat_id is None:
         return
+
     await bot.send_action(chat_id=chat_id, action=SenderAction.SENDING_PHOTO)
 
     # Читаем обычный PNG в память. В реальном проекте здесь может быть
@@ -138,7 +167,53 @@ async def cmd_upload(event: MessageCreated) -> None:
     )
 
 
-# ── Обработка входящих вложений ────────────────────────────────────────────
+@dp.message_created(Command("info"))
+async def cmd_info(event: MessageCreated) -> None:
+    """Получение метаинформации о файле через bot.get_file_info().
+
+    Ответьте командой на сообщение с вложением.
+    """
+    chat_id = event.message.recipient.chat_id
+    if chat_id is None:
+        return
+
+    replied_body = event.message.link.message if event.message.link else None
+    if not replied_body or not replied_body.attachments:
+        await event.message.answer(
+            "ℹ️ Ответьте этой командой на сообщение с файлом."
+        )
+        return
+
+    url = _get_first_attachment_url(replied_body.attachments)
+    if not url:
+        await event.message.answer("⚠️ Не удалось получить URL вложения.")
+        return
+
+    await bot.send_action(chat_id=chat_id, action=SenderAction.SENDING_FILE)
+
+    try:
+        info = await bot.get_file_info(url, timeout=10)
+    except Exception as e:
+        log.error("Ошибка инспекции: %s", e)
+        await event.message.answer("⚠️ Не удалось определить метаинформацию")
+        return
+
+    if info.status == "error":
+        # info.status == "error" только если ничего не получилось определить,
+        # Даже минимально: info.format is None
+        # info.parse_note содерит описание ошибки
+        await event.message.answer(f"⚠️ {info.parse_note}")
+        return
+
+    # Всё, что получилось узнать о файле — в строку через str(FileInfo)
+    answer = str(info)
+
+    await event.message.answer(answer)
+
+
+# ============================================================================
+# Обработка входящих вложений
+# ============================================================================
 
 
 @dp.message_created(F.message.body.attachments)
@@ -169,13 +244,24 @@ async def on_attachment(event: MessageCreated) -> None:
     chat_id = event.message.recipient.chat_id
     if chat_id is None:
         return
+
     await bot.send_action(chat_id=chat_id, action=action)
 
     # Информируем пользователя о полученном вложении
     count = len(attachments)
-    await event.message.answer(
-        f"Получено {count} вложение(й), тип: {label}. Пересылаю..."
-    )
+    reply_txt = f"Получено {count} вложение(й), тип: {label}.\n\n"
+
+    # Пытаемся получить метаинформацию через FileInspector
+    url = _get_first_attachment_url(attachments)
+    if url:
+        try:
+            info = await bot.get_file_info(url, timeout=3)
+            reply_txt += f"Первое вложение:\n{info}"
+        except aiohttp.ClientError:
+            # Не дошло даже до HTTP (нет сети, DNS)
+            reply_txt += "Не удалось получить подробности:\nСетевая ошибка"
+    reply_txt += "Пересылаю..."
+    await event.message.answer(reply_txt)
 
     # Пересылаем оригинальное сообщение обратно
     await event.message.forward(chat_id=chat_id)

maxapi/bot.py

@@ -42,6 +42,7 @@
 from .methods.send_message import SendMessage
 from .methods.subscribe_webhook import SubscribeWebhook
 from .methods.unsubscribe_webhook import UnsubscribeWebhook
+from .utils.file_inspector import FileInspector
 from .utils.message import process_input_media
 
 if TYPE_CHECKING:
@@ -80,6 +81,7 @@
     from .types.attachments.video import Video
     from .types.chats import Chat, ChatMember, Chats
     from .types.command import BotCommand
+    from .types.file_info import FileInfo
     from .types.input_media import InputMedia, InputMediaBuffer
     from .types.message import Message, Messages, NewMessageLink
     from .types.updates.message_callback import MessageForCallback
@@ -736,6 +738,42 @@ async def get_video(self, video_token: str) -> Video:
 
         return await GetVideo(bot=self, video_token=video_token).fetch()
 
+    async def get_file_info(
+        self, url: str, *, timeout: int = 10, **kwargs
+    ) -> FileInfo:
+        """
+        Получает метаинформацию о файле по URL.
+
+        Аналог ``telegram.Bot.get_file``, но с расширенными полями
+        (размеры, длительность, битрейт и т.д.). Работает с внутренними
+        и внешними URL. Для загрузки используются HTTP Range-запросы
+        (обычно 2–128 КБ вместо полного файла).
+
+        Args:
+            url: URL файла.
+            timeout: Таймаут HTTP-запроса в секундах.
+
+            kwargs:
+            - session : aiohttp-сессия (создаётся при ``None``).
+                Если вам нужно отправить авторизацию (``Authorization``,
+                ``Cookie``) на сторонний URL, укажите
+                ``allow_external_auth=True``. Без этого флага авторизация
+                отправляется только на доверенные домены
+                (``oneme.ru``, ``okcdn.ru``).
+            - max_total : Максимальный объём скачанных данных (байт).
+            - max_retries : Число повторных попыток при ``retry_on_statuses``.
+            - retry_on_statuses : HTTP-статусы, при которых повторять запрос.
+            - retry_backoff_factor : Множитель задержки между попытками
+                (1.0 → 1с, 2с, 4с).
+            - allow_external_auth : Разрешить отправку авторизации на
+                сторонние домены (по умолчанию только oneme.ru/okcdn.ru).
+
+        Returns:
+            FileInfo: Метаинформация о файле.
+        """
+        inspector = FileInspector()
+        return await inspector.inspect_url(url, timeout=timeout, **kwargs)
+
     async def send_callback(
         self,
         callback_id: str,

maxapi/types/__init__.py

@@ -27,6 +27,7 @@
     "DialogMuted",
     "DialogRemoved",
     "DialogUnmuted",
+    "FileInfo",
     "FromUserRef",
     "Icon",
     "InputMedia",
@@ -123,6 +124,7 @@
 from ..types.updates.user_added import UserAdded
 from ..types.updates.user_removed import UserRemoved
 from ..types.users import ChatAdmin, User
+from .file_info import FileInfo
 from .input_media import InputMedia, InputMediaBuffer
 
 if TYPE_CHECKING:

maxapi/types/file_info.py

@@ -0,0 +1,139 @@
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict
+
+
+class FileInfo(BaseModel):
+    """
+    Метаинформация о медиафайле.
+
+    Attributes:
+        url: URL или путь к источнику.
+        mime_type: MIME-тип (по сигнатуре или из заголовка).
+        file_name: Имя файла.
+        file_size: Размер в байтах.
+        width: Ширина кадра (изображение/видео).
+        height: Высота кадра (изображение/видео).
+        duration: Длительность в секундах (аудио/видео).
+        fps: Частота кадров (видео).
+        sample_rate: Частота дискретизации (аудио), Гц.
+        bitrate_nominal: Номинальный битрейт из метаданных, кбит/с.
+        bitrate_avg: Средний битрейт (размер / длительность), кбит/с.
+        parse_note: Описание ошибки или предупреждения от парсера.
+        format: Определённый формат контейнера/кодека (по сигнатуре).
+        status: Результат инспекции (задаёт парсер или inspect).
+    """
+
+    model_config = ConfigDict(frozen=True)
+
+    url: str
+    mime_type: str = ""
+    file_name: str = ""
+    file_size: int | None = None
+    width: int | None = None
+    height: int | None = None
+    duration: float | None = None
+    fps: float | None = None
+    sample_rate: int | None = None
+    bitrate_nominal: int | None = None
+    bitrate_avg: int | None = None
+    status: Literal["ok", "partial", "error"] = "error"
+    parse_note: str = ""
+    format: (
+        Literal[
+            "PNG",
+            "JPEG",
+            "GIF",
+            "WEBP",
+            "WEBP/VP8X",
+            "WEBP/VP8",
+            "WEBP/VP8L",
+            "MP4",
+            "AVI",
+            "MKV",
+            "WEBM",
+            "OGG",
+            "OGV",
+            "M4A",
+            "MP3",
+            "AAC",
+            "WAV",
+            "WMA",
+            "FLAC",
+        ]
+        | None
+    ) = None
+
+    @property
+    def has_dimensions(self) -> bool:
+        """True, если известны ширина и высота."""
+        return self.width is not None and self.height is not None
+
+    @property
+    def is_image(self) -> bool:
+        """True, если MIME-тип относится к изображению."""
+        return self.mime_type.startswith("image/")
+
+    @property
+    def is_audio(self) -> bool:
+        """True, если MIME-тип относится к аудио."""
+        return self.mime_type.startswith("audio/")
+
+    @property
+    def is_video(self) -> bool:
+        """True, если MIME-тип относится к видео."""
+        return self.mime_type.startswith("video/")
+
+    @property
+    def file_size_human(self) -> str:
+        """Размер файла в человекочитаемом виде."""
+        if self.file_size is None:
+            return "неизвестно"
+        if self.file_size < 1024:
+            return f"{self.file_size} байт"
+        if self.file_size < 1_048_576:
+            return f"{self.file_size / 1024:.0f} КБ"
+        if self.file_size < 1_073_741_824:
+            return f"{self.file_size / 1_048_576:.1f} МБ"
+        return f"{self.file_size / 1_073_741_824:.2f} ГБ"
+
+    def __eq__(self, other: object) -> bool:
+        """Сравнение без учёта ``url`` и ``file_name``."""
+        if not isinstance(other, FileInfo):
+            return NotImplemented
+        s = self.model_dump()
+        o = other.model_dump()
+        del s["url"]
+        del o["url"]
+        del s["file_name"]
+        del o["file_name"]
+        return s == o
+
+    def __str__(self) -> str:
+        """Форматированная строка для вывода пользователю."""
+        lines = []
+        if self.file_name:
+            lines.append(f"Имя файла: {self.file_name}")
+        else:
+            lines.append("[Без имени]")
+        lines.append(f"Размер: {self.file_size_human}")
+        if self.format:
+            lines.append(f"Формат: {self.format}")
+        if self.width and self.height:
+            lines.append(f"Размеры: {self.width}×{self.height} пикс")
+        if self.duration:
+            lines.append(f"Длительность: {self.duration} сек")
+        if self.fps:
+            lines.append(f"Частота кадров: {self.fps} к/с")
+        if self.sample_rate:
+            lines.append(f"Аудио: {self.sample_rate} Гц")
+        if self.bitrate_nominal:
+            lines.append(
+                f"Битрейт (номинальный): {self.bitrate_nominal} кбит/с"
+            )
+        if self.bitrate_avg:
+            lines.append(f"Битрейт (средний): {self.bitrate_avg} кбит/с")
+        if self.parse_note:
+            lines.append(f"⚠️ {self.parse_note}")
+
+        return "\n".join(lines)