diff --git a/app/agents/orchestrator.py b/app/agents/orchestrator.py index 2c4b090..8c4bf43 100644 --- a/app/agents/orchestrator.py +++ b/app/agents/orchestrator.py @@ -2,12 +2,18 @@ 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.utils.logger import setup_logger logger = setup_logger("orchestrator") 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)) @@ -15,8 +21,14 @@ async def run_pipeline() -> None: for topic in topics: assert topic.id is not None, "topic must be persisted before use" - if await generate_post(topic): + approved = await generate_post(topic) + if approved is not None: mark_topic_status(topic.id, STATUS_USED) + + try: + publish_post(approved.linkedin_content) + except LinkedInPublishError: + logger.exception("Failed to publish to LinkedIn for topic '%s'", topic.title) return mark_topic_status(topic.id, STATUS_REJECTED) diff --git a/app/agents/writer_agent.py b/app/agents/writer_agent.py index 584e3b6..f5d1f40 100644 --- a/app/agents/writer_agent.py +++ b/app/agents/writer_agent.py @@ -1,5 +1,6 @@ import asyncio import json +from dataclasses import dataclass from typing import TypedDict from google.genai import types @@ -16,6 +17,12 @@ logger = setup_logger("writer-agent") +@dataclass +class ApprovedPost: + linkedin_content: str + x_content: str + + class DraftPost(TypedDict): platform: str post_content: str @@ -86,7 +93,7 @@ def _draft_post(topic: Topic) -> list[DraftPost]: return validated_posts -async def generate_post(topic: Topic) -> bool: +async def generate_post(topic: Topic) -> ApprovedPost | None: """ Draft a post for `topic`, send it to Discord, and keep regenerating on "Re-try" until the reviewer Accepts or Rejects. Returns True if accepted. @@ -144,5 +151,7 @@ async def generate_post(topic: Topic) -> bool: if decision == ApprovalDecision.RETRY: continue + if decision == ApprovalDecision.ACCEPTED: + return ApprovedPost(linkedin_content=linkedin_post, x_content=x_post) - return decision == ApprovalDecision.ACCEPTED + return None diff --git a/app/config/settings.py b/app/config/settings.py index 0eb0659..9b818a5 100644 --- a/app/config/settings.py +++ b/app/config/settings.py @@ -1,6 +1,4 @@ -import datetime from pathlib import Path -from zoneinfo import ZoneInfo from pydantic_settings import BaseSettings, SettingsConfigDict @@ -10,9 +8,6 @@ class Settings(BaseSettings): model_config = SettingsConfigDict(env_file=BASE_DIR / ".env", extra="ignore") - # Current Date - DATE: str = datetime.datetime.now(ZoneInfo("Asia/Kolkata")).date().isoformat() - # Database DATABASE_URL: str diff --git a/app/publishers/linkedin.py b/app/publishers/linkedin.py new file mode 100644 index 0000000..6714e64 --- /dev/null +++ b/app/publishers/linkedin.py @@ -0,0 +1,83 @@ +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("linkedin-publisher") + +LINKEDIN_POST_URL = "https://api.linkedin.com/rest/posts" + + +class LinkedInPublishError(Exception): + """Raised when LinkedIn rejects a post, or accepts it without returning a post URN.""" + + +def _headers() -> dict[str, str]: + return { + "Authorization": f"Bearer {settings.LINKEDIN_ACCESS_TOKEN}", + "Content-Type": "application/json", + "LinkedIn-Version": settings.LINKEDIN_API_VERSION, + "X-Restli-Protocol-Version": "2.0.0", + } + + +def _is_retryable(exc: BaseException) -> bool: + """Retry on network failures and LinkedIn 5xx errors - not on 4xx (retrying won't fix a bad token or payload).""" + + 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_linkedin(payload: dict) -> httpx.Response: + response = httpx.post(LINKEDIN_POST_URL, headers=_headers(), json=payload, timeout=15) + response.raise_for_status() + return response + + +def publish_post(content: str) -> str: + """ + Publish `content` as a text post on the configured LinkedIn member profile. + Returns the new post's URN + """ + + payload = { + "author": f"urn:li:person:{settings.LINKEDIN_PERSON_ID}", + "commentary": content, + "visibility": "PUBLIC", + "distribution": { + "feedDistribution": "MAIN_FEED", + "targetEntities": [], + "thirdPartyDistributionChannels": [], + }, + "lifecycleState": "PUBLISHED", + "isReshareDisabledByAuthor": False, + } + + try: + response = _post_to_linkedin(payload) + except httpx.HTTPStatusError as e: + logger.error("LinkedIn publish failed: %s - %s", e.response.status_code, e.response.text) + raise LinkedInPublishError(f"LinkedIn API returned {e.response.status_code}: {e.response.text}") from e + + post_urn = response.headers.get("x-restli-id") + if post_urn is None: + try: + post_urn = response.json().get("id") + except ValueError: + post_urn = None + + if post_urn is None: + raise LinkedInPublishError("LinkedIn accepted the post but returned no identifiable post URN.") + + logger.info("Published to LinkedIn: %s", post_urn) + return post_urn diff --git a/tests/agents/test_orchestrator.py b/tests/agents/test_orchestrator.py index ce6a35f..061b50c 100644 --- a/tests/agents/test_orchestrator.py +++ b/tests/agents/test_orchestrator.py @@ -2,6 +2,7 @@ from app.agents import orchestrator from app.database.models.topic import STATUS_REJECTED, STATUS_USED +from app.publishers.linkedin import LinkedInPublishError class _FakeTopic: @@ -14,22 +15,27 @@ 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") mocker.patch("app.agents.orchestrator.get_topics", return_value=topics) - mock_generate_post = mocker.patch("app.agents.orchestrator.generate_post", side_effect=[True]) + 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") 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") @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") mocker.patch("app.agents.orchestrator.get_topics", return_value=topics) - mocker.patch("app.agents.orchestrator.generate_post", side_effect=[False, True]) + 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") await orchestrator.run_pipeline() @@ -43,15 +49,33 @@ 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)] - 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=[False, True]) + approved = mocker.Mock(linkedin_content="Fake LinkedIn 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") await orchestrator.run_pipeline() assert mock_get_topics.call_count == 2 +@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") + 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")) + + await orchestrator.run_pipeline() + + mock_mark_status.assert_called_once_with(1, STATUS_USED) + + @pytest.mark.asyncio async def test_run_pipeline_asserts_topic_must_be_persisted(mocker): mocker.patch("app.agents.orchestrator.get_topics", return_value=[_FakeTopic(id=None)]) diff --git a/tests/agents/test_writer_agent.py b/tests/agents/test_writer_agent.py index 1817bbb..ec5ffa5 100644 --- a/tests/agents/test_writer_agent.py +++ b/tests/agents/test_writer_agent.py @@ -52,7 +52,7 @@ def test_draft_post_raises_on_non_string_platform(mocker): @pytest.mark.asyncio -async def test_generate_post_returns_true_on_accept(mocker): +async def test_generate_post_returns_approved_post_on_accept(mocker): mocker.patch("app.agents.writer_agent._draft_post", return_value=SAMPLE_POSTS) mocker.patch("app.agents.writer_agent.send_for_approval", new=AsyncMock(return_value=MagicMock())) @@ -62,11 +62,13 @@ async def test_generate_post_returns_true_on_accept(mocker): result = await writer_agent.generate_post(_FakeTopic()) - assert result is True + assert result is not None + assert result.linkedin_content == "LinkedIn body" + assert result.x_content == "X body" @pytest.mark.asyncio -async def test_generate_post_returns_false_on_reject(mocker): +async def test_generate_post_returns_none_on_reject(mocker): mocker.patch("app.agents.writer_agent._draft_post", return_value=SAMPLE_POSTS) mocker.patch("app.agents.writer_agent.send_for_approval", new=AsyncMock(return_value=MagicMock())) @@ -76,7 +78,7 @@ async def test_generate_post_returns_false_on_reject(mocker): result = await writer_agent.generate_post(_FakeTopic()) - assert result is False + assert result is None @pytest.mark.asyncio @@ -92,7 +94,8 @@ async def test_generate_post_redrafts_on_retry_then_accepts(mocker): result = await writer_agent.generate_post(_FakeTopic()) - assert result is True + assert result is not None + assert result.linkedin_content == "LinkedIn body" assert mock_draft.call_count == 2