From 79b87a5551a88dea6393b73b234c77c1b98ac47f Mon Sep 17 00:00:00 2001 From: Hossein Date: Thu, 25 Sep 2025 02:34:01 +0330 Subject: [PATCH] update --- scraper.py | 162 ++++++++++++++++++++++++----------------------------- 1 file changed, 74 insertions(+), 88 deletions(-) diff --git a/scraper.py b/scraper.py index 1b5ff64..8bbfa24 100644 --- a/scraper.py +++ b/scraper.py @@ -3,7 +3,7 @@ # --- imports --- from sqlalchemy.orm import Session from database import InstagramAccount, InstagramPost -from utils.boxapi_client import get_media_page, BoxAPIError # ← paged client +from utils.boxapi_client import get_media_page, BoxAPIError # paged client (12 per request, next_max_id) from utils.parser_fast import parse_boxapi_items_fast from utils.seen_store import filter_unseen, mark_seen # use optimized uploaders; keep the generic fallback as well @@ -15,15 +15,18 @@ from utils.s3_uploader import ( from schemas import ScrapeRequest, ScrapedPost from typing import List, Dict, Any, Optional from datetime import datetime -from scraper_cursor import get_cursor, set_cursor # ← new helpers +from scraper_cursor import get_cursor, set_cursor # cursor helpers +# ---------------------------- +# Minimal helpers (fast path) +# ---------------------------- 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 + taken_at = item.get("taken_at") or item.get("timestamp") try: taken_at = int(taken_at) if taken_at is not None else None except Exception: @@ -31,106 +34,92 @@ def _extract_minimal(item: Dict[str, Any]) -> tuple[Optional[str], Optional[int] return ig_id, taken_at -def _collect_unseen_raw( +def _collect_unseen( username: str, - target_unseen: int, + target: int, stored_cursor: Optional[str], - since_ts: Optional[int], -) -> tuple[List[Dict[str, Any]], Optional[str], int]: + since_ts: Optional[int] = None, +) -> tuple[List[Dict[str, Any]], Optional[str]]: """ - 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 + Keep fetching BoxAPI pages until we gather `target` unseen posts or run out of pages. + - Always skip already-seen (via filter_unseen) + - Optionally skip by since_ts + Returns: (raw_unseen_items, final_cursor) """ - 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 = [] + raw_unseen: List[Dict[str, Any]] = [] + cursor = stored_cursor + first_page_done = False + + while len(raw_unseen) < target: + # newest first, then older via next_max_id/cursor + max_id = None if not first_page_done else cursor + need = min(12, max(1, target - len(raw_unseen))) + try: + page, _log, next_id = get_media_page(username=username, count=need, max_id=max_id) + except BoxAPIError: + # Stop on BoxAPI error; return what we have and current cursor + break + + first_page_done = True + cursor = next_id # advance our traversal cursor even if page empty - 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 + break + + # Optional: filter by timestamp before unseen check + filtered_page: List[Dict[str, Any]] = [] 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]: + if since_ts is not None and taken_at is not None and taken_at <= since_ts: continue - if ig_id in unseen_ids_set: - selected_raw.append(it) - added += 1 - if len(selected_raw) >= target_unseen: - break - return added + filtered_page.append(it) + ids_in_order.append(ig_id) - # Process newest page first - add_unseen_from_page(first_page) + if not ids_in_order: + if not cursor: + break + continue - # 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: + # Fast unseen check using Redis/seen store + unseen_ids_set = set(filter_unseen(username, ids_in_order)) + if unseen_ids_set: + for it in filtered_page: + ig_id, _ = _extract_minimal(it) + if ig_id and ig_id in unseen_ids_set: + raw_unseen.append(it) + if len(raw_unseen) >= target: + break + + # If no next page, stop + if not cursor: break - # keep looping until we hit target or run out (cursor becomes None) - return selected_raw, final_cursor, len(selected_raw) + return raw_unseen, cursor +# ---------------------------- +# Main entry +# ---------------------------- def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: target = int(req.max_count) - # Optional: allow caller to pass since_taken_at; coerce safely to int - since_ts_raw = getattr(req, "since_taken_at", None) + # Optional since_taken_at support since_ts: Optional[int] = None - if since_ts_raw is not None: + if hasattr(req, "since_taken_at") and req.since_taken_at is not None: try: - since_ts = int(since_ts_raw) + since_ts = int(req.since_taken_at) 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 + stored_cursor = get_cursor(db, req.username) # may be None on first run try: - selected_raw, final_cursor, unseen_count = _collect_unseen_raw( + raw_unseen, final_cursor = _collect_unseen( username=req.username, - target_unseen=target, + target=target, stored_cursor=stored_cursor, since_ts=since_ts, ) @@ -138,28 +127,25 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: print(f"❌ BoxAPI error: {e}") return [] - if not selected_raw: - # Still store cursor progress (we may have advanced even if nothing new) - set_cursor(db, req.username, final_cursor) + # Persist traversal progress even if nothing new (so next run resumes deeper) + set_cursor(db, req.username, final_cursor) + + if not raw_unseen: return [] # 2) Normalize ONLY the selected unseen raw items - normalized = parse_boxapi_items_fast(selected_raw) + normalized = parse_boxapi_items_fast(raw_unseen) - # 3) Defensive re-check: filter_unseen again after normalize by IG IDs (handles races) + # 3) Defensive re-check: filter_unseen again after normalize (race-safety) 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 [] - unseen_ids = set(filter_unseen(req.username, ids_in_order)) - final_batch = [n for n in normalized if n.get("ig_post_id") in unseen_ids][:target] + final_unseen_ids = set(filter_unseen(req.username, ids_in_order)) + batch = [n for n in normalized if n.get("ig_post_id") in final_unseen_ids][:target] - # Persist cursor progress AFTER we’ve walked pages - set_cursor(db, req.username, final_cursor) - - # Mark as seen now to avoid future reprocessing - mark_seen(req.username, [b["ig_post_id"] for b in final_batch]) + # Mark as seen now to avoid reprocessing in future runs + mark_seen(req.username, [b["ig_post_id"] for b in batch]) # 4) Ensure account row account = db.query(InstagramAccount).filter_by(username=req.username).first() @@ -176,7 +162,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: # 5) Upload + save each post in the final batch result: List[ScrapedPost] = [] - for item in final_batch: + for item in batch: post_id = str(item["ig_post_id"]) seller_id = str(req.seller_id)