From c80a10680b6e94468b97071fcfb16410111f7a08 Mon Sep 17 00:00:00 2001 From: Hossein Date: Thu, 25 Sep 2025 02:26:58 +0330 Subject: [PATCH] update --- scraper.py | 191 ++++++++++++++++++++++++++++++++++++----------------- 1 file changed, 129 insertions(+), 62 deletions(-) diff --git a/scraper.py b/scraper.py index f4e8806..1b5ff64 100644 --- a/scraper.py +++ b/scraper.py @@ -18,76 +18,150 @@ from datetime import datetime from scraper_cursor import get_cursor, set_cursor # ← new helpers -def _collect_across_pages(username: str, target: int, seed_cursor: Optional[str]) -> tuple[list[Dict[str, Any]], Optional[str]]: - """ - 1) Fetch ONE newest page (max_id=None). - 2) If still short, continue with `seed_cursor` (or newest next_max_id) to backfill older pages. - Returns (raw_items, final_cursor_to_store). - """ - collected: list[Dict[str, Any]] = [] - newest_next: Optional[str] = None - - # Step 1: newest page first (cheap check for new content) - need = min(12, target) +def _extract_minimal(item: Dict[str, Any]) -> tuple[Optional[str], Optional[int]]: + """Return (ig_id, taken_at) from a raw BoxAPI item without full parsing.""" + ig_id = item.get("id") or item.get("pk") + if ig_id is not None: + ig_id = str(ig_id) + taken_at = item.get("taken_at") or item.get("timestamp") # might be int/epoch try: - page, _log, newest_next = get_media_page(username=username, count=need, max_id=None) - if page: - collected.extend(page) + taken_at = int(taken_at) if taken_at is not None else None except Exception: - # ignore and continue with backfill-only path - pass + taken_at = None + return ig_id, taken_at - # Step 2: continue older pages from stored cursor (or newest_next if none stored) - cursor = seed_cursor or newest_next - while len(collected) < target and cursor: - need = min(12, target - len(collected)) + +def _collect_unseen_raw( + username: str, + target_unseen: int, + stored_cursor: Optional[str], + since_ts: Optional[int], +) -> tuple[List[Dict[str, Any]], Optional[str], int]: + """ + Keep requesting BoxAPI pages until we accumulate `target_unseen` UNSEEN items, + or we run out of pages. We do minimal work per page to decide quickly. + + Returns: + - selected_raw: only the unseen (and since_ts-passing) raw items in order + - final_cursor: the last next_max_id we reached (to persist) + - unseen_count: number of unseen items we actually collected + """ + selected_raw: List[Dict[str, Any]] = [] + final_cursor: Optional[str] = stored_cursor + # 1) newest page first + next_id: Optional[str] = None + try: + first_page, _log, next_id = get_media_page(username=username, count=min(12, target_unseen), max_id=None) + except Exception: + first_page = [] + + def add_unseen_from_page(page: List[Dict[str, Any]]) -> int: + """Filter a page to unseen (and since_ts), append to selected_raw, return how many added.""" + if not page: + return 0 + # Extract ids in order + ids_in_order: List[str] = [] + since_mask: List[bool] = [] + for it in page: + ig_id, taken_at = _extract_minimal(it) + if ig_id: + # optional since_ts filter (skip if not passing) + ok = True + if since_ts is not None and taken_at is not None: + ok = taken_at > since_ts + ids_in_order.append(ig_id) + since_mask.append(ok) + if not ids_in_order: + return 0 + + # Fast unseen check + unseen_ids_set = set(filter_unseen(username, ids_in_order)) + if not unseen_ids_set: + return 0 + + # Append only ids that pass since_mask AND are unseen, preserving order + added = 0 + for it in page: + ig_id, taken_at = _extract_minimal(it) + if not ig_id: + continue + idx = ids_in_order.index(ig_id) # safe: ig_id exists in list + if not since_mask[idx]: + continue + if ig_id in unseen_ids_set: + selected_raw.append(it) + added += 1 + if len(selected_raw) >= target_unseen: + break + return added + + # Process newest page first + add_unseen_from_page(first_page) + + # 2) then keep going with stored cursor (older pages). Seed with stored or newest next_id. + cursor = stored_cursor or next_id + # Only continue if we still need more + while len(selected_raw) < target_unseen and cursor: + need = min(12, target_unseen - len(selected_raw)) page, _log, cursor = get_media_page(username=username, count=need, max_id=cursor) + added = add_unseen_from_page(page) + final_cursor = cursor # we move the cursor forward as we traverse if not page: break - collected.extend(page) + # keep looping until we hit target or run out (cursor becomes None) - return collected, cursor # final cursor to persist + return selected_raw, final_cursor, len(selected_raw) def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: target = int(req.max_count) - # 1) Collect raw items across newest + backfill pages (no request schema change) - stored_cursor = get_cursor(db, req.username) # may be None on first run + # Optional: allow caller to pass since_taken_at; coerce safely to int + since_ts_raw = getattr(req, "since_taken_at", None) + since_ts: Optional[int] = None + if since_ts_raw is not None: + try: + since_ts = int(since_ts_raw) + except Exception: + since_ts = None + + # 1) Gather ONLY unseen raw items up to target (newest → older via cursor) + stored_cursor = get_cursor(db, req.username) # may be None try: - raw_items, final_cursor = _collect_across_pages(req.username, target, stored_cursor) + selected_raw, final_cursor, unseen_count = _collect_unseen_raw( + username=req.username, + target_unseen=target, + stored_cursor=stored_cursor, + since_ts=since_ts, + ) except BoxAPIError as e: print(f"❌ BoxAPI error: {e}") return [] - if not raw_items: + if not selected_raw: + # Still store cursor progress (we may have advanced even if nothing new) + set_cursor(db, req.username, final_cursor) return [] - # 2) Normalize once on the aggregated list - normalized = parse_boxapi_items_fast(raw_items) + # 2) Normalize ONLY the selected unseen raw items + normalized = parse_boxapi_items_fast(selected_raw) - # 3) Optional cursor filter (kept as-is) - since_ts = getattr(req, "since_taken_at", None) - if since_ts: - try: - since_ts = int(since_ts) - normalized = [n for n in normalized if (n.get("taken_at") or 0) > since_ts] - except Exception: - pass + # 3) Defensive re-check: filter_unseen again after normalize by IG IDs (handles races) + ids_in_order = [n.get("ig_post_id") for n in normalized if n.get("ig_post_id")] + if not ids_in_order: + set_cursor(db, req.username, final_cursor) + return [] - # 4) Filter already-seen FIRST (avoid re-uploading) — ensure filter_unseen has NO hidden 5-cap - ids_in_order = [n["ig_post_id"] for n in normalized if n.get("ig_post_id")] unseen_ids = set(filter_unseen(req.username, ids_in_order)) - unseen = [n for n in normalized if n["ig_post_id"] in unseen_ids] + final_batch = [n for n in normalized if n.get("ig_post_id") in unseen_ids][:target] - # 5) Clip to requested count and persist the backfill cursor (progress) - batch = unseen[:target] + # Persist cursor progress AFTER we’ve walked pages set_cursor(db, req.username, final_cursor) - # 6) Mark these as seen so the next run skips them - mark_seen(req.username, [b["ig_post_id"] for b in batch]) + # Mark as seen now to avoid future reprocessing + mark_seen(req.username, [b["ig_post_id"] for b in final_batch]) - # 7) Ensure account row + # 4) Ensure account row account = db.query(InstagramAccount).filter_by(username=req.username).first() if not account: account = InstagramAccount( @@ -100,46 +174,41 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: db.commit() db.refresh(account) + # 5) Upload + save each post in the final batch result: List[ScrapedPost] = [] - new_post_count = 0 - - # 8) Upload + save each post in the batch - for item in batch: - post_id = str(item["ig_post_id"]) # ENSURE clean, no slashes + for item in final_batch: + post_id = str(item["ig_post_id"]) seller_id = str(req.seller_id) - # skip if already saved (race-safety) + # Skip if already saved (DB guard) if db.query(InstagramPost).filter_by(ig_post_id=post_id).first(): continue - # media URL list (normalized parser gives remote_urls) + # Media URLs (from normalized) media_urls = list(item.get("remote_urls", [])) - # If this is a video post, keep only true video URLs (avoid thumbnail duplication) + # If video, keep only true video URLs (avoid using the thumb URL) 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] - # Upload media to S3 (optimized path) + # Upload media uploaded_urls: List[str] = [] for remote_url in media_urls: try: if item["media_type"] == "video": - # optimized 480p H.264 CRF23 (+faststart) url = upload_video_from_url_optimized(remote_url, seller_id, post_id) elif item["media_type"] in ("image", "carousel"): - # carousel items may be images; optimize to WebP url = upload_image_from_url_optimized(remote_url, seller_id, post_id) if url is None: - # some carousels include videos; try generic fallback url = upload_media_from_url(remote_url, seller_id, post_id) else: - # unknown → generic detection/fallback url = upload_media_from_url(remote_url, seller_id, post_id) - if url: uploaded_urls.append(url) except Exception as e: @@ -154,14 +223,13 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: if thumb_url: final_thumbnail = thumb_url else: - # fallback to generic (no optimize) 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}") - # Save + # Save row created_at = datetime.fromtimestamp(item["taken_at"]) if item.get("taken_at") else datetime.utcnow() post_entry = InstagramPost( ig_post_id=post_id, @@ -184,8 +252,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: created_at=created_at, )) - new_post_count += 1 - if new_post_count >= target: + if len(result) >= target: break return result