# 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