make it faster

README.md

@@ -37,13 +37,17 @@ A FastAPI application with Docker and docker-compose support.
    # nano .env  # or use your preferred editor
    ```
 
-2. **Configure BoxAPI (for Instagram scraping):**
+2. **Configure BoxAPI and Redis:**
    ```bash
    # Add your BoxAPI credentials to .env
    BOXAPI_USERNAME=your_boxapi_username
    BOXAPI_PASSWORD=your_boxapi_password
    BOXAPI_ENDPOINT=https://boxapi.ir/api/instagram/user/get_media_by_username
    BOXAPI_LOG_DIR=/app/logs/boxapi
+   BOXAPI_LOG_LEVEL=all  # or "error" for production
+   
+   # Redis for pagination (optional but recommended)
+   REDIS_URL=redis://localhost:6379/0
    ```
 
 2. **Build and start the application:**
@@ -163,8 +167,10 @@ FastApi-ISS/
 ├── utils/              # Utility modules
 │   ├── __init__.py
 │   ├── boxapi_client.py # BoxAPI client for Instagram scraping
-│   ├── boxapi_parser.py # BoxAPI response parser
+│   ├── parser_fast.py  # Fast BoxAPI response parser
+│   ├── boxapi_parser.py # Legacy BoxAPI response parser
 │   ├── boxapi_log.py   # BoxAPI response logging
+│   ├── seen_store.py   # Redis-based seen posts tracking
 │   ├── instagram_client.py # Legacy Instagram client
 │   └── s3_uploader.py  # S3 media upload functionality
 ├── .dockerignore       # Docker ignore file
@@ -181,6 +187,10 @@ The application uses environment variables for configuration. Copy `env.example`
 - `BOXAPI_PASSWORD`: Your BoxAPI password
 - `BOXAPI_ENDPOINT`: BoxAPI endpoint URL (default: `https://boxapi.ir/api/instagram/user/get_media_by_username`)
 - `BOXAPI_LOG_DIR`: Directory for BoxAPI response logs (default: `/app/logs/boxapi`)
+- `BOXAPI_LOG_LEVEL`: Logging level - "all", "error", or "none" (default: "all")
+
+### Redis Configuration (for pagination)
+- `REDIS_URL`: Redis connection URL (default: `redis://localhost:6379/0`)
 
 ### Database Configuration
 - `POSTGRES_DB`: Database name (default: `fastapi_db`)

requirements.txt

@@ -10,3 +10,5 @@ python-dotenv==1.0.0
 boto3==1.34.0
 requests==2.31.0
 Pillow>=8.1.1
+redis==5.0.1
+orjson==3.9.10 

schemas.py

@@ -8,6 +8,7 @@ class ScrapeRequest(BaseModel):
     username: str
     seller_id: UUID
     max_count: int = Field(default=5, ge=1, le=50)
+    since_taken_at: Optional[int] = None  # epoch seconds for cursor-like behavior
 
 
 class ScrapedPost(BaseModel):
@@ -15,5 +16,6 @@ class ScrapedPost(BaseModel):
     media_type: str  # "image", "video", or "carousel"
     caption: Optional[str]
     thumbnail_url: Optional[str]
-    local_media: List[str]  # ✅ list of S3 URLs only
-    created_at: datetime
+    remote_urls: List[str]  # remote URLs to be uploaded later
+    taken_at: int  # epoch seconds
+    code: Optional[str] = None

scraper.py

@@ -1,7 +1,8 @@
 from sqlalchemy.orm import Session
 from database import InstagramAccount, InstagramPost
 from utils.boxapi_client import get_media, BoxAPIError
-from utils.boxapi_parser import parse_boxapi_items
+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
 from schemas import ScrapeRequest, ScrapedPost
 from typing import List
@@ -10,18 +11,36 @@ from datetime import datetime
 
 def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
     try:
-        items, log_path = get_media(req.username, req.max_count)
+        # 1) one BoxAPI call (fast) — we keep this single round trip
+        # Ask for more to offset duplicates
+        raw_items, log_path = get_media(req.username, max(req.max_count, 12))
     except BoxAPIError as e:
         print(f"❌ BoxAPI error: {e}")
         return []
 
-    if not items:
+    if not raw_items:
         return []
 
-    # Parse and normalize the items
-    normalized_items = parse_boxapi_items(items)
+    # 2) normalize quickly
+    normalized = parse_boxapi_items_fast(raw_items)
 
-    # Get or create InstagramAccount
+    # 3) apply cursor filter if provided
+    if req.since_taken_at:
+        normalized = [n for n in normalized if n["taken_at"] < req.since_taken_at]
+
+    # 4) filter already-seen
+    ids_in_order = [n["ig_post_id"] for n in normalized if n.get("ig_post_id")]
+    unseen_ids = set(filter_unseen(req.username, ids_in_order))
+
+    unseen = [n for n in normalized if n["ig_post_id"] in unseen_ids]
+
+    # 5) take the next N unseen (newest first already)
+    batch = unseen[:req.max_count]
+
+    # 6) mark them seen (only what we return)
+    mark_seen(req.username, [b["ig_post_id"] for b in batch])
+
+    # 7) Get or create InstagramAccount
     account = db.query(InstagramAccount).filter_by(username=req.username).first()
     if not account:
         account = InstagramAccount(
@@ -37,13 +56,14 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
     result: List[ScrapedPost] = []
     new_post_count = 0
 
-    for item in normalized_items:
+    for item in batch:
         post_id = item["ig_post_id"]
         
+        # Check if already in database
         if db.query(InstagramPost).filter_by(ig_post_id=post_id).first():
             continue
 
-        # Upload remote URLs to S3
+        # Upload remote URLs to S3 (optional - can be done async later)
         media_urls: List[str] = []
         for remote_url in item["remote_urls"]:
             try:
@@ -54,9 +74,6 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
                 print(f"⚠️ Error uploading media: {e}")
                 continue
 
-        if not media_urls:
-            continue
-
         # Upload thumbnail (optional)
         final_thumbnail = item["thumbnail_url"]
         if item["thumbnail_url"]:
@@ -75,7 +92,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
             caption=item["caption"],
             thumbnail_url=final_thumbnail,
             local_media=media_urls,
-            created_at=item["taken_at"],
+            created_at=datetime.fromtimestamp(item["taken_at"]),
         )
         db.add(post_entry)
         db.commit()
