88 lines
3.1 KiB
Python
88 lines
3.1 KiB
Python
# utils/boxapi_parser.py
|
|
from typing import Any, Dict, List, Optional
|
|
from datetime import datetime
|
|
|
|
def _ts_to_dt(ts: Optional[int]) -> datetime:
|
|
# BoxAPI/IG timestamps appear to be epoch seconds
|
|
if ts is None:
|
|
return datetime.utcnow()
|
|
return datetime.utcfromtimestamp(int(ts))
|
|
|
|
def pick_image_urls(item: Dict[str, Any]) -> List[str]:
|
|
urls: List[str] = []
|
|
image_versions2 = item.get("image_versions2") or {}
|
|
for c in image_versions2.get("candidates", []):
|
|
u = c.get("url")
|
|
if u:
|
|
urls.append(u)
|
|
# Fallback to additional_candidates.first_frame if exists
|
|
add = image_versions2.get("additional_candidates") or {}
|
|
first_frame = add.get("first_frame") or {}
|
|
if first_frame.get("url"):
|
|
urls.append(first_frame["url"])
|
|
return list(dict.fromkeys(urls)) # dedupe
|
|
|
|
def pick_video_urls(item: Dict[str, Any]) -> List[str]:
|
|
urls: List[str] = []
|
|
for v in item.get("video_versions", []) or []:
|
|
u = v.get("url")
|
|
if u:
|
|
urls.append(u)
|
|
# Fallback: progressive_download_url if present
|
|
prog = (item.get("clips_metadata") or {}).get("original_sound_info") or {}
|
|
prog_url = prog.get("progressive_download_url")
|
|
if prog_url:
|
|
urls.append(prog_url)
|
|
return list(dict.fromkeys(urls))
|
|
|
|
def parse_boxapi_items(items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
|
"""
|
|
Normalized items for storage:
|
|
{
|
|
"ig_post_id": str,
|
|
"media_type": "image"|"video"|"carousel",
|
|
"caption": str,
|
|
"taken_at": datetime,
|
|
"thumbnail_url": Optional[str],
|
|
"remote_urls": List[str], # to be uploaded to S3
|
|
"code": Optional[str]
|
|
}
|
|
"""
|
|
out: List[Dict[str, Any]] = []
|
|
for it in items:
|
|
mt_raw = it.get("media_type")
|
|
if mt_raw == 1:
|
|
mtype = "image"
|
|
remote_urls = pick_image_urls(it)
|
|
elif mt_raw == 2:
|
|
mtype = "video"
|
|
remote_urls = pick_video_urls(it)
|
|
# Add a cover frame if available (first image candidate)
|
|
cover = pick_image_urls(it)[:1]
|
|
remote_urls = cover + remote_urls
|
|
elif mt_raw == 8: # carousel, if BoxAPI returns it
|
|
mtype = "carousel"
|
|
remote_urls = []
|
|
for child in it.get("carousel_media", []) or []:
|
|
if child.get("media_type") == 1:
|
|
remote_urls += pick_image_urls(child)
|
|
elif child.get("media_type") == 2:
|
|
# put a cover first if exists
|
|
cover = pick_image_urls(child)[:1]
|
|
remote_urls += cover + pick_video_urls(child)
|
|
else:
|
|
# Unknown, skip
|
|
continue
|
|
|
|
cap = (it.get("caption") or {}).get("text") or ""
|
|
thumb = (it.get("image_versions2") or {}).get("candidates", [{}])[0].get("url")
|
|
out.append({
|
|
"ig_post_id": str(it.get("pk") or it.get("id") or ""),
|
|
"media_type": mtype,
|
|
"caption": cap,
|
|
"taken_at": _ts_to_dt(it.get("taken_at")),
|
|
"thumbnail_url": thumb,
|
|
"remote_urls": remote_urls,
|
|
"code": it.get("code"),
|
|
})
|
|
return out |