Перейти к содержанию

Для разработчиков

Внутреннее устройство url-media-probe: архитектура, потоки данных, как добавлять новые форматы.

Структура проекта

url_media_probe/
├── __init__.py              # Публичный API: MediaProbe, MediaInfo, readers
├── media_probe.py           # Главный класс — цикл инспекции (_probe)
├── media_info.py            # Frozen dataclass MediaInfo — результат
├── range_readers.py         # Абстракция чтения: RangeDownloader / RangeFileReader / RangeBytesReader
├── py.typed                 # PEP 561 маркер
└── parsers/
    ├── __init__.py           # parse_media_dimensions() — точка входа в парсеры
    ├── _types.py             # ParseResult — контракт парсер → MediaProbe
    ├── _constants.py         # Лимиты чтения (MAX_HEAD, MAX_TAIL и т.д.)
    ├── image.py              # ImageParser: PNG, JPEG, GIF, WEBP
    ├── video.py              # VideoParser: MP4/M4A, AVI, MKV, WebM, OGV, OGG, HLS, M3U
    └── audio.py              # AudioParser: MP3, WAV, FLAC, AAC, WMA (+ делегирование в video для OGG/M4A)

Архитектура: поток от вызова до результата

Пользователь
    │
    ▼
MediaProbe.from_url() / from_file() / from_bytes()
    │
    │  1. Создаёт RangeReader (Downloader / File / Bytes)
    │
    ▼
MediaProbe._probe(reader)
    │
    │  2. Цикл: reader отдаёт head (и опционально tail)
    │     reader.__aiter__ → reader.head, reader.tail, reader.meta
    │
    │  3. parse_media_dimensions(head, tail, file_size)
    │     ├── ImageParser.parse()   → ParseResult | None
    │     ├── VideoParser.parse()   → ParseResult | None
    │     └── AudioParser.parse()   → ParseResult | None
    │
    │  4. _split_parse_result(raw) → dims dict, needs dict, status
    │     Если status == "ok" → _build_media_info() → MediaInfo
    │     Если status == "partial" → reader.pending_needs = needs → goto 2
    │
    ▼
MediaInfo (frozen dataclass)

ParseResult — контракт парсер → MediaProbe

url_media_probe/parsers/_types.py

Типизированный dataclass, который каждый парсер возвращает вместо dict[str, Any]:

@dataclass
class ParseResult:
    format: str | None = None          # "PNG", "MP4", "AAC" ...
    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     # кбит/с (вычисляется из size/duration)
    video_codec: str | None = None     # "h264", "vp8", "av1" ...
    video_codec_long: str | None = None
    audio_codec: str | None = None     # "aac", "mp3", "opus" ...
    audio_codec_long: str | None = None
    status: Literal["ok", "partial"] = "partial"
    parse_note: str = ""
    need_head: int = 0                 # -1 = нужен head, >0 = нужен head такого размера
    need_tail: int = 0                 # 0 = не нужен, >0 = нужен tail (от конца файла, назад)
    need_slice: tuple[int, int] | None = None  # (offset, size): слайс от известной позиции, вперёд
    need_url: str = ""                 # non-empty = нужна докачка по другому URL

Свойство ok: True если status == "ok".

Парсер возвращает ParseResult(status="ok") когда данных достаточно, или ParseResult(status="partial", need_head=N) когда нужно больше данных.

Поля need_head / need_tail / need_slice — взаимодополняющие запросы на докачку: - need_tail=N — конец файла неизвестен (например moov у потокового MP4 с mdat size=0): reader заякоривает tail на EOF и растёт назад, докачивая только недостающий префикс. - need_slice=(offset, size) — позиция вычислена из head (например начало moov = конец mdat): reader заякоривает tail на offset и растёт вперёд, при смене якоря буфер сбрасывается.

RangeReader — абстракция источника данных

url_media_probe/range_readers.py

Три реализации, единый интерфейс __aiter__. Лимиты чтения max_total (максимум байт) и max_ratio объявлены один раз в базовом классе RangeReader и наследуются всеми реализациями:

