diff --git a/backend/main.py b/backend/main.py new file mode 100644 index 0000000..dbd999f --- /dev/null +++ b/backend/main.py @@ -0,0 +1,69 @@ +""" +just download stuff lol — FastAPI backend +Run: uvicorn main:app --reload --host 0.0.0.0 --port 8000 +""" + +import asyncio +import os +from contextlib import asynccontextmanager +from pathlib import Path + +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware +from fastapi.staticfiles import StaticFiles + +from routers import youtube, instagram, info + +DOWNLOAD_DIR = Path("downloads") +DOWNLOAD_DIR.mkdir(exist_ok=True) + + +@asynccontextmanager +async def lifespan(app: FastAPI): + asyncio.create_task(_cleanup_loop()) + yield + + +app = FastAPI( + title="just download stuff lol", + description="no cap, it just works™", + version="2.0.0", + lifespan=lifespan, +) + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +app.include_router(info.router, prefix="/api/info", tags=["info"]) +app.include_router(youtube.router, prefix="/api/youtube", tags=["youtube"]) +app.include_router(instagram.router, prefix="/api/instagram", tags=["instagram"]) + +app.mount("/files", StaticFiles(directory="downloads"), name="files") + + +@app.get("/") +async def root(): + return {"status": "ok", "message": "just download stuff lol"} + + +@app.get("/health") +async def health(): + return {"status": "healthy"} + + +async def _cleanup_loop(): + """Delete files older than 1 hour, runs every 10 minutes.""" + while True: + await asyncio.sleep(600) + now = asyncio.get_event_loop().time() + for f in DOWNLOAD_DIR.iterdir(): + try: + if f.is_file() and (now - os.path.getmtime(f)) > 3600: + f.unlink() + except Exception: + pass diff --git a/backend/requirements.txt b/backend/requirements.txt new file mode 100644 index 0000000..6e4c429 --- /dev/null +++ b/backend/requirements.txt @@ -0,0 +1,8 @@ +pydantic==1.10.21 +fastapi==0.100.1 +uvicorn>=0.29.0 +yt-dlp>=2024.4.9 +instaloader>=4.10.3 +httpx>=0.27.0 +python-multipart>=0.0.9 +aiofiles>=23.2.1 diff --git a/backend/routers/info.py b/backend/routers/info.py new file mode 100644 index 0000000..a07d1d2 --- /dev/null +++ b/backend/routers/info.py @@ -0,0 +1,52 @@ +""" +/api/info — platform detection and metadata preview +""" + +import re +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel + +from services.youtube_service import fetch_yt_info +from services.instagram_service import fetch_ig_info + +router = APIRouter() + +YT_RE = re.compile(r"(youtube\.com|youtu\.be)") +IG_RE = re.compile(r"(instagram\.com|instagr\.am)") + + +class InfoRequest(BaseModel): + url: str + + +def detect_platform(url: str) -> str: + if YT_RE.search(url): + return "youtube" + if IG_RE.search(url): + return "instagram" + return "unknown" + + +@router.post("/detect") +async def detect(req: InfoRequest): + platform = detect_platform(req.url) + if platform == "unknown": + raise HTTPException(400, detail="could not detect platform from that url") + return {"platform": platform, "url": req.url} + + +@router.post("/preview") +async def preview(req: InfoRequest): + platform = detect_platform(req.url) + try: + if platform == "youtube": + data = await fetch_yt_info(req.url) + elif platform == "instagram": + data = await fetch_ig_info(req.url) + else: + raise HTTPException(400, detail="unsupported platform — only YouTube and Instagram") + except HTTPException: + raise + except Exception as e: + raise HTTPException(500, detail=f"fetch failed: {str(e)}") + return {"platform": platform, **data} diff --git a/backend/routers/instagram.py b/backend/routers/instagram.py new file mode 100644 index 0000000..36fe290 --- /dev/null +++ b/backend/routers/instagram.py @@ -0,0 +1,94 @@ +""" +/api/instagram — post, reel, story, profile, thumbnail +""" + +import asyncio +import os +from fastapi import APIRouter, HTTPException, BackgroundTasks +from fastapi.responses import FileResponse +from pydantic import BaseModel +from typing import Optional + +from services.instagram_service import ( + download_post, + download_reel, + download_profile, + download_ig_thumbnail, + fetch_ig_info, +) + +router = APIRouter() + + +class PostRequest(BaseModel): + url: str + format: str = "mp4" + start_time: Optional[str] = None + end_time: Optional[str] = None + + +class ProfileRequest(BaseModel): + username: str + + +class SimpleRequest(BaseModel): + url: str + + +async def _delete_later(path: str, delay: int = 3600): + await asyncio.sleep(delay) + try: + os.unlink(path) + except Exception: + pass + + +@router.post("/post") +async def post(req: PostRequest, bg: BackgroundTasks): + try: + path, filename = await download_post( + url=req.url, + fmt=req.format, + start_time=req.start_time, + end_time=req.end_time, + ) + except Exception as e: + raise HTTPException(500, detail=str(e)) + bg.add_task(_delete_later, path) + media_type = "video/mp4" if filename.endswith(".mp4") else "image/jpeg" + return FileResponse(path, filename=filename, media_type=media_type) + + +@router.post("/reel") +async def reel(req: PostRequest, bg: BackgroundTasks): + # reels use yt-dlp directly — more reliable than instaloader for reels + try: + path, filename = await download_reel( + url=req.url, + start_time=req.start_time, + end_time=req.end_time, + ) + except Exception as e: + raise HTTPException(500, detail=str(e)) + bg.add_task(_delete_later, path) + return FileResponse(path, filename=filename, media_type="video/mp4") + + +@router.post("/profile") +async def profile(req: ProfileRequest, bg: BackgroundTasks): + try: + path, filename = await download_profile(req.username) + except Exception as e: + raise HTTPException(500, detail=str(e)) + bg.add_task(_delete_later, path) + return FileResponse(path, filename=filename, media_type="image/jpeg") + + +@router.post("/thumbnail") +async def thumbnail(req: SimpleRequest, bg: BackgroundTasks): + try: + path, filename = await download_ig_thumbnail(req.url) + except Exception as e: + raise HTTPException(500, detail=str(e)) + bg.add_task(_delete_later, path) + return FileResponse(path, filename=filename, media_type="image/jpeg") diff --git a/backend/routers/youtube.py b/backend/routers/youtube.py new file mode 100644 index 0000000..7399703 --- /dev/null +++ b/backend/routers/youtube.py @@ -0,0 +1,114 @@ +""" +/api/youtube — video, audio, thumbnail, subtitles, channel art +""" + +import asyncio +import os +from fastapi import APIRouter, HTTPException, BackgroundTasks +from fastapi.responses import FileResponse +from pydantic import BaseModel +from typing import Optional + +from services.youtube_service import ( + download_video, + download_audio, + download_thumbnail, + download_subtitles, + download_channel_art, +) + +router = APIRouter() + + +class VideoRequest(BaseModel): + url: str + resolution: str = "best" + format: str = "mp4" + start_time: Optional[str] = None + end_time: Optional[str] = None + + +class AudioRequest(BaseModel): + url: str + quality: str = "192" + format: str = "mp3" + + +class SimpleRequest(BaseModel): + url: str + + +class SubtitleRequest(BaseModel): + url: str + lang: str = "en" + + +class ChannelRequest(BaseModel): + channel_url: str + + +async def _delete_later(path: str, delay: int = 3600): + await asyncio.sleep(delay) + try: + os.unlink(path) + except Exception: + pass + + +@router.post("/video") +async def video(req: VideoRequest, bg: BackgroundTasks): + try: + path, filename = await download_video( + url=req.url, + resolution=req.resolution, + fmt=req.format, + start_time=req.start_time, + end_time=req.end_time, + ) + except Exception as e: + raise HTTPException(500, detail=str(e)) + bg.add_task(_delete_later, path) + return FileResponse(path, filename=filename, media_type="video/mp4") + + +@router.post("/audio") +async def audio(req: AudioRequest, bg: BackgroundTasks): + try: + path, filename = await download_audio( + url=req.url, + quality=req.quality, + fmt=req.format, + ) + except Exception as e: + raise HTTPException(500, detail=str(e)) + bg.add_task(_delete_later, path) + return FileResponse(path, filename=filename, media_type="audio/mpeg") + + +@router.post("/thumbnail") +async def thumbnail(req: SimpleRequest, bg: BackgroundTasks): + try: + path, filename = await download_thumbnail(req.url) + except Exception as e: + raise HTTPException(500, detail=str(e)) + bg.add_task(_delete_later, path) + return FileResponse(path, filename=filename, media_type="image/jpeg") + + +@router.post("/subtitles") +async def subtitles(req: SubtitleRequest, bg: BackgroundTasks): + try: + path, filename = await download_subtitles(req.url, req.lang) + except Exception as e: + raise HTTPException(500, detail=str(e)) + bg.add_task(_delete_later, path) + return FileResponse(path, filename=filename, media_type="text/plain") + + +@router.post("/channel-art") +async def channel_art(req: ChannelRequest): + try: + assets = await download_channel_art(req.channel_url) + except Exception as e: + raise HTTPException(500, detail=str(e)) + return {"assets": assets} diff --git a/backend/services/instagram_service.py b/backend/services/instagram_service.py new file mode 100644 index 0000000..b3cc62f --- /dev/null +++ b/backend/services/instagram_service.py @@ -0,0 +1,335 @@ +""" +Instagram service. + +Strategy: + - Posts (images): instaloader (best quality originals) + - Posts (videos): yt-dlp first, instaloader fallback + - Reels: yt-dlp (instaloader 403s on reels consistently) + - Profile pictures: instaloader (direct profile_pic_url) + - Thumbnails: yt-dlp (fast, no auth needed) + +Instagram blocks unauthenticated API calls aggressively. +For private content set IG_USERNAME / IG_PASSWORD env vars. +""" + +import asyncio +import os +import re +import shutil +import uuid +from pathlib import Path +from typing import Optional, Tuple + +import yt_dlp +import instaloader + +DOWNLOAD_DIR = Path("downloads") +DOWNLOAD_DIR.mkdir(exist_ok=True) + +IG_USERNAME = os.getenv("IG_USERNAME", "") +IG_PASSWORD = os.getenv("IG_PASSWORD", "") + +_loader: Optional[instaloader.Instaloader] = None + +_BASE_YDL = { + "quiet": True, + "no_warnings": True, + "noprogress": True, +} + + +# ── helpers ─────────────────────────────────────────────────────────────────── + +def _run(func, *args, **kwargs): + loop = asyncio.get_event_loop() + return loop.run_in_executor(None, lambda: func(*args, **kwargs)) + + +def _uid() -> str: + return uuid.uuid4().hex[:10] + + +def _safe(name: str) -> str: + return re.sub(r'[\\/*?:"<>|]', "_", name or "file")[:80] + + +def _fmt_s(s) -> str: + if not s: + return "0:00" + m, sec = divmod(int(s), 60) + return f"{m}:{sec:02d}" + + +def _get_loader() -> instaloader.Instaloader: + global _loader + if _loader is None: + _loader = instaloader.Instaloader( + download_pictures=True, + download_videos=True, + download_video_thumbnails=False, + save_metadata=False, + post_metadata_txt_pattern="", + quiet=True, + ) + if IG_USERNAME and IG_PASSWORD: + try: + _loader.login(IG_USERNAME, IG_PASSWORD) + except Exception: + pass + return _loader + + +def _shortcode_from_url(url: str) -> str: + """ + Extract shortcode from any Instagram post/reel/tv URL. + Handles: + instagram.com/p/ABC123/ + instagram.com/reel/ABC123/ + instagram.com/tv/ABC123/ + instagram.com/reels/ABC123/ + """ + m = re.search(r"/(?:p|reel|reels|tv)/([A-Za-z0-9_-]+)", url) + if m: + return m.group(1) + raise ValueError( + f"could not extract shortcode from URL: {url}\n" + "make sure it's a post, reel, or tv URL — not a profile or explore page" + ) + + +def _username_from_url(url: str) -> str: + """ + Extract username from instagram.com/@username or instagram.com/username + Strips trailing slashes, query params, etc. + """ + # strip protocol and domain + path = re.sub(r"https?://(www\.)?instagram\.com/?", "", url) + # strip @ if present + path = path.lstrip("@") + # take first path segment + username = path.split("/")[0].split("?")[0].strip() + if not username: + raise ValueError("could not extract username from URL") + # reject if it looks like a post URL + if username in ("p", "reel", "reels", "tv", "explore", "stories"): + raise ValueError(f"that looks like a post URL, not a profile — got segment: {username}") + return username + + +def _trim_video(src: Path, dest: Path, start: Optional[str], end: Optional[str]) -> Path: + import subprocess + cmd = ["ffmpeg", "-y"] + if start: + cmd += ["-ss", start] + cmd += ["-i", str(src)] + if end: + cmd += ["-to", end] + cmd += ["-c", "copy", str(dest)] + result = subprocess.run(cmd, capture_output=True, timeout=120) + if result.returncode != 0: + raise RuntimeError(f"ffmpeg trim failed: {result.stderr.decode()[:200]}") + return dest + + +# ── info ────────────────────────────────────────────────────────────────────── + +async def fetch_ig_info(url: str) -> dict: + """ + Try yt-dlp first (works for reels/public posts without auth), + fall back to instaloader for images and private content. + """ + def _fetch_ytdlp(): + opts = {**_BASE_YDL, "skip_download": True} + with yt_dlp.YoutubeDL(opts) as ydl: + info = ydl.extract_info(url, download=False) + return { + "type": "video" if info.get("ext") != "jpg" else "image", + "title": info.get("title") or info.get("uploader") or "Instagram post", + "thumbnail": info.get("thumbnail"), + "duration": info.get("duration"), + "duration_str": _fmt_s(info.get("duration")), + "owner": info.get("uploader"), + "likes": info.get("like_count"), + } + + def _fetch_instaloader(): + shortcode = _shortcode_from_url(url) + L = _get_loader() + post = instaloader.Post.from_shortcode(L.context, shortcode) + return { + "type": "video" if post.is_video else "image", + "title": f"@{post.owner_username}", + "thumbnail": post.url, + "duration": post.video_duration if post.is_video else None, + "duration_str": _fmt_s(post.video_duration) if post.is_video else None, + "owner": post.owner_username, + "likes": post.likes, + } + + try: + return await _run(_fetch_ytdlp) + except Exception: + pass + return await _run(_fetch_instaloader) + + +# ── post (image or video) ───────────────────────────────────────────────────── + +async def download_post( + url: str, + fmt: str = "mp4", + start_time: Optional[str] = None, + end_time: Optional[str] = None, +) -> Tuple[str, str]: + uid = _uid() + + def _try_ytdlp(): + out_tmpl = str(DOWNLOAD_DIR / f"ig_post_{uid}.%(ext)s") + opts = { + **_BASE_YDL, + "format": "bestvideo+bestaudio/best", + "outtmpl": out_tmpl, + "merge_output_format": "mp4", + } + with yt_dlp.YoutubeDL(opts) as ydl: + info = ydl.extract_info(url) + owner = _safe(info.get("uploader") or "ig") + ext = "mp4" + final = DOWNLOAD_DIR / f"ig_post_{uid}.{ext}" + if not final.exists(): + candidates = list(DOWNLOAD_DIR.glob(f"ig_post_{uid}.*")) + if not candidates: + raise FileNotFoundError("yt-dlp produced no file") + final = candidates[0] + if (start_time or end_time) and final.suffix == ".mp4": + trimmed = DOWNLOAD_DIR / f"ig_post_{uid}_trim.mp4" + _trim_video(final, trimmed, start_time, end_time) + final.unlink(missing_ok=True) + final = trimmed + return str(final), f"{owner}_{uid}{final.suffix}" + + def _try_instaloader(): + tmp_dir = DOWNLOAD_DIR / f"ig_il_{uid}" + tmp_dir.mkdir(exist_ok=True) + try: + shortcode = _shortcode_from_url(url) + L = _get_loader() + post = instaloader.Post.from_shortcode(L.context, shortcode) + L.dirname_pattern = str(tmp_dir) + L.download_post(post, target=str(tmp_dir)) + + media = next( + (f for f in tmp_dir.iterdir() if f.suffix in (".mp4", ".jpg", ".jpeg", ".png")), + None, + ) + if not media: + raise FileNotFoundError("instaloader produced no media file") + + owner = _safe(post.owner_username) + dest = DOWNLOAD_DIR / f"ig_post_{uid}{media.suffix}" + + if post.is_video and (start_time or end_time): + _trim_video(media, dest, start_time, end_time) + else: + shutil.copy2(media, dest) + + return str(dest), f"{owner}_{uid}{media.suffix}" + finally: + shutil.rmtree(tmp_dir, ignore_errors=True) + + # try yt-dlp first, fall back to instaloader + try: + return await _run(_try_ytdlp) + except Exception: + pass + return await _run(_try_instaloader) + + +# ── reel (yt-dlp only — instaloader 403s on reels) ─────────────────────────── + +async def download_reel( + url: str, + start_time: Optional[str] = None, + end_time: Optional[str] = None, +) -> Tuple[str, str]: + uid = _uid() + out_tmpl = str(DOWNLOAD_DIR / f"ig_reel_{uid}.%(ext)s") + + opts = { + **_BASE_YDL, + "format": "bestvideo+bestaudio/best", + "outtmpl": out_tmpl, + "merge_output_format": "mp4", + } + + def _dl(): + with yt_dlp.YoutubeDL(opts) as ydl: + info = ydl.extract_info(url) + owner = _safe(info.get("uploader") or "ig") + + final = DOWNLOAD_DIR / f"ig_reel_{uid}.mp4" + if not final.exists(): + candidates = list(DOWNLOAD_DIR.glob(f"ig_reel_{uid}.*")) + if not candidates: + raise FileNotFoundError("yt-dlp produced no reel file") + final = candidates[0] + + if start_time or end_time: + trimmed = DOWNLOAD_DIR / f"ig_reel_{uid}_trim.mp4" + _trim_video(final, trimmed, start_time, end_time) + final.unlink(missing_ok=True) + final = trimmed + + return str(final), f"{owner}_reel_{uid}.mp4" + + return await _run(_dl) + + +# ── profile picture ─────────────────────────────────────────────────────────── + +async def download_profile(username_or_url: str) -> Tuple[str, str]: + # accept either a username string or a full instagram URL + if "instagram.com" in username_or_url: + username = _username_from_url(username_or_url) + else: + username = username_or_url.lstrip("@").strip() + + if not username: + raise ValueError("no username provided") + + uid = _uid() + + def _dl(): + import httpx + L = _get_loader() + profile = instaloader.Profile.from_username(L.context, username) + pic_url = profile.profile_pic_url + path = DOWNLOAD_DIR / f"ig_pfp_{_safe(username)}_{uid}.jpg" + data = httpx.get(pic_url, follow_redirects=True, timeout=15).content + path.write_bytes(data) + return str(path), f"{_safe(username)}_profile.jpg" + + return await _run(_dl) + + +# ── thumbnail ───────────────────────────────────────────────────────────────── + +async def download_ig_thumbnail(url: str) -> Tuple[str, str]: + uid = _uid() + + def _dl(): + import httpx + # try yt-dlp for thumbnail URL + opts = {**_BASE_YDL, "skip_download": True} + with yt_dlp.YoutubeDL(opts) as ydl: + info = ydl.extract_info(url, download=False) + thumb_url = info.get("thumbnail") + if not thumb_url: + raise ValueError("no thumbnail found for this post") + owner = _safe(info.get("uploader") or "ig") + path = DOWNLOAD_DIR / f"ig_thumb_{uid}.jpg" + data = httpx.get(thumb_url, follow_redirects=True, timeout=15).content + path.write_bytes(data) + return str(path), f"{owner}_thumbnail_{uid}.jpg" + + return await _run(_dl) diff --git a/backend/services/youtube_service.py b/backend/services/youtube_service.py new file mode 100644 index 0000000..ed43242 --- /dev/null +++ b/backend/services/youtube_service.py @@ -0,0 +1,280 @@ +""" +YouTube service — all yt-dlp operations. +""" + +import asyncio +import uuid +import re +from pathlib import Path +from typing import Optional, Tuple + +import yt_dlp + +DOWNLOAD_DIR = Path("downloads") +DOWNLOAD_DIR.mkdir(exist_ok=True) + +# shared yt-dlp options applied to every call +_BASE_OPTS = { + "quiet": True, + "no_warnings": True, + "noprogress": True, +} + + +def _run(func, *args, **kwargs): + loop = asyncio.get_event_loop() + return loop.run_in_executor(None, lambda: func(*args, **kwargs)) + + +def _uid() -> str: + return uuid.uuid4().hex[:10] + + +def _safe(name: str) -> str: + return re.sub(r'[\\/*?:"<>|]', "_", name or "file")[:80] + + +def _fmt_duration(seconds: int) -> str: + if not seconds: + return "0:00" + h, rem = divmod(int(seconds), 3600) + m, s = divmod(rem, 60) + return f"{h}:{m:02d}:{s:02d}" if h else f"{m}:{s:02d}" + + +def _parse_time(t: str) -> float: + parts = [float(p) for p in t.strip().split(":")] + if len(parts) == 3: + return parts[0] * 3600 + parts[1] * 60 + parts[2] + if len(parts) == 2: + return parts[0] * 60 + parts[1] + return parts[0] + + +# ── info ────────────────────────────────────────────────────────────────────── + +async def fetch_yt_info(url: str) -> dict: + def _fetch(): + opts = {**_BASE_OPTS, "skip_download": True} + with yt_dlp.YoutubeDL(opts) as ydl: + info = ydl.extract_info(url, download=False) + formats = info.get("formats") or [] + resolutions = sorted( + {f.get("height") for f in formats if f.get("height")}, + reverse=True, + ) + fps_options = sorted( + {int(f.get("fps")) for f in formats if f.get("fps")}, + reverse=True, + ) + return { + "title": info.get("title"), + "duration": info.get("duration"), + "duration_str": _fmt_duration(info.get("duration", 0)), + "thumbnail": info.get("thumbnail"), + "uploader": info.get("uploader"), + "view_count": info.get("view_count"), + "upload_date": info.get("upload_date"), + "available_resolutions": resolutions, + "fps_options": fps_options, + "has_subtitles": bool(info.get("subtitles")), + "subtitle_langs": list((info.get("subtitles") or {}).keys()), + } + return await _run(_fetch) + + +# ── video ───────────────────────────────────────────────────────────────────── + +async def download_video( + url: str, + resolution: str = "best", + fmt: str = "mp4", + start_time: Optional[str] = None, + end_time: Optional[str] = None, +) -> Tuple[str, str]: + uid = _uid() + out_tmpl = str(DOWNLOAD_DIR / f"yt_video_{uid}.%(ext)s") + + # build format selector + if resolution == "best": + fmt_sel = f"bestvideo[ext={fmt}]+bestaudio/bestvideo+bestaudio/best" + else: + fmt_sel = ( + f"bestvideo[height<={resolution}][ext={fmt}]+bestaudio/" + f"bestvideo[height<={resolution}]+bestaudio/best" + ) + + opts = { + **_BASE_OPTS, + "format": fmt_sel, + "outtmpl": out_tmpl, + "merge_output_format": fmt, + } + + if start_time or end_time: + start_s = _parse_time(start_time) if start_time else 0 + end_s = _parse_time(end_time) if end_time else None + + def _range_fn(info_dict, ydl): + end = end_s if end_s is not None else info_dict.get("duration", 86400) + return [{"start_time": start_s, "end_time": end}] + + opts["download_ranges"] = _range_fn + opts["force_keyframes_at_cuts"] = True + + def _dl(): + with yt_dlp.YoutubeDL(opts) as ydl: + info = ydl.extract_info(url) + title = _safe(info.get("title", "video")) + ext = info.get("ext", fmt) + # find the actual output file + final = DOWNLOAD_DIR / f"yt_video_{uid}.{ext}" + if not final.exists(): + # yt-dlp may have merged to a different ext + candidates = list(DOWNLOAD_DIR.glob(f"yt_video_{uid}.*")) + if not candidates: + raise FileNotFoundError("output file not found after download") + final = candidates[0] + return str(final), f"{title}.{final.suffix.lstrip('.')}" + + return await _run(_dl) + + +# ── audio ───────────────────────────────────────────────────────────────────── + +async def download_audio( + url: str, + quality: str = "192", + fmt: str = "mp3", +) -> Tuple[str, str]: + uid = _uid() + out_tmpl = str(DOWNLOAD_DIR / f"yt_audio_{uid}.%(ext)s") + + codec_map = {"mp3": "mp3", "opus": "opus", "flac": "flac", "m4a": "m4a"} + codec = codec_map.get(fmt, "mp3") + + opts = { + **_BASE_OPTS, + "format": "bestaudio/best", + "outtmpl": out_tmpl, + "postprocessors": [ + { + "key": "FFmpegExtractAudio", + "preferredcodec": codec, + "preferredquality": quality, + }, + {"key": "FFmpegMetadata", "add_metadata": True}, + ], + } + + def _dl(): + with yt_dlp.YoutubeDL(opts) as ydl: + info = ydl.extract_info(url) + title = _safe(info.get("title", "audio")) + final = DOWNLOAD_DIR / f"yt_audio_{uid}.{fmt}" + if not final.exists(): + candidates = list(DOWNLOAD_DIR.glob(f"yt_audio_{uid}.*")) + if not candidates: + raise FileNotFoundError("audio file not found after download") + final = candidates[0] + return str(final), f"{title}.{final.suffix.lstrip('.')}" + + return await _run(_dl) + + +# ── thumbnail ───────────────────────────────────────────────────────────────── + +async def download_thumbnail(url: str) -> Tuple[str, str]: + uid = _uid() + out_tmpl = str(DOWNLOAD_DIR / f"yt_thumb_{uid}.%(ext)s") + + opts = { + **_BASE_OPTS, + "skip_download": True, + "writethumbnail": True, + "outtmpl": out_tmpl, + } + + def _dl(): + with yt_dlp.YoutubeDL(opts) as ydl: + info = ydl.extract_info(url, download=True) + title = _safe(info.get("title", "thumbnail")) + for ext in ("jpg", "webp", "png", "jpeg"): + candidate = DOWNLOAD_DIR / f"yt_thumb_{uid}.{ext}" + if candidate.exists(): + return str(candidate), f"{title}_thumbnail.{ext}" + raise FileNotFoundError("thumbnail not found after download") + + return await _run(_dl) + + +# ── subtitles ───────────────────────────────────────────────────────────────── + +async def download_subtitles(url: str, lang: str = "en") -> Tuple[str, str]: + uid = _uid() + out_tmpl = str(DOWNLOAD_DIR / f"yt_subs_{uid}.%(ext)s") + + opts = { + **_BASE_OPTS, + "skip_download": True, + "writesubtitles": True, + "writeautomaticsub": True, + "subtitleslangs": [lang], + "subtitlesformat": "srt", + "outtmpl": out_tmpl, + } + + def _dl(): + with yt_dlp.YoutubeDL(opts) as ydl: + info = ydl.extract_info(url, download=True) + title = _safe(info.get("title", "subtitles")) + for ext in ("srt", "vtt", "ass"): + # yt-dlp names it: yt_subs_{uid}.{lang}.{ext} + candidate = DOWNLOAD_DIR / f"yt_subs_{uid}.{lang}.{ext}" + if candidate.exists(): + return str(candidate), f"{title}.{lang}.{ext}" + # fallback: any subtitle file matching uid + candidates = list(DOWNLOAD_DIR.glob(f"yt_subs_{uid}.*")) + if candidates: + return str(candidates[0]), candidates[0].name + raise FileNotFoundError(f"no subtitles found for lang={lang} — video may not have them") + + return await _run(_dl) + + +# ── channel art ─────────────────────────────────────────────────────────────── + +async def download_channel_art(channel_url: str) -> list: + def _dl(): + import httpx + + opts = {**_BASE_OPTS, "skip_download": True, "extract_flat": True} + with yt_dlp.YoutubeDL(opts) as ydl: + info = ydl.extract_info(channel_url, download=False) + + results = [] + uid = _uid() + + # avatar — last thumbnail is usually highest res + thumbnails = info.get("thumbnails") or [] + avatar_url = thumbnails[-1].get("url") if thumbnails else info.get("thumbnail") + if avatar_url: + path = DOWNLOAD_DIR / f"yt_ch_{uid}_avatar.jpg" + data = httpx.get(avatar_url, follow_redirects=True, timeout=15).content + path.write_bytes(data) + results.append({"type": "avatar", "url": f"/files/{path.name}"}) + + # banner + banner_url = info.get("header_image") or info.get("banner") + if banner_url: + path = DOWNLOAD_DIR / f"yt_ch_{uid}_banner.jpg" + data = httpx.get(banner_url, follow_redirects=True, timeout=15).content + path.write_bytes(data) + results.append({"type": "banner", "url": f"/files/{path.name}"}) + + if not results: + raise ValueError("no channel art found — try passing the full channel URL") + + return results + + return await _run(_dl) diff --git a/frontend/index.html b/frontend/index.html new file mode 100644 index 0000000..5749525 --- /dev/null +++ b/frontend/index.html @@ -0,0 +1,553 @@ + + +
+ + +