From e9853eaf5d2f785e19c0e8f33ea7796756438462 Mon Sep 17 00:00:00 2001 From: kriptoburak Date: Tue, 30 Jun 2026 05:41:13 +0300 Subject: [PATCH] Add Xquik search helper --- test_xquik.py | 93 +++++++++++++++++++++++++++++++++++++++++ xquik.py | 113 ++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 206 insertions(+) create mode 100644 test_xquik.py create mode 100644 xquik.py diff --git a/test_xquik.py b/test_xquik.py new file mode 100644 index 0000000..d69544a --- /dev/null +++ b/test_xquik.py @@ -0,0 +1,93 @@ +import json +import unittest +from urllib.error import HTTPError +from unittest import mock + +from xquik import XquikClient, search_tweets + + +class Response: + def __init__(self, payload): + self.payload = payload + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def read(self): + return json.dumps(self.payload).encode("utf-8") + + +class XquikClientTest(unittest.TestCase): + def test_search_maps_tweets(self): + seen = {} + + def fake_urlopen(request, timeout): + seen["url"] = request.full_url + seen["key"] = request.headers["X-api-key"] + seen["timeout"] = timeout + return Response( + { + "tweets": [ + { + "id": "123", + "text": "hello", + "createdAt": "2026-01-02T03:04:05Z", + "likeCount": 7, + "retweetCount": 2, + "url": "https://x.com/alice/status/123", + "author": { + "username": "alice", + "name": "Alice", + }, + } + ] + } + ) + + with mock.patch("xquik.urllib.request.urlopen", fake_urlopen): + tweets = XquikClient("test-key", timeout=4).search( + "launch", + max_results=5, + query_type="Top", + ) + + self.assertEqual(len(tweets), 1) + self.assertEqual(tweets[0].tweet_id, "123") + self.assertEqual(tweets[0].tweet, "hello") + self.assertEqual(tweets[0].name, "Alice") + self.assertEqual(tweets[0].screen_name, "alice") + self.assertEqual(tweets[0].retweet_count, 2) + self.assertEqual(tweets[0].favourites_count, 7) + self.assertIn("q=launch", seen["url"]) + self.assertIn("queryType=Top", seen["url"]) + self.assertEqual(seen["key"], "test-key") + self.assertEqual(seen["timeout"], 4) + + def test_requires_api_key(self): + with mock.patch.dict("os.environ", {}, clear=True): + with self.assertRaisesRegex(ValueError, "API key required"): + XquikClient() + + def test_search_tweets_uses_client(self): + with mock.patch.object(XquikClient, "search", return_value=["tweet"]) as search: + result = search_tweets("launch", api_key="test-key", max_results=3) + + self.assertEqual(result, ["tweet"]) + search.assert_called_once_with("launch", max_results=3, query_type="Latest") + + def test_http_error_omits_api_key(self): + def fake_urlopen(request, timeout): + raise HTTPError(request.full_url, 401, "Unauthorized", hdrs=None, fp=None) + + with mock.patch("xquik.urllib.request.urlopen", fake_urlopen): + with self.assertRaisesRegex(RuntimeError, "HTTP 401") as ctx: + XquikClient("test-key").search("launch") + + self.assertNotIn("test-key", str(ctx.exception)) + + +if __name__ == "__main__": + unittest.main() diff --git a/xquik.py b/xquik.py new file mode 100644 index 0000000..d86eced --- /dev/null +++ b/xquik.py @@ -0,0 +1,113 @@ +import datetime +import json +import os +import urllib.error +import urllib.parse +import urllib.request +import uuid + +from tweets import Tweet + + +XQUIK_BASE_URL = "https://xquik.com" +XQUIK_SEARCH_PATH = "/api/v1/x/tweets/search" +_ALLOWED_HOSTS = {"xquik.com"} +_QUERY_TYPES = {"Latest", "Top"} + + +def _int_or_zero(value): + if isinstance(value, bool): + return 0 + if isinstance(value, int): + return value + if isinstance(value, str): + try: + return int(value) + except ValueError: + return 0 + return 0 + + +def _parse_date(value): + if not isinstance(value, str) or value == "": + return datetime.datetime.utcfromtimestamp(0) + normalized = value[:-1] + "+00:00" if value.endswith("Z") else value + try: + parsed = datetime.datetime.fromisoformat(normalized) + except ValueError: + return datetime.datetime.utcfromtimestamp(0) + if parsed.tzinfo is not None: + parsed = parsed.astimezone(datetime.timezone.utc).replace(tzinfo=None) + return parsed + + +def _tweet_from_xquik(raw): + author = raw.get("author") if isinstance(raw.get("author"), dict) else {} + return Tweet( + str(uuid.uuid4()), + str(author.get("name", "")), + str(author.get("username", "")), + str(raw.get("id", "")), + str(raw.get("text", "")), + _parse_date(raw.get("createdAt")), + _int_or_zero(raw.get("retweetCount")), + _int_or_zero(raw.get("likeCount")), + str(raw.get("url", "")), + ) + + +class XquikClient: + def __init__(self, api_key=None, base_url=XQUIK_BASE_URL, timeout=15): + key = api_key or os.environ.get("XQUIK_API_KEY") + if not key: + raise ValueError("Xquik API key required. Pass api_key or set XQUIK_API_KEY.") + + parsed = urllib.parse.urlparse(base_url) + if parsed.scheme != "https" or parsed.hostname not in _ALLOWED_HOSTS: + raise ValueError("Xquik base URL must use https://xquik.com.") + + self.api_key = key + self.base_url = base_url.rstrip("/") + self.timeout = timeout + + def search(self, search_query, max_results=20, query_type="Latest"): + if query_type not in _QUERY_TYPES: + raise ValueError("query_type must be Latest or Top.") + if max_results < 1: + raise ValueError("max_results must be at least 1.") + + params = urllib.parse.urlencode( + { + "q": search_query, + "queryType": query_type, + "limit": min(max_results, 200), + } + ) + request = urllib.request.Request( + self.base_url + XQUIK_SEARCH_PATH + "?" + params, + headers={ + "Accept": "application/json", + "User-Agent": "source-nerd-twitter-scraper Xquik helper", + "x-api-key": self.api_key, + }, + ) + + try: + with urllib.request.urlopen(request, timeout=self.timeout) as response: + payload = json.loads(response.read().decode("utf-8")) + except urllib.error.HTTPError as error: + raise RuntimeError("Xquik search failed with HTTP {}.".format(error.code)) + + tweets = payload.get("tweets") if isinstance(payload, dict) else None + if not isinstance(tweets, list): + raise RuntimeError("Xquik search returned an invalid response.") + + return [_tweet_from_xquik(tweet) for tweet in tweets if isinstance(tweet, dict)] + + +def search_tweets(search_query, api_key=None, max_results=20, query_type="Latest"): + return XquikClient(api_key=api_key).search( + search_query, + max_results=max_results, + query_type=query_type, + )