From 90c28032cc12997d85dcfa95de1a4deefb426775 Mon Sep 17 00:00:00 2001 From: Hossein Date: Thu, 25 Sep 2025 02:44:14 +0330 Subject: [PATCH] ishalla fixe --- scraper.py | 157 ++++++++++++++++++++--------------------------------- 1 file changed, 58 insertions(+), 99 deletions(-) diff --git a/scraper.py b/scraper.py index 3db2ec7..a613b1f 100644 --- a/scraper.py +++ b/scraper.py @@ -1,11 +1,9 @@ # scraper.py -import os from sqlalchemy.orm import Session from database import InstagramAccount, InstagramPost 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 from utils.s3_uploader import ( upload_media_from_url, upload_image_from_url_optimized, @@ -16,23 +14,14 @@ from typing import List, Dict, Any, Optional, Set from datetime import datetime 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 +# Safety cap for very large accounts +MAX_PAGES = 500 # adjust if needed -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: - return int(v) if v is not None else None - except Exception: - return None def _db_existing_ids(db: Session, ids: List[str]) -> Set[str]: if not ids: @@ -44,106 +33,85 @@ def _db_existing_ids(db: Session, ids: List[str]) -> Set[str]: ) return {r[0] for r in rows} -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( +def _collect_raw_to_save( db: Session, username: str, - target: int, + needed: int, stored_cursor: Optional[str], since_ts: Optional[int], -) -> tuple[List[Dict[str, Any]], Optional[str], int, int]: +) -> tuple[List[Dict[str, Any]], Optional[str]]: """ - Page through BoxAPI until we collect `target` unseen posts or run out of pages. - Returns: (raw_unseen_items, final_cursor, pages_walked, unseen_count) + Keep fetching pages until we have `needed` posts that are NOT in DB. + Return (raw_items_to_save, final_cursor). """ - raw_unseen: List[Dict[str, Any]] = [] - pages_walked = 0 - final_cursor = stored_cursor # default to stored; will update when we traverse + to_save_raw: List[Dict[str, Any]] = [] + pages = 0 + cursor = stored_cursor first = True - cursor = stored_cursor # For first fetch we will use None (newest) unless you want pure backfill. - - 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))) + last_seen_cursor: Optional[str] = stored_cursor + while len(to_save_raw) < needed and pages < MAX_PAGES: + max_id = None if first else cursor + count = min(12, needed - len(to_save_raw)) or 12 try: - page, _log, next_id = get_media_page(username=username, count=need, max_id=max_id) - except BoxAPIError as e: - dprint(f"BoxAPI error on page fetch: {e}") + page, _log, next_id = get_media_page(username=username, count=count, max_id=max_id) + except BoxAPIError: break - 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 + pages += 1 + last_seen_cursor = next_id if not page: break - # Apply since_ts BEFORE unseen checks + # Optional: since_taken_at filter (raw, cheap) filtered = [] ids_in_order: List[str] = [] for it in page: igid = _ig_id_of(it) if not igid: continue - ts = _taken_at_of(it) - if since_ts is not None and ts is not None and ts <= since_ts: - 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_in_order.append(igid) if not ids_in_order: - dprint(" after since_ts filter: 0 items") - if not cursor: + if not next_id: break first = False + cursor = next_id continue - # 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)}") + # DB check: keep only items not yet saved + already = _db_existing_ids(db, ids_in_order) + for it in filtered: + igid = _ig_id_of(it) + if igid and igid not in already: + to_save_raw.append(it) + if len(to_save_raw) >= needed: + break - 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 not cursor: + if not next_id: break first = False + cursor = next_id + + return to_save_raw, last_seen_cursor - 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) def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: target = int(req.max_count) - # Parse optional since_taken_at + # Optional since_taken_at since_ts: Optional[int] = None if hasattr(req, "since_taken_at") and req.since_taken_at is not None: try: @@ -151,38 +119,28 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: except Exception: since_ts = None - stored_cursor = get_cursor(db, req.username) # may be None + # Start from stored backfill cursor (if any), but always try newest page first + stored_cursor = get_cursor(db, req.username) - # 1) Collect ONLY unseen raw items (newest page first, then older via cursor) - raw_unseen, final_cursor, pages_walked, unseen_found = _collect_unseen_raw( + # 1) Collect raw items to save (NOT in DB), paging as deep as needed + raw_to_save, final_cursor = _collect_raw_to_save( db=db, username=req.username, - target=target, + needed=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) + # Persist traversal progress (so next run can continue deeper) + set_cursor(db, req.username, final_cursor) - if not raw_unseen: - dprint("No unseen items collected; returning early.") + if not raw_to_save: return [] - # 2) Normalize only selected unseen raw items - normalized = parse_boxapi_items_fast(raw_unseen) + # 2) Normalize only what we intend to save + normalized = parse_boxapi_items_fast(raw_to_save)[:target] - # 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")] - 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] - - # Mark seen to prevent reprocessing - mark_seen(req.username, [b["ig_post_id"] for b in batch]) - - # 4) Ensure account row + # 3) Ensure account row account = db.query(InstagramAccount).filter_by(username=req.username).first() if not account: account = InstagramAccount( @@ -195,18 +153,19 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: db.commit() db.refresh(account) - # 5) Upload + save each post + # 4) Upload + save result: List[ScrapedPost] = [] - for item in batch: + for item in normalized: post_id = str(item["ig_post_id"]) seller_id = str(req.seller_id) - # DB guard + # DB guard (rare race) if db.query(InstagramPost).filter_by(ig_post_id=post_id).first(): continue media_urls = list(item.get("remote_urls", [])) + # If video, keep only actual video URLs (avoid thumbnail) if item["media_type"] == "video": thumb = (item.get("thumbnail_url") or "").strip() def is_video(u: str) -> bool: @@ -245,6 +204,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: print(f"⚠️ Failed to upload thumbnail: {e}") created_at = datetime.fromtimestamp(item["taken_at"]) if item.get("taken_at") else datetime.utcnow() + post_entry = InstagramPost( ig_post_id=post_id, account_id=account.id, @@ -269,5 +229,4 @@ 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