fixed instagram apprantly with a login and yt avater + banner downloads
This commit is contained in:
parent
5f0bb0e47e
commit
4cc354d96b
4 changed files with 236 additions and 143 deletions
|
|
@ -1,7 +1,2 @@
|
||||||
# just-download
|
# 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)
|
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
|
|
||||||
|
|
|
||||||
|
|
@ -106,9 +106,20 @@ async def subtitles(req: SubtitleRequest, bg: BackgroundTasks):
|
||||||
|
|
||||||
|
|
||||||
@router.post("/channel-art")
|
@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:
|
try:
|
||||||
assets = await download_channel_art(req.channel_url)
|
assets = await download_channel_art(req.channel_url)
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
raise HTTPException(500, detail=str(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}
|
return {"assets": assets}
|
||||||
|
|
|
||||||
|
|
@ -2,14 +2,20 @@
|
||||||
Instagram service.
|
Instagram service.
|
||||||
|
|
||||||
Strategy:
|
Strategy:
|
||||||
- Posts (images): instaloader (best quality originals)
|
- Posts (images): instaloader with saved session
|
||||||
- Posts (videos): yt-dlp first, instaloader fallback
|
- Posts (videos): yt-dlp first, instaloader fallback
|
||||||
- Reels: yt-dlp (instaloader 403s on reels consistently)
|
- Reels: yt-dlp (instaloader 403s on reels)
|
||||||
- Profile pictures: instaloader (direct profile_pic_url)
|
- Profile pics: instaloader with saved session
|
||||||
- Thumbnails: yt-dlp (fast, no auth needed)
|
- Thumbnails: yt-dlp
|
||||||
|
|
||||||
Instagram blocks unauthenticated API calls aggressively.
|
Auth: run this once on the tablet to create ig_session:
|
||||||
For private content set IG_USERNAME / IG_PASSWORD env vars.
|
python3 -c "
|
||||||
|
import instaloader, os
|
||||||
|
L = instaloader.Instaloader()
|
||||||
|
L.interactive_login('your_username')
|
||||||
|
L.save_session_to_file('ig_session')
|
||||||
|
print('saved')
|
||||||
|
"
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
|
|
@ -26,15 +32,16 @@ import instaloader
|
||||||
DOWNLOAD_DIR = Path("downloads")
|
DOWNLOAD_DIR = Path("downloads")
|
||||||
DOWNLOAD_DIR.mkdir(exist_ok=True)
|
DOWNLOAD_DIR.mkdir(exist_ok=True)
|
||||||
|
|
||||||
IG_USERNAME = os.getenv("IG_USERNAME", "")
|
SESSION_FILE = Path("ig_session")
|
||||||
IG_PASSWORD = os.getenv("IG_PASSWORD", "")
|
IG_USERNAME = os.getenv("IG_USERNAME", "")
|
||||||
|
IG_PASSWORD = os.getenv("IG_PASSWORD", "")
|
||||||
|
|
||||||
_loader: Optional[instaloader.Instaloader] = None
|
_loader: Optional[instaloader.Instaloader] = None
|
||||||
|
|
||||||
_BASE_YDL = {
|
_BASE_YDL = {
|
||||||
"quiet": True,
|
"quiet": True,
|
||||||
"no_warnings": True,
|
"no_warnings": True,
|
||||||
"noprogress": True,
|
"noprogress": True,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -62,57 +69,63 @@ def _fmt_s(s) -> str:
|
||||||
|
|
||||||
def _get_loader() -> instaloader.Instaloader:
|
def _get_loader() -> instaloader.Instaloader:
|
||||||
global _loader
|
global _loader
|
||||||
if _loader is None:
|
if _loader is not None:
|
||||||
_loader = instaloader.Instaloader(
|
return _loader
|
||||||
download_pictures=True,
|
|
||||||
download_videos=True,
|
L = instaloader.Instaloader(
|
||||||
download_video_thumbnails=False,
|
download_pictures=True,
|
||||||
save_metadata=False,
|
download_videos=True,
|
||||||
post_metadata_txt_pattern="",
|
download_video_thumbnails=False,
|
||||||
quiet=True,
|
save_metadata=False,
|
||||||
)
|
post_metadata_txt_pattern="",
|
||||||
if IG_USERNAME and IG_PASSWORD:
|
quiet=True,
|
||||||
try:
|
)
|
||||||
_loader.login(IG_USERNAME, IG_PASSWORD)
|
|
||||||
except Exception:
|
# prefer saved session file — avoids triggering suspicious login alerts
|
||||||
pass
|
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
|
return _loader
|
||||||
|
|
||||||
|
|
||||||
def _shortcode_from_url(url: str) -> str:
|
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)
|
m = re.search(r"/(?:p|reel|reels|tv)/([A-Za-z0-9_-]+)", url)
|
||||||
if m:
|
if m:
|
||||||
return m.group(1)
|
return m.group(1)
|
||||||
raise ValueError(
|
raise ValueError(
|
||||||
f"could not extract shortcode from URL: {url}\n"
|
f"could not extract shortcode from: {url}\n"
|
||||||
"make sure it's a post, reel, or tv URL — not a profile or explore page"
|
"must be a post, reel, or tv URL — not a profile or explore page"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
def _username_from_url(url: str) -> str:
|
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)
|
path = re.sub(r"https?://(www\.)?instagram\.com/?", "", url)
|
||||||
# strip @ if present
|
|
||||||
path = path.lstrip("@")
|
path = path.lstrip("@")
|
||||||
# take first path segment
|
|
||||||
username = path.split("/")[0].split("?")[0].strip()
|
username = path.split("/")[0].split("?")[0].strip()
|
||||||
if not username:
|
if not username:
|
||||||
raise ValueError("could not extract username from URL")
|
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", "accounts"):
|
||||||
if username in ("p", "reel", "reels", "tv", "explore", "stories"):
|
raise ValueError(
|
||||||
raise ValueError(f"that looks like a post URL, not a profile — got segment: {username}")
|
f"that looks like a post/page URL, not a profile — got: {username}"
|
||||||
|
)
|
||||||
return username
|
return username
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -131,19 +144,23 @@ def _trim_video(src: Path, dest: Path, start: Optional[str], end: Optional[str])
|
||||||
return dest
|
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 ──────────────────────────────────────────────────────────────────────
|
# ── info ──────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
async def fetch_ig_info(url: str) -> dict:
|
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():
|
def _fetch_ytdlp():
|
||||||
opts = {**_BASE_YDL, "skip_download": True}
|
opts = {**_BASE_YDL, "skip_download": True}
|
||||||
with yt_dlp.YoutubeDL(opts) as ydl:
|
with yt_dlp.YoutubeDL(opts) as ydl:
|
||||||
info = ydl.extract_info(url, download=False)
|
info = ydl.extract_info(url, download=False)
|
||||||
return {
|
return {
|
||||||
"type": "video" if info.get("ext") != "jpg" else "image",
|
"type": "video",
|
||||||
"title": info.get("title") or info.get("uploader") or "Instagram post",
|
"title": info.get("title") or info.get("uploader") or "Instagram post",
|
||||||
"thumbnail": info.get("thumbnail"),
|
"thumbnail": info.get("thumbnail"),
|
||||||
"duration": info.get("duration"),
|
"duration": info.get("duration"),
|
||||||
|
|
@ -173,7 +190,7 @@ async def fetch_ig_info(url: str) -> dict:
|
||||||
return await _run(_fetch_instaloader)
|
return await _run(_fetch_instaloader)
|
||||||
|
|
||||||
|
|
||||||
# ── post (image or video) ─────────────────────────────────────────────────────
|
# ── post ──────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
async def download_post(
|
async def download_post(
|
||||||
url: str,
|
url: str,
|
||||||
|
|
@ -187,25 +204,25 @@ async def download_post(
|
||||||
out_tmpl = str(DOWNLOAD_DIR / f"ig_post_{uid}.%(ext)s")
|
out_tmpl = str(DOWNLOAD_DIR / f"ig_post_{uid}.%(ext)s")
|
||||||
opts = {
|
opts = {
|
||||||
**_BASE_YDL,
|
**_BASE_YDL,
|
||||||
"format": "bestvideo+bestaudio/best",
|
"format": "bestvideo+bestaudio/best",
|
||||||
"outtmpl": out_tmpl,
|
"outtmpl": out_tmpl,
|
||||||
"merge_output_format": "mp4",
|
"merge_output_format": "mp4",
|
||||||
}
|
}
|
||||||
with yt_dlp.YoutubeDL(opts) as ydl:
|
with yt_dlp.YoutubeDL(opts) as ydl:
|
||||||
info = ydl.extract_info(url)
|
info = ydl.extract_info(url)
|
||||||
owner = _safe(info.get("uploader") or "ig")
|
owner = _safe(info.get("uploader") or "ig")
|
||||||
ext = "mp4"
|
|
||||||
final = DOWNLOAD_DIR / f"ig_post_{uid}.{ext}"
|
candidates = list(DOWNLOAD_DIR.glob(f"ig_post_{uid}.*"))
|
||||||
if not final.exists():
|
if not candidates:
|
||||||
candidates = list(DOWNLOAD_DIR.glob(f"ig_post_{uid}.*"))
|
raise FileNotFoundError("yt-dlp produced no file")
|
||||||
if not candidates:
|
final = candidates[0]
|
||||||
raise FileNotFoundError("yt-dlp produced no file")
|
|
||||||
final = candidates[0]
|
|
||||||
if (start_time or end_time) and final.suffix == ".mp4":
|
if (start_time or end_time) and final.suffix == ".mp4":
|
||||||
trimmed = DOWNLOAD_DIR / f"ig_post_{uid}_trim.mp4"
|
trimmed = DOWNLOAD_DIR / f"ig_post_{uid}_trim.mp4"
|
||||||
_trim_video(final, trimmed, start_time, end_time)
|
_trim_video(final, trimmed, start_time, end_time)
|
||||||
final.unlink(missing_ok=True)
|
final.unlink(missing_ok=True)
|
||||||
final = trimmed
|
final = trimmed
|
||||||
|
|
||||||
return str(final), f"{owner}_{uid}{final.suffix}"
|
return str(final), f"{owner}_{uid}{final.suffix}"
|
||||||
|
|
||||||
def _try_instaloader():
|
def _try_instaloader():
|
||||||
|
|
@ -219,7 +236,8 @@ async def download_post(
|
||||||
L.download_post(post, target=str(tmp_dir))
|
L.download_post(post, target=str(tmp_dir))
|
||||||
|
|
||||||
media = next(
|
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,
|
None,
|
||||||
)
|
)
|
||||||
if not media:
|
if not media:
|
||||||
|
|
@ -237,7 +255,6 @@ async def download_post(
|
||||||
finally:
|
finally:
|
||||||
shutil.rmtree(tmp_dir, ignore_errors=True)
|
shutil.rmtree(tmp_dir, ignore_errors=True)
|
||||||
|
|
||||||
# try yt-dlp first, fall back to instaloader
|
|
||||||
try:
|
try:
|
||||||
return await _run(_try_ytdlp)
|
return await _run(_try_ytdlp)
|
||||||
except Exception:
|
except Exception:
|
||||||
|
|
@ -245,7 +262,7 @@ async def download_post(
|
||||||
return await _run(_try_instaloader)
|
return await _run(_try_instaloader)
|
||||||
|
|
||||||
|
|
||||||
# ── reel (yt-dlp only — instaloader 403s on reels) ───────────────────────────
|
# ── reel ──────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
async def download_reel(
|
async def download_reel(
|
||||||
url: str,
|
url: str,
|
||||||
|
|
@ -267,12 +284,10 @@ async def download_reel(
|
||||||
info = ydl.extract_info(url)
|
info = ydl.extract_info(url)
|
||||||
owner = _safe(info.get("uploader") or "ig")
|
owner = _safe(info.get("uploader") or "ig")
|
||||||
|
|
||||||
final = DOWNLOAD_DIR / f"ig_reel_{uid}.mp4"
|
candidates = list(DOWNLOAD_DIR.glob(f"ig_reel_{uid}.*"))
|
||||||
if not final.exists():
|
if not candidates:
|
||||||
candidates = list(DOWNLOAD_DIR.glob(f"ig_reel_{uid}.*"))
|
raise FileNotFoundError("yt-dlp produced no reel file")
|
||||||
if not candidates:
|
final = candidates[0]
|
||||||
raise FileNotFoundError("yt-dlp produced no reel file")
|
|
||||||
final = candidates[0]
|
|
||||||
|
|
||||||
if start_time or end_time:
|
if start_time or end_time:
|
||||||
trimmed = DOWNLOAD_DIR / f"ig_reel_{uid}_trim.mp4"
|
trimmed = DOWNLOAD_DIR / f"ig_reel_{uid}_trim.mp4"
|
||||||
|
|
@ -288,7 +303,6 @@ async def download_reel(
|
||||||
# ── profile picture ───────────────────────────────────────────────────────────
|
# ── profile picture ───────────────────────────────────────────────────────────
|
||||||
|
|
||||||
async def download_profile(username_or_url: str) -> Tuple[str, str]:
|
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:
|
if "instagram.com" in username_or_url:
|
||||||
username = _username_from_url(username_or_url)
|
username = _username_from_url(username_or_url)
|
||||||
else:
|
else:
|
||||||
|
|
@ -300,13 +314,15 @@ async def download_profile(username_or_url: str) -> Tuple[str, str]:
|
||||||
uid = _uid()
|
uid = _uid()
|
||||||
|
|
||||||
def _dl():
|
def _dl():
|
||||||
import httpx
|
L = _get_loader()
|
||||||
L = _get_loader()
|
profile = instaloader.Profile.from_username(L.context, username)
|
||||||
profile = instaloader.Profile.from_username(L.context, username)
|
# profile_pic_url_no_iphone gives higher resolution
|
||||||
pic_url = profile.profile_pic_url
|
try:
|
||||||
path = DOWNLOAD_DIR / f"ig_pfp_{_safe(username)}_{uid}.jpg"
|
pic_url = profile.profile_pic_url_no_iphone
|
||||||
data = httpx.get(pic_url, follow_redirects=True, timeout=15).content
|
except Exception:
|
||||||
path.write_bytes(data)
|
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 str(path), f"{_safe(username)}_profile.jpg"
|
||||||
|
|
||||||
return await _run(_dl)
|
return await _run(_dl)
|
||||||
|
|
@ -318,8 +334,6 @@ async def download_ig_thumbnail(url: str) -> Tuple[str, str]:
|
||||||
uid = _uid()
|
uid = _uid()
|
||||||
|
|
||||||
def _dl():
|
def _dl():
|
||||||
import httpx
|
|
||||||
# try yt-dlp for thumbnail URL
|
|
||||||
opts = {**_BASE_YDL, "skip_download": True}
|
opts = {**_BASE_YDL, "skip_download": True}
|
||||||
with yt_dlp.YoutubeDL(opts) as ydl:
|
with yt_dlp.YoutubeDL(opts) as ydl:
|
||||||
info = ydl.extract_info(url, download=False)
|
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")
|
raise ValueError("no thumbnail found for this post")
|
||||||
owner = _safe(info.get("uploader") or "ig")
|
owner = _safe(info.get("uploader") or "ig")
|
||||||
path = DOWNLOAD_DIR / f"ig_thumb_{uid}.jpg"
|
path = DOWNLOAD_DIR / f"ig_thumb_{uid}.jpg"
|
||||||
data = httpx.get(thumb_url, follow_redirects=True, timeout=15).content
|
_download_image(thumb_url, path)
|
||||||
path.write_bytes(data)
|
|
||||||
return str(path), f"{owner}_thumbnail_{uid}.jpg"
|
return str(path), f"{owner}_thumbnail_{uid}.jpg"
|
||||||
|
|
||||||
return await _run(_dl)
|
return await _run(_dl)
|
||||||
|
|
|
||||||
|
|
@ -3,21 +3,20 @@ YouTube service — all yt-dlp operations.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import uuid
|
|
||||||
import re
|
import re
|
||||||
|
import uuid
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Optional, Tuple
|
from typing import Optional, Tuple, List
|
||||||
|
|
||||||
import yt_dlp
|
import yt_dlp
|
||||||
|
|
||||||
DOWNLOAD_DIR = Path("downloads")
|
DOWNLOAD_DIR = Path("downloads")
|
||||||
DOWNLOAD_DIR.mkdir(exist_ok=True)
|
DOWNLOAD_DIR.mkdir(exist_ok=True)
|
||||||
|
|
||||||
# shared yt-dlp options applied to every call
|
|
||||||
_BASE_OPTS = {
|
_BASE_OPTS = {
|
||||||
"quiet": True,
|
"quiet": True,
|
||||||
"no_warnings": 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]
|
return re.sub(r'[\\/*?:"<>|]', "_", name or "file")[:80]
|
||||||
|
|
||||||
|
|
||||||
def _fmt_duration(seconds: int) -> str:
|
def _fmt_duration(seconds) -> str:
|
||||||
if not seconds:
|
if not seconds:
|
||||||
return "0:00"
|
return "0:00"
|
||||||
h, rem = divmod(int(seconds), 3600)
|
h, rem = divmod(int(seconds), 3600)
|
||||||
|
|
@ -51,6 +50,14 @@ def _parse_time(t: str) -> float:
|
||||||
return parts[0]
|
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 ──────────────────────────────────────────────────────────────────────
|
# ── info ──────────────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
async def fetch_yt_info(url: str) -> dict:
|
async def fetch_yt_info(url: str) -> dict:
|
||||||
|
|
@ -92,10 +99,9 @@ async def download_video(
|
||||||
start_time: Optional[str] = None,
|
start_time: Optional[str] = None,
|
||||||
end_time: Optional[str] = None,
|
end_time: Optional[str] = None,
|
||||||
) -> Tuple[str, str]:
|
) -> Tuple[str, str]:
|
||||||
uid = _uid()
|
uid = _uid()
|
||||||
out_tmpl = str(DOWNLOAD_DIR / f"yt_video_{uid}.%(ext)s")
|
out_tmpl = str(DOWNLOAD_DIR / f"yt_video_{uid}.%(ext)s")
|
||||||
|
|
||||||
# build format selector
|
|
||||||
if resolution == "best":
|
if resolution == "best":
|
||||||
fmt_sel = f"bestvideo[ext={fmt}]+bestaudio/bestvideo+bestaudio/best"
|
fmt_sel = f"bestvideo[ext={fmt}]+bestaudio/bestvideo+bestaudio/best"
|
||||||
else:
|
else:
|
||||||
|
|
@ -106,9 +112,9 @@ async def download_video(
|
||||||
|
|
||||||
opts = {
|
opts = {
|
||||||
**_BASE_OPTS,
|
**_BASE_OPTS,
|
||||||
"format": fmt_sel,
|
"format": fmt_sel,
|
||||||
"outtmpl": out_tmpl,
|
"outtmpl": out_tmpl,
|
||||||
"merge_output_format": fmt,
|
"merge_output_format": fmt,
|
||||||
}
|
}
|
||||||
|
|
||||||
if start_time or end_time:
|
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)
|
end = end_s if end_s is not None else info_dict.get("duration", 86400)
|
||||||
return [{"start_time": start_s, "end_time": end}]
|
return [{"start_time": start_s, "end_time": end}]
|
||||||
|
|
||||||
opts["download_ranges"] = _range_fn
|
opts["download_ranges"] = _range_fn
|
||||||
opts["force_keyframes_at_cuts"] = True
|
opts["force_keyframes_at_cuts"] = True
|
||||||
|
|
||||||
def _dl():
|
def _dl():
|
||||||
with yt_dlp.YoutubeDL(opts) as ydl:
|
with yt_dlp.YoutubeDL(opts) as ydl:
|
||||||
info = ydl.extract_info(url)
|
info = ydl.extract_info(url)
|
||||||
title = _safe(info.get("title", "video"))
|
title = _safe(info.get("title", "video"))
|
||||||
ext = info.get("ext", fmt)
|
candidates = list(DOWNLOAD_DIR.glob(f"yt_video_{uid}.*"))
|
||||||
# find the actual output file
|
if not candidates:
|
||||||
final = DOWNLOAD_DIR / f"yt_video_{uid}.{ext}"
|
raise FileNotFoundError("output file not found after download")
|
||||||
if not final.exists():
|
final = candidates[0]
|
||||||
# yt-dlp may have merged to a different ext
|
return str(final), f"{title}{final.suffix}"
|
||||||
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)
|
return await _run(_dl)
|
||||||
|
|
||||||
|
|
@ -171,13 +172,11 @@ async def download_audio(
|
||||||
with yt_dlp.YoutubeDL(opts) as ydl:
|
with yt_dlp.YoutubeDL(opts) as ydl:
|
||||||
info = ydl.extract_info(url)
|
info = ydl.extract_info(url)
|
||||||
title = _safe(info.get("title", "audio"))
|
title = _safe(info.get("title", "audio"))
|
||||||
final = DOWNLOAD_DIR / f"yt_audio_{uid}.{fmt}"
|
candidates = list(DOWNLOAD_DIR.glob(f"yt_audio_{uid}.*"))
|
||||||
if not final.exists():
|
if not candidates:
|
||||||
candidates = list(DOWNLOAD_DIR.glob(f"yt_audio_{uid}.*"))
|
raise FileNotFoundError("audio file not found after download")
|
||||||
if not candidates:
|
final = candidates[0]
|
||||||
raise FileNotFoundError("audio file not found after download")
|
return str(final), f"{title}{final.suffix}"
|
||||||
final = candidates[0]
|
|
||||||
return str(final), f"{title}.{final.suffix.lstrip('.')}"
|
|
||||||
|
|
||||||
return await _run(_dl)
|
return await _run(_dl)
|
||||||
|
|
||||||
|
|
@ -216,12 +215,12 @@ async def download_subtitles(url: str, lang: str = "en") -> Tuple[str, str]:
|
||||||
|
|
||||||
opts = {
|
opts = {
|
||||||
**_BASE_OPTS,
|
**_BASE_OPTS,
|
||||||
"skip_download": True,
|
"skip_download": True,
|
||||||
"writesubtitles": True,
|
"writesubtitles": True,
|
||||||
"writeautomaticsub": True,
|
"writeautomaticsub": True,
|
||||||
"subtitleslangs": [lang],
|
"subtitleslangs": [lang],
|
||||||
"subtitlesformat": "srt",
|
"subtitlesformat": "srt",
|
||||||
"outtmpl": out_tmpl,
|
"outtmpl": out_tmpl,
|
||||||
}
|
}
|
||||||
|
|
||||||
def _dl():
|
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)
|
info = ydl.extract_info(url, download=True)
|
||||||
title = _safe(info.get("title", "subtitles"))
|
title = _safe(info.get("title", "subtitles"))
|
||||||
for ext in ("srt", "vtt", "ass"):
|
for ext in ("srt", "vtt", "ass"):
|
||||||
# yt-dlp names it: yt_subs_{uid}.{lang}.{ext}
|
|
||||||
candidate = DOWNLOAD_DIR / f"yt_subs_{uid}.{lang}.{ext}"
|
candidate = DOWNLOAD_DIR / f"yt_subs_{uid}.{lang}.{ext}"
|
||||||
if candidate.exists():
|
if candidate.exists():
|
||||||
return str(candidate), f"{title}.{lang}.{ext}"
|
return str(candidate), f"{title}.{lang}.{ext}"
|
||||||
# fallback: any subtitle file matching uid
|
|
||||||
candidates = list(DOWNLOAD_DIR.glob(f"yt_subs_{uid}.*"))
|
candidates = list(DOWNLOAD_DIR.glob(f"yt_subs_{uid}.*"))
|
||||||
if candidates:
|
if candidates:
|
||||||
return str(candidates[0]), candidates[0].name
|
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)
|
return await _run(_dl)
|
||||||
|
|
||||||
|
|
||||||
# ── channel art ───────────────────────────────────────────────────────────────
|
# ── channel art ───────────────────────────────────────────────────────────────
|
||||||
|
|
||||||
async def download_channel_art(channel_url: str) -> list:
|
async def download_channel_art(channel_url: str) -> List[dict]:
|
||||||
def _dl():
|
"""
|
||||||
import httpx
|
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:
|
with yt_dlp.YoutubeDL(opts) as ydl:
|
||||||
info = ydl.extract_info(channel_url, download=False)
|
info = ydl.extract_info(channel_url, download=False)
|
||||||
|
|
||||||
results = []
|
|
||||||
uid = _uid()
|
|
||||||
|
|
||||||
# avatar — last thumbnail is usually highest res
|
|
||||||
thumbnails = info.get("thumbnails") or []
|
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:
|
if avatar_url:
|
||||||
path = DOWNLOAD_DIR / f"yt_ch_{uid}_avatar.jpg"
|
path = DOWNLOAD_DIR / f"yt_ch_{uid}_avatar.jpg"
|
||||||
data = httpx.get(avatar_url, follow_redirects=True, timeout=15).content
|
try:
|
||||||
path.write_bytes(data)
|
_download_image(avatar_url, path)
|
||||||
results.append({"type": "avatar", "url": f"/files/{path.name}"})
|
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:
|
if banner_url:
|
||||||
path = DOWNLOAD_DIR / f"yt_ch_{uid}_banner.jpg"
|
path = DOWNLOAD_DIR / f"yt_ch_{uid}_banner.jpg"
|
||||||
data = httpx.get(banner_url, follow_redirects=True, timeout=15).content
|
try:
|
||||||
path.write_bytes(data)
|
_download_image(banner_url, path)
|
||||||
results.append({"type": "banner", "url": f"/files/{path.name}"})
|
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:
|
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 results
|
||||||
|
|
||||||
return await _run(_dl)
|
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()
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue