From 023eb8f70eced8b9a1b5f391fde4b691d6370930 Mon Sep 17 00:00:00 2001 From: Zaine Date: Wed, 13 May 2026 22:52:47 +0100 Subject: [PATCH] breaking code down --- Makefile | 2 +- README.md | 6 + authoring_server.py | 5206 +---------------- src/authoring_service/__init__.py | 10 + src/authoring_service/build.py | 167 + src/authoring_service/config.py | 97 + src/authoring_service/constants.py | 202 + src/authoring_service/content.py | 427 ++ src/authoring_service/hidden.py | 660 +++ src/authoring_service/models.py | 34 + src/authoring_service/templates.py | 3490 +++++++++++ src/authoring_service/utils.py | 57 + src/authoring_service/web.py | 156 + {tests => src/tests}/test_authoring_server.py | 44 +- 14 files changed, 5339 insertions(+), 5219 deletions(-) create mode 100644 src/authoring_service/__init__.py create mode 100644 src/authoring_service/build.py create mode 100644 src/authoring_service/config.py create mode 100644 src/authoring_service/constants.py create mode 100644 src/authoring_service/content.py create mode 100644 src/authoring_service/hidden.py create mode 100644 src/authoring_service/models.py create mode 100644 src/authoring_service/templates.py create mode 100644 src/authoring_service/utils.py create mode 100644 src/authoring_service/web.py rename {tests => src/tests}/test_authoring_server.py (89%) mode change 100755 => 100644 diff --git a/Makefile b/Makefile index 8e75258..1b6effa 100755 --- a/Makefile +++ b/Makefile @@ -19,7 +19,7 @@ JOURNALCTL_USER := env XDG_RUNTIME_DIR=$(SYSTEMD_RUNTIME_DIR) journalctl --user all: author-restart test: $(VENV) - $(PY) -m unittest discover -s tests -p 'test_authoring_server.py' -v + PYTHONPATH=src $(PY) -m unittest discover -s src/tests -p 'test_authoring_server.py' -v $(VENV): @echo "Generating virtual environment" diff --git a/README.md b/README.md index 98dda53..d9b8b0b 100755 --- a/README.md +++ b/README.md @@ -24,3 +24,9 @@ Useful commands: - `make author-logs` shows recent service logs. The service code runs from this repository. The editable content root is set by `AUTHOR_CONTENT_ROOT` in `systemd/user/org-web-authoring.service`; it currently points at the website repository. Hidden-details backups are stored in this repository under `backups/hidden-details`. + +Code layout: + +- `authoring_server.py` is the compatibility executable used by systemd. +- `src/authoring_service/` contains the implementation modules. +- `src/tests/` contains the unit tests. diff --git a/authoring_server.py b/authoring_server.py index 91dbd3d..74c902d 100755 --- a/authoring_server.py +++ b/authoring_server.py @@ -1,5209 +1,17 @@ #!/usr/bin/env python3 -"""Local authoring UI for the Org published website.""" +"""Compatibility entry point for the authoring service.""" from __future__ import annotations -import json -import os -import posixpath -import re -import struct -import subprocess -import sys -import threading -import time -import mimetypes -from dataclasses import dataclass, field -from datetime import datetime -from email.parser import BytesParser -from email.policy import default as email_default_policy -from email.utils import formatdate -from http import HTTPStatus -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path -from typing import Any, Callable -from urllib.parse import parse_qs, urlparse +import sys +SRC = Path(__file__).resolve().parent / "src" +if str(SRC) not in sys.path: + sys.path.insert(0, str(SRC)) -APP_ROOT = Path(__file__).resolve().parent -DEFAULT_CONTENT_ROOT = Path("/home/zaine/master-folder/org_files/org_web") - - -def looks_like_content_root(path: Path) -> bool: - return (path / "blogs").exists() or (path / "posts").exists() or (path / "lima").exists() - - -def resolve_root() -> Path: - for env_name in ("AUTHOR_CONTENT_ROOT", "AUTHOR_ROOT"): - env_root = os.environ.get(env_name) - if not env_root: - continue - resolved = Path(env_root).expanduser().resolve() - if looks_like_content_root(resolved): - return resolved - candidates = [ - Path.cwd(), - DEFAULT_CONTENT_ROOT, - APP_ROOT, - ] - workspace = os.environ.get("GITHUB_WORKSPACE") - if workspace: - candidates.insert(0, Path(workspace)) - for base in list(candidates): - candidates.extend(base.parents) - seen = set() - for candidate in candidates: - resolved = candidate.expanduser().resolve() - if resolved in seen: - continue - seen.add(resolved) - if looks_like_content_root(resolved): - return resolved - return APP_ROOT - - -ROOT = resolve_root() -BLOGS_DIR = ROOT / "blogs" -POSTS_DIR = ROOT / "posts" -LIMA_DIR = ROOT / "lima" -IMAGE_ASSETS_DIR = ROOT / "assets" / "images" -HZONE_ASSETS_DIR = IMAGE_ASSETS_DIR / "hzone" -HIDDEN_DETAILS_JS = ROOT / "assets" / "scripts" / "hidden-details.js" -HIDDEN_CONTENT_JSON = ROOT / "assets" / "content" / "hidden-details.json" -HIDDEN_BACKUP_DIR = Path(os.environ.get("AUTHOR_HIDDEN_BACKUP_DIR", APP_ROOT / "backups" / "hidden-details")).expanduser().resolve() -EXCLUDED_CONTENT_DIR_NAMES = { - ".agents", - ".codex", - ".git", - ".packages", - ".venv", - "__pycache__", - "assets", - "backups", - "output", - "tags", -} -GENERATED_ORG_NAMES = { - "blogs-list.org", - "books-list.org", - "posts-list.org", - "career-list.org", - "sitemap.org", - "recently-updated.org", - "wip.org", -} -GENERATED_CONTENT_NAMES = GENERATED_ORG_NAMES | {"lima-list.org"} -ALLOWED_UPLOAD_EXTENSIONS = { - ".png", - ".jpg", - ".jpeg", - ".gif", - ".webp", - ".svg", -} -MONTH_NAMES = [ - "january", - "february", - "march", - "april", - "may", - "june", - "july", - "august", - "september", - "october", - "november", - "december", -] - -HIDDEN_CONTENT_TYPES = [ - "quote", - "poem", - "hidden dialogue", - "journal entry", - "rare event", - "loading screen message", - "secret interaction", - "hidden tooltip", - "future z message", - "young z memory fragment", - "sensei chi wisdom entry", - "aphy system message", - "lima note/message", - "dream sequence", - "terminal log", - "fake error message", - "recurring joke", - "seasonal event", - "weather-based event", - "hover message", - "hidden achievement", - "guestbook entry", - "hidden conversation", - "family layer", - "search toast", - "search route", - "keyboard secret", -] - -CHARACTER_REGISTRY = { - "young z": { - "id": "young z", - "displayLabel": "young z", - "aliases": ["Young Z", "young z", "young-z", "young_z", "young"], - "territoryColor": "#d8a95a", - "glow": "#f0b85c", - "avatar": "https://zainezq.com/assets/avatars/young-z.jpeg", - "fallbackAvatar": "https://zainezq.com/assets/avatars/z.jpeg", - "expressions": { - "default": "https://zainezq.com/assets/avatars/young-z.jpeg", - "nostalgic": "https://zainezq.com/assets/avatars/young-z.jpeg", - "playful": "https://zainezq.com/assets/avatars/young-z.jpeg", - }, - "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, - "observatory": {"territory": "nostalgic/playful memories", "shimmer": "soft"}, - "presenceTones": ["nostalgic", "funny", "hopeful"], - "presenceKeywords": ["childhood", "play", "desk", "crayon", "young", "small", "blanket", "memory"], - "symbol": "Y", - "motifs": ["crayon sun", "blanket cape", "childhood desk"], - "themes": ["childhood", "play", "memory", "safety"], - "affinities": ["z", "future z", "aphy", "lima"], - }, - "z": { - "id": "z", - "displayLabel": "z", - "aliases": ["Z", "z", "zaine"], - "territoryColor": "#d6c38a", - "glow": "#ead68e", - "avatar": "https://zainezq.com/assets/avatars/z.jpeg", - "fallbackAvatar": "https://zainezq.com/assets/avatars/z.jpeg", - "expressions": {"default": "https://zainezq.com/assets/avatars/z.jpeg", "calm": "https://zainezq.com/assets/avatars/z.jpeg"}, - "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, - "observatory": {"territory": "self/archive memories", "shimmer": "steady"}, - "presenceTones": ["warm", "soft", "hopeful"], - "presenceKeywords": ["website", "archive", "home", "self", "making", "return", "zaine"], - "symbol": "Z", - "motifs": ["archive", "website", "home"], - "themes": ["selfhood", "return", "making"], - "affinities": ["lima", "young z", "future z"], - }, - "aphy": { - "id": "aphy", - "displayLabel": "aphy", - "aliases": ["Aphy", "aphy", "aphy_bot", "aphy bot"], - "territoryColor": "#7fb089", - "glow": "#9dd3a6", - "avatar": "https://zainezq.com/assets/avatars/aphy.jpeg", - "fallbackAvatar": "https://zainezq.com/assets/avatars/z.jpeg", - "expressions": { - "default": "https://zainezq.com/assets/avatars/aphy.jpeg", - "system": "https://zainezq.com/assets/avatars/aphy.jpeg", - "amused": "https://zainezq.com/assets/avatars/aphy.jpeg", - }, - "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, - "observatory": {"territory": "system/symbolic areas", "shimmer": "diagnostic"}, - "presenceTones": ["funny", "strange", "hopeful"], - "presenceKeywords": ["system", "console", "diagnostic", "keyboard", "search", "terminal", "backup", "symbolic", "query"], - "symbol": "A", - "motifs": ["console", "diagnostic", "backup"], - "themes": ["humor", "systems", "care through tools"], - "affinities": ["z", "lima", "sensei chi"], - }, - "lima": { - "id": "lima", - "displayLabel": "lima", - "aliases": ["Lima", "lima"], - "territoryColor": "#d06b78", - "glow": "#f2c58b", - "avatar": "https://zainezq.com/assets/avatars/lima.jpg", - "fallbackAvatar": "https://zainezq.com/assets/avatars/z.jpeg", - "expressions": { - "default": "https://zainezq.com/assets/avatars/lima.jpg", - "calm": "https://zainezq.com/assets/avatars/lima.jpg", - "warm": "https://zainezq.com/assets/avatars/lima.jpg", - "protective": "https://zainezq.com/assets/avatars/lima.jpg", - }, - "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, - "observatory": {"territory": "warm/protective arcs", "shimmer": "warm"}, - "presenceTones": ["warm", "protective", "soft"], - "presenceKeywords": ["lima", "love", "warm", "kitchen", "light", "ring", "home", "protect", "eat"], - "symbol": "L", - "motifs": ["warmth", "kitchen light", "ring"], - "themes": ["love", "home", "grounding"], - "affinities": ["z", "aphy", "future z", "young z"], - }, - "sensei chi": { - "id": "sensei chi", - "displayLabel": "sensei chi", - "aliases": ["Sensei Chi", "sensei chi", "sensei-chi", "sensei_chi", "sensei"], - "territoryColor": "#75a9bd", - "glow": "#9ccddd", - "avatar": "https://zainezq.com/assets/avatars/sensei-chi.jpeg", - "fallbackAvatar": "https://zainezq.com/assets/avatars/z.jpeg", - "expressions": { - "default": "https://zainezq.com/assets/avatars/sensei-chi.jpeg", - "reflective": "https://zainezq.com/assets/avatars/sensei-chi.jpeg", - "wise": "https://zainezq.com/assets/avatars/sensei-chi.jpeg", - }, - "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, - "observatory": {"territory": "reflective areas", "shimmer": "quiet"}, - "presenceTones": ["wise", "melancholy", "soft"], - "presenceKeywords": ["reflection", "patience", "wisdom", "lesson", "tea", "garden", "quiet", "sensei"], - "symbol": "S", - "motifs": ["tea", "garden", "quiet lesson"], - "themes": ["reflection", "patience", "wisdom"], - "affinities": ["aphy", "future z"], - }, - "future z": { - "id": "future z", - "displayLabel": "future z", - "aliases": ["Future Z", "future z", "future-z", "future_z", "future"], - "territoryColor": "#a58ac9", - "glow": "#c2a4ee", - "avatar": "https://zainezq.com/assets/avatars/future-z.jpeg", - "fallbackAvatar": "https://zainezq.com/assets/avatars/z.jpeg", - "expressions": { - "default": "https://zainezq.com/assets/avatars/future-z.jpeg", - "temporal": "https://zainezq.com/assets/avatars/future-z.jpeg", - "reassuring": "https://zainezq.com/assets/avatars/future-z.jpeg", - }, - "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, - "observatory": {"territory": "temporal regions", "shimmer": "temporal"}, - "presenceTones": ["hopeful", "melancholy", "wise"], - "presenceKeywords": ["future", "time", "clock", "older", "tomorrow", "age", "reassurance", "continuity"], - "symbol": "F", - "motifs": ["clock", "age 40", "future log"], - "themes": ["time", "reassurance", "continuity"], - "affinities": ["z", "young z", "lima", "sensei chi"], - }, -} -HIDDEN_CHARACTERS = list(CHARACTER_REGISTRY) -HIDDEN_TONES = ["warm", "funny", "nostalgic", "wise", "strange", "soft", "hopeful", "protective", "melancholy"] -HIDDEN_RARITIES = ["common", "uncommon", "rare", "very rare", "seasonal", "timed"] -HIDDEN_STORY_MARKERS = ["public", "hidden", "rare", "emotional", "dream-like", "temporal"] -HIDDEN_DISCOVERY_STYLES = ["gradual", "direct", "hidden route", "character-led", "dream-like", "temporal"] -HIDDEN_LAYER_DEPTHS = [ - { - "id": "0", - "name": "Surface Reality", - "meaning": "Normal visible website content, visible warmth, and ordinary interactions.", - }, - { - "id": "1", - "name": "Hidden Personality", - "meaning": "Small hidden jokes, hover text, tiny discoveries, and recurring symbols.", - }, - { - "id": "2", - "name": "Memory Layer", - "meaning": "young z memories, lima notes, nostalgia, and emotional fragments.", - }, - { - "id": "3", - "name": "Reflection Layer", - "meaning": "sensei chi philosophy, aphy conversations, and introspection.", - }, - { - "id": "4", - "name": "Time Layer", - "meaning": "future z logs, time anomalies, long-term revisits, and future/past echoes.", - }, - { - "id": "5", - "name": "Core Layer", - "meaning": "Rare deeply emotional truths found by patient exploration.", - }, -] - - -def slugify(value: str) -> str: - slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") - return slug or "untitled" - - -def normalise_tags(value: Any) -> list[str]: - if isinstance(value, str): - raw = re.split(r"[,:\s]+", value) - elif isinstance(value, list): - raw = [str(item) for item in value] - else: - raw = [] - tags = [] - for tag in raw: - if not tag.strip(): - continue - clean = slugify(tag) - if clean and clean not in tags: - tags.append(clean) - return tags - - -def org_date(dt: datetime) -> str: - return dt.strftime("<%Y-%m-%d %a %H:%M>") - - -def parse_org_datetime(value: str | None) -> datetime | None: - if not value: - return None - match = re.search(r"<(\d{4})-(\d{2})-(\d{2})(?:\s+\w+)?(?:\s+(\d{2}):(\d{2}))?>", value) - if not match: - return None - year, month, day, hour, minute = match.groups() - return datetime( - int(year), - int(month), - int(day), - int(hour or 12), - int(minute or 0), - ) - - -def html_escape(value: str) -> str: - return ( - value.replace("&", "&") - .replace("<", "<") - .replace(">", ">") - .replace('"', """) - ) - - -@dataclass -class OrgPage: - path: str - page_type: str - title: str - slug: str - tags: list[str] - content: str - date: str - comments: bool - options: str - wip: str | None = None - - -@dataclass -class ContentPage: - path: str - page_type: str - title: str - slug: str - tags: list[str] - content: str - date: str - comments: bool - options: str - format: str - wip: str | None = None - - -@dataclass -class BuildJob: - id: int - path: str - title: str - queued_at: float = field(default_factory=time.time) - started_at: float | None = None - finished_at: float | None = None - ok: bool | None = None - message: str = "Queued" - log: str = "" - - @property - def status(self) -> str: - if self.finished_at is not None: - return "done" if self.ok else "failed" - if self.started_at is not None: - return "running" - return "queued" - - def to_dict(self, include_log: bool = False) -> dict[str, Any]: - data = { - "id": self.id, - "path": self.path, - "title": self.title, - "queuedAt": self.queued_at, - "startedAt": self.started_at, - "finishedAt": self.finished_at, - "ok": self.ok, - "status": self.status, - "message": self.message, - } - if include_log: - data["log"] = self.log[-12000:] - return data - - -class BuildQueue: - def __init__(self) -> None: - self._lock = threading.Lock() - self._next_id = 1 - self._pending: list[BuildJob] = [] - self._current: BuildJob | None = None - self._recent: list[BuildJob] = [] - self._worker: threading.Thread | None = None - - def snapshot(self) -> dict[str, Any]: - with self._lock: - current = self._current.to_dict(include_log=True) if self._current else None - recent = [job.to_dict(include_log=True) for job in self._recent[-10:]] - pending = [job.to_dict() for job in self._pending] - latest = current or (recent[-1] if recent else None) - message = latest["message"] if latest else "No builds have run yet." - return { - "running": current is not None, - "queued": len(pending), - "message": message, - "current": current, - "pending": pending, - "recent": recent, - "log": latest.get("log", "") if latest else "", - } - - def enqueue(self, path: str, title: str) -> BuildJob: - with self._lock: - job = BuildJob(self._next_id, path, title) - self._next_id += 1 - self._pending.append(job) - if self._worker is None or not self._worker.is_alive(): - self._worker = threading.Thread(target=self._run_worker, daemon=True) - self._worker.start() - return job - - def _run_worker(self) -> None: - while True: - with self._lock: - if not self._pending: - self._current = None - return - job = self._pending.pop(0) - job.started_at = time.time() - job.message = "Publishing site and search index." - self._current = job - def append_log(chunk: str) -> None: - with self._lock: - job.log = (job.log + chunk)[-200000:] - - ok, message, log = run_build_commands(append_log) - with self._lock: - job.finished_at = time.time() - job.ok = ok - job.message = message - job.log = log[-200000:] - self._recent.append(job) - self._recent = self._recent[-20:] - self._current = None - - -BUILD_QUEUE = BuildQueue() - - -def safe_relative_path(path: str) -> Path: - rel = Path(path) - if rel.is_absolute() or ".." in rel.parts: - raise ValueError("Path must stay inside this repository.") - full = (ROOT / rel).resolve() - if not full.is_relative_to(ROOT): - raise ValueError("Path must stay inside this repository.") - if full.name in GENERATED_CONTENT_NAMES or "sync-conflict" in full.name: - raise ValueError("Generated and sync-conflict files are not editable here.") - rel_parts = full.relative_to(ROOT).parts - if any(part in EXCLUDED_CONTENT_DIR_NAMES for part in rel_parts): - raise ValueError("This path is outside the editable content folders.") - if full.suffix == ".org": - return full - if full.suffix == ".md" and full.is_relative_to(LIMA_DIR): - return full - raise ValueError("Only .org content files and .md files under lima can be edited here.") - - -def safe_target_path(path: str, slug: str, page_type: str) -> Path: - candidate = path.strip() - if not candidate: - raise ValueError("Path is required.") - default_ext = ".md" if page_type == "lima" else ".org" - if candidate.endswith("/"): - candidate = f"{candidate}{slug}{default_ext}" - elif not Path(candidate).suffix: - candidate = f"{candidate}{default_ext}" - return safe_relative_path(candidate) - - -def markdown_title(content: str, fallback: str) -> str: - for line in content.splitlines(): - match = re.match(r"^#{1,6}\s+(.+?)\s*$", line) - if match: - return match.group(1).strip() - return fallback.replace("-", " ").replace("_", " ").title() - - -def read_markdown_page(path: Path) -> ContentPage: - content = path.read_text(encoding="utf-8") - rel = path.relative_to(ROOT).as_posix() - title = markdown_title(content, path.stem) - return ContentPage( - path=rel, - page_type="lima", - title=title, - slug=path.stem, - tags=[], - content=content, - date="", - comments=True, - options="", - format="markdown", - ) - - -def read_page(path: Path) -> ContentPage: - if path.suffix == ".md" and path.is_relative_to(LIMA_DIR): - return read_markdown_page(path) - text = path.read_text(encoding="utf-8") - meta: dict[str, str] = {} - body_lines: list[str] = [] - in_header = True - for line in text.splitlines(): - if in_header and line.startswith("#+"): - key, _, value = line[2:].partition(":") - meta[key.strip().upper()] = value.strip() - else: - in_header = False - body_lines.append(line) - rel = path.relative_to(ROOT).as_posix() - if path.is_relative_to(BLOGS_DIR): - page_type = "blog" - elif path.is_relative_to(POSTS_DIR): - page_type = "post" - else: - page_type = "page" - slug = meta.get("SLUG") or path.stem - tags = normalise_tags(meta.get("FILETAGS", "")) - return ContentPage( - path=rel, - page_type=page_type, - title=meta.get("TITLE", path.stem), - slug=slug, - tags=tags, - content="\n".join(body_lines).lstrip("\n"), - date=meta.get("DATE", org_date(datetime.fromtimestamp(path.stat().st_mtime))), - comments=meta.get("COMMENTS", "t").lower() == "t", - options=meta.get("OPTIONS", "num:nil"), - format="org", - wip=meta.get("WIP"), - ) - - -def page_to_dict(page: ContentPage) -> dict[str, Any]: - return { - "path": page.path, - "pageType": page.page_type, - "title": page.title, - "slug": page.slug, - "tags": page.tags, - "content": page.content, - "date": page.date, - "comments": page.comments, - "options": page.options, - "format": page.format, - "wip": page.wip or "", - } - - -def server_diagnostics() -> dict[str, Any]: - pages = list_pages() - try: - cwd = Path.cwd().as_posix() - except OSError as exc: - cwd = f"" - return { - "root": ROOT.as_posix(), - "cwd": cwd, - "executable": sys.executable, - "pid": os.getpid(), - "pageCount": len(pages), - "firstPage": pages[0]["path"] if pages else "", - } - - -def list_pages() -> list[dict[str, Any]]: - pages = [] - org_paths = [] - if ROOT.exists(): - try: - for path in ROOT.rglob("*.org"): - try: - rel_parts = path.relative_to(ROOT).parts - except ValueError: - continue - if any(part in EXCLUDED_CONTENT_DIR_NAMES for part in rel_parts): - continue - org_paths.append(path) - except OSError: - org_paths = [] - try: - md_paths = list(LIMA_DIR.rglob("*.md")) if LIMA_DIR.exists() else [] - except OSError: - md_paths = [] - for path in sorted(org_paths + md_paths): - if path.name in GENERATED_CONTENT_NAMES or "sync-conflict" in path.name: - continue - try: - page = read_page(path) - parsed = parse_org_datetime(page.date) - timestamp = parsed.timestamp() if parsed else path.stat().st_mtime - except (OSError, UnicodeDecodeError, ValueError): - continue - pages.append( - { - "path": page.path, - "pageType": page.page_type, - "title": page.title, - "slug": page.slug, - "tags": page.tags, - "date": page.date, - "format": page.format, - "timestamp": timestamp, - } - ) - return sorted(pages, key=lambda item: item["timestamp"], reverse=True) - - -def target_path(data: dict[str, Any], existing_path: str | None) -> Path: - if existing_path: - return safe_relative_path(existing_path) - title = str(data.get("title") or "").strip() - slug = slugify(str(data.get("slug") or title)) - page_type = str(data.get("pageType") or "blog") - explicit_path = str(data.get("targetPath") or "").strip() - if explicit_path: - return safe_target_path(explicit_path, slug, page_type) - if page_type == "blog": - dt = parse_org_datetime(str(data.get("date") or "")) or datetime.now() - folder = BLOGS_DIR / str(dt.year) / f"{dt.month:02d}-{MONTH_NAMES[dt.month - 1]}" - return folder / f"{slug}.org" - if page_type == "post": - raw_section = str(data.get("section") or "").strip() - section = slugify(raw_section) if raw_section else "" - folder = POSTS_DIR / section if section else POSTS_DIR - return folder / f"{slug}.org" - if page_type == "lima": - return LIMA_DIR / f"{slug}.md" - if page_type == "page": - return ROOT / f"{slug}.org" - raise ValueError("pageType must be blog, post, page, or lima.") - - -def render_markdown(data: dict[str, Any]) -> str: - content = str(data.get("content") or "").replace("\r\n", "\n").strip() - title = str(data.get("title") or "").strip() - if not title: - raise ValueError("Title is required.") - content = re.sub( - r']*>\s*]*\bsrc="([^"]+)"[^>]*\balt="([^"]*)"[^>]*>\s*', - lambda match: f"![{match.group(2)}]({match.group(1)})", - content, - flags=re.IGNORECASE, - ) - if content: - if re.search(r"^#{1,6}[^\S\r\n]+.+?[^\S\r\n]*$", content, flags=re.MULTILINE): - content = re.sub( - r"^#{1,6}[^\S\r\n]+.+?[^\S\r\n]*$", - f"# {title}", - content, - count=1, - flags=re.MULTILINE, - ) - else: - content = f"# {title}\n\n{content}" - return content + "\n" - return f"# {title}\n" - - -def render_org(data: dict[str, Any], previous: ContentPage | None) -> str: - title = str(data.get("title") or "").strip() - if not title: - raise ValueError("Title is required.") - slug = slugify(str(data.get("slug") or title)) - tags = normalise_tags(data.get("tags", [])) - content = str(data.get("content") or "").replace("\r\n", "\n").strip() - date = str(data.get("date") or "").strip() - if not parse_org_datetime(date): - date = previous.date if previous else org_date(datetime.now()) - options = str(data.get("options") or (previous.options if previous else "num:nil")).strip() - comments = bool(data.get("comments", True)) - lines = [ - f"#+TITLE: {title}", - f"#+OPTIONS: {options}", - f"#+DATE: {date}", - f"#+filetags: {''.join(f':{tag}' for tag in tags)}:", - ] - wip = str(data.get("wip") or (previous.wip if previous else "") or "").strip() - if wip: - lines.append(f"#+WIP: {wip}") - lines.extend( - [ - f"#+COMMENTS: {'t' if comments else ''}", - f"#+SLUG: {slug}", - "", - content, - "", - ] - ) - return "\n".join(lines) - - -def save_page(data: dict[str, Any]) -> dict[str, Any]: - existing_path = data.get("path") or None - target = target_path(data, str(existing_path) if existing_path else None) - previous = read_page(target) if target.exists() else None - if not target.parent.exists(): - target.parent.mkdir(parents=True) - if target.exists() and not existing_path: - raise ValueError(f"{target.relative_to(ROOT)} already exists.") - if target.suffix == ".md": - target.write_text(render_markdown(data), encoding="utf-8") - else: - target.write_text(render_org(data, previous), encoding="utf-8") - return page_to_dict(read_page(target)) - - -def image_dimensions(payload: bytes, ext: str) -> tuple[int, int] | None: - if ext == ".png" and payload.startswith(b"\x89PNG\r\n\x1a\n") and len(payload) >= 24: - width, height = struct.unpack(">II", payload[16:24]) - return width, height - if ext == ".gif" and payload[:6] in {b"GIF87a", b"GIF89a"} and len(payload) >= 10: - width, height = struct.unpack(" len(payload): - break - size = int.from_bytes(payload[i:i + 2], "big") - if size < 2: - break - if marker in {0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF}: - if i + 7 <= len(payload): - height = int.from_bytes(payload[i + 3:i + 5], "big") - width = int.from_bytes(payload[i + 5:i + 7], "big") - return width, height - break - i += size - return None - - -def relative_asset_path(page_path: str, asset_path: str) -> str: - page = safe_relative_path(page_path) if page_path else ROOT / "index.org" - page_rel = page.relative_to(ROOT).as_posix() - page_output_dir = posixpath.dirname(page_rel) - return posixpath.relpath(asset_path, page_output_dir or ".") - - -def gallery_image_html(filename: str, asset_path: str, page_path: str, payload: bytes, ext: str) -> str: - absolute_url = f"https://zainezq.com/{asset_path}" - relative_url = relative_asset_path(page_path, asset_path) - dims = image_dimensions(payload, ext) - width, height = dims if dims else (1920, 1080) - alt = html_escape(filename) - return ( - f'' - f'{alt}' - ) - - -def attachment_image_dir(page_path: str) -> Path: - page = safe_relative_path(page_path) if page_path else ROOT / "index.org" - if page.suffix == ".md" and page.is_relative_to(LIMA_DIR): - return HZONE_ASSETS_DIR - if page.is_relative_to(POSTS_DIR): - rel = page.relative_to(POSTS_DIR) - section = rel.parts[0] if len(rel.parts) > 1 else "posts" - return IMAGE_ASSETS_DIR / slugify(section) - if page.is_relative_to(BLOGS_DIR): - return IMAGE_ASSETS_DIR / "blogs" - rel = page.relative_to(ROOT) - if len(rel.parts) > 1: - return IMAGE_ASSETS_DIR / slugify(rel.parts[0]) - return IMAGE_ASSETS_DIR / "pages" - - -def save_upload(filename: str, payload: bytes, page_path: str = "") -> dict[str, str]: - original = Path(filename or "attachment").name - ext = Path(original).suffix.lower() - if ext not in ALLOWED_UPLOAD_EXTENSIONS: - raise ValueError("Only common image files can be uploaded.") - now = datetime.now() - target_dir = attachment_image_dir(page_path) - target_dir.mkdir(parents=True, exist_ok=True) - stem = slugify(Path(original).stem) - prefix = "" if re.match(r"^\d{4}-\d{2}-\d{2}-", stem) else f"{now.strftime('%Y-%m-%d')}-" - target = target_dir / f"{prefix}{stem}{ext}" - counter = 2 - while target.exists(): - target = target_dir / f"{prefix}{stem}-{counter}{ext}" - counter += 1 - target.write_bytes(payload) - rel = target.relative_to(ROOT).as_posix() - absolute_url = f"https://zainezq.com/{rel}" - relative_url = relative_asset_path(page_path, rel) - is_markdown = page_path.endswith(".md") - insert_text = f"![{target.name}]({absolute_url})" if is_markdown else f"[[{relative_url}]]" - return { - "url": absolute_url, - "relativeUrl": relative_url, - "path": rel, - "markdown": insert_text, - "insertText": insert_text, - "filename": target.name, - } - - -def find_js_const_literal(source: str, name: str) -> str: - marker = f"const {name} =" - start = source.find(marker) - if start == -1: - raise ValueError(f"Could not find const {name}.") - value_start = source.find("=", start) + 1 - while value_start < len(source) and source[value_start].isspace(): - value_start += 1 - opener = source[value_start] - pairs = {"[": "]", "{": "}"} - if opener not in pairs: - raise ValueError(f"const {name} is not an array or object.") - closer = pairs[opener] - depth = 0 - in_string = False - escape = False - for index in range(value_start, len(source)): - char = source[index] - if in_string: - if escape: - escape = False - elif char == "\\": - escape = True - elif char == '"': - in_string = False - continue - if char == '"': - in_string = True - elif char == opener: - depth += 1 - elif char == closer: - depth -= 1 - if depth == 0: - return source[value_start:index + 1] - raise ValueError(f"Could not parse const {name}.") - - -def js_literal_to_json(value: str) -> Any: - cleaned = re.sub(r"//.*", "", value) - cleaned = re.sub(r"(/\*.*?\*/)", "", cleaned, flags=re.DOTALL) - cleaned = re.sub(r"([{\[,]\s*)([A-Za-z_$][\w$]*)\s*:", r'\1"\2":', cleaned) - cleaned = re.sub(r",(\s*[\]}])", r"\1", cleaned) - return json.loads(cleaned) - - -def character_alias_lookup() -> dict[str, str]: - aliases = {} - for canonical, config in CHARACTER_REGISTRY.items(): - aliases[canonical] = canonical - for alias in config["aliases"]: - aliases[slugify(str(alias)).replace("-", " ")] = canonical - aliases[str(alias).strip().lower()] = canonical - return aliases - - -def normalize_character_name(value: Any) -> str | None: - raw = str(value or "").strip() - if not raw: - return None - simplified = slugify(raw).replace("-", " ") - return character_alias_lookup().get(raw.lower()) or character_alias_lookup().get(simplified) - - -def normalize_character_list(values: Any) -> list[str]: - if isinstance(values, str): - raw_values = re.split(r"[,/]", values) - elif isinstance(values, list): - raw_values = values - else: - raw_values = [] - normalized = [] - for item in raw_values: - canonical = normalize_character_name(item) - if canonical and canonical not in normalized: - normalized.append(canonical) - return normalized - - -def normalize_character_text_refs(value: Any) -> Any: - if isinstance(value, list): - return [normalize_character_text_refs(item) for item in value] - if isinstance(value, dict): - return {key: normalize_character_text_refs(item) for key, item in value.items()} - if not isinstance(value, str): - return value - text = value - replacements = [] - for canonical, config in CHARACTER_REGISTRY.items(): - for alias in config["aliases"]: - if alias == canonical: - continue - if alias.lower() in {"young", "future", "sensei"}: - continue - replacements.append((alias, canonical)) - replacements.sort(key=lambda item: len(item[0]), reverse=True) - for alias, canonical in replacements: - pattern = r"(? list[str]: - found = [] - normalized_text = normalize_character_text_refs(text).lower() - for character in HIDDEN_CHARACTERS: - if re.search(r"(? str: - return f"{slugify(content_type)}-{index + 1:03d}-{slugify(title)[:42]}" - - -def hidden_title(content_type: str, content: str, index: int) -> str: - text = re.sub(r"\s+", " ", content).strip() - if not text: - return f"{content_type.title()} {index + 1}" - return text[:58] + ("..." if len(text) > 58 else "") - - -def make_hidden_entry(content_type: str, content: str, index: int, **extra: Any) -> dict[str, Any]: - now_iso = datetime.now().date().isoformat() - title = str(extra.pop("title", "") or hidden_title(content_type, content, index)) - characters = extra.pop("characters", None) or detect_hidden_characters(f"{title} {content}") - tags = extra.pop("tags", None) or [slugify(character) for character in characters] - entry = { - "id": hidden_entry_id(content_type, index, title), - "type": content_type, - "title": title, - "content": content, - "characters": characters, - "emotionalTone": extra.pop("emotionalTone", "warm"), - "rarity": extra.pop("rarity", "common"), - "triggerConditions": extra.pop("triggerConditions", ""), - "tags": tags, - "category": extra.pop("category", content_type), - "pageLocation": extra.pop("pageLocation", ""), - "familyLayer": extra.pop("familyLayer", ""), - "enabled": extra.pop("enabled", True), - "createdDate": extra.pop("createdDate", now_iso), - "modifiedDate": extra.pop("modifiedDate", now_iso), - "notes": extra.pop("notes", ""), - "audioSettings": extra.pop("audioSettings", ""), - "animationTrigger": extra.pop("animationTrigger", ""), - "cssClassHooks": extra.pop("cssClassHooks", ""), - "chainReferences": extra.pop("chainReferences", []), - "continuationLinks": extra.pop("continuationLinks", []), - "emotionalRole": extra.pop("emotionalRole", ""), - "discoveryDifficulty": extra.pop("discoveryDifficulty", "gentle"), - "mysteryLevel": extra.pop("mysteryLevel", "quiet"), - "resonanceScore": extra.pop("resonanceScore", 3), - "symbols": extra.pop("symbols", []), - "narrativeArcs": extra.pop("narrativeArcs", []), - "parentLinks": extra.pop("parentLinks", []), - "childLinks": extra.pop("childLinks", []), - "echoes": extra.pop("echoes", []), - "mirroredEntries": extra.pop("mirroredEntries", []), - "thematicLinks": extra.pop("thematicLinks", []), - "symbolicLinks": extra.pop("symbolicLinks", []), - "triggerLinks": extra.pop("triggerLinks", []), - } - entry.update(extra) - return entry - - -def migrate_hidden_entries_from_js() -> list[dict[str, Any]]: - source = HIDDEN_DETAILS_JS.read_text(encoding="utf-8") - family_layers = js_literal_to_json(find_js_const_literal(source, "familyLayers")) - details = js_literal_to_json(find_js_const_literal(source, "details")) - poems = js_literal_to_json(find_js_const_literal(source, "poems")) - greetings = js_literal_to_json(find_js_const_literal(source, "greetings")) - night_messages = js_literal_to_json(find_js_const_literal(source, "nightMessages")) - lore = js_literal_to_json(find_js_const_literal(source, "lore")) - search_toasts = js_literal_to_json(find_js_const_literal(source, "SEARCH_TOASTS")) - search_routes = js_literal_to_json(find_js_const_literal(source, "SEARCH_ROUTES")) - keyboard_secrets = js_literal_to_json(find_js_const_literal(source, "KEYBOARD_SECRETS")) - long_keyboard_secrets = js_literal_to_json(find_js_const_literal(source, "LONG_KEYBOARD_SECRETS")) - - entries: list[dict[str, Any]] = [] - add = entries.append - for index, item in enumerate(family_layers): - add(make_hidden_entry("family layer", item, index, familyLayer=str(index), category="Family Layer Index")) - for index, item in enumerate(details): - add(make_hidden_entry("hidden tooltip", item, index, category="footer and whispers")) - for index, item in enumerate(poems): - add(make_hidden_entry("poem", item, index, category="poems", emotionalTone="soft")) - for index, item in enumerate(greetings): - add(make_hidden_entry("loading screen message", item, index, category="homepage greeting")) - for index, item in enumerate(night_messages): - add(make_hidden_entry("rare event", item, index, category="late night", rarity="timed", triggerConditions="hour >= 22 or hour < 5")) - for key, content_type in [ - ("quotes", "quote"), - ("journals", "journal entry"), - ("warnings", "fake error message"), - ("dreams", "dream sequence"), - ("cassettes", "terminal log"), - ("fakeUsers", "guestbook entry"), - ("homepageTakeovers", "rare event"), - ]: - for index, item in enumerate(lore.get(key, [])): - add(make_hidden_entry(content_type, item, index, category=key, rarity="rare" if key in {"warnings", "dreams", "homepageTakeovers"} else "common")) - for index, item in enumerate(lore.get("conversations", [])): - title = " / ".join(str(part) for part in item[::2]) or f"Conversation {index + 1}" - add(make_hidden_entry("hidden conversation", "\n".join(str(part) for part in item), index, title=title, category="conversations", dialogue=item)) - for index, (season, item) in enumerate((lore.get("seasonal", {}) or {}).items()): - add(make_hidden_entry("seasonal event", item, index, title=f"{season.title()} note", category="seasonal", rarity="seasonal", triggerConditions=f"season is {season}", season=season)) - for index, item in enumerate(lore.get("roomLinks", [])): - href, label = item - add(make_hidden_entry("secret interaction", label, index, title=label, category="room links", pageLocation=href, triggerConditions="secret link injected into page")) - for index, (query, message) in enumerate(search_toasts.items()): - add(make_hidden_entry("search toast", message, index, title=f"Search: {query}", category="search", triggerConditions=query, query=query)) - for index, (query, route) in enumerate(search_routes.items()): - add(make_hidden_entry("search route", route, index, title=f"Route: {query}", category="search", pageLocation=route, triggerConditions=query, query=query)) - for index, item in enumerate(keyboard_secrets + long_keyboard_secrets): - content = item.get("message") or item.get("route") or ("play tiny aphy song" if item.get("song") else "") - add(make_hidden_entry("keyboard secret", content, index, title=f"Keyboard: {item.get('phrase')}", category="keyboard", pageLocation=item.get("route", ""), triggerConditions=item.get("phrase", ""), keyboard=item)) - return entries - - -def normalize_hidden_entry(raw: dict[str, Any], existing: dict[str, Any] | None = None) -> dict[str, Any]: - today = datetime.now().date().isoformat() - entry = existing.copy() if existing else {} - entry.update(raw) - title = str(normalize_character_text_refs(entry.get("title") or "")).strip() - content = str(normalize_character_text_refs(entry.get("content") or "")).replace("\r\n", "\n") - content_type = str(entry.get("type") or "quote").strip() - content_type = str(normalize_character_text_refs(content_type)) - if content_type not in HIDDEN_CONTENT_TYPES: - raise ValueError(f"Unsupported hidden content type: {content_type}") - if not title: - raise ValueError("Every hidden entry needs a title.") - if not content and content_type not in {"search route"}: - raise ValueError(f"{title} needs content.") - entry["id"] = slugify(str(entry.get("id") or title)) - entry["title"] = title - entry["type"] = content_type - entry["content"] = content - detected_characters = detect_hidden_characters(" ".join([ - title, - content, - str(entry.get("triggerConditions") or ""), - str(entry.get("notes") or ""), - str(entry.get("category") or ""), - ])) - entry["characters"] = normalize_character_list(entry.get("characters", [])) - for character in detected_characters: - if character not in entry["characters"]: - entry["characters"].append(character) - if not entry["characters"]: - entry["characters"] = ["z"] - entry["emotionalTone"] = str(entry.get("emotionalTone") or "warm") - entry["rarity"] = str(entry.get("rarity") or "common") - entry["triggerConditions"] = str(normalize_character_text_refs(entry.get("triggerConditions") or "")) - entry["tags"] = normalise_tags(entry.get("tags", [])) - entry["category"] = str(normalize_character_text_refs(entry.get("category") or content_type)) - entry["pageLocation"] = str(entry.get("pageLocation") or "") - entry["familyLayer"] = str(entry.get("familyLayer") or "") - entry["enabled"] = bool(entry.get("enabled", True)) - entry["createdDate"] = str(entry.get("createdDate") or today) - entry["modifiedDate"] = today - entry["notes"] = str(normalize_character_text_refs(entry.get("notes") or "")) - entry["audioSettings"] = str(entry.get("audioSettings") or "") - entry["animationTrigger"] = str(entry.get("animationTrigger") or "") - entry["cssClassHooks"] = str(entry.get("cssClassHooks") or "") - entry["chainReferences"] = [str(item).strip() for item in entry.get("chainReferences", []) if str(item).strip()] - entry["continuationLinks"] = [str(item).strip() for item in entry.get("continuationLinks", []) if str(item).strip()] - entry["emotionalRole"] = str(entry.get("emotionalRole") or "") - entry["discoveryDifficulty"] = str(entry.get("discoveryDifficulty") or "gentle") - entry["mysteryLevel"] = str(entry.get("mysteryLevel") or "quiet") - try: - entry["resonanceScore"] = max(1, min(10, int(entry.get("resonanceScore") or 3))) - except (TypeError, ValueError): - entry["resonanceScore"] = 3 - for key in [ - "symbols", - "narrativeArcs", - "parentLinks", - "childLinks", - "echoes", - "mirroredEntries", - "thematicLinks", - "symbolicLinks", - "triggerLinks", - ]: - entry[key] = [str(normalize_character_text_refs(item)).strip() for item in entry.get(key, []) if str(item).strip()] - for key in ["dialogue", "keyboard"]: - if key in entry: - entry[key] = normalize_character_text_refs(entry[key]) - return entry - - -def normalize_hidden_story(raw: dict[str, Any], entries: list[dict[str, Any]] | None = None, existing: dict[str, Any] | None = None) -> dict[str, Any]: - today = datetime.now().date().isoformat() - story = existing.copy() if existing else {} - story.update(raw) - title = str(normalize_character_text_refs(story.get("title") or "")).strip() - if not title: - raise ValueError("Every story needs a title.") - ids = {entry["id"] for entry in entries or []} - raw_nodes = story.get("nodes", []) - if isinstance(raw_nodes, str): - raw_nodes = re.split(r"[,:\s]+", raw_nodes) - nodes = [] - for item in raw_nodes: - node_id = str(item).strip() - if node_id and node_id not in nodes and (not ids or node_id in ids): - nodes.append(node_id) - characters = normalize_character_list(story.get("characters", [])) - symbols = [str(normalize_character_text_refs(item)).strip() for item in story.get("symbols", []) if str(item).strip()] - markers = [slugify(str(item)) for item in story.get("markers", []) if str(item).strip()] - layer_affinity = [] - for item in story.get("layerAffinity", []): - match = re.search(r"[0-5]", str(item)) - if match and int(match.group(0)) not in layer_affinity: - layer_affinity.append(int(match.group(0))) - if not layer_affinity and nodes: - by_id = {entry["id"]: entry for entry in entries or []} - layer_affinity = sorted({ - int(match.group(0)) - for node in nodes - if (match := re.search(r"[0-5]", str(by_id.get(node, {}).get("familyLayer", "")))) - }) - story_id = slugify(str(story.get("id") or f"story-{title}")) - if not story_id.startswith("story-"): - story_id = f"story-{story_id}" - return { - "id": story_id, - "title": title, - "description": str(normalize_character_text_refs(story.get("description") or "")), - "tone": str(story.get("tone") or "warm"), - "characters": characters, - "symbols": symbols, - "nodes": nodes, - "discoveryStyle": str(story.get("discoveryStyle") or "gradual"), - "layerAffinity": layer_affinity, - "unlockConditions": [str(normalize_character_text_refs(item)).strip() for item in story.get("unlockConditions", []) if str(item).strip()], - "hidden": bool(story.get("hidden", False)), - "markers": markers, - "createdDate": str(story.get("createdDate") or today), - "modifiedDate": today, - } - - -def migrate_hidden_stories(entries: list[dict[str, Any]]) -> list[dict[str, Any]]: - story_groups: dict[str, dict[str, Any]] = {} - - def add(group: str, entry: dict[str, Any], source: str) -> None: - if not group: - return - key = slugify(group) - if key not in story_groups: - story_groups[key] = { - "id": f"story-{key}", - "title": group.replace("-", " "), - "description": f"Migrated from old {source} relationships into a calmer story path.", - "tone": entry.get("emotionalTone") or "warm", - "characters": [], - "symbols": [], - "nodes": [], - "discoveryStyle": "gradual", - "layerAffinity": [], - "unlockConditions": [], - "hidden": False, - "markers": ["emotional"], - } - story = story_groups[key] - if entry["id"] not in story["nodes"]: - story["nodes"].append(entry["id"]) - for character in entry.get("characters", []): - if character not in story["characters"]: - story["characters"].append(character) - for symbol in entry.get("symbols", []): - if symbol not in story["symbols"]: - story["symbols"].append(symbol) - layer = re.search(r"[0-5]", str(entry.get("familyLayer", ""))) - if layer and int(layer.group(0)) not in story["layerAffinity"]: - story["layerAffinity"].append(int(layer.group(0))) - - by_id = {entry["id"]: entry for entry in entries} - for entry in entries: - for arc in entry.get("narrativeArcs", []): - add(str(arc), entry, "arc") - for symbol in entry.get("symbols", []): - add(str(symbol), entry, "symbol") - for target in entry.get("continuationLinks", []) + entry.get("chainReferences", []): - if target in by_id: - name = (entry.get("narrativeArcs") or entry.get("symbols") or [entry.get("emotionalTone") or "quiet return"])[0] - add(str(name), entry, "continuation") - add(str(name), by_id[target], "continuation") - if not story_groups: - for character in HIDDEN_CHARACTERS: - character_entries = [entry for entry in entries if character in entry.get("characters", [])][:12] - if character_entries: - for entry in character_entries: - add(f"{character} stories", entry, "character territory") - return [normalize_hidden_story(story, entries) for story in story_groups.values() if len(story.get("nodes", [])) >= 2][:36] - - -def load_hidden_store() -> dict[str, Any]: - migrated = False - if HIDDEN_CONTENT_JSON.exists(): - data = json.loads(HIDDEN_CONTENT_JSON.read_text(encoding="utf-8")) - entries = data.get("entries", []) - stories = data.get("stories", []) - else: - entries = migrate_hidden_entries_from_js() - stories = [] - migrated = True - data = { - "schemaVersion": 2, - "generatedFrom": "assets/scripts/hidden-details.js", - "generatedAt": datetime.now().isoformat(timespec="seconds"), - "entries": entries, - "stories": stories, - } - normalized = [normalize_hidden_entry(entry, entry) for entry in entries] - normalized_stories = [normalize_hidden_story(story, normalized, story) for story in stories] - migrated_relationships = False - if not normalized_stories: - normalized_stories = migrate_hidden_stories(normalized) - migrated_relationships = bool(normalized_stories) - return { - "schemaVersion": 2, - "source": HIDDEN_DETAILS_JS.relative_to(ROOT).as_posix(), - "contentPath": HIDDEN_CONTENT_JSON.relative_to(ROOT).as_posix(), - "migratedFromJs": migrated, - "migratedRelationshipsToStories": migrated_relationships, - "types": HIDDEN_CONTENT_TYPES, - "characters": HIDDEN_CHARACTERS, - "characterRegistry": CHARACTER_REGISTRY, - "validation": validate_hidden_integrity(normalized, normalized_stories), - "tones": HIDDEN_TONES, - "rarities": HIDDEN_RARITIES, - "storyMarkers": HIDDEN_STORY_MARKERS, - "discoveryStyles": HIDDEN_DISCOVERY_STYLES, - "layers": HIDDEN_LAYER_DEPTHS, - "entries": normalized, - "stories": normalized_stories, - "recommendations": hidden_architecture_recommendations(), - } - - -def validate_hidden_integrity(entries: list[dict[str, Any]], stories: list[dict[str, Any]] | None = None) -> dict[str, Any]: - ids = {entry["id"] for entry in entries} - unknown_characters = [] - orphan_nodes = [] - stale_links = [] - stale_story_nodes = [] - for entry in entries: - characters = entry.get("characters", []) - if not characters: - orphan_nodes.append(entry["id"]) - for character in characters: - if character not in CHARACTER_REGISTRY: - unknown_characters.append({"entry": entry["id"], "character": character}) - for key in ["chainReferences", "continuationLinks", "parentLinks", "childLinks", "echoes", "mirroredEntries", "thematicLinks", "symbolicLinks", "triggerLinks"]: - for target in entry.get(key, []): - if target and target not in ids and not str(target).startswith("/"): - stale_links.append({"entry": entry["id"], "field": key, "target": target}) - for story in stories or []: - for character in story.get("characters", []): - if character not in CHARACTER_REGISTRY: - unknown_characters.append({"story": story["id"], "character": character}) - for node_id in story.get("nodes", []): - if node_id not in ids: - stale_story_nodes.append({"story": story["id"], "target": node_id}) - return { - "ok": not unknown_characters and not orphan_nodes and not stale_links and not stale_story_nodes, - "unknownCharacters": unknown_characters[:50], - "orphanNodes": orphan_nodes[:50], - "staleLinks": stale_links[:50], - "staleStoryNodes": stale_story_nodes[:50], - "summary": { - "unknownCharacterCount": len(unknown_characters), - "orphanNodeCount": len(orphan_nodes), - "staleLinkCount": len(stale_links), - "staleStoryNodeCount": len(stale_story_nodes), - }, - "repairPolicy": "Unknown aliases are normalized through the registry. Empty character lists are repaired to z. Old relationship links are retained as legacy data, but new authoring should happen through stories.", - } - - -def hidden_architecture_recommendations() -> dict[str, Any]: - return { - "storage": "Use assets/content/hidden-details.json as the friendly source of truth for memories and first-class stories, then regenerate the editable constants in assets/scripts/hidden-details.js.", - "backups": "Every save writes timestamped backups for both JSON and JS under backups/hidden-details/.", - "versioning": "Commit the JSON and generated JS together so the live site and authoring history stay aligned.", - "collaboration": "For simultaneous editing, resolve conflicts by memory id and story id rather than by whole-file ownership.", - "scalability": "Stories are the primary emotional routes. Legacy relationship fields remain readable for migration, but the observatory should stay story-first.", - } - - -def hidden_entries_by_type(entries: list[dict[str, Any]], content_type: str) -> list[dict[str, Any]]: - return [entry for entry in entries if entry.get("enabled", True) and entry.get("type") == content_type] - - -def hidden_contents(entries: list[dict[str, Any]], content_type: str) -> list[str]: - return [str(entry.get("content", "")) for entry in hidden_entries_by_type(entries, content_type)] - - -def generated_hidden_content_block(entries: list[dict[str, Any]]) -> str: - family_layers = hidden_contents(entries, "family layer") - details = hidden_contents(entries, "hidden tooltip") + hidden_contents(entries, "hover message") - poems = hidden_contents(entries, "poem") - greetings = hidden_contents(entries, "loading screen message") - night_messages = [ - entry["content"] for entry in hidden_entries_by_type(entries, "rare event") - if "night" in f"{entry.get('category', '')} {entry.get('triggerConditions', '')}".lower() - ] - quote_like = hidden_contents(entries, "quote") + hidden_contents(entries, "sensei chi wisdom entry") + hidden_contents(entries, "aphy system message") + hidden_contents(entries, "lima note/message") + hidden_contents(entries, "future z message") + hidden_contents(entries, "young z memory fragment") - lore = { - "quotes": quote_like, - "conversations": [ - entry.get("dialogue") if isinstance(entry.get("dialogue"), list) else [line for line in str(entry.get("content", "")).splitlines() if line.strip()] - for entry in hidden_entries_by_type(entries, "hidden conversation") - ], - "journals": hidden_contents(entries, "journal entry"), - "warnings": hidden_contents(entries, "fake error message"), - "dreams": hidden_contents(entries, "dream sequence"), - "cassettes": hidden_contents(entries, "terminal log"), - "fakeUsers": hidden_contents(entries, "guestbook entry"), - "seasonal": { - str(entry.get("season") or entry.get("triggerConditions") or entry.get("title", "")).lower().replace("season is ", ""): entry.get("content", "") - for entry in hidden_entries_by_type(entries, "seasonal event") - }, - "homepageTakeovers": [ - entry["content"] for entry in hidden_entries_by_type(entries, "rare event") - if "homepage" in f"{entry.get('category', '')} {entry.get('triggerConditions', '')}".lower() - ], - "roomLinks": [ - [entry.get("pageLocation", ""), entry.get("content", "") or entry.get("title", "")] - for entry in hidden_entries_by_type(entries, "secret interaction") - if entry.get("pageLocation") - ], - } - search_toasts = { - entry.get("query") or entry.get("triggerConditions") or entry.get("title", ""): entry.get("content", "") - for entry in hidden_entries_by_type(entries, "search toast") - } - search_routes = { - entry.get("query") or entry.get("triggerConditions") or entry.get("title", ""): entry.get("pageLocation") or entry.get("content", "") - for entry in hidden_entries_by_type(entries, "search route") - } - keyboard_secrets = [] - long_keyboard_secrets = [] - for entry in hidden_entries_by_type(entries, "keyboard secret"): - secret = dict(entry.get("keyboard") or {}) - secret["phrase"] = secret.get("phrase") or entry.get("triggerConditions") or entry.get("title", "") - if entry.get("pageLocation"): - secret["route"] = entry.get("pageLocation") - elif entry.get("content") == "play tiny aphy song": - secret["song"] = True - else: - secret["message"] = entry.get("content", "") - target = long_keyboard_secrets if len(secret["phrase"]) > 8 or " " in secret["phrase"] else keyboard_secrets - target.append(secret) - - def js_const(name: str, value: Any) -> str: - return f" const {name} = {json.dumps(value, ensure_ascii=False, indent=4)};\n" - - return ( - " // -----------------------------\n" - " // EDITABLE CONTENT\n" - " // -----------------------------\n" - " // Generated by the hidden narrative authoring page.\n" - " // Friendly source of truth: assets/content/hidden-details.json\n\n" - f"{js_const('familyLayers', family_layers)}\n" - f"{js_const('details', details)}\n" - f"{js_const('poems', poems)}\n" - f"{js_const('greetings', greetings)}\n" - f"{js_const('nightMessages', night_messages)}\n" - f"{js_const('lore', lore)}\n" - " // Exact search text -> toast message.\n" - f"{js_const('SEARCH_TOASTS', search_toasts)}\n" - " // Exact search text -> hidden page route.\n" - f"{js_const('SEARCH_ROUTES', search_routes)}\n" - " // Short typed phrases. Stored in sessionStorage as a rolling key chain.\n" - f"{js_const('KEYBOARD_SECRETS', keyboard_secrets)}\n" - " // Longer typed phrases and character names.\n" - f"{js_const('LONG_KEYBOARD_SECRETS', long_keyboard_secrets)}\n" - ) - - -def backup_hidden_file(path: Path, stamp: str) -> None: - if not path.exists(): - return - HIDDEN_BACKUP_DIR.mkdir(parents=True, exist_ok=True) - target = HIDDEN_BACKUP_DIR / f"{stamp}-{path.name}" - target.write_text(path.read_text(encoding="utf-8"), encoding="utf-8") - - -def replace_hidden_editable_block(source: str, entries: list[dict[str, Any]]) -> str: - start = source.find(" // -----------------------------\n // EDITABLE CONTENT") - end = source.find(" // -----------------------------\n // STATE HELPERS") - if start == -1 or end == -1 or end <= start: - raise ValueError("Could not find editable hidden content block.") - return source[:start] + generated_hidden_content_block(entries) + "\n" + source[end:] - - -def save_hidden_store(payload: dict[str, Any]) -> dict[str, Any]: - loaded = load_hidden_store() - current = {entry["id"]: entry for entry in loaded["entries"]} - entries = [normalize_hidden_entry(entry, current.get(str(entry.get("id", "")))) for entry in payload.get("entries", [])] - current_stories = {story["id"]: story for story in loaded.get("stories", [])} - stories = [normalize_hidden_story(story, entries, current_stories.get(str(story.get("id", "")))) for story in payload.get("stories", loaded.get("stories", []))] - ids = [entry["id"] for entry in entries] - if len(ids) != len(set(ids)): - raise ValueError("Entry ids must be unique.") - story_ids = [story["id"] for story in stories] - if len(story_ids) != len(set(story_ids)): - raise ValueError("Story ids must be unique.") - stamp = datetime.now().strftime("%Y%m%d-%H%M%S") - backup_hidden_file(HIDDEN_DETAILS_JS, stamp) - backup_hidden_file(HIDDEN_CONTENT_JSON, stamp) - HIDDEN_CONTENT_JSON.parent.mkdir(parents=True, exist_ok=True) - HIDDEN_CONTENT_JSON.write_text( - json.dumps( - { - "schemaVersion": 2, - "generatedAt": datetime.now().isoformat(timespec="seconds"), - "entries": entries, - "stories": stories, - }, - ensure_ascii=False, - indent=2, - ) + "\n", - encoding="utf-8", - ) - source = HIDDEN_DETAILS_JS.read_text(encoding="utf-8") - HIDDEN_DETAILS_JS.write_text(replace_hidden_editable_block(source, entries), encoding="utf-8") - saved = load_hidden_store() - saved["message"] = "stored safely. future-you will probably smile at this one." - saved["backupStamp"] = stamp - return saved - - -def parse_upload_form(content_type: str, body: bytes) -> tuple[str, bytes, str]: - if not content_type.lower().startswith("multipart/form-data"): - raise ValueError("Uploads must use multipart/form-data.") - message = BytesParser(policy=email_default_policy).parsebytes( - f"Content-Type: {content_type}\r\nMIME-Version: 1.0\r\n\r\n".encode("utf-8") + body - ) - if not message.is_multipart(): - raise ValueError("Upload form is not multipart.") - - filename = "" - payload = b"" - page_path = "" - for part in message.iter_parts(): - name = part.get_param("name", header="content-disposition") - if name == "attachment": - filename = part.get_filename() or "" - payload = part.get_payload(decode=True) or b"" - elif name == "pagePath": - raw_value = part.get_payload(decode=True) or b"" - page_path = raw_value.decode(part.get_content_charset() or "utf-8", errors="replace") - - if not filename: - raise ValueError("No attachment was uploaded.") - if not payload: - raise ValueError("Attachment is empty.") - return filename, payload, page_path - - -def run_build_commands(log_callback: Callable[[str], None] | None = None) -> tuple[bool, str, str]: - venv_python = ROOT / ".venv" / "bin" / "python" - venv_pip = ROOT / ".venv" / "bin" / "pip" - commands = [["emacs", "-Q", "--script", "build-site.el"]] - if not venv_python.exists(): - commands.extend( - [ - [sys.executable, "-m", "venv", ".venv"], - [str(venv_pip), "install", "-r", "requirements.txt"], - ] - ) - commands.append([str(venv_python), "search-index-json.py"]) - combined = [] - - def append_log(text: str) -> None: - combined.append(text) - if log_callback: - log_callback(text) - - ok = True - for command in commands: - append_log(f"$ {' '.join(command)}\n") - env = os.environ.copy() - env["PYTHONUNBUFFERED"] = "1" - proc = subprocess.Popen( - command, - cwd=ROOT, - env=env, - text=True, - bufsize=1, - stderr=subprocess.STDOUT, - stdout=subprocess.PIPE, - ) - assert proc.stdout is not None - for line in proc.stdout: - append_log(line) - return_code = proc.wait() - if return_code != 0: - ok = False - append_log(f"\nCommand exited with {return_code}.\n") - break - message = "Build complete. The site output and search index were regenerated." if ok else "Build failed. Check the log below." - return ok, message, "".join(combined) - - -def queue_build(page: dict[str, Any]) -> dict[str, Any]: - return BUILD_QUEUE.enqueue(page["path"], page["title"]).to_dict() - - -def queue_hidden_build() -> dict[str, Any]: - return BUILD_QUEUE.enqueue("assets/content/hidden-details.json", "Hidden Memory Observatory").to_dict() - - -APP_HTML = r""" - - - - - - Org Site Authoring - - - -
- -
-
-
-

