# scraper.py import os from sqlalchemy.orm import Session from database import InstagramAccount, InstagramPost from utils.boxapi_client import get_media_page, BoxAPIError from utils.parser_fast import parse_boxapi_items_fast from utils.s3_uploader import ( upload_media_from_url, upload_image_from_url_optimized, upload_video_from_url_optimized, ) from schemas import ScrapeRequest, ScrapedPost from typing import List, Dict, Any, Optional, Tuple, Set from datetime import datetime from scraper_cursor import get_cursor, set_cursor DIAG = os.getenv("SCRAPER_DIAGNOSTICS", "0") == "1" MAX_PAGES = int(os.getenv("SCRAPER_MAX_PAGES", "400")) # safety cap def dprint(msg: str) -> None: if DIAG: print(f"[SCRAPER] {msg}") def _resolve_target(req: ScrapeRequest) -> int: # accept both "count" and "max_count" return int((getattr(req, "count", None) if getattr(req, "count", None) is not None else req.max_count)) def _db_existing_ids(db: Session, ig_ids: List[str]) -> Set[str]: if not ig_ids: return set() rows = ( db.query(InstagramPost.ig_post_id) .filter(InstagramPost.ig_post_id.in_(ig_ids)) .all() ) return {r[0] for r in rows} def _load_posts_by_ids_ordered(db: Session, account_id, ig_ids: List[str]) -> List[InstagramPost]: """Fetch already-saved posts for this account, preserving ig_ids order.""" if not ig_ids: return [] rows = ( db.query(InstagramPost) .filter(InstagramPost.account_id == account_id, InstagramPost.ig_post_id.in_(ig_ids)) .all() ) by_id = {r.ig_post_id: r for r in rows} return [by_id[i] for i in ig_ids if i in by_id] def _normalize_page(page: List[Dict[str, Any]]) -> List[Dict[str, Any]]: if not page: return [] return parse_boxapi_items_fast(page) def _collect_normalized_pages( db: Session, username: str, target: int, start_cursor: Optional[str], since_ts: Optional[int], ) -> Tuple[List[Dict[str, Any]], List[str], Optional[str], int]: """ Collect normalized items across pages until we have at least `target` items when counting: - all new (not in DB), plus - duplicates we can top up from DB later. Returns: normalized_new -> list of normalized, not in DB dup_ids_ordered -> list of IG IDs already in DB (for top-up) final_cursor -> next_max_id to persist if we traversed older pages pages_walked """ normalized_new: List[Dict[str, Any]] = [] dup_ids_ordered: List[str] = [] pages = 0 # Phase A: NEWEST first newest_cursor = None max_id = None while (len(normalized_new) + len(dup_ids_ordered) < target) and pages < MAX_PAGES: per_req = 12 # always ask 12 for fewer roundtrips try: page, _log, nxt = get_media_page(username=username, count=per_req, max_id=max_id) except BoxAPIError as e: dprint(f"BoxAPI error (A): {e}") break pages += 1 dprint(f"[A] Page#{pages}: request max_id={max_id!r} → got {len(page)} items, next_max_id={nxt!r}") if newest_cursor is None: newest_cursor = nxt norm = _normalize_page(page) if since_ts is not None: try: s = int(since_ts) norm = [n for n in norm if (n.get("taken_at") or 0) > s] except Exception: pass ids = [str(n.get("ig_post_id")) for n in norm if n.get("ig_post_id")] already = _db_existing_ids(db, ids) new_here = [n for n in norm if n.get("ig_post_id") and str(n["ig_post_id"]) not in already] dups_here = [str(n["ig_post_id"]) for n in norm if n.get("ig_post_id") and str(n["ig_post_id"]) in already] normalized_new.extend(new_here) dup_ids_ordered.extend(dups_here) dprint(f" normalized={len(norm)} new={len(new_here)} dups={len(dups_here)} " f"total_collected={len(normalized_new)+len(dup_ids_ordered)}") if len(normalized_new) >= target or (len(normalized_new) + len(dup_ids_ordered)) >= target: # enough candidates to build exact N return normalized_new, dup_ids_ordered, None, pages if not nxt: break max_id = nxt # Phase B: BACKFILL older (use stored cursor if provided, else continue from newest_cursor) backfill_cursor = start_cursor or newest_cursor final_cursor = None while (len(normalized_new) + len(dup_ids_ordered) < target) and backfill_cursor and pages < MAX_PAGES: per_req = 12 try: page, _log, nxt = get_media_page(username=username, count=per_req, max_id=backfill_cursor) except BoxAPIError as e: dprint(f"BoxAPI error (B): {e}") break pages += 1 dprint(f"[B] Page#{pages}: request max_id={backfill_cursor!r} → got {len(page)} items, next_max_id={nxt!r}") final_cursor = nxt # we advanced older pages norm = _normalize_page(page) if since_ts is not None: try: s = int(since_ts) norm = [n for n in norm if (n.get("taken_at") or 0) > s] except Exception: pass ids = [str(n.get("ig_post_id")) for n in norm if n.get("ig_post_id")] already = _db_existing_ids(db, ids) new_here = [n for n in norm if n.get("ig_post_id") and str(n["ig_post_id"]) not in already] dups_here = [str(n["ig_post_id"]) for n in norm if n.get("ig_post_id") and str(n["ig_post_id"]) in already] normalized_new.extend(new_here) dup_ids_ordered.extend(dups_here) dprint(f" normalized={len(norm)} new={len(new_here)} dups={len(dups_here)} " f"total_collected={len(normalized_new)+len(dup_ids_ordered)}") if len(normalized_new) >= target or (len(normalized_new) + len(dup_ids_ordered)) >= target: return normalized_new, dup_ids_ordered, final_cursor, pages if not nxt: dprint("No next_max_id → end of feed") break backfill_cursor = nxt return normalized_new, dup_ids_ordered, final_cursor, pages def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: target = _resolve_target(req) dprint(f"requested count={getattr(req,'count',None)} max_count={req.max_count} → resolved target={target}") # since_taken_at (optional) since_ts: Optional[int] = None if hasattr(req, "since_taken_at") and req.since_taken_at is not None: try: since_ts = int(req.since_taken_at) except Exception: since_ts = None # Ensure account row (needed for DB top-up) account = db.query(InstagramAccount).filter_by(username=req.username).first() if not account: account = InstagramAccount( username=req.username, seller_id=req.seller_id, is_active=True, last_synced=None ) db.add(account) db.commit() db.refresh(account) # Start from stored cursor (for deeper backfill) start_cursor = get_cursor(db, req.username) dprint(f"start_cursor={start_cursor!r}") # 1) Collect normalized candidates across pages normalized_new, dup_ids_ordered, final_cursor, pages_walked = _collect_normalized_pages( db=db, username=req.username, target=target, start_cursor=start_cursor, since_ts=since_ts, ) # Persist backfill cursor only if we actually traversed older pages if pages_walked > 0 and final_cursor is not None: set_cursor(db, req.username, final_cursor) result: List[ScrapedPost] = [] # 2) First, save NEW posts (upload + insert) until we reach target or run out for item in normalized_new: if len(result) >= target: break post_id = str(item["ig_post_id"]) # Skip if somehow inserted by a parallel worker if db.query(InstagramPost).filter_by(ig_post_id=post_id).first(): continue seller_id = str(req.seller_id) media_urls = list(item.get("remote_urls", [])) # If video, keep only video URLs (avoid thumb) if item["media_type"] == "video": thumb = (item.get("thumbnail_url") or "").strip() def is_video(u: str) -> bool: base = u.split("?", 1)[0].lower() return base.endswith((".mp4", ".mov", ".m4v", ".webm")) media_urls = [u for u in media_urls if is_video(u) and u != thumb] uploaded_urls: List[str] = [] for remote_url in media_urls: try: if item["media_type"] == "video": url = upload_video_from_url_optimized(remote_url, seller_id, post_id) elif item["media_type"] in ("image", "carousel"): url = upload_image_from_url_optimized(remote_url, seller_id, post_id) if url is None: url = upload_media_from_url(remote_url, seller_id, post_id) else: url = upload_media_from_url(remote_url, seller_id, post_id) if url: uploaded_urls.append(url) except Exception as e: print(f"⚠️ Error uploading media: {e}") continue created_at = datetime.fromtimestamp(item["taken_at"]) if item.get("taken_at") else datetime.utcnow() final_thumbnail = item.get("thumbnail_url") or None if final_thumbnail: try: thumb_url = upload_image_from_url_optimized(final_thumbnail, seller_id, post_id) if thumb_url: final_thumbnail = thumb_url else: alt = upload_media_from_url(final_thumbnail, seller_id, post_id) if alt: final_thumbnail = alt except Exception as e: print(f"⚠️ Failed to upload thumbnail: {e}") row = InstagramPost( ig_post_id=post_id, account_id=account.id, media_type=item["media_type"], caption=item.get("caption"), thumbnail_url=final_thumbnail, local_media=uploaded_urls, created_at=created_at, ) db.add(row) db.commit() result.append(ScrapedPost( ig_post_id=post_id, media_type=item["media_type"], caption=item.get("caption"), thumbnail_url=final_thumbnail, local_media=uploaded_urls, created_at=created_at, )) # 3) If we still need more, TOP-UP with already-saved posts from DB (NO uploads) if len(result) < target and dup_ids_ordered: missing = target - len(result) # Load the dup rows in the same order as we saw them while paging dup_rows = _load_posts_by_ids_ordered(db, account.id, dup_ids_ordered) for row in dup_rows: if missing <= 0: break # Build response directly from DB (fast) result.append(ScrapedPost( ig_post_id=row.ig_post_id, media_type=row.media_type, caption=row.caption, thumbnail_url=row.thumbnail_url, local_media=row.local_media or [], created_at=row.created_at, )) missing -= 1 # 4) If STILL not enough (end of feed + tiny DB), just return what we have if len(result) < target: dprint(f"feed_exhausted_or_insufficient_total=True returned={len(result)} requested={target}") return result