refactor: dispatch poster by campaign.type via platform strategy registry
Replace hardcoded platform dispatch with get_client(campaign["type"]) so any future campaign type (blog_post, email, etc.) routes automatically through the strategy registry. Adds dupe-guard opt-out per strategy, sub_row pre-fetch for extra metadata, and 5 new TDD tests (14 total).
This commit is contained in:
parent
de6dc645f6
commit
81a63ab0ec
2 changed files with 167 additions and 34 deletions
|
|
@ -5,83 +5,101 @@ Called by the scheduler and by the manual-trigger API endpoint.
|
|||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
from app.core.config import get_settings
|
||||
from app.db.store import Store
|
||||
from app.services.platforms import get_client, SUPPORTED_PLATFORMS
|
||||
from app.services.platforms import get_client
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _run_post(db_path: str, campaign_id: int, sub: str, triggered_by: str = "scheduler") -> dict:
|
||||
def _run_post(db_path: str, campaign_id: int, target: str,
|
||||
triggered_by: str = "scheduler") -> dict:
|
||||
"""Execute a single post attempt (blocking, runs in a thread)."""
|
||||
store = Store(db_path)
|
||||
try:
|
||||
# Dupe guard: skip if already posted to this sub this week
|
||||
if store.already_posted_this_week(campaign_id, sub):
|
||||
return {"skipped": True, "reason": f"already posted to r/{sub} this week"}
|
||||
campaign = store.get_campaign(campaign_id)
|
||||
if campaign is None:
|
||||
return {"skipped": True, "reason": f"campaign {campaign_id} not found"}
|
||||
|
||||
# Resolve the best variant for this sub
|
||||
variant = store.resolve_variant(campaign_id, sub)
|
||||
campaign_type = campaign.get("type", "reddit_post")
|
||||
|
||||
# Resolve strategy — skip if type is unknown
|
||||
try:
|
||||
strategy = get_client(campaign_type)
|
||||
except ValueError as exc:
|
||||
return {"skipped": True, "reason": str(exc)}
|
||||
|
||||
# Fetch the campaign_subs row for this target (used for extra + occurrence)
|
||||
all_subs = store.list_campaign_subs(campaign_id)
|
||||
sub_row = next((s for s in all_subs if s["sub"] == target), {})
|
||||
|
||||
# Dupe guard (opt-out allowed per strategy)
|
||||
if strategy.supports_dupe_guard() and store.already_posted_this_week(campaign_id, target):
|
||||
return {"skipped": True, "reason": f"already posted to {target!r} this week"}
|
||||
|
||||
# Resolve best content variant for this target
|
||||
variant = store.resolve_variant(campaign_id, target)
|
||||
if variant is None:
|
||||
return {"skipped": True, "reason": "no variant found for campaign"}
|
||||
|
||||
# Check sub rules
|
||||
rules = store.get_sub_rules(sub)
|
||||
# Check platform rules (Reddit-specific; harmless for other platforms)
|
||||
rules = store.get_sub_rules(target)
|
||||
if rules and rules.get("promo_allowed") == 0:
|
||||
return {"skipped": True, "reason": f"r/{sub} hard-bans promotion"}
|
||||
return {"skipped": True, "reason": f"{target!r} hard-bans promotion"}
|
||||
|
||||
# Check platform support
|
||||
campaign = store.get_campaign(campaign_id)
|
||||
platform = campaign["platform"] if campaign else "reddit"
|
||||
if platform not in SUPPORTED_PLATFORMS:
|
||||
return {"skipped": True, "reason": f"platform '{platform}' not yet implemented"}
|
||||
|
||||
# Create the pending post record
|
||||
# Create pending post record
|
||||
post = store.create_post(
|
||||
campaign_id=campaign_id,
|
||||
target=sub,
|
||||
target=target,
|
||||
variant_id=variant["id"],
|
||||
platform=platform,
|
||||
platform=campaign.get("platform", "reddit"),
|
||||
triggered_by=triggered_by,
|
||||
)
|
||||
post_id = post["id"]
|
||||
|
||||
# Execute
|
||||
# Build extra dict from sub_row
|
||||
extra = dict(sub_row)
|
||||
|
||||
# Execute strategy
|
||||
flair = variant.get("flair") or (rules.get("flair_to_use") if rules else None)
|
||||
try:
|
||||
client = get_client(platform)
|
||||
flair = variant.get("flair") or (rules.get("flair_to_use") if rules else None)
|
||||
url = client.post(
|
||||
sub=sub,
|
||||
title=variant["title"],
|
||||
body=variant["body"],
|
||||
result = strategy.execute(
|
||||
target=target,
|
||||
title=variant.get("title", ""),
|
||||
body=variant.get("body", ""),
|
||||
flair=flair,
|
||||
extra=extra,
|
||||
)
|
||||
return store.update_post_status(post_id, "success", url=url)
|
||||
return store.update_post_status(post_id, "success", url=result.url)
|
||||
except Exception as exc:
|
||||
logger.exception("Strategy %s failed for target %r", campaign_type, target)
|
||||
return store.update_post_status(post_id, "failed", error_msg=str(exc))
|
||||
finally:
|
||||
store.close()
|
||||
|
||||
|
||||
async def post_campaign_to_sub(campaign_id: int, sub: str,
|
||||
triggered_by: str = "scheduler") -> dict:
|
||||
async def post_campaign_to_sub(campaign_id: int, target: str,
|
||||
triggered_by: str = "scheduler") -> dict:
|
||||
"""Async wrapper for API and scheduler use."""
|
||||
db_path = get_settings().db_path
|
||||
return await asyncio.to_thread(_run_post, db_path, campaign_id, sub, triggered_by)
|
||||
return await asyncio.to_thread(_run_post, db_path, campaign_id, target, triggered_by)
|
||||
|
||||
|
||||
async def run_campaign(campaign_id: int, triggered_by: str = "scheduler") -> list[dict]:
|
||||
"""Post a campaign to all of its configured subs, sequentially."""
|
||||
"""Post a campaign to all of its configured targets, sequentially."""
|
||||
db_path = get_settings().db_path
|
||||
store = Store(db_path)
|
||||
try:
|
||||
subs = store.list_campaign_subs(campaign_id)
|
||||
active_subs = [s["sub"] for s in subs if s.get("active", 1)]
|
||||
active_targets = [s["sub"] for s in subs if s.get("active", 1)]
|
||||
finally:
|
||||
store.close()
|
||||
|
||||
results = []
|
||||
for sub in active_subs:
|
||||
result = await post_campaign_to_sub(campaign_id, sub, triggered_by)
|
||||
for target in active_targets:
|
||||
result = await post_campaign_to_sub(campaign_id, target, triggered_by)
|
||||
results.append(result)
|
||||
return results
|
||||
|
|
|
|||
115
tests/services/test_poster.py
Normal file
115
tests/services/test_poster.py
Normal file
|
|
@ -0,0 +1,115 @@
|
|||
from unittest.mock import MagicMock, patch
|
||||
from app.services.poster import _run_post
|
||||
|
||||
|
||||
def _make_store(
|
||||
*,
|
||||
already_posted=False,
|
||||
variant=None,
|
||||
rules=None,
|
||||
campaign_type="reddit_post",
|
||||
subs=None,
|
||||
):
|
||||
store = MagicMock()
|
||||
store.already_posted_this_week.return_value = already_posted
|
||||
store.resolve_variant.return_value = variant or {
|
||||
"id": 1,
|
||||
"title": "Test Title",
|
||||
"body": "Test body",
|
||||
"flair": None,
|
||||
}
|
||||
store.get_sub_rules.return_value = rules
|
||||
store.get_campaign.return_value = {
|
||||
"id": 1,
|
||||
"name": "Test",
|
||||
"type": campaign_type,
|
||||
"platform": "reddit",
|
||||
}
|
||||
store.create_post.return_value = {"id": 42}
|
||||
store.update_post_status.return_value = {
|
||||
"id": 42, "status": "success",
|
||||
"url": "https://reddit.com/r/test/comments/abc/",
|
||||
}
|
||||
store.list_campaign_subs.return_value = subs or []
|
||||
return store
|
||||
|
||||
|
||||
def test_run_post_dispatches_by_campaign_type(tmp_path):
|
||||
db = str(tmp_path / "test.db")
|
||||
mock_store = _make_store()
|
||||
|
||||
mock_result = MagicMock()
|
||||
mock_result.url = "https://reddit.com/r/test/comments/abc/"
|
||||
|
||||
mock_strategy = MagicMock()
|
||||
mock_strategy.supports_dupe_guard.return_value = True
|
||||
mock_strategy.execute.return_value = mock_result
|
||||
|
||||
with patch("app.services.poster.Store", return_value=mock_store):
|
||||
with patch("app.services.poster.get_client", return_value=mock_strategy) as mock_get_client:
|
||||
result = _run_post(db, campaign_id=1, target="selfhosted", triggered_by="manual")
|
||||
|
||||
mock_get_client.assert_called_once_with("reddit_post")
|
||||
mock_strategy.execute.assert_called_once()
|
||||
assert result["status"] == "success"
|
||||
|
||||
|
||||
def test_run_post_skips_when_dupe_guard_fires(tmp_path):
|
||||
db = str(tmp_path / "test.db")
|
||||
mock_store = _make_store(already_posted=True)
|
||||
|
||||
mock_strategy = MagicMock()
|
||||
mock_strategy.supports_dupe_guard.return_value = True
|
||||
|
||||
with patch("app.services.poster.Store", return_value=mock_store):
|
||||
with patch("app.services.poster.get_client", return_value=mock_strategy):
|
||||
result = _run_post(db, campaign_id=1, target="selfhosted", triggered_by="manual")
|
||||
|
||||
assert result["skipped"] is True
|
||||
mock_strategy.execute.assert_not_called()
|
||||
|
||||
|
||||
def test_run_post_skips_dupe_guard_when_strategy_opts_out(tmp_path):
|
||||
db = str(tmp_path / "test.db")
|
||||
mock_store = _make_store(already_posted=True, campaign_type="blog_post")
|
||||
|
||||
mock_result = MagicMock()
|
||||
mock_result.url = "https://circuitforge.tech/blog/test-post"
|
||||
|
||||
mock_strategy = MagicMock()
|
||||
mock_strategy.supports_dupe_guard.return_value = False
|
||||
mock_strategy.execute.return_value = mock_result
|
||||
|
||||
with patch("app.services.poster.Store", return_value=mock_store):
|
||||
with patch("app.services.poster.get_client", return_value=mock_strategy):
|
||||
result = _run_post(db, campaign_id=1, target="blog", triggered_by="scheduler")
|
||||
|
||||
mock_strategy.execute.assert_called_once()
|
||||
assert result["status"] == "success"
|
||||
|
||||
|
||||
def test_run_post_skips_banned_sub(tmp_path):
|
||||
db = str(tmp_path / "test.db")
|
||||
mock_store = _make_store(rules={"promo_allowed": 0})
|
||||
|
||||
mock_strategy = MagicMock()
|
||||
mock_strategy.supports_dupe_guard.return_value = True
|
||||
|
||||
with patch("app.services.poster.Store", return_value=mock_store):
|
||||
with patch("app.services.poster.get_client", return_value=mock_strategy):
|
||||
result = _run_post(db, campaign_id=1, target="ADHD", triggered_by="scheduler")
|
||||
|
||||
assert result["skipped"] is True
|
||||
mock_strategy.execute.assert_not_called()
|
||||
|
||||
|
||||
def test_run_post_unknown_type_skips(tmp_path):
|
||||
db = str(tmp_path / "test.db")
|
||||
mock_store = _make_store(campaign_type="future_platform")
|
||||
|
||||
with patch("app.services.poster.Store", return_value=mock_store):
|
||||
with patch("app.services.poster.get_client", side_effect=ValueError("Unknown campaign type")):
|
||||
result = _run_post(db, campaign_id=1, target="some_target", triggered_by="scheduler")
|
||||
|
||||
assert result["skipped"] is True
|
||||
assert "Unknown campaign type" in result["reason"]
|
||||
Loading…
Reference in a new issue