fixed instagram avater download + youtube avater download and yt radio mix issues most likely
This commit is contained in:
parent
49c8bac7a7
commit
384e6a903c
6 changed files with 277 additions and 194 deletions
|
|
@ -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}
|
||||
|
|
|
|||
|
|
@ -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:
|
||||
path, filename = await download_profile(req.username)
|
||||
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(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)
|
||||
|
|
|
|||
|
|
@ -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}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
- 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,8 +153,9 @@ async def fetch_ig_info(url: str) -> dict:
|
|||
"likes": info.get("like_count"),
|
||||
}
|
||||
|
||||
def _fetch_instaloader():
|
||||
shortcode = _shortcode_from_url(url)
|
||||
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 {
|
||||
|
|
@ -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)
|
||||
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)
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
||||
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
|
||||
|
|
@ -337,10 +335,8 @@ 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:
|
||||
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"))
|
||||
|
||||
|
|
@ -348,7 +344,7 @@ def _is_avatar(t: dict) -> bool:
|
|||
def _is_banner(t: dict) -> bool:
|
||||
w = t.get("width") or 0
|
||||
h = t.get("height") or 1
|
||||
url = t.get("url", "")
|
||||
url = t.get("url") or ""
|
||||
if w and h and (w / h) > 2.5:
|
||||
return True
|
||||
return "banner" in url.lower()
|
||||
|
|
|
|||
62
backend/utils.py
Normal file
62
backend/utils.py
Normal file
|
|
@ -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"
|
||||
)
|
||||
Loading…
Add table
Reference in a new issue