Add publisher for X(Twitter) to post a Post. This code is not the per…

.github/workflows/ci.yml

@@ -44,4 +44,8 @@ jobs:
                 LINKEDIN_REFRESH_TOKEN: "refresh-token"
                 LINKEDIN_PERSON_ID: "person-id"
                 LINKEDIN_API_VERSION: "api-version"
+                X_CLIENT_ID: "your_client_id"
+                X_CLIENT_SECRET: "your_client_secret"
+                X_ACCESS_TOKEN: "your_access_token_here"
+                X_REFRESH_TOKEN: "your_refresh_token"
               run: pytest -v

app/agents/orchestrator.py

@@ -2,7 +2,10 @@
 from app.database.models.topic import STATUS_REJECTED, STATUS_USED
 from app.discovery.feeds import get_topics
 from app.discovery.repository import mark_topic_status
-from app.publishers.linkedin import LinkedInPublishError, publish_post
+from app.publishers.linkedin import LinkedInPublishError
+from app.publishers.linkedin import publish_post as publish_linkedin_post
+from app.publishers.x import XPublishError
+from app.publishers.x import publish_post as publish_x_post
 from app.utils.logger import setup_logger
 
 logger = setup_logger("orchestrator")
@@ -13,7 +16,6 @@ async def run_pipeline() -> None:
     Work through topic ideas (cached first, freshly generated if needed)
     one at a time. Stops as soon as one post is approved and published.
     """
-
     while True:
         topics = get_topics()
         logger.info("Got %d candidate topic(s) to try.", len(topics))
@@ -26,16 +28,24 @@ async def run_pipeline() -> None:
                 mark_topic_status(topic.id, STATUS_USED)
 
                 try:
-                    publish_post(approved.linkedin_content)
+                    publish_linkedin_post(approved.linkedin_content)
                 except LinkedInPublishError:
                     logger.exception("Failed to publish to LinkedIn for topic '%s'", topic.title)
+
+                try:
+                    publish_x_post(approved.x_content)
+                except XPublishError:
+                    logger.exception("Failed to publish to X for topic '%s'", topic.title)
+
                 return
 
             mark_topic_status(topic.id, STATUS_REJECTED)
 
         logger.info("All cached topics rejected - will generate a fresh batch.")
 
+
 if __name__ == "__main__":
     import asyncio
 
     asyncio.run(run_pipeline())
+

app/config/settings.py

@@ -28,6 +28,12 @@ class Settings(BaseSettings):
     LINKEDIN_PERSON_ID: str
     LINKEDIN_API_VERSION: str
 
+    # X
+    X_CLIENT_ID: str
+    X_CLIENT_SECRET: str
+    X_ACCESS_TOKEN: str
+    X_REFRESH_TOKEN: str
+
 
 settings = Settings()
 

app/publishers/x.py

@@ -0,0 +1,60 @@
+import httpx
+from tenacity import retry, retry_if_exception, stop_after_attempt, wait_exponential
+
+from app.config.settings import settings
+from app.utils.logger import setup_logger
+
+logger = setup_logger("x-publisher")
+
+X_POSTS_URL = "https://api.x.com/2/tweets"
+
+
+class XPublishError(Exception):
+    """Raised when X rejects a post, or accepts it without returning a post ID."""
+
+
+def _headers() -> dict[str, str]:
+    return {
+        "Authorization": f"Bearer {settings.X_ACCESS_TOKEN}",
+        "Content-Type": "application/json",
+    }
+
+
+def _is_retryable(exc: BaseException) -> bool:
+    if isinstance(exc, httpx.TransportError):
+        return True
+    if isinstance(exc, httpx.HTTPStatusError):
+        return exc.response.status_code >= 500
+    return False
+
+
+@retry(
+    retry=retry_if_exception(_is_retryable),
+    stop=stop_after_attempt(3),
+    wait=wait_exponential(multiplier=2, min=2, max=20),
+    reraise=True,
+)
+def _post_to_x(payload: dict) -> httpx.Response:
+    response = httpx.post(X_POSTS_URL, headers=_headers(), json=payload, timeout=15)
+    response.raise_for_status()
+    return response
+
+
+def publish_post(content: str) -> str:
+    """
+    Publish `content` as a post on the configured X account.
+    Returns the new post's ID.
+    """
+    try:
+        response = _post_to_x({"text": content})
+    except httpx.HTTPStatusError as e:
+        logger.error("X publish failed: %s - %s", e.response.status_code, e.response.text)
+        raise XPublishError(f"X API returned {e.response.status_code}: {e.response.text}") from e
+
+    try:
+        post_id = response.json()["data"]["id"]
+    except (ValueError, KeyError) as e:
+        raise XPublishError("X accepted the post but returned no identifiable post ID.") from e
+
+    logger.info("Published to X: %s", post_id)
+    return post_id

scripts/linkedin_authorize.py

@@ -0,0 +1,97 @@
+"""
+Run once, manually, to obtain LinkedIn OAuth credentials for .env.
+Starts a tiny local server to catch LinkedIn's redirect automatically.
+"""
+
+import threading
+import webbrowser
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from urllib.parse import parse_qs, urlencode, urlparse
+
+import httpx
+
+CLIENT_ID = input("LinkedIn Client ID: ").strip()
+CLIENT_SECRET = input("LinkedIn Client Secret: ").strip()
+REDIRECT_URI = "http://localhost:8080/callback"
+
+captured_code: dict[str, str] = {}
+
+
+class CallbackHandler(BaseHTTPRequestHandler):
+    def do_GET(self):
+        query = parse_qs(urlparse(self.path).query)
+        code = query.get("code", [None])[0]
+        error = query.get("error_description", [None])[0]
+
+        self.send_response(200)
+        self.send_header("Content-Type", "text/html")
+        self.end_headers()
+
+        if code:
+            captured_code["code"] = code
+            self.wfile.write(b"<h2>Success! You can close this tab and return to the terminal.</h2>")
+        else:
+            self.wfile.write(f"<h2>Error: {error or 'no code received'}</h2>".encode())
+
+    def log_message(self, format, *args):
+        pass  # silence default request logging
+
+
+def _run_server_until_code_received():
+    server = HTTPServer(("localhost", 8080), CallbackHandler)
+    while "code" not in captured_code:
+        server.handle_request()
+
+
+server_thread = threading.Thread(target=_run_server_until_code_received)
+server_thread.start()
+
+auth_url = "https://www.linkedin.com/oauth/v2/authorization?" + urlencode(
+    {
+        "response_type": "code",
+        "client_id": CLIENT_ID,
+        "redirect_uri": REDIRECT_URI,
+        "scope": "openid profile email w_member_social",
+        "state": "linkedin-setup",
+    }
+)
+
+print(f"\nOpening browser for LinkedIn login...\nIf it doesn't open automatically, visit:\n{auth_url}\n")
+webbrowser.open(auth_url)
+
+server_thread.join(timeout=120)
+
+if "code" not in captured_code:
+    raise SystemExit("Timed out waiting for LinkedIn's redirect. Did you approve access in the browser?")
+
+code = captured_code["code"]
+print("Authorization code captured automatically.")
+
+token_response = httpx.post(
+    "https://www.linkedin.com/oauth/v2/accessToken",
+    data={
+        "grant_type": "authorization_code",
+        "code": code,
+        "redirect_uri": REDIRECT_URI,
+        "client_id": CLIENT_ID,
+        "client_secret": CLIENT_SECRET,
+    },
+    headers={"Content-Type": "application/x-www-form-urlencoded"},
+)
+token_response.raise_for_status()
+tokens = token_response.json()
+
+userinfo_response = httpx.get(
+    "https://api.linkedin.com/v2/userinfo",
+    headers={"Authorization": f"Bearer {tokens['access_token']}"},
+)
+userinfo_response.raise_for_status()
+person_id = userinfo_response.json()["sub"]
+
+print("\nAdd these to your .env:\n")
+print(f"LINKEDIN_CLIENT_ID={CLIENT_ID}")
+print(f"LINKEDIN_CLIENT_SECRET={CLIENT_SECRET}")
+print(f"LINKEDIN_ACCESS_TOKEN={tokens['access_token']}")
+print(f"LINKEDIN_REFRESH_TOKEN={tokens.get('refresh_token', '')}")
+print(f"LINKEDIN_PERSON_ID={person_id}")
+print("LINKEDIN_API_VERSION=202401")

scripts/x_authorize.py

@@ -0,0 +1,108 @@
+"""
+Run once, manually, to obtain X (Twitter) OAuth 2.0 User Context credentials for .env.
+Starts a tiny local server to catch X's redirect automatically.
+"""
+
+import base64
+import hashlib
+import re
+import secrets
+import threading
+import webbrowser
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from urllib.parse import parse_qs, urlencode, urlparse
+
+import httpx
+
+CLIENT_ID = input("X Client ID: ").strip()
+CLIENT_SECRET = input("X Client Secret: ").strip()
+REDIRECT_URI = "http://localhost:8081/callback"
+SCOPES = "tweet.read tweet.write users.read offline.access"
+
+captured_code: dict[str, str] = {}
+
+
+def _generate_pkce_pair() -> tuple[str, str]:
+    verifier = re.sub(r"[^a-zA-Z0-9]", "", base64.urlsafe_b64encode(secrets.token_bytes(40)).decode())
+    challenge = base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).decode().rstrip("=")
+    return verifier, challenge
+
+
+code_verifier, code_challenge = _generate_pkce_pair()
+
+
+class CallbackHandler(BaseHTTPRequestHandler):
+    def do_GET(self):
+        query = parse_qs(urlparse(self.path).query)
+        code = query.get("code", [None])[0]
+        error = query.get("error_description", [None])[0]
+
+        self.send_response(200)
+        self.send_header("Content-Type", "text/html")
+        self.end_headers()
+
+        if code:
+            captured_code["code"] = code
+            self.wfile.write(b"<h2>Success! You can close this tab and return to the terminal.</h2>")
+        else:
+            self.wfile.write(f"<h2>Error: {error or 'no code received'}</h2>".encode())
+
+    def log_message(self, format, *args):
+        pass
+
+
+def _run_server_until_code_received():
+    server = HTTPServer(("localhost", 8081), CallbackHandler)
+    while "code" not in captured_code:
+        server.handle_request()
+
+
+server_thread = threading.Thread(target=_run_server_until_code_received)
+server_thread.start()
+
+auth_url = "https://twitter.com/i/oauth2/authorize?" + urlencode(
+    {
+        "response_type": "code",
+        "client_id": CLIENT_ID,
+        "redirect_uri": REDIRECT_URI,
+        "scope": SCOPES,
+        "state": "x-setup",
+        "code_challenge": code_challenge,
+        "code_challenge_method": "S256",
+    }
+)
+
+print(f"\nOpening browser for X login...\nIf it doesn't open automatically, visit:\n{auth_url}\n")
+webbrowser.open(auth_url)
+
+server_thread.join(timeout=120)
+
+if "code" not in captured_code:
+    raise SystemExit("Timed out waiting for X's redirect. Did you approve access in the browser?")
+
+code = captured_code["code"]
+print("Authorization code captured automatically.")
+
+basic_auth = base64.b64encode(f"{CLIENT_ID}:{CLIENT_SECRET}".encode()).decode()
+
+token_response = httpx.post(
+    "https://api.x.com/2/oauth2/token",
+    data={
+        "grant_type": "authorization_code",
+        "code": code,
+        "redirect_uri": REDIRECT_URI,
+        "code_verifier": code_verifier,
+    },
+    headers={
+        "Content-Type": "application/x-www-form-urlencoded",
+        "Authorization": f"Basic {basic_auth}",
+    },
+)
+token_response.raise_for_status()
+tokens = token_response.json()
+
+print("\nAdd these to your .env:\n")
+print(f"X_CLIENT_ID={CLIENT_ID}")
+print(f"X_CLIENT_SECRET={CLIENT_SECRET}")
+print(f"X_ACCESS_TOKEN={tokens['access_token']}")
+print(f"X_REFRESH_TOKEN={tokens.get('refresh_token', '')}")