sawz/last30days-skill
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add Xquik as X/Twitter search source

Browse Source 
Add Xquik (xquik.com) as a new X/Twitter search source that uses a REST
API with full engagement metrics (likes, retweets, replies, quotes,
views, bookmarks). Uses stdlib urllib only -- no new dependencies.

- scripts/lib/xquik.py: source module with search, parse, query expansion
- tests/test_xquik.py: 32 unit tests covering all functions
- env.py: XQUIK_API_KEY config and availability check
- pipeline.py: source registration and retrieve dispatch
- normalize.py: reuses _normalize_x (same item format as Bird)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Burak Bayır
2026-04-09 22:44:10 +03:00
parent 45f596ca0c
commit 6b3de9170e
5 changed files with 530 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/lib/env.py
+15
View File
@@ -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 ''
scripts/lib/normalize.py
+1
View File
@@ -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,
scripts/lib/pipeline.py
+12
View File
@@ -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
@@ -930,6 +935,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}")
scripts/lib/xquik.py
+227
View File
@@ -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="")
tests/test_xquik.py
+275
View File
@@ -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()