Upload folder using huggingface_hub

.gitattributes

@@ -1,35 +1,35 @@
-*.7z filter=lfs diff=lfs merge=lfs -text
-*.arrow filter=lfs diff=lfs merge=lfs -text
-*.bin filter=lfs diff=lfs merge=lfs -text
-*.bz2 filter=lfs diff=lfs merge=lfs -text
-*.ckpt filter=lfs diff=lfs merge=lfs -text
-*.ftz filter=lfs diff=lfs merge=lfs -text
-*.gz filter=lfs diff=lfs merge=lfs -text
-*.h5 filter=lfs diff=lfs merge=lfs -text
-*.joblib filter=lfs diff=lfs merge=lfs -text
-*.lfs.* filter=lfs diff=lfs merge=lfs -text
-*.mlmodel filter=lfs diff=lfs merge=lfs -text
-*.model filter=lfs diff=lfs merge=lfs -text
-*.msgpack filter=lfs diff=lfs merge=lfs -text
-*.npy filter=lfs diff=lfs merge=lfs -text
-*.npz filter=lfs diff=lfs merge=lfs -text
-*.onnx filter=lfs diff=lfs merge=lfs -text
-*.ot filter=lfs diff=lfs merge=lfs -text
-*.parquet filter=lfs diff=lfs merge=lfs -text
-*.pb filter=lfs diff=lfs merge=lfs -text
-*.pickle filter=lfs diff=lfs merge=lfs -text
-*.pkl filter=lfs diff=lfs merge=lfs -text
-*.pt filter=lfs diff=lfs merge=lfs -text
-*.pth filter=lfs diff=lfs merge=lfs -text
-*.rar filter=lfs diff=lfs merge=lfs -text
-*.safetensors filter=lfs diff=lfs merge=lfs -text
-saved_model/**/* filter=lfs diff=lfs merge=lfs -text
-*.tar.* filter=lfs diff=lfs merge=lfs -text
-*.tar filter=lfs diff=lfs merge=lfs -text
-*.tflite filter=lfs diff=lfs merge=lfs -text
-*.tgz filter=lfs diff=lfs merge=lfs -text
-*.wasm filter=lfs diff=lfs merge=lfs -text
-*.xz filter=lfs diff=lfs merge=lfs -text
-*.zip filter=lfs diff=lfs merge=lfs -text
-*.zst filter=lfs diff=lfs merge=lfs -text
-*tfevents* filter=lfs diff=lfs merge=lfs -text
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ckpt filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.mlmodel filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.npy filter=lfs diff=lfs merge=lfs -text
+*.npz filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pickle filter=lfs diff=lfs merge=lfs -text
+*.pkl filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+*.safetensors filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tar filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.wasm filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zst filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text

Dockerfile

@@ -0,0 +1,16 @@
+# Read the doc: https://huggingface.co/docs/hub/spaces-sdks-docker
+# you will also find guides on how best to write your Dockerfile
+
+FROM python:3.13
+
+RUN useradd -m -u 1000 user
+USER user
+ENV PATH="/home/user/.local/bin:$PATH"
+
+WORKDIR /app
+
+COPY --chown=user ./requirements.txt requirements.txt
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+
+COPY --chown=user . /app
+CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -0,0 +1,11 @@
+---
+title: VaslNistim
+emoji: 🦀
+colorFrom: blue
+colorTo: red
+sdk: docker
+pinned: false
+license: mit
+---
+
+Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference

app.py

