Feat file info#149
Conversation
Codecov Report❌ Patch coverage is
📢 Thoughts on this report? Let us know! |
|
Здравия всем. Что-то много коммитов на присоединил. Там по сути только последний один по делу. И я понимаю что там могут быть косяки. Форматы разные — не все случаи проверены. |
|
Может надо пересоздать PR, чтобы правильно коммиты подгрузились. Чтоб лишнего не показывали. |
получение метаинформации о файлах без полной загрузки Добавлен модуль file_inspector.py с классами: - FileInfo — pydantic-модель с метаинформацией о файле (format, width, height, duration, fps, sample_rate, bitrate, error_desc). Статус ok/partial/error определяется автоматически по наличию ключевых полей. - FetchPlan — план загрузки файла. Определяет сколько байт читать из начала (initial_head), размер докачки (expand_chunk), лимиты (max_head, min_head) и нужно ли читать хвост (need_tail). Метод from_content_type() строит план по MIME-типу и размеру файла. - RangeReader — базовый класс для чтения файлов частями. Единый интерфейс: async for chunks in reader. Свойства head, tail, head_size, tail_size. - RangeDownloader(HTTP) — загрузчик по URL. Retry при 429/500/502/503/504 с экспоненциальным backoff. Прогрессивная докачка с удвоением чанка. Определяет Content-Type из заголовков, строит план, читает head с докачкой и tail через Range-запросы. - RangeFileReader — читает локальный файл (anyio). Head, tail через seek, докачка по плану. - RangeBytesReader — читает из bytes/BytesIO/NamedBytesIO. Использует memoryview без копирования данных. Для маленьких файлов читает целиком. - FileInspector — высокоуровневая обёртка. Три метода: inspect_url(url) — удалённый файл inspect_file(path) — локальный файл inspect_bytes(data) — уже загруженные байты Возвращает FileInfo. Сохраняет last_file_info, head, tail после инспекции. Парсеры встроены в FileInspector как @classmethod. Поддерживают: Изображения: JPEG, PNG, GIF, WebP (VP8/VP8L/VP8X) Видео: MP4/MOV, AVI, MKV, WEBM, OGV Аудио: MP3, AAC, WAV, WMA, FLAC, OGG, M4A Добавлен метод bot.get_file_info(url) — получение метаинформации через FileInspector без полной загрузки файла. Примеры: 05_media_bot.py — добавлен /info метаинформация о replied-вложении через info = await bot.get_file_info(url) Тесты: - Параметризованные тесты на всех фикстурах из fixtures.json - Фикстуры можно расширять и править с помощью prepare_fixtures.py - inspect_url, inspect_file, inspect_bytes — сравнение результатов - Retry-логика: 503 → retry → успех, исчерпание retry, 404 без retry - Создание сессии когда не передана - HTML-страница → error - Пустой Content-Type → определение формата по байтам
|
В общем, да тесты упали. потому что ребейз сделал. |
fix: JPEG parser fix: bot.get_file_info kwargs для передачи в RangeDownloader
There was a problem hiding this comment.
Pull request overview
Добавляется FileInspector/RangeDownloader для извлечения метаданных медиафайлов (формат, размеры, длительность, битрейты) по URL/локальному файлу/байтам без полной загрузки, а также публичная модель FileInfo и удобный метод bot.get_file_info().
Changes:
- Добавлены
FileInspector,RangeDownloaderи набор парсеров сигнатур/заголовков для популярных форматов медиа. - Добавлена pydantic-модель
FileInfoи экспорт вmaxapi.types, плюс методBot.get_file_info(). - Добавлены тесты с фикстурами и вспомогательный скрипт генерации
fixtures.json; обновлён пример05_media_bot.py.
Reviewed changes
Copilot reviewed 7 out of 10 changed files in this pull request and generated 9 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/test_utils/fileinfo/test_file_info.py | Новые параметризованные тесты FileInspector/RangeDownloader + моки aiohttp. |
| tests/test_utils/fileinfo/prepare_fixtures.py | Скрипт скачивания/генерации фикстур head/tail и expected-полей в fixtures.json. |
| tests/test_utils/fileinfo/init.py | Пакет тестов для fileinfo. |
| maxapi/utils/file_inspector.py | Основная реализация инспекции: планы загрузки, range-ридеры, парсеры форматов. |
| maxapi/types/file_info.py | Новая модель FileInfo со статусом и человекочитаемым выводом. |
| maxapi/types/init.py | Экспорт FileInfo в публичный maxapi.types. |
| maxapi/connection/base.py | Добавлен NamedBytesIO (BytesIO с .name). |
| maxapi/bot.py | Добавлен метод Bot.get_file_info(). |
| examples/05_media_bot.py | Пример дополнен командой /info и авто-инспекцией первого вложения. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| mime = exp.get("mime_type", "") | ||
| ext = mimetypes.guess_extension(mime) or ".bin" | ||
| file_name = f"{name}{ext}" | ||
|
|
||
| file_size = exp.get("file_size") or (len(head) + len(tail)) | ||
|
|
||
| full = bytearray(file_size) | ||
| full[: len(head)] = head | ||
| if tail: | ||
| full[-len(tail) :] = tail |
| return self.tail or b"self.tail empty" | ||
|
|
||
| resp.content.read = read_tail | ||
| resp.read = AsyncMock(return_value=self.tail) |
| async def test_retry_then_success(self): | ||
| """503 → retry → успех.""" | ||
| factory = MockResponseFactory( | ||
| head=b"data", tail=b"", content_type="image/jpeg", file_size=4 | ||
| ) | ||
| meta_resp = factory.make_head_response("url") # для _fetch_meta | ||
| bad = factory.make_head_response("url") | ||
| bad.status, bad.ok = 503, False | ||
| good = factory.make_head_response("url") | ||
| session = AsyncMock() | ||
| session.get = AsyncMock(side_effect=[meta_resp, bad, good]) | ||
|
|
||
| info = await FileInspector().inspect_url( | ||
| "https://x.com/x.jpg", session=session, retry_backoff_factor=0 | ||
| ) | ||
| assert info.status == "error" | ||
| assert session.get.call_count == 3 |
| @staticmethod | ||
| def _m4a_parse_mvhd_duration(data: bytes) -> int | None: | ||
| """Парсит длительность из mvhd атома.""" | ||
| if len(data) < 20: | ||
| return None | ||
|
|
||
| version = data[8] # Смещение 8 байт от начала атома | ||
|
|
||
| if version == 0: | ||
| # timescale (4 bytes) at offset 20 | ||
| if len(data) < 24: | ||
| return None | ||
| timescale = struct.unpack(">I", data[20:24])[0] | ||
| # duration (4 bytes) at offset 24 | ||
| if len(data) < 28: | ||
| return None | ||
| duration = struct.unpack(">I", data[24:28])[0] | ||
| else: # version 1 | ||
| # timescale (4 bytes) at offset 20 | ||
| if len(data) < 24: | ||
| return None | ||
| timescale = struct.unpack(">I", data[20:24])[0] | ||
| # duration (8 bytes) at offset 24 | ||
| if len(data) < 32: | ||
| return None | ||
| duration = struct.unpack(">Q", data[24:32])[0] | ||
|
|
||
| if timescale > 0: | ||
| return duration / timescale | ||
|
|
| lines = [ f"Имя файла: {self.file_name}", | ||
| f"Размер: {self.file_size_human}", | ||
| ] | ||
| _FIELD_LABELS = ( | ||
| ("format", "Формат: {}"), | ||
| ("width", " Размеры: {}×"), | ||
| ("height", None), # добавляется к width | ||
| ("duration", " Длительность: {} сек"), | ||
| ("fps", " Частота кадров: {} к/с"), | ||
| ("sample_rate", " Аудио: {} Гц"), | ||
| ("bitrate_nominal", " Битрейт (номинальный): {} кбит/с"), | ||
| ("bitrate_avg", " Битрейт (средний): {} кбит/с"), | ||
| ) | ||
| # fmt: on | ||
| for field, tmpl in _FIELD_LABELS: | ||
| value = getattr(self, field, None) | ||
| if not value: | ||
| continue | ||
| if field == "height": | ||
| lines[-1] = lines[-1].rstrip(" пикс") + f"×{value} пикс" | ||
| elif tmpl: | ||
| lines.append(tmpl.format(value)) |
There was a problem hiding this comment.
Крестик учёл. А "width" и "height" либо оба присутствуют, либо оба отсутствуют.
There was a problem hiding this comment.
Такие штуки лучше собирать через список, а потом в "×".join(...) закидывать. Тогда гарантированно 1 раз проставляться будет и крайний случай с висячими тоже сам по себе отвалится (даже если не возможен).
|
Olegt0rr, спасибо. Я тут посмотрел аналогию FileInfo в телеграмм. Получилось мало общего (в том смысле, что у нас намного шире представленная информация. Я бы даже назвал этот класс MediaInfo, но обычные файлы тоже можно посмотреть размер и тип (zip, rar, pdf, и др.)) |
Доработка выглядит весьма адекватно. По поводу того брать или нет – здесь финальное слово за @love-apples |
- file_size = exp.get("file_size") or (len(head) + len(tail)) упадёт с TypeError
- В mock tail-чтения при tail is None возвращается b"self.tail empty"
- Docstring теста
- _m4a_parse_mvhd_duration(data: bytes) -> float
- опечатки и орфография
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 7 out of 10 changed files in this pull request and generated 6 comments.
Comments suppressed due to low confidence (1)
tests/test_utils/fileinfo/prepare_fixtures.py:61
- Опечатка в комментарии: «Провеирть» → «Проверить».
# Провеирть ключи. Совпадающие заменят данные, новые добавят.
|
|
||
| async def inspect_bytes( | ||
| self, | ||
| data: bytes | BytesIO, |
There was a problem hiding this comment.
исправлено.
Кстати. Я думаю NamedBytesIO надо засунуть в types, а не как сейчас в connection.base
Добро?
| for attempt in range(self.max_retries + 1): | ||
| try: | ||
| headers = dict(self.headers) | ||
| if allow_range and range_bytes: | ||
| headers["Range"] = f"bytes=-{range_bytes}" | ||
|
|
||
| response = await self.session.get(url, headers=headers) | ||
| except ClientConnectionError as e: |
There was a problem hiding this comment.
Это не баг, а фича. Документируем
There was a problem hiding this comment.
Это не баг, а фича. Документируем.
Так же добавил проверку и спец флаг allow_external_auth для намеренного разрешения таких действий
There was a problem hiding this comment.
Я бы ограничения какие-то ставил всё же. Не понятно какая ссылка полетит в этот метод
There was a problem hiding this comment.
Ну ограничения заключаются в том, что если намеренно не передать allow_external_auth=True, то аутентификация мо;ет быть передана только на доверенные серверы _TRUSTED_DOMAINS = {"oneme.ru", "okcdn.ru"}. На другие сервера не пойдёт и будет логироваться как ошибка..
https://github.com/love-apples/maxapi/pull/149/changes#diff-88c6615f8ab45d8a9290ea86690f48c0fa0868044fcc173860049e97785072c8R509
По-умолчанию сессия создаётся (то есть не использует сессию бота). то есть чтобы отправить чего-то не туда, ещё попыхтеть придётся. Намерено передать сессию или заголовки, и намерено указать, что нужна авторизация на не доверенном сервере.
| # Получаем метаинформацию и строим план | ||
| if not self._fetched_meta: | ||
| await self._fetch_meta() | ||
| self._meta = cast(FileMeta, self._meta) | ||
| self.content_type = self._meta.content_type | ||
| self.file_name = self._meta.file_name | ||
| self.file_size = self._meta.file_size | ||
| self.plan = FetchPlan.from_content_type( | ||
| self._meta.content_type, | ||
| self._meta.file_size, | ||
| ) |
There was a problem hiding this comment.
Вот тут подумал и решил, что надо так:
-
Удалить
FetchPlanкак класс. Заменить константами:INITIAL_HEAD=4096, MAX_HEAD=256000, EXPAND_CHUNK=4096, MAX_TAIL=65536. -
Переработать
RangeDownloader.__aiter__:
- Сразу получаем минимальные GET 4096 байт + meta из заголовков (Content-Type, Length, filename). yield чанк
- Отправляем в парсер, он возвращает _need_head / _need_tail
_need_head == -1 → докачка удвоением (используем существующий)
_need_head > 0 → докачка до конкретного размера (используем существующий)
_need_tail > 0 → Отдельный Range-запрос
Без ключей — хватило
Цикл пока парсер просит ещё.
-
Адаптировать FileInspector._inspect под новую логику.
-
Выигрыш:
- Меньше кода. Меньше Классов. Понятная логика для всех случаев.
- Надёжнее: план строится по реальным данным, а не по MIME-типу от сервера (который может быть application/octet-stream или врать).
- Точнее: парсер знает сколько ему нужно или просто просит ещё.
- Меньше запросов: meta + head в одном GET (вместо двух).
- Проще расширять: новый формат — только парсер, без правок плана.
There was a problem hiding this comment.
Но самое простое решение не закрывать соединение после _fetch_meta и переиспользовать его для head + докачка. А для tail сделать отдельное RANGE соединение, которое можно сразу закрыть.
There was a problem hiding this comment.
Реализовал пока самое простое решение
There was a problem hiding this comment.
Реализовал сложное решение. Пока под вопросом как лучше.
- mp4, m4a check - опечатки
RangeDownloader теперь не делает отдельный GET в _fetch_meta(), а сохраняет ответ для использования при чтении head
Проверка на содержание Authorization или Cookie Параметр allow_external_auth для разрешения + docstrings
Упрощение (уменьшение) кода: - mp4 m4a теперь общий парсер - удалён класс MediaChunks за ненадобностью fix: prepare_fixtures отключет полную загрузку файла в фикстуру. fix: передача параметра allow_external_auth + docstrings
fix: test_retry_then_success update fixtures.json
Удалить FetchPlan как класс. Заменить константами: INITIAL_HEAD=4096, MAX_HEAD=256000, EXPAND_CHUNK=4096, MAX_TAIL=65536. Переработать RangeDownloader.__aiter__: Сразу получаем минимальные GET 4096 байт + meta из заголовков (Content-Type, Length, filename). yield чанк Отправляем в парсер, он возвращает _need_head / _need_tail _need_head == -1 → докачка удвоением (используем существующий) _need_head > 0 → докачка до конкретного размера (используем существующий) _need_tail > 0 → Отдельный Range-запрос Без ключей — хватило Цикл пока парсер просит ещё. Адаптировать FileInspector._inspect под новую логику. Выигрыш: Меньше кода. Меньше Классов. Понятная логика для всех случаев. Надёжнее: план строится по реальным данным, а не по MIME-типу от сервера (который может быть application/octet-stream или врать). Точнее: парсер знает сколько ему нужно или просто просит ещё. Меньше запросов: meta + head в одном GET (вместо двух). Проще расширять: новый формат — только парсер, без правок плана. + Переименованы поля: FileInfo error_desc -> parse_note full_read_limit -> full_read_threshold
|
Olegt0rr попроси Copilot проанализировать. Там сейчас прям сильно поменялось всё. |
|
love-apples, У меня ещё идея, может это дополнение реализовать в отдельном пакете а тут подключить как зависимость? Но у меня опыта мало в настройке выкладывания на pip. |
| async def _fetch_meta(self): | ||
| """Получает метаинформацию с retry.""" | ||
| if ( | ||
| ("Authorization" in self.headers or "Cookie" in self.headers) | ||
| and not self._is_trusted_url | ||
| and not self._allow_external_auth | ||
| ): |
| async def _read_response( | ||
| self, response: ClientResponse, size: int | ||
| ) -> bytes: | ||
| actual = min(size, self.max_total) | ||
| data = b"" | ||
| while len(data) < actual: | ||
| chunk = await response.content.read(actual - len(data)) | ||
| if not chunk: | ||
| break | ||
| data += chunk | ||
| return data |
| response = self._response | ||
| if not response: | ||
| raise RuntimeError( | ||
| "Response отсутвует. Сначала нужно запросить _fetch_meta()" |
| Args: | ||
| path: Путь к файлу. | ||
| full_read_threshold: Файлы меньше этого размера читаются целиком. | ||
| Установите в ноль, чтоюы отключить полное чтение. |
| mime = exp.get("mime_type", "") | ||
| ext = mimetypes.guess_extension(mime) or ".bin" | ||
| file_name = f"{name}{ext}" | ||
|
|
||
| file_size = exp.get("file_size") or (len(head) + len(tail) if tail else 0) | ||
|
|
||
| full = bytearray(file_size) | ||
| full[: len(head)] = head | ||
| if tail: | ||
| full[-len(tail) :] = tail |
There was a problem hiding this comment.
OK. Это и имелось ввиду. Условие написать в скобках: (len(head) + (len(tail) if tail else 0))
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
fix: Опечатка в докстринге: «чтоюы» → «чтобы».
+ Проверка на куки конкретного URL
Теперь подсчитывает всё скачанное.
|
Извиняюсь за большое количество комитов. Эксперимент. Все правки через браузер без среды тестирования. Фух. |
|
@Pankovea привет, поправь пожалуйста конфликт и покрой тестами) |
FileInspector — метаинформация о медиафайлах без полной загрузки
Что это
Быстрое определение параметров медиафайла по URL, локальному пути или байтам.
Работает без дополнительных библиотек — анализирует сигнатуры и заголовки
в первых и последних килобайтах файла. Скачивает минимум данных, докачивает
только если не хватило.
Было навеяно
get_fileиз Telegram API, в частности узнать размер файла, но разрослось в Media Info с информацией о размере кадра, длительности, частоты кадров, битрейте.Примеры использования
Как работает
План загрузки. По MIME-типу и размеру файла определяется стратегия:
сколько читать из начала, нужен ли хвост, с каким шагом докачивать.
Например, для AVI — 8 КБ начала + докачка по 4 КБ до 64 КБ.
Для MP4 — 8 КБ начала + 8 КБ хвоста (moov в конце).
Загрузка. RangeDownloader делает Range-запросы, докачивает
прогрессивно (удвоение чанка: 8 → 16 → 32 → 64 КБ), повторяет
при 429/5xx ошибках. Всё в одном keep-alive соединении.
Парсинг. Сигнатуры байт определяют формат, из заголовков
извлекаются размеры, длительность, FPS, sample rate, битрейт.
То есть по байтам (без знания имени и mimetype файла)
можно узнать что там за файл.
Результат. FileInfo — pydantic-модель. Статус ok если всё
найдено, partial если не хватило данных, error при ошибке.
Поддерживаемые форматы
Изображения: JPEG, PNG, GIF, WebP (VP8/VP8L/VP8X)
Видео: MP4/MOV, AVI, MKV, WebM, OGV
Аудио: MP3, AAC, WAV, WMA, FLAC, OGG, M4A
Что добавлено
FileInfo — модель с полями width, height, duration, fps, sample_rate,
bitrate_nominal, bitrate_avg, format, status, error_desc.
Метод str для красивого вывода.
FileInspector — три метода: inspect_url(), inspect_file(),
inspect_bytes(). Работает без доп. библиотек.
FetchPlan.from_content_type() — стратегия загрузки по MIME-типу.
RangeDownloader — HTTP-загрузчик с retry и прогрессивной докачкой.
RangeFileReader / RangeBytesReader — локальные файлы и байты.
bot.get_file_info(url) — удобный метод для ботов.
Дополнен пример 05_media_bot.py:
/info ответом на вложение — присылает для метаданные.
На все входящие вложения присылается подробные данные о формате и параметрах медиа.
Тесты
На всех форматах с моками HTTP, retry, ошибок.
Фикстуры в fixtures.json. Содержат в себе head и tail байты
реальных файлов для тестирования парсинга.
Скрипт prepare_fixtures.py для обновления фикстур
по ссылке в интернете или из локального файла.