diff --git a/backend/routers/info.py b/backend/routers/info.py index a07d1d2..6abffd4 100644 --- a/backend/routers/info.py +++ b/backend/routers/info.py @@ -6,8 +6,9 @@ import re from fastapi import APIRouter, HTTPException from pydantic import BaseModel +from utils import clean_yt_url, clean_ig_url, is_ig_profile_url, extract_ig_username from services.youtube_service import fetch_yt_info -from services.instagram_service import fetch_ig_info +from services.instagram_service import fetch_ig_post_info, fetch_ig_profile_info router = APIRouter() @@ -37,16 +38,29 @@ async def detect(req: InfoRequest): @router.post("/preview") async def preview(req: InfoRequest): - platform = detect_platform(req.url) + url = req.url.strip() + platform = detect_platform(url) + try: if platform == "youtube": - data = await fetch_yt_info(req.url) + url = clean_yt_url(url) + data = await fetch_yt_info(url) + elif platform == "instagram": - data = await fetch_ig_info(req.url) + url = clean_ig_url(url) + if is_ig_profile_url(url): + # profile page — show username + avatar + username = extract_ig_username(url) + data = await fetch_ig_profile_info(username) + else: + # post / reel / tv + data = await fetch_ig_post_info(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 index 36fe290..da7bc29 100644 --- a/backend/routers/instagram.py +++ b/backend/routers/instagram.py @@ -1,5 +1,5 @@ """ -/api/instagram — post, reel, story, profile, thumbnail +/api/instagram — post, reel, profile, thumbnail """ import asyncio @@ -9,12 +9,12 @@ from fastapi.responses import FileResponse from pydantic import BaseModel from typing import Optional +from utils import clean_ig_url, extract_ig_username from services.instagram_service import ( download_post, download_reel, download_profile, download_ig_thumbnail, - fetch_ig_info, ) router = APIRouter() @@ -28,6 +28,7 @@ class PostRequest(BaseModel): class ProfileRequest(BaseModel): + # accepts either a username ("oiupoyt") or full URL ("instagram.com/oiupoyt/") username: str @@ -45,12 +46,11 @@ async def _delete_later(path: str, delay: int = 3600): @router.post("/post") async def post(req: PostRequest, bg: BackgroundTasks): + url = clean_ig_url(req.url.strip()) try: path, filename = await download_post( - url=req.url, - fmt=req.format, - start_time=req.start_time, - end_time=req.end_time, + url=url, fmt=req.format, + start_time=req.start_time, end_time=req.end_time, ) except Exception as e: raise HTTPException(500, detail=str(e)) @@ -61,12 +61,10 @@ async def post(req: PostRequest, bg: BackgroundTasks): @router.post("/reel") async def reel(req: PostRequest, bg: BackgroundTasks): - # reels use yt-dlp directly — more reliable than instaloader for reels + url = clean_ig_url(req.url.strip()) try: path, filename = await download_reel( - url=req.url, - start_time=req.start_time, - end_time=req.end_time, + url=url, start_time=req.start_time, end_time=req.end_time, ) except Exception as e: raise HTTPException(500, detail=str(e)) @@ -76,8 +74,25 @@ async def reel(req: PostRequest, bg: BackgroundTasks): @router.post("/profile") async def profile(req: ProfileRequest, bg: BackgroundTasks): + """ + Accepts a username ("oiupoyt") or full URL ("https://www.instagram.com/oiupoyt/"). + Downloads the profile picture as a JPEG. + """ + raw = req.username.strip() + # if it looks like a URL, extract username from it + if "instagram.com" in raw: + try: + username = extract_ig_username(raw) + except ValueError as e: + raise HTTPException(400, detail=str(e)) + else: + username = raw.lstrip("@") + + if not username: + raise HTTPException(400, detail="no username provided") + try: - path, filename = await download_profile(req.username) + path, filename = await download_profile(username) except Exception as e: raise HTTPException(500, detail=str(e)) bg.add_task(_delete_later, path) @@ -86,8 +101,9 @@ async def profile(req: ProfileRequest, bg: BackgroundTasks): @router.post("/thumbnail") async def thumbnail(req: SimpleRequest, bg: BackgroundTasks): + url = clean_ig_url(req.url.strip()) try: - path, filename = await download_ig_thumbnail(req.url) + path, filename = await download_ig_thumbnail(url) except Exception as e: raise HTTPException(500, detail=str(e)) bg.add_task(_delete_later, path) diff --git a/backend/routers/youtube.py b/backend/routers/youtube.py index 93be74e..f5f78e5 100644 --- a/backend/routers/youtube.py +++ b/backend/routers/youtube.py @@ -9,6 +9,7 @@ from fastapi.responses import FileResponse from pydantic import BaseModel from typing import Optional +from utils import clean_yt_url from services.youtube_service import ( download_video, download_audio, @@ -44,7 +45,7 @@ class SubtitleRequest(BaseModel): class ChannelRequest(BaseModel): - channel_url: str + channel_url: str # youtube.com/@handle or youtube.com/channel/UCxxxxxx async def _delete_later(path: str, delay: int = 3600): @@ -57,9 +58,10 @@ async def _delete_later(path: str, delay: int = 3600): @router.post("/video") async def video(req: VideoRequest, bg: BackgroundTasks): + url = clean_yt_url(req.url.strip()) try: path, filename = await download_video( - url=req.url, + url=url, resolution=req.resolution, fmt=req.format, start_time=req.start_time, @@ -73,12 +75,9 @@ async def video(req: VideoRequest, bg: BackgroundTasks): @router.post("/audio") async def audio(req: AudioRequest, bg: BackgroundTasks): + url = clean_yt_url(req.url.strip()) try: - path, filename = await download_audio( - url=req.url, - quality=req.quality, - fmt=req.format, - ) + path, filename = await download_audio(url=url, quality=req.quality, fmt=req.format) except Exception as e: raise HTTPException(500, detail=str(e)) bg.add_task(_delete_later, path) @@ -87,8 +86,9 @@ async def audio(req: AudioRequest, bg: BackgroundTasks): @router.post("/thumbnail") async def thumbnail(req: SimpleRequest, bg: BackgroundTasks): + url = clean_yt_url(req.url.strip()) try: - path, filename = await download_thumbnail(req.url) + path, filename = await download_thumbnail(url) except Exception as e: raise HTTPException(500, detail=str(e)) bg.add_task(_delete_later, path) @@ -97,8 +97,9 @@ async def thumbnail(req: SimpleRequest, bg: BackgroundTasks): @router.post("/subtitles") async def subtitles(req: SubtitleRequest, bg: BackgroundTasks): + url = clean_yt_url(req.url.strip()) try: - path, filename = await download_subtitles(req.url, req.lang) + path, filename = await download_subtitles(url, req.lang) except Exception as e: raise HTTPException(500, detail=str(e)) bg.add_task(_delete_later, path) @@ -108,18 +109,22 @@ async def subtitles(req: SubtitleRequest, bg: BackgroundTasks): @router.post("/channel-art") 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/..."}, ...]} + Downloads avatar (JPG) and banner (JPG) for a YouTube channel. + Returns JSON — frontend opens each asset URL to trigger download. + + { + "assets": [ + {"type": "avatar", "filename": "...", "url": "/files/..."}, + {"type": "banner", "filename": "...", "url": "/files/..."} + ] + } """ try: - assets = await download_channel_art(req.channel_url) + assets = await download_channel_art(req.channel_url.strip()) 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) + bg.add_task(_delete_later, f"downloads/{asset['filename']}") return {"assets": assets} diff --git a/backend/services/instagram_service.py b/backend/services/instagram_service.py index 307e1b3..4b1f804 100644 --- a/backend/services/instagram_service.py +++ b/backend/services/instagram_service.py @@ -2,17 +2,16 @@ Instagram service. Strategy: - - 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 + - Posts/images: instaloader (with session) → yt-dlp fallback + - Reels/videos: yt-dlp (instaloader 403s on reels) + - Profile pics: instaloader (uses username directly, no shortcode) + - Thumbnails: yt-dlp -Auth: run this once on the tablet to create ig_session: +Session setup (run once on the tablet): python3 -c " - import instaloader, os + import instaloader L = instaloader.Instaloader() - L.interactive_login('your_username') + L.interactive_login('your_ig_username') L.save_session_to_file('ig_session') print('saved') " @@ -81,17 +80,15 @@ def _get_loader() -> instaloader.Instaloader: 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") + print("[ig] session loaded from file") _loader = L return _loader except Exception as e: - print(f"[ig] session file load failed: {e}") + print(f"[ig] session file failed: {e}") - # fall back to env var credentials if IG_USERNAME and IG_PASSWORD: try: L.login(IG_USERNAME, IG_PASSWORD) @@ -101,32 +98,17 @@ def _get_loader() -> instaloader.Instaloader: except Exception as e: print(f"[ig] password login failed: {e}") - print("[ig] WARNING: no session — most IG content requires login now") + print("[ig] WARNING: unauthenticated — most IG content will fail") _loader = L return _loader -def _shortcode_from_url(url: str) -> str: - 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}\n" - "must be a post, reel, or tv URL — not a profile or explore page" - ) - - -def _username_from_url(url: str) -> str: - path = re.sub(r"https?://(www\.)?instagram\.com/?", "", url) - path = path.lstrip("@") - username = path.split("/")[0].split("?")[0].strip() - if not username: - raise ValueError("could not extract username from URL") - 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 +def _download_image_url(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 def _trim_video(src: Path, dest: Path, start: Optional[str], end: Optional[str]) -> Path: @@ -144,18 +126,20 @@ 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 +def _first_media(directory: Path) -> Optional[Path]: + return next( + (f for f in sorted(directory.iterdir()) + if f.suffix in (".mp4", ".jpg", ".jpeg", ".png")), + None, + ) -# ── info ────────────────────────────────────────────────────────────────────── +# ── post info (shortcode-based) ─────────────────────────────────────────────── -async def fetch_ig_info(url: str) -> dict: - def _fetch_ytdlp(): +async def fetch_ig_post_info(url: str) -> dict: + """Info for post / reel / tv URLs.""" + + def _via_ytdlp(): opts = {**_BASE_YDL, "skip_download": True} with yt_dlp.YoutubeDL(opts) as ydl: info = ydl.extract_info(url, download=False) @@ -169,9 +153,10 @@ async def fetch_ig_info(url: str) -> dict: "likes": info.get("like_count"), } - def _fetch_instaloader(): - shortcode = _shortcode_from_url(url) - L = _get_loader() + def _via_instaloader(): + from utils import extract_ig_shortcode + shortcode = extract_ig_shortcode(url) + L = _get_loader() post = instaloader.Post.from_shortcode(L.context, shortcode) return { "type": "video" if post.is_video else "image", @@ -184,13 +169,36 @@ async def fetch_ig_info(url: str) -> dict: } try: - return await _run(_fetch_ytdlp) + return await _run(_via_ytdlp) except Exception: pass - return await _run(_fetch_instaloader) + return await _run(_via_instaloader) -# ── post ────────────────────────────────────────────────────────────────────── +# ── profile info (username-based) ───────────────────────────────────────────── + +async def fetch_ig_profile_info(username: str) -> dict: + """Info for profile pages — uses username directly, no shortcode.""" + + def _fetch(): + L = _get_loader() + profile = instaloader.Profile.from_username(L.context, username) + try: + pic_url = profile.profile_pic_url_no_iphone + except Exception: + pic_url = profile.profile_pic_url + return { + "type": "profile", + "title": f"@{profile.username}", + "thumbnail": pic_url, + "owner": profile.username, + "likes": profile.followers, # repurpose likes field for follower count + } + + return await _run(_fetch) + + +# ── post download ───────────────────────────────────────────────────────────── async def download_post( url: str, @@ -200,7 +208,7 @@ async def download_post( ) -> Tuple[str, str]: uid = _uid() - def _try_ytdlp(): + def _via_ytdlp(): out_tmpl = str(DOWNLOAD_DIR / f"ig_post_{uid}.%(ext)s") opts = { **_BASE_YDL, @@ -211,58 +219,48 @@ async def download_post( with yt_dlp.YoutubeDL(opts) as ydl: info = ydl.extract_info(url) owner = _safe(info.get("uploader") or "ig") - 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(): + def _via_instaloader(): + from utils import extract_ig_shortcode tmp_dir = DOWNLOAD_DIR / f"ig_il_{uid}" tmp_dir.mkdir(exist_ok=True) try: - shortcode = _shortcode_from_url(url) - L = _get_loader() + shortcode = extract_ig_shortcode(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 sorted(tmp_dir.iterdir()) - if f.suffix in (".mp4", ".jpg", ".jpeg", ".png")), - None, - ) + media = _first_media(tmp_dir) 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: - return await _run(_try_ytdlp) + return await _run(_via_ytdlp) except Exception: pass - return await _run(_try_instaloader) + return await _run(_via_instaloader) -# ── reel ────────────────────────────────────────────────────────────────────── +# ── reel download (yt-dlp only) ─────────────────────────────────────────────── async def download_reel( url: str, @@ -283,46 +281,38 @@ async def download_reel( with yt_dlp.YoutubeDL(opts) as ydl: info = ydl.extract_info(url) owner = _safe(info.get("uploader") or "ig") - 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]: - 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") +# ── profile picture download ────────────────────────────────────────────────── +async def download_profile(username: str) -> Tuple[str, str]: + """ + Downloads the profile picture for a given username. + username should already be clean (no @ prefix, no URL). + """ uid = _uid() def _dl(): 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) + _download_image_url(pic_url, path) return str(path), f"{_safe(username)}_profile.jpg" return await _run(_dl) @@ -342,7 +332,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" - _download_image(thumb_url, path) + _download_image_url(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 b5bf909..65596f9 100644 --- a/backend/services/youtube_service.py +++ b/backend/services/youtube_service.py @@ -50,7 +50,7 @@ def _parse_time(t: str) -> float: return parts[0] -def _download_image(url: str, dest: Path) -> Path: +def _download_image_url(url: str, dest: Path) -> Path: import httpx resp = httpx.get(url, follow_redirects=True, timeout=20) resp.raise_for_status() @@ -58,6 +58,11 @@ def _download_image(url: str, dest: Path) -> Path: return dest +def _first_file(pattern: str) -> Optional[Path]: + candidates = list(DOWNLOAD_DIR.glob(pattern)) + return candidates[0] if candidates else None + + # ── info ────────────────────────────────────────────────────────────────────── async def fetch_yt_info(url: str) -> dict: @@ -132,10 +137,9 @@ async def download_video( with yt_dlp.YoutubeDL(opts) as ydl: info = ydl.extract_info(url) title = _safe(info.get("title", "video")) - candidates = list(DOWNLOAD_DIR.glob(f"yt_video_{uid}.*")) - if not candidates: + final = _first_file(f"yt_video_{uid}.*") + if not final: raise FileNotFoundError("output file not found after download") - final = candidates[0] return str(final), f"{title}{final.suffix}" return await _run(_dl) @@ -150,20 +154,14 @@ async def download_audio( ) -> 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") + codec = {"mp3": "mp3", "opus": "opus", "flac": "flac", "m4a": "m4a"}.get(fmt, "mp3") opts = { **_BASE_OPTS, "format": "bestaudio/best", "outtmpl": out_tmpl, "postprocessors": [ - { - "key": "FFmpegExtractAudio", - "preferredcodec": codec, - "preferredquality": quality, - }, + {"key": "FFmpegExtractAudio", "preferredcodec": codec, "preferredquality": quality}, {"key": "FFmpegMetadata", "add_metadata": True}, ], } @@ -172,10 +170,9 @@ async def download_audio( with yt_dlp.YoutubeDL(opts) as ydl: info = ydl.extract_info(url) title = _safe(info.get("title", "audio")) - candidates = list(DOWNLOAD_DIR.glob(f"yt_audio_{uid}.*")) - if not candidates: + final = _first_file(f"yt_audio_{uid}.*") + if not final: raise FileNotFoundError("audio file not found after download") - final = candidates[0] return str(final), f"{title}{final.suffix}" return await _run(_dl) @@ -231,12 +228,10 @@ async def download_subtitles(url: str, lang: str = "en") -> Tuple[str, str]: candidate = DOWNLOAD_DIR / f"yt_subs_{uid}.{lang}.{ext}" if candidate.exists(): return str(candidate), f"{title}.{lang}.{ext}" - 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" - ) + final = _first_file(f"yt_subs_{uid}.*") + if final: + return str(final), final.name + raise FileNotFoundError(f"no subtitles found for lang={lang}") return await _run(_dl) @@ -245,15 +240,22 @@ async def download_subtitles(url: str, lang: str = "en") -> Tuple[str, str]: 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. + Given a channel URL (youtube.com/@handle or youtube.com/channel/UCxxxxx), + downloads the avatar and banner as separate JPEG files. + + Returns: + [ + {"type": "avatar", "filename": "channelname_avatar.jpg", "url": "/files/..."}, + {"type": "banner", "filename": "channelname_banner.jpg", "url": "/files/..."}, + ] """ def _dl(): - uid = _uid() - results = [] + import httpx + uid = _uid() + + # fetch channel metadata — extract_flat avoids fetching all videos opts = { **_BASE_OPTS, "skip_download": True, @@ -262,71 +264,67 @@ async def download_channel_art(channel_url: str) -> List[dict]: with yt_dlp.YoutubeDL(opts) as ydl: info = ydl.extract_info(channel_url, download=False) - thumbnails = info.get("thumbnails") or [] + channel_name = _safe(info.get("channel") or info.get("uploader") or "channel") + thumbnails = info.get("thumbnails") or [] + results = [] # ── avatar ──────────────────────────────────────────────────────────── - # avatars are square-ish; YouTube avatar thumbnails often come from - # ggpht.com or have "photo" in the URL + # YouTube channel avatars come from ggpht.com and are square 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"] + + # first try thumbnails tagged as square / avatar-like + square = [t for t in thumbnails if _is_avatar(t)] + if square: + # highest resolution square thumbnail + avatar_url = max(square, key=lambda t: (t.get("width") or 0) * (t.get("height") or 0))["url"] + + # fallback: any thumbnail from ggpht (Google's image CDN for avatars) + if not avatar_url: + ggpht = [t for t in thumbnails if "ggpht" in (t.get("url") or "")] + if ggpht: + avatar_url = ggpht[-1]["url"] + + # last resort: channel thumbnail field + if not avatar_url: + avatar_url = info.get("thumbnail") if avatar_url: - path = DOWNLOAD_DIR / f"yt_ch_{uid}_avatar.jpg" + dest = DOWNLOAD_DIR / f"{channel_name}_avatar_{uid}.jpg" try: - _download_image(avatar_url, path) + _download_image_url(avatar_url, dest) results.append({ "type": "avatar", - "filename": path.name, - "url": f"/files/{path.name}", + "filename": dest.name, + "url": f"/files/{dest.name}", }) except Exception as e: - print(f"[yt] avatar download failed: {e}") + print(f"[yt channel] 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") - ) + 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"] + # look for very wide thumbnails (banner aspect ratio) + wide = [t for t in thumbnails if _is_banner(t)] + if wide: + banner_url = max(wide, key=lambda t: t.get("width") or 0)["url"] if banner_url: - path = DOWNLOAD_DIR / f"yt_ch_{uid}_banner.jpg" + dest = DOWNLOAD_DIR / f"{channel_name}_banner_{uid}.jpg" try: - _download_image(banner_url, path) + _download_image_url(banner_url, dest) results.append({ "type": "banner", - "filename": path.name, - "url": f"/files/{path.name}", + "filename": dest.name, + "url": f"/files/{dest.name}", }) except Exception as e: - print(f"[yt] banner download failed: {e}") + print(f"[yt channel] banner download failed: {e}") if not results: raise ValueError( - "no channel art found — pass a full channel URL like " - "youtube.com/@channelname or youtube.com/channel/UCxxxxxx" + "no channel art found — make sure you're using a channel URL " + "like youtube.com/@channelname or youtube.com/channel/UCxxxxxx" ) return results @@ -335,20 +333,18 @@ async def download_channel_art(channel_url: str) -> List[dict]: 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 + w = t.get("width") or 0 + h = t.get("height") or 1 + url = t.get("url") or "" + if w and h and 0.8 < (w / h) < 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", "") + w = t.get("width") or 0 + h = t.get("height") or 1 + url = t.get("url") or "" if w and h and (w / h) > 2.5: return True return "banner" in url.lower() diff --git a/backend/utils.py b/backend/utils.py new file mode 100644 index 0000000..11290b1 --- /dev/null +++ b/backend/utils.py @@ -0,0 +1,62 @@ +""" +Shared URL utilities. +""" + +import re + + +def clean_yt_url(url: str) -> str: + """ + Strip playlist/radio params from YouTube URLs so yt-dlp + treats them as single videos. + + youtube.com/watch?v=ABC&list=RD...&start_radio=1 + → youtube.com/watch?v=ABC + """ + # if it has a video id, keep only that + m = re.search(r"[?&]v=([A-Za-z0-9_-]{11})", url) + if m: + return f"https://www.youtube.com/watch?v={m.group(1)}" + # youtu.be short links — keep as-is (no query needed) + m2 = re.match(r"https?://youtu\.be/([A-Za-z0-9_-]{11})", url) + if m2: + return f"https://youtu.be/{m2.group(1)}" + # channel or playlist URL — return as-is + return url + + +def clean_ig_url(url: str) -> str: + """Strip query params and trailing slashes from Instagram URLs.""" + url = url.split("?")[0].rstrip("/") + return url + + +def is_ig_profile_url(url: str) -> bool: + """ + Returns True if the URL points to a profile page rather than a post/reel. + Profile: instagram.com/username + Post: instagram.com/p/ABC instagram.com/reel/ABC + """ + path = re.sub(r"https?://(www\.)?instagram\.com/?", "", url).lstrip("/") + first_segment = path.split("/")[0].split("?")[0].lstrip("@") + return first_segment not in ("p", "reel", "reels", "tv", "explore", "stories", "accounts", "") + + +def extract_ig_username(url: str) -> str: + """Pull username from instagram.com/username or instagram.com/@username.""" + path = re.sub(r"https?://(www\.)?instagram\.com/?", "", url).lstrip("/") + username = path.split("/")[0].split("?")[0].lstrip("@").strip() + if not username: + raise ValueError("could not extract username from URL") + return username + + +def extract_ig_shortcode(url: str) -> str: + """Pull shortcode from post/reel/tv URLs.""" + 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} — " + "expected a post, reel, or tv URL" + )