66 lines
2.2 KiB
Python
66 lines
2.2 KiB
Python
# utils/parser_fast.py
|
|
from typing import Any, Dict, List, Tuple, Optional
|
|
|
|
def pick_image_url(item: Dict[str, Any]) -> Optional[str]:
|
|
cands = ((item.get("image_versions2") or {}).get("candidates") or [])
|
|
return cands[0].get("url") if cands else None
|
|
|
|
def pick_video_url(item: Dict[str, Any]) -> Optional[str]:
|
|
# choose the largest height (usually last or explicitly max)
|
|
vv = item.get("video_versions") or []
|
|
if not vv:
|
|
return None
|
|
# faster than sorted: single pass to get max by height
|
|
best = vv[0]
|
|
bh = best.get("height") or 0
|
|
for v in vv[1:]:
|
|
h = v.get("height") or 0
|
|
if h > bh:
|
|
bh, best = h, v
|
|
return best.get("url")
|
|
|
|
def normalize_item(it: Dict[str, Any]) -> Dict[str, Any]:
|
|
mt = it.get("media_type")
|
|
if mt == 1:
|
|
media_type = "image"
|
|
url = pick_image_url(it)
|
|
urls = [url] if url else []
|
|
elif mt == 2:
|
|
media_type = "video"
|
|
# put a cover first if exists
|
|
cover = pick_image_url(it)
|
|
vurl = pick_video_url(it)
|
|
urls = ([cover] if cover else []) + ([vurl] if vurl else [])
|
|
elif mt == 8: # carousel
|
|
media_type = "carousel"
|
|
urls = []
|
|
for child in it.get("carousel_media") or []:
|
|
if child.get("media_type") == 1:
|
|
u = pick_image_url(child)
|
|
if u: urls.append(u)
|
|
elif child.get("media_type") == 2:
|
|
cover = pick_image_url(child)
|
|
vurl = pick_video_url(child)
|
|
if cover: urls.append(cover)
|
|
if vurl: urls.append(vurl)
|
|
else:
|
|
return {}
|
|
|
|
return {
|
|
"ig_post_id": str(it.get("pk") or it.get("id") or ""),
|
|
"media_type": media_type,
|
|
"caption": ((it.get("caption") or {}).get("text") or ""),
|
|
"taken_at": int(it.get("taken_at") or 0),
|
|
"thumbnail_url": pick_image_url(it),
|
|
"remote_urls": [u for u in urls if u],
|
|
"code": it.get("code"),
|
|
}
|
|
|
|
def parse_boxapi_items_fast(items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
out = []
|
|
for it in items:
|
|
n = normalize_item(it)
|
|
if n: out.append(n)
|
|
# keep newest first
|
|
out.sort(key=lambda x: x["taken_at"], reverse=True)
|
|
return out |