Примеры

Здесь собраны практические примеры использования url-media-probe.

Определение видео по URL

import asyncio
from url_media_probe import MediaProbe

async def main():
    probe = MediaProbe()
    info = await probe.from_url("https://example.com/video.mp4")

    print(info.format)    # "MP4"
    print(info.width)     # 1920
    print(info.height)    # 1080
    print(info.duration)  # 15.0
    print(info.fps)       # 25.0
    print(info.status)    # "ok"

    # Сколько данных было скачано
    print(f"Скачано: {probe.downloaded_human}")

    # Если нужно скачать файл целиком — используем уже скачанные данные
    if info.status == "ok":
        path = await probe.download_file("./downloads")
        print(f"Сохранён: {path}")

asyncio.run(main())

Определение локального файла

Локальные файлы по умолчанию читаются целиком (до 20 МБ) — для анимированных GIF/WebP длительность определяется точно, без экстраполяции.

import asyncio
from url_media_probe import MediaProbe

async def main():
    probe = MediaProbe()
    info = await probe.from_file("/path/to/photo.jpg")

    print(info.format)    # "JPEG"
    print(info.width)     # 4032
    print(info.height)    # 3024
    print(info.is_image)  # True

asyncio.run(main())

Определение из байтов

import asyncio
from url_media_probe import MediaProbe

async def main():
    probe = MediaProbe()

    # Из bytes
    data = open("audio.mp3", "rb").read()
    info = await probe.from_bytes(data)
    print(info.format)  # "MP3"

    # Из BytesIO
    from io import BytesIO
    buf = BytesIO(data)
    buf.name = "audio.mp3"
    info = await probe.from_bytes(buf)
    print(info.format)  # "MP3"

asyncio.run(main())

Парсинг по готовым head/tail

Если у вас уже скачаны начало и конец файла:

import asyncio
from url_media_probe import MediaProbe

async def main():
    probe = MediaProbe()

    head = b"..."  # первые байты файла
    tail = b"..."  # последние байты файла
    file_size = 1_048_576  # 1 MB

    info = await probe.from_bytes_range(head, tail, file_size, file_name="video.mp4")
    print(info.format)  # "MP4"

asyncio.run(main())

Кастомные HTTP-заголовки

import asyncio
import aiohttp
from url_media_probe import MediaProbe

async def main():
    probe = MediaProbe()

    # Создаём сессию с кастомными заголовками
    headers = {"Authorization": "Bearer token123"}
    async with aiohttp.ClientSession(headers=headers) as session:
        info = await probe.from_url(
            "https://example.com/protected/video.mp4",
            session=session,
            timeout=15,
            max_total=512_000,  # максимум 512 KB для определения параметров
            max_ratio=0.1,      # относительный лимит: не больше 10% размера файла (итог — минимум с max_total); head всегда 4 KB
        )
        print(info.format)

asyncio.run(main())

Режим «только заголовки» (без скачивания тела)

Если нужны только Content-Type, размер и имя файла — без скачивания тела:

import asyncio
from url_media_probe import MediaProbe

async def main():
    probe = MediaProbe()

    info = await probe.from_url(
        "https://example.com/video.mp4",
        max_ratio=0,  # или max_total=0
    )
    print(info.mime_type)  # "video/mp4"
    print(info.file_size)  # 1048576
    print(info.status)     # "ok"

asyncio.run(main())

Проверка типа файла

info = await probe.from_url("https://example.com/file.bin")

if info.is_image:
    print(f"Изображение: {info.width}x{info.height}")
elif info.is_video:
    print(f"Видео: {info.duration} сек, {info.fps} FPS")
elif info.is_audio:
    print(f"Аудио: {info.duration} сек, {info.sample_rate} Гц")
else:
    print(f"Неизвестный тип: {info.mime_type}")

Использование с aiohttp session

import asyncio
import aiohttp
from url_media_probe import MediaProbe

async def main():
    probe = MediaProbe()

    async with aiohttp.ClientSession() as session:
        info = await probe.from_url(
            "https://example.com/video.mp4",
            session=session,
        )
        print(info.format)

        # Сессия остаётся доступной — можно использовать повторно
        info2 = await probe.from_url(
            "https://example.com/audio.mp3",
            session=session,
        )
        print(info2.format)

asyncio.run(main())

Обработка ошибок

Библиотека никогда не бросает исключения — все ошибки попадают в MediaInfo.status / parse_note.

Проверка статуса

info = await probe.from_url("https://example.com/video.mp4")

if info.status == "ok":
    print(info)
elif info.status == "partial":
    print(f"Частичные данные: {info.parse_note}")
elif info.status == "error":
    print(f"Ошибка: {info.parse_note}")

Типичные ошибки и их parse_note

info = await probe.from_url("https://example.com/file.bin")

match info.parse_note:
    case note if "Сетевая ошибка" in note:
        print("Нет доступа к серверу или таймаут")
    case note if "не является медиа" in note:
        print("По ссылке оказалась HTML-страница, а не медиафайл")
    case note if "не поддерживается" in note:
        print("Формат файла не распознан парсером")
    case note if "Недостаточно данных" in note:
        print("Файл слишком мал, данные не извлечены")
    case note if "не найден" in note:
        print("Локальный файл не существует")
    case _:
        print(f"Другая ошибка: {info.parse_note}")

Проброс сетевых ошибок (raise_on_network)

Если удобнее работать с исключениями — включите raise_on_network:

import aiohttp
import url_media_probe.media_probe as cfg
from url_media_probe import MediaProbe

# Глобально — ко всем экземплярам по умолчанию
cfg.raise_on_network = True

probe = MediaProbe()
try:
    info = await probe.from_url("https://example.com/video.mp4")
except aiohttp.ClientError as e:
    print(f"Сетевая ошибка: {e}")
    # при необходимости — отключить для следующего вызова
    probe_fallback = MediaProbe(raise_on_network=False)
    info = await probe_fallback.from_url("https://backup.example.com/video.mp4")

Пакетная проверка URLs

import asyncio
from url_media_probe import MediaProbe

async def check_url(probe, url):
    info = await probe.from_url(url)
    return url, info.status, info.format

async def main():
    probe = MediaProbe()
    urls = [
        "https://example.com/video.mp4",
        "https://example.com/audio.mp3",
        "https://example.com/missing.jpg",
    ]

    results = await asyncio.gather(
        *(check_url(probe, url) for url in urls),
        return_exceptions=True,
    )

    for result in results:
        if isinstance(result, Exception):
            print(f"Исключение: {result}")
        else:
            url, status, fmt = result
            print(f"{url}: {status} ({fmt})")

asyncio.run(main())