# social_insights/twitterx.py from typing import Dict, Any, List, Optional, Tuple from .base import SocialProvider, OAuthToken, DateRange from .utils import http_get, http_post, clamp_int import datetime TW_OAUTH = "https://twitter.com/i/oauth2/authorize" TW_TOKEN = "https://api.twitter.com/2/oauth2/token" TW_API = "https://api.twitter.com/2" TW_TEMP_TOKEN = "https://ap.bestbrain.ai/x/token" TW_TEMP_REFRESH = "https://ap.bestbrain.ai/x/refresh" # scopes: tweet.read users.read offline.access like.read list.read # (posting needs tweet.write; impressions generally not available unless Ads/elevated) TW_SCOPES_DEFAULT = [ "tweet.read","users.read","offline.access","like.read" ] class TwitterXProvider(SocialProvider): def auth_url(self, state: str, scopes: List[str]) -> str: scope = "%20".join(scopes or TW_SCOPES_DEFAULT) return (f"{TW_OAUTH}?response_type=code&client_id={self.client_id}" f"&redirect_uri={self.redirect_uri}&scope={scope}&state={state}" f"&code_challenge=plain_dummy&code_challenge_method=plain") # replace with PKCE def exchange_code(self, code: str) -> OAuthToken: data = http_post(TW_TOKEN, headers={"Content-Type": "application/x-www-form-urlencoded"}, data={ "grant_type": "authorization_code", "code": code, "redirect_uri": self.redirect_uri, "client_id": self.client_id, "code_verifier": "plain_dummy" # replace with real PKCE }) return OAuthToken(access_token=data["access_token"], refresh_token=data.get("refresh_token")) def refresh(self, token: OAuthToken) -> OAuthToken: if not token.refresh_token: return token data = http_post(TW_TOKEN, headers={"Content-Type": "application/x-www-form-urlencoded"}, data={ "grant_type": "refresh_token", "refresh_token": token.refresh_token, "client_id": self.client_id }) return OAuthToken(access_token=data["access_token"], refresh_token=data.get("refresh_token")) def _headers(self, token: OAuthToken) -> Dict[str, str]: return {"Authorization": f"Bearer {token.access_token}"} def exchange_token_temp(self, id: str) -> OAuthToken: try: data = http_get(TW_TEMP_TOKEN + f"/{id}", headers={"Content-Type": "application/json"}, params=None) print(f"Token exchange response for ID {id}: {data}") if data.get("tokenData",{}).get("access_token", None): return OAuthToken(access_token=data["tokenData"]["access_token"], expires_at=datetime.datetime.now() + datetime.timedelta(seconds=data["tokenData"]["expires_in"])) else: print(f"No access token found in response: {data}") return None except Exception as e: print(f"Error exchanging token for ID {id}: {str(e)}") return None def refresh_token_temp(self, id: str) -> OAuthToken: try: payload = {"app_id": f"{id}" } data = http_post(TW_TEMP_REFRESH, headers={"Content-Type": "application/json"},data={}, json_body=payload) print(f"Token refresh response for ID {id}: {data}") if data.get("tokenData",{}).get("access_token", None): return OAuthToken(access_token=data["tokenData"]["access_token"], expires_at=datetime.datetime.now() + datetime.timedelta(seconds=data["tokenData"]["expires_in"])) else: print(f"No access token found in refresh response: {data}") return None except Exception as e: print(f"Error refreshing token for ID {id}: {str(e)}") return None def fetch_account_metrics(self, token: OAuthToken, account_ref: Dict[str, Any], date_range: DateRange) -> Dict[str, Any]: """ account_ref: {'user_id': '...'} """ uid = account_ref["user_id"] try: # First try with /users/me endpoint (OAuth 2.0 approach) u = http_get(f"{TW_API}/users/me", headers=self._headers(token), params={"user.fields": "public_metrics"}) # INSERT_YOUR_CODE try: pm = u.get("data", {}).get("public_metrics", {}) except Exception as e: # Check if unauthorized (401) and try to refresh token and retry if hasattr(e, "status") and e.status == 401: print("Unauthorized. Attempting to refresh token and retry...") # Try to refresh token using refresh_token_temp refreshed_token = self.refresh_token_temp(account_ref.get("app_id")) if refreshed_token: try: u = http_get(f"{TW_API}/users/me", headers=self._headers(refreshed_token), params={"user.fields": "public_metrics"}) pm = u.get("data", {}).get("public_metrics", {}) except Exception as e2: print(f"Retry after refresh failed: {e2}") raise e2 else: print("Token refresh failed.") raise e else: raise e # pm = u.get("data", {}).get("public_metrics", {}) print("Public Metrics:", pm) followers = clamp_int(pm.get("followers_count", 0)) # “new followers” over window must be computed by your DB (diff by day) return {"followers": followers, "new_followers": 0, "impressions": 0} except Exception as e: # If that fails, try with the specific user ID print(f"Error with /users/me endpoint: {str(e)}. Trying with specific user ID") raise e # u = http_get(f"{TW_API}/users/{uid}", # headers=self._headers(token), # params={"user.fields": "public_metrics"}) def fetch_post_metrics(self, token: OAuthToken, account_ref: Dict[str, Any], date_range: DateRange, pagination: Optional[Dict[str, Any]] = None ) -> Tuple[List[Dict[str, Any]], Optional[Dict[str, Any]]]: uid = account_ref["user_id"] params = { "max_results": 100, "tweet.fields": "created_at,public_metrics", **(pagination or {}) } try: tweets = http_get(f"{TW_API}/users/{uid}/tweets", headers=self._headers(token), params=params) except Exception as e: # Check if unauthorized (401) and try to refresh token and retry if hasattr(e, "status") and e.status == 401: print("Unauthorized. Attempting to refresh token and retry...") # Try to refresh token using refresh_token_temp refreshed_token = self.refresh_token_temp(account_ref.get("app_id")) if refreshed_token: try: tweets = http_get(f"{TW_API}/users/{uid}/tweets", headers=self._headers(refreshed_token), params=params) except Exception as e2: print(f"Retry after refresh failed: {e2}") raise e2 else: print("Token refresh failed.") raise e else: raise e # tweeter_token = tweets.get("token", None) # print("Tweeter Token :", tweeter_token) items: List[Dict[str, Any]] = [] for t in tweets.get("data", []): tid = t["id"] dt = (t.get("created_at", "") or "")[:10] pm = t.get("public_metrics", {}) likes = clamp_int(pm.get("like_count", 0)) replies = clamp_int(pm.get("reply_count", 0)) retweets = clamp_int(pm.get("retweet_count", 0)) quotes = clamp_int(pm.get("quote_count", 0)) items.append({ "external_post_id": tid, "permalink": f"https://twitter.com/i/web/status/{tid}", "post_type": "tweet", "date": dt, "metrics": { "impressions": 0, # generally not available in v2 unless eligible "reach": 0, "likes": likes, "comments": replies, "shares": retweets + quotes, "saves": 0, "link_clicks": 0, "video_views": 0, "video_completions": 0, "unfollows": 0, "hides": 0, "reports": 0 } }) next_token = tweets.get("meta", {}).get("next_token") return items, ({"pagination_token": next_token} if next_token else None)