This commit is contained in:
Hossein 2025-09-25 02:52:44 +03:30
parent 90c28032cc
commit bbc9bc670d
2 changed files with 62 additions and 64 deletions

View File

@ -1,5 +1,6 @@
# scraper.py # scraper.py
import os
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from database import InstagramAccount, InstagramPost from database import InstagramAccount, InstagramPost
from utils.boxapi_client import get_media_page, BoxAPIError from utils.boxapi_client import get_media_page, BoxAPIError
@ -14,63 +15,60 @@ from typing import List, Dict, Any, Optional, Set
from datetime import datetime from datetime import datetime
from scraper_cursor import get_cursor, set_cursor from scraper_cursor import get_cursor, set_cursor
# Safety cap for very large accounts DIAG = os.getenv("SCRAPER_DIAGNOSTICS", "0") == "1"
MAX_PAGES = 500 # adjust if needed MAX_PAGES = int(os.getenv("SCRAPER_MAX_PAGES", "400")) # safety cap
def dprint(msg: str) -> None:
if DIAG:
print(f"[SCRAPER] {msg}")
def _ig_id_of(it: Dict[str, Any]) -> Optional[str]: def _ig_id(it: Dict[str, Any]) -> Optional[str]:
v = it.get("id") or it.get("pk") v = it.get("id") or it.get("pk")
return str(v) if v is not None else None return str(v) if v is not None else None
def _db_existing_ids(db: Session, ids: List[str]) -> Set[str]: def _db_existing_ids(db: Session, ids: List[str]) -> Set[str]:
if not ids: if not ids:
return set() return set()
rows = ( rows = db.query(InstagramPost.ig_post_id)\
db.query(InstagramPost.ig_post_id) .filter(InstagramPost.ig_post_id.in_(ids)).all()
.filter(InstagramPost.ig_post_id.in_(ids))
.all()
)
return {r[0] for r in rows} return {r[0] for r in rows}
def _collect_raw_not_in_db(
def _collect_raw_to_save(
db: Session, db: Session,
username: str, username: str,
needed: int, needed: int,
stored_cursor: Optional[str],
since_ts: Optional[int], since_ts: Optional[int],
) -> tuple[List[Dict[str, Any]], Optional[str]]: ) -> tuple[List[Dict[str, Any]], Optional[str], int]:
""" """
Keep fetching pages until we have `needed` posts that are NOT in DB. Walk pages linearly: newest first, then older via next_max_id.
Return (raw_items_to_save, final_cursor). Keep only posts not present in DB. Stop when we have `needed` or no more pages.
Returns (raw_to_save, final_cursor, pages_walked)
""" """
to_save_raw: List[Dict[str, Any]] = [] raw_to_save: List[Dict[str, Any]] = []
pages = 0 pages = 0
cursor = stored_cursor max_id = None # None → newest page
first = True final_cursor: Optional[str] = None
last_seen_cursor: Optional[str] = stored_cursor
while len(to_save_raw) < needed and pages < MAX_PAGES: while len(raw_to_save) < needed and pages < MAX_PAGES:
max_id = None if first else cursor count = min(12, max(1, needed - len(raw_to_save))) # fetch up to 12
count = min(12, needed - len(to_save_raw)) or 12
try: try:
page, _log, next_id = get_media_page(username=username, count=count, max_id=max_id) page, _log, next_max = get_media_page(username=username, count=count, max_id=max_id)
except BoxAPIError: except BoxAPIError as e:
dprint(f"BoxAPI error: {e}")
break break
pages += 1 pages += 1
last_seen_cursor = next_id dprint(f"Page#{pages}: got {len(page)} items, next_max_id={next_max}")
if not page: if not page:
break break
# Optional: since_taken_at filter (raw, cheap) # Optional cutoff by timestamp (cheap check)
filtered = [] filtered = []
ids_in_order: List[str] = [] ids_ordered: List[str] = []
for it in page: for it in page:
igid = _ig_id_of(it) pid = _ig_id(it)
if not igid: if not pid:
continue continue
if since_ts is not None: if since_ts is not None:
ts = it.get("taken_at") or it.get("timestamp") ts = it.get("taken_at") or it.get("timestamp")
@ -81,37 +79,29 @@ def _collect_raw_to_save(
if ts is not None and ts <= since_ts: if ts is not None and ts <= since_ts:
continue continue
filtered.append(it) filtered.append(it)
ids_in_order.append(igid) ids_ordered.append(pid)
if not ids_in_order: if ids_ordered:
if not next_id: already = _db_existing_ids(db, ids_ordered)
break for it in filtered:
first = False pid = _ig_id(it)
cursor = next_id if pid and pid not in already:
continue raw_to_save.append(it)
if len(raw_to_save) >= needed:
break
# DB check: keep only items not yet saved # advance pagination
already = _db_existing_ids(db, ids_in_order) final_cursor = next_max
for it in filtered: if not next_max:
igid = _ig_id_of(it)
if igid and igid not in already:
to_save_raw.append(it)
if len(to_save_raw) >= needed:
break
if not next_id:
break break
max_id = next_max
first = False return raw_to_save, final_cursor, pages
cursor = next_id
return to_save_raw, last_seen_cursor
def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]: def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
target = int(req.max_count) target = int(req.max_count)
# Optional since_taken_at # since_taken_at (optional)
since_ts: Optional[int] = None since_ts: Optional[int] = None
if hasattr(req, "since_taken_at") and req.since_taken_at is not None: if hasattr(req, "since_taken_at") and req.since_taken_at is not None:
try: try:
@ -119,25 +109,23 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
except Exception: except Exception:
since_ts = None since_ts = None
# Start from stored backfill cursor (if any), but always try newest page first # 1) BRUTE-FORCE scan: newest → older, pick posts NOT in DB until we reach target
stored_cursor = get_cursor(db, req.username) raw_to_save, final_cursor, pages_walked = _collect_raw_not_in_db(
# 1) Collect raw items to save (NOT in DB), paging as deep as needed
raw_to_save, final_cursor = _collect_raw_to_save(
db=db, db=db,
username=req.username, username=req.username,
needed=target, needed=target,
stored_cursor=stored_cursor,
since_ts=since_ts, since_ts=since_ts,
) )
# Persist traversal progress (so next run can continue deeper) # Persist cursor if we walked at least one page
set_cursor(db, req.username, final_cursor) if pages_walked > 0:
set_cursor(db, req.username, final_cursor)
if not raw_to_save: if not raw_to_save:
dprint("No new posts to save.")
return [] return []
# 2) Normalize only what we intend to save # 2) Normalize only the ones we intend to save
normalized = parse_boxapi_items_fast(raw_to_save)[:target] normalized = parse_boxapi_items_fast(raw_to_save)[:target]
# 3) Ensure account row # 3) Ensure account row
@ -159,13 +147,12 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
post_id = str(item["ig_post_id"]) post_id = str(item["ig_post_id"])
seller_id = str(req.seller_id) seller_id = str(req.seller_id)
# DB guard (rare race) # DB guard in case a parallel worker just saved it
if db.query(InstagramPost).filter_by(ig_post_id=post_id).first(): if db.query(InstagramPost).filter_by(ig_post_id=post_id).first():
continue continue
media_urls = list(item.get("remote_urls", [])) media_urls = list(item.get("remote_urls", []))
# If video, keep only actual video URLs (avoid thumbnail)
if item["media_type"] == "video": if item["media_type"] == "video":
thumb = (item.get("thumbnail_url") or "").strip() thumb = (item.get("thumbnail_url") or "").strip()
def is_video(u: str) -> bool: def is_video(u: str) -> bool:
@ -229,4 +216,4 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
if len(result) >= target: if len(result) >= target:
break break
return result return result

View File

@ -12,9 +12,16 @@ BOXAPI_ENDPOINT = os.environ.get(
"https://boxapi.ir/api/instagram/user/get_media_by_username", "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): class BoxAPIError(Exception):
pass pass
def _dprint(msg: str) -> None:
if BOXAPI_DEBUG:
print(f"[BOXAPI] {msg}")
def get_media_page( def get_media_page(
username: str, username: str,
@ -38,6 +45,8 @@ def get_media_page(
if max_id: if max_id:
payload["max_id"] = max_id payload["max_id"] = max_id
_dprint(f"→ POST {BOXAPI_ENDPOINT} payload={payload}")
resp_json: Optional[Dict[str, Any]] = None resp_json: Optional[Dict[str, Any]] = None
resp_text: Optional[str] = None resp_text: Optional[str] = None
status_code: Optional[int] = None status_code: Optional[int] = None
@ -86,6 +95,8 @@ def get_media_page(
f"Unexpected response format from BoxAPI ({snapshot}). 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 return items, log_path, next_max_id
except BoxAPIError: except BoxAPIError: