enought

scraper.py

@@ -1,112 +1,149 @@
 # scraper.py
 
-# --- imports ---
+import os
 from sqlalchemy.orm import Session
 from database import InstagramAccount, InstagramPost
-from utils.boxapi_client import get_media_page, BoxAPIError  # paged client (12 per request, next_max_id)
+from utils.boxapi_client import get_media_page, BoxAPIError
 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
 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
+from typing import List, Dict, Any, Optional, Set
 from datetime import datetime
-from scraper_cursor import get_cursor, set_cursor  # cursor helpers
+from scraper_cursor import get_cursor, set_cursor
 
+DIAG = os.getenv("SCRAPER_DIAGNOSTICS", "0") == "1"
+MAX_PAGES = int(os.getenv("SCRAPER_MAX_PAGES", "200"))  # hard safety
 
-# ----------------------------
+def dprint(msg: str) -> None:
-# Minimal helpers (fast path)
+    if DIAG:
-# ----------------------------
+        print(f"[SCRAPER] {msg}")
-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."""
+def _ig_id_of(it: Dict[str, Any]) -> Optional[str]:
-    ig_id = item.get("id") or item.get("pk")
+    v = it.get("id") or it.get("pk")
-    if ig_id is not None:
+    return str(v) if v is not None else None
-        ig_id = str(ig_id)
+
-    taken_at = item.get("taken_at") or item.get("timestamp")
+def _taken_at_of(it: Dict[str, Any]) -> Optional[int]:
+    v = it.get("taken_at") or it.get("timestamp")
     try:
-        taken_at = int(taken_at) if taken_at is not None else None
+        return int(v) if v is not None else None
     except Exception:
-        taken_at = None
+        return None
-    return ig_id, taken_at
 
+def _db_existing_ids(db: Session, ids: List[str]) -> Set[str]:
+    if not ids:
+        return set()
+    rows = (
+        db.query(InstagramPost.ig_post_id)
+        .filter(InstagramPost.ig_post_id.in_(ids))
+        .all()
+    )
+    return {r[0] for r in rows}
 
-def _collect_unseen(
+def _ordered_unseen_ids(db: Session, username: str, ids_in_order: List[str]) -> List[str]:
+    """
+    Combine Redis fast-path + DB guard to get truly unseen ids in original order.
+    This eliminates mismatch issues between Redis key format and DB.
+    """
+    # Redis fast path
+    redis_unseen = set(filter_unseen(username, ids_in_order))
+    if not redis_unseen:
+        return []
+
+    # DB guard (in case posts were stored by another worker/process)
+    already_in_db = _db_existing_ids(db, ids_in_order)
+    truly_unseen = redis_unseen.difference(already_in_db)
+
+    # Preserve original order
+    return [pid for pid in ids_in_order if pid in truly_unseen]
+
+def _collect_unseen_raw(
+    db: Session,
     username: str,
     target: int,
     stored_cursor: Optional[str],
-    since_ts: Optional[int] = None,
+    since_ts: Optional[int],
-) -> tuple[List[Dict[str, Any]], Optional[str]]:
+) -> tuple[List[Dict[str, Any]], Optional[str], int, int]:
     """
