ishalla fixe
This commit is contained in:
parent
2e55f2d99e
commit
90c28032cc
157
scraper.py
157
scraper.py
@ -1,11 +1,9 @@
|
||||
# scraper.py
|
||||
|
||||
import os
|
||||
from sqlalchemy.orm import Session
|
||||
from database import InstagramAccount, InstagramPost
|
||||
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
|
||||
from utils.s3_uploader import (
|
||||
upload_media_from_url,
|
||||
upload_image_from_url_optimized,
|
||||
@ -16,23 +14,14 @@ from typing import List, Dict, Any, Optional, 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", "200")) # hard safety
|
||||
# Safety cap for very large accounts
|
||||
MAX_PAGES = 500 # adjust if needed
|
||||
|
||||
def dprint(msg: str) -> None:
|
||||
if DIAG:
|
||||
print(f"[SCRAPER] {msg}")
|
||||
|
||||
def _ig_id_of(it: Dict[str, Any]) -> Optional[str]:
|
||||
v = it.get("id") or it.get("pk")
|
||||
return str(v) if v is not None else None
|
||||
|
||||
def _taken_at_of(it: Dict[str, Any]) -> Optional[int]:
|
||||
v = it.get("taken_at") or it.get("timestamp")
|
||||
try:
|
||||
return int(v) if v is not None else None
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
def _db_existing_ids(db: Session, ids: List[str]) -> Set[str]:
|
||||
if not ids:
|
||||
@ -44,106 +33,85 @@ def _db_existing_ids(db: Session, ids: List[str]) -> Set[str]:
|
||||
)
|
||||
return {r[0] for r in rows}
|
||||
|
||||
def _ordered_unseen_ids(db: Session, username: str, ids_in_order: List[str]) -> List[str]:
|
||||
"""
|
||||
Combine Redis fast-path + DB guard to get truly unseen ids in original order.
|
||||
This eliminates mismatch issues between Redis key format and DB.
|
||||
"""
|
||||
# Redis fast path
|
||||
redis_unseen = set(filter_unseen(username, ids_in_order))
|
||||
if not redis_unseen:
|
||||
return []
|
||||
|
||||
# DB guard (in case posts were stored by another worker/process)
|
||||
already_in_db = _db_existing_ids(db, ids_in_order)
|
||||
truly_unseen = redis_unseen.difference(already_in_db)
|
||||
|
||||
# Preserve original order
|
||||
return [pid for pid in ids_in_order if pid in truly_unseen]
|
||||
|
||||
def _collect_unseen_raw(
|
||||
def _collect_raw_to_save(
|
||||
db: Session,
|
||||
username: str,
|
||||
target: int,
|
||||
needed: int,
|
||||
stored_cursor: Optional[str],
|
||||
since_ts: Optional[int],
|
||||
) -> tuple[List[Dict[str, Any]], Optional[str], int, int]:
|
||||
) -> tuple[List[Dict[str, Any]], Optional[str]]:
|
||||
"""
|
||||
Page through BoxAPI until we collect `target` unseen posts or run out of pages.
|
||||
Returns: (raw_unseen_items, final_cursor, pages_walked, unseen_count)
|
||||
Keep fetching pages until we have `needed` posts that are NOT in DB.
|
||||
Return (raw_items_to_save, final_cursor).
|
||||
"""
|
||||
raw_unseen: List[Dict[str, Any]] = []
|
||||
pages_walked = 0
|
||||
final_cursor = stored_cursor # default to stored; will update when we traverse
|
||||
to_save_raw: List[Dict[str, Any]] = []
|
||||
pages = 0
|
||||
cursor = stored_cursor
|
||||
first = True
|
||||
cursor = stored_cursor # For first fetch we will use None (newest) unless you want pure backfill.
|
||||
|
||||
while len(raw_unseen) < target and pages_walked < MAX_PAGES:
|
||||
max_id = None if first else cursor # newest first, then backfill using cursor
|
||||
need = min(12, max(1, target - len(raw_unseen)))
|
||||
last_seen_cursor: Optional[str] = stored_cursor
|
||||
|
||||
while len(to_save_raw) < needed and pages < MAX_PAGES:
|
||||
max_id = None if first else cursor
|
||||
count = min(12, needed - len(to_save_raw)) or 12
|
||||
try:
|
||||
page, _log, next_id = get_media_page(username=username, count=need, max_id=max_id)
|
||||
except BoxAPIError as e:
|
||||
dprint(f"BoxAPI error on page fetch: {e}")
|
||||
page, _log, next_id = get_media_page(username=username, count=count, max_id=max_id)
|
||||
except BoxAPIError:
|
||||
break
|
||||
|
||||
pages_walked += 1
|
||||
dprint(f"Page#{pages_walked} fetch → max_id={max_id} → next_max_id={next_id}, raw_count={len(page)}")
|
||||
|
||||
# Move traversal cursor forward only if we actually fetched a page
|
||||
cursor = next_id
|
||||
if page:
|
||||
final_cursor = next_id
|
||||
pages += 1
|
||||
last_seen_cursor = next_id
|
||||
|
||||
if not page:
|
||||
break
|
||||
|
||||
# Apply since_ts BEFORE unseen checks
|
||||
# Optional: since_taken_at filter (raw, cheap)
|
||||
filtered = []
|
||||
ids_in_order: List[str] = []
|
||||
for it in page:
|
||||
igid = _ig_id_of(it)
|
||||
if not igid:
|
||||
continue
|
||||
ts = _taken_at_of(it)
|
||||
if since_ts is not None and ts is not None and ts <= since_ts:
|
||||
continue
|
||||
if since_ts is not None:
|
||||
ts = it.get("taken_at") or it.get("timestamp")
|
||||
try:
|
||||
ts = int(ts) if ts is not None else None
|
||||
except Exception:
|
||||
ts = None
|
||||
if ts is not None and ts <= since_ts:
|
||||
continue
|
||||
filtered.append(it)
|
||||
ids_in_order.append(igid)
|
||||
|
||||
if not ids_in_order:
|
||||
dprint(" after since_ts filter: 0 items")
|
||||
if not cursor:
|
||||
if not next_id:
|
||||
break
|
||||
first = False
|
||||
cursor = next_id
|
||||
continue
|
||||
|
||||
# Combined unseen check (Redis + DB)
|
||||
ordered_unseen_ids = _ordered_unseen_ids(db, username, ids_in_order)
|
||||
dprint(f" ids_in_order={len(ids_in_order)} unseen_after_redis+db={len(ordered_unseen_ids)}")
|
||||
# DB check: keep only items not yet saved
|
||||
already = _db_existing_ids(db, ids_in_order)
|
||||
for it in filtered:
|
||||
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 ordered_unseen_ids:
|
||||
unseen_set = set(ordered_unseen_ids)
|
||||
for it in filtered:
|
||||
igid = _ig_id_of(it)
|
||||
if igid in unseen_set:
|
||||
raw_unseen.append(it)
|
||||
if len(raw_unseen) >= target:
|
||||
break
|
||||
|
||||
if not cursor:
|
||||
if not next_id:
|
||||
break
|
||||
|
||||
first = False
|
||||
cursor = next_id
|
||||
|
||||
return to_save_raw, last_seen_cursor
|
||||
|
||||
dprint(f"Collected unseen={len(raw_unseen)} across pages={pages_walked}, final_cursor={final_cursor}")
|
||||
return raw_unseen, final_cursor, pages_walked, len(raw_unseen)
|
||||
|
||||
def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
|
||||
target = int(req.max_count)
|
||||
|
||||
# Parse optional since_taken_at
|
||||
# Optional since_taken_at
|
||||
since_ts: Optional[int] = None
|
||||
if hasattr(req, "since_taken_at") and req.since_taken_at is not None:
|
||||
try:
|
||||
@ -151,38 +119,28 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
|
||||
except Exception:
|
||||
since_ts = None
|
||||
|
||||
stored_cursor = get_cursor(db, req.username) # may be None
|
||||
# Start from stored backfill cursor (if any), but always try newest page first
|
||||
stored_cursor = get_cursor(db, req.username)
|
||||
|
||||
# 1) Collect ONLY unseen raw items (newest page first, then older via cursor)
|
||||
raw_unseen, final_cursor, pages_walked, unseen_found = _collect_unseen_raw(
|
||||
# 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,
|
||||
username=req.username,
|
||||
target=target,
|
||||
needed=target,
|
||||
stored_cursor=stored_cursor,
|
||||
since_ts=since_ts,
|
||||
)
|
||||
|
||||
# Persist traversal progress only if we actually walked at least one page
|
||||
# (This avoids accidentally overwriting a good cursor when nothing fetched.)
|
||||
if pages_walked > 0:
|
||||
set_cursor(db, req.username, final_cursor)
|
||||
# Persist traversal progress (so next run can continue deeper)
|
||||
set_cursor(db, req.username, final_cursor)
|
||||
|
||||
if not raw_unseen:
|
||||
dprint("No unseen items collected; returning early.")
|
||||
if not raw_to_save:
|
||||
return []
|
||||
|
||||
# 2) Normalize only selected unseen raw items
|
||||
normalized = parse_boxapi_items_fast(raw_unseen)
|
||||
# 2) Normalize only what we intend to save
|
||||
normalized = parse_boxapi_items_fast(raw_to_save)[:target]
|
||||
|
||||
# 3) Defensive unseen re-check (race safety) AFTER normalize
|
||||
ids_in_order = [n.get("ig_post_id") for n in normalized if n.get("ig_post_id")]
|
||||
final_unseen_ids = _ordered_unseen_ids(db, req.username, ids_in_order)
|
||||
batch = [n for n in normalized if n.get("ig_post_id") in set(final_unseen_ids)][:target]
|
||||
|
||||
# Mark seen to prevent reprocessing
|
||||
mark_seen(req.username, [b["ig_post_id"] for b in batch])
|
||||
|
||||
# 4) Ensure account row
|
||||
# 3) Ensure account row
|
||||
account = db.query(InstagramAccount).filter_by(username=req.username).first()
|
||||
if not account:
|
||||
account = InstagramAccount(
|
||||
@ -195,18 +153,19 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
|
||||
db.commit()
|
||||
db.refresh(account)
|
||||
|
||||
# 5) Upload + save each post
|
||||
# 4) Upload + save
|
||||
result: List[ScrapedPost] = []
|
||||
for item in batch:
|
||||
for item in normalized:
|
||||
post_id = str(item["ig_post_id"])
|
||||
seller_id = str(req.seller_id)
|
||||
|
||||
# DB guard
|
||||
# DB guard (rare race)
|
||||
if db.query(InstagramPost).filter_by(ig_post_id=post_id).first():
|
||||
continue
|
||||
|
||||
media_urls = list(item.get("remote_urls", []))
|
||||
|
||||
# If video, keep only actual video URLs (avoid thumbnail)
|
||||
if item["media_type"] == "video":
|
||||
thumb = (item.get("thumbnail_url") or "").strip()
|
||||
def is_video(u: str) -> bool:
|
||||
@ -245,6 +204,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
|
||||
print(f"⚠️ Failed to upload thumbnail: {e}")
|
||||
|
||||
created_at = datetime.fromtimestamp(item["taken_at"]) if item.get("taken_at") else datetime.utcnow()
|
||||
|
||||
post_entry = InstagramPost(
|
||||
ig_post_id=post_id,
|
||||
account_id=account.id,
|
||||
@ -269,5 +229,4 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
|
||||
if len(result) >= target:
|
||||
break
|
||||
|
||||
dprint(f"Returning {len(result)} items (requested {target}).")
|
||||
return result
|
||||
|
||||
Loading…
Reference in New Issue
Block a user