khastam

docker-compose.yml

@@ -16,6 +16,9 @@ services:
       - DATABASE_URL=${DATABASE_URL}
       - FFMPEG_BIN=/usr/bin/ffmpeg
       - FFPROBE_BIN=/usr/bin/ffprobe
+      - BOXAPI_DEBUG=1
+      - SCRAPER_DIAGNOSTICS=1
+      - SCRAPER_MAX_PAGES=400
     restart: unless-stopped
     depends_on:
       postgres:

schemas.py

@@ -6,7 +6,10 @@ from datetime import datetime
 class ScrapeRequest(BaseModel):
     username: str
     seller_id: UUID
+    # existing field
     max_count: int = Field(default=5, ge=1, le=50)
+    # NEW: accept "count" from older/other callers; ignored if not present
+    count: Optional[int] = Field(default=None, ge=1, le=50)
 
 class ScrapedPost(BaseModel):
     ig_post_id: str

scraper.py

@@ -1,112 +1,95 @@
 # scraper.py
+import os
+from typing import List, Dict, Any, Optional, Tuple, Set
+from datetime import datetime
 
-# --- imports ---
 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 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.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 datetime import datetime
-from scraper_cursor import get_cursor, set_cursor  # cursor helpers
+
+DIAG = os.getenv("SCRAPER_DIAGNOSTICS", "0") == "1"
+MAX_PAGES = int(os.getenv("SCRAPER_MAX_PAGES", "400"))  # safety cap
+PER_REQ = 12  # BoxAPI cap
 
 
-# ----------------------------
-# 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")
-    try:
-        taken_at = int(taken_at) if taken_at is not None else None
-    except Exception:
-        taken_at = None
-    return ig_id, taken_at
+def dprint(msg: str) -> None:
+    if DIAG:
+        print(f"[SCRAPER] {msg}")
 
 
-def _collect_unseen(
-    username: str,
-    target: int,
-    stored_cursor: Optional[str],
-    since_ts: Optional[int] = None,
-) -> tuple[List[Dict[str, Any]], Optional[str]]:
-    """
-    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)
-    """
-    raw_unseen: List[Dict[str, Any]] = []
-    cursor = stored_cursor
-    first_page_done = False
+# ------------------------- cursor helpers (persisted) -------------------------
 
-    while len(raw_unseen) < target:
-        # newest first, then older via next_max_id/cursor
-        max_id = None if not first_page_done else 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
-            break
+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
 
-        first_page_done = True
-        cursor = next_id  # advance our traversal cursor even if page empty
 
+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 _normalize_page(page: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
     if not page:
-            break
-
-        # Optional: filter by timestamp before unseen check
-        filtered_page: List[Dict[str, Any]] = []
-        ids_in_order: List[str] = []
-        for it in page:
-            ig_id, taken_at = _extract_minimal(it)
-            if not ig_id:
-                continue
-            if since_ts is not None and taken_at is not None and taken_at <= since_ts:
-                continue
-            filtered_page.append(it)
-            ids_in_order.append(ig_id)
-
-        if not ids_in_order:
-            if not cursor:
-                break
-            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:
-                    raw_unseen.append(it)
-                    if len(raw_unseen) >= target:
-                        break
-
-        # If no next page, stop
-        if not cursor:
-            break
-
-    return raw_unseen, cursor
+        return []
+    return parse_boxapi_items_fast(page)
 
 
-# ----------------------------
-# Main entry
-# ----------------------------
+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 _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]:
-    target = int(req.max_count)
+    username = req.username
+    target = _resolve_target(req)
+    seller_id = str(req.seller_id)
 
-    # Optional since_taken_at support
+    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:
@@ -114,76 +97,113 @@ 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
+    # 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]] = []
+
     try:
-        raw_unseen, final_cursor = _collect_unseen(
-            username=req.username,
-            target=target,
-            stored_cursor=stored_cursor,
-            since_ts=since_ts,
+        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 []
 
-    # Persist traversal progress even if nothing new (so next run resumes deeper)
-    set_cursor(db, req.username, final_cursor)
-
-    if not raw_unseen:
-        return []
-
-    # 2) Normalize ONLY the selected unseen raw items
-    normalized = parse_boxapi_items_fast(raw_unseen)
-
-    # 3) Defensive re-check: filter_unseen again after normalize (race-safety)
-    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 = 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(req.username, [b["ig_post_id"] for b in batch])
-
-    # 4) 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)
-
-    # 5) Upload + save each post in the final batch
+    # Save up to `target` new posts (upload media, insert rows)
     result: List[ScrapedPost] = []
-    for item in batch:
+    for item in normalized_new[:target]:
         post_id = str(item["ig_post_id"])
-        seller_id = str(req.seller_id)
 
-        # Skip if already saved (DB guard)
-        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
 
-        # Media URLs (from normalized)
+        # Upload media
         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:
@@ -191,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)
@@ -201,7 +221,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
                 print(f"⚠️ Error uploading media: {e}")
                 continue
 
-        # Upload/optimize thumbnail (if present)
+        # Thumbnail
         final_thumbnail = item.get("thumbnail_url") or None
         if final_thumbnail:
             try:
@@ -215,9 +235,9 @@ 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(
+
+        row = InstagramPost(
             ig_post_id=post_id,
             account_id=account.id,
             media_type=item["media_type"],
@@ -226,7 +246,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
             local_media=uploaded_urls,
             created_at=created_at,
         )
-        db.add(post_entry)
+        db.add(row)
         db.commit()
 
         result.append(ScrapedPost(
@@ -241,4 +261,5 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
         if len(result) >= target:
             break
 
+    dprint(f"DONE username={username} returned={len(result)} requested={target} pages={pages} last_cursor={final_cursor!r}")
     return result

utils/boxapi_client.py

@@ -2,42 +2,45 @@
 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",
 )
+BOXAPI_DEBUG = os.getenv("BOXAPI_DEBUG", "0") == "1"
+
 
 class BoxAPIError(Exception):
     pass
 
 
+def _dprint(msg: str) -> None:
+    if BOXAPI_DEBUG:
+        print(f"[BOXAPI] {msg}")
+
+
 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
 
+    _dprint(f"→ POST {BOXAPI_ENDPOINT} payload={payload}")
+
     resp_json: Optional[Dict[str, Any]] = None
     resp_text: Optional[str] = None
     status_code: Optional[int] = None
@@ -52,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")
@@ -75,17 +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:
@@ -99,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)
@@ -111,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))
@@ -124,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]