"""publisher — themed content calendar with a hard human gate on publishing. State machine: draft -> approved -> published. Only ``approved`` items ever publish. AUTONOMOUS_MODE has NO effect on this component: content always requires explicit human approval (documented in AUDIT.md). Channels: nostr (NIP-01, primary) — facebook/instagram adapters exist but are disabled unless explicitly enabled via env (DECISIONS: default deployment never touches Meta). """ from __future__ import annotations from datetime import UTC, datetime, timedelta from cayceops.components.publisher import images, nostr from cayceops.config import Config from cayceops.db import session_scope from cayceops.models import ContentItem, ContentStatus, DaemonRun, audit # rotation: one theme per slot; channel rotation nostr -> facebook -> instagram THEMES = [ ("lightning-abroad", "bitcoinhotels.org/guides/lightning-abroad"), ("seed-strategy", "hardcap.news/learn/travel-security-kit"), ("atm-fees", "bitcoinhotels.org/guides/best-bitcoin-atms-by-city"), ("elsalvador-budget", "bitcoinhotels.org/guides/living-on-sats-in-el-salvador"), ] _CAPTION_TEMPLATE = "{hook}\n\n{body}\n\n{link}" def _caption(theme: str, link: str) -> str: hooks = { "lightning-abroad": "Paying abroad with Lightning beats your bank.", "seed-strategy": "The seed strategy you pick decides how the border story goes.", "atm-fees": "The first ATM you find charges 10-15% over spot.", "elsalvador-budget": "A real weekly budget on a Bitcoin standard.", } bodies = { "lightning-abroad": ( "No foreign transaction fees, no dynamic currency conversion, " "no declines. Field-tested guide." ), "seed-strategy": ( "Steel plate, memorized + decoy, or split custody — " "the full breakdown, brutally honest." ), "atm-fees": ( "The prepared traveler pays 4-6%. The difference is knowing " "the operator and the no-KYC cap." ), "elsalvador-budget": ( "Surf hostels, comidas, coworking — the corners where you still " "need dollars, and the 90-day protocol." ), } return _CAPTION_TEMPLATE.format( hook=hooks.get(theme, theme.replace("-", " ").title()), body=bodies.get(theme, "Field-tested notes."), link=link, ) # --- calendar planning ---------------------------------------------------------- def _plan(cfg: Config, sess, days: int = 4) -> dict: """Fill the calendar: one draft per theme/channel pair not already queued. Idempotent by (theme, channel): existing draft/approved items block a duplicate. Drafts are scheduled one per day, starting tomorrow. """ created = 0 for idx, (theme, link) in enumerate(THEMES[:days]): channel = ["nostr", "facebook", "instagram", "nostr"][idx % 4] exists = ( sess.query(ContentItem) .filter( ContentItem.theme == theme, ContentItem.channel == channel, ContentItem.status != ContentStatus.PUBLISHED, ) .one_or_none() ) if exists: continue item = ContentItem( theme=theme, channel=channel, caption=_caption(theme, link), link=link, status=ContentStatus.DRAFT, scheduled_for=datetime.now(UTC).replace(hour=13, minute=0, second=0, microsecond=0) + timedelta(days=idx + 1), ) sess.add(item) sess.flush() audit( sess, actor="daemon:publisher", action="plan_draft", detail={"item": item.id, "theme": theme, "channel": channel}, ) created += 1 return {"created": created} # --- human gate ------------------------------------------------------------------- def _approve(item_id: int, actor: str, sess) -> int: item = sess.get(ContentItem, item_id) if item is None or item.status != ContentStatus.DRAFT: audit( sess, actor=actor, action="approve_content", result="error", detail={"reason": f"item {item_id} not in draft state"}, ) return 1 item.status = ContentStatus.APPROVED audit( sess, actor=actor, action="approve_content", detail={"item": item.id, "channel": item.channel}, ) return 0 # --- image rendering ---------------------------------------------------------------- def _render_image(item: ContentItem, cfg: Config) -> str | None: """Render the brand image for a nostr item (per its theme).""" brand = "hc" if "hardcap" in (item.link or "") else "bh" title = item.theme.replace("-", " ").upper().split() lines = [" ".join(title[:2])] if title else [item.theme] out = images.render_brand_image( brand=brand, kicker="C A Y C E", lines=lines, tagline=[(item.link or "")[:70]], url=(item.link or "")[:60], out_path=f"{cfg.media_dir}/item-{item.id}.jpg", ) item.media_path = out return out # --- publishing ---------------------------------------------------------------------- def _publish_nostr(item: ContentItem, cfg: Config, sess, sender=None) -> str: if not cfg.nostr_private_key or not cfg.nostr_relays: raise RuntimeError( "nostr channel requires CAYCE_OPS_NOSTR_PRIVATE_KEY and CAYCE_OPS_NOSTR_RELAYS" ) secret = nostr.normalize_key(cfg.nostr_private_key) event = nostr.build_event(secret, item.caption, tags=[["t", item.theme]]) if not nostr.verify_event(event): # pragma: no cover — build/verify symmetry raise RuntimeError("event failed self-verification before publish") results = (sender or nostr.publish)(cfg.nostr_relays, event) audit( sess, actor="daemon:publisher", action="nostr_relay_results", detail={"event": event["id"], "relays": results}, ) return event["id"] def _publish_due(cfg: Config, sess, nostr_sender=None) -> dict: """Publish approved items that are due. Non-nostr channels are skipped unless their adapters are explicitly enabled (default: skipped, audited).""" stats = {"published": 0, "skipped": 0, "failed": 0} due = ( sess.query(ContentItem) .filter(ContentItem.status == ContentStatus.APPROVED) .all() ) now = datetime.now(UTC) for item in due: if item.scheduled_for and item.scheduled_for > now: continue # not due yet try: if item.channel == "nostr": item.post_ref = _publish_nostr(item, cfg, sess, sender=nostr_sender) else: audit( sess, actor="daemon:publisher", action="publish_skipped", detail={ "item": item.id, "channel": item.channel, "reason": f"{item.channel} adapter disabled by default", }, ) stats["skipped"] += 1 continue except Exception as exc: # noqa: BLE001 — publish boundary: audit and continue item.error = str(exc)[:300] audit( sess, actor="daemon:publisher", action="publish_failed", result="error", detail={"item": item.id, "error": item.error}, ) stats["failed"] += 1 continue item.status = ContentStatus.PUBLISHED item.published_at = datetime.now(UTC) audit( sess, actor="daemon:publisher", action="publish_content", detail={"item": item.id, "channel": item.channel, "ref": item.post_ref}, ) stats["published"] += 1 return stats # --- CLI wrappers --------------------------------------------------------------------- def plan(cfg: Config | None = None) -> int: from cayceops.config import load from cayceops.db import engine cfg = cfg or load() try: with session_scope() as sess: run_row = DaemonRun(daemon="publisher") sess.add(run_row) sess.flush() stats = _plan(cfg, sess) run_row.status = "ok" run_row.detail = stats run_row.finished_at = datetime.now(UTC) return 0 finally: engine().dispose() def publish_due(cfg: Config | None = None) -> int: from cayceops.config import load from cayceops.db import engine cfg = cfg or load() try: with session_scope() as sess: run_row = DaemonRun(daemon="publisher") sess.add(run_row) sess.flush() stats = _publish_due(cfg, sess) run_row.status = "ok" run_row.detail = stats run_row.finished_at = datetime.now(UTC) return 0 finally: engine().dispose() def make_image(item_id: int, cfg: Config | None = None) -> int: from cayceops.config import load from cayceops.db import engine cfg = cfg or load() try: with session_scope() as sess: item = sess.get(ContentItem, item_id) if item is None: return 1 print(_render_image(item, cfg)) return 0 finally: engine().dispose() def approve(item_id: int, actor: str) -> int: from cayceops.db import engine try: with session_scope() as sess: return _approve(item_id, actor, sess) finally: engine().dispose()