@@ -0,0 +1,100 @@
+from threading import Thread
+import gradio as gr
+import speedtest
+import platform
+import psutil
+import os
+
+
+def starter():
+    os.system("python3 Kmain.py")
+
+
+thread = Thread(target=starter)
+thread.start()
+
+
+def get_system_usage():
+    # Get CPU usage as a percentage
+    cpu_usage = psutil.cpu_percent(interval=1)
+
+    # Get memory usage
+    memory = psutil.virtual_memory()
+    total_memory = memory.total
+    available_memory = memory.available
+    memory_usage = total_memory - available_memory
+
+    # Get disk usage
+    disk = psutil.disk_usage('/')
+    total_disk = disk.total
+    used_disk = disk.used
+    disk_usage = used_disk
+
+    # Prepare the system usage information as a string
+    system_usage = f"CPU Usage: {cpu_usage}%\n"
+    system_usage += f"Memory Usage: {memory_usage} bytes\n"
+    system_usage += f"Disk Usage: {disk_usage} bytes"
+
+    return system_usage
+
+
+def get_internet_speed():
+    st = speedtest.Speedtest()
+    st.get_best_server()
+    download_speed = st.download() / 1_000_000 / 8 # Convert to MBps
+    upload_speed = st.upload() / 1_000_000 / 8 # Convert to MBps
+    return download_speed, upload_speed
+    
+
+def get_system_information():
+    system = platform.system()
+    release = platform.release()
+    version = platform.version()
+    machine = platform.machine()
+    processor = platform.processor()
+
+    memory = psutil.virtual_memory()
+    memory_total = memory.total
+    memory_used = memory.used
+    memory_percent = memory.percent
+
+    disk = psutil.disk_usage('/')
+    disk_total = disk.total
+    disk_used = disk.used
+    disk_percent = disk.percent
+
+    cpu_percent = psutil.cpu_percent()
+
+    download_speed, upload_speed = get_internet_speed()
+
+    information = (f"System Information:\n"
+                   f"- System: {system}\n"
+                   f"- Release: {release}\n"
+                   f"- Version: {version}\n"
+                   f"- Machine: {machine}\n"
+                   f"- Processor: {processor}\n"
+                   f"\n"
+                   f"Memory Usage:\n"
+                   f"- Memory Total: {memory_total} bytes\n"
+                   f"- Memory Used: {memory_used} bytes\n"
+                   f"- Memory Percent: {memory_percent}%\n"
+                   f"\n"
+                   f"Disk Usage:\n"
+                   f"- Disk Total: {disk_total} bytes\n"
+                   f"- Disk Used: {disk_used} bytes\n"
+                   f"- Disk Percent: {disk_percent}%\n"
+                   f"\n"
+                   f"CPU Usage:\n"
+                   f"- CPU Percent: {cpu_percent}%\n"
+                   f"\n"
+                   f"- Internet Usage:\n"
+                   f"Download Speed: {download_speed:.2f} MBps\n"
+                   f"Upload Speed: {upload_speed:.2f} MBps"
+                  )
+
+    return information
+
+
+
+iface = gr.Interface(fn=get_system_information, inputs=None, outputs="text")
+iface.launch(server_name="0.0.0.0", server_port=7860)

main.py

@@ -0,0 +1,37 @@
+import os
+from video_dl import get_video_formats, download_video
+from balethon import Client, conditions
+from balethon.objects.message import Message
+import json
+import re
+
+
+url = "https://www.youtube.com/watch?v=GVPhBSmMuq8&pp=ugUEEgJlbg%3D%3D"
+data = get_video_formats(url=url)
+# print(json.dumps(data, indent=2))
+# download_video(url=url, format_id="278", output="video.mp4", ffmpeg_location=r"C:\ffmpeg-master-latest-win64-gpl\bin\ffmpeg.exe")
+
+
+
+bot = Client(os.environ["BotToken"])
+
+def is_youtube_link(event: str) -> bool:
+    link = False
+    if event.text:
+        # Regular expression to match various YouTube URL formats
+        youtube_regex = r'^(https?://)?(www\.|m\.)?(youtube\.com|youtu\.be)/'
+        if re.match(youtube_regex, event.text):
+            link = True
+    return link
+
+
+
+@bot.on_message(condition=conditions.Condition(is_youtube_link))
+async def greet(message: Message):
+    user_message = message.text
+    formats = get_video_formats(url=user_message)
+    output_path = download_video(url=user_message, format_id=formats[-1]["format_id"])
+    # await message.reply("Hello")
+    await message.reply_video(output_path)
+
+bot.run()

requirements.txt

@@ -0,0 +1,7 @@
+h2
+gradio-client
+gradio
+speedtest
+cryptography
+psutil
+yt-dlp

video_dl.py

