diff --git a/README.md b/README.md index 1937a53..075a090 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,2 @@ # just-download simple frontend + backend to download various aspects (thumnails, audio, full video, etc) from a youtube link and instagram(broken a bit, will work lator) - - -- youtube works for now expect avater + banner extraction -- instagram doesnt work at all (will be fixed lator) -- change the ip:port in index.html to your backend route diff --git a/backend/routers/youtube.py b/backend/routers/youtube.py index 7399703..93be74e 100644 --- a/backend/routers/youtube.py +++ b/backend/routers/youtube.py @@ -106,9 +106,20 @@ async def subtitles(req: SubtitleRequest, bg: BackgroundTasks): @router.post("/channel-art") -async def channel_art(req: ChannelRequest): +async def channel_art(req: ChannelRequest, bg: BackgroundTasks): + """ + Returns JSON with download URLs for avatar and banner images. + Frontend fetches each URL separately to trigger downloads. + Response: {"assets": [{"type": "avatar", "url": "/files/..."}, ...]} + """ try: assets = await download_channel_art(req.channel_url) except Exception as e: raise HTTPException(500, detail=str(e)) + + # schedule cleanup for each file + for asset in assets: + path = f"downloads/{asset['filename']}" + bg.add_task(_delete_later, path) + return {"assets": assets} diff --git a/backend/services/instagram_service.py b/backend/services/instagram_service.py index b3cc62f..307e1b3 100644 --- a/backend/services/instagram_service.py +++ b/backend/services/instagram_service.py @@ -2,14 +2,20 @@ 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) + - Posts (images): instaloader with saved session + - Posts (videos): yt-dlp first, instaloader fallback + - Reels: yt-dlp (instaloader 403s on reels) + - Profile pics: instaloader with saved session + - Thumbnails: yt-dlp -Instagram blocks unauthenticated API calls aggressively. -For private content set IG_USERNAME / IG_PASSWORD env vars. +Auth: run this once on the tablet to create ig_session: + python3 -c " + import instaloader, os + L = instaloader.Instaloader() + L.interactive_login('your_username') + L.save_session_to_file('ig_session') + print('saved') + " """ import asyncio @@ -26,15 +32,16 @@ import instaloader DOWNLOAD_DIR = Path("downloads") DOWNLOAD_DIR.mkdir(exist_ok=True) -IG_USERNAME = os.getenv("IG_USERNAME", "") -IG_PASSWORD = os.getenv("IG_PASSWORD", "") +SESSION_FILE = Path("ig_session") +IG_USERNAME = os.getenv("IG_USERNAME", "") +IG_PASSWORD = os.getenv("IG_PASSWORD", "") _loader: Optional[instaloader.Instaloader] = None _BASE_YDL = { - "quiet": True, + "quiet": True, "no_warnings": True, - "noprogress": True, + "noprogress": True, } @@ -62,57 +69,63 @@ def _fmt_s(s) -> str: 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 + if _loader is not None: + return _loader + + L = instaloader.Instaloader( + download_pictures=True, + download_videos=True, + download_video_thumbnails=False, + save_metadata=False, + post_metadata_txt_pattern="", + quiet=True, + ) + + # prefer saved session file — avoids triggering suspicious login alerts + if SESSION_FILE.exists(): + try: + L.load_session_from_file(IG_USERNAME or "user", str(SESSION_FILE)) + print("[ig] loaded session from file") + _loader = L + return _loader + except Exception as e: + print(f"[ig] session file load failed: {e}") + + # fall back to env var credentials + if IG_USERNAME and IG_PASSWORD: + try: + L.login(IG_USERNAME, IG_PASSWORD) + print("[ig] logged in with password") + _loader = L + return _loader + except Exception as e: + print(f"[ig] password login failed: {e}") + + print("[ig] WARNING: no session — most IG content requires login now") + _loader = L 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" + f"could not extract shortcode from: {url}\n" + "must be 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}") + if username in ("p", "reel", "reels", "tv", "explore", "stories", "accounts"): + raise ValueError( + f"that looks like a post/page URL, not a profile — got: {username}" + ) return username @@ -131,19 +144,23 @@ def _trim_video(src: Path, dest: Path, start: Optional[str], end: Optional[str]) return dest +def _download_image(url: str, dest: Path) -> Path: + import httpx + resp = httpx.get(url, follow_redirects=True, timeout=20) + resp.raise_for_status() + dest.write_bytes(resp.content) + 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", + "type": "video", "title": info.get("title") or info.get("uploader") or "Instagram post", "thumbnail": info.get("thumbnail"), "duration": info.get("duration"), @@ -173,7 +190,7 @@ async def fetch_ig_info(url: str) -> dict: return await _run(_fetch_instaloader) -# ── post (image or video) ───────────────────────────────────────────────────── +# ── post ────────────────────────────────────────────────────────────────────── async def download_post( url: str, @@ -187,25 +204,25 @@ async def download_post( out_tmpl = str(DOWNLOAD_DIR / f"ig_post_{uid}.%(ext)s") opts = { **_BASE_YDL, - "format": "bestvideo+bestaudio/best", - "outtmpl": out_tmpl, + "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] + + 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(): @@ -219,7 +236,8 @@ async def download_post( 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")), + (f for f in sorted(tmp_dir.iterdir()) + if f.suffix in (".mp4", ".jpg", ".jpeg", ".png")), None, ) if not media: @@ -237,7 +255,6 @@ async def download_post( finally: shutil.rmtree(tmp_dir, ignore_errors=True) - # try yt-dlp first, fall back to instaloader try: return await _run(_try_ytdlp) except Exception: @@ -245,7 +262,7 @@ async def download_post( return await _run(_try_instaloader) -# ── reel (yt-dlp only — instaloader 403s on reels) ─────────────────────────── +# ── reel ────────────────────────────────────────────────────────────────────── async def download_reel( url: str, @@ -267,12 +284,10 @@ async def download_reel( 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] + 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" @@ -288,7 +303,6 @@ async def download_reel( # ── 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: @@ -300,13 +314,15 @@ async def download_profile(username_or_url: str) -> Tuple[str, str]: 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) + L = _get_loader() + profile = instaloader.Profile.from_username(L.context, username) + # profile_pic_url_no_iphone gives higher resolution + try: + pic_url = profile.profile_pic_url_no_iphone + except Exception: + pic_url = profile.profile_pic_url + path = DOWNLOAD_DIR / f"ig_pfp_{_safe(username)}_{uid}.jpg" + _download_image(pic_url, path) return str(path), f"{_safe(username)}_profile.jpg" return await _run(_dl) @@ -318,8 +334,6 @@ 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) @@ -328,8 +342,7 @@ async def download_ig_thumbnail(url: str) -> Tuple[str, str]: 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) + _download_image(thumb_url, path) 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 index ed43242..b5bf909 100644 --- a/backend/services/youtube_service.py +++ b/backend/services/youtube_service.py @@ -3,21 +3,20 @@ YouTube service — all yt-dlp operations. """ import asyncio -import uuid import re +import uuid from pathlib import Path -from typing import Optional, Tuple +from typing import Optional, Tuple, List 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, + "quiet": True, "no_warnings": True, - "noprogress": True, + "noprogress": True, } @@ -34,7 +33,7 @@ def _safe(name: str) -> str: return re.sub(r'[\\/*?:"<>|]', "_", name or "file")[:80] -def _fmt_duration(seconds: int) -> str: +def _fmt_duration(seconds) -> str: if not seconds: return "0:00" h, rem = divmod(int(seconds), 3600) @@ -51,6 +50,14 @@ def _parse_time(t: str) -> float: return parts[0] +def _download_image(url: str, dest: Path) -> Path: + import httpx + resp = httpx.get(url, follow_redirects=True, timeout=20) + resp.raise_for_status() + dest.write_bytes(resp.content) + return dest + + # ── info ────────────────────────────────────────────────────────────────────── async def fetch_yt_info(url: str) -> dict: @@ -92,10 +99,9 @@ async def download_video( start_time: Optional[str] = None, end_time: Optional[str] = None, ) -> Tuple[str, str]: - uid = _uid() + 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: @@ -106,9 +112,9 @@ async def download_video( opts = { **_BASE_OPTS, - "format": fmt_sel, - "outtmpl": out_tmpl, - "merge_output_format": fmt, + "format": fmt_sel, + "outtmpl": out_tmpl, + "merge_output_format": fmt, } if start_time or end_time: @@ -119,23 +125,18 @@ async def download_video( 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["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('.')}" + 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}" return await _run(_dl) @@ -171,13 +172,11 @@ async def download_audio( 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('.')}" + 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}" return await _run(_dl) @@ -216,12 +215,12 @@ async def download_subtitles(url: str, lang: str = "en") -> Tuple[str, str]: opts = { **_BASE_OPTS, - "skip_download": True, - "writesubtitles": True, - "writeautomaticsub": True, - "subtitleslangs": [lang], - "subtitlesformat": "srt", - "outtmpl": out_tmpl, + "skip_download": True, + "writesubtitles": True, + "writeautomaticsub": True, + "subtitleslangs": [lang], + "subtitlesformat": "srt", + "outtmpl": out_tmpl, } def _dl(): @@ -229,52 +228,127 @@ async def download_subtitles(url: str, lang: str = "en") -> Tuple[str, str]: 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") + 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 +async def download_channel_art(channel_url: str) -> List[dict]: + """ + Downloads channel avatar and banner as separate JPEG files. + Returns list of {"type", "filename", "url"} dicts. + Avatar and banner are identified by aspect ratio heuristics. + """ - opts = {**_BASE_OPTS, "skip_download": True, "extract_flat": True} + def _dl(): + uid = _uid() + results = [] + + 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") + + # ── avatar ──────────────────────────────────────────────────────────── + # avatars are square-ish; YouTube avatar thumbnails often come from + # ggpht.com or have "photo" in the URL + avatar_url = None + avatar_candidates = [ + t for t in thumbnails + if t.get("url") and _is_avatar(t) + ] + if avatar_candidates: + # pick highest resolution among avatar candidates + avatar_url = max( + avatar_candidates, + key=lambda t: (t.get("width") or 0) * (t.get("height") or 0) + )["url"] + elif thumbnails: + # last resort: use the last thumbnail + avatar_url = thumbnails[-1]["url"] + 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}"}) + try: + _download_image(avatar_url, path) + results.append({ + "type": "avatar", + "filename": path.name, + "url": f"/files/{path.name}", + }) + except Exception as e: + print(f"[yt] avatar download failed: {e}") + + # ── banner ──────────────────────────────────────────────────────────── + # banners are very wide (ratio > 3:1), or keyed explicitly + banner_url = ( + info.get("header_image") + or info.get("banner") + ) + if not banner_url: + banner_candidates = [ + t for t in thumbnails + if t.get("url") and _is_banner(t) + ] + if banner_candidates: + banner_url = max( + banner_candidates, + key=lambda t: (t.get("width") or 0) + )["url"] - # 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}"}) + try: + _download_image(banner_url, path) + results.append({ + "type": "banner", + "filename": path.name, + "url": f"/files/{path.name}", + }) + except Exception as e: + print(f"[yt] banner download failed: {e}") if not results: - raise ValueError("no channel art found — try passing the full channel URL") + raise ValueError( + "no channel art found — pass a full channel URL like " + "youtube.com/@channelname or youtube.com/channel/UCxxxxxx" + ) return results return await _run(_dl) + + +def _is_avatar(t: dict) -> bool: + w = t.get("width") or 0 + h = t.get("height") or 1 + url = t.get("url", "") + if w and h: + ratio = w / h + if 0.8 < ratio < 1.3: + return True + return any(kw in url for kw in ("ggpht", "photo", "avatar")) + + +def _is_banner(t: dict) -> bool: + w = t.get("width") or 0 + h = t.get("height") or 1 + url = t.get("url", "") + if w and h and (w / h) > 2.5: + return True + return "banner" in url.lower()