"""X (Twitter) API v2 with OAuth 2.0 user context (PKCE). Threads are reply chains; images and a video go on the first post (v2 media upload: one call for an image, initialize / append / finalize / status for a video).""" import base64 import hashlib import secrets import time from urllib.parse import urlencode import httpx from sqlalchemy.orm import Session from app.config import settings from app.models import SocialAccount from app.services.crypto import decrypt from app.services.social import SocialError, expired, save_tokens, scope_set, token AUTH_URL = "https://x.com/i/oauth2/authorize" TOKEN_URL = "https://api.x.com/2/oauth2/token" API = "https://api.x.com/2" SCOPES = "tweet.read tweet.write users.read offline.access media.write" CHUNK = 4 * 1024 * 1024 # video upload segments PROCESSING_WAIT = 600 # seconds X may take to process a video before we give up def redirect_uri() -> str: return f"{settings.api_url.rstrip('/')}/api/social/x/callback" def configured() -> bool: return bool(settings.x_enabled and settings.x_client_id and settings.x_client_secret) def pkce_pair() -> tuple[str, str]: verifier = secrets.token_urlsafe(64)[:96] challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).decode().rstrip("=") return verifier, challenge def authorize_url(state: str, challenge: str) -> str: return AUTH_URL + "?" + urlencode({ "response_type": "code", "client_id": settings.x_client_id, "redirect_uri": redirect_uri(), "scope": SCOPES, "state": state, "code_challenge": challenge, "code_challenge_method": "S256", }) def _token_request(data: dict) -> dict: resp = httpx.post(TOKEN_URL, data={**data, "client_id": settings.x_client_id}, auth=(settings.x_client_id, settings.x_client_secret), timeout=20) if resp.status_code >= 400: raise SocialError(f"X token request failed: {resp.text[:300]}", permanent=resp.status_code in (400, 401)) return resp.json() def exchange_code(code: str, verifier: str) -> dict: tokens = _token_request({"grant_type": "authorization_code", "code": code, "redirect_uri": redirect_uri(), "code_verifier": verifier}) me = httpx.get(f"{API}/users/me", params={"user.fields": "profile_image_url"}, headers={"Authorization": f"Bearer {tokens['access_token']}"}, timeout=20) me.raise_for_status() user = me.json()["data"] return {**tokens, "external_id": user["id"], "handle": user["username"], "avatar_url": user.get("profile_image_url")} def refresh(db: Session, account: SocialAccount) -> None: """Swap the refresh token for new tokens (X rotates the refresh token too). A refused refresh means the connection is gone: reconnect.""" if not account.refresh_token: raise SocialError("The X connection expired. Reconnect X.", reconnect=True) try: tokens = _token_request({"grant_type": "refresh_token", "refresh_token": decrypt(account.refresh_token)}) except SocialError as exc: if exc.permanent: raise SocialError("The X connection expired. Reconnect X.", reconnect=True) from exc raise save_tokens(db, account, tokens["access_token"], tokens.get("refresh_token"), tokens.get("expires_in")) def _access(db: Session, account: SocialAccount) -> str: if expired(account): refresh(db, account) return token(account) def check_alive(db: Session, account: SocialAccount) -> None: """One cheap call: raises SocialError(reconnect=True) when X no longer accepts this connection.""" resp = _call("GET", f"{API}/users/me", headers={"Authorization": f"Bearer {_access(db, account)}"}, timeout=20) if resp.status_code == 401: raise SocialError("X no longer accepts this connection. Reconnect X.", reconnect=True) def _call(method: str, url: str, **kw) -> httpx.Response: """Every request to X goes through here (tests replace it).""" return httpx.request(method, url, timeout=kw.pop("timeout", 60), **kw) def can_post_media(account: SocialAccount) -> bool: """Accounts connected before media.write was asked for have to reconnect to post images or videos.""" return "media.write" in scope_set(account.scopes) def _media_error(resp: httpx.Response) -> SocialError: if resp.status_code in (401, 403): return SocialError("X refused the upload. Reconnect X to post images and videos.", reconnect=True) return SocialError(f"X media upload {resp.status_code}: {resp.text[:300]}", permanent=400 <= resp.status_code < 500 and resp.status_code != 429) def _upload_image(headers: dict, m) -> str: resp = _call("POST", f"{API}/media/upload", headers=headers, data={"media_category": "tweet_image"}, files={"media": (m.filename, m.read(), m.content_type)}) if resp.status_code >= 400: raise _media_error(resp) return resp.json()["data"]["id"] def _upload_video(headers: dict, m) -> str: resp = _call("POST", f"{API}/media/upload/initialize", headers=headers, json={"media_type": m.content_type, "total_bytes": m.size, "media_category": "tweet_video"}) if resp.status_code >= 400: raise _media_error(resp) media_id = resp.json()["data"]["id"] for i, start in enumerate(range(0, m.size, CHUNK)): resp = _call("POST", f"{API}/media/upload/{media_id}/append", headers=headers, data={"segment_index": i}, files={"media": (m.filename, m.read(start, start + CHUNK), "application/octet-stream")}) if resp.status_code >= 400: raise _media_error(resp) resp = _call("POST", f"{API}/media/upload/{media_id}/finalize", headers=headers) if resp.status_code >= 400: raise _media_error(resp) info = (resp.json().get("data") or {}).get("processing_info") waited = 0 while info and info.get("state") in ("pending", "in_progress"): if waited >= PROCESSING_WAIT: raise SocialError("X took too long to process the video.") pause = int(info.get("check_after_secs") or 5) time.sleep(pause) waited += pause resp = _call("GET", f"{API}/media/upload", headers=headers, params={"command": "STATUS", "media_id": media_id}) if resp.status_code >= 400: raise _media_error(resp) info = (resp.json().get("data") or {}).get("processing_info") if info and info.get("state") == "failed": reason = (info.get("error") or {}).get("message") or "X could not process the video" raise SocialError(f"{reason}. X takes videos up to 2:20.", permanent=True) return media_id def publish(db: Session, account: SocialAccount, posts: list[str], media: list | None = None) -> tuple[str, str]: headers = {"Authorization": f"Bearer {_access(db, account)}"} media_ids: list[str] = [] if media: if not can_post_media(account): raise SocialError("Reconnect X to post images and videos.", permanent=True) media_ids = [(_upload_video if m.kind == "video" else _upload_image)(headers, m) for m in media] first_id, reply_to = None, None for text in posts: body: dict = {"text": text} if text else {} # X takes a post with media and no text, but not an empty text if reply_to: body["reply"] = {"in_reply_to_tweet_id": reply_to} elif media_ids: body["media"] = {"media_ids": media_ids} # the images or video go on the first post resp = _call("POST", f"{API}/tweets", json=body, headers=headers, timeout=20) if resp.status_code == 401: raise SocialError("X rejected the connection. Reconnect X.", reconnect=True) if resp.status_code >= 400: raise SocialError(f"X {resp.status_code}: {resp.text[:300]}", permanent=resp.status_code in (400, 403)) reply_to = resp.json()["data"]["id"] first_id = first_id or reply_to return first_id, f"https://x.com/{account.handle}/status/{first_id}" def metrics(db: Session, account: SocialAccount, external_id: str) -> dict: resp = httpx.get(f"{API}/tweets", params={"ids": external_id, "tweet.fields": "public_metrics"}, headers={"Authorization": f"Bearer {_access(db, account)}"}, timeout=20) if resp.status_code >= 400: return {} data = (resp.json().get("data") or [{}])[0].get("public_metrics") or {} return {"likes": data.get("like_count", 0), "reposts": data.get("retweet_count", 0), "replies": data.get("reply_count", 0), "impressions": data.get("impression_count", 0)}