feat: inspect public video sources with yt-dlp

app/inspection.py

@@ -0,0 +1,169 @@
+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,
+    }