| 1 | """Instagram Reels search via ScrapeCreators API for /last30days. |
| 2 | |
| 3 | Uses ScrapeCreators REST API to search Instagram Reels by keyword, extract |
| 4 | engagement metrics (views, likes, comments), and fetch video transcripts. |
| 5 | |
| 6 | Requires SCRAPECREATORS_API_KEY in config. 100 free API calls, then PAYG. |
| 7 | API docs: https://scrapecreators.com/docs |
| 8 | """ |
| 9 | |
| 10 | import os |
| 11 | import re |
| 12 | import sys |
| 13 | from datetime import datetime |
| 14 | from typing import Any, Dict, List, Optional, Set |
| 15 | |
| 16 | from . import dates, http, log |
| 17 | from .query import infer_query_intent |
| 18 | from .relevance import token_overlap_relevance as _compute_relevance |
| 19 | |
| 20 | SCRAPECREATORS_BASE = "https://api.scrapecreators.com" |
| 21 | |
| 22 | # Depth configurations: how many results to fetch / captions to extract |
| 23 | DEPTH_CONFIG = { |
| 24 | "quick": {"results_per_page": 10, "max_captions": 3}, |
| 25 | "default": {"results_per_page": 20, "max_captions": 5}, |
| 26 | "deep": {"results_per_page": 40, "max_captions": 8}, |
| 27 | } |
| 28 | |
| 29 | # Max words to keep from each caption |
| 30 | CAPTION_MAX_WORDS = 500 |
| 31 | |
| 32 | # Default transcript fetch timeout (seconds). SC's |
| 33 | # /v2/instagram/media/transcript regularly takes >15s on real workloads, |
| 34 | # so the default is generous; override via LAST30DAYS_TRANSCRIPT_TIMEOUT. |
| 35 | DEFAULT_TRANSCRIPT_TIMEOUT = 30 |
| 36 | |
| 37 | |
| 38 | def _resolve_transcript_timeout( |
| 39 | timeout: Optional[float] = None, |
| 40 | config: Optional[Dict[str, Any]] = None, |
| 41 | ) -> float: |
| 42 | """Resolve the IG transcript-fetch timeout. |
| 43 | |
| 44 | Priority (highest wins): |
| 45 | 1. Explicit ``timeout`` kwarg |
| 46 | 2. ``LAST30DAYS_TRANSCRIPT_TIMEOUT`` in os.environ |
| 47 | 3. ``LAST30DAYS_TRANSCRIPT_TIMEOUT`` in caller-supplied config dict |
| 48 | 4. ``DEFAULT_TRANSCRIPT_TIMEOUT`` (30s) |
| 49 | |
| 50 | Mirrors the ``os.environ.get(X) or config.get(X)`` pattern used for |
| 51 | LAST30DAYS_STORE in last30days.py so the env var works whether it's |
| 52 | shell-exported or set in ~/.config/last30days/.env. |
| 53 | """ |
| 54 | if timeout is not None: |
| 55 | try: |
| 56 | return float(timeout) |
| 57 | except (TypeError, ValueError): |
| 58 | pass |
| 59 | raw = os.environ.get("LAST30DAYS_TRANSCRIPT_TIMEOUT") |
| 60 | if not raw and config: |
| 61 | raw = config.get("LAST30DAYS_TRANSCRIPT_TIMEOUT") |
| 62 | if raw: |
| 63 | try: |
| 64 | return float(raw) |
| 65 | except (TypeError, ValueError): |
| 66 | pass |
| 67 | return float(DEFAULT_TRANSCRIPT_TIMEOUT) |
| 68 | |
| 69 | |
| 70 | def _extract_core_subject(topic: str) -> str: |
| 71 | """Extract core subject from verbose query for Instagram search.""" |
| 72 | from .query import VIRAL_NOISE, extract_core_subject |
| 73 | return extract_core_subject(topic, noise=VIRAL_NOISE) |
| 74 | |
| 75 | |
| 76 | def _to_hashtag_form(query: str) -> str: |
| 77 | """Collapse a multi-word query to hashtag form (no spaces, lowercase). |
| 78 | |
| 79 | SC's /v2/instagram/reels/search wraps Google Search and is documented |
| 80 | to be flaky on multi-token queries. Single-token queries map to a |
| 81 | hashtag page lookup which is the stable path. Used as a 500-retry |
| 82 | fallback before the request bubbles up as a silent failure. |
| 83 | """ |
| 84 | return ''.join(query.split()).lower() |
| 85 | |
| 86 | |
| 87 | def expand_instagram_queries(topic: str, depth: str) -> List[str]: |
| 88 | """Generate multiple Instagram search queries from a topic. |
| 89 | |
| 90 | Mirrors reddit.py's expand_reddit_queries() pattern: |
| 91 | 1. Extract core subject (strip noise words) |
| 92 | 2. Include original topic if different from core |
| 93 | 3. Add intent-specific OR-joined content-type variants |
| 94 | 4. Cap by depth: 1 for quick, 2 for default, 3 for deep |
| 95 | |
| 96 | Returns 1-3 query strings depending on depth. |
| 97 | """ |
| 98 | core = _extract_core_subject(topic) |
| 99 | queries = [core] |
| 100 | |
| 101 | # Include cleaned original topic as variant if different from core |
| 102 | original_clean = topic.strip().rstrip('?!.') |
| 103 | if core.lower() != original_clean.lower() and len(original_clean.split()) <= 8: |
| 104 | queries.append(original_clean) |
| 105 | |
| 106 | qtype = infer_query_intent(topic) |
| 107 | |
| 108 | # Intent-specific Instagram content-type variants |
| 109 | if qtype == "breaking_news": |
| 110 | queries.append(f"{core} reaction OR edit") |
| 111 | elif qtype == "opinion": |
| 112 | queries.append(f"{core} reaction OR edit") |
| 113 | elif qtype == "product": |
| 114 | queries.append(f"{core} review OR haul") |
| 115 | elif qtype == "comparison": |
| 116 | queries.append(f"{core} vs OR compared") |
| 117 | elif qtype == "how_to": |
| 118 | queries.append(f"{core} tutorial OR hack") |
| 119 | else: |
| 120 | queries.append(f"{core} reaction OR edit") |
| 121 | |
| 122 | # Deep depth: add viral content variant |
| 123 | if depth == "deep": |
| 124 | queries.append(f"{core} viral OR trending OR reel") |
| 125 | |
| 126 | # Cap by depth budget |
| 127 | caps = {"quick": 1, "default": 2, "deep": 3} |
| 128 | cap = caps.get(depth, 2) |
| 129 | return queries[:cap] |
| 130 | |
| 131 | |
| 132 | def _log(msg: str): |
| 133 | log.source_log("Instagram", msg, tty_only=False) |
| 134 | |
| 135 | |
| 136 | def _parse_date(item: Dict[str, Any]) -> Optional[str]: |
| 137 | """Parse date from ScrapeCreators Instagram item to YYYY-MM-DD. |
| 138 | |
| 139 | Handles taken_at as ISO string (e.g. "2026-02-26T16:00:00.000Z") |
| 140 | or unix timestamp. |
| 141 | """ |
| 142 | ts = item.get("taken_at") |
| 143 | if not ts: |
| 144 | return None |
| 145 | |
| 146 | # Try ISO string first (ScrapeCreators reels/search returns this) |
| 147 | if isinstance(ts, str): |
| 148 | try: |
| 149 | # Handle "2026-02-26T16:00:00.000Z" format |
| 150 | dt = datetime.fromisoformat(ts.replace("Z", "+00:00")) |
| 151 | return dt.strftime("%Y-%m-%d") |
| 152 | except (ValueError, TypeError): |
| 153 | pass |
| 154 | # Try just the date portion |
| 155 | if len(ts) >= 10: |
| 156 | return ts[:10] |
| 157 | |
| 158 | # Fall back to unix timestamp |
| 159 | try: |
| 160 | return dates.timestamp_to_date(int(ts)) |
| 161 | except (ValueError, TypeError): |
| 162 | pass |
| 163 | |
| 164 | return None |
| 165 | |
| 166 | |
| 167 | def _extract_hashtags(caption_text: str) -> List[str]: |
| 168 | """Extract hashtags from Instagram caption text.""" |
| 169 | if not caption_text: |
| 170 | return [] |
| 171 | return re.findall(r'#(\w+)', caption_text) |
| 172 | |
| 173 | |
| 174 | def _parse_items(raw_items: List[Dict[str, Any]], core_topic: str) -> List[Dict[str, Any]]: |
| 175 | """Parse raw Instagram items into normalized dicts.""" |
| 176 | items = [] |
| 177 | for raw in raw_items: |
| 178 | if not isinstance(raw, dict): |
| 179 | continue |
| 180 | |
| 181 | # Extract reel ID and shortcode |
| 182 | reel_pk = str(raw.get("id", raw.get("pk", ""))) |
| 183 | shortcode = raw.get("shortcode", raw.get("code", "")) |
| 184 | |
| 185 | # Caption text -- can be a string or dict depending on endpoint |
| 186 | caption_obj = raw.get("caption", "") |
| 187 | if isinstance(caption_obj, dict): |
| 188 | text = caption_obj.get("text", "") |
| 189 | elif isinstance(caption_obj, str): |
| 190 | text = caption_obj |
| 191 | else: |
| 192 | text = raw.get("desc", raw.get("text", "")) |
| 193 | |
| 194 | # Engagement metrics |
| 195 | play_count = raw.get("video_play_count") or raw.get("video_view_count") or raw.get("play_count") or 0 |
| 196 | like_count = raw.get("like_count") or 0 |
| 197 | comment_count = raw.get("comment_count") or 0 |
| 198 | |
| 199 | # Author info -- 'owner' in reels/search, 'user' in user/reels |
| 200 | owner_raw = raw.get("owner") or raw.get("user") |
| 201 | if isinstance(owner_raw, dict): |
| 202 | author_name = owner_raw.get("username", "") |
| 203 | elif isinstance(owner_raw, str): |
| 204 | author_name = owner_raw |
| 205 | else: |
| 206 | author_name = "" |
| 207 | |
| 208 | # Duration |
| 209 | duration = raw.get("video_duration") |
| 210 | |
| 211 | # Date |
| 212 | date_str = _parse_date(raw) |
| 213 | |
| 214 | # Hashtags from caption text |
| 215 | hashtags = _extract_hashtags(text) |
| 216 | |
| 217 | # Compute relevance with hashtag boost |
| 218 | relevance = _compute_relevance(core_topic, text, hashtags) |
| 219 | |
| 220 | # Build URL -- prefer API-provided url, fallback to shortcode |
| 221 | url = raw.get("url", "") |
| 222 | if not url and shortcode: |
| 223 | url = f"https://www.instagram.com/reel/{shortcode}" |
| 224 | |
| 225 | items.append({ |
| 226 | "video_id": reel_pk, |
| 227 | "text": text, |
| 228 | "url": url, |
| 229 | "author_name": author_name, |
| 230 | "date": date_str, |
| 231 | "engagement": { |
| 232 | "views": play_count, |
| 233 | "likes": like_count, |
| 234 | "comments": comment_count, |
| 235 | }, |
| 236 | "hashtags": hashtags, |
| 237 | "duration": duration, |
| 238 | "relevance": relevance, |
| 239 | "why_relevant": f"Instagram: {text[:60]}" if text else f"Instagram: {core_topic}", |
| 240 | "caption_snippet": "", # populated by fetch_captions |
| 241 | }) |
| 242 | return items |
| 243 | |
| 244 | |
| 245 | def _user_reels( |
| 246 | handle: str, |
| 247 | token: str, |
| 248 | ) -> List[Dict[str, Any]]: |
| 249 | """Fetch an Instagram user's recent reels via ScrapeCreators. |
| 250 | |
| 251 | Args: |
| 252 | handle: Instagram username (without @) |
| 253 | token: ScrapeCreators API key |
| 254 | |
| 255 | Returns: |
| 256 | List of raw Instagram reel dicts. |
| 257 | """ |
| 258 | _log(f"User reels: @{handle}") |
| 259 | reels_url = f"{SCRAPECREATORS_BASE}/v1/instagram/user/reels" |
| 260 | try: |
| 261 | data = http.get( |
| 262 | reels_url, |
| 263 | params={"handle": handle}, |
| 264 | headers=http.scrapecreators_headers(token), |
| 265 | timeout=30, |
| 266 | retries=2, |
| 267 | ) |
| 268 | except Exception as e: |
| 269 | _log(f"User reels error for @{handle}: {e}") |
| 270 | return [] |
| 271 | |
| 272 | raw_items = data.get("items") or data.get("reels") or data.get("data") or [] |
| 273 | _log(f" -> {len(raw_items)} reels from @{handle}") |
| 274 | |
| 275 | # User-reels responses wrap each reel in a ``media`` envelope, while |
| 276 | # other Instagram endpoints already return flat reel dictionaries. |
| 277 | return [ |
| 278 | item["media"] |
| 279 | if isinstance(item, dict) and isinstance(item.get("media"), dict) |
| 280 | else item |
| 281 | for item in raw_items |
| 282 | ] |
| 283 | |
| 284 | |
| 285 | def search_instagram( |
| 286 | topic: str, |
| 287 | from_date: str, |
| 288 | to_date: str, |
| 289 | depth: str = "default", |
| 290 | token: str = None, |
| 291 | ) -> Dict[str, Any]: |
| 292 | """Search Instagram Reels via ScrapeCreators API. |
| 293 | |
| 294 | Args: |
| 295 | topic: Search topic |
| 296 | from_date: Start date (YYYY-MM-DD) |
| 297 | to_date: End date (YYYY-MM-DD) |
| 298 | depth: 'quick', 'default', or 'deep' |
| 299 | token: ScrapeCreators API key |
| 300 | |
| 301 | Returns: |
| 302 | Dict with 'items' list and optional 'error'. |
| 303 | """ |
| 304 | if not token: |
| 305 | return {"items": [], "error": "No SCRAPECREATORS_API_KEY configured"} |
| 306 | |
| 307 | config = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) |
| 308 | core_topic = _extract_core_subject(topic) |
| 309 | |
| 310 | _log(f"Searching Instagram for '{core_topic}' (depth={depth}, count={config['results_per_page']})") |
| 311 | |
| 312 | try: |
| 313 | data = http.get( |
| 314 | f"{SCRAPECREATORS_BASE}/v2/instagram/reels/search", |
| 315 | params={"query": core_topic}, |
| 316 | headers=http.scrapecreators_headers(token), |
| 317 | timeout=30, |
| 318 | retries=2, |
| 319 | ) |
| 320 | except http.HTTPError as e: |
| 321 | # SC's v2 reels search wraps Google Search and 500s frequently on |
| 322 | # multi-token queries. Single tokens hit the stable hashtag-page |
| 323 | # path. Retry once with hashtag form before bubbling up. |
| 324 | if getattr(e, "status_code", None) == 500 and ' ' in core_topic: |
| 325 | _log(f"IG search 500 on '{core_topic}', retrying with hashtag form") |
| 326 | try: |
| 327 | data = http.get( |
| 328 | f"{SCRAPECREATORS_BASE}/v2/instagram/reels/search", |
| 329 | params={"query": _to_hashtag_form(core_topic)}, |
| 330 | headers=http.scrapecreators_headers(token), |
| 331 | timeout=30, |
| 332 | retries=2, |
| 333 | ) |
| 334 | except Exception as retry_e: |
| 335 | _log(f"IG search retry failed: {retry_e}") |
| 336 | return {"items": [], "error": f"{type(retry_e).__name__}: {retry_e}"} |
| 337 | else: |
| 338 | _log(f"ScrapeCreators error: {e}") |
| 339 | return {"items": [], "error": f"{type(e).__name__}: {e}"} |
| 340 | except Exception as e: |
| 341 | _log(f"ScrapeCreators error: {e}") |
| 342 | return {"items": [], "error": f"{type(e).__name__}: {e}"} |
| 343 | |
| 344 | # Items are in the 'reels' array (ScrapeCreators v2 response) |
| 345 | raw_items = data.get("reels") or data.get("items") or data.get("data") or [] |
| 346 | |
| 347 | # Limit to configured count |
| 348 | raw_items = raw_items[:config["results_per_page"]] |
| 349 | |
| 350 | # Parse items |
| 351 | items = _parse_items(raw_items, core_topic) |
| 352 | |
| 353 | # Hard date filter |
| 354 | in_range = [i for i in items if i["date"] and from_date <= i["date"] <= to_date] |
| 355 | out_of_range = len(items) - len(in_range) |
| 356 | if in_range: |
| 357 | items = in_range |
| 358 | if out_of_range: |
| 359 | _log(f"Filtered {out_of_range} reels outside date range") |
| 360 | else: |
| 361 | _log(f"No reels within date range, keeping all {len(items)}") |
| 362 | |
| 363 | # Sort by views descending |
| 364 | items.sort(key=lambda x: x["engagement"]["views"], reverse=True) |
| 365 | |
| 366 | _log(f"Found {len(items)} Instagram reels") |
| 367 | return {"items": items} |
| 368 | |
| 369 | |
| 370 | def fetch_captions( |
| 371 | video_items: List[Dict[str, Any]], |
| 372 | token: str, |
| 373 | depth: str = "default", |
| 374 | timeout: Optional[float] = None, |
| 375 | config: Optional[Dict[str, Any]] = None, |
| 376 | ) -> Dict[str, str]: |
| 377 | """Fetch transcripts for top N Instagram reels via ScrapeCreators. |
| 378 | |
| 379 | Strategy: |
| 380 | 1. Use the 'text' field (caption) as baseline |
| 381 | 2. For top N, call /v2/instagram/media/transcript for spoken-word captions |
| 382 | |
| 383 | Args: |
| 384 | video_items: Items from search_instagram() |
| 385 | token: ScrapeCreators API key |
| 386 | depth: Depth level for caption limit |
| 387 | timeout: Optional per-request transcript timeout in seconds. When |
| 388 | None, resolves from LAST30DAYS_TRANSCRIPT_TIMEOUT (env or |
| 389 | config), defaulting to DEFAULT_TRANSCRIPT_TIMEOUT (30s). |
| 390 | config: Optional config dict (from env.get_config()) used as a |
| 391 | fallback source for LAST30DAYS_TRANSCRIPT_TIMEOUT when the |
| 392 | value is not exported in os.environ. |
| 393 | |
| 394 | Returns: |
| 395 | Dict mapping video_id -> caption text (truncated to 500 words) |
| 396 | """ |
| 397 | depth_cfg = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) |
| 398 | max_captions = depth_cfg["max_captions"] |
| 399 | transcript_timeout = _resolve_transcript_timeout(timeout, config) |
| 400 | |
| 401 | if not video_items or not token: |
| 402 | return {} |
| 403 | |
| 404 | top_items = video_items[:max_captions] |
| 405 | _log(f"Enriching captions for {len(top_items)} reels") |
| 406 | |
| 407 | captions = {} |
| 408 | |
| 409 | # First pass: use text field as caption (always available, free) |
| 410 | for item in top_items: |
| 411 | vid = item["video_id"] |
| 412 | text = item.get("text", "") |
| 413 | if text: |
| 414 | words = text.split() |
| 415 | if len(words) > CAPTION_MAX_WORDS: |
| 416 | text = ' '.join(words[:CAPTION_MAX_WORDS]) + '...' |
| 417 | captions[vid] = text |
| 418 | |
| 419 | # Second pass: try to get spoken-word transcripts (1 credit each) |
| 420 | for item in top_items: |
| 421 | vid = item["video_id"] |
| 422 | url = item.get("url", "") |
| 423 | if not url: |
| 424 | continue |
| 425 | try: |
| 426 | # Isolate transcript fetch errors from the pipeline-level |
| 427 | # capture_failures() context so an individual reel's 400 doesn't |
| 428 | # poison the entire source outcome (#829). |
| 429 | with http.capture_failures() as _tf: |
| 430 | data = http.get( |
| 431 | f"{SCRAPECREATORS_BASE}/v2/instagram/media/transcript", |
| 432 | params={"url": url}, |
| 433 | headers=http.scrapecreators_headers(token), |
| 434 | timeout=transcript_timeout, |
| 435 | retries=1, |
| 436 | ) |
| 437 | transcripts = data.get("transcripts") or [] |
| 438 | if transcripts and isinstance(transcripts, list): |
| 439 | transcript_text = " ".join( |
| 440 | t.get("text", "") for t in transcripts |
| 441 | if isinstance(t, dict) and t.get("text") |
| 442 | ) |
| 443 | if transcript_text: |
| 444 | words = transcript_text.split() |
| 445 | if len(words) > CAPTION_MAX_WORDS: |
| 446 | transcript_text = ' '.join(words[:CAPTION_MAX_WORDS]) + '...' |
| 447 | captions[vid] = transcript_text |
| 448 | except Exception as e: |
| 449 | _log(f"Transcript fetch failed for {vid}: {e}") |
| 450 | |
| 451 | got = sum(1 for v in captions.values() if v) |
| 452 | _log(f"Got captions for {got}/{len(top_items)} reels") |
| 453 | return captions |
| 454 | |
| 455 | |
| 456 | def search_and_enrich( |
| 457 | topic: str, |
| 458 | from_date: str, |
| 459 | to_date: str, |
| 460 | depth: str = "default", |
| 461 | token: str = None, |
| 462 | ig_creators: List[str] | None = None, |
| 463 | ) -> Dict[str, Any]: |
| 464 | """Full Instagram search: find reels, then fetch captions for top results. |
| 465 | |
| 466 | Uses expand_instagram_queries() to generate multiple search queries, |
| 467 | runs ScrapeCreators for each, and merges/deduplicates results by video ID. |
| 468 | |
| 469 | Args: |
| 470 | topic: Search topic (raw topic, not planner's narrowed query) |
| 471 | from_date: Start date (YYYY-MM-DD) |
| 472 | to_date: End date (YYYY-MM-DD) |
| 473 | depth: 'quick', 'default', or 'deep' |
| 474 | token: ScrapeCreators API key |
| 475 | ig_creators: Optional list of Instagram creator handles to fetch reels from |
| 476 | |
| 477 | Returns: |
| 478 | Dict with 'items' list. Each item has a 'caption_snippet' field. |
| 479 | """ |
| 480 | core_topic = _extract_core_subject(topic) |
| 481 | seen_ids: Set[str] = set() |
| 482 | items: List[Dict[str, Any]] = [] |
| 483 | last_error = None |
| 484 | |
| 485 | # Step 0: Creator reels (high-signal, runs first) |
| 486 | if ig_creators and token: |
| 487 | for creator in ig_creators: |
| 488 | raw_items = _user_reels(creator, token) |
| 489 | parsed = _parse_items(raw_items, core_topic) |
| 490 | for item in parsed: |
| 491 | vid = item.get("video_id", "") |
| 492 | if vid and vid not in seen_ids: |
| 493 | seen_ids.add(vid) |
| 494 | items.append(item) |
| 495 | |
| 496 | # Step 1: Multi-query keyword search — run ScrapeCreators for each expanded query |
| 497 | queries = expand_instagram_queries(topic, depth) |
| 498 | for q in queries: |
| 499 | search_result = search_instagram(q, from_date, to_date, depth, token) |
| 500 | if search_result.get("error"): |
| 501 | last_error = search_result["error"] |
| 502 | for item in search_result.get("items", []): |
| 503 | vid = item.get("video_id", "") |
| 504 | if vid and vid not in seen_ids: |
| 505 | seen_ids.add(vid) |
| 506 | items.append(item) |
| 507 | |
| 508 | # Sort merged results by views descending |
| 509 | items.sort(key=lambda x: x.get("engagement", {}).get("views") or 0, reverse=True) |
| 510 | |
| 511 | if not items: |
| 512 | return {"items": [], "error": last_error} |
| 513 | |
| 514 | # Step 2: Fetch captions for top N |
| 515 | captions = fetch_captions(items, token, depth) |
| 516 | |
| 517 | # Step 3: Attach captions to items |
| 518 | for item in items: |
| 519 | vid = item["video_id"] |
| 520 | caption = captions.get(vid) |
| 521 | if caption: |
| 522 | item["caption_snippet"] = caption |
| 523 | |
| 524 | return {"items": items, "error": last_error} |
| 525 | |
| 526 | |
| 527 | def parse_instagram_response(response: Dict[str, Any]) -> List[Dict[str, Any]]: |
| 528 | """Parse Instagram search response to normalized format. |
| 529 | |
| 530 | Returns: |
| 531 | List of item dicts ready for normalization. |
| 532 | """ |
| 533 | return response.get("items", []) |
| 534 | |
| 535 | |
| 536 | # --------------------------------------------------------------------------- |
| 537 | # Comments (ScrapeCreators, opt-in via INCLUDE_SOURCES=instagram_comments) |
| 538 | # --------------------------------------------------------------------------- |
| 539 | |
| 540 | |
| 541 | def _ig_total_engagement(item: Dict[str, Any]) -> int: |
| 542 | """Sum an Instagram item's engagement for picking which posts to enrich.""" |
| 543 | eng = item.get("engagement", {}) or {} |
| 544 | return (eng.get("views") or 0) + (eng.get("likes") or 0) + (eng.get("comments") or 0) |
| 545 | |
| 546 | |
| 547 | def enrich_with_comments( |
| 548 | items: List[Dict[str, Any]], |
| 549 | token: str, |
| 550 | max_posts: int = 3, |
| 551 | max_comments: int = 5, |
| 552 | ) -> List[Dict[str, Any]]: |
| 553 | """Enrich top Instagram posts with comment data from ScrapeCreators. |
| 554 | |
| 555 | Mirrors ``tiktok.enrich_with_comments`` / ``youtube_yt.enrich_with_comments``: |
| 556 | for the top N posts by engagement, fetch comments and attach them as a |
| 557 | ``top_comments`` field (highest-liked first). Failures never crash the run. |
| 558 | """ |
| 559 | if not items or not token or max_posts <= 0: |
| 560 | return items |
| 561 | |
| 562 | ranked = sorted(items, key=_ig_total_engagement, reverse=True) |
| 563 | top_items = ranked[:max_posts] |
| 564 | _log(f"Enriching comments for {len(top_items)} Instagram posts") |
| 565 | |
| 566 | from concurrent.futures import ThreadPoolExecutor, as_completed |
| 567 | |
| 568 | def _enrich_one(item: dict) -> bool: |
| 569 | post_url = item.get("url", "") |
| 570 | if not post_url: |
| 571 | return False |
| 572 | try: |
| 573 | comments = _fetch_post_comments(post_url, token, max_comments) |
| 574 | if comments: |
| 575 | item["top_comments"] = comments |
| 576 | return True |
| 577 | except Exception as exc: |
| 578 | _log(f"Comment enrichment failed for {post_url}: {exc}") |
| 579 | return False |
| 580 | |
| 581 | enriched_count = 0 |
| 582 | with ThreadPoolExecutor(max_workers=min(4, len(top_items))) as executor: |
| 583 | futures = {http.submit_with_context(executor, _enrich_one, item): item for item in top_items} |
| 584 | for future in as_completed(futures): |
| 585 | if future.result(): |
| 586 | enriched_count += 1 |
| 587 | |
| 588 | _log(f"Enriched {enriched_count}/{len(top_items)} posts with comments") |
| 589 | return items |
| 590 | |
| 591 | |
| 592 | def _fetch_post_comments( |
| 593 | post_url: str, |
| 594 | token: str, |
| 595 | max_comments: int = 5, |
| 596 | ) -> List[Dict[str, Any]]: |
| 597 | """Fetch comments for a single Instagram post/reel via ScrapeCreators. |
| 598 | |
| 599 | SC endpoint: GET /v2/instagram/post/comments?url=<post_or_reel_url> |
| 600 | Response shape: { comments: [{text, comment_like_count, child_comment_count, |
| 601 | created_at, user{username, ...}}], cursor } |
| 602 | |
| 603 | Returns: |
| 604 | List of comment dicts with author, text, comment_like_count (likes), date, |
| 605 | highest-liked first. Empty list on any error — never crashes the pipeline. |
| 606 | """ |
| 607 | try: |
| 608 | data = http.get( |
| 609 | f"{SCRAPECREATORS_BASE}/v2/instagram/post/comments", |
| 610 | params={"url": post_url}, |
| 611 | headers=http.scrapecreators_headers(token), |
| 612 | timeout=30, |
| 613 | retries=2, |
| 614 | ) |
| 615 | except Exception as exc: |
| 616 | _log(f"Comment fetch error for {post_url}: {exc}") |
| 617 | return [] |
| 618 | |
| 619 | raw_comments = data.get("comments") or data.get("data") or [] |
| 620 | # Sort by like count desc so normalize sees the highest-signal first. |
| 621 | raw_comments = sorted( |
| 622 | raw_comments, |
| 623 | key=lambda c: c.get("comment_like_count", 0) or 0, |
| 624 | reverse=True, |
| 625 | ) |
| 626 | out: List[Dict[str, Any]] = [] |
| 627 | for c in raw_comments[:max_comments]: |
| 628 | if not isinstance(c, dict): |
| 629 | continue |
| 630 | text = c.get("text") or "" |
| 631 | if not text: |
| 632 | continue |
| 633 | user = c.get("user") if isinstance(c.get("user"), dict) else {} |
| 634 | author = user.get("username") or "" |
| 635 | created_at = c.get("created_at") or "" |
| 636 | # created_at is ISO 8601 (e.g. "2026-07-04T14:27:58.000Z"); take the date. |
| 637 | date_str = created_at[:10] if isinstance(created_at, str) and len(created_at) >= 10 else "" |
| 638 | out.append({ |
| 639 | "author": author, |
| 640 | "text": text[:400], |
| 641 | "comment_like_count": c.get("comment_like_count", 0) or 0, |
| 642 | "date": date_str, |
| 643 | }) |
| 644 | return out |
| 645 |