diff --git a/database.py b/database.py index 2bf3b53..a67c65c 100644 --- a/database.py +++ b/database.py @@ -1,32 +1,40 @@ -from sqlalchemy import create_engine, Column, Integer, String, Float, DateTime, Boolean, Text, ForeignKey +# database.py +import os +import uuid +from sqlalchemy import ( + create_engine, Column, Integer, String, Float, DateTime, Boolean, Text, + ForeignKey, UniqueConstraint, Index +) from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, relationship from sqlalchemy.sql import func from sqlalchemy.dialects.postgresql import UUID, JSONB -import os -import uuid -# Database URL from environment variable +# ----------------------------------------------------------------------------- +# Database URL +# ----------------------------------------------------------------------------- DATABASE_URL = os.getenv("DATABASE_URL") if not DATABASE_URL: - # Fallback to constructing from individual variables POSTGRES_USER = os.getenv("POSTGRES_USER", "fastapi_user") POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD", "fastapi_password") POSTGRES_HOST = os.getenv("POSTGRES_HOST", "postgres") POSTGRES_PORT = os.getenv("POSTGRES_PORT", "5432") POSTGRES_DB = os.getenv("POSTGRES_DB", "fastapi_db") - DATABASE_URL = f"postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_HOST}:{POSTGRES_PORT}/{POSTGRES_DB}" + DATABASE_URL = ( + f"postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}" + f"@{POSTGRES_HOST}:{POSTGRES_PORT}/{POSTGRES_DB}" + ) -# Create SQLAlchemy engine +# ----------------------------------------------------------------------------- +# Engine / Session / Base +# ----------------------------------------------------------------------------- engine = create_engine(DATABASE_URL) - -# Create SessionLocal class SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) - -# Create Base class Base = declarative_base() -# Basic Item model (for backward compatibility) +# ----------------------------------------------------------------------------- +# Basic Item model (kept for backward compatibility / health checks) +# ----------------------------------------------------------------------------- class Item(Base): __tablename__ = "items" @@ -37,37 +45,84 @@ class Item(Base): created_at = Column(DateTime(timezone=True), server_default=func.now()) updated_at = Column(DateTime(timezone=True), onupdate=func.now()) -# Instagram Account Model +# ----------------------------------------------------------------------------- +# Cursor for BoxAPI pagination (persist next_max_id per username) +# ----------------------------------------------------------------------------- +class InstaCursor(Base): + __tablename__ = "insta_cursors" + + id = Column(Integer, primary_key=True) + username = Column(String(150), nullable=False, unique=True) + next_max_id = Column(String(255), nullable=True) + updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now()) + + __table_args__ = ( + UniqueConstraint("username", name="uq_insta_cursor_username"), + Index("ix_insta_cursors_username", "username"), + ) + +# ----------------------------------------------------------------------------- +# Instagram Account +# ----------------------------------------------------------------------------- class InstagramAccount(Base): __tablename__ = "instagram_accounts" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) seller_id = Column(UUID(as_uuid=True), nullable=False) - username = Column(String(100), nullable=False, unique=True) + username = Column(String(100), nullable=False, unique=True, index=True) is_active = Column(Boolean, default=True) last_synced = Column(DateTime(timezone=True), nullable=True) - # Relationship - posts = relationship("InstagramPost", back_populates="account", cascade="all, delete-orphan") + posts = relationship( + "InstagramPost", + back_populates="account", + cascade="all, delete-orphan", + passive_deletes=True, + ) -# Instagram Post Model + __table_args__ = ( + UniqueConstraint("username", name="uq_instagram_accounts_username"), + Index("ix_instagram_accounts_username", "username"), + Index("ix_instagram_accounts_seller_id", "seller_id"), + ) + +# ----------------------------------------------------------------------------- +# Instagram Post +# ----------------------------------------------------------------------------- class InstagramPost(Base): __tablename__ = "instagram_posts" id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) - account_id = Column(UUID(as_uuid=True), ForeignKey("instagram_accounts.id"), nullable=False) - ig_post_id = Column(String(100), nullable=False, unique=True) + account_id = Column( + UUID(as_uuid=True), + ForeignKey("instagram_accounts.id", ondelete="CASCADE"), + nullable=False, + index=True, + ) + ig_post_id = Column(String(100), nullable=False, unique=True, index=True) media_type = Column(String(10), nullable=False) # 'image', 'video', 'carousel' caption = Column(Text, nullable=True) thumbnail_url = Column(Text, nullable=True) - local_media = Column(JSONB, nullable=True) # List of media items with type and local path - created_at = Column(DateTime(timezone=True), nullable=False) # Original post time on Instagram - fetched_at = Column(DateTime(timezone=True), server_default=func.now()) # Time when scraped - # Relationship + # List[str] of public S3 URLs (stored as JSON array) + local_media = Column(JSONB, nullable=True) + + # Original post time on Instagram + created_at = Column(DateTime(timezone=True), nullable=False, index=True) + # When we scraped it + fetched_at = Column(DateTime(timezone=True), server_default=func.now()) + account = relationship("InstagramAccount", back_populates="posts") -# Dependency to get database session + __table_args__ = ( + UniqueConstraint("ig_post_id", name="uq_instagram_posts_ig_post_id"), + Index("ix_instagram_posts_created_at", "created_at"), + Index("ix_instagram_posts_media_type", "media_type"), + ) + +# ----------------------------------------------------------------------------- +# Session dependency +# ----------------------------------------------------------------------------- def get_db(): db = SessionLocal() try: @@ -75,6 +130,8 @@ def get_db(): finally: db.close() -# Create tables +# ----------------------------------------------------------------------------- +# Table creation +# ----------------------------------------------------------------------------- def create_tables(): - Base.metadata.create_all(bind=engine) \ No newline at end of file + Base.metadata.create_all(bind=engine) diff --git a/scraper.py b/scraper.py index 6e946ea..f4e8806 100644 --- a/scraper.py +++ b/scraper.py @@ -1,7 +1,9 @@ +# scraper.py + # --- imports --- from sqlalchemy.orm import Session from database import InstagramAccount, InstagramPost -from utils.boxapi_client import get_media, BoxAPIError +from utils.boxapi_client import get_media_page, BoxAPIError # ← paged client 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 @@ -11,13 +13,49 @@ from utils.s3_uploader import ( upload_video_from_url_optimized, ) from schemas import ScrapeRequest, ScrapedPost -from typing import List +from typing import List, Dict, Any, Optional from datetime import datetime +from scraper_cursor import get_cursor, set_cursor # ← new helpers + + +def _collect_across_pages(username: str, target: int, seed_cursor: Optional[str]) -> tuple[list[Dict[str, Any]], Optional[str]]: + """ + 1) Fetch ONE newest page (max_id=None). + 2) If still short, continue with `seed_cursor` (or newest next_max_id) to backfill older pages. + Returns (raw_items, final_cursor_to_store). + """ + collected: list[Dict[str, Any]] = [] + newest_next: Optional[str] = None + + # Step 1: newest page first (cheap check for new content) + need = min(12, target) + try: + page, _log, newest_next = get_media_page(username=username, count=need, max_id=None) + if page: + collected.extend(page) + except Exception: + # ignore and continue with backfill-only path + pass + + # Step 2: continue older pages from stored cursor (or newest_next if none stored) + cursor = seed_cursor or newest_next + while len(collected) < target and cursor: + need = min(12, target - len(collected)) + page, _log, cursor = get_media_page(username=username, count=need, max_id=cursor) + if not page: + break + collected.extend(page) + + return collected, cursor # final cursor to persist def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: + target = int(req.max_count) + + # 1) Collect raw items across newest + backfill pages (no request schema change) + stored_cursor = get_cursor(db, req.username) # may be None on first run try: - raw_items, log_path = get_media(req.username, max(req.max_count, 30)) + raw_items, final_cursor = _collect_across_pages(req.username, target, stored_cursor) except BoxAPIError as e: print(f"❌ BoxAPI error: {e}") return [] @@ -25,9 +63,10 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: if not raw_items: return [] + # 2) Normalize once on the aggregated list normalized = parse_boxapi_items_fast(raw_items) - # optional cursor filter + # 3) Optional cursor filter (kept as-is) since_ts = getattr(req, "since_taken_at", None) if since_ts: try: @@ -36,16 +75,19 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: except Exception: pass - # filter seen + # 4) Filter already-seen FIRST (avoid re-uploading) — ensure filter_unseen has NO hidden 5-cap 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] - # pick next N - batch = unseen[:req.max_count] + # 5) Clip to requested count and persist the backfill cursor (progress) + batch = unseen[:target] + set_cursor(db, req.username, final_cursor) + + # 6) Mark these as seen so the next run skips them mark_seen(req.username, [b["ig_post_id"] for b in batch]) - # ensure account + # 7) Ensure account row account = db.query(InstagramAccount).filter_by(username=req.username).first() if not account: account = InstagramAccount( @@ -61,18 +103,19 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: result: List[ScrapedPost] = [] new_post_count = 0 + # 8) Upload + save each post in the batch for item in batch: post_id = str(item["ig_post_id"]) # ENSURE clean, no slashes seller_id = str(req.seller_id) - # skip if already saved + # skip if already saved (race-safety) if db.query(InstagramPost).filter_by(ig_post_id=post_id).first(): continue # media URL list (normalized parser gives remote_urls) media_urls = list(item.get("remote_urls", [])) - # If this is a video post, keep only true video URLs + # If this is a video post, keep only true video URLs (avoid thumbnail duplication) if item["media_type"] == "video": thumb = (item.get("thumbnail_url") or "").strip() def is_video(u: str) -> bool: @@ -142,7 +185,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: )) new_post_count += 1 - if new_post_count >= req.max_count: + if new_post_count >= target: break return result diff --git a/scraper_cursor.py b/scraper_cursor.py new file mode 100644 index 0000000..e15b83e --- /dev/null +++ b/scraper_cursor.py @@ -0,0 +1,17 @@ +# scraper_cursor.py +from typing import Optional +from sqlalchemy.orm import Session +from database import InstaCursor + +def get_cursor(db: Session, username: str) -> Optional[str]: + row = db.query(InstaCursor).filter_by(username=username).first() + return row.next_max_id if row else None + +def set_cursor(db: Session, username: str, next_max_id: Optional[str]) -> None: + row = db.query(InstaCursor).filter_by(username=username).first() + if row is None: + row = InstaCursor(username=username, next_max_id=next_max_id) + db.add(row) + else: + row.next_max_id = next_max_id + db.commit() diff --git a/utils/boxapi_client.py b/utils/boxapi_client.py index e671e08..d3037db 100644 --- a/utils/boxapi_client.py +++ b/utils/boxapi_client.py @@ -1,27 +1,43 @@ -import os, requests +# utils/boxapi_client.py +import os +import requests from typing import Any, Dict, List, Tuple, Optional from utils.boxapi_log import log_boxapi_response BOXAPI_USERNAME = os.environ.get("BOXAPI_USERNAME", "") BOXAPI_PASSWORD = os.environ.get("BOXAPI_PASSWORD", "") -# Your .env can override to /get_media; default matches your cURL (/get_media_by_username) +# 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" + "https://boxapi.ir/api/instagram/user/get_media_by_username", ) class BoxAPIError(Exception): pass -def get_media(username: str, count: int = 10) -> Tuple[List[Dict[str, Any]], str]: + +def get_media_page( + username: str, + count: int = 10, + max_id: Optional[str] = None, +) -> Tuple[List[Dict[str, Any]], str, Optional[str]]: """ - Calls BoxAPI and returns (items, log_path). + Perform a single BoxAPI request and return a page of results. + 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.") - payload = {"username": username, "count": count} - resp = None + # BoxAPI hard 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 + resp_json: Optional[Dict[str, Any]] = None resp_text: Optional[str] = None status_code: Optional[int] = None @@ -35,29 +51,30 @@ def get_media(username: str, count: int = 10) -> Tuple[List[Dict[str, Any]], str ) status_code = resp.status_code resp_text = resp.text - # Try to parse JSON; if it fails, we'll log text and raise + + # Try to parse JSON; if it fails, log and raise try: resp_json = resp.json() except Exception: log_path = log_boxapi_response( username, payload, status_code, response_text=resp_text, - note="Non-JSON response" + note="Non-JSON response (error)" ) raise BoxAPIError(f"BoxAPI non-JSON response. See log: {log_path}") - # Expected shape: status -> response -> status_code + body.items + # Expected shape: status -> response -> status_code + body.items + body.next_max_id top_status = resp_json.get("status") inner = resp_json.get("response") or {} inner_code = inner.get("status_code") body = inner.get("body") or {} items = body.get("items") or [] + next_max_id = body.get("next_max_id") log_path = log_boxapi_response( username, payload, inner_code, response_json=resp_json, note="OK" ) if top_status != "done" or inner_code != 200 or not isinstance(items, list): - # Include a concise snapshot of keys to help debugging snapshot = { "top_status": top_status, "inner_code": inner_code, @@ -65,24 +82,53 @@ def get_media(username: str, count: int = 10) -> Tuple[List[Dict[str, Any]], str "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 ({snapshot}). See log: {log_path}" + ) - return items, log_path + return items, log_path, next_max_id except BoxAPIError: raise except Exception as e: log_path = log_boxapi_response( - username, payload, status_code, response_json=resp_json, response_text=resp_text, - note=f"HTTP/parse error: {e}" + username, payload, status_code, + response_json=resp_json, response_text=resp_text, + note=f"HTTP/parse error (error): {e}" ) raise BoxAPIError(f"BoxAPI request failed ({e}). See log: {log_path}") -# Legacy function for backward compatibility -def fetch_boxapi_posts(username: str, count: int = 30) -> List[Dict]: + +def get_media(username: str, count: int = 10) -> Tuple[List[Dict[str, Any]], str]: + """ + Backward-compatible wrapper that fetches ONE page (latest) without pagination. + Returns: (items, log_path) + """ + items, log_path, _next = get_media_page(username=username, count=count, max_id=None) + return items, log_path + + +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: - items, _ = get_media(username, count) - return items + target = max(1, int(count)) + collected: List[Dict[str, Any]] = [] + next_id: Optional[str] = None + + while len(collected) < target: + need = min(12, target - len(collected)) + items, _log, next_id = get_media_page(username=username, count=need, max_id=next_id) + if not items: + break + collected.extend(items) + if not next_id: # no more pages + break + + return collected[:target] + except BoxAPIError as e: print(f"❌ Failed to fetch posts from BoxAPI: {e}") - return [] \ No newline at end of file + return []