fixed instagram avater probs and yt avater too ig

This commit is contained in:
katR 2026-04-11 10:31:43 +05:30
parent 3f398f7f71
commit f862e1d670
2 changed files with 207 additions and 93 deletions

View file

@ -2,16 +2,17 @@
Instagram service. Instagram service.
Strategy: Strategy:
- Posts/images: instaloader (with session) → yt-dlp fallback - Post/reel info: yt-dlp first, instaloader fallback
- Reels/videos: yt-dlp (instaloader 403s on reels) - Post download: yt-dlp first, instaloader fallback
- Profile pics: instaloader (uses username directly, no shortcode) - Reel download: yt-dlp only (instaloader 403s on reels)
- Thumbnails: yt-dlp - Profile picture: yt-dlp (avoids instaloader rate limits on Profile.from_username)
- Thumbnail: yt-dlp
Session setup (run once on the tablet): Session setup (run once on tablet, needed for instaloader fallback):
python3 -c " python3 -c "
import instaloader import instaloader
L = instaloader.Instaloader() L = instaloader.Instaloader()
L.interactive_login('your_ig_username') L.interactive_login('your_username')
L.save_session_to_file('ig_session') L.save_session_to_file('ig_session')
print('saved') print('saved')
" "
@ -66,6 +67,37 @@ def _fmt_s(s) -> str:
return f"{m}:{sec:02d}" return f"{m}:{sec:02d}"
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:
import subprocess
cmd = ["ffmpeg", "-y"]
if start:
cmd += ["-ss", start]
cmd += ["-i", str(src)]
if end:
cmd += ["-to", end]
cmd += ["-c", "copy", str(dest)]
result = subprocess.run(cmd, capture_output=True, timeout=120)
if result.returncode != 0:
raise RuntimeError(f"ffmpeg trim failed: {result.stderr.decode()[:200]}")
return dest
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,
)
def _get_loader() -> instaloader.Instaloader: def _get_loader() -> instaloader.Instaloader:
global _loader global _loader
if _loader is not None: if _loader is not None:
@ -103,42 +135,9 @@ def _get_loader() -> instaloader.Instaloader:
return _loader return _loader
def _download_image_url(url: str, dest: Path) -> Path: # ── post info ─────────────────────────────────────────────────────────────────
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:
import subprocess
cmd = ["ffmpeg", "-y"]
if start:
cmd += ["-ss", start]
cmd += ["-i", str(src)]
if end:
cmd += ["-to", end]
cmd += ["-c", "copy", str(dest)]
result = subprocess.run(cmd, capture_output=True, timeout=120)
if result.returncode != 0:
raise RuntimeError(f"ffmpeg trim failed: {result.stderr.decode()[:200]}")
return dest
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,
)
# ── post info (shortcode-based) ───────────────────────────────────────────────
async def fetch_ig_post_info(url: str) -> dict: async def fetch_ig_post_info(url: str) -> dict:
"""Info for post / reel / tv URLs."""
def _via_ytdlp(): def _via_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:
@ -175,12 +174,40 @@ async def fetch_ig_post_info(url: str) -> dict:
return await _run(_via_instaloader) return await _run(_via_instaloader)
# ── profile info (username-based) ───────────────────────────────────────────── # ── profile info ──────────────────────────────────────────────────────────────
async def fetch_ig_profile_info(username: str) -> dict: async def fetch_ig_profile_info(username: str) -> dict:
"""Info for profile pages — uses username directly, no shortcode.""" """
Fetch profile info using yt-dlp's Instagram user extractor.
This avoids instaloader's Profile.from_username() which gets
rate-limited and returns "does not exist" even for real accounts.
"""
def _via_ytdlp():
# yt-dlp can extract Instagram user pages
profile_url = f"https://www.instagram.com/{username}/"
opts = {
**_BASE_YDL,
"skip_download": True,
"extract_flat": True,
"playlist_items": "0",
}
with yt_dlp.YoutubeDL(opts) as ydl:
info = ydl.extract_info(profile_url, download=False)
def _fetch(): # yt-dlp returns uploader_url, thumbnail from profile pages
thumb = info.get("thumbnail") or (
info.get("thumbnails") or [{}]
)[-1].get("url")
return {
"type": "profile",
"title": f"@{username}",
"thumbnail": thumb,
"owner": username,
"likes": info.get("channel_follower_count") or 0,
}
def _via_instaloader():
L = _get_loader() L = _get_loader()
profile = instaloader.Profile.from_username(L.context, username) profile = instaloader.Profile.from_username(L.context, username)
try: try:
@ -192,10 +219,15 @@ async def fetch_ig_profile_info(username: str) -> dict:
"title": f"@{profile.username}", "title": f"@{profile.username}",
"thumbnail": pic_url, "thumbnail": pic_url,
"owner": profile.username, "owner": profile.username,
"likes": profile.followers, # repurpose likes field for follower count "likes": profile.followers,
} }
return await _run(_fetch) try:
return await _run(_via_ytdlp)
except Exception:
pass
# instaloader as last resort
return await _run(_via_instaloader)
# ── post download ───────────────────────────────────────────────────────────── # ── post download ─────────────────────────────────────────────────────────────
@ -260,7 +292,7 @@ async def download_post(
return await _run(_via_instaloader) return await _run(_via_instaloader)
# ── reel download (yt-dlp only) ─────────────────────────────────────────────── # ── reel download ─────────────────────────────────────────────────────────────
async def download_reel( async def download_reel(
url: str, url: str,
@ -299,12 +331,39 @@ async def download_reel(
async def download_profile(username: str) -> Tuple[str, str]: async def download_profile(username: str) -> Tuple[str, str]:
""" """
Downloads the profile picture for a given username. Downloads profile picture using yt-dlp's Instagram extractor,
username should already be clean (no @ prefix, no URL). falling back to instaloader if yt-dlp can't find a usable image URL.
""" """
uid = _uid() uid = _uid()
def _dl(): def _via_ytdlp():
profile_url = f"https://www.instagram.com/{username}/"
opts = {
**_BASE_YDL,
"skip_download": True,
"extract_flat": True,
"playlist_items": "0",
}
with yt_dlp.YoutubeDL(opts) as ydl:
info = ydl.extract_info(profile_url, download=False)
# try thumbnail fields in order of preference
thumb = (
info.get("thumbnail")
or next(
(t.get("url") for t in reversed(info.get("thumbnails") or [])
if t.get("url")),
None,
)
)
if not thumb:
raise ValueError("yt-dlp found no thumbnail for this profile")
path = DOWNLOAD_DIR / f"ig_pfp_{_safe(username)}_{uid}.jpg"
_download_image_url(thumb, path)
return str(path), f"{_safe(username)}_profile.jpg"
def _via_instaloader():
L = _get_loader() L = _get_loader()
profile = instaloader.Profile.from_username(L.context, username) profile = instaloader.Profile.from_username(L.context, username)
try: try:
@ -315,7 +374,11 @@ async def download_profile(username: str) -> Tuple[str, str]:
_download_image_url(pic_url, path) _download_image_url(pic_url, path)
return str(path), f"{_safe(username)}_profile.jpg" return str(path), f"{_safe(username)}_profile.jpg"
return await _run(_dl) try:
return await _run(_via_ytdlp)
except Exception:
pass
return await _run(_via_instaloader)
# ── thumbnail ───────────────────────────────────────────────────────────────── # ── thumbnail ─────────────────────────────────────────────────────────────────

View file

@ -240,14 +240,15 @@ async def download_subtitles(url: str, lang: str = "en") -> Tuple[str, str]:
async def download_channel_art(channel_url: str) -> List[dict]: async def download_channel_art(channel_url: str) -> List[dict]:
""" """
Given a channel URL (youtube.com/@handle or youtube.com/channel/UCxxxxx), Downloads avatar + banner for a YouTube channel as separate JPEGs.
downloads the avatar and banner as separate JPEG files.
Returns: Approach:
[ 1. Use yt-dlp to get channel metadata (uploader, thumbnails list)
{"type": "avatar", "filename": "channelname_avatar.jpg", "url": "/files/..."}, 2. Parse thumbnails list — YouTube returns avatars (square, ggpht CDN)
{"type": "banner", "filename": "channelname_banner.jpg", "url": "/files/..."}, and banners (wide ratio) as separate entries
] 3. Download each directly with httpx
Returns list of {"type", "filename", "url"} dicts.
""" """
def _dl(): def _dl():
@ -255,38 +256,53 @@ async def download_channel_art(channel_url: str) -> List[dict]:
uid = _uid() uid = _uid()
# fetch channel metadata — extract_flat avoids fetching all videos # fetch channel page metadata only — no video list needed
# DO NOT use extract_flat here, it causes yt-dlp to return
# video entries instead of channel metadata
opts = { opts = {
**_BASE_OPTS, **_BASE_OPTS,
"skip_download": True, "skip_download": True,
"extract_flat": True, "extract_flat": False,
"playlist_items": "0", # fetch 0 videos — just channel info
} }
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)
channel_name = _safe(info.get("channel") or info.get("uploader") or "channel") channel_name = _safe(
thumbnails = info.get("thumbnails") or [] info.get("channel")
results = [] or info.get("uploader")
or info.get("title")
or "channel"
)
thumbnails = info.get("thumbnails") or []
results = []
print(f"[yt channel] found {len(thumbnails)} thumbnails for {channel_name}")
for t in thumbnails:
print(f" w={t.get('width')} h={t.get('height')} url={str(t.get('url',''))[:80]}")
# ── avatar ──────────────────────────────────────────────────────────── # ── avatar ────────────────────────────────────────────────────────────
# YouTube channel avatars come from ggpht.com and are square # YouTube avatar thumbnails:
# - hosted on ggpht.com or lh3.googleusercontent.com
# - square aspect ratio (width == height)
avatar_url = None avatar_url = None
avatar_candidates = [
t for t in thumbnails
if t.get("url") and _is_avatar_thumb(t)
]
if avatar_candidates:
# pick highest resolution
best = max(
avatar_candidates,
key=lambda t: (t.get("width") or 0) * (t.get("height") or 0),
)
avatar_url = best["url"]
# for ggpht URLs, request a larger size by tweaking the size param
avatar_url = _upscale_ggpht(avatar_url, 800)
# first try thumbnails tagged as square / avatar-like # fallback: channel's own thumbnail field
square = [t for t in thumbnails if _is_avatar(t)] if not avatar_url and info.get("thumbnail"):
if square: avatar_url = info["thumbnail"]
# 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: if avatar_url:
dest = DOWNLOAD_DIR / f"{channel_name}_avatar_{uid}.jpg" dest = DOWNLOAD_DIR / f"{channel_name}_avatar_{uid}.jpg"
@ -297,17 +313,22 @@ async def download_channel_art(channel_url: str) -> List[dict]:
"filename": dest.name, "filename": dest.name,
"url": f"/files/{dest.name}", "url": f"/files/{dest.name}",
}) })
print(f"[yt channel] avatar saved: {dest.name}")
except Exception as e: except Exception as e:
print(f"[yt channel] avatar download failed: {e}") print(f"[yt channel] avatar download failed: {e}")
# ── banner ──────────────────────────────────────────────────────────── # ── banner ────────────────────────────────────────────────────────────
# Banners are wide (ratio > 3:1) or explicitly keyed
banner_url = info.get("header_image") or info.get("banner") banner_url = info.get("header_image") or info.get("banner")
if not banner_url: if not banner_url:
# look for very wide thumbnails (banner aspect ratio) banner_candidates = [
wide = [t for t in thumbnails if _is_banner(t)] t for t in thumbnails
if wide: if t.get("url") and _is_banner_thumb(t)
banner_url = max(wide, key=lambda t: t.get("width") or 0)["url"] ]
if banner_candidates:
best = max(banner_candidates, key=lambda t: t.get("width") or 0)
banner_url = best["url"]
if banner_url: if banner_url:
dest = DOWNLOAD_DIR / f"{channel_name}_banner_{uid}.jpg" dest = DOWNLOAD_DIR / f"{channel_name}_banner_{uid}.jpg"
@ -318,13 +339,14 @@ async def download_channel_art(channel_url: str) -> List[dict]:
"filename": dest.name, "filename": dest.name,
"url": f"/files/{dest.name}", "url": f"/files/{dest.name}",
}) })
print(f"[yt channel] banner saved: {dest.name}")
except Exception as e: except Exception as e:
print(f"[yt channel] banner download failed: {e}") print(f"[yt channel] banner download failed: {e}")
if not results: if not results:
raise ValueError( raise ValueError(
"no channel art found — make sure you're using a channel URL " "no channel art found — use a channel URL like "
"like youtube.com/@channelname or youtube.com/channel/UCxxxxxx" "youtube.com/@channelname or youtube.com/channel/UCxxxxxx"
) )
return results return results
@ -332,19 +354,48 @@ async def download_channel_art(channel_url: str) -> List[dict]:
return await _run(_dl) return await _run(_dl)
def _is_avatar(t: dict) -> bool: def _is_avatar_thumb(t: dict) -> bool:
"""True if this thumbnail looks like a channel avatar (square, ggpht CDN)."""
url = t.get("url") or ""
w = t.get("width") or 0 w = t.get("width") or 0
h = t.get("height") or 1 h = t.get("height") or 1
url = t.get("url") or ""
if w and h and 0.8 < (w / h) < 1.3: # ggpht / lh3.googleusercontent = Google's avatar CDN
if any(cdn in url for cdn in ("ggpht.com", "lh3.googleusercontent.com")):
return True return True
return any(kw in url for kw in ("ggpht", "photo", "avatar"))
# square-ish and small-to-medium (avatars are usually ≤ 900px)
if w and h and 0.85 < (w / h) < 1.15 and w <= 900:
return True
return False
def _is_banner(t: dict) -> bool: def _is_banner_thumb(t: dict) -> bool:
"""True if this thumbnail looks like a channel banner (very wide)."""
url = t.get("url") or ""
w = t.get("width") or 0 w = t.get("width") or 0
h = t.get("height") or 1 h = t.get("height") or 1
url = t.get("url") or ""
if w and h and (w / h) > 2.5: if "banner" in url.lower():
return True return True
return "banner" in url.lower()
# banners are typically 2560×1440 or similar ultra-wide
if w and h and (w / h) > 3.0:
return True
return False
def _upscale_ggpht(url: str, size: int) -> str:
"""
ggpht URLs end with =s{size} or =s{size}-c.
Replace with a larger size for higher resolution avatars.
e.g. =s88-c-k → =s800-c-k
"""
# match =sNNN optionally followed by other params
upscaled = re.sub(r"=s\d+", f"=s{size}", url)
if upscaled != url:
return upscaled
# if no size param found, just return original
return url