Implement prodl-core shared package

packages/prodl-core/prodl_core/__init__.py

@@ -0,0 +1,26 @@
+from .models import (
+    VideoFormat,
+    VideoInfo,
+    SubtitleInfo,
+    DownloadOptions,
+    DownloadResult,
+    StreamInfo,
+)
+from .downloader import ProDLDownloader
+from .thumbnail import fetch_thumbnail
+from .subtitles import list_subtitles
+from .sponsorblock import get_segments
+
+__version__ = "0.1.0"
+__all__ = [
+    "ProDLDownloader",
+    "VideoFormat",
+    "VideoInfo",
+    "SubtitleInfo",
+    "DownloadOptions",
+    "DownloadResult",
+    "StreamInfo",
+    "fetch_thumbnail",
+    "list_subtitles",
+    "get_segments",
+]

packages/prodl-core/prodl_core/downloader.py

@@ -0,0 +1,235 @@
+import asyncio
+import glob
+import os
+from typing import Any
+import yt_dlp
+
+from .models import (
+    DownloadOptions,
+    DownloadResult,
+    StreamInfo,
+    SubtitleInfo,
+    VideoFormat,
+    VideoInfo,
+)
+
+
+class ProDLDownloader:
+
+    @staticmethod
+    async def get_info(url: str) -> VideoInfo:
+        """
+        Use yt-dlp to extract video metadata, formats, and subtitles.
+        """
+        opts = {
+            'quiet': True,
+            'no_warnings': True,
+        }
+
+        def _extract() -> VideoInfo:
+            with yt_dlp.YoutubeDL(opts) as ydl:
+                info = ydl.extract_info(url, download=False)
+                if info is None:
+                    raise ValueError(f"Could not extract info for URL: {url}")
+
+                # Format extraction & filtering
+                raw_formats = info.get("formats") or []
+                video_formats: list[VideoFormat] = []
+
+                for fmt in raw_formats:
+                    vcodec = fmt.get("vcodec")
+                    acodec = fmt.get("acodec")
+
+                    has_video = vcodec is not None and vcodec != "none"
+                    has_audio = acodec is not None and acodec != "none"
+
+                    if not (has_video or has_audio):
+                        continue
+
+                    res = fmt.get("resolution")
+                    if not res:
+                        w = fmt.get("width")
+                        h = fmt.get("height")
+                        if w and h:
+                            res = f"{w}x{h}"
+                        elif has_video:
+                            res = "video only"
+                        else:
+                            res = "audio only"
+
+                    filesize = fmt.get("filesize") or fmt.get("filesize_approx")
+                    note = fmt.get("format_note") or fmt.get("note") or ""
+
+                    video_formats.append(
+                        VideoFormat(
+                            format_id=str(fmt.get("format_id", "")),
+                            ext=str(fmt.get("ext", "")),
+                            resolution=str(res),
+                            filesize=filesize,
+                            note=str(note),
+                            has_video=has_video,
+                            has_audio=has_audio,
+                        )
+                    )
+
+                # Subtitles extraction (manual + auto)
+                subtitles_list: list[SubtitleInfo] = []
+
+                raw_subs = info.get("subtitles") or {}
+                for lang, lang_formats in raw_subs.items():
+                    lang_name = lang
+                    if isinstance(lang_formats, list) and len(lang_formats) > 0:
+                        lang_name = lang_formats[0].get("name") or lang
+                    subtitles_list.append(
+                        SubtitleInfo(lang=str(lang), lang_name=str(lang_name), is_auto=False)
+                    )
+
+                raw_auto = info.get("automatic_captions") or {}
+                for lang, lang_formats in raw_auto.items():
+                    lang_name = lang
+                    if isinstance(lang_formats, list) and len(lang_formats) > 0:
+                        lang_name = lang_formats[0].get("name") or lang
+                    subtitles_list.append(
+                        SubtitleInfo(lang=str(lang), lang_name=str(lang_name), is_auto=True)
+                    )
+
+                return VideoInfo(
+                    title=str(info.get("title") or ""),
+                    duration=info.get("duration"),
+                    thumbnail=info.get("thumbnail"),
+                    uploader=info.get("uploader"),
+                    view_count=info.get("view_count"),
+                    webpage_url=str(info.get("webpage_url") or url),
+                    extractor=str(info.get("extractor") or ""),
+                    formats=video_formats,
+                    subtitles=subtitles_list,
+                )
+
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(None, _extract)
+
+    @staticmethod
+    async def get_stream_info(opts: DownloadOptions) -> StreamInfo:
+        """
+        Extract direct CDN URL for video stream without downloading.
+        """
+        ydl_opts = {
+            'quiet': True,
+            'format': opts.format_id,
+            'skip_download': True,
+        }
+
+        def _extract() -> StreamInfo:
+            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                info = ydl.extract_info(opts.url, download=False)
+                if info is None:
+                    raise ValueError(f"Could not extract stream info for URL: {opts.url}")
+
+                direct_url = info.get("url")
+                if not direct_url:
+                    req_dl = info.get("requested_downloads")
+                    if req_dl and isinstance(req_dl, list) and len(req_dl) > 0:
+                        direct_url = req_dl[0].get("url")
+
+                if not direct_url:
+                    raise ValueError(f"Direct stream URL not found for: {opts.url}")
+
+                filesize = info.get("filesize") or info.get("filesize_approx")
+
+                return StreamInfo(
+                    url=str(direct_url),
+                    title=str(info.get("title") or ""),
+                    ext=str(info.get("ext") or ""),
+                    filesize=filesize,
+                )
+
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(None, _extract)
+
+    @staticmethod
+    async def download(opts: DownloadOptions) -> DownloadResult:
+        """
+        Download video to local disk based on options.
+        """
+        os.makedirs(opts.output_dir, exist_ok=True)
+
+        ydl_opts: dict[str, Any] = {
+            'quiet': True,
+            'no_warnings': True,
+            'outtmpl': f'{opts.output_dir}/%(id)s.%(ext)s',
+            'merge_output_format': 'mp4',
+        }
+
+        if opts.audio_only:
+            ydl_opts['format'] = 'bestaudio/best'
+            ydl_opts['postprocessors'] = [
+                {
+                    'key': 'FFmpegExtractAudio',
+                    'preferredcodec': 'mp3',
+                    'preferredquality': '192',
+                }
+            ]
+        else:
+            ydl_opts['format'] = opts.format_id
+
+        if opts.subtitles:
+            ydl_opts['writesubtitles'] = True
+            ydl_opts['writeautomaticsub'] = opts.auto_subs
+            ydl_opts['subtitleslangs'] = opts.sub_langs
+            ydl_opts['embedsubtitles'] = opts.sub_embed
+
+        if opts.thumbnail:
+            ydl_opts['writethumbnail'] = True
+            ydl_opts['embedthumbnail'] = opts.thumb_embed
+
+        if opts.sponsorblock:
+            if opts.sb_action == "cut":
+                ydl_opts['sponsorblock_remove'] = opts.sb_categories
+            elif opts.sb_action == "skip":
+                ydl_opts['sponsorblock_mark'] = opts.sb_categories
+            elif opts.sb_action == "mark":
+                ydl_opts['sponsorblock_chapter_title'] = opts.sb_categories
+
+        def _download() -> DownloadResult:
+            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                info = ydl.extract_info(opts.url, download=True)
+                if info is None:
+                    raise ValueError(f"Download failed for URL: {opts.url}")
+
+                video_id = info.get("id")
+                title = str(info.get("title") or "")
+
+                # Find output file by glob pattern
+                matching_files = glob.glob(f"{opts.output_dir}/{video_id}.*")
+
+                # Exclude temporary download files or info json if any
+                matching_files = [
+                    f for f in matching_files
+                    if not f.endswith(".part") and not f.endswith(".info.json") and not f.endswith(".ytdl")
+                ]
+
+                if matching_files:
+                    filepath = matching_files[0]
+                else:
+                    # Fallback to requested download filename or expected filepath
+                    filepath = f"{opts.output_dir}/{video_id}.mp4"
+
+                filename = os.path.basename(filepath)
+                ext = os.path.splitext(filepath)[1].lstrip(".")
+
+                filesize = None
+                if os.path.exists(filepath):
+                    filesize = os.path.getsize(filepath)
+                else:
+                    filesize = info.get("filesize") or info.get("filesize_approx")
+
+                return DownloadResult(
+                    filepath=filepath,
+                    filename=filename,
+                    title=title,
+                    ext=ext,
+                    filesize=filesize,
+                )
+
+        loop = asyncio.get_running_loop()
+        return await loop.run_in_executor(None, _download)

packages/prodl-core/prodl_core/models.py

@@ -0,0 +1,69 @@
+from typing import Literal
+from pydantic import BaseModel
+
+
+class VideoFormat(BaseModel):
+    format_id: str
+    ext: str
+    resolution: str
+    filesize: int | None
+    note: str
+    has_video: bool
+    has_audio: bool
+
+
+class SubtitleInfo(BaseModel):
+    lang: str
+    lang_name: str
+    is_auto: bool  # YouTube auto-generated
+
+
+class VideoInfo(BaseModel):
+    title: str
+    duration: int | None
+    thumbnail: str | None
+    uploader: str | None
+    view_count: int | None
+    webpage_url: str
+    extractor: str
+    formats: list[VideoFormat]
+    subtitles: list[SubtitleInfo]  # available subs
+
+
+class DownloadOptions(BaseModel):
+    url: str
+    format_id: str = "bestvideo[ext=mp4]+bestaudio[ext=m4a]/best"
+    audio_only: bool = False
+
+    # Subtitles
+    subtitles: bool = False
+    sub_langs: list[str] = ["en", "ar"]
+    sub_embed: bool = False
+    auto_subs: bool = True
+
+    # Thumbnail
+    thumbnail: bool = False
+    thumb_embed: bool = True  # embed in MP3 as cover art
+
+    # SponsorBlock
+    sponsorblock: bool = False
+    sb_action: Literal["skip", "cut", "mark"] = "cut"
+    sb_categories: list[str] = ["sponsor", "intro", "outro"]
+
+    # Output
+    output_dir: str = "/tmp/prodl_downloads"
+
+
+class DownloadResult(BaseModel):
+    filepath: str
+    filename: str
+    title: str
+    ext: str
+    filesize: int | None
+
+
+class StreamInfo(BaseModel):
+    url: str  # direct stream URL from yt-dlp
+    title: str
+    ext: str
+    filesize: int | None

packages/prodl-core/prodl_core/sponsorblock.py

@@ -0,0 +1,28 @@
+import json
+import httpx
+
+
+async def get_segments(video_id: str, categories: list[str]) -> list[dict]:
+    """
+    Call SponsorBlock API directly (optional utility).
+    GET https://sponsor.ajay.app/api/skipSegments
+        ?videoID={video_id}
+        &categories={json_encoded_categories}
+
+    Use httpx async client.
+    Return list of segment dicts as-is from the API.
+    If 404 (no segments found) → return empty list.
+    """
+    encoded_categories = json.dumps(categories)
+    params = {
+        "videoID": video_id,
+        "categories": encoded_categories,
+    }
+    url = "https://sponsor.ajay.app/api/skipSegments"
+
+    async with httpx.AsyncClient() as client:
+        response = await client.get(url, params=params)
+        if response.status_code == 404:
+            return []
+        response.raise_for_status()
+        return response.json()