diff --git a/scraper.py b/scraper.py index 3da0d3b..48bd496 100644 --- a/scraper.py +++ b/scraper.py @@ -22,38 +22,45 @@ def dprint(msg: str) -> None: if DIAG: print(f"[SCRAPER] {msg}") -def _ig_id(it: Dict[str, Any]) -> Optional[str]: - v = it.get("id") or it.get("pk") - return str(v) if v is not None else None - -def _db_existing_ids(db: Session, ids: List[str]) -> Set[str]: - if not ids: +def _db_existing_ids_by_ig(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_(ids)).all() + rows = ( + db.query(InstagramPost.ig_post_id) + .filter(InstagramPost.ig_post_id.in_(ig_ids)) + .all() + ) return {r[0] for r in rows} -def _collect_raw_not_in_db( +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 _collect_normalized_not_in_db( db: Session, username: str, - needed: int, + target: int, + start_cursor: Optional[str], since_ts: Optional[int], ) -> tuple[List[Dict[str, Any]], Optional[str], int]: """ - Walk pages linearly: newest first (max_id=None), then older via next_max_id. - Keep only posts NOT in DB. Stop when we have `needed` or no more pages. - Returns (raw_to_save, final_cursor, pages_walked) + 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) """ - raw_to_save: List[Dict[str, Any]] = [] + to_save_norm: List[Dict[str, Any]] = [] pages = 0 - first = True - max_id: Optional[str] = None # None → newest page - final_cursor: Optional[str] = None + max_id = start_cursor # None => newest + final_cursor: Optional[str] = start_cursor - while len(raw_to_save) < needed and pages < MAX_PAGES: - count = min(12, max(1, needed - len(raw_to_save))) # 1..12 + while len(to_save_norm) < target and pages < MAX_PAGES: + # Always ask up to 12; progress even when many are duplicates + per_req = 12 try: - page, _log, next_max = get_media_page(username=username, count=count, max_id=max_id) + page, _log, next_max = get_media_page(username=username, count=per_req, max_id=max_id) except BoxAPIError as e: dprint(f"BoxAPI error: {e}") break @@ -61,66 +68,52 @@ def _collect_raw_not_in_db( pages += 1 dprint(f"Page#{pages}: request max_id={max_id!r} → got {len(page)} items, next_max_id={next_max!r}") - # after first fetch, always continue with next_max_id - first = False - if not page: - # no items on this page, stop if no further cursor + # no data; if there's a next, follow it once; else stop if not next_max: - dprint("Empty page and no next_max_id → stopping") break - # otherwise, advance cursor and continue (rare) max_id = next_max final_cursor = next_max continue - # Optional time cutoff - filtered: List[Dict[str, Any]] = [] - ids_ordered: List[str] = [] - for it in page: - pid = _ig_id(it) - if not pid: + # Normalize THIS page (critical: now ig_post_id matches DB) + norm_page = parse_boxapi_items_fast(page) + + # Optional timestamp cutoff + if since_ts is not None: + try: + s = int(since_ts) + 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] + + # 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)}") + + 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 - if since_ts is not None: - ts = it.get("taken_at") or it.get("timestamp") - try: - ts = int(ts) if ts is not None else None - except Exception: - ts = None - if ts is not None and ts <= since_ts: - continue - filtered.append(it) - ids_ordered.append(pid) + to_save_norm.append(n) + if len(to_save_norm) >= target: + break - if ids_ordered: - already = _db_existing_ids(db, ids_ordered) - dprint(f" from {len(ids_ordered)} IDs, already_in_db={len(already)}") - for it in filtered: - pid = _ig_id(it) - if pid and pid not in already: - raw_to_save.append(it) - if len(raw_to_save) >= needed: - break - else: - dprint(" no valid IDs after filtering") - - # advance pagination cursor for the next loop + # advance cursor final_cursor = next_max if not next_max: - dprint("No next_max_id → reached end of feed") + dprint("No next_max_id → end of feed") break + max_id = next_max - max_id = next_max # ← KEY: chase older page next round - - dprint(f"Collected to_save={len(raw_to_save)} across pages={pages}, final_cursor={final_cursor!r}") - return raw_to_save, 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]: - # Prefer "count" if provided by the caller, otherwise fall back to "max_count" - target = int((req.count if getattr(req, "count", None) is not None else req.max_count)) - - # Optional debug line - print(f"[SCRAPER] requested count={getattr(req, 'count', None)} max_count={req.max_count} → resolved target={target}") + 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 @@ -130,26 +123,28 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: except Exception: since_ts = None - # 1) BRUTE-FORCE scan: newest → older, pick posts NOT in DB until we reach target - raw_to_save, final_cursor, pages_walked = _collect_raw_not_in_db( + # 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, - needed=target, + target=target, + start_cursor=start_cursor, since_ts=since_ts, ) - # Persist cursor if we walked at least one page + # Persist cursor if we walked any page if pages_walked > 0: set_cursor(db, req.username, final_cursor) - if not raw_to_save: - dprint("No new posts to save.") + if not normalized_to_save: + dprint("No new normalized posts to save.") return [] - # 2) Normalize only the ones we intend to save - normalized = parse_boxapi_items_fast(raw_to_save)[:target] - - # 3) Ensure account row + # 2) Ensure account row account = db.query(InstagramAccount).filter_by(username=req.username).first() if not account: account = InstagramAccount( @@ -162,18 +157,20 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: db.commit() db.refresh(account) - # 4) Upload + save + # 3) Upload + save result: List[ScrapedPost] = [] - for item in normalized: + for item in normalized_to_save[:target]: post_id = str(item["ig_post_id"]) seller_id = str(req.seller_id) - # DB guard in case a parallel worker just saved it + # DB guard (parallel runner safety) if db.query(InstagramPost).filter_by(ig_post_id=post_id).first(): continue + # Media URLs media_urls = list(item.get("remote_urls", [])) + # If video, keep only video URLs (avoid thumb as media) if item["media_type"] == "video": thumb = (item.get("thumbnail_url") or "").strip() def is_video(u: str) -> bool: @@ -181,6 +178,7 @@ 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: @@ -198,6 +196,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: print(f"⚠️ Error uploading media: {e}") continue + # Thumbnail final_thumbnail = item.get("thumbnail_url") or None if final_thumbnail: try: @@ -213,6 +212,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: created_at = datetime.fromtimestamp(item["taken_at"]) if item.get("taken_at") else datetime.utcnow() + # Insert post_entry = InstagramPost( ig_post_id=post_id, account_id=account.id,