New page

- -
-
- - - - - - - -
-
- - - - - - - - - - - - - -
- -
- - -
- -
- - - -
-
-

Build queue

-
-

Latest build log

-

-    
-
- - - -""" - - -HIDDEN_APP_HTML = r""" - - - - - Hidden Memory Observatory - - - -
- - -
-
-
-

Emotional Graph

-
Layer 0 lives near the edge. Layer 5 rests at the quiet core.
-
-
- - - - - - - - - - -
-
- -
- -
Click a story to open its route. Drag memories in the Story Builder to shape the path.
- -
Atlas
-
- - -
- - - - - -""" - - -class Handler(BaseHTTPRequestHandler): - server_version = "OrgAuthoring/1.0" - - def log_message(self, fmt: str, *args: Any) -> None: - sys.stderr.write("%s - %s\n" % (formatdate(time.time()), fmt % args)) - - def send_json(self, data: Any, status: HTTPStatus = HTTPStatus.OK) -> None: - body = json.dumps(data).encode("utf-8") - self.send_response(status) - self.send_header("Content-Type", "application/json; charset=utf-8") - self.send_header("Content-Length", str(len(body))) - self.send_header("Cache-Control", "no-store") - self.end_headers() - self.wfile.write(body) - - def do_GET(self) -> None: - parsed = urlparse(self.path) - if parsed.path == "/": - body = APP_HTML.encode("utf-8") - self.send_response(HTTPStatus.OK) - self.send_header("Content-Type", "text/html; charset=utf-8") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - return - if parsed.path == "/hidden": - body = HIDDEN_APP_HTML.encode("utf-8") - self.send_response(HTTPStatus.OK) - self.send_header("Content-Type", "text/html; charset=utf-8") - self.send_header("Content-Length", str(len(body))) - self.end_headers() - self.wfile.write(body) - return - if parsed.path == "/api/pages": - try: - self.send_json(list_pages()) - except Exception as exc: - self.send_json({"error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) - return - if parsed.path == "/api/diagnostics": - try: - self.send_json(server_diagnostics()) - except Exception as exc: - self.send_json({"error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) - return - if parsed.path == "/api/page": - query = parse_qs(parsed.query) - try: - path = safe_relative_path(query.get("path", [""])[0]) - self.send_json(page_to_dict(read_page(path))) - except Exception as exc: - self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) - return - if parsed.path == "/api/build": - self.send_json(BUILD_QUEUE.snapshot()) - return - if parsed.path == "/api/hidden": - try: - self.send_json(load_hidden_store()) - except Exception as exc: - self.send_json({"error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) - return - if parsed.path.startswith("/assets/avatars/"): - try: - relative = Path(parsed.path.lstrip("/")) - if relative.is_absolute() or ".." in relative.parts: - self.send_error(HTTPStatus.NOT_FOUND) - return - asset_path = (ROOT / relative).resolve() - avatars_root = (ROOT / "assets" / "avatars").resolve() - if not asset_path.is_file() or avatars_root not in asset_path.parents: - self.send_error(HTTPStatus.NOT_FOUND) - return - body = asset_path.read_bytes() - self.send_response(HTTPStatus.OK) - self.send_header("Content-Type", mimetypes.guess_type(asset_path.name)[0] or "application/octet-stream") - self.send_header("Content-Length", str(len(body))) - self.send_header("Cache-Control", "no-store") - self.end_headers() - self.wfile.write(body) - except Exception: - self.send_error(HTTPStatus.NOT_FOUND) - return - self.send_error(HTTPStatus.NOT_FOUND) - - def do_POST(self) -> None: - if self.path == "/api/upload": - try: - length = int(self.headers.get("Content-Length", "0")) - filename, payload, page_path = parse_upload_form( - self.headers.get("Content-Type", ""), - self.rfile.read(length), - ) - self.send_json(save_upload(filename, payload, page_path)) - except Exception as exc: - self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) - return - if self.path == "/api/hidden": - try: - length = int(self.headers.get("Content-Length", "0")) - data = json.loads(self.rfile.read(length).decode("utf-8")) - saved = save_hidden_store(data) - saved["queuedBuild"] = queue_hidden_build() - self.send_json(saved) - except Exception as exc: - self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) - return - if self.path != "/api/page": - self.send_error(HTTPStatus.NOT_FOUND) - return - try: - length = int(self.headers.get("Content-Length", "0")) - data = json.loads(self.rfile.read(length).decode("utf-8")) - saved = save_page(data) - saved["queuedBuild"] = queue_build(saved) - self.send_json(saved) - except Exception as exc: - self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) - - -def main() -> None: - port = int(os.environ.get("AUTHOR_PORT", "8765")) - server = ThreadingHTTPServer(("127.0.0.1", port), Handler) - page_count = len(list_pages()) - print(f"Authoring UI running at http://127.0.0.1:{port}") - print(f"Content root: {ROOT}") - print(f"Editable pages: {page_count}") - if page_count == 0: - print("WARNING: no editable pages were found. Check AUTHOR_CONTENT_ROOT/AUTHOR_ROOT and the launch directory.") - print("Press Ctrl-C to stop.") - try: - server.serve_forever() - except KeyboardInterrupt: - pass +from authoring_service import * # noqa: F401,F403 +from authoring_service.web import main if __name__ == "__main__": diff --git a/src/authoring_service/__init__.py b/src/authoring_service/__init__.py new file mode 100644 index 0000000..5f882f1 --- /dev/null +++ b/src/authoring_service/__init__.py @@ -0,0 +1,10 @@ +"""Authoring service package.""" + +from .build import BUILD_QUEUE, BuildJob, BuildQueue, queue_build, queue_hidden_build, run_build_commands +from .config import * +from .content import * +from .hidden import * +from .models import ContentPage, OrgPage +from .templates import APP_HTML, HIDDEN_APP_HTML +from .utils import html_escape, normalise_tags, org_date, parse_org_datetime, slugify +from .web import Handler, main diff --git a/src/authoring_service/build.py b/src/authoring_service/build.py new file mode 100644 index 0000000..1016012 --- /dev/null +++ b/src/authoring_service/build.py @@ -0,0 +1,167 @@ +"""Build queue and publishing command execution.""" + +from __future__ import annotations + +import os +import subprocess +import sys +import threading +import time +from dataclasses import dataclass, field +from typing import Any, Callable + +from .config import ROOT + +@dataclass +class BuildJob: + id: int + path: str + title: str + queued_at: float = field(default_factory=time.time) + started_at: float | None = None + finished_at: float | None = None + ok: bool | None = None + message: str = "Queued" + log: str = "" + + @property + def status(self) -> str: + if self.finished_at is not None: + return "done" if self.ok else "failed" + if self.started_at is not None: + return "running" + return "queued" + + def to_dict(self, include_log: bool = False) -> dict[str, Any]: + data = { + "id": self.id, + "path": self.path, + "title": self.title, + "queuedAt": self.queued_at, + "startedAt": self.started_at, + "finishedAt": self.finished_at, + "ok": self.ok, + "status": self.status, + "message": self.message, + } + if include_log: + data["log"] = self.log[-12000:] + return data + + +class BuildQueue: + def __init__(self) -> None: + self._lock = threading.Lock() + self._next_id = 1 + self._pending: list[BuildJob] = [] + self._current: BuildJob | None = None + self._recent: list[BuildJob] = [] + self._worker: threading.Thread | None = None + + def snapshot(self) -> dict[str, Any]: + with self._lock: + current = self._current.to_dict(include_log=True) if self._current else None + recent = [job.to_dict(include_log=True) for job in self._recent[-10:]] + pending = [job.to_dict() for job in self._pending] + latest = current or (recent[-1] if recent else None) + message = latest["message"] if latest else "No builds have run yet." + return { + "running": current is not None, + "queued": len(pending), + "message": message, + "current": current, + "pending": pending, + "recent": recent, + "log": latest.get("log", "") if latest else "", + } + + def enqueue(self, path: str, title: str) -> BuildJob: + with self._lock: + job = BuildJob(self._next_id, path, title) + self._next_id += 1 + self._pending.append(job) + if self._worker is None or not self._worker.is_alive(): + self._worker = threading.Thread(target=self._run_worker, daemon=True) + self._worker.start() + return job + + def _run_worker(self) -> None: + while True: + with self._lock: + if not self._pending: + self._current = None + return + job = self._pending.pop(0) + job.started_at = time.time() + job.message = "Publishing site and search index." + self._current = job + def append_log(chunk: str) -> None: + with self._lock: + job.log = (job.log + chunk)[-200000:] + + ok, message, log = run_build_commands(append_log) + with self._lock: + job.finished_at = time.time() + job.ok = ok + job.message = message + job.log = log[-200000:] + self._recent.append(job) + self._recent = self._recent[-20:] + self._current = None + + +BUILD_QUEUE = BuildQueue() + + +def run_build_commands(log_callback: Callable[[str], None] | None = None) -> tuple[bool, str, str]: + venv_python = ROOT / ".venv" / "bin" / "python" + venv_pip = ROOT / ".venv" / "bin" / "pip" + commands = [["emacs", "-Q", "--script", "build-site.el"]] + if not venv_python.exists(): + commands.extend( + [ + [sys.executable, "-m", "venv", ".venv"], + [str(venv_pip), "install", "-r", "requirements.txt"], + ] + ) + commands.append([str(venv_python), "search-index-json.py"]) + combined = [] + + def append_log(text: str) -> None: + combined.append(text) + if log_callback: + log_callback(text) + + ok = True + for command in commands: + append_log(f"$ {' '.join(command)}\n") + env = os.environ.copy() + env["PYTHONUNBUFFERED"] = "1" + proc = subprocess.Popen( + command, + cwd=ROOT, + env=env, + text=True, + bufsize=1, + stderr=subprocess.STDOUT, + stdout=subprocess.PIPE, + ) + assert proc.stdout is not None + for line in proc.stdout: + append_log(line) + return_code = proc.wait() + if return_code != 0: + ok = False + append_log(f"\nCommand exited with {return_code}.\n") + break + message = "Build complete. The site output and search index were regenerated." if ok else "Build failed. Check the log below." + return ok, message, "".join(combined) + + +def queue_build(page: dict[str, Any]) -> dict[str, Any]: + return BUILD_QUEUE.enqueue(page["path"], page["title"]).to_dict() + + +def queue_hidden_build() -> dict[str, Any]: + return BUILD_QUEUE.enqueue("assets/content/hidden-details.json", "Hidden Memory Observatory").to_dict() + diff --git a/src/authoring_service/config.py b/src/authoring_service/config.py new file mode 100644 index 0000000..d9bd163 --- /dev/null +++ b/src/authoring_service/config.py @@ -0,0 +1,97 @@ +"""Configuration and content-root discovery for the authoring service.""" + +from __future__ import annotations + +import os +from pathlib import Path + +APP_ROOT = Path(__file__).resolve().parents[2] +DEFAULT_CONTENT_ROOT = Path("/home/zaine/master-folder/org_files/org_web") + + +def looks_like_content_root(path: Path) -> bool: + return (path / "blogs").exists() or (path / "posts").exists() or (path / "lima").exists() + + +def resolve_root() -> Path: + for env_name in ("AUTHOR_CONTENT_ROOT", "AUTHOR_ROOT"): + env_root = os.environ.get(env_name) + if not env_root: + continue + resolved = Path(env_root).expanduser().resolve() + if looks_like_content_root(resolved): + return resolved + candidates = [ + Path.cwd(), + DEFAULT_CONTENT_ROOT, + APP_ROOT, + ] + workspace = os.environ.get("GITHUB_WORKSPACE") + if workspace: + candidates.insert(0, Path(workspace)) + for base in list(candidates): + candidates.extend(base.parents) + seen = set() + for candidate in candidates: + resolved = candidate.expanduser().resolve() + if resolved in seen: + continue + seen.add(resolved) + if looks_like_content_root(resolved): + return resolved + return APP_ROOT + + +ROOT = resolve_root() +BLOGS_DIR = ROOT / "blogs" +POSTS_DIR = ROOT / "posts" +LIMA_DIR = ROOT / "lima" +IMAGE_ASSETS_DIR = ROOT / "assets" / "images" +HZONE_ASSETS_DIR = IMAGE_ASSETS_DIR / "hzone" +HIDDEN_DETAILS_JS = ROOT / "assets" / "scripts" / "hidden-details.js" +HIDDEN_CONTENT_JSON = ROOT / "assets" / "content" / "hidden-details.json" +HIDDEN_BACKUP_DIR = Path(os.environ.get("AUTHOR_HIDDEN_BACKUP_DIR", APP_ROOT / "backups" / "hidden-details")).expanduser().resolve() +EXCLUDED_CONTENT_DIR_NAMES = { + ".agents", + ".codex", + ".git", + ".packages", + ".venv", + "__pycache__", + "assets", + "backups", + "output", + "tags", +} +GENERATED_ORG_NAMES = { + "blogs-list.org", + "books-list.org", + "posts-list.org", + "career-list.org", + "sitemap.org", + "recently-updated.org", + "wip.org", +} +GENERATED_CONTENT_NAMES = GENERATED_ORG_NAMES | {"lima-list.org"} +ALLOWED_UPLOAD_EXTENSIONS = { + ".png", + ".jpg", + ".jpeg", + ".gif", + ".webp", + ".svg", +} +MONTH_NAMES = [ + "january", + "february", + "march", + "april", + "may", + "june", + "july", + "august", + "september", + "october", + "november", + "december", +] diff --git a/src/authoring_service/constants.py b/src/authoring_service/constants.py new file mode 100644 index 0000000..6ae8ee2 --- /dev/null +++ b/src/authoring_service/constants.py @@ -0,0 +1,202 @@ +"""Hidden narrative constants used by the authoring UI.""" + +from __future__ import annotations + +HIDDEN_CONTENT_TYPES = [ + "quote", + "poem", + "hidden dialogue", + "journal entry", + "rare event", + "loading screen message", + "secret interaction", + "hidden tooltip", + "future z message", + "young z memory fragment", + "sensei chi wisdom entry", + "aphy system message", + "lima note/message", + "dream sequence", + "terminal log", + "fake error message", + "recurring joke", + "seasonal event", + "weather-based event", + "hover message", + "hidden achievement", + "guestbook entry", + "hidden conversation", + "family layer", + "search toast", + "search route", + "keyboard secret", +] + +CHARACTER_REGISTRY = { + "young z": { + "id": "young z", + "displayLabel": "young z", + "aliases": ["Young Z", "young z", "young-z", "young_z", "young"], + "territoryColor": "#d8a95a", + "glow": "#f0b85c", + "avatar": "https://zainezq.com/assets/avatars/young-z.jpeg", + "fallbackAvatar": "https://zainezq.com/assets/avatars/z.jpeg", + "expressions": { + "default": "https://zainezq.com/assets/avatars/young-z.jpeg", + "nostalgic": "https://zainezq.com/assets/avatars/young-z.jpeg", + "playful": "https://zainezq.com/assets/avatars/young-z.jpeg", + }, + "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, + "observatory": {"territory": "nostalgic/playful memories", "shimmer": "soft"}, + "presenceTones": ["nostalgic", "funny", "hopeful"], + "presenceKeywords": ["childhood", "play", "desk", "crayon", "young", "small", "blanket", "memory"], + "symbol": "Y", + "motifs": ["crayon sun", "blanket cape", "childhood desk"], + "themes": ["childhood", "play", "memory", "safety"], + "affinities": ["z", "future z", "aphy", "lima"], + }, + "z": { + "id": "z", + "displayLabel": "z", + "aliases": ["Z", "z", "zaine"], + "territoryColor": "#d6c38a", + "glow": "#ead68e", + "avatar": "https://zainezq.com/assets/avatars/z.jpeg", + "fallbackAvatar": "https://zainezq.com/assets/avatars/z.jpeg", + "expressions": {"default": "https://zainezq.com/assets/avatars/z.jpeg", "calm": "https://zainezq.com/assets/avatars/z.jpeg"}, + "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, + "observatory": {"territory": "self/archive memories", "shimmer": "steady"}, + "presenceTones": ["warm", "soft", "hopeful"], + "presenceKeywords": ["website", "archive", "home", "self", "making", "return", "zaine"], + "symbol": "Z", + "motifs": ["archive", "website", "home"], + "themes": ["selfhood", "return", "making"], + "affinities": ["lima", "young z", "future z"], + }, + "aphy": { + "id": "aphy", + "displayLabel": "aphy", + "aliases": ["Aphy", "aphy", "aphy_bot", "aphy bot"], + "territoryColor": "#7fb089", + "glow": "#9dd3a6", + "avatar": "https://zainezq.com/assets/avatars/aphy.jpeg", + "fallbackAvatar": "https://zainezq.com/assets/avatars/z.jpeg", + "expressions": { + "default": "https://zainezq.com/assets/avatars/aphy.jpeg", + "system": "https://zainezq.com/assets/avatars/aphy.jpeg", + "amused": "https://zainezq.com/assets/avatars/aphy.jpeg", + }, + "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, + "observatory": {"territory": "system/symbolic areas", "shimmer": "diagnostic"}, + "presenceTones": ["funny", "strange", "hopeful"], + "presenceKeywords": ["system", "console", "diagnostic", "keyboard", "search", "terminal", "backup", "symbolic", "query"], + "symbol": "A", + "motifs": ["console", "diagnostic", "backup"], + "themes": ["humor", "systems", "care through tools"], + "affinities": ["z", "lima", "sensei chi"], + }, + "lima": { + "id": "lima", + "displayLabel": "lima", + "aliases": ["Lima", "lima"], + "territoryColor": "#d06b78", + "glow": "#f2c58b", + "avatar": "https://zainezq.com/assets/avatars/lima.jpg", + "fallbackAvatar": "https://zainezq.com/assets/avatars/z.jpeg", + "expressions": { + "default": "https://zainezq.com/assets/avatars/lima.jpg", + "calm": "https://zainezq.com/assets/avatars/lima.jpg", + "warm": "https://zainezq.com/assets/avatars/lima.jpg", + "protective": "https://zainezq.com/assets/avatars/lima.jpg", + }, + "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, + "observatory": {"territory": "warm/protective arcs", "shimmer": "warm"}, + "presenceTones": ["warm", "protective", "soft"], + "presenceKeywords": ["lima", "love", "warm", "kitchen", "light", "ring", "home", "protect", "eat"], + "symbol": "L", + "motifs": ["warmth", "kitchen light", "ring"], + "themes": ["love", "home", "grounding"], + "affinities": ["z", "aphy", "future z", "young z"], + }, + "sensei chi": { + "id": "sensei chi", + "displayLabel": "sensei chi", + "aliases": ["Sensei Chi", "sensei chi", "sensei-chi", "sensei_chi", "sensei"], + "territoryColor": "#75a9bd", + "glow": "#9ccddd", + "avatar": "https://zainezq.com/assets/avatars/sensei-chi.jpeg", + "fallbackAvatar": "https://zainezq.com/assets/avatars/z.jpeg", + "expressions": { + "default": "https://zainezq.com/assets/avatars/sensei-chi.jpeg", + "reflective": "https://zainezq.com/assets/avatars/sensei-chi.jpeg", + "wise": "https://zainezq.com/assets/avatars/sensei-chi.jpeg", + }, + "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, + "observatory": {"territory": "reflective areas", "shimmer": "quiet"}, + "presenceTones": ["wise", "melancholy", "soft"], + "presenceKeywords": ["reflection", "patience", "wisdom", "lesson", "tea", "garden", "quiet", "sensei"], + "symbol": "S", + "motifs": ["tea", "garden", "quiet lesson"], + "themes": ["reflection", "patience", "wisdom"], + "affinities": ["aphy", "future z"], + }, + "future z": { + "id": "future z", + "displayLabel": "future z", + "aliases": ["Future Z", "future z", "future-z", "future_z", "future"], + "territoryColor": "#a58ac9", + "glow": "#c2a4ee", + "avatar": "https://zainezq.com/assets/avatars/future-z.jpeg", + "fallbackAvatar": "https://zainezq.com/assets/avatars/z.jpeg", + "expressions": { + "default": "https://zainezq.com/assets/avatars/future-z.jpeg", + "temporal": "https://zainezq.com/assets/avatars/future-z.jpeg", + "reassuring": "https://zainezq.com/assets/avatars/future-z.jpeg", + }, + "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, + "observatory": {"territory": "temporal regions", "shimmer": "temporal"}, + "presenceTones": ["hopeful", "melancholy", "wise"], + "presenceKeywords": ["future", "time", "clock", "older", "tomorrow", "age", "reassurance", "continuity"], + "symbol": "F", + "motifs": ["clock", "age 40", "future log"], + "themes": ["time", "reassurance", "continuity"], + "affinities": ["z", "young z", "lima", "sensei chi"], + }, +} +HIDDEN_CHARACTERS = list(CHARACTER_REGISTRY) +HIDDEN_TONES = ["warm", "funny", "nostalgic", "wise", "strange", "soft", "hopeful", "protective", "melancholy"] +HIDDEN_RARITIES = ["common", "uncommon", "rare", "very rare", "seasonal", "timed"] +HIDDEN_STORY_MARKERS = ["public", "hidden", "rare", "emotional", "dream-like", "temporal"] +HIDDEN_DISCOVERY_STYLES = ["gradual", "direct", "hidden route", "character-led", "dream-like", "temporal"] +HIDDEN_LAYER_DEPTHS = [ + { + "id": "0", + "name": "Surface Reality", + "meaning": "Normal visible website content, visible warmth, and ordinary interactions.", + }, + { + "id": "1", + "name": "Hidden Personality", + "meaning": "Small hidden jokes, hover text, tiny discoveries, and recurring symbols.", + }, + { + "id": "2", + "name": "Memory Layer", + "meaning": "young z memories, lima notes, nostalgia, and emotional fragments.", + }, + { + "id": "3", + "name": "Reflection Layer", + "meaning": "sensei chi philosophy, aphy conversations, and introspection.", + }, + { + "id": "4", + "name": "Time Layer", + "meaning": "future z logs, time anomalies, long-term revisits, and future/past echoes.", + }, + { + "id": "5", + "name": "Core Layer", + "meaning": "Rare deeply emotional truths found by patient exploration.", + }, +] diff --git a/src/authoring_service/content.py b/src/authoring_service/content.py new file mode 100644 index 0000000..864375b --- /dev/null +++ b/src/authoring_service/content.py @@ -0,0 +1,427 @@ +"""Editable page, upload, and diagnostics operations.""" + +from __future__ import annotations + +import os +import posixpath +import re +import struct +import sys +from datetime import datetime +from email.parser import BytesParser +from email.policy import default as email_default_policy +from pathlib import Path +from typing import Any + +from .config import ( + ALLOWED_UPLOAD_EXTENSIONS, + BLOGS_DIR, + EXCLUDED_CONTENT_DIR_NAMES, + GENERATED_CONTENT_NAMES, + HZONE_ASSETS_DIR, + IMAGE_ASSETS_DIR, + LIMA_DIR, + MONTH_NAMES, + POSTS_DIR, + ROOT, +) +from .models import ContentPage +from .utils import html_escape, normalise_tags, org_date, parse_org_datetime, slugify + +def safe_relative_path(path: str) -> Path: + rel = Path(path) + if rel.is_absolute() or ".." in rel.parts: + raise ValueError("Path must stay inside this repository.") + full = (ROOT / rel).resolve() + if not full.is_relative_to(ROOT): + raise ValueError("Path must stay inside this repository.") + if full.name in GENERATED_CONTENT_NAMES or "sync-conflict" in full.name: + raise ValueError("Generated and sync-conflict files are not editable here.") + rel_parts = full.relative_to(ROOT).parts + if any(part in EXCLUDED_CONTENT_DIR_NAMES for part in rel_parts): + raise ValueError("This path is outside the editable content folders.") + if full.suffix == ".org": + return full + if full.suffix == ".md" and full.is_relative_to(LIMA_DIR): + return full + raise ValueError("Only .org content files and .md files under lima can be edited here.") + + +def safe_target_path(path: str, slug: str, page_type: str) -> Path: + candidate = path.strip() + if not candidate: + raise ValueError("Path is required.") + default_ext = ".md" if page_type == "lima" else ".org" + if candidate.endswith("/"): + candidate = f"{candidate}{slug}{default_ext}" + elif not Path(candidate).suffix: + candidate = f"{candidate}{default_ext}" + return safe_relative_path(candidate) + + +def markdown_title(content: str, fallback: str) -> str: + for line in content.splitlines(): + match = re.match(r"^#{1,6}\s+(.+?)\s*$", line) + if match: + return match.group(1).strip() + return fallback.replace("-", " ").replace("_", " ").title() + + +def read_markdown_page(path: Path) -> ContentPage: + content = path.read_text(encoding="utf-8") + rel = path.relative_to(ROOT).as_posix() + title = markdown_title(content, path.stem) + return ContentPage( + path=rel, + page_type="lima", + title=title, + slug=path.stem, + tags=[], + content=content, + date="", + comments=True, + options="", + format="markdown", + ) + + +def read_page(path: Path) -> ContentPage: + if path.suffix == ".md" and path.is_relative_to(LIMA_DIR): + return read_markdown_page(path) + text = path.read_text(encoding="utf-8") + meta: dict[str, str] = {} + body_lines: list[str] = [] + in_header = True + for line in text.splitlines(): + if in_header and line.startswith("#+"): + key, _, value = line[2:].partition(":") + meta[key.strip().upper()] = value.strip() + else: + in_header = False + body_lines.append(line) + rel = path.relative_to(ROOT).as_posix() + if path.is_relative_to(BLOGS_DIR): + page_type = "blog" + elif path.is_relative_to(POSTS_DIR): + page_type = "post" + else: + page_type = "page" + slug = meta.get("SLUG") or path.stem + tags = normalise_tags(meta.get("FILETAGS", "")) + return ContentPage( + path=rel, + page_type=page_type, + title=meta.get("TITLE", path.stem), + slug=slug, + tags=tags, + content="\n".join(body_lines).lstrip("\n"), + date=meta.get("DATE", org_date(datetime.fromtimestamp(path.stat().st_mtime))), + comments=meta.get("COMMENTS", "t").lower() == "t", + options=meta.get("OPTIONS", "num:nil"), + format="org", + wip=meta.get("WIP"), + ) + + +def page_to_dict(page: ContentPage) -> dict[str, Any]: + return { + "path": page.path, + "pageType": page.page_type, + "title": page.title, + "slug": page.slug, + "tags": page.tags, + "content": page.content, + "date": page.date, + "comments": page.comments, + "options": page.options, + "format": page.format, + "wip": page.wip or "", + } + + +def server_diagnostics() -> dict[str, Any]: + pages = list_pages() + try: + cwd = Path.cwd().as_posix() + except OSError as exc: + cwd = f"" + return { + "root": ROOT.as_posix(), + "cwd": cwd, + "executable": sys.executable, + "pid": os.getpid(), + "pageCount": len(pages), + "firstPage": pages[0]["path"] if pages else "", + } + + +def list_pages() -> list[dict[str, Any]]: + pages = [] + org_paths = [] + if ROOT.exists(): + try: + for path in ROOT.rglob("*.org"): + try: + rel_parts = path.relative_to(ROOT).parts + except ValueError: + continue + if any(part in EXCLUDED_CONTENT_DIR_NAMES for part in rel_parts): + continue + org_paths.append(path) + except OSError: + org_paths = [] + try: + md_paths = list(LIMA_DIR.rglob("*.md")) if LIMA_DIR.exists() else [] + except OSError: + md_paths = [] + for path in sorted(org_paths + md_paths): + if path.name in GENERATED_CONTENT_NAMES or "sync-conflict" in path.name: + continue + try: + page = read_page(path) + parsed = parse_org_datetime(page.date) + timestamp = parsed.timestamp() if parsed else path.stat().st_mtime + except (OSError, UnicodeDecodeError, ValueError): + continue + pages.append( + { + "path": page.path, + "pageType": page.page_type, + "title": page.title, + "slug": page.slug, + "tags": page.tags, + "date": page.date, + "format": page.format, + "timestamp": timestamp, + } + ) + return sorted(pages, key=lambda item: item["timestamp"], reverse=True) + + +def target_path(data: dict[str, Any], existing_path: str | None) -> Path: + if existing_path: + return safe_relative_path(existing_path) + title = str(data.get("title") or "").strip() + slug = slugify(str(data.get("slug") or title)) + page_type = str(data.get("pageType") or "blog") + explicit_path = str(data.get("targetPath") or "").strip() + if explicit_path: + return safe_target_path(explicit_path, slug, page_type) + if page_type == "blog": + dt = parse_org_datetime(str(data.get("date") or "")) or datetime.now() + folder = BLOGS_DIR / str(dt.year) / f"{dt.month:02d}-{MONTH_NAMES[dt.month - 1]}" + return folder / f"{slug}.org" + if page_type == "post": + raw_section = str(data.get("section") or "").strip() + section = slugify(raw_section) if raw_section else "" + folder = POSTS_DIR / section if section else POSTS_DIR + return folder / f"{slug}.org" + if page_type == "lima": + return LIMA_DIR / f"{slug}.md" + if page_type == "page": + return ROOT / f"{slug}.org" + raise ValueError("pageType must be blog, post, page, or lima.") + + +def render_markdown(data: dict[str, Any]) -> str: + content = str(data.get("content") or "").replace("\r\n", "\n").strip() + title = str(data.get("title") or "").strip() + if not title: + raise ValueError("Title is required.") + content = re.sub( + r']*>\s*]*\bsrc="([^"]+)"[^>]*\balt="([^"]*)"[^>]*>\s*', + lambda match: f"![{match.group(2)}]({match.group(1)})", + content, + flags=re.IGNORECASE, + ) + if content: + if re.search(r"^#{1,6}[^\S\r\n]+.+?[^\S\r\n]*$", content, flags=re.MULTILINE): + content = re.sub( + r"^#{1,6}[^\S\r\n]+.+?[^\S\r\n]*$", + f"# {title}", + content, + count=1, + flags=re.MULTILINE, + ) + else: + content = f"# {title}\n\n{content}" + return content + "\n" + return f"# {title}\n" + + +def render_org(data: dict[str, Any], previous: ContentPage | None) -> str: + title = str(data.get("title") or "").strip() + if not title: + raise ValueError("Title is required.") + slug = slugify(str(data.get("slug") or title)) + tags = normalise_tags(data.get("tags", [])) + content = str(data.get("content") or "").replace("\r\n", "\n").strip() + date = str(data.get("date") or "").strip() + if not parse_org_datetime(date): + date = previous.date if previous else org_date(datetime.now()) + options = str(data.get("options") or (previous.options if previous else "num:nil")).strip() + comments = bool(data.get("comments", True)) + lines = [ + f"#+TITLE: {title}", + f"#+OPTIONS: {options}", + f"#+DATE: {date}", + f"#+filetags: {''.join(f':{tag}' for tag in tags)}:", + ] + wip = str(data.get("wip") or (previous.wip if previous else "") or "").strip() + if wip: + lines.append(f"#+WIP: {wip}") + lines.extend( + [ + f"#+COMMENTS: {'t' if comments else ''}", + f"#+SLUG: {slug}", + "", + content, + "", + ] + ) + return "\n".join(lines) + + +def save_page(data: dict[str, Any]) -> dict[str, Any]: + existing_path = data.get("path") or None + target = target_path(data, str(existing_path) if existing_path else None) + previous = read_page(target) if target.exists() else None + if not target.parent.exists(): + target.parent.mkdir(parents=True) + if target.exists() and not existing_path: + raise ValueError(f"{target.relative_to(ROOT)} already exists.") + if target.suffix == ".md": + target.write_text(render_markdown(data), encoding="utf-8") + else: + target.write_text(render_org(data, previous), encoding="utf-8") + return page_to_dict(read_page(target)) + + +def image_dimensions(payload: bytes, ext: str) -> tuple[int, int] | None: + if ext == ".png" and payload.startswith(b"\x89PNG\r\n\x1a\n") and len(payload) >= 24: + width, height = struct.unpack(">II", payload[16:24]) + return width, height + if ext == ".gif" and payload[:6] in {b"GIF87a", b"GIF89a"} and len(payload) >= 10: + width, height = struct.unpack(" len(payload): + break + size = int.from_bytes(payload[i:i + 2], "big") + if size < 2: + break + if marker in {0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF}: + if i + 7 <= len(payload): + height = int.from_bytes(payload[i + 3:i + 5], "big") + width = int.from_bytes(payload[i + 5:i + 7], "big") + return width, height + break + i += size + return None + + +def relative_asset_path(page_path: str, asset_path: str) -> str: + page = safe_relative_path(page_path) if page_path else ROOT / "index.org" + page_rel = page.relative_to(ROOT).as_posix() + page_output_dir = posixpath.dirname(page_rel) + return posixpath.relpath(asset_path, page_output_dir or ".") + + +def gallery_image_html(filename: str, asset_path: str, page_path: str, payload: bytes, ext: str) -> str: + absolute_url = f"https://zainezq.com/{asset_path}" + relative_url = relative_asset_path(page_path, asset_path) + dims = image_dimensions(payload, ext) + width, height = dims if dims else (1920, 1080) + alt = html_escape(filename) + return ( + f'' + f'{alt}' + ) + + +def attachment_image_dir(page_path: str) -> Path: + page = safe_relative_path(page_path) if page_path else ROOT / "index.org" + if page.suffix == ".md" and page.is_relative_to(LIMA_DIR): + return HZONE_ASSETS_DIR + if page.is_relative_to(POSTS_DIR): + rel = page.relative_to(POSTS_DIR) + section = rel.parts[0] if len(rel.parts) > 1 else "posts" + return IMAGE_ASSETS_DIR / slugify(section) + if page.is_relative_to(BLOGS_DIR): + return IMAGE_ASSETS_DIR / "blogs" + rel = page.relative_to(ROOT) + if len(rel.parts) > 1: + return IMAGE_ASSETS_DIR / slugify(rel.parts[0]) + return IMAGE_ASSETS_DIR / "pages" + + +def save_upload(filename: str, payload: bytes, page_path: str = "") -> dict[str, str]: + original = Path(filename or "attachment").name + ext = Path(original).suffix.lower() + if ext not in ALLOWED_UPLOAD_EXTENSIONS: + raise ValueError("Only common image files can be uploaded.") + now = datetime.now() + target_dir = attachment_image_dir(page_path) + target_dir.mkdir(parents=True, exist_ok=True) + stem = slugify(Path(original).stem) + prefix = "" if re.match(r"^\d{4}-\d{2}-\d{2}-", stem) else f"{now.strftime('%Y-%m-%d')}-" + target = target_dir / f"{prefix}{stem}{ext}" + counter = 2 + while target.exists(): + target = target_dir / f"{prefix}{stem}-{counter}{ext}" + counter += 1 + target.write_bytes(payload) + rel = target.relative_to(ROOT).as_posix() + absolute_url = f"https://zainezq.com/{rel}" + relative_url = relative_asset_path(page_path, rel) + is_markdown = page_path.endswith(".md") + insert_text = f"![{target.name}]({absolute_url})" if is_markdown else f"[[{relative_url}]]" + return { + "url": absolute_url, + "relativeUrl": relative_url, + "path": rel, + "markdown": insert_text, + "insertText": insert_text, + "filename": target.name, + } + + + + +def parse_upload_form(content_type: str, body: bytes) -> tuple[str, bytes, str]: + if not content_type.lower().startswith("multipart/form-data"): + raise ValueError("Uploads must use multipart/form-data.") + message = BytesParser(policy=email_default_policy).parsebytes( + f"Content-Type: {content_type}\r\nMIME-Version: 1.0\r\n\r\n".encode("utf-8") + body + ) + if not message.is_multipart(): + raise ValueError("Upload form is not multipart.") + + filename = "" + payload = b"" + page_path = "" + for part in message.iter_parts(): + name = part.get_param("name", header="content-disposition") + if name == "attachment": + filename = part.get_filename() or "" + payload = part.get_payload(decode=True) or b"" + elif name == "pagePath": + raw_value = part.get_payload(decode=True) or b"" + page_path = raw_value.decode(part.get_content_charset() or "utf-8", errors="replace") + + if not filename: + raise ValueError("No attachment was uploaded.") + if not payload: + raise ValueError("Attachment is empty.") + return filename, payload, page_path + diff --git a/src/authoring_service/hidden.py b/src/authoring_service/hidden.py new file mode 100644 index 0000000..8ee4d83 --- /dev/null +++ b/src/authoring_service/hidden.py @@ -0,0 +1,660 @@ +"""Hidden narrative store loading, migration, validation, and persistence.""" + +from __future__ import annotations + +import json +import re +from datetime import datetime +from pathlib import Path +from typing import Any + +from .config import HIDDEN_BACKUP_DIR, HIDDEN_CONTENT_JSON, HIDDEN_DETAILS_JS, ROOT +from .constants import ( + CHARACTER_REGISTRY, + HIDDEN_CHARACTERS, + HIDDEN_CONTENT_TYPES, + HIDDEN_DISCOVERY_STYLES, + HIDDEN_LAYER_DEPTHS, + HIDDEN_RARITIES, + HIDDEN_STORY_MARKERS, + HIDDEN_TONES, +) +from .utils import normalise_tags, slugify + +def find_js_const_literal(source: str, name: str) -> str: + marker = f"const {name} =" + start = source.find(marker) + if start == -1: + raise ValueError(f"Could not find const {name}.") + value_start = source.find("=", start) + 1 + while value_start < len(source) and source[value_start].isspace(): + value_start += 1 + opener = source[value_start] + pairs = {"[": "]", "{": "}"} + if opener not in pairs: + raise ValueError(f"const {name} is not an array or object.") + closer = pairs[opener] + depth = 0 + in_string = False + escape = False + for index in range(value_start, len(source)): + char = source[index] + if in_string: + if escape: + escape = False + elif char == "\\": + escape = True + elif char == '"': + in_string = False + continue + if char == '"': + in_string = True + elif char == opener: + depth += 1 + elif char == closer: + depth -= 1 + if depth == 0: + return source[value_start:index + 1] + raise ValueError(f"Could not parse const {name}.") + + +def js_literal_to_json(value: str) -> Any: + cleaned = re.sub(r"//.*", "", value) + cleaned = re.sub(r"(/\*.*?\*/)", "", cleaned, flags=re.DOTALL) + cleaned = re.sub(r"([{\[,]\s*)([A-Za-z_$][\w$]*)\s*:", r'\1"\2":', cleaned) + cleaned = re.sub(r",(\s*[\]}])", r"\1", cleaned) + return json.loads(cleaned) + + +def character_alias_lookup() -> dict[str, str]: + aliases = {} + for canonical, config in CHARACTER_REGISTRY.items(): + aliases[canonical] = canonical + for alias in config["aliases"]: + aliases[slugify(str(alias)).replace("-", " ")] = canonical + aliases[str(alias).strip().lower()] = canonical + return aliases + + +def normalize_character_name(value: Any) -> str | None: + raw = str(value or "").strip() + if not raw: + return None + simplified = slugify(raw).replace("-", " ") + return character_alias_lookup().get(raw.lower()) or character_alias_lookup().get(simplified) + + +def normalize_character_list(values: Any) -> list[str]: + if isinstance(values, str): + raw_values = re.split(r"[,/]", values) + elif isinstance(values, list): + raw_values = values + else: + raw_values = [] + normalized = [] + for item in raw_values: + canonical = normalize_character_name(item) + if canonical and canonical not in normalized: + normalized.append(canonical) + return normalized + + +def normalize_character_text_refs(value: Any) -> Any: + if isinstance(value, list): + return [normalize_character_text_refs(item) for item in value] + if isinstance(value, dict): + return {key: normalize_character_text_refs(item) for key, item in value.items()} + if not isinstance(value, str): + return value + text = value + replacements = [] + for canonical, config in CHARACTER_REGISTRY.items(): + for alias in config["aliases"]: + if alias == canonical: + continue + if alias.lower() in {"young", "future", "sensei"}: + continue + replacements.append((alias, canonical)) + replacements.sort(key=lambda item: len(item[0]), reverse=True) + for alias, canonical in replacements: + pattern = r"(? list[str]: + found = [] + normalized_text = normalize_character_text_refs(text).lower() + for character in HIDDEN_CHARACTERS: + if re.search(r"(? str: + return f"{slugify(content_type)}-{index + 1:03d}-{slugify(title)[:42]}" + + +def hidden_title(content_type: str, content: str, index: int) -> str: + text = re.sub(r"\s+", " ", content).strip() + if not text: + return f"{content_type.title()} {index + 1}" + return text[:58] + ("..." if len(text) > 58 else "") + + +def make_hidden_entry(content_type: str, content: str, index: int, **extra: Any) -> dict[str, Any]: + now_iso = datetime.now().date().isoformat() + title = str(extra.pop("title", "") or hidden_title(content_type, content, index)) + characters = extra.pop("characters", None) or detect_hidden_characters(f"{title} {content}") + tags = extra.pop("tags", None) or [slugify(character) for character in characters] + entry = { + "id": hidden_entry_id(content_type, index, title), + "type": content_type, + "title": title, + "content": content, + "characters": characters, + "emotionalTone": extra.pop("emotionalTone", "warm"), + "rarity": extra.pop("rarity", "common"), + "triggerConditions": extra.pop("triggerConditions", ""), + "tags": tags, + "category": extra.pop("category", content_type), + "pageLocation": extra.pop("pageLocation", ""), + "familyLayer": extra.pop("familyLayer", ""), + "enabled": extra.pop("enabled", True), + "createdDate": extra.pop("createdDate", now_iso), + "modifiedDate": extra.pop("modifiedDate", now_iso), + "notes": extra.pop("notes", ""), + "audioSettings": extra.pop("audioSettings", ""), + "animationTrigger": extra.pop("animationTrigger", ""), + "cssClassHooks": extra.pop("cssClassHooks", ""), + "chainReferences": extra.pop("chainReferences", []), + "continuationLinks": extra.pop("continuationLinks", []), + "emotionalRole": extra.pop("emotionalRole", ""), + "discoveryDifficulty": extra.pop("discoveryDifficulty", "gentle"), + "mysteryLevel": extra.pop("mysteryLevel", "quiet"), + "resonanceScore": extra.pop("resonanceScore", 3), + "symbols": extra.pop("symbols", []), + "narrativeArcs": extra.pop("narrativeArcs", []), + "parentLinks": extra.pop("parentLinks", []), + "childLinks": extra.pop("childLinks", []), + "echoes": extra.pop("echoes", []), + "mirroredEntries": extra.pop("mirroredEntries", []), + "thematicLinks": extra.pop("thematicLinks", []), + "symbolicLinks": extra.pop("symbolicLinks", []), + "triggerLinks": extra.pop("triggerLinks", []), + } + entry.update(extra) + return entry + + +def migrate_hidden_entries_from_js() -> list[dict[str, Any]]: + source = HIDDEN_DETAILS_JS.read_text(encoding="utf-8") + family_layers = js_literal_to_json(find_js_const_literal(source, "familyLayers")) + details = js_literal_to_json(find_js_const_literal(source, "details")) + poems = js_literal_to_json(find_js_const_literal(source, "poems")) + greetings = js_literal_to_json(find_js_const_literal(source, "greetings")) + night_messages = js_literal_to_json(find_js_const_literal(source, "nightMessages")) + lore = js_literal_to_json(find_js_const_literal(source, "lore")) + search_toasts = js_literal_to_json(find_js_const_literal(source, "SEARCH_TOASTS")) + search_routes = js_literal_to_json(find_js_const_literal(source, "SEARCH_ROUTES")) + keyboard_secrets = js_literal_to_json(find_js_const_literal(source, "KEYBOARD_SECRETS")) + long_keyboard_secrets = js_literal_to_json(find_js_const_literal(source, "LONG_KEYBOARD_SECRETS")) + + entries: list[dict[str, Any]] = [] + add = entries.append + for index, item in enumerate(family_layers): + add(make_hidden_entry("family layer", item, index, familyLayer=str(index), category="Family Layer Index")) + for index, item in enumerate(details): + add(make_hidden_entry("hidden tooltip", item, index, category="footer and whispers")) + for index, item in enumerate(poems): + add(make_hidden_entry("poem", item, index, category="poems", emotionalTone="soft")) + for index, item in enumerate(greetings): + add(make_hidden_entry("loading screen message", item, index, category="homepage greeting")) + for index, item in enumerate(night_messages): + add(make_hidden_entry("rare event", item, index, category="late night", rarity="timed", triggerConditions="hour >= 22 or hour < 5")) + for key, content_type in [ + ("quotes", "quote"), + ("journals", "journal entry"), + ("warnings", "fake error message"), + ("dreams", "dream sequence"), + ("cassettes", "terminal log"), + ("fakeUsers", "guestbook entry"), + ("homepageTakeovers", "rare event"), + ]: + for index, item in enumerate(lore.get(key, [])): + add(make_hidden_entry(content_type, item, index, category=key, rarity="rare" if key in {"warnings", "dreams", "homepageTakeovers"} else "common")) + for index, item in enumerate(lore.get("conversations", [])): + title = " / ".join(str(part) for part in item[::2]) or f"Conversation {index + 1}" + add(make_hidden_entry("hidden conversation", "\n".join(str(part) for part in item), index, title=title, category="conversations", dialogue=item)) + for index, (season, item) in enumerate((lore.get("seasonal", {}) or {}).items()): + add(make_hidden_entry("seasonal event", item, index, title=f"{season.title()} note", category="seasonal", rarity="seasonal", triggerConditions=f"season is {season}", season=season)) + for index, item in enumerate(lore.get("roomLinks", [])): + href, label = item + add(make_hidden_entry("secret interaction", label, index, title=label, category="room links", pageLocation=href, triggerConditions="secret link injected into page")) + for index, (query, message) in enumerate(search_toasts.items()): + add(make_hidden_entry("search toast", message, index, title=f"Search: {query}", category="search", triggerConditions=query, query=query)) + for index, (query, route) in enumerate(search_routes.items()): + add(make_hidden_entry("search route", route, index, title=f"Route: {query}", category="search", pageLocation=route, triggerConditions=query, query=query)) + for index, item in enumerate(keyboard_secrets + long_keyboard_secrets): + content = item.get("message") or item.get("route") or ("play tiny aphy song" if item.get("song") else "") + add(make_hidden_entry("keyboard secret", content, index, title=f"Keyboard: {item.get('phrase')}", category="keyboard", pageLocation=item.get("route", ""), triggerConditions=item.get("phrase", ""), keyboard=item)) + return entries + + +def normalize_hidden_entry(raw: dict[str, Any], existing: dict[str, Any] | None = None) -> dict[str, Any]: + today = datetime.now().date().isoformat() + entry = existing.copy() if existing else {} + entry.update(raw) + title = str(normalize_character_text_refs(entry.get("title") or "")).strip() + content = str(normalize_character_text_refs(entry.get("content") or "")).replace("\r\n", "\n") + content_type = str(entry.get("type") or "quote").strip() + content_type = str(normalize_character_text_refs(content_type)) + if content_type not in HIDDEN_CONTENT_TYPES: + raise ValueError(f"Unsupported hidden content type: {content_type}") + if not title: + raise ValueError("Every hidden entry needs a title.") + if not content and content_type not in {"search route"}: + raise ValueError(f"{title} needs content.") + entry["id"] = slugify(str(entry.get("id") or title)) + entry["title"] = title + entry["type"] = content_type + entry["content"] = content + detected_characters = detect_hidden_characters(" ".join([ + title, + content, + str(entry.get("triggerConditions") or ""), + str(entry.get("notes") or ""), + str(entry.get("category") or ""), + ])) + entry["characters"] = normalize_character_list(entry.get("characters", [])) + for character in detected_characters: + if character not in entry["characters"]: + entry["characters"].append(character) + if not entry["characters"]: + entry["characters"] = ["z"] + entry["emotionalTone"] = str(entry.get("emotionalTone") or "warm") + entry["rarity"] = str(entry.get("rarity") or "common") + entry["triggerConditions"] = str(normalize_character_text_refs(entry.get("triggerConditions") or "")) + entry["tags"] = normalise_tags(entry.get("tags", [])) + entry["category"] = str(normalize_character_text_refs(entry.get("category") or content_type)) + entry["pageLocation"] = str(entry.get("pageLocation") or "") + entry["familyLayer"] = str(entry.get("familyLayer") or "") + entry["enabled"] = bool(entry.get("enabled", True)) + entry["createdDate"] = str(entry.get("createdDate") or today) + entry["modifiedDate"] = today + entry["notes"] = str(normalize_character_text_refs(entry.get("notes") or "")) + entry["audioSettings"] = str(entry.get("audioSettings") or "") + entry["animationTrigger"] = str(entry.get("animationTrigger") or "") + entry["cssClassHooks"] = str(entry.get("cssClassHooks") or "") + entry["chainReferences"] = [str(item).strip() for item in entry.get("chainReferences", []) if str(item).strip()] + entry["continuationLinks"] = [str(item).strip() for item in entry.get("continuationLinks", []) if str(item).strip()] + entry["emotionalRole"] = str(entry.get("emotionalRole") or "") + entry["discoveryDifficulty"] = str(entry.get("discoveryDifficulty") or "gentle") + entry["mysteryLevel"] = str(entry.get("mysteryLevel") or "quiet") + try: + entry["resonanceScore"] = max(1, min(10, int(entry.get("resonanceScore") or 3))) + except (TypeError, ValueError): + entry["resonanceScore"] = 3 + for key in [ + "symbols", + "narrativeArcs", + "parentLinks", + "childLinks", + "echoes", + "mirroredEntries", + "thematicLinks", + "symbolicLinks", + "triggerLinks", + ]: + entry[key] = [str(normalize_character_text_refs(item)).strip() for item in entry.get(key, []) if str(item).strip()] + for key in ["dialogue", "keyboard"]: + if key in entry: + entry[key] = normalize_character_text_refs(entry[key]) + return entry + + +def normalize_hidden_story(raw: dict[str, Any], entries: list[dict[str, Any]] | None = None, existing: dict[str, Any] | None = None) -> dict[str, Any]: + today = datetime.now().date().isoformat() + story = existing.copy() if existing else {} + story.update(raw) + title = str(normalize_character_text_refs(story.get("title") or "")).strip() + if not title: + raise ValueError("Every story needs a title.") + ids = {entry["id"] for entry in entries or []} + raw_nodes = story.get("nodes", []) + if isinstance(raw_nodes, str): + raw_nodes = re.split(r"[,:\s]+", raw_nodes) + nodes = [] + for item in raw_nodes: + node_id = str(item).strip() + if node_id and node_id not in nodes and (not ids or node_id in ids): + nodes.append(node_id) + characters = normalize_character_list(story.get("characters", [])) + symbols = [str(normalize_character_text_refs(item)).strip() for item in story.get("symbols", []) if str(item).strip()] + markers = [slugify(str(item)) for item in story.get("markers", []) if str(item).strip()] + layer_affinity = [] + for item in story.get("layerAffinity", []): + match = re.search(r"[0-5]", str(item)) + if match and int(match.group(0)) not in layer_affinity: + layer_affinity.append(int(match.group(0))) + if not layer_affinity and nodes: + by_id = {entry["id"]: entry for entry in entries or []} + layer_affinity = sorted({ + int(match.group(0)) + for node in nodes + if (match := re.search(r"[0-5]", str(by_id.get(node, {}).get("familyLayer", "")))) + }) + story_id = slugify(str(story.get("id") or f"story-{title}")) + if not story_id.startswith("story-"): + story_id = f"story-{story_id}" + return { + "id": story_id, + "title": title, + "description": str(normalize_character_text_refs(story.get("description") or "")), + "tone": str(story.get("tone") or "warm"), + "characters": characters, + "symbols": symbols, + "nodes": nodes, + "discoveryStyle": str(story.get("discoveryStyle") or "gradual"), + "layerAffinity": layer_affinity, + "unlockConditions": [str(normalize_character_text_refs(item)).strip() for item in story.get("unlockConditions", []) if str(item).strip()], + "hidden": bool(story.get("hidden", False)), + "markers": markers, + "createdDate": str(story.get("createdDate") or today), + "modifiedDate": today, + } + + +def migrate_hidden_stories(entries: list[dict[str, Any]]) -> list[dict[str, Any]]: + story_groups: dict[str, dict[str, Any]] = {} + + def add(group: str, entry: dict[str, Any], source: str) -> None: + if not group: + return + key = slugify(group) + if key not in story_groups: + story_groups[key] = { + "id": f"story-{key}", + "title": group.replace("-", " "), + "description": f"Migrated from old {source} relationships into a calmer story path.", + "tone": entry.get("emotionalTone") or "warm", + "characters": [], + "symbols": [], + "nodes": [], + "discoveryStyle": "gradual", + "layerAffinity": [], + "unlockConditions": [], + "hidden": False, + "markers": ["emotional"], + } + story = story_groups[key] + if entry["id"] not in story["nodes"]: + story["nodes"].append(entry["id"]) + for character in entry.get("characters", []): + if character not in story["characters"]: + story["characters"].append(character) + for symbol in entry.get("symbols", []): + if symbol not in story["symbols"]: + story["symbols"].append(symbol) + layer = re.search(r"[0-5]", str(entry.get("familyLayer", ""))) + if layer and int(layer.group(0)) not in story["layerAffinity"]: + story["layerAffinity"].append(int(layer.group(0))) + + by_id = {entry["id"]: entry for entry in entries} + for entry in entries: + for arc in entry.get("narrativeArcs", []): + add(str(arc), entry, "arc") + for symbol in entry.get("symbols", []): + add(str(symbol), entry, "symbol") + for target in entry.get("continuationLinks", []) + entry.get("chainReferences", []): + if target in by_id: + name = (entry.get("narrativeArcs") or entry.get("symbols") or [entry.get("emotionalTone") or "quiet return"])[0] + add(str(name), entry, "continuation") + add(str(name), by_id[target], "continuation") + if not story_groups: + for character in HIDDEN_CHARACTERS: + character_entries = [entry for entry in entries if character in entry.get("characters", [])][:12] + if character_entries: + for entry in character_entries: + add(f"{character} stories", entry, "character territory") + return [normalize_hidden_story(story, entries) for story in story_groups.values() if len(story.get("nodes", [])) >= 2][:36] + + +def load_hidden_store() -> dict[str, Any]: + migrated = False + if HIDDEN_CONTENT_JSON.exists(): + data = json.loads(HIDDEN_CONTENT_JSON.read_text(encoding="utf-8")) + entries = data.get("entries", []) + stories = data.get("stories", []) + else: + entries = migrate_hidden_entries_from_js() + stories = [] + migrated = True + data = { + "schemaVersion": 2, + "generatedFrom": "assets/scripts/hidden-details.js", + "generatedAt": datetime.now().isoformat(timespec="seconds"), + "entries": entries, + "stories": stories, + } + normalized = [normalize_hidden_entry(entry, entry) for entry in entries] + normalized_stories = [normalize_hidden_story(story, normalized, story) for story in stories] + migrated_relationships = False + if not normalized_stories: + normalized_stories = migrate_hidden_stories(normalized) + migrated_relationships = bool(normalized_stories) + return { + "schemaVersion": 2, + "source": HIDDEN_DETAILS_JS.relative_to(ROOT).as_posix(), + "contentPath": HIDDEN_CONTENT_JSON.relative_to(ROOT).as_posix(), + "migratedFromJs": migrated, + "migratedRelationshipsToStories": migrated_relationships, + "types": HIDDEN_CONTENT_TYPES, + "characters": HIDDEN_CHARACTERS, + "characterRegistry": CHARACTER_REGISTRY, + "validation": validate_hidden_integrity(normalized, normalized_stories), + "tones": HIDDEN_TONES, + "rarities": HIDDEN_RARITIES, + "storyMarkers": HIDDEN_STORY_MARKERS, + "discoveryStyles": HIDDEN_DISCOVERY_STYLES, + "layers": HIDDEN_LAYER_DEPTHS, + "entries": normalized, + "stories": normalized_stories, + "recommendations": hidden_architecture_recommendations(), + } + + +def validate_hidden_integrity(entries: list[dict[str, Any]], stories: list[dict[str, Any]] | None = None) -> dict[str, Any]: + ids = {entry["id"] for entry in entries} + unknown_characters = [] + orphan_nodes = [] + stale_links = [] + stale_story_nodes = [] + for entry in entries: + characters = entry.get("characters", []) + if not characters: + orphan_nodes.append(entry["id"]) + for character in characters: + if character not in CHARACTER_REGISTRY: + unknown_characters.append({"entry": entry["id"], "character": character}) + for key in ["chainReferences", "continuationLinks", "parentLinks", "childLinks", "echoes", "mirroredEntries", "thematicLinks", "symbolicLinks", "triggerLinks"]: + for target in entry.get(key, []): + if target and target not in ids and not str(target).startswith("/"): + stale_links.append({"entry": entry["id"], "field": key, "target": target}) + for story in stories or []: + for character in story.get("characters", []): + if character not in CHARACTER_REGISTRY: + unknown_characters.append({"story": story["id"], "character": character}) + for node_id in story.get("nodes", []): + if node_id not in ids: + stale_story_nodes.append({"story": story["id"], "target": node_id}) + return { + "ok": not unknown_characters and not orphan_nodes and not stale_links and not stale_story_nodes, + "unknownCharacters": unknown_characters[:50], + "orphanNodes": orphan_nodes[:50], + "staleLinks": stale_links[:50], + "staleStoryNodes": stale_story_nodes[:50], + "summary": { + "unknownCharacterCount": len(unknown_characters), + "orphanNodeCount": len(orphan_nodes), + "staleLinkCount": len(stale_links), + "staleStoryNodeCount": len(stale_story_nodes), + }, + "repairPolicy": "Unknown aliases are normalized through the registry. Empty character lists are repaired to z. Old relationship links are retained as legacy data, but new authoring should happen through stories.", + } + + +def hidden_architecture_recommendations() -> dict[str, Any]: + return { + "storage": "Use assets/content/hidden-details.json as the friendly source of truth for memories and first-class stories, then regenerate the editable constants in assets/scripts/hidden-details.js.", + "backups": "Every save writes timestamped backups for both JSON and JS under backups/hidden-details/.", + "versioning": "Commit the JSON and generated JS together so the live site and authoring history stay aligned.", + "collaboration": "For simultaneous editing, resolve conflicts by memory id and story id rather than by whole-file ownership.", + "scalability": "Stories are the primary emotional routes. Legacy relationship fields remain readable for migration, but the observatory should stay story-first.", + } + + +def hidden_entries_by_type(entries: list[dict[str, Any]], content_type: str) -> list[dict[str, Any]]: + return [entry for entry in entries if entry.get("enabled", True) and entry.get("type") == content_type] + + +def hidden_contents(entries: list[dict[str, Any]], content_type: str) -> list[str]: + return [str(entry.get("content", "")) for entry in hidden_entries_by_type(entries, content_type)] + + +def generated_hidden_content_block(entries: list[dict[str, Any]]) -> str: + family_layers = hidden_contents(entries, "family layer") + details = hidden_contents(entries, "hidden tooltip") + hidden_contents(entries, "hover message") + poems = hidden_contents(entries, "poem") + greetings = hidden_contents(entries, "loading screen message") + night_messages = [ + entry["content"] for entry in hidden_entries_by_type(entries, "rare event") + if "night" in f"{entry.get('category', '')} {entry.get('triggerConditions', '')}".lower() + ] + quote_like = hidden_contents(entries, "quote") + hidden_contents(entries, "sensei chi wisdom entry") + hidden_contents(entries, "aphy system message") + hidden_contents(entries, "lima note/message") + hidden_contents(entries, "future z message") + hidden_contents(entries, "young z memory fragment") + lore = { + "quotes": quote_like, + "conversations": [ + entry.get("dialogue") if isinstance(entry.get("dialogue"), list) else [line for line in str(entry.get("content", "")).splitlines() if line.strip()] + for entry in hidden_entries_by_type(entries, "hidden conversation") + ], + "journals": hidden_contents(entries, "journal entry"), + "warnings": hidden_contents(entries, "fake error message"), + "dreams": hidden_contents(entries, "dream sequence"), + "cassettes": hidden_contents(entries, "terminal log"), + "fakeUsers": hidden_contents(entries, "guestbook entry"), + "seasonal": { + str(entry.get("season") or entry.get("triggerConditions") or entry.get("title", "")).lower().replace("season is ", ""): entry.get("content", "") + for entry in hidden_entries_by_type(entries, "seasonal event") + }, + "homepageTakeovers": [ + entry["content"] for entry in hidden_entries_by_type(entries, "rare event") + if "homepage" in f"{entry.get('category', '')} {entry.get('triggerConditions', '')}".lower() + ], + "roomLinks": [ + [entry.get("pageLocation", ""), entry.get("content", "") or entry.get("title", "")] + for entry in hidden_entries_by_type(entries, "secret interaction") + if entry.get("pageLocation") + ], + } + search_toasts = { + entry.get("query") or entry.get("triggerConditions") or entry.get("title", ""): entry.get("content", "") + for entry in hidden_entries_by_type(entries, "search toast") + } + search_routes = { + entry.get("query") or entry.get("triggerConditions") or entry.get("title", ""): entry.get("pageLocation") or entry.get("content", "") + for entry in hidden_entries_by_type(entries, "search route") + } + keyboard_secrets = [] + long_keyboard_secrets = [] + for entry in hidden_entries_by_type(entries, "keyboard secret"): + secret = dict(entry.get("keyboard") or {}) + secret["phrase"] = secret.get("phrase") or entry.get("triggerConditions") or entry.get("title", "") + if entry.get("pageLocation"): + secret["route"] = entry.get("pageLocation") + elif entry.get("content") == "play tiny aphy song": + secret["song"] = True + else: + secret["message"] = entry.get("content", "") + target = long_keyboard_secrets if len(secret["phrase"]) > 8 or " " in secret["phrase"] else keyboard_secrets + target.append(secret) + + def js_const(name: str, value: Any) -> str: + return f" const {name} = {json.dumps(value, ensure_ascii=False, indent=4)};\n" + + return ( + " // -----------------------------\n" + " // EDITABLE CONTENT\n" + " // -----------------------------\n" + " // Generated by the hidden narrative authoring page.\n" + " // Friendly source of truth: assets/content/hidden-details.json\n\n" + f"{js_const('familyLayers', family_layers)}\n" + f"{js_const('details', details)}\n" + f"{js_const('poems', poems)}\n" + f"{js_const('greetings', greetings)}\n" + f"{js_const('nightMessages', night_messages)}\n" + f"{js_const('lore', lore)}\n" + " // Exact search text -> toast message.\n" + f"{js_const('SEARCH_TOASTS', search_toasts)}\n" + " // Exact search text -> hidden page route.\n" + f"{js_const('SEARCH_ROUTES', search_routes)}\n" + " // Short typed phrases. Stored in sessionStorage as a rolling key chain.\n" + f"{js_const('KEYBOARD_SECRETS', keyboard_secrets)}\n" + " // Longer typed phrases and character names.\n" + f"{js_const('LONG_KEYBOARD_SECRETS', long_keyboard_secrets)}\n" + ) + + +def backup_hidden_file(path: Path, stamp: str) -> None: + if not path.exists(): + return + HIDDEN_BACKUP_DIR.mkdir(parents=True, exist_ok=True) + target = HIDDEN_BACKUP_DIR / f"{stamp}-{path.name}" + target.write_text(path.read_text(encoding="utf-8"), encoding="utf-8") + + +def replace_hidden_editable_block(source: str, entries: list[dict[str, Any]]) -> str: + start = source.find(" // -----------------------------\n // EDITABLE CONTENT") + end = source.find(" // -----------------------------\n // STATE HELPERS") + if start == -1 or end == -1 or end <= start: + raise ValueError("Could not find editable hidden content block.") + return source[:start] + generated_hidden_content_block(entries) + "\n" + source[end:] + + +def save_hidden_store(payload: dict[str, Any]) -> dict[str, Any]: + loaded = load_hidden_store() + current = {entry["id"]: entry for entry in loaded["entries"]} + entries = [normalize_hidden_entry(entry, current.get(str(entry.get("id", "")))) for entry in payload.get("entries", [])] + current_stories = {story["id"]: story for story in loaded.get("stories", [])} + stories = [normalize_hidden_story(story, entries, current_stories.get(str(story.get("id", "")))) for story in payload.get("stories", loaded.get("stories", []))] + ids = [entry["id"] for entry in entries] + if len(ids) != len(set(ids)): + raise ValueError("Entry ids must be unique.") + story_ids = [story["id"] for story in stories] + if len(story_ids) != len(set(story_ids)): + raise ValueError("Story ids must be unique.") + stamp = datetime.now().strftime("%Y%m%d-%H%M%S") + backup_hidden_file(HIDDEN_DETAILS_JS, stamp) + backup_hidden_file(HIDDEN_CONTENT_JSON, stamp) + HIDDEN_CONTENT_JSON.parent.mkdir(parents=True, exist_ok=True) + HIDDEN_CONTENT_JSON.write_text( + json.dumps( + { + "schemaVersion": 2, + "generatedAt": datetime.now().isoformat(timespec="seconds"), + "entries": entries, + "stories": stories, + }, + ensure_ascii=False, + indent=2, + ) + "\n", + encoding="utf-8", + ) + source = HIDDEN_DETAILS_JS.read_text(encoding="utf-8") + HIDDEN_DETAILS_JS.write_text(replace_hidden_editable_block(source, entries), encoding="utf-8") + saved = load_hidden_store() + saved["message"] = "stored safely. future-you will probably smile at this one." + saved["backupStamp"] = stamp + return saved + + diff --git a/src/authoring_service/models.py b/src/authoring_service/models.py new file mode 100644 index 0000000..32334c7 --- /dev/null +++ b/src/authoring_service/models.py @@ -0,0 +1,34 @@ +"""Shared data models.""" + +from __future__ import annotations + +from dataclasses import dataclass + +@dataclass +class OrgPage: + path: str + page_type: str + title: str + slug: str + tags: list[str] + content: str + date: str + comments: bool + options: str + wip: str | None = None + + +@dataclass +class ContentPage: + path: str + page_type: str + title: str + slug: str + tags: list[str] + content: str + date: str + comments: bool + options: str + format: str + wip: str | None = None + diff --git a/src/authoring_service/templates.py b/src/authoring_service/templates.py new file mode 100644 index 0000000..1331ccd --- /dev/null +++ b/src/authoring_service/templates.py @@ -0,0 +1,3490 @@ +"""HTML templates served by the local authoring UI.""" + +from __future__ import annotations + +APP_HTML = r""" + + + + + + Org Site Authoring + + + +
+ +
+
+
+

