Jeffrey Sperling
96948cc7c0
Replace duplicated STOPWORDS, SYNONYMS, _tokenize, and _compute_relevance in four modules with imports from the shared relevance.py module. Existing tests pass unchanged since modules re-export the functions under the same names via import aliases.
371 lines
12 KiB
Python
371 lines
12 KiB
Python
"""Instagram Reels search via ScrapeCreators API for /last30days.
|
|
|
|
Uses ScrapeCreators REST API to search Instagram Reels by keyword, extract
|
|
engagement metrics (views, likes, comments), and fetch video transcripts.
|
|
|
|
Requires SCRAPECREATORS_API_KEY in config. 100 free credits, then PAYG.
|
|
API docs: https://scrapecreators.com/docs
|
|
"""
|
|
|
|
import re
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Dict, List, Optional, Set
|
|
|
|
try:
|
|
import requests as _requests
|
|
except ImportError:
|
|
_requests = None
|
|
|
|
from . import http
|
|
|
|
SCRAPECREATORS_BASE = "https://api.scrapecreators.com"
|
|
|
|
# Depth configurations: how many results to fetch / captions to extract
|
|
DEPTH_CONFIG = {
|
|
"quick": {"results_per_page": 10, "max_captions": 3},
|
|
"default": {"results_per_page": 20, "max_captions": 5},
|
|
"deep": {"results_per_page": 40, "max_captions": 8},
|
|
}
|
|
|
|
# Max words to keep from each caption
|
|
CAPTION_MAX_WORDS = 500
|
|
|
|
from .relevance import (
|
|
STOPWORDS,
|
|
SYNONYMS,
|
|
token_overlap_relevance as _compute_relevance,
|
|
tokenize as _tokenize,
|
|
)
|
|
|
|
|
|
def _extract_core_subject(topic: str) -> str:
|
|
"""Extract core subject from verbose query for Instagram search."""
|
|
from .query import extract_core_subject
|
|
_INSTAGRAM_NOISE = frozenset({
|
|
'best', 'top', 'good', 'great', 'awesome', 'killer',
|
|
'latest', 'new', 'news', 'update', 'updates',
|
|
'trending', 'hottest', 'popular', 'viral',
|
|
'practices', 'features',
|
|
'recommendations', 'advice',
|
|
'prompt', 'prompts', 'prompting',
|
|
'methods', 'strategies', 'approaches',
|
|
})
|
|
return extract_core_subject(topic, noise=_INSTAGRAM_NOISE)
|
|
|
|
|
|
def _log(msg: str):
|
|
"""Log to stderr (only in interactive terminals; spinner handles non-TTY)."""
|
|
if sys.stderr.isatty():
|
|
sys.stderr.write(f"[Instagram] {msg}\n")
|
|
sys.stderr.flush()
|
|
|
|
|
|
def _sc_headers(token: str) -> Dict[str, str]:
|
|
"""Build ScrapeCreators request headers."""
|
|
return {
|
|
"x-api-key": token,
|
|
"Content-Type": "application/json",
|
|
}
|
|
|
|
|
|
def _parse_date(item: Dict[str, Any]) -> Optional[str]:
|
|
"""Parse date from ScrapeCreators Instagram item to YYYY-MM-DD.
|
|
|
|
Handles taken_at as ISO string (e.g. "2026-02-26T16:00:00.000Z")
|
|
or unix timestamp.
|
|
"""
|
|
ts = item.get("taken_at")
|
|
if not ts:
|
|
return None
|
|
|
|
# Try ISO string first (ScrapeCreators reels/search returns this)
|
|
if isinstance(ts, str):
|
|
try:
|
|
# Handle "2026-02-26T16:00:00.000Z" format
|
|
dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
|
|
return dt.strftime("%Y-%m-%d")
|
|
except (ValueError, TypeError):
|
|
pass
|
|
# Try just the date portion
|
|
if len(ts) >= 10:
|
|
return ts[:10]
|
|
|
|
# Fall back to unix timestamp
|
|
try:
|
|
dt = datetime.fromtimestamp(int(ts), tz=timezone.utc)
|
|
return dt.strftime("%Y-%m-%d")
|
|
except (ValueError, TypeError, OSError):
|
|
pass
|
|
|
|
return None
|
|
|
|
|
|
def _extract_hashtags(caption_text: str) -> List[str]:
|
|
"""Extract hashtags from Instagram caption text."""
|
|
if not caption_text:
|
|
return []
|
|
return re.findall(r'#(\w+)', caption_text)
|
|
|
|
|
|
def search_instagram(
|
|
topic: str,
|
|
from_date: str,
|
|
to_date: str,
|
|
depth: str = "default",
|
|
token: str = None,
|
|
) -> Dict[str, Any]:
|
|
"""Search Instagram Reels via ScrapeCreators API.
|
|
|
|
Args:
|
|
topic: Search topic
|
|
from_date: Start date (YYYY-MM-DD)
|
|
to_date: End date (YYYY-MM-DD)
|
|
depth: 'quick', 'default', or 'deep'
|
|
token: ScrapeCreators API key
|
|
|
|
Returns:
|
|
Dict with 'items' list and optional 'error'.
|
|
"""
|
|
if not token:
|
|
return {"items": [], "error": "No SCRAPECREATORS_API_KEY configured"}
|
|
|
|
config = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"])
|
|
core_topic = _extract_core_subject(topic)
|
|
|
|
_log(f"Searching Instagram for '{core_topic}' (depth={depth}, count={config['results_per_page']})")
|
|
|
|
if not _requests:
|
|
_log("requests library not installed, falling back to urllib")
|
|
try:
|
|
from urllib.parse import urlencode
|
|
params = urlencode({"query": core_topic})
|
|
url = f"{SCRAPECREATORS_BASE}/v2/instagram/reels/search?{params}"
|
|
headers = _sc_headers(token)
|
|
headers["User-Agent"] = http.USER_AGENT
|
|
data = http.get(url, headers=headers, timeout=30, retries=2)
|
|
except Exception as e:
|
|
_log(f"ScrapeCreators error (urllib): {e}")
|
|
return {"items": [], "error": f"{type(e).__name__}: {e}"}
|
|
else:
|
|
try:
|
|
resp = _requests.get(
|
|
f"{SCRAPECREATORS_BASE}/v2/instagram/reels/search",
|
|
params={"query": core_topic},
|
|
headers=_sc_headers(token),
|
|
timeout=30,
|
|
)
|
|
resp.raise_for_status()
|
|
data = resp.json()
|
|
except Exception as e:
|
|
_log(f"ScrapeCreators error: {e}")
|
|
return {"items": [], "error": f"{type(e).__name__}: {e}"}
|
|
|
|
# Items are in the 'reels' array (ScrapeCreators v2 response)
|
|
raw_items = data.get("reels") or data.get("items") or data.get("data") or []
|
|
|
|
# Limit to configured count
|
|
raw_items = raw_items[:config["results_per_page"]]
|
|
|
|
# Parse items
|
|
items = []
|
|
for raw in raw_items:
|
|
if not isinstance(raw, dict):
|
|
continue
|
|
|
|
# Extract reel ID and shortcode
|
|
reel_pk = str(raw.get("id", raw.get("pk", "")))
|
|
shortcode = raw.get("shortcode", raw.get("code", ""))
|
|
|
|
# Caption text — can be a string or dict depending on endpoint
|
|
caption_obj = raw.get("caption", "")
|
|
if isinstance(caption_obj, dict):
|
|
text = caption_obj.get("text", "")
|
|
elif isinstance(caption_obj, str):
|
|
text = caption_obj
|
|
else:
|
|
text = raw.get("desc", raw.get("text", ""))
|
|
|
|
# Engagement metrics
|
|
play_count = raw.get("video_play_count") or raw.get("video_view_count") or raw.get("play_count") or 0
|
|
like_count = raw.get("like_count") or 0
|
|
comment_count = raw.get("comment_count") or 0
|
|
|
|
# Author info — 'owner' in reels/search, 'user' in user/reels
|
|
owner = raw.get("owner") or raw.get("user") or {}
|
|
author_name = owner.get("username", "")
|
|
|
|
# Duration
|
|
duration = raw.get("video_duration")
|
|
|
|
# Date
|
|
date_str = _parse_date(raw)
|
|
|
|
# Hashtags from caption text
|
|
hashtags = _extract_hashtags(text)
|
|
|
|
# Compute relevance with hashtag boost
|
|
relevance = _compute_relevance(core_topic, text, hashtags)
|
|
|
|
# Build URL — prefer API-provided url, fallback to shortcode
|
|
url = raw.get("url", "")
|
|
if not url and shortcode:
|
|
url = f"https://www.instagram.com/reel/{shortcode}"
|
|
|
|
items.append({
|
|
"video_id": reel_pk,
|
|
"text": text,
|
|
"url": url,
|
|
"author_name": author_name,
|
|
"date": date_str,
|
|
"engagement": {
|
|
"views": play_count,
|
|
"likes": like_count,
|
|
"comments": comment_count,
|
|
},
|
|
"hashtags": hashtags,
|
|
"duration": duration,
|
|
"relevance": relevance,
|
|
"why_relevant": f"Instagram: {text[:60]}" if text else f"Instagram: {core_topic}",
|
|
"caption_snippet": "", # populated by fetch_captions
|
|
})
|
|
|
|
# Hard date filter
|
|
in_range = [i for i in items if i["date"] and from_date <= i["date"] <= to_date]
|
|
out_of_range = len(items) - len(in_range)
|
|
if in_range:
|
|
items = in_range
|
|
if out_of_range:
|
|
_log(f"Filtered {out_of_range} reels outside date range")
|
|
else:
|
|
_log(f"No reels within date range, keeping all {len(items)}")
|
|
|
|
# Sort by views descending
|
|
items.sort(key=lambda x: x["engagement"]["views"], reverse=True)
|
|
|
|
_log(f"Found {len(items)} Instagram reels")
|
|
return {"items": items}
|
|
|
|
|
|
def fetch_captions(
|
|
video_items: List[Dict[str, Any]],
|
|
token: str,
|
|
depth: str = "default",
|
|
) -> Dict[str, str]:
|
|
"""Fetch transcripts for top N Instagram reels via ScrapeCreators.
|
|
|
|
Strategy:
|
|
1. Use the 'text' field (caption) as baseline
|
|
2. For top N, call /v2/instagram/media/transcript for spoken-word captions
|
|
|
|
Args:
|
|
video_items: Items from search_instagram()
|
|
token: ScrapeCreators API key
|
|
depth: Depth level for caption limit
|
|
|
|
Returns:
|
|
Dict mapping video_id -> caption text (truncated to 500 words)
|
|
"""
|
|
config = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"])
|
|
max_captions = config["max_captions"]
|
|
|
|
if not video_items or not token or not _requests:
|
|
return {}
|
|
|
|
top_items = video_items[:max_captions]
|
|
_log(f"Enriching captions for {len(top_items)} reels")
|
|
|
|
captions = {}
|
|
|
|
# First pass: use text field as caption (always available, free)
|
|
for item in top_items:
|
|
vid = item["video_id"]
|
|
text = item.get("text", "")
|
|
if text:
|
|
words = text.split()
|
|
if len(words) > CAPTION_MAX_WORDS:
|
|
text = ' '.join(words[:CAPTION_MAX_WORDS]) + '...'
|
|
captions[vid] = text
|
|
|
|
# Second pass: try to get spoken-word transcripts (1 credit each)
|
|
for item in top_items:
|
|
vid = item["video_id"]
|
|
url = item.get("url", "")
|
|
if not url:
|
|
continue
|
|
try:
|
|
resp = _requests.get(
|
|
f"{SCRAPECREATORS_BASE}/v2/instagram/media/transcript",
|
|
params={"url": url},
|
|
headers=_sc_headers(token),
|
|
timeout=15,
|
|
)
|
|
if resp.status_code == 200:
|
|
data = resp.json()
|
|
transcripts = data.get("transcripts") or []
|
|
if transcripts and isinstance(transcripts, list):
|
|
# Combine all transcript segments
|
|
transcript_text = " ".join(
|
|
t.get("text", "") for t in transcripts
|
|
if isinstance(t, dict) and t.get("text")
|
|
)
|
|
if transcript_text:
|
|
words = transcript_text.split()
|
|
if len(words) > CAPTION_MAX_WORDS:
|
|
transcript_text = ' '.join(words[:CAPTION_MAX_WORDS]) + '...'
|
|
captions[vid] = transcript_text
|
|
except Exception as e:
|
|
_log(f"Transcript fetch failed for {vid}: {e}")
|
|
|
|
got = sum(1 for v in captions.values() if v)
|
|
_log(f"Got captions for {got}/{len(top_items)} reels")
|
|
return captions
|
|
|
|
|
|
def search_and_enrich(
|
|
topic: str,
|
|
from_date: str,
|
|
to_date: str,
|
|
depth: str = "default",
|
|
token: str = None,
|
|
) -> Dict[str, Any]:
|
|
"""Full Instagram search: find reels, then fetch captions for top results.
|
|
|
|
Args:
|
|
topic: Search topic
|
|
from_date: Start date (YYYY-MM-DD)
|
|
to_date: End date (YYYY-MM-DD)
|
|
depth: 'quick', 'default', or 'deep'
|
|
token: ScrapeCreators API key
|
|
|
|
Returns:
|
|
Dict with 'items' list. Each item has a 'caption_snippet' field.
|
|
"""
|
|
# Step 1: Search
|
|
search_result = search_instagram(topic, from_date, to_date, depth, token)
|
|
items = search_result.get("items", [])
|
|
|
|
if not items:
|
|
return search_result
|
|
|
|
# Step 2: Fetch captions for top N
|
|
captions = fetch_captions(items, token, depth)
|
|
|
|
# Step 3: Attach captions to items
|
|
for item in items:
|
|
vid = item["video_id"]
|
|
caption = captions.get(vid)
|
|
if caption:
|
|
item["caption_snippet"] = caption
|
|
|
|
return {"items": items, "error": search_result.get("error")}
|
|
|
|
|
|
def parse_instagram_response(response: Dict[str, Any]) -> List[Dict[str, Any]]:
|
|
"""Parse Instagram search response to normalized format.
|
|
|
|
Returns:
|
|
List of item dicts ready for normalization.
|
|
"""
|
|
return response.get("items", [])
|