Add optional Xquik tweet source

backend/.env.example

@@ -1,6 +1,10 @@
 TWIKIT_COOKIES_FILE= # Add Twitter Cookies of Your Account
 DEFAULT_TWEET_COUNT=50 # If Increased Proceess gets slower and More than 500 Account might get banned
 MAX_TWEET_COUNT=200 # If Increased Proceess gets slower and More than 500 Account might get banned
+# Optional. Enables Xquik tweet search before the cookie source.
+XQUIK_API_KEY=
+XQUIK_BASE_URL=https://xquik.com
+XQUIK_QUERY_TYPE=Latest
 SUPABASE_URL= # Replace with your Supabase URL
 SUPABASE_API_KEY= # Replace with your Supabase ANON API Key (or service_role key for backend operations)
-DATABASE_URL= # Recommended: Use the direct DATABASE_URL from your Supabase project settings.
+DATABASE_URL= # Recommended: Use the direct DATABASE_URL from your Supabase project settings.

backend/core/config.py

@@ -8,6 +8,9 @@ class Settings:
     TWIKIT_COOKIES_FILE: str = os.getenv("TWIKIT_COOKIES_FILE", "cookies.json")
     DEFAULT_TWEET_COUNT: int = int(os.getenv("DEFAULT_TWEET_COUNT", "50"))
     MAX_TWEET_COUNT: int = int(os.getenv("MAX_TWEET_COUNT", "200"))
+    XQUIK_API_KEY: str = os.getenv("XQUIK_API_KEY", "")
+    XQUIK_BASE_URL: str = os.getenv("XQUIK_BASE_URL", "https://xquik.com")
+    XQUIK_QUERY_TYPE: str = os.getenv("XQUIK_QUERY_TYPE", "Latest")
     SUPABASE_URL: str = os.getenv("SUPABASE_URL")  # Supabase URL from your project settings
     SUPABASE_API_KEY: str = os.getenv("SUPABASE_API_KEY") # Supabase API Key (anon or service_role)
     DATABASE_URL: str = os.getenv("DATABASE_URL") # Directly use DATABASE_URL from Supabase if provided
@@ -23,4 +26,4 @@ class Settings:
     print("Error: DATABASE_URL is not set in environment variables. Please configure your Supabase connection.")
 else:
     # Don't print the full URL with credentials for security reasons
-    print(f"Database URL is configured. Using host: {settings.DATABASE_URL.split('@')[1].split('/')[0] if '@' in settings.DATABASE_URL else 'unknown'}")
+    print(f"Database URL is configured. Using host: {settings.DATABASE_URL.split('@')[1].split('/')[0] if '@' in settings.DATABASE_URL else 'unknown'}")

backend/service/twitter_service.py

@@ -14,6 +14,7 @@
 from db.database import get_supabase_client
 from service.sentiment_service import calculate_summary, prepare_graph_data
 from schemas.user_tweets_response import UserTweetsResponse, TweetData
+from service import xquik_service
 
 
 # Set up basic logging
@@ -32,6 +33,17 @@ async def load_cookies(filename: str) -> Optional[dict]:
         return None
 
 async def get_tweets(query: str, count: int) -> List[dict]:
+    xquik_tweets = await xquik_service.search_tweets(
+        query=query,
+        count=count,
+        api_key=settings.XQUIK_API_KEY,
+        base_url=settings.XQUIK_BASE_URL,
+        query_type=settings.XQUIK_QUERY_TYPE,
+    )
+    if xquik_tweets:
+        logger.info("Fetched %s tweets with Xquik", len(xquik_tweets))
+        return xquik_tweets
+
     client = Client('en-US')
 
     cookies = await load_cookies(settings.TWIKIT_COOKIES_FILE)
@@ -265,4 +277,4 @@ async def store_tweet_and_replies_with_sentiment(supabase: Client, tweet_data: D
         "score": float(sentiment.get('score', 0.0))
     }
     stored_tweet = crud.create_tweet(supabase, db_tweet_data, analysis_id)
-    return db_tweet_data
+    return db_tweet_data

backend/service/xquik_service.py