Класс Источник Лимиты по умолчанию
RangeDownloader HTTP URL max_total=256 000, max_ratio=0.1
RangeFileReader Локальный файл max_total=20 971 520, max_ratio=1.0 (чтение целиком)
RangeBytesReader bytes / BytesIO max_total=20 971 520, max_ratio=1.0 (чтение целиком)

Каждый reader при итерации заполняет: - reader.head — начало файла (bytes) - reader.tail — единый буфер хвоста (bytes): либо конец файла (якорь на EOF), либо слайс от вычисленной парсером позиции (moov у MP4) - reader.tail_start — абсолютное смещение tail[0] в файле (для EOF-хвоста file_size - len(tail)) - reader.meta — ReaderMeta (content_type, file_name, file_size) - reader.pending_needs — dict с запросами парсера ("_need_head", "_need_tail", "_need_slice")

Общий бюджет. Все reader-ы считают бюджет по одному правилу (RangeReader._budget()): file_size неизвестен → max_total; иначе min(max_total, round(file_size · max_ratio)), но не меньше INITIAL_HEAD. Head, tail и слайс суммарно не превышают бюджет, каждая порция ограничена остатком бюджета.

Начальный head:

  • RangeDownloader._initial_head_size() — минимум: INITIAL_HEAD (4 КБ) или размер файла, если он меньше (в пределах бюджета). head заранее почти не скачивается: парсер после первой итерации сам решает, что нужно, и reader качает ровно запрошенное.
  • RangeReader._initial_head_size() (файлы/bytes) — минимум INITIAL_HEAD или размер файла, если он меньше.

tail = RangeReader._tail_limit() — верхняя граница объёма tail (не «заказ» на всю порцию сразу: парсеры сообщают need_tail, а reader растёт от уже скачанного хвоста до запрошенного):

min(MAX_TAIL, _budget())

Для HTTP-дефолта (256 КБ, 0.1) бюджет для больших файлов — round(256 000 · 0.1) ≈ 25.6 КБ < MAX_TAIL, поэтому tail по умолчанию ~25.6 КБ. Для локального чтения (max_ratio=1.0, max_total=20 МБ) бюджет равен размеру файла, и tail упирается в лимит MAX_TAIL = 64 КБ.

Резервирования места под tail в бюджете нет: head минимален, поэтому бюджет тратится только на то, что реально запросил парсер (head/tail/слайс).

Если размер файла неизвестен (нет Content-Length) — бюджет равен max_total.

max_ratio=0 или max_total=0 (HTTP) → режим «только заголовки»: тело не скачивается, MediaProbe.from_url() строит MediaInfo(status="ok") по reader.meta.

Цикл итерации: 1. Reader читает начальный блок по _initial_head_size() 2. Парсер анализирует, возвращает ParseResult 3. Если status == "partial", reader выполняет запросы из pending_needs: _need_head (докачка head), _need_tail (рост tail назад от EOF), _need_slice (слайс от известной позиции вперёд) 4. Повторяет до status == "ok" или исчерпания лимитов (MAX_HEAD для head, _tail_limit() для tail, бюджет — общий лимит скачивания)

Парсеры: общая архитектура

Точка входа

parsers/__init__.py → parse_media_dimensions(head, tail, file_size, tail_start=0):

ImageParser.parse() → None? → VideoParser.parse() → None? → AudioParser.parse() → None?

Порядок важен: изображения проверяются первыми (самые быстрые), затем видео, затем аудио.

ImageParser (parsers/image.py)

Статический парсер — анализирует только head (не требует tail).

Методы-детекторы сигнатур → методы-парсеры:

Сигнатура Метод Что извлекает
\x89PNG + IHDR _png_parse width, height
RIFF + WEBP _webp_parse width, height, duration, fps (для анимации)
\xff\xd8 (JPEG) _jpeg_parse width, height (по маркеру SOF)
GIF87a/GIF89a _gif_parse_info width, height, duration, fps (по GCE)