-    Keep fetching BoxAPI pages until we gather `target` unseen posts or run out of pages.
+    Page through BoxAPI until we collect `target` unseen posts or run out of pages.
-    - Always skip already-seen (via filter_unseen)
+    Returns: (raw_unseen_items, final_cursor, pages_walked, unseen_count)
-    - Optionally skip by since_ts
-    Returns: (raw_unseen_items, final_cursor)
     """
     raw_unseen: List[Dict[str, Any]] = []
-    cursor = stored_cursor
+    pages_walked = 0
-    first_page_done = False
+    final_cursor = stored_cursor  # default to stored; will update when we traverse
+    first = True
+    cursor = stored_cursor  # For first fetch we will use None (newest) unless you want pure backfill.
 
-    while len(raw_unseen) < target:
+    while len(raw_unseen) < target and pages_walked < MAX_PAGES:
-        # newest first, then older via next_max_id/cursor
+        max_id = None if first else cursor  # newest first, then backfill using 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:
+        except BoxAPIError as e:
-            # Stop on BoxAPI error; return what we have and current cursor
+            dprint(f"BoxAPI error on page fetch: {e}")
             break
 
-        first_page_done = True
+        pages_walked += 1
-        cursor = next_id  # advance our traversal cursor even if page empty
+        dprint(f"Page#{pages_walked} fetch → max_id={max_id} → next_max_id={next_id}, raw_count={len(page)}")
+
+        # Move traversal cursor forward only if we actually fetched a page
+        cursor = next_id
+        if page:
+            final_cursor = next_id
 
         if not page:
             break
 
-        # Optional: filter by timestamp before unseen check
+        # Apply since_ts BEFORE unseen checks
-        filtered_page: List[Dict[str, Any]] = []
+        filtered = []
         ids_in_order: List[str] = []
         for it in page:
-            ig_id, taken_at = _extract_minimal(it)
+            igid = _ig_id_of(it)
-            if not ig_id:
+            if not igid:
                 continue
-            if since_ts is not None and taken_at is not None and taken_at <= since_ts:
+            ts = _taken_at_of(it)
+            if since_ts is not None and ts is not None and ts <= since_ts:
                 continue
-            filtered_page.append(it)
+            filtered.append(it)
-            ids_in_order.append(ig_id)
+            ids_in_order.append(igid)
 
         if not ids_in_order:
+            dprint("  after since_ts filter: 0 items")
             if not cursor:
                 break
+            first = False
             continue
 
-        # Fast unseen check using Redis/seen store
+        # Combined unseen check (Redis + DB)
-        unseen_ids_set = set(filter_unseen(username, ids_in_order))
+        ordered_unseen_ids = _ordered_unseen_ids(db, username, ids_in_order)
-        if unseen_ids_set:
+        dprint(f"  ids_in_order={len(ids_in_order)} unseen_after_redis+db={len(ordered_unseen_ids)}")
-            for it in filtered_page:
+
-                ig_id, _ = _extract_minimal(it)
+        if ordered_unseen_ids:
-                if ig_id and ig_id in unseen_ids_set:
+            unseen_set = set(ordered_unseen_ids)
+            for it in filtered:
+                igid = _ig_id_of(it)
+                if igid in unseen_set:
                     raw_unseen.append(it)
                     if len(raw_unseen) >= target:
                         break
 
-        # If no next page, stop
         if not cursor:
             break
 
-    return raw_unseen, cursor
+        first = False
 
+    dprint(f"Collected unseen={len(raw_unseen)} across pages={pages_walked}, final_cursor={final_cursor}")
+    return raw_unseen, final_cursor, pages_walked, len(raw_unseen)
 
-# ----------------------------
-# Main entry
-# ----------------------------
 def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
     target = int(req.max_count)
 
-    # Optional since_taken_at support
+    # Parse optional since_taken_at
     since_ts: Optional[int] = None
     if hasattr(req, "since_taken_at") and req.since_taken_at is not None:
         try:
@@ -114,37 +151,35 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
         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:
+    # 1) Collect ONLY unseen raw items (newest page first, then older via cursor)
-        raw_unseen, final_cursor = _collect_unseen(
+    raw_unseen, final_cursor, pages_walked, unseen_found = _collect_unseen_raw(
+        db=db,
         username=req.username,
         target=target,
         stored_cursor=stored_cursor,
         since_ts=since_ts,
     )
-    except BoxAPIError as e:
-        print(f"❌ BoxAPI error: {e}")
-        return []
 
-    # Persist traversal progress even if nothing new (so next run resumes deeper)
+    # Persist traversal progress only if we actually walked at least one page
+    # (This avoids accidentally overwriting a good cursor when nothing fetched.)
+    if pages_walked > 0:
         set_cursor(db, req.username, final_cursor)
 
     if not raw_unseen:
+        dprint("No unseen items collected; returning early.")
         return []
 
-    # 2) Normalize ONLY the selected unseen raw items
+    # 2) Normalize only selected unseen raw items
     normalized = parse_boxapi_items_fast(raw_unseen)
 
-    # 3) Defensive re-check: filter_unseen again after normalize (race-safety)
+    # 3) Defensive unseen re-check (race safety) AFTER normalize
     ids_in_order = [n.get("ig_post_id") for n in normalized if n.get("ig_post_id")]
-    if not ids_in_order:
+    final_unseen_ids = _ordered_unseen_ids(db, req.username, ids_in_order)
-        return []
+    batch = [n for n in normalized if n.get("ig_post_id") in set(final_unseen_ids)][:target]
 
-    final_unseen_ids = set(filter_unseen(req.username, ids_in_order))
+    # Mark seen to prevent reprocessing
-    batch = [n for n in normalized if n.get("ig_post_id") in final_unseen_ids][:target]
-
-    # 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
@@ -160,30 +195,25 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
         db.commit()
         db.refresh(account)
 
-    # 5) Upload + save each post in the final batch
+    # 5) Upload + save each post
     result: List[ScrapedPost] = []
     for item in batch:
         post_id = str(item["ig_post_id"])
         seller_id = str(req.seller_id)
 
-        # Skip if already saved (DB guard)
+        # DB guard
         if db.query(InstagramPost).filter_by(ig_post_id=post_id).first():
             continue
 
-        # Media URLs (from normalized)
         media_urls = list(item.get("remote_urls", []))
 
-        # 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
         uploaded_urls: List[str] = []
         for remote_url in media_urls:
             try:
@@ -201,7 +231,6 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
                 print(f"⚠️ Error uploading media: {e}")
                 continue
 
-        # Upload/optimize thumbnail (if present)
         final_thumbnail = item.get("thumbnail_url") or None
         if final_thumbnail:
             try:
@@ -215,7 +244,6 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
             except Exception as e:
                 print(f"⚠️ Failed to upload thumbnail: {e}")
 
-        # 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,
@@ -241,4 +269,5 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
         if len(result) >= target:
             break
 
+    dprint(f"Returning {len(result)} items (requested {target}).")
     return result