# 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.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, 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", "400")) # safety cap def dprint(msg: str) -> None: if DIAG: print(f"[SCRAPER] {msg}") def _ig_id(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 _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_raw_not_in_db( db: Session, username: str, needed: int, since_ts: Optional[int], ) -> tuple[List[Dict[str, Any]], Optional[str], int]: """ Walk pages linearly: newest first (max_id=None), then older via next_max_id. Keep only posts NOT in DB. Stop when we have `needed` or no more pages. Returns (raw_to_save, final_cursor, pages_walked) """ raw_to_save: List[Dict[str, Any]] = [] pages = 0 first = True max_id: Optional[str] = None # None → newest page final_cursor: Optional[str] = None while len(raw_to_save) < needed and pages < MAX_PAGES: count = min(12, max(1, needed - len(raw_to_save))) # 1..12 try: page, _log, next_max = get_media_page(username=username, count=count, max_id=max_id) except BoxAPIError as e: dprint(f"BoxAPI error: {e}") break pages += 1 dprint(f"Page#{pages}: request max_id={max_id!r} → got {len(page)} items, next_max_id={next_max!r}") # after first fetch, always continue with next_max_id first = False if not page: # no items on this page, stop if no further cursor if not next_max: dprint("Empty page and no next_max_id → stopping") break # otherwise, advance cursor and continue (rare) max_id = next_max final_cursor = next_max continue # Optional time cutoff filtered: List[Dict[str, Any]] = [] ids_ordered: List[str] = [] for it in page: pid = _ig_id(it) if not pid: 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_ordered.append(pid) if ids_ordered: already = _db_existing_ids(db, ids_ordered) dprint(f" from {len(ids_ordered)} IDs, already_in_db={len(already)}") for it in filtered: pid = _ig_id(it) if pid and pid not in already: raw_to_save.append(it) if len(raw_to_save) >= needed: break else: dprint(" no valid IDs after filtering") # advance pagination cursor for the next loop final_cursor = next_max if not next_max: dprint("No next_max_id → reached end of feed") break max_id = next_max # ← KEY: chase older page next round dprint(f"Collected to_save={len(raw_to_save)} across pages={pages}, final_cursor={final_cursor!r}") return raw_to_save, final_cursor, pages def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: # Prefer "count" if provided by the caller, otherwise fall back to "max_count" target = int((req.count if getattr(req, "count", None) is not None else req.max_count)) # Optional debug line print(f"[SCRAPER] requested count={getattr(req, 'count', None)} max_count={req.max_count} → resolved target={target}") # since_taken_at (optional) since_ts: Optional[int] = None if hasattr(req, "since_taken_at") and req.since_taken_at is not None: try: since_ts = int(req.since_taken_at) except Exception: since_ts = None # 1) BRUTE-FORCE scan: newest → older, pick posts NOT in DB until we reach target raw_to_save, final_cursor, pages_walked = _collect_raw_not_in_db( db=db, username=req.username, needed=target, since_ts=since_ts, ) # Persist cursor if we walked at least one page if pages_walked > 0: set_cursor(db, req.username, final_cursor) if not raw_to_save: dprint("No new posts to save.") return [] # 2) Normalize only the ones we intend to save normalized = parse_boxapi_items_fast(raw_to_save)[:target] # 3) Ensure account row account = db.query(InstagramAccount).filter_by(username=req.username).first() if not account: account = InstagramAccount( username=req.username, seller_id=req.seller_id, is_active=True, last_synced=None ) db.add(account) db.commit() db.refresh(account) # 4) Upload + save result: List[ScrapedPost] = [] for item in normalized: post_id = str(item["ig_post_id"]) seller_id = str(req.seller_id) # DB guard in case a parallel worker just saved it if db.query(InstagramPost).filter_by(ig_post_id=post_id).first(): continue media_urls = list(item.get("remote_urls", [])) 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] uploaded_urls: List[str] = [] for remote_url in media_urls: try: if item["media_type"] == "video": url = upload_video_from_url_optimized(remote_url, seller_id, post_id) elif item["media_type"] in ("image", "carousel"): url = upload_image_from_url_optimized(remote_url, seller_id, post_id) if url is None: url = upload_media_from_url(remote_url, seller_id, post_id) else: url = upload_media_from_url(remote_url, seller_id, post_id) if url: uploaded_urls.append(url) except Exception as e: print(f"⚠️ Error uploading media: {e}") continue final_thumbnail = item.get("thumbnail_url") or None if final_thumbnail: try: thumb_url = upload_image_from_url_optimized(final_thumbnail, seller_id, post_id) if thumb_url: final_thumbnail = thumb_url else: 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}") 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, media_type=item["media_type"], caption=item.get("caption"), thumbnail_url=final_thumbnail, local_media=uploaded_urls, created_at=created_at, ) db.add(post_entry) db.commit() result.append(ScrapedPost( ig_post_id=post_id, media_type=item["media_type"], caption=item.get("caption"), thumbnail_url=final_thumbnail, local_media=uploaded_urls, created_at=created_at, )) if len(result) >= target: break return result