Анимированные форматы (GIF, WebP) могут запросить need_head для докачки, если текущего head недостаточно для определения длительности.

VideoParser (parsers/video.py)

Крупнейший парсер (~1226 строк). Обрабатывает контейнерные форматы.

Формат Метод Ключевые техники
MP4/M4A _mp4_m4a_parse_info Парсинг атомов: ftyp → moov → mvhd (duration) → trak → tkhd (dims) + mdia → stbl → stsd (codec)
AVI _avi_parse_info RIFF: hdrl → avih (duration) + strl → strh (stream type, fps, sample_rate) + strf (codec, dims)
MKV/WebM _webm_mkv_parse_info EBML: Segment → Info (duration) + Tracks (codec, dims, sample_rate)
OGV/OGG _ogg_parse_info OggS pages → Vorbis/Theora/Opus headers → granule positions для duration
HLS _hls_parse_info Парсит m3u8 плейлист, определяет duration из EXT-X-PLAYLIST-TYPE и EXT-X-ENDLIST
M3U _m3u_parse_info Парсит простые m3u плейлисты (без HLS), определяет duration по сумме песен

Детектирование формата в parse(): проверяет сигнатуры в порядке: 1. MP4 (ftyp или moov/mdat) 2. RIFF (AVI или WAV) 3. OggS 4. EBML (MKV/WebM) 5. #EXTM3U (HLS/M3U)

Внутренние хелперы MP4: - _mp4_parse_atoms — рекурсивный парсер атомов (box format) - _mp4_moov_parse — обходит moov → trak → mdia, собирает dims dict - _mp4_parse_tkhd — извлекает width/height из Track Header Box - _mp4_parse_stsd_codec — извлекает codec из Sample Description Box (avc1→h264, hev1→h265, mp4a→aac, opus→opus) - _mp4_parse_mvhd_duration — из Media Header Box - _mp4_parse_tkhd_duration — из Track Header Box v1

Внутренние хелперы AVI: - _parse_avih — main AVI header (duration, total_frames, width, height) - _parse_strl — обходит stream list (vids/auds), вызывает _parse_strh + _parse_strf - _parse_strh — stream header (fccHandler → video codec, sample_rate) - _parse_strf — stream format (wFormatTag → audio codec, dims)

Кодек-мапы: - _MP4_CODEC_MAP — "avc1" → "h264", "mp4a" → "aac", "opus" → "opus" и т.д. - _AVI_VIDEO_CODEC_MAP — "mp4v" → "mpeg4", "cvid" → "cinepak", "av01" → "av1" и т.д. - _AVI_AUDIO_FMT_MAP — 1 → "pcm_s16le", 0x55 → "mp3", 0x0001 → "pcm" и т.д.

AudioParser (parsers/audio.py)

Формат Метод Ключевые техники
MP3 _mp3_parse_info Поиск frame sync (0xFFE0+), Xing/VBRI заголовки для duration, ID3 теги
WAV _wav_parse_info RIFF: fmt chunk (sample_rate, byte_rate) + data chunk (audio_bytes → duration)
FLAC _flac_parse_info fLaC: STREAMINFO block (sample_rate, total_samples → duration)
AAC _aac_parse_info ADTS frame sync (0xFFF0+), подсчёт кадров, экстраполяция duration
WMA _wma_parse_info ASF: FilePropertiesObject (duration) + StreamPropertiesObject (sample_rate, bitrate)

Делегирование: _ogg_ogv_check → VideoParser._ogg_parse_info, _m4a_check/_mp4_check → VideoParser._mp4_m4a_parse_info.

Цикл инспекции: детали

MediaProbe._probe(reader) в media_probe.py:

for chunk in reader:                    # reader отдаёт head/tail порциями
    if not reader.head → error
    if looks_like_html(head) → error

    raw = parse_media_dimensions(head, tail, file_size, tail_start)   # ParseResult | None
    if raw is None → partial (неизвестный формат)

    dims, needs, status = _split_parse_result(raw)        # разделяет data + service fields

    if status == "ok" → _build_media_info(dims) → return MediaInfo

    reader.pending_needs = needs       # передаёт запросы reader'у для докачки

