From 2e55f2d99e62f3c1d4a2059928ee7298600f18fb Mon Sep 17 00:00:00 2001 From: Hossein Date: Thu, 25 Sep 2025 02:37:17 +0330 Subject: [PATCH] enought --- scraper.py | 195 ++++++++++++++++++++++++++++++----------------------- 1 file changed, 112 insertions(+), 83 deletions(-) diff --git a/scraper.py b/scraper.py index 8bbfa24..3db2ec7 100644 --- a/scraper.py +++ b/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 -# ---------------------------- -# 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") +def dprint(msg: str) -> None: + if DIAG: + print(f"[SCRAPER] {msg}") + +def _ig_id_of(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 _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 ig_id, taken_at + return None +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, -) -> tuple[List[Dict[str, Any]], Optional[str]]: + since_ts: Optional[int], +) -> tuple[List[Dict[str, Any]], Optional[str], int, int]: """ - 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) + Page through BoxAPI until we collect `target` unseen posts or run out of pages. + Returns: (raw_unseen_items, final_cursor, pages_walked, unseen_count) """ raw_unseen: List[Dict[str, Any]] = [] - cursor = stored_cursor - first_page_done = False + pages_walked = 0 + 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: - # newest first, then older via next_max_id/cursor - max_id = None if not first_page_done else cursor + while len(raw_unseen) < target and pages_walked < MAX_PAGES: + max_id = None if first else cursor # newest first, then backfill using 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 + except BoxAPIError as e: + dprint(f"BoxAPI error on page fetch: {e}") break - first_page_done = True - cursor = next_id # advance our traversal cursor even if page empty + pages_walked += 1 + 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 - filtered_page: List[Dict[str, Any]] = [] + # Apply since_ts BEFORE unseen checks + filtered = [] ids_in_order: List[str] = [] for it in page: - ig_id, taken_at = _extract_minimal(it) - if not ig_id: + igid = _ig_id_of(it) + 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) - ids_in_order.append(ig_id) + filtered.append(it) + 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 - 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: + # Combined unseen check (Redis + DB) + ordered_unseen_ids = _ordered_unseen_ids(db, username, ids_in_order) + dprint(f" ids_in_order={len(ids_in_order)} unseen_after_redis+db={len(ordered_unseen_ids)}") + + if ordered_unseen_ids: + 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 on first run - try: - raw_unseen, final_cursor = _collect_unseen( - username=req.username, - target=target, - stored_cursor=stored_cursor, - since_ts=since_ts, - ) - except BoxAPIError as e: - print(f"❌ BoxAPI error: {e}") - return [] + stored_cursor = get_cursor(db, req.username) # may be None - # Persist traversal progress even if nothing new (so next run resumes deeper) - set_cursor(db, req.username, final_cursor) + # 1) Collect ONLY unseen raw items (newest page first, then older via cursor) + 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, + ) + + # 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: - return [] + final_unseen_ids = _ordered_unseen_ids(db, req.username, ids_in_order) + 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)) - 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 to prevent reprocessing 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