diff --git a/scraper.py b/scraper.py index 470c926..c96bedd 100644 --- a/scraper.py +++ b/scraper.py @@ -1,8 +1,12 @@ # scraper.py - import os +from typing import List, Dict, Any, Optional, Tuple, Set +from datetime import datetime + from sqlalchemy.orm import Session -from database import InstagramAccount, InstagramPost +from sqlalchemy import select + +from database import InstagramAccount, InstagramPost, InstaCursor from utils.boxapi_client import get_media_page, BoxAPIError from utils.parser_fast import parse_boxapi_items_fast from utils.s3_uploader import ( @@ -11,144 +15,81 @@ from utils.s3_uploader import ( upload_video_from_url_optimized, ) from schemas import ScrapeRequest, ScrapedPost -from typing import List, Dict, Any, Optional, Tuple, 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 +PER_REQ = 12 # BoxAPI cap + def dprint(msg: str) -> None: if DIAG: print(f"[SCRAPER] {msg}") + +# ------------------------- cursor helpers (persisted) ------------------------- + +def _get_cursor(db: Session, username: str) -> Optional[str]: + row = db.execute( + select(InstaCursor).where(InstaCursor.username == username) + ).scalars().first() + return row.next_max_id if row else None + + +def _set_cursor(db: Session, username: str, next_max_id: Optional[str]) -> None: + if next_max_id is None: + return + row = db.execute( + select(InstaCursor).where(InstaCursor.username == username) + ).scalars().first() + if row: + row.next_max_id = next_max_id + else: + row = InstaCursor(username=username, next_max_id=next_max_id) + db.add(row) + db.commit() + + +# ------------------------------ small utilities ------------------------------ + def _resolve_target(req: ScrapeRequest) -> int: # accept both "count" and "max_count" return int((getattr(req, "count", None) if getattr(req, "count", None) is not None else req.max_count)) -def _db_existing_ids(db: Session, ig_ids: List[str]) -> Set[str]: - if not ig_ids: - return set() - rows = ( - db.query(InstagramPost.ig_post_id) - .filter(InstagramPost.ig_post_id.in_(ig_ids)) - .all() - ) - return {r[0] for r in rows} - -def _load_posts_by_ids_ordered(db: Session, account_id, ig_ids: List[str]) -> List[InstagramPost]: - """Fetch already-saved posts for this account, preserving ig_ids order.""" - if not ig_ids: - return [] - rows = ( - db.query(InstagramPost) - .filter(InstagramPost.account_id == account_id, InstagramPost.ig_post_id.in_(ig_ids)) - .all() - ) - by_id = {r.ig_post_id: r for r in rows} - return [by_id[i] for i in ig_ids if i in by_id] def _normalize_page(page: List[Dict[str, Any]]) -> List[Dict[str, Any]]: if not page: return [] return parse_boxapi_items_fast(page) -def _collect_normalized_pages( - db: Session, - username: str, - target: int, - start_cursor: Optional[str], - since_ts: Optional[int], -) -> Tuple[List[Dict[str, Any]], List[str], Optional[str], int]: - """ - Collect pages while counting ONLY *new* posts toward `target`. - Keep dup IDs to optionally top-up from DB after paging. - Returns: - normalized_new -> normalized items not in DB (in order found) - dup_ids_ordered -> ig_post_id list already in DB (in order found) - final_cursor -> last next_max_id seen (persist to continue older) - pages_walked -> how many API pages we pulled - """ - normalized_new: List[Dict[str, Any]] = [] - dup_ids_ordered: List[str] = [] - pages = 0 - # Start from stored cursor if present to keep drilling older across runs - max_id = start_cursor - newest_cursor = None - final_cursor: Optional[str] = None - seen_cursors: Set[Optional[str]] = set() +def _db_existing_ids(db: Session, ig_ids: List[str]) -> Set[str]: + if not ig_ids: + return set() + rows = db.query(InstagramPost.ig_post_id).filter(InstagramPost.ig_post_id.in_(ig_ids)).all() + return {r[0] for r in rows} - while len(normalized_new) < target and pages < MAX_PAGES: - per_req = 12 # BoxAPI cap per request - try: - page, _log, nxt = get_media_page(username=username, count=per_req, 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={nxt!r}") - if newest_cursor is None: - newest_cursor = nxt +def _get_or_create_account(db: Session, username: str, seller_id) -> InstagramAccount: + acc = db.query(InstagramAccount).filter_by(username=username).first() + if acc: + return acc + acc = InstagramAccount(username=username, seller_id=seller_id, is_active=True, last_synced=None) + db.add(acc) + db.commit() + db.refresh(acc) + return acc - if not page: - dprint("No items → stopping.") - break - - norm = _normalize_page(page) - - # optional time filter - if since_ts is not None: - try: - s = int(since_ts) - norm = [n for n in norm if (n.get("taken_at") or 0) > s] - except Exception: - pass - - ids = [str(n.get("ig_post_id")) for n in norm if n.get("ig_post_id")] - already = _db_existing_ids(db, ids) - new_here = [n for n in norm if n.get("ig_post_id") and str(n["ig_post_id"]) not in already] - dups_here = [str(n["ig_post_id"]) for n in norm if n.get("ig_post_id") and str(n["ig_post_id"]) in already] - - # Append in discovery order - normalized_new.extend(new_here) - dup_ids_ordered.extend(dups_here) - - dprint( - f" normalized={len(norm)} new={len(new_here)} dups={len(dups_here)} " - f"new_collected_total={len(normalized_new)}/{target}" - ) - - # Stop only when we have enough NEW items - if len(normalized_new) >= target: - final_cursor = nxt # where to resume next time (older side) - break - - # Advance to older page; if none, we're done - if not nxt: - dprint("No next_max_id → end of feed") - final_cursor = None - break - - # Loop guard: avoid cursor cycles - if nxt in seen_cursors: - dprint("Repeating cursor detected → stopping.") - final_cursor = nxt - break - - seen_cursors.add(nxt) - final_cursor = nxt - max_id = nxt # CRITICAL: go older - - return normalized_new, dup_ids_ordered, final_cursor, pages +# --------------------------------- MAIN -------------------------------------- def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: + username = req.username target = _resolve_target(req) - dprint(f"requested count={getattr(req,'count',None)} max_count={req.max_count} → resolved target={target}") + seller_id = str(req.seller_id) - # since_taken_at (optional) + dprint(f"requested count={getattr(req,'count',None)} max_count={req.max_count} → target={target}") + + # Optional time filter since_ts: Optional[int] = None if hasattr(req, "since_taken_at") and req.since_taken_at is not None: try: @@ -156,52 +97,106 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: except Exception: since_ts = None - # Ensure account row (needed for DB top-up) - 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) + # Account (needed for saving) + account = _get_or_create_account(db, username, req.seller_id) - # Start from stored cursor (for deeper backfill) - start_cursor = get_cursor(db, req.username) - dprint(f"start_cursor={start_cursor!r}") + # Cursor: start where we left off (older pages) + max_id: Optional[str] = _get_cursor(db, username) + dprint(f"start_cursor={max_id!r}") - # 1) Collect normalized candidates across pages - normalized_new, dup_ids_ordered, final_cursor, pages_walked = _collect_normalized_pages( - db=db, - username=req.username, - target=target, - start_cursor=start_cursor, - since_ts=since_ts, - ) + # Page through BoxAPI until we collect enough NEW items (older than what we have) + pages = 0 + final_cursor: Optional[str] = None + seen_cursors: Set[Optional[str]] = set() + if max_id is not None: + seen_cursors.add(max_id) # guard against cycles - # Persist backfill cursor only if we actually traversed older pages - if pages_walked > 0 and final_cursor is not None: - set_cursor(db, req.username, final_cursor) + normalized_new: List[Dict[str, Any]] = [] + try: + while len(normalized_new) < target and pages < MAX_PAGES: + pages += 1 + page, _log, nxt = get_media_page(username=username, count=PER_REQ, max_id=max_id) + dprint(f"[PAGE]#{pages}: request max_id={max_id!r} → got {len(page)} items, next_max_id={nxt!r}") + + if not page: + dprint("No items → stopping.") + break + + norm = _normalize_page(page) + + # optional time filter + if since_ts is not None: + try: + s = int(since_ts) + norm = [n for n in norm if (n.get("taken_at") or 0) > s] + except Exception: + pass + + ids = [str(n.get("ig_post_id")) for n in norm if n.get("ig_post_id")] + already = _db_existing_ids(db, ids) + new_here = [n for n in norm if n.get("ig_post_id") and str(n["ig_post_id"]) not in already] + + normalized_new.extend(new_here) + + dprint( + f" normalized={len(norm)} new_here={len(new_here)} " + f"new_total={len(normalized_new)}/{target}" + ) + + if len(normalized_new) >= target: + final_cursor = nxt # where to resume next time (older side) + break + + if not nxt: + dprint("No next_max_id → end of feed.") + final_cursor = None + break + + if nxt in seen_cursors: + dprint("Repeating cursor detected → stopping.") + final_cursor = nxt + break + + seen_cursors.add(nxt) + final_cursor = nxt + max_id = nxt # go older + + except BoxAPIError as e: + print(f"❌ BoxAPI error: {e}") + # continue with whatever we gathered + + # Persist cursor only if we actually moved + if pages > 0 and final_cursor: + _set_cursor(db, username, final_cursor) + + if not normalized_new: + dprint("No NEW posts found.") + return [] + + # Save up to `target` new posts (upload media, insert rows) result: List[ScrapedPost] = [] - - # 2) First, save NEW posts (upload + insert) until we reach target or run out - for item in normalized_new: - if len(result) >= target: - break - + for item in normalized_new[:target]: post_id = str(item["ig_post_id"]) - # Skip if somehow inserted by a parallel worker - if db.query(InstagramPost).filter_by(ig_post_id=post_id).first(): + + # Race guard: if another worker already saved it, return from DB so we still reach N + existing = db.query(InstagramPost).filter_by(ig_post_id=post_id).first() + if existing: + result.append(ScrapedPost( + ig_post_id=existing.ig_post_id, + media_type=existing.media_type, + caption=existing.caption, + thumbnail_url=existing.thumbnail_url, + local_media=existing.local_media or [], + created_at=existing.created_at, + )) + if len(result) >= target: + break continue - seller_id = str(req.seller_id) + # Upload media media_urls = list(item.get("remote_urls", [])) - # If video, keep only video URLs (avoid thumb) if item["media_type"] == "video": thumb = (item.get("thumbnail_url") or "").strip() def is_video(u: str) -> bool: @@ -216,7 +211,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: 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: + if not url: url = upload_media_from_url(remote_url, seller_id, post_id) else: url = upload_media_from_url(remote_url, seller_id, post_id) @@ -226,7 +221,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: print(f"⚠️ Error uploading media: {e}") continue - created_at = datetime.fromtimestamp(item["taken_at"]) if item.get("taken_at") else datetime.utcnow() + # Thumbnail final_thumbnail = item.get("thumbnail_url") or None if final_thumbnail: try: @@ -240,6 +235,8 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: 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() + row = InstagramPost( ig_post_id=post_id, account_id=account.id, @@ -261,27 +258,8 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: created_at=created_at, )) - # 3) If we still need more, TOP-UP with already-saved posts from DB (NO uploads) - if len(result) < target and dup_ids_ordered: - missing = target - len(result) - # Load the dup rows in the same order as we saw them while paging - dup_rows = _load_posts_by_ids_ordered(db, account.id, dup_ids_ordered) - for row in dup_rows: - if missing <= 0: - break - # Build response directly from DB (fast) - result.append(ScrapedPost( - ig_post_id=row.ig_post_id, - media_type=row.media_type, - caption=row.caption, - thumbnail_url=row.thumbnail_url, - local_media=row.local_media or [], - created_at=row.created_at, - )) - missing -= 1 - - # 4) If STILL not enough (end of feed + tiny DB), just return what we have - if len(result) < target: - dprint(f"feed_exhausted_or_insufficient_total=True returned={len(result)} requested={target}") + if len(result) >= target: + break + dprint(f"DONE username={username} returned={len(result)} requested={target} pages={pages} last_cursor={final_cursor!r}") return result diff --git a/utils/boxapi_client.py b/utils/boxapi_client.py index 3a3e0d3..4562470 100644 --- a/utils/boxapi_client.py +++ b/utils/boxapi_client.py @@ -2,22 +2,21 @@ import os import requests from typing import Any, Dict, List, Tuple, Optional -from utils.boxapi_log import log_boxapi_response +from utils.boxapi_log import log_boxapi_response # keep your existing logger BOXAPI_USERNAME = os.environ.get("BOXAPI_USERNAME", "") BOXAPI_PASSWORD = os.environ.get("BOXAPI_PASSWORD", "") -# Default matches /get_media_by_username; override in .env if needed BOXAPI_ENDPOINT = os.environ.get( "BOXAPI_ENDPOINT", "https://boxapi.ir/api/instagram/user/get_media_by_username", ) - -# NEW: simple debug flag BOXAPI_DEBUG = os.getenv("BOXAPI_DEBUG", "0") == "1" + class BoxAPIError(Exception): pass + def _dprint(msg: str) -> None: if BOXAPI_DEBUG: print(f"[BOXAPI] {msg}") @@ -25,22 +24,17 @@ def _dprint(msg: str) -> None: def get_media_page( username: str, - count: int = 10, + count: int = 12, max_id: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], str, Optional[str]]: """ - Perform a single BoxAPI request and return a page of results. + Fetch ONE page from BoxAPI. Returns: (items, log_path, next_max_id) - Notes: - - BoxAPI returns at most 12 items per request. - - Pass `max_id` to fetch older pages (use the previous page's next_max_id). """ if not BOXAPI_USERNAME or not BOXAPI_PASSWORD: raise BoxAPIError("Missing BOXAPI_USERNAME/BOXAPI_PASSWORD in environment.") - # BoxAPI hard cap per request - per_req = max(1, min(int(count or 1), 12)) - + per_req = max(1, min(int(count or 1), 12)) # BoxAPI cap per request payload: Dict[str, Any] = {"username": username, "count": per_req} if max_id: payload["max_id"] = max_id @@ -61,17 +55,18 @@ def get_media_page( status_code = resp.status_code resp_text = resp.text - # Try to parse JSON; if it fails, log and raise try: resp_json = resp.json() except Exception: + # Non-JSON – log and raise log_path = log_boxapi_response( username, payload, status_code, response_text=resp_text, note="Non-JSON response (error)" ) raise BoxAPIError(f"BoxAPI non-JSON response. See log: {log_path}") - # Expected shape: status -> response -> status_code + body.items + body.next_max_id + # Expected shape: + # {"status":"done", "response":{"status_code":200, "body":{"items":[...], "next_max_id":"..."}}} top_status = resp_json.get("status") inner = resp_json.get("response") or {} inner_code = inner.get("status_code") @@ -84,19 +79,16 @@ def get_media_page( ) if top_status != "done" or inner_code != 200 or not isinstance(items, list): - snapshot = { + snap = { "top_status": top_status, "inner_code": inner_code, "has_body": isinstance(body, dict), "items_type": str(type(items)), "items_len": len(items) if isinstance(items, list) else None, } - raise BoxAPIError( - f"Unexpected response format from BoxAPI ({snapshot}). See log: {log_path}" - ) + raise BoxAPIError(f"Unexpected response format from BoxAPI ({snap}). See log: {log_path}") _dprint(f"← items={len(items)} next_max_id={next_max_id!r}") - return items, log_path, next_max_id except BoxAPIError: @@ -110,9 +102,9 @@ def get_media_page( raise BoxAPIError(f"BoxAPI request failed ({e}). See log: {log_path}") -def get_media(username: str, count: int = 10) -> Tuple[List[Dict[str, Any]], str]: +def get_media(username: str, count: int = 12) -> Tuple[List[Dict[str, Any]], str]: """ - Backward-compatible wrapper that fetches ONE page (latest) without pagination. + Backward-compatible helper: fetch ONE latest page without pagination. Returns: (items, log_path) """ items, log_path, _next = get_media_page(username=username, count=count, max_id=None) @@ -122,7 +114,6 @@ def get_media(username: str, count: int = 10) -> Tuple[List[Dict[str, Any]], str def fetch_boxapi_posts(username: str, count: int = 30) -> List[Dict[str, Any]]: """ Backward-compatible helper that PAGES until it collects `count` items (or exhausts). - Uses get_media_page(...) under the hood (12-per-request windows). """ try: target = max(1, int(count)) @@ -135,7 +126,7 @@ def fetch_boxapi_posts(username: str, count: int = 30) -> List[Dict[str, Any]]: if not items: break collected.extend(items) - if not next_id: # no more pages + if not next_id: break return collected[:target]