diff --git a/automation/anwarpy/publication_meta.py b/automation/anwarpy/publication_meta.py index 9ae7507..728624f 100644 --- a/automation/anwarpy/publication_meta.py +++ b/automation/anwarpy/publication_meta.py @@ -129,10 +129,18 @@ def same_meta(expected_raw: str, actual_raw: str) -> bool: def mark_publishing(raw: str, slug: str) -> str: meta = parse_meta(raw) - if meta["status"] != "pending": - raise ValueError("only pending markers can be claimed") - meta["status"] = "publishing" - meta["slug"] = clean(slug) + if meta["status"] not in ("pending", "republish"): + raise ValueError("only pending or republish markers can be claimed") + normalized_slug = clean(slug) + if meta["status"] == "republish" and normalized_slug != meta["slug"]: + raise ValueError("republish slug cannot change") + meta = { + "status": "publishing", + "access": meta["access"], + "category": meta["category"], + "queued_at": meta["queued_at"], + "slug": normalized_slug, + } return render_meta(meta) @@ -148,8 +156,13 @@ def mark_published(raw: str, published_at: str, url: str) -> str: def mark_blocked(raw: str, reason: str) -> str: meta = parse_meta(raw) - if meta["status"] != "pending": - raise ValueError("only pending markers can be blocked") - meta["status"] = "blocked" - meta["blocked_reason"] = clean(reason) + if meta["status"] not in ("pending", "publishing"): + raise ValueError("only pending or publishing markers can be blocked") + meta = { + "status": "blocked", + "access": meta["access"], + "category": meta["category"], + "queued_at": meta["queued_at"], + "blocked_reason": clean(reason), + } return render_meta(meta) diff --git a/automation/anwarpy/publish_one.py b/automation/anwarpy/publish_one.py new file mode 100644 index 0000000..9302dc2 --- /dev/null +++ b/automation/anwarpy/publish_one.py @@ -0,0 +1,321 @@ +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 diff --git a/automation/anwarpy/test_publication_meta.py b/automation/anwarpy/test_publication_meta.py index 80e7909..daa7b2e 100644 --- a/automation/anwarpy/test_publication_meta.py +++ b/automation/anwarpy/test_publication_meta.py @@ -52,6 +52,10 @@ class PublicationMetaTests(unittest.TestCase): def test_blocked_is_terminal_and_carries_a_bounded_reason(self): blocked = mark_blocked(PENDING, "cover") self.assertEqual(parse_meta(blocked)["blocked_reason"], "cover") + publishing = mark_publishing(PENDING, "arabic-ai-guide") + blocked_after_claim = parse_meta(mark_blocked(publishing, "generation")) + self.assertEqual(blocked_after_claim["status"], "blocked") + self.assertNotIn("slug", blocked_after_claim) with self.assertRaises(ValueError): mark_blocked(PENDING, "network") diff --git a/automation/anwarpy/test_publish_one.py b/automation/anwarpy/test_publish_one.py new file mode 100644 index 0000000..0a2d19f --- /dev/null +++ b/automation/anwarpy/test_publish_one.py @@ -0,0 +1,269 @@ +import fcntl +import tempfile +import unittest +from pathlib import Path + +from page_state import find_marker +from publish_one import ( + AmbiguousWrite, + DeterministicFailure, + PublicVerificationError, + Publisher, +) + + +WORKSPACE = "workspace" +PENDING_OLD = "anwarpy: status=pending | access=free | category=نماذج | queued_at=2026-10-08T09:00:00+03:00" +PENDING_NEW = "anwarpy: status=pending | access=vip | category=أدوات | queued_at=2026-10-09T09:00:00+03:00" +BLOCKED_OLDER = "anwarpy: status=blocked | access=free | category=نماذج | queued_at=2026-10-07T09:00:00+03:00 | blocked_reason=cover" +REPUBLISH = "anwarpy: status=republish | access=vip | category=أدوات | queued_at=2026-10-10T09:00:00+03:00 | slug=existing-guide | published_at=2026-10-09T12:00:00+03:00 | url=https://anwarpy.ai/posts/existing-guide" + + +def blocks(marker, evidence="Public source text"): + return [ + {"id": "body", "type": "paragraph", "text": evidence, "depth": 0}, + {"id": "marker", "type": "paragraph", "text": marker, "depth": 0}, + ] + + +class FakeAppFlowy: + def __init__(self, page_markers): + self.pages = [ + {"view_id": page_id, "name": title} + for page_id, (title, _marker) in page_markers.items() + ] + self.block_map = { + page_id: blocks(marker) for page_id, (_title, marker) in page_markers.items() + } + self.list_calls = 0 + self.edits = [] + + def list_pages(self): + self.list_calls += 1 + return self.pages + + def get_blocks(self, page_id): + return [dict(block) for block in self.block_map[page_id]] + + def edit_block(self, page_id, block_id, text): + self.edits.append((page_id, block_id, text)) + for block in self.block_map[page_id]: + if block["id"] == block_id: + block["text"] = text + return + raise KeyError(block_id) + + def marker(self, page_id): + return find_marker(self.get_blocks(page_id)).meta + + +class FakeGenerator: + def __init__(self, slugs=None): + self.slugs = slugs or {} + self.calls = [] + + def generate(self, title, evidence, meta): + self.calls.append((title, evidence, dict(meta))) + slug = self.slugs.get(title, "generated-guide") + return { + "title": f"دليل {title}", + "slug": slug, + "dek": "شرح عربي عملي ومباشر", + "teaser": "مقدمة مجانية مفيدة" if meta["access"] == "vip" else "", + "sections": [ + {"heading": "المشكلة", "paragraphs": ["شرح واضح للمشكلة والحل."], "code": None}, + {"heading": "التحقق", "paragraphs": ["تحقق من النتيجة بخطوة قابلة للقياس."], "code": None}, + {"heading": "القيود", "paragraphs": ["راجع القيود قبل الاستخدام."], "code": None}, + ], + } + + +class FakeMedia: + def __init__(self): + self.calls = [] + + def render(self, article, meta): + self.calls.append((article["slug"], dict(meta))) + return { + "coverImage": {"filename": f"{article['slug']}.jpg", "dataBase64": "/9j/2Q=="}, + "video_path": f"/tmp/{article['slug']}.mp4", + } + + +class FakeApi: + def __init__(self, records=None, timeout_after_create=False, public_ok=True): + self.records = records or {} + self.timeout_after_create = timeout_after_create + self.public_ok = public_ok + self.create_calls = [] + self.update_calls = [] + self.verify_calls = [] + + def lookup(self, slug): + record = self.records.get(slug) + return dict(record) if record else None + + def create(self, body): + self.create_calls.append(dict(body)) + record = { + "id": len(self.records) + 1, + "title": body["title"], + "slug": body["slug"], + "status": "published" if body["publish"] else "draft", + "access": body["access"], + "category": body["category"], + "publishedAt": "2026-10-09T12:20:11+00:00", + "url": f"https://anwarpy.ai/posts/{body['slug']}", + } + self.records[body["slug"]] = record + if self.timeout_after_create: + self.timeout_after_create = False + raise AmbiguousWrite("timeout after create") + return dict(record) + + def update(self, slug, body): + self.update_calls.append((slug, dict(body))) + old = self.records[slug] + old.update({ + "title": body.get("title", old.get("title")), + "status": "published" if body.get("publish") else old["status"], + "access": body.get("access", old["access"]), + "category": body.get("category", old["category"]), + "publishedAt": old.get("publishedAt") or "2026-10-09T12:20:11+00:00", + "url": f"https://anwarpy.ai/posts/{slug}", + }) + return dict(old) + + def verify_public(self, record): + self.verify_calls.append(dict(record)) + return self.public_ok + + +class PublisherTests(unittest.TestCase): + def make_publisher(self, app, api, generator=None, media=None, lock_path=None): + return Publisher( + appflowy=app, + api=api, + generator=generator or FakeGenerator(), + media=media or FakeMedia(), + lock_path=lock_path or Path(tempfile.mkdtemp()) / "publisher.lock", + ) + + def test_lock_loser_makes_no_external_calls(self): + lock_path = Path(tempfile.mkdtemp()) / "publisher.lock" + app = FakeAppFlowy({"page": ("Topic", PENDING_OLD)}) + api = FakeApi() + with lock_path.open("a+") as handle: + fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + result = self.make_publisher(app, api, lock_path=lock_path).run() + self.assertEqual(result, "busy") + self.assertEqual(app.list_calls, 0) + + def test_fifo_uses_queued_at_and_skips_blocked(self): + app = FakeAppFlowy({ + "blocked": ("Blocked", BLOCKED_OLDER), + "new": ("New", PENDING_NEW), + "old": ("Old", PENDING_OLD), + }) + generator = FakeGenerator({"Old": "old-guide", "New": "new-guide"}) + result = self.make_publisher(app, FakeApi(), generator=generator).run() + self.assertEqual(result, "published:old-guide") + self.assertEqual(generator.calls[0][0], "Old") + self.assertEqual(app.marker("blocked")["status"], "blocked") + self.assertEqual(app.marker("new")["status"], "pending") + + def test_pending_absent_slug_posts_once_and_rerun_is_idle(self): + app = FakeAppFlowy({"page": ("Topic", PENDING_OLD)}) + api = FakeApi() + publisher = self.make_publisher(app, api) + self.assertEqual(publisher.run(), "published:generated-guide") + self.assertEqual(len(api.create_calls), 1) + self.assertEqual(app.marker("page")["status"], "published") + self.assertEqual(publisher.run(), "idle") + self.assertEqual(len(api.create_calls), 1) + + def test_existing_draft_is_promoted_without_post(self): + record = { + "id": 1, "title": "Draft", "slug": "generated-guide", "status": "draft", + "access": "free", "category": "نماذج", "publishedAt": None, + } + app = FakeAppFlowy({"page": ("Topic", PENDING_OLD)}) + api = FakeApi({"generated-guide": record}) + self.assertEqual(self.make_publisher(app, api).run(), "published:generated-guide") + self.assertEqual(api.create_calls, []) + self.assertEqual(len(api.update_calls), 1) + + def test_republish_puts_existing_slug_and_never_posts(self): + record = { + "id": 7, "title": "Old", "slug": "existing-guide", "status": "published", + "access": "vip", "category": "أدوات", "publishedAt": "2026-10-09T09:00:00Z", + "url": "https://anwarpy.ai/posts/existing-guide", + } + app = FakeAppFlowy({"page": ("Topic", REPUBLISH)}) + api = FakeApi({"existing-guide": record}) + generator = FakeGenerator({"Topic": "existing-guide"}) + self.assertEqual(self.make_publisher(app, api, generator=generator).run(), "published:existing-guide") + self.assertEqual(api.create_calls, []) + self.assertEqual(len(api.update_calls), 1) + + def test_ambiguous_create_reconciles_without_second_post(self): + app = FakeAppFlowy({"page": ("Topic", PENDING_OLD)}) + api = FakeApi(timeout_after_create=True) + self.assertEqual(self.make_publisher(app, api).run(), "published:generated-guide") + self.assertEqual(len(api.create_calls), 1) + + def test_public_failure_leaves_publishing_and_does_not_repost(self): + app = FakeAppFlowy({"page": ("Topic", PENDING_OLD)}) + api = FakeApi(public_ok=False) + publisher = self.make_publisher(app, api) + with self.assertRaises(PublicVerificationError): + publisher.run() + self.assertEqual(app.marker("page")["status"], "publishing") + self.assertEqual(len(api.create_calls), 1) + with self.assertRaises(PublicVerificationError): + publisher.run() + self.assertEqual(len(api.create_calls), 1) + + def test_api_metadata_mismatch_leaves_publishing(self): + app = FakeAppFlowy({"page": ("Topic", PENDING_OLD)}) + api = FakeApi() + original_create = api.create + + def mismatched(body): + result = original_create(body) + result["access"] = "vip" + api.records[body["slug"]]["access"] = "vip" + return result + + api.create = mismatched + with self.assertRaisesRegex(ValueError, "access mismatch"): + self.make_publisher(app, api).run() + self.assertEqual(app.marker("page")["status"], "publishing") + self.assertEqual(len(api.create_calls), 1) + + def test_generation_failure_blocks_page_without_api_write(self): + app = FakeAppFlowy({"page": ("Topic", PENDING_OLD)}) + api = FakeApi() + generator = FakeGenerator() + + def fail(*_args): + raise DeterministicFailure("generation", "model failed") + + generator.generate = fail + self.assertEqual(self.make_publisher(app, api, generator=generator).run(), "blocked:generation") + self.assertEqual(app.marker("page")["status"], "blocked") + self.assertEqual(api.create_calls, []) + + def test_malformed_marker_is_reported_and_valid_page_can_run(self): + app = FakeAppFlowy({ + "bad": ("Bad", "Anwarpy: pending"), + "good": ("Good", PENDING_OLD), + }) + publisher = self.make_publisher(app, FakeApi(), generator=FakeGenerator({"Good": "good-guide"})) + self.assertEqual(publisher.run(), "published:good-guide") + self.assertEqual(len(publisher.marker_errors), 1) + self.assertIn("bad", publisher.marker_errors[0]) + + +if __name__ == "__main__": + unittest.main()