diff --git a/scripts/lib/env.py b/scripts/lib/env.py index a5af3ec..755a002 100644 --- a/scripts/lib/env.py +++ b/scripts/lib/env.py @@ -261,6 +261,7 @@ def get_config() -> dict[str, Any]: ('SERPER_API_KEY', None), ('OPENROUTER_API_KEY', None), ('PARALLEL_API_KEY', None), + ('XQUIK_API_KEY', None), ('FROM_BROWSER', None), ('SETUP_COMPLETE', None), ('INCLUDE_SOURCES', None), @@ -613,3 +614,17 @@ def is_pinterest_available(config: dict[str, Any]) -> bool: def get_pinterest_token(config: dict[str, Any]) -> str: """Get Pinterest API token (same ScrapeCreators key as TikTok/Instagram).""" return config.get('SCRAPECREATORS_API_KEY') or '' + + +# Xquik +def is_xquik_available(config: dict[str, Any]) -> bool: + """Check if Xquik X search source is available. + + Requires XQUIK_API_KEY (API key from xquik.com). + """ + return bool(config.get('XQUIK_API_KEY')) + + +def get_xquik_token(config: dict[str, Any]) -> str: + """Get Xquik API key.""" + return config.get('XQUIK_API_KEY') or '' diff --git a/scripts/lib/normalize.py b/scripts/lib/normalize.py index 052a934..68221bb 100644 --- a/scripts/lib/normalize.py +++ b/scripts/lib/normalize.py @@ -46,6 +46,7 @@ def normalize_source_items( "bluesky": lambda s, i, idx, fd, td: _normalize_microblog(s, i, idx, fd, td, "BS", "Bluesky post"), "truthsocial": lambda s, i, idx, fd, td: _normalize_microblog(s, i, idx, fd, td, "TS", "Truth Social post"), "threads": lambda s, i, idx, fd, td: _normalize_microblog(s, i, idx, fd, td, "TH", "Threads post"), + "xquik": _normalize_x, "pinterest": _normalize_pinterest, "polymarket": _normalize_polymarket, "grounding": _normalize_grounding, diff --git a/scripts/lib/pipeline.py b/scripts/lib/pipeline.py index 960f215..c1a1cc3 100644 --- a/scripts/lib/pipeline.py +++ b/scripts/lib/pipeline.py @@ -39,6 +39,7 @@ from . import ( truthsocial, xai_x, xiaohongshu_api, + xquik, youtube_yt, ) from .cluster import cluster_candidates @@ -56,6 +57,7 @@ SEARCH_ALIAS = { "truth": "truthsocial", "web": "grounding", "xhs": "xiaohongshu", + "xquik": "xquik", } MAX_SOURCE_FETCHES: dict[str, int] = {"x": 2} @@ -74,6 +76,7 @@ MOCK_AVAILABLE_SOURCES = [ "xiaohongshu", "github", "perplexity", + "xquik", ] @@ -117,6 +120,8 @@ def available_sources(config: dict[str, Any], requested_sources: list[str] | Non available.append("threads") if requested_sources and "pinterest" in requested_sources and env.is_pinterest_available(config): available.append("pinterest") + if env.is_xquik_available(config): + available.append("xquik") return available @@ -935,6 +940,13 @@ def _retrieve_stream( ), {} if source == "perplexity": return perplexity.search(subquery.search_query, date_range, config, deep=config.get("_deep_research", False)) + if source == "xquik": + result = xquik.search_xquik( + subquery.search_query, from_date, to_date, + depth=depth, + token=env.get_xquik_token(config), + ) + return xquik.parse_xquik_response(result), {} raise RuntimeError(f"Unsupported source: {source}") diff --git a/scripts/lib/xquik.py b/scripts/lib/xquik.py new file mode 100644 index 0000000..80b6ae3 --- /dev/null +++ b/scripts/lib/xquik.py @@ -0,0 +1,227 @@ +"""Xquik X search source for the v3.0.0 last30days pipeline. + +Uses the Xquik REST API (https://xquik.com/api/v1) to search X/Twitter +with full engagement metrics (likes, retweets, replies, quotes, views, +bookmarks). Requires an API key from xquik.com. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any, Dict, List + +from . import http, log +from .relevance import token_overlap_relevance as _compute_relevance + +# Depth configurations: number of results to request per query +DEPTH_CONFIG = { + "quick": {"limit": 10, "queries": 1}, + "default": {"limit": 20, "queries": 2}, + "deep": {"limit": 40, "queries": 3}, +} + +_BASE_URL = "https://xquik.com/api/v1" + + +def _log(msg: str): + log.source_log("Xquik", msg, tty_only=False) + + +def _extract_core_subject(topic: str) -> str: + """Extract core subject for X search queries.""" + from .query import extract_core_subject + return extract_core_subject(topic, max_words=5, strip_suffixes=True) + + +def expand_xquik_queries(topic: str, depth: str) -> List[str]: + """Generate query variants based on depth. + + Args: + topic: Research topic + depth: "quick", "default", or "deep" + + Returns: + List of query strings (1 for quick, 2 for default, 3 for deep). + """ + core = _extract_core_subject(topic) + queries = [core] + + # Add original topic if meaningfully different + if topic.lower().strip() != core.lower().strip(): + queries.append(topic.strip()) + + # Add compound term variant for deep searches + if len(queries) < 3: + from .query import extract_compound_terms + compounds = extract_compound_terms(topic) + if compounds: + or_parts = " OR ".join(f'"{t}"' for t in compounds[:3]) + queries.append(f"({or_parts})") + + cap = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"])["queries"] + return queries[:cap] + + +def search_xquik( + topic: str, + from_date: str, + to_date: str, + depth: str = "default", + token: str = "", +) -> Dict[str, Any]: + """Search X via Xquik REST API. + + Args: + topic: Search topic + from_date: Start date (YYYY-MM-DD) + to_date: End date (YYYY-MM-DD) + depth: Research depth - "quick", "default", or "deep" + token: Xquik API key + + Returns: + Dict with "items" list and optional "error" string. + """ + if not token: + return {"items": [], "error": "No XQUIK_API_KEY configured"} + + cfg = DEPTH_CONFIG.get(depth, DEPTH_CONFIG["default"]) + queries = expand_xquik_queries(topic, depth) + all_items: List[Dict[str, Any]] = [] + seen_ids: set[str] = set() + + for query_text in queries: + try: + url = f"{_BASE_URL}/x/tweets/search" + # Build query with date filter + q = f"{query_text} since:{from_date} until:{to_date}" + params = f"q={_url_encode(q)}&queryType=Top&limit={cfg['limit']}" + full_url = f"{url}?{params}" + + _log(f"Searching: {query_text}") + response = http.get( + full_url, + headers={"X-Api-Key": token}, + timeout=30, + retries=2, + ) + + tweets = response.get("tweets", []) + if not isinstance(tweets, list): + continue + + for i, tweet in enumerate(tweets): + if not isinstance(tweet, dict): + continue + tweet_id = str(tweet.get("id", "")) + if tweet_id in seen_ids: + continue + seen_ids.add(tweet_id) + + item = _parse_tweet(tweet, i + len(all_items), query_text) + if item: + all_items.append(item) + + except http.HTTPError as exc: + status = getattr(exc, "status_code", None) + if status in (401, 403): + return {"items": [], "error": f"Xquik auth failed ({status})"} + _log(f"HTTP error for query '{query_text}': {exc}") + except Exception as exc: + _log(f"Error for query '{query_text}': {exc}") + + return {"items": all_items} + + +def search_and_enrich( + topic: str, + from_date: str, + to_date: str, + depth: str = "default", + token: str = "", +) -> Dict[str, Any]: + """Search X via Xquik and return results. + + Xquik API returns full engagement data by default, so no separate + enrichment step is needed. + """ + return search_xquik(topic, from_date, to_date, depth=depth, token=token) + + +def parse_xquik_response(response: Dict[str, Any]) -> List[Dict[str, Any]]: + """Extract items from search response. + + Args: + response: Response dict from search_xquik() + + Returns: + List of normalized item dicts. + """ + return response.get("items", []) + + +def _parse_tweet( + tweet: Dict[str, Any], index: int, query: str +) -> Dict[str, Any] | None: + """Parse a single tweet from the API response into the standard item format.""" + author = tweet.get("author") or {} + username = str(author.get("username", "")).lstrip("@") + tweet_id = str(tweet.get("id", "")) + + # Build URL + url = "" + if username and tweet_id: + url = f"https://x.com/{username}/status/{tweet_id}" + if not url: + return None + + # Parse date + date = None + created_at = tweet.get("createdAt") or "" + if created_at: + try: + if len(created_at) > 10 and created_at[10] == "T": + dt = datetime.fromisoformat(created_at.replace("Z", "+00:00")) + else: + dt = datetime.strptime(created_at, "%a %b %d %H:%M:%S %z %Y") + date = dt.strftime("%Y-%m-%d") + except (ValueError, TypeError): + pass + + text = str(tweet.get("text", "")).strip()[:500] + + # Build engagement dict with full metrics + engagement = { + "likes": _safe_int(tweet.get("likeCount")), + "reposts": _safe_int(tweet.get("retweetCount")), + "replies": _safe_int(tweet.get("replyCount")), + "quotes": _safe_int(tweet.get("quoteCount")), + "views": _safe_int(tweet.get("viewCount")), + "bookmarks": _safe_int(tweet.get("bookmarkCount")), + } + + return { + "id": f"XQ{index + 1}", + "text": text, + "url": url, + "author_handle": username, + "date": date, + "engagement": engagement, + "relevance": _compute_relevance(query, text) if query else 0.7, + "why_relevant": "", + } + + +def _safe_int(value: Any) -> int | None: + """Convert value to int, returning None on failure.""" + if value is None: + return None + try: + return int(value) + except (ValueError, TypeError): + return None + + +def _url_encode(text: str) -> str: + """URL-encode a string using stdlib.""" + from urllib.parse import quote + return quote(text, safe="") diff --git a/tests/test_xquik.py b/tests/test_xquik.py new file mode 100644 index 0000000..b54d097 --- /dev/null +++ b/tests/test_xquik.py @@ -0,0 +1,275 @@ +import sys +import unittest +from pathlib import Path +from unittest.mock import patch + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) + +from lib.xquik import ( + DEPTH_CONFIG, + _parse_tweet, + _safe_int, + expand_xquik_queries, + parse_xquik_response, + search_xquik, +) + + +class TestExpandXquikQueries(unittest.TestCase): + def test_quick_returns_one_query(self): + queries = expand_xquik_queries("latest trends in AI agents", "quick") + self.assertEqual(len(queries), 1) + + def test_default_returns_up_to_two_queries(self): + queries = expand_xquik_queries("multi-agent systems research", "default") + self.assertLessEqual(len(queries), 2) + self.assertGreaterEqual(len(queries), 1) + + def test_deep_returns_up_to_three_queries(self): + queries = expand_xquik_queries("best AI coding assistants 2026", "deep") + self.assertLessEqual(len(queries), 3) + self.assertGreaterEqual(len(queries), 1) + + def test_single_word_topic(self): + queries = expand_xquik_queries("Bitcoin", "quick") + self.assertEqual(len(queries), 1) + self.assertIn("bitcoin", queries[0].lower()) + + +class TestParseTweet(unittest.TestCase): + def test_valid_tweet(self): + tweet = { + "id": "123456", + "text": "This is a test tweet about AI agents", + "createdAt": "2026-03-15T12:00:00Z", + "likeCount": 42, + "retweetCount": 10, + "replyCount": 5, + "quoteCount": 2, + "viewCount": 5000, + "bookmarkCount": 8, + "author": {"username": "testuser", "name": "Test User"}, + } + item = _parse_tweet(tweet, 0, "AI agents") + self.assertIsNotNone(item) + self.assertEqual(item["id"], "XQ1") + self.assertEqual(item["url"], "https://x.com/testuser/status/123456") + self.assertEqual(item["author_handle"], "testuser") + self.assertEqual(item["date"], "2026-03-15") + self.assertEqual(item["engagement"]["likes"], 42) + self.assertEqual(item["engagement"]["reposts"], 10) + self.assertEqual(item["engagement"]["replies"], 5) + self.assertEqual(item["engagement"]["quotes"], 2) + self.assertEqual(item["engagement"]["views"], 5000) + self.assertEqual(item["engagement"]["bookmarks"], 8) + self.assertGreater(item["relevance"], 0) + + def test_missing_author_returns_none(self): + tweet = {"id": "123", "text": "test"} + item = _parse_tweet(tweet, 0, "test") + self.assertIsNone(item) + + def test_at_prefix_stripped(self): + tweet = { + "id": "456", + "text": "hello", + "author": {"username": "@someone"}, + } + item = _parse_tweet(tweet, 0, "hello") + self.assertIsNotNone(item) + self.assertEqual(item["author_handle"], "someone") + + def test_zero_engagement_preserved(self): + tweet = { + "id": "789", + "text": "zero likes tweet", + "author": {"username": "user"}, + "likeCount": 0, + "retweetCount": 0, + "replyCount": 0, + "quoteCount": 0, + "viewCount": 0, + "bookmarkCount": 0, + } + item = _parse_tweet(tweet, 0, "test") + self.assertIsNotNone(item) + self.assertEqual(item["engagement"]["likes"], 0) + self.assertEqual(item["engagement"]["reposts"], 0) + self.assertEqual(item["engagement"]["views"], 0) + + def test_none_engagement_values(self): + tweet = { + "id": "101", + "text": "minimal tweet", + "author": {"username": "user"}, + } + item = _parse_tweet(tweet, 0, "test") + self.assertIsNotNone(item) + self.assertIsNone(item["engagement"]["likes"]) + self.assertIsNone(item["engagement"]["views"]) + + def test_text_truncated_at_500(self): + tweet = { + "id": "102", + "text": "x" * 600, + "author": {"username": "user"}, + } + item = _parse_tweet(tweet, 0, "test") + self.assertIsNotNone(item) + self.assertEqual(len(item["text"]), 500) + + def test_twitter_date_format(self): + tweet = { + "id": "103", + "text": "old format", + "createdAt": "Wed Jan 15 14:30:00 +0000 2026", + "author": {"username": "user"}, + } + item = _parse_tweet(tweet, 0, "test") + self.assertIsNotNone(item) + self.assertEqual(item["date"], "2026-01-15") + + def test_invalid_date_graceful(self): + tweet = { + "id": "104", + "text": "bad date", + "createdAt": "not-a-date", + "author": {"username": "user"}, + } + item = _parse_tweet(tweet, 0, "test") + self.assertIsNotNone(item) + self.assertIsNone(item["date"]) + + def test_empty_author_dict(self): + tweet = {"id": "105", "text": "test", "author": {}} + item = _parse_tweet(tweet, 0, "test") + self.assertIsNone(item) + + def test_index_offset(self): + tweet = { + "id": "106", + "text": "test", + "author": {"username": "user"}, + } + item = _parse_tweet(tweet, 4, "test") + self.assertIsNotNone(item) + self.assertEqual(item["id"], "XQ5") + + +class TestSafeInt(unittest.TestCase): + def test_int_passthrough(self): + self.assertEqual(_safe_int(42), 42) + + def test_string_int(self): + self.assertEqual(_safe_int("100"), 100) + + def test_none_returns_none(self): + self.assertIsNone(_safe_int(None)) + + def test_invalid_string(self): + self.assertIsNone(_safe_int("abc")) + + def test_zero(self): + self.assertEqual(_safe_int(0), 0) + + def test_float_truncates(self): + self.assertEqual(_safe_int(3.7), 3) + + +class TestParseXquikResponse(unittest.TestCase): + def test_extracts_items(self): + response = {"items": [{"id": "1"}, {"id": "2"}]} + items = parse_xquik_response(response) + self.assertEqual(len(items), 2) + + def test_empty_response(self): + self.assertEqual(parse_xquik_response({}), []) + + def test_error_response(self): + response = {"items": [], "error": "something went wrong"} + self.assertEqual(parse_xquik_response(response), []) + + +class TestSearchXquik(unittest.TestCase): + def test_no_token_returns_error(self): + result = search_xquik("test", "2026-01-01", "2026-03-01", token="") + self.assertEqual(result["items"], []) + self.assertIn("XQUIK_API_KEY", result["error"]) + + @patch("lib.xquik.http.get") + def test_successful_search(self, mock_get): + mock_get.return_value = { + "tweets": [ + { + "id": "111", + "text": "AI agents are amazing", + "createdAt": "2026-02-15T10:00:00Z", + "likeCount": 50, + "retweetCount": 12, + "replyCount": 3, + "quoteCount": 1, + "viewCount": 2000, + "bookmarkCount": 5, + "author": {"username": "aidev"}, + }, + ], + "has_next_page": False, + } + result = search_xquik("AI agents", "2026-02-01", "2026-03-01", token="test-key") + self.assertEqual(len(result["items"]), 1) + self.assertEqual(result["items"][0]["author_handle"], "aidev") + self.assertEqual(result["items"][0]["engagement"]["likes"], 50) + self.assertNotIn("error", result) + + @patch("lib.xquik.http.get") + def test_deduplicates_across_queries(self, mock_get): + tweet = { + "id": "222", + "text": "duplicate tweet", + "author": {"username": "user"}, + } + mock_get.return_value = {"tweets": [tweet]} + result = search_xquik("test topic", "2026-01-01", "2026-03-01", depth="default", token="key") + # Even with multiple queries, same tweet ID should appear only once + ids = [item.get("id") for item in result["items"]] + # All items should have unique XQ ids (deduped by tweet ID) + self.assertEqual(len(ids), len(set(ids))) + + @patch("lib.xquik.http.get") + def test_auth_error_returns_error(self, mock_get): + from lib import http as http_mod + mock_get.side_effect = http_mod.HTTPError("Unauthorized", status_code=401) + result = search_xquik("test", "2026-01-01", "2026-03-01", token="bad-key") + self.assertEqual(result["items"], []) + self.assertIn("auth failed", result.get("error", "")) + + @patch("lib.xquik.http.get") + def test_empty_tweets_list(self, mock_get): + mock_get.return_value = {"tweets": []} + result = search_xquik("obscure topic", "2026-01-01", "2026-03-01", token="key") + self.assertEqual(result["items"], []) + self.assertNotIn("error", result) + + @patch("lib.xquik.http.get") + def test_non_list_tweets_skipped(self, mock_get): + mock_get.return_value = {"tweets": "not a list"} + result = search_xquik("test", "2026-01-01", "2026-03-01", token="key") + self.assertEqual(result["items"], []) + + +class TestDepthConfig(unittest.TestCase): + def test_all_depths_have_limit_and_queries(self): + for depth_name, cfg in DEPTH_CONFIG.items(): + self.assertIn("limit", cfg, f"{depth_name} missing 'limit'") + self.assertIn("queries", cfg, f"{depth_name} missing 'queries'") + + def test_deep_has_highest_limit(self): + self.assertGreater(DEPTH_CONFIG["deep"]["limit"], DEPTH_CONFIG["default"]["limit"]) + self.assertGreater(DEPTH_CONFIG["default"]["limit"], DEPTH_CONFIG["quick"]["limit"]) + + def test_deep_has_most_queries(self): + self.assertGreater(DEPTH_CONFIG["deep"]["queries"], DEPTH_CONFIG["quick"]["queries"]) + + +if __name__ == "__main__": + unittest.main()