# После цикла: если есть need_url → докачка по другому URL (HLS master playlist)
# Финальный fallback → _build_media_info(dims, status) → MediaInfo

_split_parse_result(raw): преобразует ParseResult в tuple (dims_dict, needs_dict, status): - dims_dict — только пользовательские поля (format, width, height, duration, fps, sample_rate, bitrate_nominal, video_codec, audio_codec, parse_note) - needs_dict — {"_need_head": N} и/или {"_need_tail": N} и/или {"_need_slice": (offset, size)} - Обратная совместимость: если raw — dict (старый формат), парсит ключи _need_head, _need_tail, _need_slice, _status

_build_media_info(dims, ...): создаёт MediaInfo из dims dict: - bitrate_avg вычисляется как file_size / duration * 8 / 1024 (если не задан) - duration округляется: до 1 decimal если < 10 сек, иначе до целого - MIME-тип определяется из _FORMAT_TO_MIME (формат → mime)

Как добавить новый формат

Изображение

  1. Добавить метод-детектор _xxx_check(data: bytes) -> bool в ImageParser
  2. Добавить метод-парсер _xxx_parse(data: bytes) -> ParseResult | None
  3. Вызвать детектор и парсер в ImageParser.parse() (порядок важен!)
  4. Добавить fixture в tests/fixtures/fixtures.json
  5. Добавить тест в tests/test_parsers.py

Видео/аудио (контейнер)

  1. Добавить сигнатуру в VideoParser.parse() (метод-детектор + порядок)
  2. Добавить метод _xxx_parse_info(head, tail, file_size) -> ParseResult | None
  3. Реализовать парсинг заголовков контейнера
  4. Заполнить ParseResult: format, width/height, duration, fps, sample_rate, bitrate_nominal, video_codec, audio_codec
  5. Добавить формат в _FORMAT_TO_MIME (media_probe.py) и Literal-тип format в MediaInfo
  6. Добавить fixture + тесты

Аудио-кодек (внутри контейнера)

  1. Добавить кодек-мап (аналог _MP4_CODEC_MAP или _AVI_AUDIO_FMT_MAP)
  2. Парсить кодек изSample Description Box / Stream Format chunk
  3. Заполнять audio_codec (краткий) и audio_codec_long (полное название)

Константы

url_media_probe/parsers/_constants.py:

Константа Значение Описание
INITIAL_HEAD 4096 Минимальный head (и значение при неизвестном размере)
MAX_HEAD 256 000 Максимальный объём head (лимит докачки)
MAX_TAIL 65 536 Верхний предел tail (лимит _tail_limit())
MOOV_TAIL_START 4096 Первый чанк хвоста потокового MP4 (moov в конце)
OGG_TAIL_START 4096 Первый чанк хвоста OGG (хватит на последнюю страницу)
MP3_TAIL_FRAME 8192 Докачка MP3, когда в head не нашёлся фрейм
MID_REUSE_MIN 65 536 Мин. объём tail-слайса для переиспользования при смене якоря
EXPAND_CHUNK 4096 Размер шага докачки head
NEED_HEAD "_need_head" Ключ в pending_needs (для reader)
NEED_TAIL "_need_tail" Ключ в pending_needs (для reader)
NEED_SLICE "_need_slice" Ключ для запроса слайса от известной позиции
NEED_URL "_need_url" Ключ для запроса докачки по URL
PARSE_STATUS "_status" Ключ статуса (legacy, используется в _split_parse_result)

Дополнительно лимиты max_total и max_ratio задаются в RangeReader.__init__ (см. выше). Бюджет — min(max_total, file_size · max_ratio) (не меньше INITIAL_HEAD): после его исчерпания reader не читает больше данных. Резервирования нет: head минимален, и бюджет тратится только на запрошенное парсером (каждая порция ограничена остатком budget − downloaded).

Фикстуры и тесты

