diff --git a/scripts/last30days.py b/scripts/last30days.py index f57c813..dc9c926 100644 --- a/scripts/last30days.py +++ b/scripts/last30days.py @@ -151,6 +151,7 @@ from lib import ( render, schema, score, + scrapecreators_x, ui, tiktok, instagram, @@ -357,6 +358,25 @@ def _search_x( return x_items, raw_response, x_error + # Use ScrapeCreators if specified + if x_source == "scrapecreators": + try: + raw_response = scrapecreators_x.search_x( + topic, from_date, to_date, + depth=depth, + token=config.get("SCRAPECREATORS_API_KEY"), + ) + except Exception as e: + raw_response = {"error": str(e)} + x_error = f"{type(e).__name__}: {e}" + + x_items = scrapecreators_x.parse_x_response(raw_response or {}) + + if raw_response and isinstance(raw_response, dict) and raw_response.get("error") and not x_error: + x_error = raw_response["error"] + + return x_items, raw_response, x_error + # Use xAI (original behavior) try: raw_response = xai_x.search_x( @@ -1481,14 +1501,14 @@ def main(): # Check available sources (accounting for Bird auto-detection) available = env.get_available_sources(config) - # Override available if Bird is ready - if x_source == 'bird': + # Override available if Bird or ScrapeCreators provides X + if x_source in ('bird', 'scrapecreators'): if available == 'reddit': - available = 'both' # Now have both Reddit + X (via Bird) + available = 'both' # Now have both Reddit + X elif available == 'reddit-web': - available = 'all' # Reddit + X (via Bird) + Web + available = 'all' # Reddit + X + Web elif available == 'web': - available = 'x-web' # X via Bird + Web + available = 'x-web' # X + Web # Mock mode can work without keys if args.mock: @@ -1724,7 +1744,7 @@ def main(): if x_source_status["bird_installed"]: source_info["x_skip_reason"] = "Bird installed but not authenticated — log into x.com in browser" else: - source_info["x_skip_reason"] = "No Bird CLI or XAI_API_KEY (Node.js 22+ needed for Bird)" + source_info["x_skip_reason"] = "No Bird CLI, XAI_API_KEY, or SCRAPECREATORS_API_KEY" if not has_ytdlp: source_info["youtube_skip_reason"] = "yt-dlp not installed — fix: brew install yt-dlp" elif has_ytdlp and not report.youtube: diff --git a/scripts/lib/env.py b/scripts/lib/env.py index 3cc8837..a270439 100644 --- a/scripts/lib/env.py +++ b/scripts/lib/env.py @@ -355,7 +355,8 @@ def get_missing_keys(config: Dict[str, Any]) -> str: from . import bird_x has_bird = bird_x.is_bird_installed() and bird_x.is_bird_authenticated() - has_x = has_xai or has_bird + has_sc_x = bool(config.get('SCRAPECREATORS_API_KEY')) + has_x = has_xai or has_bird or has_sc_x if has_reddit and has_x and has_web: return 'none' @@ -434,7 +435,7 @@ def validate_sources(requested: str, available: str, include_web: bool = False) def get_x_source(config: Dict[str, Any]) -> Optional[str]: """Determine the best available X/Twitter source. - Priority: Bird (free) → xAI (paid API) + Priority: Bird (free) → xAI (paid API) → ScrapeCreators (shared key) Args: config: Configuration dict from get_config() @@ -442,6 +443,7 @@ def get_x_source(config: Dict[str, Any]) -> Optional[str]: Returns: 'bird' if Bird is installed and authenticated, 'xai' if XAI_API_KEY is configured, + 'scrapecreators' if SCRAPECREATORS_API_KEY is configured, None if no X source available. """ # Import here to avoid circular dependency @@ -457,6 +459,10 @@ def get_x_source(config: Dict[str, Any]) -> Optional[str]: if config.get('XAI_API_KEY'): return 'xai' + # Fall back to ScrapeCreators (same key as Reddit/TikTok/Instagram) + if config.get('SCRAPECREATORS_API_KEY'): + return 'scrapecreators' + return None @@ -559,11 +565,15 @@ def get_x_source_status(config: Dict[str, Any]) -> Dict[str, Any]: bird_status = bird_x.get_bird_status() xai_available = bool(config.get('XAI_API_KEY')) + sc_available = bool(config.get('SCRAPECREATORS_API_KEY')) + # Determine active source if bird_status["authenticated"]: source = 'bird' elif xai_available: source = 'xai' + elif sc_available: + source = 'scrapecreators' else: source = None @@ -573,5 +583,6 @@ def get_x_source_status(config: Dict[str, Any]) -> Dict[str, Any]: "bird_authenticated": bird_status["authenticated"], "bird_username": bird_status["username"], "xai_available": xai_available, + "scrapecreators_available": sc_available, "can_install_bird": bird_status["can_install"], } diff --git a/scripts/lib/scrapecreators_x.py b/scripts/lib/scrapecreators_x.py new file mode 100644 index 0000000..3abd7ef --- /dev/null +++ b/scripts/lib/scrapecreators_x.py @@ -0,0 +1,231 @@ +"""X/Twitter search via ScrapeCreators API for /last30days. + +Uses ScrapeCreators REST API to search Twitter/X by keyword. +Same API key as Reddit, TikTok, and Instagram - one key covers all social sources. + +Requires SCRAPECREATORS_API_KEY in config. +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 + +SCRAPECREATORS_BASE = "https://api.scrapecreators.com/v1/twitter" + +DEPTH_CONFIG = { + "quick": {"results_per_page": 10}, + "default": {"results_per_page": 20}, + "deep": {"results_per_page": 40}, +} + +STOPWORDS = frozenset({ + 'the', 'a', 'an', 'to', 'for', 'how', 'is', 'in', 'of', 'on', + 'and', 'with', 'from', 'by', 'at', 'this', 'that', 'it', 'my', + 'your', 'i', 'me', 'we', 'you', 'what', 'are', 'do', 'can', + 'its', 'be', 'or', 'not', 'no', 'so', 'if', 'but', 'about', + 'all', 'just', 'get', 'has', 'have', 'was', 'will', +}) + +SYNONYMS = { + 'js': {'javascript'}, 'javascript': {'js'}, + 'ts': {'typescript'}, 'typescript': {'ts'}, + 'ai': {'artificial', 'intelligence'}, + 'ml': {'machine', 'learning'}, + 'react': {'reactjs'}, 'reactjs': {'react'}, +} + + +def _tokenize(text: str) -> Set[str]: + """Lowercase, strip punctuation, remove stopwords, drop single-char tokens.""" + words = re.sub(r'[^\w\s]', ' ', text.lower()).split() + tokens = {w for w in words if w not in STOPWORDS and len(w) > 1} + expanded = set(tokens) + for t in tokens: + if t in SYNONYMS: + expanded.update(SYNONYMS[t]) + return expanded + + +def _compute_relevance(query: str, text: str) -> float: + """Compute relevance as ratio of query tokens found in text. Floors at 0.1.""" + q_tokens = _tokenize(query) + t_tokens = _tokenize(text) + if not q_tokens: + return 0.5 + overlap = len(q_tokens & t_tokens) + ratio = overlap / len(q_tokens) + return max(0.1, min(1.0, ratio)) + + +def _extract_core_subject(topic: str) -> str: + """Extract core subject from verbose query for Twitter search.""" + text = topic.lower().strip() + prefixes = [ + 'what are the best', 'what is the best', 'what are the latest', + 'what are people saying about', 'what do people think about', + 'how do i use', 'how to use', 'how to', + 'what are', 'what is', 'tips for', 'best practices for', + ] + for p in prefixes: + if text.startswith(p + ' '): + text = text[len(p):].strip() + noise = { + 'best', 'top', 'good', 'great', 'awesome', + 'latest', 'new', 'news', 'update', 'updates', + 'trending', 'hottest', 'popular', 'viral', + 'practices', 'features', 'recommendations', 'advice', + } + words = text.split() + filtered = [w for w in words if w not in noise] + result = ' '.join(filtered) if filtered else text + return result.rstrip('?!.') + + +def _log(msg: str): + if sys.stderr.isatty(): + sys.stderr.write(f"[X/SC] {msg}\n") + sys.stderr.flush() + + +def _sc_headers(token: str) -> Dict[str, str]: + return { + "x-api-key": token, + "Content-Type": "application/json", + } + + +def _parse_date(item: Dict[str, Any]) -> Optional[str]: + """Parse date from ScrapeCreators Twitter item to YYYY-MM-DD.""" + # Try created_at string (e.g. "Wed Oct 10 20:19:24 +0000 2018") + created_at = item.get("created_at") + if created_at and isinstance(created_at, str): + try: + dt = datetime.strptime(created_at, "%a %b %d %H:%M:%S %z %Y") + return dt.strftime("%Y-%m-%d") + except (ValueError, TypeError): + pass + + # Try unix timestamp + ts = item.get("timestamp") or item.get("created_at_timestamp") + if ts: + try: + dt = datetime.fromtimestamp(int(ts), tz=timezone.utc) + return dt.strftime("%Y-%m-%d") + except (ValueError, TypeError, OSError): + pass + + # Try ISO format + for key in ("created_at", "date"): + val = item.get(key) + if val and isinstance(val, str): + try: + dt = datetime.fromisoformat(val.replace("Z", "+00:00")) + return dt.strftime("%Y-%m-%d") + except (ValueError, TypeError): + pass + + return None + + +def search_x( + topic: str, + from_date: str, + to_date: str, + depth: str = "default", + token: str = None, +) -> Dict[str, Any]: + """Search X/Twitter via ScrapeCreators API. + + Returns: + Dict with 'items' list (in normalize_x_items format) and optional 'error'. + """ + if not token: + return {"items": [], "error": "No SCRAPECREATORS_API_KEY configured"} + + if not _requests: + return {"items": [], "error": "requests library not installed"} + + config = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) + core_topic = _extract_core_subject(topic) + + _log(f"Searching X for '{core_topic}' (depth={depth}, count={config['results_per_page']})") + + try: + resp = _requests.get( + f"{SCRAPECREATORS_BASE}/search/tweets", + params={"query": core_topic, "sort_by": "relevance"}, + 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}"} + + raw_items = data.get("tweets") or data.get("data") or data.get("results") or [] + raw_items = raw_items[:config["results_per_page"]] + + items = [] + for i, raw in enumerate(raw_items): + tweet_id = str(raw.get("id") or raw.get("tweet_id") or raw.get("id_str") or f"sc-x-{i}") + text = raw.get("full_text") or raw.get("text") or "" + user = raw.get("user") or raw.get("author") or {} + author_handle = user.get("screen_name") or user.get("username") or "" + + # Engagement metrics + likes = raw.get("favorite_count") or raw.get("likes") or 0 + retweets = raw.get("retweet_count") or raw.get("retweets") or 0 + replies = raw.get("reply_count") or raw.get("replies") or 0 + quotes = raw.get("quote_count") or raw.get("quotes") or 0 + + date_str = _parse_date(raw) + relevance = _compute_relevance(core_topic, text) + + url = "" + if author_handle and tweet_id and not tweet_id.startswith("sc-x-"): + url = f"https://x.com/{author_handle}/status/{tweet_id}" + + items.append({ + "id": tweet_id, + "text": text, + "url": url, + "author_handle": author_handle, + "date": date_str, + "engagement": { + "likes": likes, + "reposts": retweets, + "replies": replies, + "quotes": quotes, + }, + "relevance": relevance, + "why_relevant": f"X: @{author_handle}: {text[:60]}" if text else f"X: {core_topic}", + }) + + # 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} tweets outside date range") + else: + _log(f"No tweets within date range, keeping all {len(items)}") + + # Sort by engagement (likes + retweets) + items.sort(key=lambda x: (x["engagement"]["likes"] + x["engagement"]["reposts"]), reverse=True) + + _log(f"Found {len(items)} tweets") + return {"items": items} + + +def parse_x_response(response: Dict[str, Any]) -> List[Dict[str, Any]]: + """Parse search response to normalized format.""" + return response.get("items", []) diff --git a/tests/test_scrapecreators_x.py b/tests/test_scrapecreators_x.py new file mode 100644 index 0000000..e899555 --- /dev/null +++ b/tests/test_scrapecreators_x.py @@ -0,0 +1,113 @@ +"""Tests for scrapecreators_x module.""" + +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).parent.parent / "scripts")) +from lib import scrapecreators_x + + +class TestTokenize(unittest.TestCase): + def test_lowercases(self): + tokens = scrapecreators_x._tokenize("Claude AI") + self.assertIn("claude", tokens) + + def test_strips_stopwords(self): + tokens = scrapecreators_x._tokenize("the best AI tool") + self.assertNotIn("the", tokens) + self.assertIn("best", tokens) # 'best' is not a stopword in tokenizer + + def test_removes_single_char(self): + tokens = scrapecreators_x._tokenize("a b cd ef") + self.assertNotIn("a", tokens) + self.assertNotIn("b", tokens) + self.assertIn("cd", tokens) + + def test_expands_synonyms(self): + tokens = scrapecreators_x._tokenize("ai research") + self.assertIn("artificial", tokens) + self.assertIn("intelligence", tokens) + + +class TestComputeRelevance(unittest.TestCase): + def test_exact_match_high(self): + score = scrapecreators_x._compute_relevance("claude code", "claude code is amazing") + self.assertGreaterEqual(score, 0.8) + + def test_no_match_low(self): + score = scrapecreators_x._compute_relevance("claude code", "pizza recipes today") + self.assertLessEqual(score, 0.2) + + def test_empty_query_returns_neutral(self): + score = scrapecreators_x._compute_relevance("", "some text") + self.assertEqual(score, 0.5) + + def test_floor_at_01(self): + score = scrapecreators_x._compute_relevance("abcdef ghijkl", "xyz") + self.assertGreaterEqual(score, 0.1) + + +class TestExtractCoreSubject(unittest.TestCase): + def test_strips_prefix(self): + result = scrapecreators_x._extract_core_subject("what are people saying about claude") + self.assertEqual(result, "claude") + + def test_strips_noise(self): + result = scrapecreators_x._extract_core_subject("latest trending news claude code") + self.assertNotIn("latest", result) + self.assertNotIn("trending", result) + self.assertIn("claude", result) + + def test_preserves_core(self): + result = scrapecreators_x._extract_core_subject("react native") + self.assertEqual(result, "react native") + + +class TestParseDate(unittest.TestCase): + def test_twitter_format(self): + item = {"created_at": "Wed Oct 10 20:19:24 +0000 2018"} + self.assertEqual(scrapecreators_x._parse_date(item), "2018-10-10") + + def test_unix_timestamp(self): + item = {"timestamp": 1705363200} + self.assertEqual(scrapecreators_x._parse_date(item), "2024-01-16") + + def test_iso_format(self): + item = {"created_at": "2024-06-15T12:00:00Z"} + self.assertEqual(scrapecreators_x._parse_date(item), "2024-06-15") + + def test_none_returns_none(self): + self.assertIsNone(scrapecreators_x._parse_date({})) + + +class TestSearchX(unittest.TestCase): + def test_no_token_returns_error(self): + result = scrapecreators_x.search_x("test", "2024-01-01", "2024-12-31") + self.assertEqual(result["items"], []) + self.assertIn("No SCRAPECREATORS_API_KEY", result["error"]) + + def test_parse_x_response(self): + response = {"items": [{"id": "1", "text": "hello"}]} + items = scrapecreators_x.parse_x_response(response) + self.assertEqual(len(items), 1) + self.assertEqual(items[0]["text"], "hello") + + def test_parse_empty_response(self): + items = scrapecreators_x.parse_x_response({}) + self.assertEqual(items, []) + + +class TestDepthConfig(unittest.TestCase): + def test_all_depths_exist(self): + for depth in ("quick", "default", "deep"): + self.assertIn(depth, scrapecreators_x.DEPTH_CONFIG) + + def test_deep_has_more_results(self): + quick = scrapecreators_x.DEPTH_CONFIG["quick"]["results_per_page"] + deep = scrapecreators_x.DEPTH_CONFIG["deep"]["results_per_page"] + self.assertGreater(deep, quick) + + +if __name__ == "__main__": + unittest.main()