from __future__ import annotations from contextlib import contextmanager from dataclasses import dataclass from datetime import datetime import fcntl import math from pathlib import Path import re from typing import Any, Iterable from zoneinfo import ZoneInfo from page_state import MarkerError, PageMarker, find_marker from publication_meta import ( SLUG_RE, mark_blocked, mark_published, mark_publishing, same_meta, ) from sanitize import assert_publishable, sanitize_text RIYADH = ZoneInfo("Asia/Riyadh") class AmbiguousWrite(RuntimeError): """The remote side effect may have succeeded; reconcile before retrying.""" class PublicVerificationError(RuntimeError): pass class DeterministicFailure(RuntimeError): def __init__(self, reason: str, message: str): super().__init__(message) self.reason = reason @dataclass(frozen=True) class Candidate: page: dict[str, Any] marker: PageMarker blocks: list[dict[str, Any]] @contextmanager def exclusive_run_lock(path: Path): path.parent.mkdir(parents=True, exist_ok=True) with path.open("a+") as handle: try: fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: yield False return try: yield True finally: fcntl.flock(handle.fileno(), fcntl.LOCK_UN) def _queued_at(candidate: Candidate) -> datetime: return datetime.fromisoformat(candidate.marker.meta["queued_at"].replace("Z", "+00:00")) def _lexical(article: dict[str, Any]) -> dict[str, Any]: children: list[dict[str, Any]] = [] for section in article["sections"]: children.append({ "type": "heading", "tag": "h2", "version": 1, "children": [{"type": "text", "version": 1, "text": section["heading"]}], }) for paragraph in section["paragraphs"]: children.append({ "type": "paragraph", "version": 1, "children": [{"type": "text", "version": 1, "text": paragraph}], }) if section.get("code"): children.append({ "type": "code", "version": 1, "language": "text", "children": [{"type": "text", "version": 1, "text": section["code"]}], }) return { "root": { "type": "root", "format": "", "indent": 0, "version": 1, "direction": "rtl", "children": children, } } def _validate_article(article: Any, access: str, expected_slug: str | None = None) -> dict[str, Any]: if not isinstance(article, dict): raise DeterministicFailure("article_schema", "article must be an object") for field in ("title", "slug", "dek", "sections"): if field not in article: raise DeterministicFailure("article_schema", f"article field missing: {field}") for field in ("title", "slug", "dek"): if not isinstance(article[field], str) or not article[field].strip(): raise DeterministicFailure("article_schema", f"article field invalid: {field}") if not SLUG_RE.fullmatch(article["slug"]): raise DeterministicFailure("article_schema", "article slug must be lowercase ASCII kebab-case") if expected_slug is not None and article["slug"] != expected_slug: raise DeterministicFailure("article_schema", "republish slug cannot change") if access == "vip" and (not isinstance(article.get("teaser"), str) or not article["teaser"].strip()): raise DeterministicFailure("article_schema", "VIP teaser is required") sections = article["sections"] if not isinstance(sections, list) or not 3 <= len(sections) <= 6: raise DeterministicFailure("article_schema", "article must contain 3 to 6 sections") for section in sections: if not isinstance(section, dict): raise DeterministicFailure("article_schema", "section must be an object") if not isinstance(section.get("heading"), str) or not section["heading"].strip(): raise DeterministicFailure("article_schema", "section heading is required") paragraphs = section.get("paragraphs") if not isinstance(paragraphs, list) or not paragraphs or not all( isinstance(value, str) and value.strip() for value in paragraphs ): raise DeterministicFailure("article_schema", "section paragraphs are invalid") return article def _riyadh_iso(value: str) -> str: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) if parsed.tzinfo is None or parsed.utcoffset() is None: raise ValueError("publishedAt must include a timezone") return parsed.astimezone(RIYADH).isoformat() class Publisher: def __init__(self, *, appflowy, api, generator, media, lock_path: Path, allowed_hosts: Iterable[str] = ("anwarpy.ai",)): self.appflowy = appflowy self.api = api self.generator = generator self.media = media self.lock_path = lock_path self.allowed_hosts = set(allowed_hosts) self.marker_errors: list[str] = [] def run(self) -> str: with exclusive_run_lock(self.lock_path) as acquired: if not acquired: return "busy" candidates = self._candidates() publishing = sorted( (candidate for candidate in candidates if candidate.marker.meta["status"] == "publishing"), key=lambda candidate: (_queued_at(candidate), candidate.page["view_id"]), ) if publishing: return self._reconcile(publishing[0]) eligible = sorted( ( candidate for candidate in candidates if candidate.marker.meta["status"] in ("pending", "republish") ), key=lambda candidate: (_queued_at(candidate), candidate.page["view_id"]), ) if not eligible: return "idle" return self._publish(eligible[0]) def _candidates(self) -> list[Candidate]: found: list[Candidate] = [] self.marker_errors = [] for page in self.appflowy.list_pages(): page_id = page["view_id"] blocks = self.appflowy.get_blocks(page_id) try: marker = find_marker(blocks) except MarkerError as error: self.marker_errors.append(f"{page_id}: {error}") continue if marker is not None: found.append(Candidate(page=page, marker=marker, blocks=blocks)) return found def _read_marker(self, candidate: Candidate) -> PageMarker: marker = find_marker(self.appflowy.get_blocks(candidate.page["view_id"])) if marker is None: raise ValueError("publication marker disappeared") return marker def _edit_and_verify(self, candidate: Candidate, marker_text: str) -> PageMarker: self.appflowy.edit_block(candidate.page["view_id"], candidate.marker.block_id, marker_text) readback = self._read_marker(candidate) if not same_meta(marker_text, readback.raw): raise ValueError("AppFlowy marker read-back mismatch") return readback def _evidence(self, candidate: Candidate) -> str: texts = [ str(block.get("text") or "") for block in candidate.blocks if block.get("id") != candidate.marker.block_id ] sanitized = sanitize_text("\n\n".join(texts), allowed_hosts=self.allowed_hosts) assert_publishable(sanitized, allowed_hosts=self.allowed_hosts) if not sanitized.strip(): raise DeterministicFailure("sanitizer", "source is empty after sanitization") return sanitized def _block(self, candidate: Candidate, reason: str) -> str: current = self._read_marker(candidate) blocked = mark_blocked(current.raw, reason) self._edit_and_verify(Candidate(candidate.page, current, candidate.blocks), blocked) return f"blocked:{reason}" def _publish(self, candidate: Candidate) -> str: original = candidate.marker meta = original.meta try: evidence = self._evidence(candidate) article = _validate_article( self.generator.generate(candidate.page["name"], evidence, meta), meta["access"], meta.get("slug") if meta["status"] == "republish" else None, ) media = self.media.render(article, meta) if not isinstance(media, dict) or "coverImage" not in media or "video_path" not in media: raise DeterministicFailure("cover", "media output is incomplete") except DeterministicFailure as error: return self._block(candidate, error.reason) except ValueError: return self._block(candidate, "sanitizer") current = self._read_marker(candidate) if not same_meta(original.raw, current.raw): raise ValueError("AppFlowy marker changed during generation") claimed_text = mark_publishing(current.raw, article["slug"]) claimed = self._edit_and_verify(Candidate(candidate.page, current, candidate.blocks), claimed_text) claimed_candidate = Candidate(candidate.page, claimed, candidate.blocks) body = self._post_body(article, meta, media["coverImage"]) existing = self.api.lookup(article["slug"]) record = None if meta["status"] == "republish": if existing is None: raise AmbiguousWrite("republish target is missing; POST is forbidden") record = self.api.update(article["slug"], body) elif existing is None: try: record = self.api.create(body) except AmbiguousWrite: record = self.api.lookup(article["slug"]) if record is None: raise elif existing.get("status") == "draft": record = self.api.update(article["slug"], {"publish": True}) else: record = existing return self._finish(claimed_candidate, record) def _reconcile(self, candidate: Candidate) -> str: slug = candidate.marker.meta["slug"] record = self.api.lookup(slug) if record is None: return f"awaiting-reconciliation:{slug}" if record.get("status") == "draft": record = self.api.update(slug, {"publish": True}) result = self._finish(candidate, record) return result def _finish(self, candidate: Candidate, record: Any) -> str: if not isinstance(record, dict): raise ValueError("agent API returned an invalid record") expected = candidate.marker.meta slug = expected["slug"] for field in ("slug", "access", "category"): if record.get(field) != expected[field]: raise ValueError(f"agent API {field} mismatch") if record.get("status") != "published": raise ValueError("agent API post is not published") published_at = record.get("publishedAt") if not isinstance(published_at, str): raise ValueError("agent API publishedAt is missing") url = record.get("url") or f"https://anwarpy.ai/posts/{slug}" if not self.api.verify_public(record): raise PublicVerificationError(f"public verification failed for {slug}") published = mark_published(candidate.marker.raw, _riyadh_iso(published_at), url) self._edit_and_verify(candidate, published) return f"published:{slug}" @staticmethod def _post_body(article: dict[str, Any], meta: dict[str, str], cover_image: dict[str, Any]) -> dict[str, Any]: words = re.findall(r"\S+", " ".join( paragraph for section in article["sections"] for paragraph in section["paragraphs"] )) body: dict[str, Any] = { "title": article["title"], "slug": article["slug"], "dek": article["dek"], "content": _lexical(article), "category": meta["category"], "access": meta["access"], "readingMinutes": max(1, math.ceil(len(words) / 200)), "publish": True, "coverImage": cover_image, } if meta["access"] == "vip": body["teaser"] = article["teaser"] if body["category"] != meta["category"]: raise ValueError("category changed") if body["access"] != meta["access"]: raise ValueError("access changed") return body