Spaces:
Runtime error
Runtime error
| """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)} | |