120 lines
4.0 KiB
Python
120 lines
4.0 KiB
Python
from sqlalchemy.orm import Session
|
|
from database import InstagramAccount, InstagramPost
|
|
from utils.boxapi_client import get_media, 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
|
|
from schemas import ScrapeRequest, ScrapedPost
|
|
from typing import List
|
|
from datetime import datetime
|
|
|
|
|
|
def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
|
|
try:
|
|
# 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 raw_items:
|
|
return []
|
|
|
|
# 2) normalize quickly
|
|
normalized = parse_boxapi_items_fast(raw_items)
|
|
|
|
# 3) apply cursor filter if provided (safe, optional)
|
|
since_ts = getattr(req, "since_taken_at", None)
|
|
if since_ts:
|
|
try:
|
|
since_ts = int(since_ts)
|
|
normalized = [n for n in normalized if (n.get("taken_at") or 0) > since_ts]
|
|
except Exception:
|
|
# If since_taken_at is invalid, ignore it
|
|
pass
|
|
|
|
# 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(
|
|
username=req.username,
|
|
seller_id=req.seller_id,
|
|
is_active=True,
|
|
last_synced=None
|
|
)
|
|
db.add(account)
|
|
db.commit()
|
|
db.refresh(account)
|
|
|
|
result: List[ScrapedPost] = []
|
|
new_post_count = 0
|
|
|
|
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 (optional - can be done async later)
|
|
media_urls: List[str] = []
|
|
for remote_url in item["remote_urls"]:
|
|
try:
|
|
local_url = upload_media_from_url(remote_url, str(req.seller_id), f"{post_id}/")
|
|
if local_url:
|
|
media_urls.append(local_url)
|
|
except Exception as e:
|
|
print(f"⚠️ Error uploading media: {e}")
|
|
continue
|
|
|
|
# Upload thumbnail (optional)
|
|
final_thumbnail = item["thumbnail_url"]
|
|
if item["thumbnail_url"]:
|
|
try:
|
|
thumb_uploaded = upload_media_from_url(item["thumbnail_url"], str(req.seller_id), f"{post_id}/thumbnail")
|
|
if thumb_uploaded:
|
|
final_thumbnail = thumb_uploaded
|
|
except Exception as e:
|
|
print(f"⚠️ Failed to upload thumbnail: {e}")
|
|
|
|
# Save to DB
|
|
post_entry = InstagramPost(
|
|
ig_post_id=post_id,
|
|
account_id=account.id,
|
|
media_type=item["media_type"],
|
|
caption=item["caption"],
|
|
thumbnail_url=final_thumbnail,
|
|
local_media=media_urls,
|
|
created_at=datetime.fromtimestamp(item["taken_at"]),
|
|
)
|
|
db.add(post_entry)
|
|
db.commit()
|
|
|
|
result.append(ScrapedPost(
|
|
ig_post_id=post_id,
|
|
media_type=item["media_type"],
|
|
caption=item["caption"],
|
|
thumbnail_url=final_thumbnail,
|
|
local_media=media_urls, # Return S3 URLs only
|
|
created_at=datetime.fromtimestamp(item["taken_at"])
|
|
))
|
|
|
|
new_post_count += 1
|
|
if new_post_count >= req.max_count:
|
|
break
|
|
|
|
return result
|