Compare commits
No commits in common. "7af226c3674518a368936b3e850c5a307b3a5f96" and "01e5990f584475363ae76a1984eab4d836e19d48" have entirely different histories.
7af226c367
...
01e5990f58
8 changed files with 8 additions and 238 deletions
|
|
@ -1,12 +0,0 @@
|
||||||
-- 017_variant_blog_fields.sql
|
|
||||||
|
|
||||||
-- slug: Directus URL slug. Generated from title via slugify() if NULL.
|
|
||||||
ALTER TABLE campaign_variants ADD COLUMN slug TEXT;
|
|
||||||
|
|
||||||
-- tags: JSON array of tag strings. e.g. '["self-hosting","peregrine"]'
|
|
||||||
-- NULL = no tags.
|
|
||||||
ALTER TABLE campaign_variants ADD COLUMN tags TEXT;
|
|
||||||
|
|
||||||
-- seo_description: Short meta description for search engines (max ~160 chars).
|
|
||||||
-- NULL = Directus default.
|
|
||||||
ALTER TABLE campaign_variants ADD COLUMN seo_description TEXT;
|
|
||||||
|
|
@ -180,9 +180,6 @@ class Store:
|
||||||
title: str,
|
title: str,
|
||||||
body: str,
|
body: str,
|
||||||
flair: str | None = None,
|
flair: str | None = None,
|
||||||
slug: str | None = None,
|
|
||||||
tags: str | None = None,
|
|
||||||
seo_description: str | None = None,
|
|
||||||
) -> dict:
|
) -> dict:
|
||||||
existing = self._fetchone(
|
existing = self._fetchone(
|
||||||
"SELECT * FROM campaign_variants WHERE campaign_id = ? AND sub_pattern = ?",
|
"SELECT * FROM campaign_variants WHERE campaign_id = ? AND sub_pattern = ?",
|
||||||
|
|
@ -190,15 +187,15 @@ class Store:
|
||||||
)
|
)
|
||||||
if existing:
|
if existing:
|
||||||
self.conn.execute(
|
self.conn.execute(
|
||||||
"UPDATE campaign_variants SET title=?, body=?, flair=?, slug=?, tags=?, seo_description=? WHERE id=?",
|
"UPDATE campaign_variants SET title=?, body=?, flair=? WHERE id=?",
|
||||||
(title, body, flair, slug, tags, seo_description, existing["id"]),
|
(title, body, flair, existing["id"]),
|
||||||
)
|
)
|
||||||
self.conn.commit()
|
self.conn.commit()
|
||||||
return self._fetchone("SELECT * FROM campaign_variants WHERE id=?", (existing["id"],))
|
return self._fetchone("SELECT * FROM campaign_variants WHERE id=?", (existing["id"],))
|
||||||
return self._insert_returning(
|
return self._insert_returning(
|
||||||
"INSERT INTO campaign_variants (campaign_id, sub_pattern, title, body, flair, slug, tags, seo_description)"
|
"INSERT INTO campaign_variants (campaign_id, sub_pattern, title, body, flair)"
|
||||||
" VALUES (?,?,?,?,?,?,?,?) RETURNING *",
|
" VALUES (?,?,?,?,?) RETURNING *",
|
||||||
(campaign_id, sub_pattern, title, body, flair, slug, tags, seo_description),
|
(campaign_id, sub_pattern, title, body, flair),
|
||||||
)
|
)
|
||||||
|
|
||||||
def update_variant(self, variant_id: int, **fields) -> dict | None:
|
def update_variant(self, variant_id: int, **fields) -> dict | None:
|
||||||
|
|
|
||||||
|
|
@ -3,14 +3,13 @@ from __future__ import annotations
|
||||||
from app.services.platforms.base import PostingStrategy, PostResult
|
from app.services.platforms.base import PostingStrategy, PostResult
|
||||||
from app.services.platforms.reddit_post import RedditPostStrategy
|
from app.services.platforms.reddit_post import RedditPostStrategy
|
||||||
from app.services.platforms.reddit_comment import RedditCommentStrategy
|
from app.services.platforms.reddit_comment import RedditCommentStrategy
|
||||||
from app.services.platforms.blog_post import BlogPostStrategy
|
|
||||||
|
|
||||||
_REGISTRY: dict[str, PostingStrategy] = {
|
_REGISTRY: dict[str, PostingStrategy] = {
|
||||||
s.campaign_type: s()
|
s.campaign_type: s()
|
||||||
for s in [
|
for s in [
|
||||||
RedditPostStrategy,
|
RedditPostStrategy,
|
||||||
RedditCommentStrategy,
|
RedditCommentStrategy,
|
||||||
BlogPostStrategy,
|
# BlogPostStrategy — added in Plan C
|
||||||
]
|
]
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,70 +0,0 @@
|
||||||
from __future__ import annotations
|
|
||||||
|
|
||||||
import json
|
|
||||||
import logging
|
|
||||||
|
|
||||||
from app.services.directus import publish_blog_post
|
|
||||||
from app.services.platforms.base import PostingStrategy, PostResult
|
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
|
||||||
|
|
||||||
_BLOG_BASE_URL = "https://circuitforge.tech/blog"
|
|
||||||
|
|
||||||
|
|
||||||
class BlogPostStrategy(PostingStrategy):
|
|
||||||
"""Publish a blog post to the CircuitForge website via Directus.
|
|
||||||
|
|
||||||
target: arbitrary label string (e.g. "blog/main") — used for logging only.
|
|
||||||
title: post title
|
|
||||||
body: post body (Markdown)
|
|
||||||
extra: optional dict with keys:
|
|
||||||
slug (str) — Directus URL slug; generated from title if absent
|
|
||||||
tags (list|str) — tag list or JSON-encoded list string
|
|
||||||
seo_description (str) — meta description
|
|
||||||
"""
|
|
||||||
|
|
||||||
campaign_type = "blog_post"
|
|
||||||
|
|
||||||
def execute(
|
|
||||||
self,
|
|
||||||
*,
|
|
||||||
target: str,
|
|
||||||
title: str,
|
|
||||||
body: str,
|
|
||||||
flair: str | None = None,
|
|
||||||
extra: dict | None = None,
|
|
||||||
) -> PostResult:
|
|
||||||
extra = extra or {}
|
|
||||||
|
|
||||||
slug = extra.get("slug") or None
|
|
||||||
seo_description = extra.get("seo_description") or None
|
|
||||||
|
|
||||||
# tags may be stored as a JSON string in the DB; decode it
|
|
||||||
raw_tags = extra.get("tags")
|
|
||||||
tags: list[str] | None = None
|
|
||||||
if isinstance(raw_tags, str):
|
|
||||||
try:
|
|
||||||
tags = json.loads(raw_tags)
|
|
||||||
except json.JSONDecodeError:
|
|
||||||
logger.warning("Could not parse tags JSON: %r", raw_tags)
|
|
||||||
elif isinstance(raw_tags, list):
|
|
||||||
tags = raw_tags
|
|
||||||
|
|
||||||
item = publish_blog_post(
|
|
||||||
title=title,
|
|
||||||
body=body,
|
|
||||||
slug=slug,
|
|
||||||
tags=tags,
|
|
||||||
seo_description=seo_description,
|
|
||||||
)
|
|
||||||
|
|
||||||
published_slug = item.get("slug", slug or "")
|
|
||||||
url = f"{_BLOG_BASE_URL}/{published_slug}"
|
|
||||||
|
|
||||||
return PostResult(
|
|
||||||
url=url,
|
|
||||||
metadata={"directus_id": item.get("id"), "slug": published_slug},
|
|
||||||
)
|
|
||||||
|
|
||||||
def supports_dupe_guard(self) -> bool:
|
|
||||||
return False
|
|
||||||
|
|
@ -77,11 +77,8 @@ def _run_post(db_path: str, campaign_id: int, target: str,
|
||||||
)
|
)
|
||||||
post_id = post["id"]
|
post_id = post["id"]
|
||||||
|
|
||||||
# Build extra dict from sub_row; merge variant-level blog fields (blog_post strategy uses them)
|
# Build extra dict from sub_row
|
||||||
extra = dict(sub_row)
|
extra = dict(sub_row)
|
||||||
for field in ("slug", "tags", "seo_description"):
|
|
||||||
if variant.get(field) is not None:
|
|
||||||
extra.setdefault(field, variant[field])
|
|
||||||
|
|
||||||
# Execute strategy
|
# Execute strategy
|
||||||
flair = variant.get("flair") or (rules.get("flair_to_use") if rules else None)
|
flair = variant.get("flair") or (rules.get("flair_to_use") if rules else None)
|
||||||
|
|
|
||||||
|
|
@ -299,7 +299,7 @@ What part of the process drains you most? I'm building specifically for this com
|
||||||
|
|
||||||
SUB_RULES = [
|
SUB_RULES = [
|
||||||
{"sub": "selfhosted", "promo_allowed": True, "flair_required": False, "rule_warning": False, "notes": "Promo OK with context. Engages well with self-hosted + local inference framing."},
|
{"sub": "selfhosted", "promo_allowed": True, "flair_required": False, "rule_warning": False, "notes": "Promo OK with context. Engages well with self-hosted + local inference framing."},
|
||||||
{"sub": "solarpunk", "promo_allowed": False, "flair_required": True, "flair_to_use": "Action / DIY / Activism", "rule_warning": False, "notes": "Post removed by mods 2026-04-23 (est.) — suspected project age issue. No explanation given. Re-evaluate after project has more post history."},
|
{"sub": "solarpunk", "promo_allowed": True, "flair_required": True, "flair_to_use": "Action / DIY / Activism", "rule_warning": False, "notes": "Flair required. Use coordinate-click for Add button (shadow DOM)."},
|
||||||
{"sub": "opensource", "promo_allowed": True, "flair_required": True, "flair_to_use": "Promotional", "rule_warning": True, "notes": "Rule-warning dialog appears after Post click. wait_for(visible) + Submit without editing."},
|
{"sub": "opensource", "promo_allowed": True, "flair_required": True, "flair_to_use": "Promotional", "rule_warning": True, "notes": "Rule-warning dialog appears after Post click. wait_for(visible) + Submit without editing."},
|
||||||
{"sub": "AuDHD", "promo_allowed": True, "flair_required": False, "rule_warning": False, "notes": "No hard promo ban. Personally-relevant content qualifies."},
|
{"sub": "AuDHD", "promo_allowed": True, "flair_required": False, "rule_warning": False, "notes": "No hard promo ban. Personally-relevant content qualifies."},
|
||||||
{"sub": "privacytoolsIO", "promo_allowed": True, "flair_required": False, "rule_warning": False, "notes": "Self-promo OK with privacy context. No flair needed."},
|
{"sub": "privacytoolsIO", "promo_allowed": True, "flair_required": False, "rule_warning": False, "notes": "Self-promo OK with privacy context. No flair needed."},
|
||||||
|
|
@ -409,44 +409,6 @@ def seed(store: Store) -> None:
|
||||||
)
|
)
|
||||||
print(" variant: '*'")
|
print(" variant: '*'")
|
||||||
|
|
||||||
print()
|
|
||||||
print("Seeding blog campaigns...")
|
|
||||||
|
|
||||||
import json as _json
|
|
||||||
|
|
||||||
blog_campaign = store.get_or_create_campaign(
|
|
||||||
name="CircuitForge | Weekly product update — blog",
|
|
||||||
product="circuitforge",
|
|
||||||
platform="blog",
|
|
||||||
type="blog_post",
|
|
||||||
cron_schedule=None, # manual trigger only to start
|
|
||||||
)
|
|
||||||
print(f" campaign: {blog_campaign['name']!r} (id={blog_campaign['id']})")
|
|
||||||
store.upsert_campaign_sub(
|
|
||||||
campaign_id=blog_campaign["id"],
|
|
||||||
sub="blog/main",
|
|
||||||
sort_order=0,
|
|
||||||
)
|
|
||||||
print(" sub: 'blog/main'")
|
|
||||||
store.upsert_variant(
|
|
||||||
campaign_id=blog_campaign["id"],
|
|
||||||
sub_pattern="*",
|
|
||||||
title="[DRAFT] Weekly CircuitForge Update",
|
|
||||||
body=(
|
|
||||||
"# What we shipped this week\n\n"
|
|
||||||
"_Replace this with actual update content._\n\n"
|
|
||||||
"## Products\n\n"
|
|
||||||
"- **Peregrine:** ...\n"
|
|
||||||
"- **Kiwi:** ...\n"
|
|
||||||
"- **Snipe:** ...\n\n"
|
|
||||||
"## Coming up\n\n"
|
|
||||||
"..."
|
|
||||||
),
|
|
||||||
tags=_json.dumps(["updates", "circuitforge"]),
|
|
||||||
seo_description="Weekly product update from CircuitForge.",
|
|
||||||
)
|
|
||||||
print(" variant: '*'")
|
|
||||||
|
|
||||||
print()
|
print()
|
||||||
print("Seeding sub rules...")
|
print("Seeding sub rules...")
|
||||||
for rule in SUB_RULES:
|
for rule in SUB_RULES:
|
||||||
|
|
|
||||||
|
|
@ -1,93 +0,0 @@
|
||||||
import json
|
|
||||||
from unittest.mock import patch
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
|
|
||||||
from app.services.platforms.blog_post import BlogPostStrategy
|
|
||||||
from app.services.platforms.base import PostResult
|
|
||||||
|
|
||||||
|
|
||||||
def _make_directus_response(slug: str = "test-post", post_id: int = 42) -> dict:
|
|
||||||
return {"id": post_id, "slug": slug, "title": "Test Post", "status": "published"}
|
|
||||||
|
|
||||||
|
|
||||||
def test_campaign_type():
|
|
||||||
assert BlogPostStrategy.campaign_type == "blog_post"
|
|
||||||
|
|
||||||
|
|
||||||
def test_does_not_support_dupe_guard():
|
|
||||||
assert BlogPostStrategy().supports_dupe_guard() is False
|
|
||||||
|
|
||||||
|
|
||||||
def test_execute_publishes_immediately():
|
|
||||||
mock_response = _make_directus_response(slug="my-post")
|
|
||||||
|
|
||||||
with patch("app.services.platforms.blog_post.publish_blog_post", return_value=mock_response) as mock_pub:
|
|
||||||
strategy = BlogPostStrategy()
|
|
||||||
result = strategy.execute(
|
|
||||||
target="blog/main",
|
|
||||||
title="My Test Post",
|
|
||||||
body="# Hello\n\nThis is the post body.",
|
|
||||||
)
|
|
||||||
|
|
||||||
mock_pub.assert_called_once_with(
|
|
||||||
title="My Test Post",
|
|
||||||
body="# Hello\n\nThis is the post body.",
|
|
||||||
slug=None,
|
|
||||||
tags=None,
|
|
||||||
seo_description=None,
|
|
||||||
)
|
|
||||||
assert isinstance(result, PostResult)
|
|
||||||
assert result.url == "https://circuitforge.tech/blog/my-post"
|
|
||||||
assert result.metadata["directus_id"] == 42
|
|
||||||
|
|
||||||
|
|
||||||
def test_execute_passes_extra_fields():
|
|
||||||
mock_response = _make_directus_response(slug="custom-slug")
|
|
||||||
|
|
||||||
with patch("app.services.platforms.blog_post.publish_blog_post", return_value=mock_response) as mock_pub:
|
|
||||||
strategy = BlogPostStrategy()
|
|
||||||
result = strategy.execute(
|
|
||||||
target="blog/main",
|
|
||||||
title="Custom Slug Post",
|
|
||||||
body="Body content here.",
|
|
||||||
extra={
|
|
||||||
"slug": "custom-slug",
|
|
||||||
"tags": ["self-hosting", "peregrine"],
|
|
||||||
"seo_description": "A short description for SEO.",
|
|
||||||
},
|
|
||||||
)
|
|
||||||
|
|
||||||
mock_pub.assert_called_once_with(
|
|
||||||
title="Custom Slug Post",
|
|
||||||
body="Body content here.",
|
|
||||||
slug="custom-slug",
|
|
||||||
tags=["self-hosting", "peregrine"],
|
|
||||||
seo_description="A short description for SEO.",
|
|
||||||
)
|
|
||||||
assert result.url == "https://circuitforge.tech/blog/custom-slug"
|
|
||||||
|
|
||||||
|
|
||||||
def test_execute_parses_tags_from_json_string():
|
|
||||||
"""Tags stored as JSON string in campaign_variants.tags should be decoded."""
|
|
||||||
mock_response = _make_directus_response(slug="json-tags-post")
|
|
||||||
|
|
||||||
with patch("app.services.platforms.blog_post.publish_blog_post", return_value=mock_response) as mock_pub:
|
|
||||||
strategy = BlogPostStrategy()
|
|
||||||
strategy.execute(
|
|
||||||
target="blog/main",
|
|
||||||
title="JSON Tags Post",
|
|
||||||
body="Body.",
|
|
||||||
extra={"tags": '["self-hosting","kiwi"]'}, # stored as JSON string
|
|
||||||
)
|
|
||||||
|
|
||||||
call_args = mock_pub.call_args
|
|
||||||
assert call_args.kwargs["tags"] == ["self-hosting", "kiwi"]
|
|
||||||
|
|
||||||
|
|
||||||
def test_execute_raises_on_directus_error():
|
|
||||||
with patch("app.services.platforms.blog_post.publish_blog_post",
|
|
||||||
side_effect=RuntimeError("Directus 502")):
|
|
||||||
strategy = BlogPostStrategy()
|
|
||||||
with pytest.raises(RuntimeError, match="Directus 502"):
|
|
||||||
strategy.execute(target="blog/main", title="T", body="B")
|
|
||||||
|
|
@ -1,7 +1,6 @@
|
||||||
import pytest
|
import pytest
|
||||||
from app.services.platforms import get_client, SUPPORTED_PLATFORMS
|
from app.services.platforms import get_client, SUPPORTED_PLATFORMS
|
||||||
from app.services.platforms.reddit_post import RedditPostStrategy
|
from app.services.platforms.reddit_post import RedditPostStrategy
|
||||||
from app.services.platforms.blog_post import BlogPostStrategy
|
|
||||||
|
|
||||||
|
|
||||||
def test_get_client_returns_reddit_post_strategy():
|
def test_get_client_returns_reddit_post_strategy():
|
||||||
|
|
@ -16,12 +15,3 @@ def test_get_client_unknown_type_raises():
|
||||||
|
|
||||||
def test_supported_platforms_contains_reddit_post():
|
def test_supported_platforms_contains_reddit_post():
|
||||||
assert "reddit_post" in SUPPORTED_PLATFORMS
|
assert "reddit_post" in SUPPORTED_PLATFORMS
|
||||||
|
|
||||||
|
|
||||||
def test_get_client_returns_blog_post_strategy():
|
|
||||||
client = get_client("blog_post")
|
|
||||||
assert isinstance(client, BlogPostStrategy)
|
|
||||||
|
|
||||||
|
|
||||||
def test_supported_platforms_contains_blog_post():
|
|
||||||
assert "blog_post" in SUPPORTED_PLATFORMS
|
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue