cayce-ops/cayceops/components/publisher/publisher.py

"""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()