245 lines
8.4 KiB
Python
245 lines
8.4 KiB
Python
# scraper.py
|
|
|
|
# --- 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 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
|
|
|
|
|
|
# ----------------------------
|
|
# 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 _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
|
|
|
|
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
|
|
|
|
first_page_done = True
|
|
cursor = next_id # advance our traversal cursor even if page empty
|
|
|
|
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
|
|
|
|
|
|
# ----------------------------
|
|
# Main entry
|
|
# ----------------------------
|
|
def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
|
|
target = int(req.max_count)
|
|
|
|
# Optional since_taken_at support
|
|
since_ts: Optional[int] = None
|
|
if hasattr(req, "since_taken_at") and req.since_taken_at is not None:
|
|
try:
|
|
since_ts = int(req.since_taken_at)
|
|
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
|
|
try:
|
|
raw_unseen, final_cursor = _collect_unseen(
|
|
username=req.username,
|
|
target=target,
|
|
stored_cursor=stored_cursor,
|
|
since_ts=since_ts,
|
|
)
|
|
except BoxAPIError as e:
|
|
print(f"❌ BoxAPI error: {e}")
|
|
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
|
|
result: List[ScrapedPost] = []
|
|
for item in batch:
|
|
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():
|
|
continue
|
|
|
|
# Media URLs (from normalized)
|
|
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:
|
|
if item["media_type"] == "video":
|
|
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:
|
|
url = upload_media_from_url(remote_url, seller_id, post_id)
|
|
else:
|
|
url = upload_media_from_url(remote_url, seller_id, post_id)
|
|
if url:
|
|
uploaded_urls.append(url)
|
|
except Exception as e:
|
|
print(f"⚠️ Error uploading media: {e}")
|
|
continue
|
|
|
|
# Upload/optimize thumbnail (if present)
|
|
final_thumbnail = item.get("thumbnail_url") or None
|
|
if final_thumbnail:
|
|
try:
|
|
thumb_url = upload_image_from_url_optimized(final_thumbnail, seller_id, post_id)
|
|
if thumb_url:
|
|
final_thumbnail = thumb_url
|
|
else:
|
|
alt = upload_media_from_url(final_thumbnail, seller_id, post_id)
|
|
if alt:
|
|
final_thumbnail = alt
|
|
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(
|
|
ig_post_id=post_id,
|
|
account_id=account.id,
|
|
media_type=item["media_type"],
|
|
caption=item.get("caption"),
|
|
thumbnail_url=final_thumbnail,
|
|
local_media=uploaded_urls,
|
|
created_at=created_at,
|
|
)
|
|
db.add(post_entry)
|
|
db.commit()
|
|
|
|
result.append(ScrapedPost(
|
|
ig_post_id=post_id,
|
|
media_type=item["media_type"],
|
|
caption=item.get("caption"),
|
|
thumbnail_url=final_thumbnail,
|
|
local_media=uploaded_urls,
|
|
created_at=created_at,
|
|
))
|
|
|
|
if len(result) >= target:
|
|
break
|
|
|
|
return result
|