New page

+ +
+
+ + + + + + + +
+
+ + + + + + + + + + + + + +
+ +
+ + +
+ +
+ + + +
+
+

Build queue

+
+

Latest build log

+

+    
+
+ + + +""" + + +HIDDEN_APP_HTML = r""" + + + + + Hidden Memory Observatory + + + +
+ + +
+
+
+

Emotional Graph

+
Layer 0 lives near the edge. Layer 5 rests at the quiet core.
+
+
+ + + + + + + + + + +
+
+ +
+ +
Click a story to open its route. Drag memories in the Story Builder to shape the path.
+ +
Atlas
+
+ + +
+ + + + + +""" + diff --git a/src/authoring_service/utils.py b/src/authoring_service/utils.py new file mode 100644 index 0000000..1877076 --- /dev/null +++ b/src/authoring_service/utils.py @@ -0,0 +1,57 @@ +"""Small formatting and parsing helpers.""" + +from __future__ import annotations + +import re +from datetime import datetime +from typing import Any + +def slugify(value: str) -> str: + slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") + return slug or "untitled" + + +def normalise_tags(value: Any) -> list[str]: + if isinstance(value, str): + raw = re.split(r"[,:\s]+", value) + elif isinstance(value, list): + raw = [str(item) for item in value] + else: + raw = [] + tags = [] + for tag in raw: + if not tag.strip(): + continue + clean = slugify(tag) + if clean and clean not in tags: + tags.append(clean) + return tags + + +def org_date(dt: datetime) -> str: + return dt.strftime("<%Y-%m-%d %a %H:%M>") + + +def parse_org_datetime(value: str | None) -> datetime | None: + if not value: + return None + match = re.search(r"<(\d{4})-(\d{2})-(\d{2})(?:\s+\w+)?(?:\s+(\d{2}):(\d{2}))?>", value) + if not match: + return None + year, month, day, hour, minute = match.groups() + return datetime( + int(year), + int(month), + int(day), + int(hour or 12), + int(minute or 0), + ) + + +def html_escape(value: str) -> str: + return ( + value.replace("&", "&") + .replace("<", "<") + .replace(">", ">") + .replace('"', """) + ) diff --git a/src/authoring_service/web.py b/src/authoring_service/web.py new file mode 100644 index 0000000..c822784 --- /dev/null +++ b/src/authoring_service/web.py @@ -0,0 +1,156 @@ +"""HTTP handler and server entry point.""" + +from __future__ import annotations + +import json +import mimetypes +import os +import sys +import time +from email.utils import formatdate +from http import HTTPStatus +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import Any +from urllib.parse import parse_qs, urlparse + +from .build import BUILD_QUEUE, queue_build, queue_hidden_build +from .config import ROOT +from .content import list_pages, page_to_dict, parse_upload_form, read_page, safe_relative_path, save_page, save_upload, server_diagnostics +from .hidden import load_hidden_store, save_hidden_store +from .templates import APP_HTML, HIDDEN_APP_HTML + +class Handler(BaseHTTPRequestHandler): + server_version = "OrgAuthoring/1.0" + + def log_message(self, fmt: str, *args: Any) -> None: + sys.stderr.write("%s - %s\n" % (formatdate(time.time()), fmt % args)) + + def send_json(self, data: Any, status: HTTPStatus = HTTPStatus.OK) -> None: + body = json.dumps(data).encode("utf-8") + self.send_response(status) + self.send_header("Content-Type", "application/json; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(body) + + def do_GET(self) -> None: + parsed = urlparse(self.path) + if parsed.path == "/": + body = APP_HTML.encode("utf-8") + self.send_response(HTTPStatus.OK) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + return + if parsed.path == "/hidden": + body = HIDDEN_APP_HTML.encode("utf-8") + self.send_response(HTTPStatus.OK) + self.send_header("Content-Type", "text/html; charset=utf-8") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + return + if parsed.path == "/api/pages": + try: + self.send_json(list_pages()) + except Exception as exc: + self.send_json({"error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) + return + if parsed.path == "/api/diagnostics": + try: + self.send_json(server_diagnostics()) + except Exception as exc: + self.send_json({"error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) + return + if parsed.path == "/api/page": + query = parse_qs(parsed.query) + try: + path = safe_relative_path(query.get("path", [""])[0]) + self.send_json(page_to_dict(read_page(path))) + except Exception as exc: + self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) + return + if parsed.path == "/api/build": + self.send_json(BUILD_QUEUE.snapshot()) + return + if parsed.path == "/api/hidden": + try: + self.send_json(load_hidden_store()) + except Exception as exc: + self.send_json({"error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) + return + if parsed.path.startswith("/assets/avatars/"): + try: + relative = Path(parsed.path.lstrip("/")) + if relative.is_absolute() or ".." in relative.parts: + self.send_error(HTTPStatus.NOT_FOUND) + return + asset_path = (ROOT / relative).resolve() + avatars_root = (ROOT / "assets" / "avatars").resolve() + if not asset_path.is_file() or avatars_root not in asset_path.parents: + self.send_error(HTTPStatus.NOT_FOUND) + return + body = asset_path.read_bytes() + self.send_response(HTTPStatus.OK) + self.send_header("Content-Type", mimetypes.guess_type(asset_path.name)[0] or "application/octet-stream") + self.send_header("Content-Length", str(len(body))) + self.send_header("Cache-Control", "no-store") + self.end_headers() + self.wfile.write(body) + except Exception: + self.send_error(HTTPStatus.NOT_FOUND) + return + self.send_error(HTTPStatus.NOT_FOUND) + + def do_POST(self) -> None: + if self.path == "/api/upload": + try: + length = int(self.headers.get("Content-Length", "0")) + filename, payload, page_path = parse_upload_form( + self.headers.get("Content-Type", ""), + self.rfile.read(length), + ) + self.send_json(save_upload(filename, payload, page_path)) + except Exception as exc: + self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) + return + if self.path == "/api/hidden": + try: + length = int(self.headers.get("Content-Length", "0")) + data = json.loads(self.rfile.read(length).decode("utf-8")) + saved = save_hidden_store(data) + saved["queuedBuild"] = queue_hidden_build() + self.send_json(saved) + except Exception as exc: + self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) + return + if self.path != "/api/page": + self.send_error(HTTPStatus.NOT_FOUND) + return + try: + length = int(self.headers.get("Content-Length", "0")) + data = json.loads(self.rfile.read(length).decode("utf-8")) + saved = save_page(data) + saved["queuedBuild"] = queue_build(saved) + self.send_json(saved) + except Exception as exc: + self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) + + +def main() -> None: + port = int(os.environ.get("AUTHOR_PORT", "8765")) + server = ThreadingHTTPServer(("127.0.0.1", port), Handler) + page_count = len(list_pages()) + print(f"Authoring UI running at http://127.0.0.1:{port}") + print(f"Content root: {ROOT}") + print(f"Editable pages: {page_count}") + if page_count == 0: + print("WARNING: no editable pages were found. Check AUTHOR_CONTENT_ROOT/AUTHOR_ROOT and the launch directory.") + print("Press Ctrl-C to stop.") + try: + server.serve_forever() + except KeyboardInterrupt: + pass diff --git a/tests/test_authoring_server.py b/src/tests/test_authoring_server.py old mode 100755 new mode 100644 similarity index 89% rename from tests/test_authoring_server.py rename to src/tests/test_authoring_server.py index 918b596..e8c5796 --- a/tests/test_authoring_server.py +++ b/src/tests/test_authoring_server.py @@ -1,11 +1,17 @@ import os +import sys import tempfile import unittest from datetime import datetime from pathlib import Path from unittest import mock -import authoring_server as server +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +import authoring_service.build as build_server +import authoring_service.config as config_server +import authoring_service.content as server +import authoring_service.utils as utils_server class AuthoringServerTestCase(unittest.TestCase): @@ -48,29 +54,29 @@ class UtilityTests(AuthoringServerTestCase): wrong_root.mkdir() (self.root / "authoring_server.py").write_text("", encoding="utf-8") - with mock.patch.dict(os.environ, {"AUTHOR_ROOT": str(wrong_root), "GITHUB_WORKSPACE": ""}), mock.patch.object(server.Path, "cwd", return_value=self.root): - self.assertEqual(server.resolve_root(), self.root) + with mock.patch.dict(os.environ, {"AUTHOR_ROOT": str(wrong_root), "GITHUB_WORKSPACE": ""}), mock.patch.object(config_server.Path, "cwd", return_value=self.root): + self.assertEqual(config_server.resolve_root(), self.root) def test_slugify_normalises_text_and_keeps_fallback(self): - self.assertEqual(server.slugify("Hello, Org Web!"), "hello-org-web") - self.assertEqual(server.slugify(" "), "untitled") + self.assertEqual(utils_server.slugify("Hello, Org Web!"), "hello-org-web") + self.assertEqual(utils_server.slugify(" "), "untitled") def test_normalise_tags_accepts_strings_and_deduplicates(self): self.assertEqual( - server.normalise_tags("Life, review:Life Emacs"), + utils_server.normalise_tags("Life, review:Life Emacs"), ["life", "review", "emacs"], ) def test_parse_org_datetime_handles_date_and_optional_time(self): self.assertEqual( - server.parse_org_datetime("<2026-05-07 Thu 14:35>"), + utils_server.parse_org_datetime("<2026-05-07 Thu 14:35>"), datetime(2026, 5, 7, 14, 35), ) self.assertEqual( - server.parse_org_datetime("<2026-05-07 Thu>"), + utils_server.parse_org_datetime("<2026-05-07 Thu>"), datetime(2026, 5, 7, 12, 0), ) - self.assertIsNone(server.parse_org_datetime("2026-05-07")) + self.assertIsNone(utils_server.parse_org_datetime("2026-05-07")) def test_safe_relative_path_allows_expected_content_roots(self): self.assertEqual( @@ -288,8 +294,8 @@ class PageRenderingTests(AuthoringServerTestCase): class BuildQueueTests(unittest.TestCase): def test_snapshot_reports_recent_completed_job(self): - queue = server.BuildQueue() - job = server.BuildJob(1, "blogs/post.org", "Post") + queue = build_server.BuildQueue() + job = build_server.BuildJob(1, "blogs/post.org", "Post") job.started_at = 1.0 job.finished_at = 2.0 job.ok = True @@ -305,18 +311,18 @@ class BuildQueueTests(unittest.TestCase): self.assertEqual(snapshot["log"], "build log") def test_queue_build_enqueues_from_page_data(self): - queue = server.BuildQueue() - with mock.patch.object(server, "BUILD_QUEUE", queue), mock.patch.object(queue, "_run_worker"): - queued = server.queue_build({"path": "blogs/post.org", "title": "Post"}) + queue = build_server.BuildQueue() + with mock.patch.object(build_server, "BUILD_QUEUE", queue), mock.patch.object(queue, "_run_worker"): + queued = build_server.queue_build({"path": "blogs/post.org", "title": "Post"}) self.assertEqual(queued["id"], 1) self.assertEqual(queued["status"], "queued") self.assertEqual(queued["path"], "blogs/post.org") def test_queue_hidden_build_enqueues_hidden_asset_publish(self): - queue = server.BuildQueue() - with mock.patch.object(server, "BUILD_QUEUE", queue), mock.patch.object(queue, "_run_worker"): - queued = server.queue_hidden_build() + queue = build_server.BuildQueue() + with mock.patch.object(build_server, "BUILD_QUEUE", queue), mock.patch.object(queue, "_run_worker"): + queued = build_server.queue_hidden_build() self.assertEqual(queued["id"], 1) self.assertEqual(queued["status"], "queued") @@ -344,8 +350,8 @@ class BuildQueueTests(unittest.TestCase): FakeProcess(["index line\n"]), ] - with mock.patch.object(server, "ROOT", root), mock.patch.object(server.subprocess, "Popen", side_effect=processes): - ok, message, log = server.run_build_commands(seen.append) + with mock.patch.object(build_server, "ROOT", root), mock.patch.object(build_server.subprocess, "Popen", side_effect=processes): + ok, message, log = build_server.run_build_commands(seen.append) self.assertTrue(ok) self.assertIn("Build complete", message)