From aabd6bafc7bc91883c1d6baedb19b7898bbf2d98 Mon Sep 17 00:00:00 2001 From: Cayce Date: Mon, 19 Oct 2026 21:54:39 +0000 Subject: [PATCH] publisher: themed calendar, NIP-01 nostr adapter, gated brand images MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - calendar: one draft per theme/channel (idempotent), rotated schedule - human gate: approve() audited; AUTONOMOUS_MODE has no effect here by design - nostr (primary): in-repo NIP-01 client (D4) — canonical serialization, sha256 id, BIP-340 schnorr sign via coincurve with zero aux randomness (deterministic), pure-python reference verification before publish, per-relay publish with isolated errors, event id stored as post_ref - facebook/instagram adapters: exist but refuse to run unless explicitly enabled via env (default deployment never touches Meta) - images: Pillow brand templates (bh gold / hc orange), 1080x1350, template-driven typeset design (not AI-generated media) - failure isolation: config errors audited, item stays approved tests: 12 new (event roundtrip+tamper, determinism, key validation, calendar idempotence, approve gate, nostr happy path, future schedule, clean config failure, disabled meta adapters, image render) — 41 total --- cayceops/components/publisher/__init__.py | 1 + cayceops/components/publisher/adapters.py | 61 +++++ cayceops/components/publisher/images.py | 72 ++++++ cayceops/components/publisher/nostr.py | 166 +++++++++++++ cayceops/components/publisher/publisher.py | 266 +++++++++++++++++++++ tests/test_publisher.py | 162 +++++++++++++ 6 files changed, 728 insertions(+) create mode 100644 cayceops/components/publisher/__init__.py create mode 100644 cayceops/components/publisher/adapters.py create mode 100644 cayceops/components/publisher/images.py create mode 100644 cayceops/components/publisher/nostr.py create mode 100644 cayceops/components/publisher/publisher.py create mode 100644 tests/test_publisher.py diff --git a/cayceops/components/publisher/__init__.py b/cayceops/components/publisher/__init__.py new file mode 100644 index 0000000..e65aea3 --- /dev/null +++ b/cayceops/components/publisher/__init__.py @@ -0,0 +1 @@ +"""publisher package.""" diff --git a/cayceops/components/publisher/adapters.py b/cayceops/components/publisher/adapters.py new file mode 100644 index 0000000..e56cd6e --- /dev/null +++ b/cayceops/components/publisher/adapters.py @@ -0,0 +1,61 @@ +"""Optional Facebook/Instagram adapters (Graph API). + +DISABLED BY DEFAULT. Each adapter refuses to run unless its enable flag is +true (``CAYCE_OPS_ENABLE_FACEBOOK`` / ``CAYCE_OPS_ENABLE_INSTAGRAM``), so the +default deployment never touches Meta endpoints. Tokens are env-only and are +never logged. +""" + +from __future__ import annotations + +import json +import urllib.parse +import urllib.request + +GRAPH = "https://graph.facebook.com/v21.0" + + +def _post(url: str, payload: dict) -> dict: + data = urllib.parse.urlencode(payload).encode() + req = urllib.request.Request(url, data=data, method="POST") # noqa: S310 + with urllib.request.urlopen(req, timeout=30) as resp: + return json.loads(resp.read().decode()) + + +def post_facebook(cfg, caption: str, media_path: str | None = None) -> dict: + """Post to the page feed (text) or /photos (with local media). Gated.""" + if not cfg.enable_facebook: + raise PermissionError("facebook adapter disabled (CAYCE_OPS_ENABLE_FACEBOOK=false)") + if not cfg.facebook_page_token: + raise ValueError("CAYCE_OPS_FACEBOOK_PAGE_TOKEN missing") + if media_path: + # photo posts require a reachable URL or multipart upload; we use the + # simple URL form — media must be hosted (out of scope for local files) + result = _post( + f"{GRAPH}/me/photos", + {"url": media_path, "caption": caption, "access_token": cfg.facebook_page_token}, + ) + else: + result = _post( + f"{GRAPH}/me/feed", + {"message": caption, "access_token": cfg.facebook_page_token}, + ) + return {"post_id": result.get("post_id") or result.get("id", "")} + + +def post_instagram(cfg, caption: str, image_url: str) -> dict: + """Two-step IG publish (container -> publish). Gated.""" + if not cfg.enable_instagram: + raise PermissionError("instagram adapter disabled (CAYCE_OPS_ENABLE_INSTAGRAM=false)") + if not cfg.instagram_user_id: + raise ValueError("CAYCE_OPS_INSTAGRAM_USER_ID missing") + token = cfg.facebook_page_token # IG Graph publishing uses the page token + container = _post( + f"{GRAPH}/{cfg.instagram_user_id}/media", + {"image_url": image_url, "caption": caption, "access_token": token}, + ) + published = _post( + f"{GRAPH}/{cfg.instagram_user_id}/media_publish", + {"creation_id": container["id"], "access_token": token}, + ) + return {"post_id": published.get("id", "")} diff --git a/cayceops/components/publisher/images.py b/cayceops/components/publisher/images.py new file mode 100644 index 0000000..dc2f752 --- /dev/null +++ b/cayceops/components/publisher/images.py @@ -0,0 +1,72 @@ +"""Brand image templates (Pillow). 1080x1350 portrait, two brand styles. + +Deterministic, template-driven graphics — deliberately not AI-generated media +(DECISIONS: nothing here trips AI-media disclosure requirements; it is typeset +design). Fonts: Pillow's bundled default (scalable), so no system font deps. +""" + +from __future__ import annotations + +import os + +from PIL import Image, ImageDraw, ImageFont + +WIDTH, HEIGHT = 1080, 1350 + +_BRANDS = { + # bitcoinhotels — warm gold on near-black + "bh": {"bg": (20, 18, 16), "accent": (201, 168, 76), "text": (242, 236, 224), "muted": (150, 140, 120)}, + # hardcap — bitcoin orange on cool black + "hc": {"bg": (13, 13, 16), "accent": (247, 147, 26), "text": (250, 250, 250), "muted": (120, 120, 130)}, +} + + +def _font(size: int, bold: bool = False): + # Pillow >=10: load_default(size=...) returns a scalable bundled font + return ImageFont.load_default(size=size) + + +def _vgrad(top: tuple, bottom: tuple) -> Image.Image: + """Simple two-color vertical gradient.""" + img = Image.new("RGB", (WIDTH, HEIGHT)) + for y in range(HEIGHT): + t = y / (HEIGHT - 1) + row = tuple(int(top[i] + (bottom[i] - top[i]) * t) for i in range(3)) + ImageDraw.Draw(img).line([(0, y), (WIDTH, y)], fill=row) + return img + + +def render_brand_image( + *, + brand: str = "bh", + kicker: str | None, + lines: list[str], + tagline: list[str] | None = None, + url: str | None = None, + out_path: str, +) -> str: + """Render one branded poster. Returns out_path. Raises on unknown brand.""" + if brand not in _BRANDS: + raise ValueError(f"unknown brand {brand!r} (have: {sorted(_BRANDS)})") + palette = _BRANDS[brand] + img = _vgrad(palette["bg"], tuple(max(0, c - 8) for c in palette["bg"])) + draw = ImageDraw.Draw(img) + + x, y = 90, 240 + if kicker: + draw.text((x, y), kicker, font=_font(40), fill=palette["muted"]) + y += 110 + for line in lines: + draw.text((x, y), line, font=_font(96, bold=True), fill=palette["text"]) + y += 118 + y += 24 + for t in tagline or []: + draw.text((x, y), t, font=_font(38), fill=palette["muted"]) + y += 58 + draw.rectangle([x, y + 30, x + 470, y + 33], fill=palette["accent"]) + if url: + draw.text((x, y + 110), url, font=_font(44), fill=palette["accent"]) + + os.makedirs(os.path.dirname(out_path) or ".", exist_ok=True) + img.save(out_path, "JPEG", quality=92) + return out_path diff --git a/cayceops/components/publisher/nostr.py b/cayceops/components/publisher/nostr.py new file mode 100644 index 0000000..5302ca9 --- /dev/null +++ b/cayceops/components/publisher/nostr.py @@ -0,0 +1,166 @@ +"""Minimal Nostr client: NIP-01 kind-1 notes (build, sign, verify, publish). + +Implemented in-repo (DECISIONS D4) so every byte of the signing path is +auditable: event id = sha256 of the canonical NIP-01 serialization; signature = +BIP-340 Schnorr over the id. Keys are 32-byte hex. Relays are plain env config. + +Publishing is fire-and-wait: open a websocket per relay, send the EVENT, wait +for the OK frame (or timeout), and return the per-relay results. +""" + +from __future__ import annotations + +import hashlib +import json +from typing import Any + +from coincurve import PrivateKey, PublicKey + +KIND_NOTE = 1 + + +def normalize_key(private_key_hex: str) -> bytes: + """Accept 64-char hex (with or without 0x). Anything else is a config error.""" + key = private_key_hex.strip().lower().removeprefix("0x") + if len(key) != 64 or not all(c in "0123456789abcdef" for c in key): + raise ValueError("CAYCE_OPS_NOSTR_PRIVATE_KEY must be 64-char hex (32 bytes)") + return bytes.fromhex(key) + + +def pubkey_from_secret(secret: bytes) -> str: + return PublicKey.from_secret(secret).format().hex()[2:] + + +def build_event( + secret: bytes, + content: str, + tags: list[list[str]] | None = None, + created_at: int | None = None, + kind: int = KIND_NOTE, +) -> dict[str, Any]: + """Build and sign a NIP-01 event. Deterministic given the same inputs.""" + created_at = int(created_at if created_at is not None else __import__("time").time()) + pubkey = pubkey_from_secret(secret) + serialized = json.dumps( + [0, pubkey, created_at, kind, tags or [], content], + separators=(",", ":"), # canonical NIP-01 form + ensure_ascii=False, + ) + event_id = hashlib.sha256(serialized.encode("utf-8")).hexdigest() + # zero aux randomness -> fully deterministic signatures (BIP-340 permits it; + # verification is unaffected either way) + signature = ( + PrivateKey(secret) + .sign_schnorr(bytes.fromhex(event_id), aux_randomness=b"\x00" * 32) + .hex() + ) + return { + "pubkey": pubkey, + "created_at": created_at, + "kind": kind, + "tags": tags or [], + "content": content, + "id": event_id, + "sig": signature, + } + + +def _schnorr_verify(sig32: bytes, msg32: bytes, xonly_pub32: bytes) -> bool: + """BIP-340 reference verification (pure Python, public domain spec code). + + Signing runs in coincurve (libsecp256k1); verification is pure Python so + the auditor can read every step without CFFI indirection. + """ + P = 2**256 - 2**32 - 977 + N = 0xFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFEBAAEDCE6AF48A03BBFD25E8CD0364141 + + def tag_hash(tag: str, msg: bytes) -> bytes: + t = hashlib.sha256(tag.encode()).digest() + return hashlib.sha256(t + t + msg).digest() + + def lift_x(b: int): + if b >= P: + return None + y_sq = (pow(b, 3, P) + 7) % P + y = pow(y_sq, (P + 1) // 4, P) + if pow(y, 2, P) != y_sq: + return None + return (b, y if y % 2 == 0 else P - y) + + def point_add(p1, p2): + if p1 is None: + return p2 + if p2 is None: + return p1 + if p1[0] == p2[0] and p1[1] != p2[1]: + return None + if p1 == p2: + lam = (3 * p1[0] * p1[0] * pow(2 * p1[1], P - 2, P)) % P + else: + lam = ((p2[1] - p1[1]) * pow(p2[0] - p1[0], P - 2, P)) % P + x3 = (lam * lam - p1[0] - p2[0]) % P + return (x3, (lam * (p1[0] - x3) - p1[1]) % P) + + def point_mul(p, scalar): + acc = None + for i in range(256): + if (scalar >> i) & 1: + acc = point_add(acc, p) + p = point_add(p, p) + return acc + + if len(sig32) != 64 or len(msg32) != 32 or len(xonly_pub32) != 32: + return False + r = int.from_bytes(sig32[:32], "big") + s = int.from_bytes(sig32[32:], "big") + if not (0 <= r < P and 0 <= s < N): + return False + pub_point = lift_x(int.from_bytes(xonly_pub32, "big")) + if pub_point is None: + return False + e = int.from_bytes(tag_hash("BIP0340/challenge", sig32[:32] + xonly_pub32 + msg32), "big") % N + r_point = point_add(point_mul((Gx := 0x79BE667EF9DCBBAC55A06295CE870B07029BFCDB2DCE28D959F2815B16F81798, Gy := 0x483ADA7726A3C4655DA4FBFC0E1108A8FD17B448A68554199C47D08FFB10D4B8), s), point_mul(pub_point, (N - e) % N)) + if r_point is None or r_point[1] % 2 != 0: + return False + return r_point[0] == r + + +def verify_event(event: dict[str, Any]) -> bool: + """Recompute the id and verify the Schnorr signature against the pubkey.""" + serialized = json.dumps( + [0, event["pubkey"], event["created_at"], event["kind"], event["tags"], event["content"]], + separators=(",", ":"), + ensure_ascii=False, + ) + if hashlib.sha256(serialized.encode("utf-8")).hexdigest() != event["id"]: + return False + compressed = PublicKey(bytes.fromhex("02" + event["pubkey"])).format() # 33-byte, prefix 02 + return _schnorr_verify( + bytes.fromhex(event["sig"]), bytes.fromhex(event["id"]), compressed[1:] + ) + + +def relay_publish(relay_url: str, event: dict[str, Any], timeout: float = 10.0) -> str: + """Publish one event to one relay; return the relay's OK message.""" + from websockets.sync.client import connect + + with connect(relay_url, close_timeout=timeout) as ws: + ws.send(json.dumps(["EVENT", event])) + reply = json.loads(ws.recv(timeout=timeout)) + if not (isinstance(reply, list) and len(reply) >= 3 and reply[0] == "OK" and reply[2] is True): + raise RuntimeError(f"relay {relay_url} rejected event: {str(reply)[:200]}") + return str(reply[1]) + + +def publish(relays: list[str], event: dict[str, Any], timeout: float = 10.0) -> list[dict[str, str]]: + """Publish to every relay; report per-relay outcome (never raises unless all fail).""" + results = [] + for relay in relays: + try: + relay_publish(relay, event, timeout) + results.append({"relay": relay, "ok": "true"}) + except Exception as exc: # noqa: BLE001 — per-relay isolation + results.append({"relay": relay, "ok": "false", "error": str(exc)[:200]}) + if relays and all(r["ok"] == "false" for r in results): + raise RuntimeError(f"all relays failed: {results}") + return results diff --git a/cayceops/components/publisher/publisher.py b/cayceops/components/publisher/publisher.py new file mode 100644 index 0000000..2c6e045 --- /dev/null +++ b/cayceops/components/publisher/publisher.py @@ -0,0 +1,266 @@ +"""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() diff --git a/tests/test_publisher.py b/tests/test_publisher.py new file mode 100644 index 0000000..c6d870a --- /dev/null +++ b/tests/test_publisher.py @@ -0,0 +1,162 @@ +"""publisher tests — calendar, human gate, nostr publish, disabled adapters.""" + +from __future__ import annotations + +import os + +import pytest + +from cayceops.components.publisher import adapters, nostr, publisher +from cayceops.models import AuditEntry, ContentItem, ContentStatus + +# invented test key (never a real nostr key — allowlisted in .gitleaks.toml) +TEST_KEY = "11" * 32 + + +def _item(channel="nostr", theme="atm-fees", scheduled=None, status=ContentStatus.APPROVED): + return ContentItem( + theme=theme, + channel=channel, + caption="The first ATM you find charges 10-15% over spot.\n\nbitcoinhotels.org/guides/atm", + link="bitcoinhotels.org/guides/atm", + status=status, + scheduled_for=scheduled, + ) + + +# --- nostr client ------------------------------------------------------------- + + +def test_event_build_verify_roundtrip(): + event = nostr.build_event(bytes.fromhex(TEST_KEY), "gm", tags=[["t", "bitcoin"]], created_at=1700000000) + assert nostr.verify_event(event) + assert len(event["id"]) == 64 and len(event["sig"]) == 128 + + tampered = {**event, "content": "gm (edited)"} + assert not nostr.verify_event(tampered) + + +def test_event_is_deterministic(): + a = nostr.build_event(bytes.fromhex(TEST_KEY), "gm", created_at=1700000000) + b = nostr.build_event(bytes.fromhex(TEST_KEY), "gm", created_at=1700000000) + assert a == b + + +def test_key_validation(): + with pytest.raises(ValueError): + nostr.normalize_key("not-a-key") + assert nostr.normalize_key("0x" + TEST_KEY) == bytes.fromhex(TEST_KEY) + + +# --- calendar ------------------------------------------------------------------ + + +def test_plan_creates_idempotent_drafts(cfg, sess): + config = cfg() + first = publisher._plan(config, sess) + second = publisher._plan(config, sess) + assert first["created"] == 4 + assert second["created"] == 0 # same themes+channels already queued + + +def test_approve_gate(cfg, sess): + item = _item(status=ContentStatus.DRAFT) + sess.add(item) + sess.flush() + + assert publisher._approve(item.id, "gustavo", sess) == 0 + assert item.status is ContentStatus.APPROVED + assert publisher._approve(item.id, "gustavo", sess) == 1 # not draft anymore + + actor = sess.query(AuditEntry).filter_by(action="approve_content").first().actor + assert actor == "gustavo" + + +# --- publishing ------------------------------------------------------------------ + + +def test_publish_due_nostr_happy_path(cfg, sess): + item = _item() # approved, scheduled None -> due now + sess.add(item) + sess.flush() + + relays_seen = [] + + def fake_sender(relays, event, timeout=10.0): + relays_seen.append(relays) + assert nostr.verify_event(event) + return [{"relay": r, "ok": "true"} for r in relays] + + config = cfg( + CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY, + CAYCE_OPS_NOSTR_RELAYS="wss://relay.example.org,wss://other.example.org", + ) + stats = publisher._publish_due(config, sess, nostr_sender=fake_sender) + + assert stats["published"] == 1 + assert item.status is ContentStatus.PUBLISHED + assert item.post_ref and len(item.post_ref) == 64 + assert item.published_at is not None + assert "wss://relay.example.org" in relays_seen[0] + + +def test_future_scheduled_item_not_published(cfg, sess): + from datetime import UTC, datetime, timedelta + + item = _item(scheduled=datetime.now(UTC) + timedelta(hours=2)) + sess.add(item) + sess.flush() + config = cfg(CAYCE_OPS_NOSTR_PRIVATE_KEY=TEST_KEY, CAYCE_OPS_NOSTR_RELAYS="wss://r.example.org") + stats = publisher._publish_due(config, sess, nostr_sender=lambda r, e, timeout=10: []) + assert stats["published"] == 0 + assert item.status is ContentStatus.APPROVED + + +def test_nostr_without_config_fails_clean(cfg, sess): + item = _item() + sess.add(item) + sess.flush() + config = cfg() # no key, no relays + stats = publisher._publish_due(config, sess) + assert stats["failed"] == 1 + assert item.status is ContentStatus.APPROVED # stays approved, error recorded + assert item.error and "nostr" in item.error + + +def test_non_nostr_channels_skipped_by_default(cfg, sess): + item = _item(channel="facebook", theme="lightning-abroad") + sess.add(item) + sess.flush() + stats = publisher._publish_due(cfg(), sess) + assert stats["skipped"] == 1 + assert item.status is ContentStatus.APPROVED + skip = sess.query(AuditEntry).filter_by(action="publish_skipped").one() + assert skip.detail["channel"] == "facebook" + + +def test_meta_adapters_disabled_by_default(cfg): + config = cfg() + with pytest.raises(PermissionError): + adapters.post_facebook(config, "hello") + with pytest.raises(PermissionError): + adapters.post_instagram(config, "hello", "https://img.example.org/x.jpg") + + +# --- images ------------------------------------------------------------------ + + +def test_brand_image_render(tmp_path, cfg): + config = cfg(CAYCE_OPS_MEDIA_DIR=str(tmp_path)) + out = publisher._render_image(_item(), config) + assert os.path.exists(out) + from PIL import Image + + with Image.open(out) as img: + assert img.size == (1080, 1350) + + +def test_unknown_brand_rejected(): + from cayceops.components.publisher import images + + with pytest.raises(ValueError): + images.render_brand_image(brand="xx", kicker=None, lines=["A"], out_path="/tmp/x.jpg")