breaking code down
All checks were successful
Build Authoring Service / build (push) Successful in 8s

This commit is contained in:
2026-05-13 22:52:47 +01:00
parent 5822ac2c6d
commit 023eb8f70e
14 changed files with 5339 additions and 5219 deletions

View File

@@ -19,7 +19,7 @@ JOURNALCTL_USER := env XDG_RUNTIME_DIR=$(SYSTEMD_RUNTIME_DIR) journalctl --user
all: author-restart all: author-restart
test: $(VENV) 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): $(VENV):
@echo "Generating virtual environment" @echo "Generating virtual environment"

View File

@@ -24,3 +24,9 @@ Useful commands:
- `make author-logs` shows recent service logs. - `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`. 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.

File diff suppressed because it is too large Load Diff

View File

@@ -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

View File

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

View File

@@ -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",
]

View File

@@ -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.",
},
]

View File

@@ -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"<unavailable: {exc}>"
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'<a\b[^>]*>\s*<img\b[^>]*\bsrc="([^"]+)"[^>]*\balt="([^"]*)"[^>]*>\s*</a>',
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("<HH", payload[6:10])
return width, height
if ext in {".jpg", ".jpeg"} and payload.startswith(b"\xff\xd8"):
i = 2
while i + 9 < len(payload):
if payload[i] != 0xFF:
i += 1
continue
marker = payload[i + 1]
i += 2
if marker in {0xD8, 0xD9}:
continue
if i + 2 > 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'<a href="{absolute_url}" data-img="{absolute_url}" data-alt="{alt}" '
f'data-width="{width}" data-height="{height}">'
f'<img src="{relative_url}" alt="{alt}" style="cursor: zoom-in;"></a>'
)
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

View File

@@ -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"(?<![A-Za-z0-9_-])" + re.escape(alias) + r"(?![A-Za-z0-9_-])"
text = re.sub(pattern, canonical, text, flags=re.IGNORECASE)
return text
def detect_hidden_characters(text: str) -> list[str]:
found = []
normalized_text = normalize_character_text_refs(text).lower()
for character in HIDDEN_CHARACTERS:
if re.search(r"(?<![A-Za-z0-9_-])" + re.escape(character) + r"(?![A-Za-z0-9_-])", normalized_text):
found.append(character)
return found
def hidden_entry_id(content_type: str, index: int, title: str) -> 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

View File

@@ -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

File diff suppressed because it is too large Load Diff

View File

@@ -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("&", "&amp;")
.replace("<", "&lt;")
.replace(">", "&gt;")
.replace('"', "&quot;")
)

View File

@@ -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

View File

@@ -1,11 +1,17 @@
import os import os
import sys
import tempfile import tempfile
import unittest import unittest
from datetime import datetime from datetime import datetime
from pathlib import Path from pathlib import Path
from unittest import mock 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): class AuthoringServerTestCase(unittest.TestCase):
@@ -48,29 +54,29 @@ class UtilityTests(AuthoringServerTestCase):
wrong_root.mkdir() wrong_root.mkdir()
(self.root / "authoring_server.py").write_text("", encoding="utf-8") (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): 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(server.resolve_root(), self.root) self.assertEqual(config_server.resolve_root(), self.root)
def test_slugify_normalises_text_and_keeps_fallback(self): def test_slugify_normalises_text_and_keeps_fallback(self):
self.assertEqual(server.slugify("Hello, Org Web!"), "hello-org-web") self.assertEqual(utils_server.slugify("Hello, Org Web!"), "hello-org-web")
self.assertEqual(server.slugify(" "), "untitled") self.assertEqual(utils_server.slugify(" "), "untitled")
def test_normalise_tags_accepts_strings_and_deduplicates(self): def test_normalise_tags_accepts_strings_and_deduplicates(self):
self.assertEqual( self.assertEqual(
server.normalise_tags("Life, review:Life Emacs"), utils_server.normalise_tags("Life, review:Life Emacs"),
["life", "review", "emacs"], ["life", "review", "emacs"],
) )
def test_parse_org_datetime_handles_date_and_optional_time(self): def test_parse_org_datetime_handles_date_and_optional_time(self):
self.assertEqual( 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), datetime(2026, 5, 7, 14, 35),
) )
self.assertEqual( 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), 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): def test_safe_relative_path_allows_expected_content_roots(self):
self.assertEqual( self.assertEqual(
@@ -288,8 +294,8 @@ class PageRenderingTests(AuthoringServerTestCase):
class BuildQueueTests(unittest.TestCase): class BuildQueueTests(unittest.TestCase):
def test_snapshot_reports_recent_completed_job(self): def test_snapshot_reports_recent_completed_job(self):
queue = server.BuildQueue() queue = build_server.BuildQueue()
job = server.BuildJob(1, "blogs/post.org", "Post") job = build_server.BuildJob(1, "blogs/post.org", "Post")
job.started_at = 1.0 job.started_at = 1.0
job.finished_at = 2.0 job.finished_at = 2.0
job.ok = True job.ok = True
@@ -305,18 +311,18 @@ class BuildQueueTests(unittest.TestCase):
self.assertEqual(snapshot["log"], "build log") self.assertEqual(snapshot["log"], "build log")
def test_queue_build_enqueues_from_page_data(self): def test_queue_build_enqueues_from_page_data(self):
queue = server.BuildQueue() queue = build_server.BuildQueue()
with mock.patch.object(server, "BUILD_QUEUE", queue), mock.patch.object(queue, "_run_worker"): with mock.patch.object(build_server, "BUILD_QUEUE", queue), mock.patch.object(queue, "_run_worker"):
queued = server.queue_build({"path": "blogs/post.org", "title": "Post"}) queued = build_server.queue_build({"path": "blogs/post.org", "title": "Post"})
self.assertEqual(queued["id"], 1) self.assertEqual(queued["id"], 1)
self.assertEqual(queued["status"], "queued") self.assertEqual(queued["status"], "queued")
self.assertEqual(queued["path"], "blogs/post.org") self.assertEqual(queued["path"], "blogs/post.org")
def test_queue_hidden_build_enqueues_hidden_asset_publish(self): def test_queue_hidden_build_enqueues_hidden_asset_publish(self):
queue = server.BuildQueue() queue = build_server.BuildQueue()
with mock.patch.object(server, "BUILD_QUEUE", queue), mock.patch.object(queue, "_run_worker"): with mock.patch.object(build_server, "BUILD_QUEUE", queue), mock.patch.object(queue, "_run_worker"):
queued = server.queue_hidden_build() queued = build_server.queue_hidden_build()
self.assertEqual(queued["id"], 1) self.assertEqual(queued["id"], 1)
self.assertEqual(queued["status"], "queued") self.assertEqual(queued["status"], "queued")
@@ -344,8 +350,8 @@ class BuildQueueTests(unittest.TestCase):
FakeProcess(["index line\n"]), FakeProcess(["index line\n"]),
] ]
with mock.patch.object(server, "ROOT", root), mock.patch.object(server.subprocess, "Popen", side_effect=processes): with mock.patch.object(build_server, "ROOT", root), mock.patch.object(build_server.subprocess, "Popen", side_effect=processes):
ok, message, log = server.run_build_commands(seen.append) ok, message, log = build_server.run_build_commands(seen.append)
self.assertTrue(ok) self.assertTrue(ok)
self.assertIn("Build complete", message) self.assertIn("Build complete", message)