ytdlp/app/inspection.py

import ipaddress
import socket
from collections.abc import Mapping
from typing import Any
from urllib.parse import urlsplit

import yt_dlp
from yt_dlp.utils import DownloadError


class UnsafeUrlError(ValueError):
    """Raised when a submitted URL is outside the permitted network boundary."""


class InspectionError(RuntimeError):
    """Raised when yt-dlp cannot inspect a permitted URL."""


def validate_public_url(url: str) -> str:
    """Validate the initial URL and require every resolved address to be public."""
    try:
        parsed = urlsplit(url)
        port = parsed.port
    except ValueError as error:
        raise UnsafeUrlError("The URL is malformed.") from error

    if parsed.scheme not in {"http", "https"}:
        raise UnsafeUrlError("Only HTTP and HTTPS URLs are allowed.")
    if not parsed.hostname:
        raise UnsafeUrlError("The URL must include a hostname.")
    if parsed.username is not None or parsed.password is not None:
        raise UnsafeUrlError("Credentials are not allowed in source URLs.")
    if port is not None and port not in {80, 443}:
        raise UnsafeUrlError("Only the standard HTTP and HTTPS ports are allowed.")

    try:
        addresses = {
            result[4][0]
            for result in socket.getaddrinfo(
                parsed.hostname,
                port or (443 if parsed.scheme == "https" else 80),
                type=socket.SOCK_STREAM,
            )
        }
    except socket.gaierror as error:
        raise UnsafeUrlError("The source hostname could not be resolved.") from error

    if not addresses:
        raise UnsafeUrlError("The source hostname did not resolve to an address.")

    for address in addresses:
        try:
            ip = ipaddress.ip_address(address)
        except ValueError as error:
            raise UnsafeUrlError("The source hostname returned an invalid address.") from error
        if not ip.is_global:
            raise UnsafeUrlError("The source hostname resolves to a non-public address.")

    return url


def _format_summary(media_format: Mapping[str, Any]) -> dict[str, Any]:
    filesize = media_format.get("filesize") or media_format.get("filesize_approx")

    return {
        "format_id": str(media_format.get("format_id") or "unknown"),
        "extension": media_format.get("ext"),
        "width": media_format.get("width"),
        "height": media_format.get("height"),
        "fps": media_format.get("fps"),
        "video_codec": media_format.get("vcodec"),
        "audio_codec": media_format.get("acodec"),
        "filesize": filesize,
    }


def _normalized_formats(raw_formats: object) -> list[dict[str, Any]]:
    formats: list[dict[str, Any]] = []
    seen: set[tuple[object, ...]] = set()

    if not isinstance(raw_formats, list):
        return formats

    for media_format in raw_formats:
        if not isinstance(media_format, dict):
            continue
        if (
            media_format.get("vcodec") in {None, "none"}
            and media_format.get("acodec") in {None, "none"}
        ):
            continue

        summary = _format_summary(media_format)
        identity = (
            summary["extension"],
            summary["width"],
            summary["height"],
            summary["fps"],
            summary["video_codec"],
            summary["audio_codec"],
        )
        if identity in seen:
            continue
        seen.add(identity)
        formats.append(summary)

    return formats


def _select_single_video(info: dict[str, Any]) -> dict[str, Any]:
    entries = info.get("entries")
    if entries is None and info.get("_type") not in {"playlist", "multi_video"}:
        return info

    videos = [entry for entry in entries or [] if isinstance(entry, dict)]
    if len(videos) != 1:
        raise InspectionError("Playlists and multi-video sources are not supported.")

    video = dict(videos[0])
    for field in ("title", "uploader", "channel", "thumbnail", "webpage_url"):
        if not video.get(field) and info.get(field):
            video[field] = info[field]
    return video


def inspect_url(url: str) -> dict[str, Any]:
    """Inspect one public video URL without downloading its media."""
    safe_url = validate_public_url(url)
    options: dict[str, Any] = {
        "extract_flat": False,
        "extractor_retries": 1,
        "ignoreconfig": True,
        "noplaylist": True,
        "quiet": True,
        "retries": 1,
        "skip_download": True,
        "socket_timeout": 15,
        "no_warnings": True,
    }

    try:
        with yt_dlp.YoutubeDL(options) as downloader:
            raw_info = downloader.extract_info(safe_url, download=False)
            info = downloader.sanitize_info(raw_info)
    except DownloadError as error:
        raise InspectionError("yt-dlp could not inspect this URL.") from error
    except Exception as error:
        raise InspectionError("The source inspection failed unexpectedly.") from error

    if not isinstance(info, dict):
        raise InspectionError("The source returned an unsupported response.")

    info = _select_single_video(info)

    formats = _normalized_formats(info.get("formats"))

    return {
        "id": str(info.get("id") or "unknown"),
        "title": info.get("title") or "Untitled",
        "extractor": info.get("extractor_key") or info.get("extractor"),
        "uploader": info.get("uploader") or info.get("channel"),
        "duration": info.get("duration"),
        "upload_date": info.get("upload_date"),
        "source_url": safe_url,
        "webpage_url": info.get("webpage_url") or safe_url,
        "thumbnail": info.get("thumbnail"),
        "is_live": bool(info.get("is_live")),
        "formats": formats,
    }