khastam

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,86 +15,109 @@ 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
+def _db_existing_ids(db: Session, ig_ids: List[str]) -> Set[str]:
-    max_id = start_cursor
+    if not ig_ids:
-    newest_cursor = None
+        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 _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
+
+
+# --------------------------------- MAIN --------------------------------------
+
+def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
+    username = req.username
+    target = _resolve_target(req)
+    seller_id = str(req.seller_id)
+
+    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:
+            since_ts = int(req.since_taken_at)
+        except Exception:
+            since_ts = None
+
+    # Account (needed for saving)
+    account = _get_or_create_account(db, username, req.seller_id)
+
+    # Cursor: start where we left off (older pages)
+    max_id: Optional[str] = _get_cursor(db, username)
+    dprint(f"start_cursor={max_id!r}")
+
+    # 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
+
+    normalized_new: List[Dict[str, Any]] = []
 
-    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)
+        while len(normalized_new) < target and pages < MAX_PAGES:
-        except BoxAPIError as e:
-            dprint(f"BoxAPI error: {e}")
-            break
-
             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 newest_cursor is None:
-            newest_cursor = nxt
 
             if not page:
                 dprint("No items → stopping.")
@@ -109,29 +136,23 @@ def _collect_normalized_pages(
             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"    normalized={len(norm)} new_here={len(new_here)} "
-            f"new_collected_total={len(normalized_new)}/{target}"
+                f"new_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")
+                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
@@ -139,69 +160,43 @@ def _collect_normalized_pages(
 
             seen_cursors.add(nxt)
             final_cursor = nxt
-        max_id = nxt  # CRITICAL: go older
+            max_id = nxt  # go older
 
-    return normalized_new, dup_ids_ordered, final_cursor, pages
+    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)
 
-def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
+    if not normalized_new:
-    target = _resolve_target(req)
+        dprint("No NEW posts found.")
-    dprint(f"requested count={getattr(req,'count',None)} max_count={req.max_count} → resolved target={target}")
+        return []
-
-    # 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
-
-    # 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)
-
-    # Start from stored cursor (for deeper backfill)
-    start_cursor = get_cursor(db, req.username)
-    dprint(f"start_cursor={start_cursor!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,
-    )
-
-    # 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)
 
+    # Save up to `target` new posts (upload media, insert rows)
     result: List[ScrapedPost] = []
+    for item in normalized_new[:target]:
+        post_id = str(item["ig_post_id"])
 
-    # 2) First, save NEW posts (upload + insert) until we reach target or run out
+        # Race guard: if another worker already saved it, return from DB so we still reach N
-    for item in normalized_new:
+        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
-
-        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():
             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:
-    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}")
 
+    dprint(f"DONE username={username} returned={len(result)} requested={target} pages={pages} last_cursor={final_cursor!r}")
     return result

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))  # BoxAPI cap per request
-    per_req = max(1, min(int(count or 1), 12))
-
     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(
+            raise BoxAPIError(f"Unexpected response format from BoxAPI ({snap}). See log: {log_path}")
-                f"Unexpected response format from BoxAPI ({snapshot}). 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]