diff --git a/backend/.env.example b/backend/.env.example index ec5360f..4202b65 100644 --- a/backend/.env.example +++ b/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. \ No newline at end of file +DATABASE_URL= # Recommended: Use the direct DATABASE_URL from your Supabase project settings. diff --git a/backend/core/config.py b/backend/core/config.py index d3e0283..6ff8b51 100644 --- a/backend/core/config.py +++ b/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'}") \ No newline at end of file + print(f"Database URL is configured. Using host: {settings.DATABASE_URL.split('@')[1].split('/')[0] if '@' in settings.DATABASE_URL else 'unknown'}") diff --git a/backend/service/twitter_service.py b/backend/service/twitter_service.py index fcc49d0..e8c11ec 100644 --- a/backend/service/twitter_service.py +++ b/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 \ No newline at end of file + return db_tweet_data diff --git a/backend/service/xquik_service.py b/backend/service/xquik_service.py new file mode 100644 index 0000000..fc1c46b --- /dev/null +++ b/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)] diff --git a/backend/tests/test_xquik_service.py b/backend/tests/test_xquik_service.py new file mode 100644 index 0000000..e49f3b0 --- /dev/null +++ b/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()