diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index ef79902..2ff4faa 100644 --- a/.github/workflows/ci.yml +++ b/.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 \ No newline at end of file diff --git a/app/agents/orchestrator.py b/app/agents/orchestrator.py index 8c4bf43..0a6bd46 100644 --- a/app/agents/orchestrator.py +++ b/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()) + \ No newline at end of file diff --git a/app/config/settings.py b/app/config/settings.py index 9b818a5..da67c86 100644 --- a/app/config/settings.py +++ b/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() diff --git a/app/publishers/x.py b/app/publishers/x.py new file mode 100644 index 0000000..f57d941 --- /dev/null +++ b/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 diff --git a/scripts/linkedin_authorize.py b/scripts/linkedin_authorize.py new file mode 100644 index 0000000..5f18a01 --- /dev/null +++ b/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"

Success! You can close this tab and return to the terminal.

") + else: + self.wfile.write(f"

Error: {error or 'no code received'}

".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") \ No newline at end of file diff --git a/scripts/x_authorize.py b/scripts/x_authorize.py new file mode 100644 index 0000000..1780277 --- /dev/null +++ b/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"

Success! You can close this tab and return to the terminal.

") + else: + self.wfile.write(f"

Error: {error or 'no code received'}

".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', '')}") diff --git a/tests/agents/test_orchestrator.py b/tests/agents/test_orchestrator.py index 061b50c..2ff6fd3 100644 --- a/tests/agents/test_orchestrator.py +++ b/tests/agents/test_orchestrator.py @@ -3,6 +3,7 @@ from app.agents import orchestrator from app.database.models.topic import STATUS_REJECTED, STATUS_USED from app.publishers.linkedin import LinkedInPublishError +from app.publishers.x import XPublishError class _FakeTopic: @@ -15,27 +16,30 @@ def __init__(self, id, title="Idea", rank=1): @pytest.mark.asyncio async def test_run_pipeline_stops_on_first_accepted_topic(mocker): topics = [_FakeTopic(id=1), _FakeTopic(id=2)] - approved = mocker.Mock(linkedin_content="Fake LinkedIn content") + approved = mocker.Mock(linkedin_content="Fake LinkedIn content", x_content="Fake X content") mocker.patch("app.agents.orchestrator.get_topics", return_value=topics) mock_generate_post = mocker.patch("app.agents.orchestrator.generate_post", side_effect=[approved]) mock_mark_status = mocker.patch("app.agents.orchestrator.mark_topic_status") - mock_publish = mocker.patch("app.agents.orchestrator.publish_post") + mock_publish_linkedin = mocker.patch("app.agents.orchestrator.publish_linkedin_post") + mock_publish_x = mocker.patch("app.agents.orchestrator.publish_x_post") await orchestrator.run_pipeline() mock_generate_post.assert_awaited_once_with(topics[0]) mock_mark_status.assert_called_once_with(1, STATUS_USED) - mock_publish.assert_called_once_with("Fake LinkedIn content") + mock_publish_linkedin.assert_called_once_with("Fake LinkedIn content") + mock_publish_x.assert_called_once_with("Fake X content") @pytest.mark.asyncio async def test_run_pipeline_tries_next_topic_after_rejection(mocker): topics = [_FakeTopic(id=1), _FakeTopic(id=2)] - approved = mocker.Mock(linkedin_content="Fake LinkedIn content") + approved = mocker.Mock(linkedin_content="Fake LinkedIn content", x_content="Fake X content") mocker.patch("app.agents.orchestrator.get_topics", return_value=topics) mocker.patch("app.agents.orchestrator.generate_post", side_effect=[None, approved]) mock_mark_status = mocker.patch("app.agents.orchestrator.mark_topic_status") - mocker.patch("app.agents.orchestrator.publish_post") + mocker.patch("app.agents.orchestrator.publish_linkedin_post") + mocker.patch("app.agents.orchestrator.publish_x_post") await orchestrator.run_pipeline() @@ -49,13 +53,14 @@ async def test_run_pipeline_tries_next_topic_after_rejection(mocker): async def test_run_pipeline_regenerates_when_all_topics_rejected(mocker): first_batch = [_FakeTopic(id=1)] second_batch = [_FakeTopic(id=2)] - approved = mocker.Mock(linkedin_content="Fake LinkedIn content") + approved = mocker.Mock(linkedin_content="Fake LinkedIn content", x_content="Fake X content") mock_get_topics = mocker.patch( "app.agents.orchestrator.get_topics", side_effect=[first_batch, second_batch] ) mocker.patch("app.agents.orchestrator.generate_post", side_effect=[None, approved]) mocker.patch("app.agents.orchestrator.mark_topic_status") - mocker.patch("app.agents.orchestrator.publish_post") + mocker.patch("app.agents.orchestrator.publish_linkedin_post") + mocker.patch("app.agents.orchestrator.publish_x_post") await orchestrator.run_pipeline() @@ -65,11 +70,28 @@ async def test_run_pipeline_regenerates_when_all_topics_rejected(mocker): @pytest.mark.asyncio async def test_run_pipeline_logs_and_continues_when_linkedin_publish_fails(mocker): topics = [_FakeTopic(id=1)] - approved = mocker.Mock(linkedin_content="Fake LinkedIn content") + approved = mocker.Mock(linkedin_content="Fake LinkedIn content", x_content="Fake X content") mocker.patch("app.agents.orchestrator.get_topics", return_value=topics) mocker.patch("app.agents.orchestrator.generate_post", side_effect=[approved]) mock_mark_status = mocker.patch("app.agents.orchestrator.mark_topic_status") - mocker.patch("app.agents.orchestrator.publish_post", side_effect=LinkedInPublishError("boom")) + mocker.patch("app.agents.orchestrator.publish_linkedin_post", side_effect=LinkedInPublishError("boom")) + mock_publish_x = mocker.patch("app.agents.orchestrator.publish_x_post") + + await orchestrator.run_pipeline() + + mock_mark_status.assert_called_once_with(1, STATUS_USED) + mock_publish_x.assert_called_once_with("Fake X content") + + +@pytest.mark.asyncio +async def test_run_pipeline_logs_and_continues_when_x_publish_fails(mocker): + topics = [_FakeTopic(id=1)] + approved = mocker.Mock(linkedin_content="Fake LinkedIn content", x_content="Fake X content") + mocker.patch("app.agents.orchestrator.get_topics", return_value=topics) + mocker.patch("app.agents.orchestrator.generate_post", side_effect=[approved]) + mock_mark_status = mocker.patch("app.agents.orchestrator.mark_topic_status") + mocker.patch("app.agents.orchestrator.publish_linkedin_post") + mocker.patch("app.agents.orchestrator.publish_x_post", side_effect=XPublishError("boom")) await orchestrator.run_pipeline() @@ -83,3 +105,4 @@ async def test_run_pipeline_asserts_topic_must_be_persisted(mocker): with pytest.raises(AssertionError): await orchestrator.run_pipeline() + \ No newline at end of file