Структура

tests/
├── fixtures/
│   ├── fixtures.json          # Все фикстуры: base64 head/tail + expected поля
│   └── prepare_fixtures.py    # Скрипт для скачивания и создания фикстур
├── conftest.py                # Pytest-фикстуры: all_fixtures, fixture_bytes_io
├── helpers.py                 # make_mock_session, make_fixture_named_bytes_io и т.д.
├── test_parsers.py            # Unit-тесты парсеров (прямой вызов методов)
├── test_media_probe.py        # Интеграционные тесты MediaProbe
├── test_media_info.py         # Тесты MediaInfo
└── test_range_readers.py      # Тесты reader'ов

Фикстуры

Физических бинарных файлов нет. Все данные хранятся в fixtures.json как base64:

{
  "mp4_h264": {
    "head_b64": "AAAAG3ZpZ...",      // начало файла (base64)
    "tail_b64": "AAAADHZpZ...",      // конец файла (base64, опционально)
    "url": "https://example.com/v.mp4",
    "mime_type": "video/mp4",
    "file_size": 1048576,
    "format": "MP4",
    "width": 1920,
    "height": 1080,
    "duration": 15.0,
    "video_codec": "h264",
    "audio_codec": "aac"
  }
}

Конвенция: только video_codec и audio_codec (короткие имена), без _long вариантов.

Как добавить новую фикстуру

Используется tests/fixtures/prepare_fixtures.py.

Автоматический режим (рекомендуется):

  1. Добавить запись в SAMPLE_URLS в prepare_fixtures.py:
SAMPLE_URLS = {
    ...
    "my_new_format": {
        "url": "https://example.com/sample.webm",
    },
}
  1. Запустить:
cd tests/fixtures
python prepare_fixtures.py my_new_format

Скрипт скачает файл, запустит MediaProbe, получит все expected-поля автоматически.

  1. Проверить полученные значения в fixtures.json вручную перед коммитом.

Ручной режим (USE_AUTOMATIC_EXPECTED = False):

Указать expected-поля явно:

"my_new_format": {
    "url": "https://example.com/sample.webm",
    "expected": {"format": "WEBM", "width": 1280, "height": 720},
    "head_size": 8192,
}

Тесты

Unit-тесты парсеров (test_parsers.py): - Прямой вызов методов парсеров (ImageParser._jpeg_parse, VideoParser._mp4_m4a_parse_info и т.д.) - Тесты edge cases и покрытие веток - Используют hand-crafted байты или данные из fixtures.json

Интеграционные тесты MediaProbe (test_media_probe.py):

Три подхода к тестированию:

  1. from_bytes_range — самый простой для fixture-тестов:
@pytest.mark.parametrize("name", FIXTURES_ID)
async def test_format_detected(self, name, all_fixtures):
    head, tail, exp = all_fixtures[name]
    probe = MediaProbe()
    info = await probe.from_bytes_range(head, tail, exp["file_size"])
    assert info.format == exp["format"]

Не требует mock session, тестирует полный pipeline: reader → _probe → parsers → MediaInfo.

  1. from_url + mock session — для тестирования HTTP-специфичного поведения:
session = make_mock_session(head, tail, mime, file_size)
info = await probe.from_url("https://example.com/test", session=session)

Используется для: retry, ошибки сети, HTML-страницы, Range-запросы.

  1. from_bytes / from_file — для тестирования других reader'ов.

Тесты reader'ов (test_range_readers.py): - Тесты RangeDownloader, RangeFileReader, RangeBytesReader - Моки HTTP-ответов, edge cases чтения

Соглашения по коду

  • Все парсеры возвращают ParseResult
  • bitrate_nominal — номинальный битрейт из метаданных (не вычисляется)
  • bitrate_avg — вычисляется в _build_media_info как file_size / duration
  • Все docstrings и сообщения на русском языке
  • format в ParseResult — строковый идентификатор ("PNG", "MP4"), не MIME
  • status: "ok" — данные полные, "partial" — нужны дополнительные данные