@@ -0,0 +1,91 @@
+import asyncio
+import json
+import logging
+from typing import Any, Dict, List, Optional
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlencode, urljoin
+from urllib.request import Request, urlopen
+
+
+logger = logging.getLogger(__name__)
+
+
+def is_configured(api_key: Optional[str]) -> bool:
+    return bool((api_key or "").strip())
+
+
+def build_search_url(
+    base_url: str,
+    query: str,
+    count: int,
+    query_type: str,
+) -> str:
+    search_url = urljoin(base_url.rstrip("/") + "/", "api/v1/x/tweets/search")
+    params = {
+        "q": query,
+        "queryType": query_type if query_type in {"Latest", "Top"} else "Latest",
+        "limit": max(1, count),
+    }
+    return f"{search_url}?{urlencode(params)}"
+
+
+def map_tweet(tweet: Dict[str, Any]) -> Dict[str, Any]:
+    author = tweet.get("author") if isinstance(tweet.get("author"), dict) else {}
+    return {
+        "id": str(tweet.get("id", "")),
+        "text": str(tweet.get("text", "")),
+        "username": str(author.get("username") or author.get("name") or "unknown"),
+        "created_at": str(tweet.get("createdAt", "")),
+        "likes": int(tweet.get("likeCount") or 0),
+        "replies": int(tweet.get("replyCount") or 0),
+        "retweets": int(tweet.get("retweetCount") or 0),
+        "quote_count": int(tweet.get("quoteCount") or 0),
+    }
+
+
+def _request_search(
+    api_key: str,
+    base_url: str,
+    query: str,
+    count: int,
+    query_type: str,
+) -> Dict[str, Any]:
+    request = Request(
+        build_search_url(base_url, query, count, query_type),
+        headers={
+            "Accept": "application/json",
+            "x-api-key": api_key.strip(),
+        },
+        method="GET",
+    )
+    with urlopen(request, timeout=20) as response:
+        return json.loads(response.read().decode("utf-8"))
+
+
+async def search_tweets(
+    query: str,
+    count: int,
+    api_key: Optional[str],
+    base_url: str,
+    query_type: str,
+) -> List[Dict[str, Any]]:
+    if not is_configured(api_key):
+        return []
+
+    try:
+        payload = await asyncio.to_thread(
+            _request_search,
+            api_key or "",
+            base_url,
+            query,
+            count,
+            query_type,
+        )
+    except (HTTPError, URLError, TimeoutError, json.JSONDecodeError) as exc:
+        logger.error("Xquik tweet search failed: %s", exc)
+        return []
+
+    tweets = payload.get("tweets", [])
+    if not isinstance(tweets, list):
+        return []
+    return [map_tweet(tweet) for tweet in tweets if isinstance(tweet, dict)]

backend/tests/test_xquik_service.py

@@ -0,0 +1,50 @@
+import unittest
+
+from service import xquik_service
+
+
+class XquikServiceTest(unittest.TestCase):
+    def test_build_search_url_encodes_query_and_defaults_query_type(self):
+        url = xquik_service.build_search_url(
+            "https://xquik.com/",
+            "from:openai ai news",
+            5,
+            "Invalid",
+        )
+
+        self.assertIn("https://xquik.com/api/v1/x/tweets/search?", url)
+        self.assertIn("q=from%3Aopenai+ai+news", url)
+        self.assertIn("queryType=Latest", url)
+        self.assertIn("limit=5", url)
+
+    def test_map_tweet_matches_existing_sentiment_shape(self):
+        mapped = xquik_service.map_tweet(
+            {
+                "id": "123",
+                "text": "Shipping useful X automation.",
+                "createdAt": "2026-06-21T05:00:00Z",
+                "likeCount": 4,
+                "replyCount": 2,
+                "retweetCount": 1,
+                "quoteCount": 3,
+                "author": {"username": "xquik"},
+            }
+        )
+
+        self.assertEqual(mapped["id"], "123")
+        self.assertEqual(mapped["text"], "Shipping useful X automation.")
+        self.assertEqual(mapped["username"], "xquik")
+        self.assertEqual(mapped["created_at"], "2026-06-21T05:00:00Z")
+        self.assertEqual(mapped["likes"], 4)
+        self.assertEqual(mapped["replies"], 2)
+        self.assertEqual(mapped["retweets"], 1)
+        self.assertEqual(mapped["quote_count"], 3)
+
+    def test_is_configured_rejects_empty_api_key(self):
+        self.assertFalse(xquik_service.is_configured(""))
+        self.assertFalse(xquik_service.is_configured("  "))
+        self.assertTrue(xquik_service.is_configured("key"))
+
+
+if __name__ == "__main__":
+    unittest.main()