diff --git a/scraper.py b/scraper.py index 48bd496..3bf78ad 100644 --- a/scraper.py +++ b/scraper.py @@ -11,7 +11,7 @@ from utils.s3_uploader import ( upload_video_from_url_optimized, ) from schemas import ScrapeRequest, ScrapedPost -from typing import List, Dict, Any, Optional, Set +from typing import List, Dict, Any, Optional, Tuple, Set from datetime import datetime from scraper_cursor import get_cursor, set_cursor @@ -22,7 +22,11 @@ def dprint(msg: str) -> None: if DIAG: print(f"[SCRAPER] {msg}") -def _db_existing_ids_by_ig(db: Session, ig_ids: List[str]) -> Set[str]: +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 = ( @@ -32,88 +36,135 @@ def _db_existing_ids_by_ig(db: Session, ig_ids: List[str]) -> Set[str]: ) return {r[0] for r in rows} -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 _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 _collect_normalized_not_in_db( +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]], Optional[str], int]: +) -> Tuple[List[Dict[str, Any]], List[str], Optional[str], int]: """ - Walk pages: start from start_cursor (if any) else newest (None). - For each page: - - normalize - - DB-check by normalized ig_post_id - - accumulate unseen until reaching target or no next_max_id - Returns (normalized_to_save, final_cursor, pages_walked) + 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 """ - to_save_norm: List[Dict[str, Any]] = [] + normalized_new: List[Dict[str, Any]] = [] + dup_ids_ordered: List[str] = [] pages = 0 - max_id = start_cursor # None => newest - final_cursor: Optional[str] = start_cursor - while len(to_save_norm) < target and pages < MAX_PAGES: - # Always ask up to 12; progress even when many are duplicates - per_req = 12 + # 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, next_max = get_media_page(username=username, count=per_req, max_id=max_id) + page, _log, nxt = get_media_page(username=username, count=per_req, max_id=max_id) except BoxAPIError as e: - dprint(f"BoxAPI error: {e}") + dprint(f"BoxAPI error (A): {e}") break pages += 1 - dprint(f"Page#{pages}: request max_id={max_id!r} → got {len(page)} items, next_max_id={next_max!r}") + 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 - if not page: - # no data; if there's a next, follow it once; else stop - if not next_max: - break - max_id = next_max - final_cursor = next_max - continue - - # Normalize THIS page (critical: now ig_post_id matches DB) - norm_page = parse_boxapi_items_fast(page) - - # Optional timestamp cutoff + 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: - s = None - if s is not None: - norm_page = [n for n in norm_page if (n.get("taken_at") or 0) > s] + pass - # DB duplicate filter using normalized ig_post_id - norm_ids = [str(n.get("ig_post_id")) for n in norm_page if n.get("ig_post_id")] - already = _db_existing_ids_by_ig(db, norm_ids) - dprint(f" normalized ids={len(norm_ids)} already_in_db={len(already)}") + 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] - for n in norm_page: - igid = str(n.get("ig_post_id")) if n.get("ig_post_id") else None - if not igid or igid in already: - continue - to_save_norm.append(n) - if len(to_save_norm) >= target: - break + normalized_new.extend(new_here) + dup_ids_ordered.extend(dups_here) - # advance cursor - final_cursor = next_max - if not next_max: + 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 - max_id = next_max + backfill_cursor = nxt + + return normalized_new, dup_ids_ordered, final_cursor, pages - dprint(f"Collected to_save={len(to_save_norm)} across pages={pages}, final_cursor={final_cursor!r}") - return to_save_norm, 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}") + 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 @@ -123,28 +174,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: except Exception: since_ts = None - # Start from stored cursor so the next call continues deeper - start_cursor = get_cursor(db, req.username) - dprint(f"start_cursor={start_cursor!r}") - - # 1) Collect normalized posts NOT in DB, paging deeper as needed - normalized_to_save, final_cursor, pages_walked = _collect_normalized_not_in_db( - db=db, - username=req.username, - target=target, - start_cursor=start_cursor, - since_ts=since_ts, - ) - - # Persist cursor if we walked any page - if pages_walked > 0: - set_cursor(db, req.username, final_cursor) - - if not normalized_to_save: - dprint("No new normalized posts to save.") - return [] - - # 2) Ensure account row + # Ensure account row (needed for DB top-up) account = db.query(InstagramAccount).filter_by(username=req.username).first() if not account: account = InstagramAccount( @@ -157,20 +187,39 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: db.commit() db.refresh(account) - # 3) Upload + save - result: List[ScrapedPost] = [] - for item in normalized_to_save[:target]: - post_id = str(item["ig_post_id"]) - seller_id = str(req.seller_id) + # Start from stored cursor (for deeper backfill) + start_cursor = get_cursor(db, req.username) + dprint(f"start_cursor={start_cursor!r}") - # DB guard (parallel runner safety) + # 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 - # Media URLs + seller_id = str(req.seller_id) media_urls = list(item.get("remote_urls", [])) - # If video, keep only video URLs (avoid thumb as media) + # 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: @@ -178,7 +227,6 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: return base.endswith((".mp4", ".mov", ".m4v", ".webm")) media_urls = [u for u in media_urls if is_video(u) and u != thumb] - # Upload uploaded_urls: List[str] = [] for remote_url in media_urls: try: @@ -196,7 +244,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: print(f"⚠️ Error uploading media: {e}") continue - # Thumbnail + 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: @@ -210,10 +258,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: except Exception as e: print(f"⚠️ Failed to upload thumbnail: {e}") - created_at = datetime.fromtimestamp(item["taken_at"]) if item.get("taken_at") else datetime.utcnow() - - # Insert - post_entry = InstagramPost( + row = InstagramPost( ig_post_id=post_id, account_id=account.id, media_type=item["media_type"], @@ -222,7 +267,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: local_media=uploaded_urls, created_at=created_at, ) - db.add(post_entry) + db.add(row) db.commit() result.append(ScrapedPost( @@ -234,7 +279,27 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: created_at=created_at, )) - if len(result) >= target: - break + # 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 - return result \ No newline at end of file + # 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