@@ -85,8 +102,9 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
             media_type=item["media_type"],
             caption=item["caption"],
             thumbnail_url=final_thumbnail,
-            local_media=media_urls,
-            created_at=item["taken_at"]
+            remote_urls=item["remote_urls"],  # Return remote URLs for async processing
+            taken_at=item["taken_at"],
+            code=item.get("code")
         ))
 
         new_post_count += 1

utils/boxapi_log.py

@@ -1,8 +1,9 @@
 # utils/boxapi_log.py
-import json, os, datetime
+import orjson, os, datetime
 from typing import Any, Dict, Optional
 
 LOG_ROOT = os.environ.get("BOXAPI_LOG_DIR", "/app/logs/boxapi")
+LOG_LEVEL = os.environ.get("BOXAPI_LOG_LEVEL", "all")  # "all", "error", "none"
 
 def _ensure_dir(path: str) -> None:
     os.makedirs(path, exist_ok=True)
@@ -15,6 +16,10 @@ def log_boxapi_response(
     response_text: Optional[str] = None,
     note: str = ""
 ) -> str:
+    # Skip logging if level is "none" or if level is "error" and this is not an error
+    if LOG_LEVEL == "none" or (LOG_LEVEL == "error" and "error" not in note.lower()):
+        return ""
+    
     ts = datetime.datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
     safe_user = (username or "unknown").replace("/", "_")
     dirpath = os.path.join(LOG_ROOT, safe_user)
@@ -30,6 +35,6 @@ def log_boxapi_response(
         "response_json": response_json,
         "response_text": response_text if response_json is None else None,
     }
-    with open(filepath, "w", encoding="utf-8") as f:
-        json.dump(doc, f, ensure_ascii=False, indent=2)
+    with open(filepath, "wb") as f:
+        f.write(orjson.dumps(doc, option=orjson.OPT_INDENT_2))
     return filepath 

utils/parser_fast.py

@@ -0,0 +1,66 @@
+# utils/parser_fast.py
+from typing import Any, Dict, List, Tuple, Optional
+
+def pick_image_url(item: Dict[str, Any]) -> Optional[str]:
+    cands = ((item.get("image_versions2") or {}).get("candidates") or [])
+    return cands[0].get("url") if cands else None
+
+def pick_video_url(item: Dict[str, Any]) -> Optional[str]:
+    # choose the largest height (usually last or explicitly max)
+    vv = item.get("video_versions") or []
+    if not vv: 
+        return None
+    # faster than sorted: single pass to get max by height
+    best = vv[0]
+    bh = best.get("height") or 0
+    for v in vv[1:]:
+        h = v.get("height") or 0
+        if h > bh:
+            bh, best = h, v
+    return best.get("url")
+
+def normalize_item(it: Dict[str, Any]) -> Dict[str, Any]:
+    mt = it.get("media_type")
+    if mt == 1:
+        media_type = "image"
+        url = pick_image_url(it)
+        urls = [url] if url else []
+    elif mt == 2:
+        media_type = "video"
+        # put a cover first if exists
+        cover = pick_image_url(it)
+        vurl = pick_video_url(it)
+        urls = ([cover] if cover else []) + ([vurl] if vurl else [])
+    elif mt == 8:  # carousel
+        media_type = "carousel"
+        urls = []
+        for child in it.get("carousel_media") or []:
+            if child.get("media_type") == 1:
+                u = pick_image_url(child)
+                if u: urls.append(u)
+            elif child.get("media_type") == 2:
+                cover = pick_image_url(child)
+                vurl = pick_video_url(child)
+                if cover: urls.append(cover)
+                if vurl: urls.append(vurl)
+    else:
+        return {}
+
+    return {
+        "ig_post_id": str(it.get("pk") or it.get("id") or ""),
+        "media_type": media_type,
+        "caption": ((it.get("caption") or {}).get("text") or ""),
+        "taken_at": int(it.get("taken_at") or 0),
+        "thumbnail_url": pick_image_url(it),
+        "remote_urls": [u for u in urls if u],
+        "code": it.get("code"),
+    }
+
+def parse_boxapi_items_fast(items: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+    out = []
+    for it in items:
+        n = normalize_item(it)
+        if n: out.append(n)
+    # keep newest first
+    out.sort(key=lambda x: x["taken_at"], reverse=True)
+    return out 

utils/seen_store.py

@@ -0,0 +1,22 @@
+# utils/seen_store.py
+import os, redis
+from typing import Iterable
+
+REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379/0")
+r = redis.from_url(REDIS_URL, decode_responses=True)
+
+def seen_key(username: str) -> str:
+    return f"ig:seen:{username}"
+
+def mark_seen(username: str, ids: Iterable[str]) -> None:
+    if not ids: return
+    r.sadd(seen_key(username), *ids)
+
+def filter_unseen(username: str, ids: Iterable[str]) -> list[str]:
+    # pipeline for speed
+    key = seen_key(username)
+    pipe = r.pipeline()
+    for i in ids:
+        pipe.sismember(key, i)
+    flags = pipe.execute()
+    return [i for i, seen in zip(ids, flags) if not seen]