@@ -0,0 +1,525 @@
+"""
+Video format extraction and download utilities using yt-dlp.
+
+Provides two core functions:
+- get_video_formats: Extract available video format metadata, deduplicated
+  by resolution+FPS with largest file size per combination
+- download_video: Download a specific video format by ID with automatic
+  best audio merging for video-only formats.
+
+Both functions support custom ffmpeg location for format merging.
+
+Requires yt-dlp to be installed: pip install yt-dlp
+"""
+
+from __future__ import annotations
+
+import logging
+from pathlib import Path
+from typing import Any, TypedDict
+
+import yt_dlp
+from yt_dlp.utils import DownloadError, ExtractorError
+
+logger = logging.getLogger(__name__)
+
+
+class VideoFormatInfo(TypedDict):
+    """Structure for video format metadata returned by get_video_formats."""
+    format_id: str
+    width: int | None
+    height: int | None
+    size: int | None
+    fps: float | None
+    duration: float | None
+
+
+class VideoDownloadError(Exception):
+    """Base exception for video download operations."""
+    pass
+
+
+class FormatExtractionError(VideoDownloadError):
+    """Raised when format extraction fails."""
+    pass
+
+
+class DownloadFailedError(VideoDownloadError):
+    """Raised when the actual download fails."""
+    pass
+
+
+class InvalidURLError(VideoDownloadError):
+    """Raised when the provided URL is invalid or unsupported."""
+    pass
+
+
+class FormatNotFoundError(VideoDownloadError):
+    """Raised when the requested format ID does not exist."""
+    pass
+
+
+class FFmpegNotFoundError(VideoDownloadError):
+    """Raised when the specified ffmpeg binary does not exist."""
+    pass
+
+
+def _validate_url(url: str) -> None:
+    """Validate that the URL has an acceptable scheme.
+
+    Args:
+        url: The URL string to validate.
+
+    Raises:
+        InvalidURLError: If the URL scheme is not http or https.
+    """
+    from urllib.parse import urlparse
+
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        raise InvalidURLError(
+            f"Unsupported URL scheme '{parsed.scheme}'. "
+            f"Only http and https are allowed. Got: {url}"
+        )
+    if not parsed.netloc:
+        raise InvalidURLError(f"URL has no hostname. Got: {url}")
+
+
+def _sanitize_output_path(output: str | Path | None) -> Path | None:
+    """Sanitize and resolve the output path.
+
+    Args:
+        output: User-provided output path string or Path object, or None.
+
+    Returns:
+        Resolved absolute Path object, or None if output is None.
+
+    Raises:
+        ValueError: If the path contains directory traversal attempts
+                    or is otherwise invalid.
+    """
+    if output is None:
+        return None
+
+    path = Path(output).expanduser().resolve()
+
+    # Basic path traversal protection — resolve should handle .. but
+    # ensure the resolved path is still within reasonable bounds
+    try:
+        path.relative_to(path.anchor or "/")
+    except ValueError:
+        raise ValueError(f"Invalid output path: {output}")
+
+    # Create parent directories if they don't exist
+    path.parent.mkdir(parents=True, exist_ok=True)
+
+    return path
+
+
+def _validate_ffmpeg_path(ffmpeg_location: str | Path | None) -> str | None:
+    """Validate and resolve the ffmpeg binary path.
+
+    Args:
+        ffmpeg_location: Path to ffmpeg executable, or None to use
+                         system default.
+
+    Returns:
+        Resolved absolute path as string, or None for system default.
+
+    Raises:
+        FFmpegNotFoundError: If the specified path does not exist or
+                             is not a file.
+    """
+    if ffmpeg_location is None:
+        return None
+
+    ffmpeg_path = Path(ffmpeg_location).expanduser().resolve()
+
+    if not ffmpeg_path.exists():
+        raise FFmpegNotFoundError(
+            f"ffmpeg binary not found at: {ffmpeg_path}"
+        )
+    if not ffmpeg_path.is_file():
+        raise FFmpegNotFoundError(
+            f"ffmpeg path exists but is not a file: {ffmpeg_path}"
+        )
+
+    logger.debug("Using custom ffmpeg at: %s", ffmpeg_path)
+    return str(ffmpeg_path)
+
+
+def _build_base_ydl_opts(
+    ffmpeg_location: str | None = None,
+) -> dict[str, Any]:
+    """Build the base yt-dlp options dictionary.
+
+    Args:
+        ffmpeg_location: Optional path to ffmpeg binary. If None, yt-dlp
+                         will search the system PATH.
+
+    Returns:
+        Base options dictionary for yt-dlp configuration.
+    """
+    opts: dict[str, Any] = {
+        "quiet": True,
+        "no_warnings": True,
+        "noplaylist": True,
+        "compat_opts": ["no-certifi"],
+    }
+
+    if ffmpeg_location is not None:
+        opts["ffmpeg_location"] = ffmpeg_location
+
+    return opts
+
+
+def _get_deduplication_key(
+    fmt: dict[str, Any],
+) -> tuple[int | None, int | None, float | None]:
+    """Extract the deduplication key (width, height, fps) from a format dict.
+
+    Args:
+        fmt: A raw format dictionary from yt-dlp.
+
+    Returns:
+        A tuple of (width, height, fps) used for grouping similar formats.
+    """
+    return (
+        fmt.get("width"),
+        fmt.get("height"),
+        fmt.get("fps"),
+    )
+
+
+def _deduplicate_formats_by_largest_size(
+    formats: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """Deduplicate formats by resolution+FPS, keeping the largest size variant.
+
+    Audio-only formats (null width and height) are removed entirely.
+    Within each (width, height, fps) group, the format with the largest
+    non-null size is kept. If all formats in a group have null size,
+    the first one encountered is retained.
+
+    Args:
+        formats: List of raw format dictionaries from yt-dlp.
+
+    Returns:
+        Deduplicated list of format dictionaries, sorted by resolution
+        descending, then by size descending.
+    """
+    # Step 1: Remove audio-only formats and unavailable formats
+    video_formats = [
+        fmt for fmt in formats
+        if fmt.get("width") is not None
+        and fmt.get("height") is not None
+        and fmt.get("availability") != "unavailable"
+    ]
+
+    if not video_formats:
+        return []
+
+    # Step 2: Group by (width, height, fps) and keep largest size
+    grouped: dict[
+        tuple[int | None, int | None, float | None], dict[str, Any]
+    ] = {}
+
+    for fmt in video_formats:
+        key = _get_deduplication_key(fmt)
+        current_size = fmt.get("filesize") or fmt.get("filesize_approx")
+
+        if key not in grouped:
+            grouped[key] = fmt
+        else:
+            existing_size = (
+                grouped[key].get("filesize")
+                or grouped[key].get("filesize_approx")
+            )
+
+            # If current format has a known size and existing doesn't,
+            # or current size is larger — replace
+            if current_size is not None:
+                if existing_size is None or current_size > existing_size:
+                    grouped[key] = fmt
+            # If both are None, keep the first one (already in grouped)
+
+    # Step 3: Sort by resolution descending, then size descending
+    def sort_key(
+        item: tuple[
+            tuple[int | None, int | None, float | None], dict[str, Any]
+        ],
+    ) -> tuple[int, int, int]:
+        (width, height, __), fmt = item
+        # Use 0 for None values to place them at the end
+        resolved_width = width if width is not None else 0
+        resolved_height = height if height is not None else 0
+        size = fmt.get("filesize") or fmt.get("filesize_approx") or 0
+        # Sort by width descending, then height descending, then size descending
+        return (-resolved_width, -resolved_height, -size)
+
+    sorted_grouped = sorted(grouped.items(), key=sort_key)
+
+    return [fmt for _, fmt in sorted_grouped]
+
+
+def get_video_formats(
+    url: str,
+    ffmpeg_location: str | Path | None = None,
+) -> list[VideoFormatInfo]:
+    """Extract available video format metadata from a URL using yt-dlp.
+
+    Fetches format information without downloading the video. Formats are
+    deduplicated by unique (width, height, fps) combinations, keeping only
+    the variant with the largest file size for each combination. Audio-only
+    formats are excluded. Results are sorted by resolution descending.
+
+    Args:
+        url: The video URL to extract formats from. Must use http or https.
+        ffmpeg_location: Optional path to ffmpeg executable. If provided,
+                         will be used for any format merging operations.
+                         If None, yt-dlp searches the system PATH.
+                         Example: "C:\\ffmpeg-master-latest-win64-gpl\\bin\\ffmpeg.exe"
+
+    Returns:
+        A list of VideoFormatInfo dictionaries, one per unique
+        (width, height, fps) combination, sorted by resolution descending.
+        Returns an empty list if no video formats are found.
+
+    Raises:
+        InvalidURLError: If the URL scheme is not http/https.
+        FFmpegNotFoundError: If the specified ffmpeg_location does not exist.
+        FormatExtractionError: If extraction fails for any reason
+                               (network error, unsupported site, etc).
+
+    Example:
+        >>> formats = get_video_formats(
+        ...     "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
+        ...     ffmpeg_location="C:\\\\ffmpeg\\\\bin\\\\ffmpeg.exe"
+        ... )
+        >>> for fmt in formats:
+        ...     print(f"{fmt['width']}x{fmt['height']} @ {fmt['fps']}fps")
+    """
+    _validate_url(url)
+    resolved_ffmpeg = _validate_ffmpeg_path(ffmpeg_location)
+
+    ydl_opts = _build_base_ydl_opts(ffmpeg_location=resolved_ffmpeg)
+    ydl_opts["extract_flat"] = False
+    ydl_opts["skip_download"] = True
+    ydl_opts["cookiefile"] = None
+
+    try:
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(url, download=False)
+    except (DownloadError, ExtractorError) as exc:
+        raise FormatExtractionError(
+            f"Failed to extract formats from URL: {exc}"
+        ) from exc
+    except Exception as exc:
+        raise FormatExtractionError(
+            f"Unexpected error during format extraction: {exc}"
+        ) from exc
+
+    if info is None:
+        raise FormatExtractionError(
+            f"No information could be extracted from: {url}"
+        )
+
+    formats_raw: list[dict[str, Any]] = info.get("formats") or []
+
+    # Deduplicate: keep only largest size per (width, height, fps) group
+    deduplicated = _deduplicate_formats_by_largest_size(formats_raw)
+
+    result: list[VideoFormatInfo] = []
+    for fmt in deduplicated:
+        if type(fmt.get("fps")) == float:
+            continue
+        format_info: VideoFormatInfo = {
+            "format_id": str(fmt.get("format_id", "")),
+            "width": fmt.get("width"),
+            "height": fmt.get("height"),
+            "size": fmt.get("filesize") or fmt.get("filesize_approx"),
+            "fps": fmt.get("fps"),
+            "duration": info.get("duration"),
+        }
+        result.append(format_info)
+
+    return result
+
+
+def _format_has_audio(format_info: dict[str, Any]) -> bool:
+    """Determine if a format entry already contains an audio stream.
+
+    Checks audio codec presence and video codec absence to identify
+    whether the format includes audio. Formats with no video codec
+    are audio-only, and formats with an audio codec are muxed.
+
+    Args:
+        format_info: A single format dictionary from yt-dlp extraction.
+
+    Returns:
+        True if the format contains audio, False if it is video-only.
+    """
+    # Audio-only formats have no video codec
+    if format_info.get("vcodec") == "none":
+        return True
+
+    # Muxed formats have an audio codec present
+    if format_info.get("acodec") and format_info["acodec"] != "none":
+        return True
+
+    # Video-only formats have vcodec but no acodec (or acodec="none")
+    return False
+
+
+def download_video(
+    url: str,
+    format_id: str,
+    output: str | Path | None = None,
+    ffmpeg_location: str | Path | None = None,
+) -> str:
+    """Download a video in the specified format using yt-dlp.
+
+    Downloads the video with the given format ID. If the requested format
+    is video-only (no audio track), the best available audio is
+    automatically selected and merged into the final output using ffmpeg.
+
+    If an output path is provided, the file will be saved to that exact
+    location. Otherwise, yt-dlp's default naming template is used.
+
+    Args:
+        url: The video URL to download. Must use http or https.
+        format_id: The yt-dlp format ID string (e.g., "137", "22").
+                   Video-only formats will automatically have audio
+                   merged in.
+        output: Optional output file path. Directories will be created
+                if they don't exist. Should include the file extension.
+        ffmpeg_location: Optional path to ffmpeg executable. Required
+                         if system doesn't have ffmpeg in PATH. Used for
+                         merging video and audio streams.
+                         Example: "C:\\\\ffmpeg-master-latest-win64-gpl\\\\bin\\\\ffmpeg.exe"
+
+    Returns:
+        The absolute path (as a string) where the video was saved.
+
+    Raises:
+        InvalidURLError: If the URL scheme is not http/https.
+        FFmpegNotFoundError: If the specified ffmpeg_location does not exist.
+        FormatNotFoundError: If the specified format_id is not found.
+        DownloadFailedError: If the download fails for any reason.
+        ValueError: If the output path is invalid.
+
+    Example:
+        >>> path = download_video(
+        ...     "https://www.youtube.com/watch?v=dQw4w9WgXcQ",
+        ...     format_id="137",
+        ...     output="/tmp/video.mp4",
+        ...     ffmpeg_location="C:\\\\ffmpeg\\\\bin\\\\ffmpeg.exe"
+        ... )
+        >>> print(f"Saved to: {path}")
+    """
+    _validate_url(url)
+    sanitized_output = _sanitize_output_path(output)
+    resolved_ffmpeg = _validate_ffmpeg_path(ffmpeg_location)
+
+    # Extract format metadata to check for audio and validate format_id
+    try:
+        ydl_opts = _build_base_ydl_opts(ffmpeg_location=resolved_ffmpeg)
+        ydl_opts["extract_flat"] = False
+        ydl_opts["skip_download"] = True
+
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(url, download=False)
+    except (DownloadError, ExtractorError) as exc:
+        raise DownloadFailedError(
+            f"Cannot verify format availability: {exc}"
+        ) from exc
+    except Exception as exc:
+        raise DownloadFailedError(
+            f"Unexpected error during format check: {exc}"
+        ) from exc
+
+    if info is None:
+        raise DownloadFailedError(
+            "No information could be extracted from the URL."
+        )
+
+    # Find the requested format in available formats
+    formats_raw: list[dict[str, Any]] = info.get("formats") or []
+    available_ids = {str(fmt.get("format_id", "")) for fmt in formats_raw}
+
+    if format_id not in available_ids:
+        raise FormatNotFoundError(
+            f"Format ID '{format_id}' not found. "
+            f"Available IDs: {sorted(available_ids)}"
+        )
+
+    # Determine if we need to add audio
+    selected_format = next(
+        fmt for fmt in formats_raw
+        if str(fmt.get("format_id", "")) == format_id
+    )
+
+    if _format_has_audio(selected_format):
+        # Format already has audio — use as-is
+        final_format = format_id
+        logger.info(
+            "Format '%s' already includes audio. Downloading as-is.",
+            format_id,
+        )
+    else:
+        # Video-only format — merge with best audio
+        final_format = f"{format_id}+bestaudio"
+        logger.info(
+            "Format '%s' is video-only. Merging with best audio → '%s'.",
+            format_id,
+            final_format,
+        )
+
+    # Build yt-dlp options for download
+    ydl_opts = _build_base_ydl_opts(ffmpeg_location=resolved_ffmpeg)
+    ydl_opts["format"] = final_format
+    ydl_opts["merge_output_format"] = "mp4"
+    ydl_opts["overwrites"] = True
+
+    # If user specifies an output path, use it exactly
+    if sanitized_output is not None:
+        ydl_opts["outtmpl"] = str(sanitized_output)
+        ydl_opts["paths"] = {}
+
+    try:
+        with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+            info = ydl.extract_info(url, download=True)
+    except (DownloadError, ExtractorError) as exc:
+        raise DownloadFailedError(
+            f"Download failed for format '{final_format}': {exc}"
+        ) from exc
+    except Exception as exc:
+        raise DownloadFailedError(
+            f"Unexpected error during download: {exc}"
+        ) from exc
+
+    if info is None:
+        raise DownloadFailedError(
+            "Download completed but no information was returned."
+        )
+
+    # Determine the actual output path
+    if sanitized_output is not None:
+        final_path = sanitized_output
+    else:
+        requested_downloads = info.get("requested_downloads") or []
+        if requested_downloads:
+            final_path = Path(requested_downloads[0].get("filepath", ""))
+        else:
+            final_path = Path(ydl.prepare_filename(info))
+
+    final_path = final_path.resolve()
+
+    if not final_path.exists():
+        raise DownloadFailedError(
+            f"Download appeared to succeed but file not found at: "
+            f"{final_path}"
+        )
+
+    logger.info("Video downloaded successfully to: %s", final_path)
+    return str(final_path)