| 1 | """TikTok search via ScrapeCreators API for /last30days. |
| 2 | |
| 3 | Uses ScrapeCreators REST API to search TikTok by keyword, extract engagement |
| 4 | metrics (views, likes, comments, shares), 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 re |
| 11 | import sys |
| 12 | from typing import Any, Dict, List, Optional, Set |
| 13 | |
| 14 | from . import dates, http, log |
| 15 | |
| 16 | SCRAPECREATORS_BASE = "https://api.scrapecreators.com/v1/tiktok" |
| 17 | |
| 18 | # Depth configurations: how many results to fetch / captions to extract |
| 19 | DEPTH_CONFIG = { |
| 20 | "quick": {"results_per_page": 10, "max_captions": 3}, |
| 21 | "default": {"results_per_page": 20, "max_captions": 5}, |
| 22 | "deep": {"results_per_page": 40, "max_captions": 8}, |
| 23 | } |
| 24 | |
| 25 | # Max words to keep from each caption |
| 26 | CAPTION_MAX_WORDS = 500 |
| 27 | |
| 28 | from .query import infer_query_intent |
| 29 | from .relevance import token_overlap_relevance as _compute_relevance |
| 30 | |
| 31 | |
| 32 | def _extract_core_subject(topic: str) -> str: |
| 33 | """Extract core subject from verbose query for TikTok search.""" |
| 34 | from .query import VIRAL_NOISE, extract_core_subject |
| 35 | return extract_core_subject(topic, noise=VIRAL_NOISE) |
| 36 | |
| 37 | |
| 38 | def expand_tiktok_queries(topic: str, depth: str) -> List[str]: |
| 39 | """Generate multiple TikTok search queries from a topic. |
| 40 | |
| 41 | Mirrors reddit.py's expand_reddit_queries() pattern: |
| 42 | 1. Extract core subject (strip noise words) |
| 43 | 2. Include original topic if different from core |
| 44 | 3. Add intent-specific OR-joined content-type variants |
| 45 | 4. Cap by depth: 1 for quick, 2 for default, 3 for deep |
| 46 | |
| 47 | Returns 1-3 query strings depending on depth. |
| 48 | """ |
| 49 | core = _extract_core_subject(topic) |
| 50 | queries = [core] |
| 51 | |
| 52 | # Include cleaned original topic as variant if different from core |
| 53 | original_clean = topic.strip().rstrip('?!.') |
| 54 | if core.lower() != original_clean.lower() and len(original_clean.split()) <= 8: |
| 55 | queries.append(original_clean) |
| 56 | |
| 57 | qtype = infer_query_intent(topic) |
| 58 | |
| 59 | # Intent-specific TikTok content-type variants |
| 60 | if qtype in ("breaking_news", "opinion"): |
| 61 | queries.append(f"{core} edit OR reaction OR trend") |
| 62 | elif qtype == "product": |
| 63 | queries.append(f"{core} review OR haul OR unboxing") |
| 64 | elif qtype == "comparison": |
| 65 | queries.append(f"{core} vs OR compared OR which is better") |
| 66 | elif qtype == "how_to": |
| 67 | queries.append(f"{core} tutorial OR hack OR tip") |
| 68 | else: |
| 69 | queries.append(f"{core} edit OR reaction OR trend") |
| 70 | |
| 71 | # Deep depth: add viral content variant |
| 72 | if depth == "deep": |
| 73 | queries.append(f"{core} viral OR fyp OR trending") |
| 74 | |
| 75 | # Cap by depth budget |
| 76 | caps = {"quick": 1, "default": 2, "deep": 3} |
| 77 | cap = caps.get(depth, 2) |
| 78 | return queries[:cap] |
| 79 | |
| 80 | |
| 81 | def _log(msg: str): |
| 82 | log.source_log("TikTok", msg, tty_only=False) |
| 83 | |
| 84 | |
| 85 | def _parse_date(item: Dict[str, Any]) -> Optional[str]: |
| 86 | """Parse date from ScrapeCreators TikTok item to YYYY-MM-DD.""" |
| 87 | ts = item.get("create_time") |
| 88 | if ts: |
| 89 | try: |
| 90 | return dates.timestamp_to_date(int(ts)) |
| 91 | except (ValueError, TypeError): |
| 92 | pass |
| 93 | return None |
| 94 | |
| 95 | |
| 96 | def _clean_webvtt(text: str) -> str: |
| 97 | """Strip WebVTT timestamps and headers from transcript text.""" |
| 98 | if not text: |
| 99 | return "" |
| 100 | lines = text.split('\n') |
| 101 | cleaned = [] |
| 102 | for line in lines: |
| 103 | line = line.strip() |
| 104 | if not line: |
| 105 | continue |
| 106 | if line.startswith('WEBVTT'): |
| 107 | continue |
| 108 | if re.match(r'^\d{2}:\d{2}', line): |
| 109 | continue |
| 110 | if '-->' in line: |
| 111 | continue |
| 112 | cleaned.append(line) |
| 113 | return ' '.join(cleaned) |
| 114 | |
| 115 | |
| 116 | def _parse_items(raw_items: List[Dict[str, Any]], core_topic: str) -> List[Dict[str, Any]]: |
| 117 | """Parse raw TikTok items into normalized dicts.""" |
| 118 | items = [] |
| 119 | for raw in raw_items: |
| 120 | video_id = str(raw.get("aweme_id", "")) |
| 121 | text = raw.get("desc", "") |
| 122 | |
| 123 | stats = raw.get("statistics") if isinstance(raw.get("statistics"), dict) else {} |
| 124 | play_count = stats.get("play_count") if stats.get("play_count") is not None else 0 |
| 125 | digg_count = stats.get("digg_count") if stats.get("digg_count") is not None else 0 |
| 126 | comment_count = stats.get("comment_count") if stats.get("comment_count") is not None else 0 |
| 127 | share_count = stats.get("share_count") if stats.get("share_count") is not None else 0 |
| 128 | |
| 129 | author_raw = raw.get("author") |
| 130 | if isinstance(author_raw, dict): |
| 131 | author_name = author_raw.get("unique_id", "") |
| 132 | elif isinstance(author_raw, str): |
| 133 | author_name = author_raw |
| 134 | else: |
| 135 | author_name = "" |
| 136 | |
| 137 | share_url = raw.get("share_url", "") |
| 138 | text_extra = raw.get("text_extra") or [] |
| 139 | hashtag_names = [t.get("hashtag_name", "") for t in text_extra |
| 140 | if isinstance(t, dict) and t.get("hashtag_name")] |
| 141 | |
| 142 | video_raw = raw.get("video") |
| 143 | duration = video_raw.get("duration") if isinstance(video_raw, dict) else None |
| 144 | |
| 145 | date_str = _parse_date(raw) |
| 146 | |
| 147 | # Compute relevance with hashtag boost |
| 148 | relevance = _compute_relevance(core_topic, text, hashtag_names) |
| 149 | |
| 150 | # Build URL: prefer share_url, fallback to constructed URL |
| 151 | url = share_url.split("?")[0] if share_url else "" |
| 152 | if not url and author_name and video_id: |
| 153 | url = f"https://www.tiktok.com/@{author_name}/video/{video_id}" |
| 154 | |
| 155 | items.append({ |
| 156 | "video_id": video_id, |
| 157 | "text": text, |
| 158 | "url": url, |
| 159 | "author_name": author_name, |
| 160 | "date": date_str, |
| 161 | "engagement": { |
| 162 | "views": play_count, |
| 163 | "likes": digg_count, |
| 164 | "comments": comment_count, |
| 165 | "shares": share_count, |
| 166 | }, |
| 167 | "hashtags": hashtag_names, |
| 168 | "duration": duration, |
| 169 | "relevance": relevance, |
| 170 | "why_relevant": f"TikTok: {text[:60]}" if text else f"TikTok: {core_topic}", |
| 171 | "caption_snippet": "", # populated by fetch_captions |
| 172 | }) |
| 173 | return items |
| 174 | |
| 175 | |
| 176 | def _hashtag_search( |
| 177 | hashtag: str, |
| 178 | token: str, |
| 179 | ) -> List[Dict[str, Any]]: |
| 180 | """Search TikTok by hashtag via ScrapeCreators. |
| 181 | |
| 182 | Args: |
| 183 | hashtag: Hashtag name (without #) |
| 184 | token: ScrapeCreators API key |
| 185 | |
| 186 | Returns: |
| 187 | List of raw TikTok item dicts (aweme_info format). |
| 188 | """ |
| 189 | _log(f"Hashtag search: #{hashtag}") |
| 190 | try: |
| 191 | data = http.get( |
| 192 | f"{SCRAPECREATORS_BASE}/search/hashtag", |
| 193 | params={"hashtag": hashtag}, |
| 194 | headers=http.scrapecreators_headers(token), |
| 195 | timeout=30, |
| 196 | retries=2, |
| 197 | ) |
| 198 | except Exception as e: |
| 199 | _log(f"Hashtag search error for #{hashtag}: {e}") |
| 200 | return [] |
| 201 | |
| 202 | raw_items = data.get("aweme_list") or data.get("data") or [] |
| 203 | _log(f" -> {len(raw_items)} results for #{hashtag}") |
| 204 | return raw_items |
| 205 | |
| 206 | |
| 207 | def _profile_videos( |
| 208 | handle: str, |
| 209 | token: str, |
| 210 | count: int = 10, |
| 211 | ) -> List[Dict[str, Any]]: |
| 212 | """Fetch a TikTok creator's recent videos via ScrapeCreators. |
| 213 | |
| 214 | Args: |
| 215 | handle: TikTok username (without @) |
| 216 | token: ScrapeCreators API key |
| 217 | count: Max videos to return |
| 218 | |
| 219 | Returns: |
| 220 | List of raw TikTok item dicts (aweme_info format). |
| 221 | """ |
| 222 | _log(f"Profile videos: @{handle}") |
| 223 | profile_url = "https://api.scrapecreators.com/v3/tiktok/profile/videos" |
| 224 | try: |
| 225 | data = http.get( |
| 226 | profile_url, |
| 227 | params={"handle": handle, "sort_by": "latest"}, |
| 228 | headers=http.scrapecreators_headers(token), |
| 229 | timeout=30, |
| 230 | retries=2, |
| 231 | ) |
| 232 | except Exception as e: |
| 233 | _log(f"Profile videos error for @{handle}: {e}") |
| 234 | return [] |
| 235 | |
| 236 | raw_items = data.get("aweme_list") or data.get("data") or [] |
| 237 | _log(f" -> {len(raw_items)} videos from @{handle}") |
| 238 | return raw_items[:count] |
| 239 | |
| 240 | |
| 241 | def search_tiktok( |
| 242 | topic: str, |
| 243 | from_date: str, |
| 244 | to_date: str, |
| 245 | depth: str = "default", |
| 246 | token: str = None, |
| 247 | ) -> Dict[str, Any]: |
| 248 | """Search TikTok via ScrapeCreators API. |
| 249 | |
| 250 | Args: |
| 251 | topic: Search topic |
| 252 | from_date: Start date (YYYY-MM-DD) |
| 253 | to_date: End date (YYYY-MM-DD) |
| 254 | depth: 'quick', 'default', or 'deep' |
| 255 | token: ScrapeCreators API key |
| 256 | |
| 257 | Returns: |
| 258 | Dict with 'items' list and optional 'error'. |
| 259 | """ |
| 260 | if not token: |
| 261 | return {"items": [], "error": "No SCRAPECREATORS_API_KEY configured"} |
| 262 | |
| 263 | config = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) |
| 264 | core_topic = _extract_core_subject(topic) |
| 265 | |
| 266 | _log(f"Searching TikTok for '{core_topic}' (depth={depth}, count={config['results_per_page']})") |
| 267 | |
| 268 | try: |
| 269 | data = http.get( |
| 270 | f"{SCRAPECREATORS_BASE}/search/keyword", |
| 271 | params={"query": core_topic, "sort_by": "relevance"}, |
| 272 | headers=http.scrapecreators_headers(token), |
| 273 | timeout=30, |
| 274 | retries=2, |
| 275 | ) |
| 276 | except Exception as e: |
| 277 | _log(f"ScrapeCreators error: {e}") |
| 278 | return {"items": [], "error": f"{type(e).__name__}: {e}"} |
| 279 | |
| 280 | # Items are nested under aweme_info |
| 281 | raw_entries = data.get("search_item_list") or data.get("data") or [] |
| 282 | raw_items = [] |
| 283 | for entry in raw_entries: |
| 284 | if isinstance(entry, dict): |
| 285 | info = entry.get("aweme_info", entry) |
| 286 | raw_items.append(info) |
| 287 | |
| 288 | # Limit to configured count |
| 289 | raw_items = raw_items[:config["results_per_page"]] |
| 290 | |
| 291 | # Parse items |
| 292 | items = _parse_items(raw_items, core_topic) |
| 293 | |
| 294 | # Hard date filter |
| 295 | in_range = [i for i in items if i["date"] and from_date <= i["date"] <= to_date] |
| 296 | out_of_range = len(items) - len(in_range) |
| 297 | if in_range: |
| 298 | items = in_range |
| 299 | if out_of_range: |
| 300 | _log(f"Filtered {out_of_range} videos outside date range") |
| 301 | else: |
| 302 | _log(f"No videos within date range, keeping all {len(items)}") |
| 303 | |
| 304 | # Sort by views descending |
| 305 | items.sort(key=lambda x: x["engagement"]["views"], reverse=True) |
| 306 | |
| 307 | _log(f"Found {len(items)} TikTok videos") |
| 308 | return {"items": items} |
| 309 | |
| 310 | |
| 311 | def fetch_captions( |
| 312 | video_items: List[Dict[str, Any]], |
| 313 | token: str, |
| 314 | depth: str = "default", |
| 315 | ) -> Dict[str, str]: |
| 316 | """Fetch transcripts for top N TikTok videos via ScrapeCreators. |
| 317 | |
| 318 | Strategy: |
| 319 | 1. Use the 'text' field (video description) as baseline caption |
| 320 | 2. For top N, call /video/transcript for spoken-word captions |
| 321 | |
| 322 | Args: |
| 323 | video_items: Items from search_tiktok() |
| 324 | token: ScrapeCreators API key |
| 325 | depth: Depth level for caption limit |
| 326 | |
| 327 | Returns: |
| 328 | Dict mapping video_id -> caption text (truncated to 500 words) |
| 329 | """ |
| 330 | config = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) |
| 331 | max_captions = config["max_captions"] |
| 332 | |
| 333 | if not video_items or not token: |
| 334 | return {} |
| 335 | |
| 336 | top_items = video_items[:max_captions] |
| 337 | _log(f"Enriching captions for {len(top_items)} videos") |
| 338 | |
| 339 | captions = {} |
| 340 | |
| 341 | # First pass: use text field as caption (always available, free) |
| 342 | for item in top_items: |
| 343 | vid = item["video_id"] |
| 344 | text = item.get("text", "") |
| 345 | if text: |
| 346 | words = text.split() |
| 347 | if len(words) > CAPTION_MAX_WORDS: |
| 348 | text = ' '.join(words[:CAPTION_MAX_WORDS]) + '...' |
| 349 | captions[vid] = text |
| 350 | |
| 351 | # Second pass: try to get spoken-word transcripts (1 credit each) |
| 352 | for item in top_items: |
| 353 | vid = item["video_id"] |
| 354 | url = item.get("url", "") |
| 355 | if not url: |
| 356 | continue |
| 357 | try: |
| 358 | # Isolate transcript fetch errors from the pipeline-level |
| 359 | # capture_failures() context so an individual video's 400 |
| 360 | # doesn't poison the entire source outcome. |
| 361 | with http.capture_failures() as _tf: |
| 362 | data = http.get( |
| 363 | f"{SCRAPECREATORS_BASE}/video/transcript", |
| 364 | params={"url": url}, |
| 365 | headers=http.scrapecreators_headers(token), |
| 366 | timeout=15, |
| 367 | retries=1, |
| 368 | ) |
| 369 | transcript = data.get("transcript") |
| 370 | if transcript: |
| 371 | if isinstance(transcript, list): |
| 372 | transcript = " ".join(str(s) for s in transcript) |
| 373 | transcript = _clean_webvtt(transcript) |
| 374 | if transcript: |
| 375 | words = transcript.split() |
| 376 | if len(words) > CAPTION_MAX_WORDS: |
| 377 | transcript = ' '.join(words[:CAPTION_MAX_WORDS]) + '...' |
| 378 | captions[vid] = transcript |
| 379 | except Exception as e: |
| 380 | _log(f"Transcript fetch failed for {vid}: {e}") |
| 381 | |
| 382 | got = sum(1 for v in captions.values() if v) |
| 383 | _log(f"Got captions for {got}/{len(top_items)} videos") |
| 384 | return captions |
| 385 | |
| 386 | |
| 387 | def search_and_enrich( |
| 388 | topic: str, |
| 389 | from_date: str, |
| 390 | to_date: str, |
| 391 | depth: str = "default", |
| 392 | token: str = None, |
| 393 | hashtags: List[str] | None = None, |
| 394 | creators: List[str] | None = None, |
| 395 | ) -> Dict[str, Any]: |
| 396 | """Full TikTok search: find videos, then fetch captions for top results. |
| 397 | |
| 398 | Uses expand_tiktok_queries() to generate multiple search queries, |
| 399 | runs ScrapeCreators for each, and merges/deduplicates results by video ID. |
| 400 | |
| 401 | Args: |
| 402 | topic: Search topic (raw topic, not planner's narrowed query) |
| 403 | from_date: Start date (YYYY-MM-DD) |
| 404 | to_date: End date (YYYY-MM-DD) |
| 405 | depth: 'quick', 'default', or 'deep' |
| 406 | token: ScrapeCreators API key |
| 407 | hashtags: Optional list of TikTok hashtags to search (without #) |
| 408 | creators: Optional list of TikTok creator handles to fetch videos from |
| 409 | |
| 410 | Returns: |
| 411 | Dict with 'items' list. Each item has a 'caption_snippet' field. |
| 412 | """ |
| 413 | core_topic = _extract_core_subject(topic) |
| 414 | seen_ids: Set[str] = set() |
| 415 | items: List[Dict[str, Any]] = [] |
| 416 | last_error = None |
| 417 | |
| 418 | # Step 0a: Hashtag search (high-signal, runs first) |
| 419 | if hashtags and token: |
| 420 | for hashtag in hashtags: |
| 421 | raw_items = _hashtag_search(hashtag, token) |
| 422 | parsed = _parse_items(raw_items, core_topic) |
| 423 | for item in parsed: |
| 424 | vid = item.get("video_id", "") |
| 425 | if vid and vid not in seen_ids: |
| 426 | seen_ids.add(vid) |
| 427 | items.append(item) |
| 428 | |
| 429 | # Step 0b: Creator profile videos (high-signal) |
| 430 | if creators and token: |
| 431 | for creator in creators: |
| 432 | raw_items = _profile_videos(creator, token) |
| 433 | parsed = _parse_items(raw_items, core_topic) |
| 434 | for item in parsed: |
| 435 | vid = item.get("video_id", "") |
| 436 | if vid and vid not in seen_ids: |
| 437 | seen_ids.add(vid) |
| 438 | items.append(item) |
| 439 | |
| 440 | # Step 1: Multi-query keyword search — run ScrapeCreators for each expanded query |
| 441 | queries = expand_tiktok_queries(topic, depth) |
| 442 | for q in queries: |
| 443 | search_result = search_tiktok(q, from_date, to_date, depth, token) |
| 444 | if search_result.get("error"): |
| 445 | last_error = search_result["error"] |
| 446 | for item in search_result.get("items", []): |
| 447 | vid = item.get("video_id", "") |
| 448 | if vid and vid not in seen_ids: |
| 449 | seen_ids.add(vid) |
| 450 | items.append(item) |
| 451 | |
| 452 | # Sort merged results by views descending |
| 453 | items.sort(key=lambda x: x.get("engagement", {}).get("views") or 0, reverse=True) |
| 454 | |
| 455 | if not items: |
| 456 | return {"items": [], "error": last_error} |
| 457 | |
| 458 | # Step 2: Fetch captions for top N |
| 459 | captions = fetch_captions(items, token, depth) |
| 460 | |
| 461 | # Step 3: Attach captions to items |
| 462 | for item in items: |
| 463 | vid = item["video_id"] |
| 464 | caption = captions.get(vid) |
| 465 | if caption: |
| 466 | item["caption_snippet"] = caption |
| 467 | |
| 468 | return {"items": items, "error": last_error} |
| 469 | |
| 470 | |
| 471 | def parse_tiktok_response(response: Dict[str, Any]) -> List[Dict[str, Any]]: |
| 472 | """Parse TikTok search response to normalized format. |
| 473 | |
| 474 | Returns: |
| 475 | List of item dicts ready for normalization. |
| 476 | """ |
| 477 | return response.get("items", []) |
| 478 | |
| 479 | |
| 480 | def _tiktok_total_engagement(item: Dict[str, Any]) -> int: |
| 481 | """Total engagement for ranking which posts deserve comment enrichment.""" |
| 482 | eng = item.get("engagement", {}) |
| 483 | return (eng.get("views", 0) or 0) + (eng.get("likes", 0) or 0) + (eng.get("comments", 0) or 0) |
| 484 | |
| 485 | |
| 486 | def enrich_with_comments( |
| 487 | items: List[Dict[str, Any]], |
| 488 | token: str, |
| 489 | max_posts: int = 3, |
| 490 | max_comments: int = 5, |
| 491 | ) -> List[Dict[str, Any]]: |
| 492 | """Enrich top TikTok posts with comment data from ScrapeCreators. |
| 493 | |
| 494 | For the top N posts by engagement, fetches comments via the SC API |
| 495 | and attaches them as a ``top_comments`` field on each item. Mirrors |
| 496 | youtube_yt.enrich_with_comments. |
| 497 | |
| 498 | Args: |
| 499 | items: TikTok items from search_tiktok() |
| 500 | token: ScrapeCreators API key |
| 501 | max_posts: How many posts to enrich with comments |
| 502 | max_comments: Max comments to keep per post |
| 503 | |
| 504 | Returns: |
| 505 | Items list (mutated in place) with top_comments added to enriched items. |
| 506 | """ |
| 507 | if not items or not token or max_posts <= 0: |
| 508 | return items |
| 509 | |
| 510 | ranked = sorted(items, key=_tiktok_total_engagement, reverse=True) |
| 511 | top_items = ranked[:max_posts] |
| 512 | _log(f"Enriching comments for {len(top_items)} TikTok posts") |
| 513 | |
| 514 | from concurrent.futures import ThreadPoolExecutor, as_completed |
| 515 | |
| 516 | def _enrich_one(item: dict) -> bool: |
| 517 | post_url = item.get("url", "") |
| 518 | if not post_url: |
| 519 | return False |
| 520 | try: |
| 521 | comments = _fetch_post_comments(post_url, token, max_comments) |
| 522 | if comments: |
| 523 | item["top_comments"] = comments |
| 524 | return True |
| 525 | except Exception as exc: |
| 526 | _log(f"Comment enrichment failed for {post_url}: {exc}") |
| 527 | return False |
| 528 | |
| 529 | enriched_count = 0 |
| 530 | with ThreadPoolExecutor(max_workers=min(4, len(top_items))) as executor: |
| 531 | futures = {http.submit_with_context(executor, _enrich_one, item): item for item in top_items} |
| 532 | for future in as_completed(futures): |
| 533 | if future.result(): |
| 534 | enriched_count += 1 |
| 535 | |
| 536 | _log(f"Enriched {enriched_count}/{len(top_items)} posts with comments") |
| 537 | return items |
| 538 | |
| 539 | |
| 540 | def _fetch_post_comments( |
| 541 | post_url: str, |
| 542 | token: str, |
| 543 | max_comments: int = 5, |
| 544 | ) -> List[Dict[str, Any]]: |
| 545 | """Fetch comments for a single TikTok post via ScrapeCreators. |
| 546 | |
| 547 | SC endpoint: GET /v1/tiktok/video/comments?url=<video_url> |
| 548 | Response shape: { comments: [{text, user.nickname, digg_count, create_time, ...}], cursor, total } |
| 549 | |
| 550 | Args: |
| 551 | post_url: Canonical TikTok post URL (share_url form works) |
| 552 | token: ScrapeCreators API key |
| 553 | max_comments: Maximum comments to return |
| 554 | |
| 555 | Returns: |
| 556 | List of comment dicts with author, text, digg_count (likes), date. |
| 557 | Empty list on any error — comment failures never crash the pipeline. |
| 558 | """ |
| 559 | try: |
| 560 | data = http.get( |
| 561 | f"{SCRAPECREATORS_BASE}/video/comments", |
| 562 | params={"url": post_url, "trim": "true"}, |
| 563 | headers=http.scrapecreators_headers(token), |
| 564 | timeout=30, |
| 565 | retries=2, |
| 566 | ) |
| 567 | except Exception as exc: |
| 568 | _log(f"Comment fetch error for {post_url}: {exc}") |
| 569 | return [] |
| 570 | |
| 571 | raw_comments = data.get("comments") or data.get("data") or [] |
| 572 | # Sort by digg_count desc so normalize sees the highest-signal first. |
| 573 | raw_comments = sorted( |
| 574 | raw_comments, |
| 575 | key=lambda c: c.get("digg_count", 0) or 0, |
| 576 | reverse=True, |
| 577 | ) |
| 578 | out: List[Dict[str, Any]] = [] |
| 579 | for c in raw_comments[:max_comments]: |
| 580 | text = c.get("text") or "" |
| 581 | if not text: |
| 582 | continue |
| 583 | user = c.get("user") if isinstance(c.get("user"), dict) else {} |
| 584 | # Prefer unique_id (the @handle) over nickname (display name) so |
| 585 | # downstream render can cite @handle consistently across platforms. |
| 586 | author = user.get("unique_id") or user.get("nickname") or "" |
| 587 | create_time = c.get("create_time") |
| 588 | date_str = "" |
| 589 | if create_time: |
| 590 | try: |
| 591 | date_str = dates.timestamp_to_date(int(create_time)) or "" |
| 592 | except (ValueError, TypeError): |
| 593 | date_str = "" |
| 594 | out.append({ |
| 595 | "author": author, |
| 596 | "text": text[:400], |
| 597 | "digg_count": c.get("digg_count", 0) or 0, |
| 598 | "date": date_str, |
| 599 | }) |
| 600 | return out |
| 601 |