fazli/FastApi-ISS
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update

Browse Source
This commit is contained in:
Hossein 2025-09-25 02:26:58 +03:30
parent ba0c636ffc
commit c80a10680b
1 changed files with 129 additions and 62 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

191
scraper.py
View File

@ -18,76 +18,150 @@ 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)
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") # might be int/epoch
try:
page, _log, newest_next = get_media_page(username=username, count=need, max_id=None)
if page:
collected.extend(page)
taken_at = int(taken_at) if taken_at is not None else None
except Exception:
# ignore and continue with backfill-only path
pass
taken_at = None
return ig_id, taken_at
# 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))
def _collect_unseen_raw(
username: str,
target_unseen: int,
stored_cursor: Optional[str],
since_ts: Optional[int],
) -> tuple[List[Dict[str, Any]], Optional[str], int]:
"""
Keep requesting BoxAPI pages until we accumulate `target_unseen` UNSEEN items,
or we run out of pages. We do minimal work per page to decide quickly.
Returns:
- selected_raw: only the unseen (and since_ts-passing) raw items in order
- final_cursor: the last next_max_id we reached (to persist)
- unseen_count: number of unseen items we actually collected
"""
selected_raw: List[Dict[str, Any]] = []
final_cursor: Optional[str] = stored_cursor
# 1) newest page first
next_id: Optional[str] = None
try:
first_page, _log, next_id = get_media_page(username=username, count=min(12, target_unseen), max_id=None)
except Exception:
first_page = []
def add_unseen_from_page(page: List[Dict[str, Any]]) -> int:
"""Filter a page to unseen (and since_ts), append to selected_raw, return how many added."""
if not page:
return 0
# Extract ids in order
ids_in_order: List[str] = []
since_mask: List[bool] = []
for it in page:
ig_id, taken_at = _extract_minimal(it)
if ig_id:
# optional since_ts filter (skip if not passing)
ok = True
if since_ts is not None and taken_at is not None:
ok = taken_at > since_ts
ids_in_order.append(ig_id)
since_mask.append(ok)
if not ids_in_order:
return 0
# Fast unseen check
unseen_ids_set = set(filter_unseen(username, ids_in_order))
if not unseen_ids_set:
return 0
# Append only ids that pass since_mask AND are unseen, preserving order
added = 0
for it in page:
ig_id, taken_at = _extract_minimal(it)
if not ig_id:
continue
idx = ids_in_order.index(ig_id) # safe: ig_id exists in list
if not since_mask[idx]:
continue
if ig_id in unseen_ids_set:
selected_raw.append(it)
added += 1
if len(selected_raw) >= target_unseen:
break
return added
# Process newest page first
add_unseen_from_page(first_page)
# 2) then keep going with stored cursor (older pages). Seed with stored or newest next_id.
cursor = stored_cursor or next_id
# Only continue if we still need more
while len(selected_raw) < target_unseen and cursor:
need = min(12, target_unseen - len(selected_raw))
page, _log, cursor = get_media_page(username=username, count=need, max_id=cursor)
added = add_unseen_from_page(page)
final_cursor = cursor # we move the cursor forward as we traverse
if not page:
break
collected.extend(page)
# keep looping until we hit target or run out (cursor becomes None)
return collected, cursor # final cursor to persist
return selected_raw, final_cursor, len(selected_raw)
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
# Optional: allow caller to pass since_taken_at; coerce safely to int
since_ts_raw = getattr(req, "since_taken_at", None)
since_ts: Optional[int] = None
if since_ts_raw is not None:
try:
since_ts = int(since_ts_raw)
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
try:
raw_items, final_cursor = _collect_across_pages(req.username, target, stored_cursor)
selected_raw, final_cursor, unseen_count = _collect_unseen_raw(
username=req.username,
target_unseen=target,
stored_cursor=stored_cursor,
since_ts=since_ts,
)
except BoxAPIError as e:
print(f"❌ BoxAPI error: {e}")
return []
if not raw_items:
if not selected_raw:
# Still store cursor progress (we may have advanced even if nothing new)
set_cursor(db, req.username, final_cursor)
return []
# 2) Normalize once on the aggregated list
normalized = parse_boxapi_items_fast(raw_items)
# 2) Normalize ONLY the selected unseen raw items
normalized = parse_boxapi_items_fast(selected_raw)
# 3) Optional cursor filter (kept as-is)
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:
pass
# 3) Defensive re-check: filter_unseen again after normalize by IG IDs (handles races)
ids_in_order = [n.get("ig_post_id") for n in normalized if n.get("ig_post_id")]
if not ids_in_order:
set_cursor(db, req.username, final_cursor)
return []
# 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]
final_batch = [n for n in normalized if n.get("ig_post_id") in unseen_ids][:target]
# 5) Clip to requested count and persist the backfill cursor (progress)
batch = unseen[:target]
# Persist cursor progress AFTER we’ve walked pages
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])
# Mark as seen now to avoid future reprocessing
mark_seen(req.username, [b["ig_post_id"] for b in final_batch])
# 7) Ensure account row
# 4) Ensure account row
account = db.query(InstagramAccount).filter_by(username=req.username).first()
if not account:
account = InstagramAccount(
@ -100,46 +174,41 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
db.commit()
db.refresh(account)
# 5) Upload + save each post in the final batch
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
for item in final_batch:
post_id = str(item["ig_post_id"])
seller_id = str(req.seller_id)
# skip if already saved (race-safety)
# Skip if already saved (DB guard)
if db.query(InstagramPost).filter_by(ig_post_id=post_id).first():
continue
# media URL list (normalized parser gives remote_urls)
# Media URLs (from normalized)
media_urls = list(item.get("remote_urls", []))
# If this is a video post, keep only true video URLs (avoid thumbnail duplication)
# 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 to S3 (optimized path)
# Upload media
uploaded_urls: List[str] = []
for remote_url in media_urls:
try:
if item["media_type"] == "video":
# optimized 480p H.264 CRF23 (+faststart)
url = upload_video_from_url_optimized(remote_url, seller_id, post_id)
elif item["media_type"] in ("image", "carousel"):
# carousel items may be images; optimize to WebP
url = upload_image_from_url_optimized(remote_url, seller_id, post_id)
if url is None:
# some carousels include videos; try generic fallback
url = upload_media_from_url(remote_url, seller_id, post_id)
else:
# unknown → generic detection/fallback
url = upload_media_from_url(remote_url, seller_id, post_id)
if url:
uploaded_urls.append(url)
except Exception as e:
@ -154,14 +223,13 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
if thumb_url:
final_thumbnail = thumb_url
else:
# fallback to generic (no optimize)
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
# 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,
@ -184,8 +252,7 @@ def scrape_and_store(db: Session, req: ScrapeRequest) -> List[ScrapedPost]:
created_at=created_at,
))
new_post_count += 1
if new_post_count >= target:
if len(result) >= target:
break
return result