Improvements

This commit is contained in:
Zaine
2026-07-09 16:34:28 +01:00
parent d7d0054af4
commit e9e3954f31
20 changed files with 6691 additions and 6686 deletions

View File

@@ -1,22 +1,22 @@
name: Build Authoring Service name: Build Authoring Service
on: on:
push: push:
branches: branches:
- main - main
schedule: schedule:
- cron: "0 0 * * *" - cron: "0 0 * * *"
workflow_dispatch: workflow_dispatch:
jobs: jobs:
build: build:
runs-on: site-build runs-on: site-build
steps: steps:
- name: Check out repo - name: Check out repo
uses: actions/checkout@v4 uses: actions/checkout@v4
- name: Build site - name: Build site
run: make run: make

10
.gitignore vendored
View File

@@ -1,5 +1,5 @@
.venv/ .venv/
__pycache__/ __pycache__/
*.py[cod] *.py[cod]
.pytest_cache/ .pytest_cache/
backups/ backups/

211
Makefile
View File

@@ -1,103 +1,108 @@
.PHONY: test author author-install author-start author-stop author-stop-legacy author-restart author-status author-logs author-health author-diagnostics clean-venv help .PHONY: test author author-install author-start author-stop author-stop-legacy author-restart author-status author-logs author-health author-diagnostics clean-venv help
VENV := .venv VENV := .venv
PY := $(VENV)/bin/python PY := $(VENV)/bin/python
PIP := $(VENV)/bin/pip PIP := $(VENV)/bin/pip
AUTHOR_PORT := 8765 AUTHOR_PORT := 8765
AUTHOR_URL := http://127.0.0.1:$(AUTHOR_PORT) AUTHOR_URL := http://127.0.0.1:$(AUTHOR_PORT)
AUTHOR_PAGES_CHECK := /tmp/authoring-server-pages.json AUTHOR_PAGES_CHECK := /tmp/authoring-server-pages.json
AUTHOR_SERVICE := org-web-authoring.service AUTHOR_SERVICE := org-web-authoring.service
AUTHOR_UNIT_SRC := systemd/user/$(AUTHOR_SERVICE) AUTHOR_UNIT_SRC := systemd/user/$(AUTHOR_SERVICE)
SYSTEMD_USER_DIR := $(HOME)/.config/systemd/user SYSTEMD_USER_DIR := $(HOME)/.config/systemd/user
AUTHOR_UNIT_DST := $(SYSTEMD_USER_DIR)/$(AUTHOR_SERVICE) AUTHOR_UNIT_DST := $(SYSTEMD_USER_DIR)/$(AUTHOR_SERVICE)
USER_ID := $(shell id -u) USER_ID := $(shell id -u)
SYSTEMD_RUNTIME_DIR ?= /run/user/$(USER_ID) SYSTEMD_RUNTIME_DIR ?= /run/user/$(USER_ID)
SYSTEMCTL_USER := env XDG_RUNTIME_DIR=$(SYSTEMD_RUNTIME_DIR) systemctl --user SYSTEMCTL_USER := env XDG_RUNTIME_DIR=$(SYSTEMD_RUNTIME_DIR) systemctl --user
JOURNALCTL_USER := env XDG_RUNTIME_DIR=$(SYSTEMD_RUNTIME_DIR) journalctl --user JOURNALCTL_USER := env XDG_RUNTIME_DIR=$(SYSTEMD_RUNTIME_DIR) journalctl --user
all: author-restart all: author-restart
test: $(VENV) test: $(VENV)
PYTHONPATH=src $(PY) -m unittest discover -s src/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"
python3 -m venv $(VENV) python3 -m venv $(VENV)
$(PIP) install -r requirements.txt $(PIP) install -r requirements.txt
author: author-start author: author-start
author-install: $(VENV) author-install: $(VENV)
@echo "Installing $(AUTHOR_SERVICE) for the current user..." @echo "Installing $(AUTHOR_SERVICE) for the current user..."
@mkdir -p "$(SYSTEMD_USER_DIR)" @mkdir -p "$(SYSTEMD_USER_DIR)"
install -m 0644 "$(AUTHOR_UNIT_SRC)" "$(AUTHOR_UNIT_DST)" install -m 0644 "$(AUTHOR_UNIT_SRC)" "$(AUTHOR_UNIT_DST)"
$(SYSTEMCTL_USER) daemon-reload $(SYSTEMCTL_USER) daemon-reload
$(SYSTEMCTL_USER) enable "$(AUTHOR_SERVICE)" $(SYSTEMCTL_USER) enable "$(AUTHOR_SERVICE)"
author-start: author-install author-stop-legacy author-start: author-install author-stop-legacy
@echo "Starting authoring UI with systemd..." @echo "Starting authoring UI with systemd..."
$(SYSTEMCTL_USER) start "$(AUTHOR_SERVICE)" $(SYSTEMCTL_USER) start "$(AUTHOR_SERVICE)"
@$(MAKE) --no-print-directory author-health @$(MAKE) --no-print-directory author-health
author-stop: author-dev:
@echo "Stopping authoring UI with systemd..." @echo "Starting authoring UI without systemd..."
$(SYSTEMCTL_USER) stop "$(AUTHOR_SERVICE)" || true $(PY) authoring_server.py
@$(MAKE) --no-print-directory author-stop-legacy @$(MAKE) --no-print-directory author-health
author-stop-legacy: author-stop:
@echo "Stopping any legacy process on authoring port $(AUTHOR_PORT)..." @echo "Stopping authoring UI with systemd..."
@fuser -k $(AUTHOR_PORT)/tcp >/dev/null 2>&1 || true $(SYSTEMCTL_USER) stop "$(AUTHOR_SERVICE)" || true
@$(MAKE) --no-print-directory author-stop-legacy
author-restart: author-install author-stop-legacy
@echo "Restarting authoring UI with systemd..." author-stop-legacy:
$(SYSTEMCTL_USER) restart "$(AUTHOR_SERVICE)" @echo "Stopping any legacy process on authoring port $(AUTHOR_PORT)..."
@$(MAKE) --no-print-directory author-health @fuser -k $(AUTHOR_PORT)/tcp >/dev/null 2>&1 || true
author-health: author-restart: author-install author-stop-legacy
@echo "Waiting for authoring UI and page list to respond..." @echo "Restarting authoring UI with systemd..."
@for i in 1 2 3 4 5 6 7 8 9 10; do \ $(SYSTEMCTL_USER) restart "$(AUTHOR_SERVICE)"
if ! $(SYSTEMCTL_USER) is-active --quiet "$(AUTHOR_SERVICE)"; then \ @$(MAKE) --no-print-directory author-health
echo "Authoring UI service is not active. Service status:"; \
$(SYSTEMCTL_USER) status --no-pager "$(AUTHOR_SERVICE)" || true; \ author-health:
echo "Recent logs:"; \ @echo "Waiting for authoring UI and page list to respond..."
$(JOURNALCTL_USER) -u "$(AUTHOR_SERVICE)" -n 80 --no-pager || true; \ @for i in 1 2 3 4 5 6 7 8 9 10; do \
exit 1; \ if ! $(SYSTEMCTL_USER) is-active --quiet "$(AUTHOR_SERVICE)"; then \
fi; \ echo "Authoring UI service is not active. Service status:"; \
if curl -fsS "$(AUTHOR_URL)/api/pages" -o "$(AUTHOR_PAGES_CHECK)" && $(PY) -c 'import json,sys; data=json.load(sys.stdin); sys.exit(0 if isinstance(data, list) and data else 1)' < "$(AUTHOR_PAGES_CHECK)"; then \ $(SYSTEMCTL_USER) status --no-pager "$(AUTHOR_SERVICE)" || true; \
echo "Authoring UI is running on $(AUTHOR_URL)/"; \ echo "Recent logs:"; \
exit 0; \ $(JOURNALCTL_USER) -u "$(AUTHOR_SERVICE)" -n 80 --no-pager || true; \
fi; \ exit 1; \
sleep 1; \ fi; \
done; \ if curl -fsS "$(AUTHOR_URL)/api/pages" -o "$(AUTHOR_PAGES_CHECK)" && $(PY) -c 'import json,sys; data=json.load(sys.stdin); sys.exit(0 if isinstance(data, list) and data else 1)' < "$(AUTHOR_PAGES_CHECK)"; then \
echo "Authoring UI failed to respond. Service status:"; \ echo "Authoring UI is running on $(AUTHOR_URL)/"; \
$(SYSTEMCTL_USER) status --no-pager "$(AUTHOR_SERVICE)" || true; \ exit 0; \
echo "Recent logs:"; \ fi; \
$(JOURNALCTL_USER) -u "$(AUTHOR_SERVICE)" -n 80 --no-pager || true; \ sleep 1; \
exit 1 done; \
echo "Authoring UI failed to respond. Service status:"; \
author-status: $(SYSTEMCTL_USER) status --no-pager "$(AUTHOR_SERVICE)" || true; \
@$(SYSTEMCTL_USER) status --no-pager "$(AUTHOR_SERVICE)" || true echo "Recent logs:"; \
@curl -fsS "$(AUTHOR_URL)/api/pages" -o "$(AUTHOR_PAGES_CHECK)" && $(PY) -c 'import json,sys; print(f"Pages: {len(json.load(sys.stdin))}")' < "$(AUTHOR_PAGES_CHECK)" || true $(JOURNALCTL_USER) -u "$(AUTHOR_SERVICE)" -n 80 --no-pager || true; \
exit 1
author-diagnostics:
@curl -fsS "$(AUTHOR_URL)/api/diagnostics" || true author-status:
@$(SYSTEMCTL_USER) status --no-pager "$(AUTHOR_SERVICE)" || true
author-logs: @curl -fsS "$(AUTHOR_URL)/api/pages" -o "$(AUTHOR_PAGES_CHECK)" && $(PY) -c 'import json,sys; print(f"Pages: {len(json.load(sys.stdin))}")' < "$(AUTHOR_PAGES_CHECK)" || true
$(JOURNALCTL_USER) -u "$(AUTHOR_SERVICE)" -n 100 --no-pager
author-diagnostics:
clean-venv: @curl -fsS "$(AUTHOR_URL)/api/diagnostics" || true
@echo "Cleaning virtual environment..."
rm -rf $(VENV) author-logs:
$(JOURNALCTL_USER) -u "$(AUTHOR_SERVICE)" -n 100 --no-pager
help:
@echo "Available targets:" clean-venv:
@echo " make test - Run authoring server tests" @echo "Cleaning virtual environment..."
@echo " make author - Install and start the authoring UI service" rm -rf $(VENV)
@echo " make author-install - Install and enable the authoring UI service"
@echo " make author-stop - Stop the authoring UI service" help:
@echo " make author-restart - Restart the authoring UI service" @echo "Available targets:"
@echo " make author-status - Show authoring UI service status and page count" @echo " make test - Run authoring server tests"
@echo " make author-diagnostics - Show authoring UI runtime diagnostics" @echo " make author - Install and start the authoring UI service"
@echo " make author-logs - Show recent authoring UI service logs" @echo " make author-install - Install and enable the authoring UI service"
@echo " make clean-venv - Remove virtual environment" @echo " make author-stop - Stop the authoring UI service"
@echo " make author-restart - Restart the authoring UI service"
@echo " make author-status - Show authoring UI service status and page count"
@echo " make author-diagnostics - Show authoring UI runtime diagnostics"
@echo " make author-logs - Show recent authoring UI service logs"
@echo " make clean-venv - Remove virtual environment"

View File

@@ -1,33 +1,33 @@
## Authoring service ## Authoring service
Standalone local authoring UI for the Org website content in: Standalone local authoring UI for the Org website content in:
`/home/zaine/master-folder/org-platform/org_web` `/home/zaine/master-folder/org-platform/org_web`
Run: Run:
```sh ```sh
make author make author
``` ```
Then open `http://127.0.0.1:8765`. Then open `http://127.0.0.1:8765`.
The service is installed as the user systemd unit `org-web-authoring.service`. The service is installed as the user systemd unit `org-web-authoring.service`.
Useful commands: Useful commands:
- `make test` runs the authoring server unit tests. - `make test` runs the authoring server unit tests.
- `make author-install` installs and enables the user service. - `make author-install` installs and enables the user service.
- `make author-restart` restarts the service. - `make author-restart` restarts the service.
- `make author-status` shows service status and editable page count. - `make author-status` shows service status and editable page count.
- `make author-diagnostics` shows runtime diagnostics. - `make author-diagnostics` shows runtime diagnostics.
- `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: Code layout:
- `authoring_server.py` is the compatibility executable used by systemd. - `authoring_server.py` is the compatibility executable used by systemd.
- `src/authoring_service/` contains the implementation modules. - `src/authoring_service/` contains the implementation modules.
- `src/tests/` contains the unit tests. - `src/tests/` contains the unit tests.
- `docs/hidden-memory-architecture.md` documents the Fragment/Memory/Story model used by the Hidden Memory Observatory. - `docs/hidden-memory-architecture.md` documents the Fragment/Memory/Story model used by the Hidden Memory Observatory.

View File

@@ -1,6 +1,6 @@
{ {
"schemaVersion": 3, "schemaVersion": 3,
"generatedAt": "2026-06-17T20:42:16", "generatedAt": "2026-07-09T16:33:00",
"entries": [], "entries": [],
"stories": [] "stories": []
} }

View File

@@ -1,18 +1,18 @@
#!/usr/bin/env python3 #!/usr/bin/env python3
"""Compatibility entry point for the authoring service.""" """Compatibility entry point for the authoring service."""
from __future__ import annotations from __future__ import annotations
from pathlib import Path from pathlib import Path
import sys import sys
SRC = Path(__file__).resolve().parent / "src" SRC = Path(__file__).resolve().parent / "src"
if str(SRC) not in sys.path: if str(SRC) not in sys.path:
sys.path.insert(0, str(SRC)) sys.path.insert(0, str(SRC))
from authoring_service import * # noqa: F401,F403 from authoring_service import * # noqa: F401,F403
from authoring_service.web import main from authoring_service.web import main
if __name__ == "__main__": if __name__ == "__main__":
main() main()

View File

@@ -1,82 +1,82 @@
# Hidden Memory Architecture # Hidden Memory Architecture
The hidden ecosystem has one friendly source of truth: The hidden ecosystem has one friendly source of truth:
- `assets/content/hidden-details.json` in `org_web` - `assets/content/hidden-details.json` in `org_web`
The live site still consumes generated constants in: The live site still consumes generated constants in:
- `assets/scripts/hidden-details.js` in `org_web` - `assets/scripts/hidden-details.js` in `org_web`
## Canonical Content ## Canonical Content
The old `type` field is now treated as the runtime delivery kind. It answers: where does this need to go in the existing JavaScript arrays? The old `type` field is now treated as the runtime delivery kind. It answers: where does this need to go in the existing JavaScript arrays?
The newer `contentClass` field answers: what is this thing conceptually? The newer `contentClass` field answers: what is this thing conceptually?
Use these classes: Use these classes:
- `fragment`: tiny reusable language, such as tooltips, quotes, whispers, loading lines, and small dialogue. - `fragment`: tiny reusable language, such as tooltips, quotes, whispers, loading lines, and small dialogue.
- `memory`: a richer emotional moment or observation that can stand as an observatory node. - `memory`: a richer emotional moment or observation that can stand as an observatory node.
- `story material`: dialogue or sequence material that is primarily useful inside a story route. - `story material`: dialogue or sequence material that is primarily useful inside a story route.
- `interaction`: a trigger, route, keyboard secret, search response, seasonal event, or play-system behavior. - `interaction`: a trigger, route, keyboard secret, search response, seasonal event, or play-system behavior.
- `lore`: world material such as dreams, terminal logs, guestbook entries, or hidden archive objects. - `lore`: world material such as dreams, terminal logs, guestbook entries, or hidden archive objects.
- `system layer`: structural observatory material, such as the layer guide. - `system layer`: structural observatory material, such as the layer guide.
## Stories ## Stories
Stories are curated routes. They should reference entries by id through `nodes`. Stories are curated routes. They should reference entries by id through `nodes`.
Do not duplicate paragraphs inside a story when an existing fragment or memory can be referenced. A story gives order, title, tone, unlock conditions, and emotional shape. Do not duplicate paragraphs inside a story when an existing fragment or memory can be referenced. A story gives order, title, tone, unlock conditions, and emotional shape.
## Surfaces ## Surfaces
The `surfaces` field says where an entry is reusable: `tooltip`, `quote`, `story`, `observatory`, `search`, `keyboard`, `play`, `dream`, `temporal`, and similar surfaces. The `surfaces` field says where an entry is reusable: `tooltip`, `quote`, `story`, `observatory`, `search`, `keyboard`, `play`, `dream`, `temporal`, and similar surfaces.
A tooltip fragment can appear in the footer without becoming a full story node. A memory can appear in the observatory and in a story without being forced into a rotating quote. An interaction can trigger a hidden route without pretending to be a narrative scene. A tooltip fragment can appear in the footer without becoming a full story node. A memory can appear in the observatory and in a story without being forced into a rotating quote. An interaction can trigger a hidden route without pretending to be a narrative scene.
## Observatory Roles ## Observatory Roles
Use `observatoryRole` to keep rendering calm: Use `observatoryRole` to keep rendering calm:
- `ambient`: small lights and flavor, usually fragments. - `ambient`: small lights and flavor, usually fragments.
- `node`: substantial emotional points, usually memories or story material. - `node`: substantial emotional points, usually memories or story material.
- `event`: triggered behavior, routes, search, keyboard, seasonal, and play content. - `event`: triggered behavior, routes, search, keyboard, seasonal, and play content.
- `guide`: layer/system material. - `guide`: layer/system material.
The observatory should render stories as the main territories, then reveal entries inside them. Zoomed-out views should favor routes and clusters; detailed views can show individual nodes, ambient fragments, and triggered events. The observatory should render stories as the main territories, then reveal entries inside them. Zoomed-out views should favor routes and clusters; detailed views can show individual nodes, ambient fragments, and triggered events.
## Examples ## Examples
`lima left this page a little steadier than she found it.` `lima left this page a little steadier than she found it.`
- Class: `fragment` - Class: `fragment`
- Runtime kind: `hidden tooltip` - Runtime kind: `hidden tooltip`
- Surfaces: `tooltip` - Surfaces: `tooltip`
- Observatory role: `ambient` - Observatory role: `ambient`
`July 2022, on the way to uni induction, lima pops into my life` `July 2022, on the way to uni induction, lima pops into my life`
- Class: `memory` - Class: `memory`
- Runtime kind: `journal entry` - Runtime kind: `journal entry`
- Surfaces: `story`, `observatory` - Surfaces: `story`, `observatory`
- Observatory role: `node` - Observatory role: `node`
`search query "lima" opens a hidden route` `search query "lima" opens a hidden route`
- Class: `interaction` - Class: `interaction`
- Runtime kind: `search route` - Runtime kind: `search route`
- Surfaces: `search`, `hidden route` - Surfaces: `search`, `hidden route`
- Observatory role: `event` - Observatory role: `event`
`The story of love between two souls` `The story of love between two souls`
- Class: story record, not an entry class - Class: story record, not an entry class
- References: ordered entry ids in `nodes` - References: ordered entry ids in `nodes`
- Purpose: emotional route, not duplicated content - Purpose: emotional route, not duplicated content
## Authoring Rule ## Authoring Rule
Write the smallest canonical thing that is emotionally honest. Write the smallest canonical thing that is emotionally honest.
If it is one line, make it a fragment. If it is a moment with weight, make it a memory. If it happens because of a trigger, make it an interaction. If it is a route through existing things, make it a story. If it is one line, make it a fragment. If it is a moment with weight, make it a memory. If it happens because of a trigger, make it an interaction. If it is a route through existing things, make it a story.

View File

@@ -1,3 +1,3 @@
beautifulsoup4 beautifulsoup4
lxml lxml
legacy-cgi legacy-cgi

View File

@@ -1,10 +1,10 @@
"""Authoring service package.""" """Authoring service package."""
from .build import BUILD_QUEUE, BuildJob, BuildQueue, queue_build, queue_hidden_build, run_build_commands from .build import BUILD_QUEUE, BuildJob, BuildQueue, queue_build, queue_hidden_build, run_build_commands
from .config import * from .config import *
from .content import * from .content import *
from .hidden import * from .hidden import *
from .models import ContentPage, OrgPage from .models import ContentPage, OrgPage
from .templates import APP_HTML, HIDDEN_APP_HTML from .templates import APP_HTML, HIDDEN_APP_HTML
from .utils import html_escape, normalise_tags, org_date, parse_org_datetime, slugify from .utils import html_escape, normalise_tags, org_date, parse_org_datetime, slugify
from .web import Handler, main from .web import Handler, main

View File

@@ -1,167 +1,167 @@
"""Build queue and publishing command execution.""" """Build queue and publishing command execution."""
from __future__ import annotations from __future__ import annotations
import os import os
import subprocess import subprocess
import sys import sys
import threading import threading
import time import time
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Any, Callable from typing import Any, Callable
from .config import ROOT from .config import ROOT
@dataclass @dataclass
class BuildJob: class BuildJob:
id: int id: int
path: str path: str
title: str title: str
queued_at: float = field(default_factory=time.time) queued_at: float = field(default_factory=time.time)
started_at: float | None = None started_at: float | None = None
finished_at: float | None = None finished_at: float | None = None
ok: bool | None = None ok: bool | None = None
message: str = "Queued" message: str = "Queued"
log: str = "" log: str = ""
@property @property
def status(self) -> str: def status(self) -> str:
if self.finished_at is not None: if self.finished_at is not None:
return "done" if self.ok else "failed" return "done" if self.ok else "failed"
if self.started_at is not None: if self.started_at is not None:
return "running" return "running"
return "queued" return "queued"
def to_dict(self, include_log: bool = False) -> dict[str, Any]: def to_dict(self, include_log: bool = False) -> dict[str, Any]:
data = { data = {
"id": self.id, "id": self.id,
"path": self.path, "path": self.path,
"title": self.title, "title": self.title,
"queuedAt": self.queued_at, "queuedAt": self.queued_at,
"startedAt": self.started_at, "startedAt": self.started_at,
"finishedAt": self.finished_at, "finishedAt": self.finished_at,
"ok": self.ok, "ok": self.ok,
"status": self.status, "status": self.status,
"message": self.message, "message": self.message,
} }
if include_log: if include_log:
data["log"] = self.log[-12000:] data["log"] = self.log[-12000:]
return data return data
class BuildQueue: class BuildQueue:
def __init__(self) -> None: def __init__(self) -> None:
self._lock = threading.Lock() self._lock = threading.Lock()
self._next_id = 1 self._next_id = 1
self._pending: list[BuildJob] = [] self._pending: list[BuildJob] = []
self._current: BuildJob | None = None self._current: BuildJob | None = None
self._recent: list[BuildJob] = [] self._recent: list[BuildJob] = []
self._worker: threading.Thread | None = None self._worker: threading.Thread | None = None
def snapshot(self) -> dict[str, Any]: def snapshot(self) -> dict[str, Any]:
with self._lock: with self._lock:
current = self._current.to_dict(include_log=True) if self._current else None 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:]] recent = [job.to_dict(include_log=True) for job in self._recent[-10:]]
pending = [job.to_dict() for job in self._pending] pending = [job.to_dict() for job in self._pending]
latest = current or (recent[-1] if recent else None) latest = current or (recent[-1] if recent else None)
message = latest["message"] if latest else "No builds have run yet." message = latest["message"] if latest else "No builds have run yet."
return { return {
"running": current is not None, "running": current is not None,
"queued": len(pending), "queued": len(pending),
"message": message, "message": message,
"current": current, "current": current,
"pending": pending, "pending": pending,
"recent": recent, "recent": recent,
"log": latest.get("log", "") if latest else "", "log": latest.get("log", "") if latest else "",
} }
def enqueue(self, path: str, title: str) -> BuildJob: def enqueue(self, path: str, title: str) -> BuildJob:
with self._lock: with self._lock:
job = BuildJob(self._next_id, path, title) job = BuildJob(self._next_id, path, title)
self._next_id += 1 self._next_id += 1
self._pending.append(job) self._pending.append(job)
if self._worker is None or not self._worker.is_alive(): if self._worker is None or not self._worker.is_alive():
self._worker = threading.Thread(target=self._run_worker, daemon=True) self._worker = threading.Thread(target=self._run_worker, daemon=True)
self._worker.start() self._worker.start()
return job return job
def _run_worker(self) -> None: def _run_worker(self) -> None:
while True: while True:
with self._lock: with self._lock:
if not self._pending: if not self._pending:
self._current = None self._current = None
return return
job = self._pending.pop(0) job = self._pending.pop(0)
job.started_at = time.time() job.started_at = time.time()
job.message = "Publishing site and search index." job.message = "Publishing site and search index."
self._current = job self._current = job
def append_log(chunk: str) -> None: def append_log(chunk: str) -> None:
with self._lock: with self._lock:
job.log = (job.log + chunk)[-200000:] job.log = (job.log + chunk)[-200000:]
ok, message, log = run_build_commands(append_log) ok, message, log = run_build_commands(append_log)
with self._lock: with self._lock:
job.finished_at = time.time() job.finished_at = time.time()
job.ok = ok job.ok = ok
job.message = message job.message = message
job.log = log[-200000:] job.log = log[-200000:]
self._recent.append(job) self._recent.append(job)
self._recent = self._recent[-20:] self._recent = self._recent[-20:]
self._current = None self._current = None
BUILD_QUEUE = BuildQueue() BUILD_QUEUE = BuildQueue()
def run_build_commands(log_callback: Callable[[str], None] | None = None) -> tuple[bool, str, str]: def run_build_commands(log_callback: Callable[[str], None] | None = None) -> tuple[bool, str, str]:
venv_python = ROOT / ".venv" / "bin" / "python" venv_python = ROOT / ".venv" / "bin" / "python"
venv_pip = ROOT / ".venv" / "bin" / "pip" venv_pip = ROOT / ".venv" / "bin" / "pip"
commands = [["emacs", "-Q", "--script", "build-site.el"]] commands = [["emacs", "-Q", "--script", "build-site.el"]]
if not venv_python.exists(): if not venv_python.exists():
commands.extend( commands.extend(
[ [
[sys.executable, "-m", "venv", ".venv"], [sys.executable, "-m", "venv", ".venv"],
[str(venv_pip), "install", "-r", "requirements.txt"], [str(venv_pip), "install", "-r", "requirements.txt"],
] ]
) )
commands.append([str(venv_python), "search-index-json.py"]) commands.append([str(venv_python), "search-index-json.py"])
combined = [] combined = []
def append_log(text: str) -> None: def append_log(text: str) -> None:
combined.append(text) combined.append(text)
if log_callback: if log_callback:
log_callback(text) log_callback(text)
ok = True ok = True
for command in commands: for command in commands:
append_log(f"$ {' '.join(command)}\n") append_log(f"$ {' '.join(command)}\n")
env = os.environ.copy() env = os.environ.copy()
env["PYTHONUNBUFFERED"] = "1" env["PYTHONUNBUFFERED"] = "1"
proc = subprocess.Popen( proc = subprocess.Popen(
command, command,
cwd=ROOT, cwd=ROOT,
env=env, env=env,
text=True, text=True,
bufsize=1, bufsize=1,
stderr=subprocess.STDOUT, stderr=subprocess.STDOUT,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
) )
assert proc.stdout is not None assert proc.stdout is not None
for line in proc.stdout: for line in proc.stdout:
append_log(line) append_log(line)
return_code = proc.wait() return_code = proc.wait()
if return_code != 0: if return_code != 0:
ok = False ok = False
append_log(f"\nCommand exited with {return_code}.\n") append_log(f"\nCommand exited with {return_code}.\n")
break break
message = "Build complete. The site output and search index were regenerated." if ok else "Build failed. Check the log below." 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) return ok, message, "".join(combined)
def queue_build(page: dict[str, Any]) -> dict[str, Any]: def queue_build(page: dict[str, Any]) -> dict[str, Any]:
return BUILD_QUEUE.enqueue(page["path"], page["title"]).to_dict() return BUILD_QUEUE.enqueue(page["path"], page["title"]).to_dict()
def queue_hidden_build() -> dict[str, Any]: def queue_hidden_build() -> dict[str, Any]:
return BUILD_QUEUE.enqueue("assets/content/hidden-details.json", "Hidden Memory Observatory").to_dict() return BUILD_QUEUE.enqueue("assets/content/hidden-details.json", "Hidden Memory Observatory").to_dict()

View File

@@ -1,97 +1,97 @@
"""Configuration and content-root discovery for the authoring service.""" """Configuration and content-root discovery for the authoring service."""
from __future__ import annotations from __future__ import annotations
import os import os
from pathlib import Path from pathlib import Path
APP_ROOT = Path(__file__).resolve().parents[2] APP_ROOT = Path(__file__).resolve().parents[2]
DEFAULT_CONTENT_ROOT = Path("/home/zaine/master-folder/org-platform/org_web") DEFAULT_CONTENT_ROOT = Path("/home/zaine/master-folder/org-platform/org_web")
def looks_like_content_root(path: Path) -> bool: def looks_like_content_root(path: Path) -> bool:
return (path / "blogs").exists() or (path / "posts").exists() or (path / "lima").exists() return (path / "blogs").exists() or (path / "posts").exists() or (path / "lima").exists()
def resolve_root() -> Path: def resolve_root() -> Path:
for env_name in ("AUTHOR_CONTENT_ROOT", "AUTHOR_ROOT"): for env_name in ("AUTHOR_CONTENT_ROOT", "AUTHOR_ROOT"):
env_root = os.environ.get(env_name) env_root = os.environ.get(env_name)
if not env_root: if not env_root:
continue continue
resolved = Path(env_root).expanduser().resolve() resolved = Path(env_root).expanduser().resolve()
if looks_like_content_root(resolved): if looks_like_content_root(resolved):
return resolved return resolved
candidates = [ candidates = [
Path.cwd(), Path.cwd(),
DEFAULT_CONTENT_ROOT, DEFAULT_CONTENT_ROOT,
APP_ROOT, APP_ROOT,
] ]
workspace = os.environ.get("GITHUB_WORKSPACE") workspace = os.environ.get("GITHUB_WORKSPACE")
if workspace: if workspace:
candidates.insert(0, Path(workspace)) candidates.insert(0, Path(workspace))
for base in list(candidates): for base in list(candidates):
candidates.extend(base.parents) candidates.extend(base.parents)
seen = set() seen = set()
for candidate in candidates: for candidate in candidates:
resolved = candidate.expanduser().resolve() resolved = candidate.expanduser().resolve()
if resolved in seen: if resolved in seen:
continue continue
seen.add(resolved) seen.add(resolved)
if looks_like_content_root(resolved): if looks_like_content_root(resolved):
return resolved return resolved
return APP_ROOT return APP_ROOT
ROOT = resolve_root() ROOT = resolve_root()
BLOGS_DIR = ROOT / "blogs" BLOGS_DIR = ROOT / "blogs"
POSTS_DIR = ROOT / "posts" POSTS_DIR = ROOT / "posts"
LIMA_DIR = ROOT / "lima" LIMA_DIR = ROOT / "lima"
IMAGE_ASSETS_DIR = ROOT / "assets" / "images" IMAGE_ASSETS_DIR = ROOT / "assets" / "images"
HZONE_ASSETS_DIR = IMAGE_ASSETS_DIR / "hzone" HZONE_ASSETS_DIR = IMAGE_ASSETS_DIR / "hzone"
HIDDEN_DETAILS_JS = ROOT / "assets" / "scripts" / "hidden-details.js" HIDDEN_DETAILS_JS = ROOT / "assets" / "scripts" / "hidden-details.js"
HIDDEN_CONTENT_JSON = ROOT / "assets" / "content" / "hidden-details.json" 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() HIDDEN_BACKUP_DIR = Path(os.environ.get("AUTHOR_HIDDEN_BACKUP_DIR", APP_ROOT / "backups" / "hidden-details")).expanduser().resolve()
EXCLUDED_CONTENT_DIR_NAMES = { EXCLUDED_CONTENT_DIR_NAMES = {
".agents", ".agents",
".codex", ".codex",
".git", ".git",
".packages", ".packages",
".venv", ".venv",
"__pycache__", "__pycache__",
"assets", "assets",
"backups", "backups",
"output", "output",
"tags", "tags",
} }
GENERATED_ORG_NAMES = { GENERATED_ORG_NAMES = {
"blogs-list.org", "blogs-list.org",
"books-list.org", "books-list.org",
"posts-list.org", "posts-list.org",
"career-list.org", "career-list.org",
"sitemap.org", "sitemap.org",
"recently-updated.org", "recently-updated.org",
"wip.org", "wip.org",
} }
GENERATED_CONTENT_NAMES = GENERATED_ORG_NAMES | {"lima-list.org"} GENERATED_CONTENT_NAMES = GENERATED_ORG_NAMES | {"lima-list.org"}
ALLOWED_UPLOAD_EXTENSIONS = { ALLOWED_UPLOAD_EXTENSIONS = {
".png", ".png",
".jpg", ".jpg",
".jpeg", ".jpeg",
".gif", ".gif",
".webp", ".webp",
".svg", ".svg",
} }
MONTH_NAMES = [ MONTH_NAMES = [
"january", "january",
"february", "february",
"march", "march",
"april", "april",
"may", "may",
"june", "june",
"july", "july",
"august", "august",
"september", "september",
"october", "october",
"november", "november",
"december", "december",
] ]

View File

@@ -1,203 +1,203 @@
"""Hidden narrative constants used by the authoring UI.""" """Hidden narrative constants used by the authoring UI."""
from __future__ import annotations from __future__ import annotations
HIDDEN_CONTENT_TYPES = [ HIDDEN_CONTENT_TYPES = [
"tooltip", "tooltip",
"quote", "quote",
"whisper", "whisper",
"poem", "poem",
"observation", "observation",
"dialogue", "dialogue",
"secret search", "secret search",
"symbolic fragment", "symbolic fragment",
"ambient memory", "ambient memory",
"hidden interaction", "hidden interaction",
] ]
HIDDEN_CONTENT_CLASSES = [ HIDDEN_CONTENT_CLASSES = [
"fragment", "fragment",
"memory", "memory",
"story material", "story material",
"interaction", "interaction",
"lore", "lore",
"system layer", "system layer",
] ]
HIDDEN_SURFACES = [ HIDDEN_SURFACES = [
"tooltip", "tooltip",
"quote", "quote",
"poem", "poem",
"story", "story",
"observatory", "observatory",
"constellation route", "constellation route",
"hidden route", "hidden route",
"hidden interaction", "hidden interaction",
"search", "search",
"keyboard", "keyboard",
"play", "play",
"dream", "dream",
"temporal", "temporal",
"future z", "future z",
"seasonal", "seasonal",
"loading", "loading",
"guestbook", "guestbook",
"terminal", "terminal",
"layer guide", "layer guide",
] ]
TYPE_ARCHITECTURE = { TYPE_ARCHITECTURE = {
"tooltip": {"contentClass": "fragment", "surfaces": ["tooltip"], "observatory": "ambient"}, "tooltip": {"contentClass": "fragment", "surfaces": ["tooltip"], "observatory": "ambient"},
"quote": {"contentClass": "fragment", "surfaces": ["quote", "observatory"], "observatory": "ambient"}, "quote": {"contentClass": "fragment", "surfaces": ["quote", "observatory"], "observatory": "ambient"},
"whisper": {"contentClass": "fragment", "surfaces": ["tooltip", "quote", "observatory"], "observatory": "ambient"}, "whisper": {"contentClass": "fragment", "surfaces": ["tooltip", "quote", "observatory"], "observatory": "ambient"},
"poem": {"contentClass": "fragment", "surfaces": ["poem", "observatory"], "observatory": "ambient"}, "poem": {"contentClass": "fragment", "surfaces": ["poem", "observatory"], "observatory": "ambient"},
"observation": {"contentClass": "memory", "surfaces": ["story", "observatory"], "observatory": "node"}, "observation": {"contentClass": "memory", "surfaces": ["story", "observatory"], "observatory": "node"},
"dialogue": {"contentClass": "memory", "surfaces": ["story", "observatory"], "observatory": "node"}, "dialogue": {"contentClass": "memory", "surfaces": ["story", "observatory"], "observatory": "node"},
"secret search": {"contentClass": "interaction", "surfaces": ["search", "hidden route"], "observatory": "event"}, "secret search": {"contentClass": "interaction", "surfaces": ["search", "hidden route"], "observatory": "event"},
"symbolic fragment": {"contentClass": "fragment", "surfaces": ["story", "observatory"], "observatory": "ambient"}, "symbolic fragment": {"contentClass": "fragment", "surfaces": ["story", "observatory"], "observatory": "ambient"},
"ambient memory": {"contentClass": "memory", "surfaces": ["story", "observatory"], "observatory": "node"}, "ambient memory": {"contentClass": "memory", "surfaces": ["story", "observatory"], "observatory": "node"},
"hidden interaction": {"contentClass": "interaction", "surfaces": ["hidden route", "keyboard", "play"], "observatory": "event"}, "hidden interaction": {"contentClass": "interaction", "surfaces": ["hidden route", "keyboard", "play"], "observatory": "event"},
} }
LEGACY_CONTENT_TYPE_MAP = { LEGACY_CONTENT_TYPE_MAP = {
"hidden tooltip": "tooltip", "hidden tooltip": "tooltip",
"hover message": "tooltip", "hover message": "tooltip",
"loading screen message": "whisper", "loading screen message": "whisper",
"hidden dialogue": "dialogue", "hidden dialogue": "dialogue",
"hidden conversation": "dialogue", "hidden conversation": "dialogue",
"journal entry": "observation", "journal entry": "observation",
"future z message": "ambient memory", "future z message": "ambient memory",
"young z memory fragment": "ambient memory", "young z memory fragment": "ambient memory",
"sensei chi wisdom entry": "quote", "sensei chi wisdom entry": "quote",
"aphy system message": "whisper", "aphy system message": "whisper",
"lima note/message": "whisper", "lima note/message": "whisper",
"dream sequence": "symbolic fragment", "dream sequence": "symbolic fragment",
"guestbook entry": "observation", "guestbook entry": "observation",
"terminal log": "hidden interaction", "terminal log": "hidden interaction",
"fake error message": "hidden interaction", "fake error message": "hidden interaction",
"recurring joke": "whisper", "recurring joke": "whisper",
"rare event": "hidden interaction", "rare event": "hidden interaction",
"secret interaction": "hidden interaction", "secret interaction": "hidden interaction",
"seasonal event": "hidden interaction", "seasonal event": "hidden interaction",
"weather-based event": "hidden interaction", "weather-based event": "hidden interaction",
"hidden achievement": "hidden interaction", "hidden achievement": "hidden interaction",
"search toast": "secret search", "search toast": "secret search",
"search route": "secret search", "search route": "secret search",
"keyboard secret": "hidden interaction", "keyboard secret": "hidden interaction",
"family layer": "observation", "family layer": "observation",
} }
CHARACTER_REGISTRY = { CHARACTER_REGISTRY = {
"young z": { "young z": {
"id": "young z", "id": "young z",
"displayLabel": "young z", "displayLabel": "young z",
"aliases": ["Young Z", "young z", "young-z", "young_z", "young"], "aliases": ["Young Z", "young z", "young-z", "young_z", "young"],
"territoryColor": "#d8a95a", "territoryColor": "#d8a95a",
"glow": "#f0b85c", "glow": "#f0b85c",
"sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82},
"observatory": {"territory": "nostalgic/playful memories", "shimmer": "soft"}, "observatory": {"territory": "nostalgic/playful memories", "shimmer": "soft"},
"presenceTones": ["nostalgic", "funny", "hopeful"], "presenceTones": ["nostalgic", "funny", "hopeful"],
"presenceKeywords": ["childhood", "play", "desk", "crayon", "young", "small", "blanket", "memory"], "presenceKeywords": ["childhood", "play", "desk", "crayon", "young", "small", "blanket", "memory"],
"symbol": "Y", "symbol": "Y",
"motifs": ["crayon sun", "blanket cape", "childhood desk"], "motifs": ["crayon sun", "blanket cape", "childhood desk"],
"themes": ["childhood", "play", "memory", "safety"], "themes": ["childhood", "play", "memory", "safety"],
"affinities": ["z", "future z", "aphy", "lima"], "affinities": ["z", "future z", "aphy", "lima"],
}, },
"z": { "z": {
"id": "z", "id": "z",
"displayLabel": "z", "displayLabel": "z",
"aliases": ["Z", "z", "zaine"], "aliases": ["Z", "z", "zaine"],
"territoryColor": "#d6c38a", "territoryColor": "#d6c38a",
"glow": "#ead68e", "glow": "#ead68e",
"sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82},
"observatory": {"territory": "system/symbolic areas", "shimmer": "diagnostic"}, "observatory": {"territory": "system/symbolic areas", "shimmer": "diagnostic"},
"presenceTones": ["funny", "strange", "hopeful"], "presenceTones": ["funny", "strange", "hopeful"],
"presenceKeywords": ["system", "console", "diagnostic", "keyboard", "search", "terminal", "backup", "symbolic", "query"], "presenceKeywords": ["system", "console", "diagnostic", "keyboard", "search", "terminal", "backup", "symbolic", "query"],
"symbol": "A", "symbol": "A",
"motifs": ["console", "diagnostic", "backup"], "motifs": ["console", "diagnostic", "backup"],
"themes": ["humor", "systems", "care through tools"], "themes": ["humor", "systems", "care through tools"],
"affinities": ["z", "lima", "sensei chi"], "affinities": ["z", "lima", "sensei chi"],
}, },
"lima": { "lima": {
"id": "lima", "id": "lima",
"displayLabel": "lima", "displayLabel": "lima",
"aliases": ["Lima", "lima"], "aliases": ["Lima", "lima"],
"territoryColor": "#d06b78", "territoryColor": "#d06b78",
"glow": "#f2c58b", "glow": "#f2c58b",
"sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82},
"observatory": {"territory": "warm/protective arcs", "shimmer": "warm"}, "observatory": {"territory": "warm/protective arcs", "shimmer": "warm"},
"presenceTones": ["warm", "protective", "soft"], "presenceTones": ["warm", "protective", "soft"],
"presenceKeywords": ["lima", "love", "warm", "kitchen", "light", "ring", "home", "protect", "eat"], "presenceKeywords": ["lima", "love", "warm", "kitchen", "light", "ring", "home", "protect", "eat"],
"symbol": "L", "symbol": "L",
"motifs": ["warmth", "kitchen light", "ring"], "motifs": ["warmth", "kitchen light", "ring"],
"themes": ["love", "home", "grounding"], "themes": ["love", "home", "grounding"],
"affinities": ["z", "aphy", "future z", "young z"], "affinities": ["z", "aphy", "future z", "young z"],
}, },
"sensei chi": { "sensei chi": {
"id": "sensei chi", "id": "sensei chi",
"displayLabel": "sensei chi", "displayLabel": "sensei chi",
"aliases": ["Sensei Chi", "sensei chi", "sensei-chi", "sensei_chi", "sensei"], "aliases": ["Sensei Chi", "sensei chi", "sensei-chi", "sensei_chi", "sensei"],
"territoryColor": "#75a9bd", "territoryColor": "#75a9bd",
"glow": "#9ccddd", "glow": "#9ccddd",
"sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82},
"observatory": {"territory": "reflective areas", "shimmer": "quiet"}, "observatory": {"territory": "reflective areas", "shimmer": "quiet"},
"presenceTones": ["wise", "melancholy", "soft"], "presenceTones": ["wise", "melancholy", "soft"],
"presenceKeywords": ["reflection", "patience", "wisdom", "lesson", "tea", "garden", "quiet", "sensei"], "presenceKeywords": ["reflection", "patience", "wisdom", "lesson", "tea", "garden", "quiet", "sensei"],
"symbol": "S", "symbol": "S",
"motifs": ["tea", "garden", "quiet lesson"], "motifs": ["tea", "garden", "quiet lesson"],
"themes": ["reflection", "patience", "wisdom"], "themes": ["reflection", "patience", "wisdom"],
"affinities": ["aphy", "future z"], "affinities": ["aphy", "future z"],
}, },
"future z": { "future z": {
"id": "future z", "id": "future z",
"displayLabel": "future z", "displayLabel": "future z",
"aliases": ["Future Z", "future z", "future-z", "future_z", "future"], "aliases": ["Future Z", "future z", "future-z", "future_z", "future"],
"territoryColor": "#a58ac9", "territoryColor": "#a58ac9",
"glow": "#c2a4ee", "glow": "#c2a4ee",
"sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82}, "sizes": {"tiny": 18, "medium": 42, "close": 56, "focus": 82},
"observatory": {"territory": "temporal regions", "shimmer": "temporal"}, "observatory": {"territory": "temporal regions", "shimmer": "temporal"},
"presenceTones": ["hopeful", "melancholy", "wise"], "presenceTones": ["hopeful", "melancholy", "wise"],
"presenceKeywords": ["future", "time", "clock", "older", "tomorrow", "age", "reassurance", "continuity"], "presenceKeywords": ["future", "time", "clock", "older", "tomorrow", "age", "reassurance", "continuity"],
"symbol": "F", "symbol": "F",
"motifs": ["clock", "age 40", "future log"], "motifs": ["clock", "age 40", "future log"],
"themes": ["time", "reassurance", "continuity"], "themes": ["time", "reassurance", "continuity"],
"affinities": ["z", "young z", "lima", "sensei chi"], "affinities": ["z", "young z", "lima", "sensei chi"],
}, },
} }
HIDDEN_CHARACTERS = list(CHARACTER_REGISTRY) HIDDEN_CHARACTERS = list(CHARACTER_REGISTRY)
HIDDEN_TONES = ["warm", "funny", "nostalgic", "wise", "strange", "soft", "hopeful", "protective", "melancholy"] HIDDEN_TONES = ["warm", "funny", "nostalgic", "wise", "strange", "soft", "hopeful", "protective", "melancholy"]
HIDDEN_RARITIES = ["common", "uncommon", "rare", "very rare", "seasonal", "timed"] HIDDEN_RARITIES = ["common", "uncommon", "rare", "very rare", "seasonal", "timed"]
HIDDEN_STORY_MARKERS = ["public", "hidden", "rare", "emotional", "dream-like", "temporal"] HIDDEN_STORY_MARKERS = ["public", "hidden", "rare", "emotional", "dream-like", "temporal"]
HIDDEN_DISCOVERY_STYLES = ["gradual", "direct", "hidden route", "character-led", "dream-like", "temporal"] HIDDEN_DISCOVERY_STYLES = ["gradual", "direct", "hidden route", "character-led", "dream-like", "temporal"]
HIDDEN_LAYER_DEPTHS = [ HIDDEN_LAYER_DEPTHS = [
{ {
"id": "0", "id": "0",
"name": "Surface Reality", "name": "Surface Reality",
"meaning": "Normal visible website content, visible warmth, and ordinary interactions.", "meaning": "Visible content, ordinary interactions, and the world that every visitor sees.",
}, },
{ {
"id": "1", "id": "1",
"name": "Hidden Personality", "name": "Hidden Echoes",
"meaning": "Small hidden jokes, hover text, tiny discoveries, and recurring symbols.", "meaning": "Small discoveries, recurring symbols, tooltips, jokes, and fragments beneath the surface.",
}, },
{ {
"id": "2", "id": "2",
"name": "Memory Layer", "name": "Memory Archive",
"meaning": "young z memories, lima notes, nostalgia, and emotional fragments.", "meaning": "Personal recollections, nostalgia, young z memories, and emotional fragments.",
}, },
{ {
"id": "3", "id": "3",
"name": "Reflection Layer", "name": "Reflection Garden",
"meaning": "sensei chi philosophy, aphy conversations, and introspection.", "meaning": "Wisdom, philosophy, questions, and moments of introspection.",
}, },
{ {
"id": "4", "id": "4",
"name": "Time Layer", "name": "Temporal Currents",
"meaning": "future z logs, time anomalies, long-term revisits, and future/past echoes.", "meaning": "Future echoes, revisits, time anomalies, and conversations spanning different moments.",
}, },
{ {
"id": "5", "id": "5",
"name": "Core Layer", "name": "Heartspace",
"meaning": "Rare deeply emotional truths found by patient exploration.", "meaning": "Rare emotional truths, enduring relationships, and the quiet centre connecting everything.",
}, },
] ]

View File

@@ -1,427 +1,427 @@
"""Editable page, upload, and diagnostics operations.""" """Editable page, upload, and diagnostics operations."""
from __future__ import annotations from __future__ import annotations
import os import os
import posixpath import posixpath
import re import re
import struct import struct
import sys import sys
from datetime import datetime from datetime import datetime
from email.parser import BytesParser from email.parser import BytesParser
from email.policy import default as email_default_policy from email.policy import default as email_default_policy
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from .config import ( from .config import (
ALLOWED_UPLOAD_EXTENSIONS, ALLOWED_UPLOAD_EXTENSIONS,
BLOGS_DIR, BLOGS_DIR,
EXCLUDED_CONTENT_DIR_NAMES, EXCLUDED_CONTENT_DIR_NAMES,
GENERATED_CONTENT_NAMES, GENERATED_CONTENT_NAMES,
HZONE_ASSETS_DIR, HZONE_ASSETS_DIR,
IMAGE_ASSETS_DIR, IMAGE_ASSETS_DIR,
LIMA_DIR, LIMA_DIR,
MONTH_NAMES, MONTH_NAMES,
POSTS_DIR, POSTS_DIR,
ROOT, ROOT,
) )
from .models import ContentPage from .models import ContentPage
from .utils import html_escape, normalise_tags, org_date, parse_org_datetime, slugify from .utils import html_escape, normalise_tags, org_date, parse_org_datetime, slugify
def safe_relative_path(path: str) -> Path: def safe_relative_path(path: str) -> Path:
rel = Path(path) rel = Path(path)
if rel.is_absolute() or ".." in rel.parts: if rel.is_absolute() or ".." in rel.parts:
raise ValueError("Path must stay inside this repository.") raise ValueError("Path must stay inside this repository.")
full = (ROOT / rel).resolve() full = (ROOT / rel).resolve()
if not full.is_relative_to(ROOT): if not full.is_relative_to(ROOT):
raise ValueError("Path must stay inside this repository.") raise ValueError("Path must stay inside this repository.")
if full.name in GENERATED_CONTENT_NAMES or "sync-conflict" in full.name: if full.name in GENERATED_CONTENT_NAMES or "sync-conflict" in full.name:
raise ValueError("Generated and sync-conflict files are not editable here.") raise ValueError("Generated and sync-conflict files are not editable here.")
rel_parts = full.relative_to(ROOT).parts rel_parts = full.relative_to(ROOT).parts
if any(part in EXCLUDED_CONTENT_DIR_NAMES for part in rel_parts): if any(part in EXCLUDED_CONTENT_DIR_NAMES for part in rel_parts):
raise ValueError("This path is outside the editable content folders.") raise ValueError("This path is outside the editable content folders.")
if full.suffix == ".org": if full.suffix == ".org":
return full return full
if full.suffix == ".md" and full.is_relative_to(LIMA_DIR): if full.suffix == ".md" and full.is_relative_to(LIMA_DIR):
return full return full
raise ValueError("Only .org content files and .md files under lima can be edited here.") 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: def safe_target_path(path: str, slug: str, page_type: str) -> Path:
candidate = path.strip() candidate = path.strip()
if not candidate: if not candidate:
raise ValueError("Path is required.") raise ValueError("Path is required.")
default_ext = ".md" if page_type == "lima" else ".org" default_ext = ".md" if page_type == "lima" else ".org"
if candidate.endswith("/"): if candidate.endswith("/"):
candidate = f"{candidate}{slug}{default_ext}" candidate = f"{candidate}{slug}{default_ext}"
elif not Path(candidate).suffix: elif not Path(candidate).suffix:
candidate = f"{candidate}{default_ext}" candidate = f"{candidate}{default_ext}"
return safe_relative_path(candidate) return safe_relative_path(candidate)
def markdown_title(content: str, fallback: str) -> str: def markdown_title(content: str, fallback: str) -> str:
for line in content.splitlines(): for line in content.splitlines():
match = re.match(r"^#{1,6}\s+(.+?)\s*$", line) match = re.match(r"^#{1,6}\s+(.+?)\s*$", line)
if match: if match:
return match.group(1).strip() return match.group(1).strip()
return fallback.replace("-", " ").replace("_", " ").title() return fallback.replace("-", " ").replace("_", " ").title()
def read_markdown_page(path: Path) -> ContentPage: def read_markdown_page(path: Path) -> ContentPage:
content = path.read_text(encoding="utf-8") content = path.read_text(encoding="utf-8")
rel = path.relative_to(ROOT).as_posix() rel = path.relative_to(ROOT).as_posix()
title = markdown_title(content, path.stem) title = markdown_title(content, path.stem)
return ContentPage( return ContentPage(
path=rel, path=rel,
page_type="lima", page_type="lima",
title=title, title=title,
slug=path.stem, slug=path.stem,
tags=[], tags=[],
content=content, content=content,
date="", date="",
comments=True, comments=True,
options="", options="",
format="markdown", format="markdown",
) )
def read_page(path: Path) -> ContentPage: def read_page(path: Path) -> ContentPage:
if path.suffix == ".md" and path.is_relative_to(LIMA_DIR): if path.suffix == ".md" and path.is_relative_to(LIMA_DIR):
return read_markdown_page(path) return read_markdown_page(path)
text = path.read_text(encoding="utf-8") text = path.read_text(encoding="utf-8")
meta: dict[str, str] = {} meta: dict[str, str] = {}
body_lines: list[str] = [] body_lines: list[str] = []
in_header = True in_header = True
for line in text.splitlines(): for line in text.splitlines():
if in_header and line.startswith("#+"): if in_header and line.startswith("#+"):
key, _, value = line[2:].partition(":") key, _, value = line[2:].partition(":")
meta[key.strip().upper()] = value.strip() meta[key.strip().upper()] = value.strip()
else: else:
in_header = False in_header = False
body_lines.append(line) body_lines.append(line)
rel = path.relative_to(ROOT).as_posix() rel = path.relative_to(ROOT).as_posix()
if path.is_relative_to(BLOGS_DIR): if path.is_relative_to(BLOGS_DIR):
page_type = "blog" page_type = "blog"
elif path.is_relative_to(POSTS_DIR): elif path.is_relative_to(POSTS_DIR):
page_type = "post" page_type = "post"
else: else:
page_type = "page" page_type = "page"
slug = meta.get("SLUG") or path.stem slug = meta.get("SLUG") or path.stem
tags = normalise_tags(meta.get("FILETAGS", "")) tags = normalise_tags(meta.get("FILETAGS", ""))
return ContentPage( return ContentPage(
path=rel, path=rel,
page_type=page_type, page_type=page_type,
title=meta.get("TITLE", path.stem), title=meta.get("TITLE", path.stem),
slug=slug, slug=slug,
tags=tags, tags=tags,
content="\n".join(body_lines).lstrip("\n"), content="\n".join(body_lines).lstrip("\n"),
date=meta.get("DATE", org_date(datetime.fromtimestamp(path.stat().st_mtime))), date=meta.get("DATE", org_date(datetime.fromtimestamp(path.stat().st_mtime))),
comments=meta.get("COMMENTS", "t").lower() == "t", comments=meta.get("COMMENTS", "t").lower() == "t",
options=meta.get("OPTIONS", "num:nil"), options=meta.get("OPTIONS", "num:nil"),
format="org", format="org",
wip=meta.get("WIP"), wip=meta.get("WIP"),
) )
def page_to_dict(page: ContentPage) -> dict[str, Any]: def page_to_dict(page: ContentPage) -> dict[str, Any]:
return { return {
"path": page.path, "path": page.path,
"pageType": page.page_type, "pageType": page.page_type,
"title": page.title, "title": page.title,
"slug": page.slug, "slug": page.slug,
"tags": page.tags, "tags": page.tags,
"content": page.content, "content": page.content,
"date": page.date, "date": page.date,
"comments": page.comments, "comments": page.comments,
"options": page.options, "options": page.options,
"format": page.format, "format": page.format,
"wip": page.wip or "", "wip": page.wip or "",
} }
def server_diagnostics() -> dict[str, Any]: def server_diagnostics() -> dict[str, Any]:
pages = list_pages() pages = list_pages()
try: try:
cwd = Path.cwd().as_posix() cwd = Path.cwd().as_posix()
except OSError as exc: except OSError as exc:
cwd = f"<unavailable: {exc}>" cwd = f"<unavailable: {exc}>"
return { return {
"root": ROOT.as_posix(), "root": ROOT.as_posix(),
"cwd": cwd, "cwd": cwd,
"executable": sys.executable, "executable": sys.executable,
"pid": os.getpid(), "pid": os.getpid(),
"pageCount": len(pages), "pageCount": len(pages),
"firstPage": pages[0]["path"] if pages else "", "firstPage": pages[0]["path"] if pages else "",
} }
def list_pages() -> list[dict[str, Any]]: def list_pages() -> list[dict[str, Any]]:
pages = [] pages = []
org_paths = [] org_paths = []
if ROOT.exists(): if ROOT.exists():
try: try:
for path in ROOT.rglob("*.org"): for path in ROOT.rglob("*.org"):
try: try:
rel_parts = path.relative_to(ROOT).parts rel_parts = path.relative_to(ROOT).parts
except ValueError: except ValueError:
continue continue
if any(part in EXCLUDED_CONTENT_DIR_NAMES for part in rel_parts): if any(part in EXCLUDED_CONTENT_DIR_NAMES for part in rel_parts):
continue continue
org_paths.append(path) org_paths.append(path)
except OSError: except OSError:
org_paths = [] org_paths = []
try: try:
md_paths = list(LIMA_DIR.rglob("*.md")) if LIMA_DIR.exists() else [] md_paths = list(LIMA_DIR.rglob("*.md")) if LIMA_DIR.exists() else []
except OSError: except OSError:
md_paths = [] md_paths = []
for path in sorted(org_paths + md_paths): for path in sorted(org_paths + md_paths):
if path.name in GENERATED_CONTENT_NAMES or "sync-conflict" in path.name: if path.name in GENERATED_CONTENT_NAMES or "sync-conflict" in path.name:
continue continue
try: try:
page = read_page(path) page = read_page(path)
parsed = parse_org_datetime(page.date) parsed = parse_org_datetime(page.date)
timestamp = parsed.timestamp() if parsed else path.stat().st_mtime timestamp = parsed.timestamp() if parsed else path.stat().st_mtime
except (OSError, UnicodeDecodeError, ValueError): except (OSError, UnicodeDecodeError, ValueError):
continue continue
pages.append( pages.append(
{ {
"path": page.path, "path": page.path,
"pageType": page.page_type, "pageType": page.page_type,
"title": page.title, "title": page.title,
"slug": page.slug, "slug": page.slug,
"tags": page.tags, "tags": page.tags,
"date": page.date, "date": page.date,
"format": page.format, "format": page.format,
"timestamp": timestamp, "timestamp": timestamp,
} }
) )
return sorted(pages, key=lambda item: item["timestamp"], reverse=True) return sorted(pages, key=lambda item: item["timestamp"], reverse=True)
def target_path(data: dict[str, Any], existing_path: str | None) -> Path: def target_path(data: dict[str, Any], existing_path: str | None) -> Path:
if existing_path: if existing_path:
return safe_relative_path(existing_path) return safe_relative_path(existing_path)
title = str(data.get("title") or "").strip() title = str(data.get("title") or "").strip()
slug = slugify(str(data.get("slug") or title)) slug = slugify(str(data.get("slug") or title))
page_type = str(data.get("pageType") or "blog") page_type = str(data.get("pageType") or "blog")
explicit_path = str(data.get("targetPath") or "").strip() explicit_path = str(data.get("targetPath") or "").strip()
if explicit_path: if explicit_path:
return safe_target_path(explicit_path, slug, page_type) return safe_target_path(explicit_path, slug, page_type)
if page_type == "blog": if page_type == "blog":
dt = parse_org_datetime(str(data.get("date") or "")) or datetime.now() 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]}" folder = BLOGS_DIR / str(dt.year) / f"{dt.month:02d}-{MONTH_NAMES[dt.month - 1]}"
return folder / f"{slug}.org" return folder / f"{slug}.org"
if page_type == "post": if page_type == "post":
raw_section = str(data.get("section") or "").strip() raw_section = str(data.get("section") or "").strip()
section = slugify(raw_section) if raw_section else "" section = slugify(raw_section) if raw_section else ""
folder = POSTS_DIR / section if section else POSTS_DIR folder = POSTS_DIR / section if section else POSTS_DIR
return folder / f"{slug}.org" return folder / f"{slug}.org"
if page_type == "lima": if page_type == "lima":
return LIMA_DIR / f"{slug}.md" return LIMA_DIR / f"{slug}.md"
if page_type == "page": if page_type == "page":
return ROOT / f"{slug}.org" return ROOT / f"{slug}.org"
raise ValueError("pageType must be blog, post, page, or lima.") raise ValueError("pageType must be blog, post, page, or lima.")
def render_markdown(data: dict[str, Any]) -> str: def render_markdown(data: dict[str, Any]) -> str:
content = str(data.get("content") or "").replace("\r\n", "\n").strip() content = str(data.get("content") or "").replace("\r\n", "\n").strip()
title = str(data.get("title") or "").strip() title = str(data.get("title") or "").strip()
if not title: if not title:
raise ValueError("Title is required.") raise ValueError("Title is required.")
content = re.sub( content = re.sub(
r'<a\b[^>]*>\s*<img\b[^>]*\bsrc="([^"]+)"[^>]*\balt="([^"]*)"[^>]*>\s*</a>', r'<a\b[^>]*>\s*<img\b[^>]*\bsrc="([^"]+)"[^>]*\balt="([^"]*)"[^>]*>\s*</a>',
lambda match: f"![{match.group(2)}]({match.group(1)})", lambda match: f"![{match.group(2)}]({match.group(1)})",
content, content,
flags=re.IGNORECASE, flags=re.IGNORECASE,
) )
if content: if content:
if re.search(r"^#{1,6}[^\S\r\n]+.+?[^\S\r\n]*$", content, flags=re.MULTILINE): if re.search(r"^#{1,6}[^\S\r\n]+.+?[^\S\r\n]*$", content, flags=re.MULTILINE):
content = re.sub( content = re.sub(
r"^#{1,6}[^\S\r\n]+.+?[^\S\r\n]*$", r"^#{1,6}[^\S\r\n]+.+?[^\S\r\n]*$",
f"# {title}", f"# {title}",
content, content,
count=1, count=1,
flags=re.MULTILINE, flags=re.MULTILINE,
) )
else: else:
content = f"# {title}\n\n{content}" content = f"# {title}\n\n{content}"
return content + "\n" return content + "\n"
return f"# {title}\n" return f"# {title}\n"
def render_org(data: dict[str, Any], previous: ContentPage | None) -> str: def render_org(data: dict[str, Any], previous: ContentPage | None) -> str:
title = str(data.get("title") or "").strip() title = str(data.get("title") or "").strip()
if not title: if not title:
raise ValueError("Title is required.") raise ValueError("Title is required.")
slug = slugify(str(data.get("slug") or title)) slug = slugify(str(data.get("slug") or title))
tags = normalise_tags(data.get("tags", [])) tags = normalise_tags(data.get("tags", []))
content = str(data.get("content") or "").replace("\r\n", "\n").strip() content = str(data.get("content") or "").replace("\r\n", "\n").strip()
date = str(data.get("date") or "").strip() date = str(data.get("date") or "").strip()
if not parse_org_datetime(date): if not parse_org_datetime(date):
date = previous.date if previous else org_date(datetime.now()) date = previous.date if previous else org_date(datetime.now())
options = str(data.get("options") or (previous.options if previous else "num:nil")).strip() options = str(data.get("options") or (previous.options if previous else "num:nil")).strip()
comments = bool(data.get("comments", True)) comments = bool(data.get("comments", True))
lines = [ lines = [
f"#+TITLE: {title}", f"#+TITLE: {title}",
f"#+OPTIONS: {options}", f"#+OPTIONS: {options}",
f"#+DATE: {date}", f"#+DATE: {date}",
f"#+filetags: {''.join(f':{tag}' for tag in tags)}:", f"#+filetags: {''.join(f':{tag}' for tag in tags)}:",
] ]
wip = str(data.get("wip") or (previous.wip if previous else "") or "").strip() wip = str(data.get("wip") or (previous.wip if previous else "") or "").strip()
if wip: if wip:
lines.append(f"#+WIP: {wip}") lines.append(f"#+WIP: {wip}")
lines.extend( lines.extend(
[ [
f"#+COMMENTS: {'t' if comments else ''}", f"#+COMMENTS: {'t' if comments else ''}",
f"#+SLUG: {slug}", f"#+SLUG: {slug}",
"", "",
content, content,
"", "",
] ]
) )
return "\n".join(lines) return "\n".join(lines)
def save_page(data: dict[str, Any]) -> dict[str, Any]: def save_page(data: dict[str, Any]) -> dict[str, Any]:
existing_path = data.get("path") or None existing_path = data.get("path") or None
target = target_path(data, str(existing_path) if existing_path else None) target = target_path(data, str(existing_path) if existing_path else None)
previous = read_page(target) if target.exists() else None previous = read_page(target) if target.exists() else None
if not target.parent.exists(): if not target.parent.exists():
target.parent.mkdir(parents=True) target.parent.mkdir(parents=True)
if target.exists() and not existing_path: if target.exists() and not existing_path:
raise ValueError(f"{target.relative_to(ROOT)} already exists.") raise ValueError(f"{target.relative_to(ROOT)} already exists.")
if target.suffix == ".md": if target.suffix == ".md":
target.write_text(render_markdown(data), encoding="utf-8") target.write_text(render_markdown(data), encoding="utf-8")
else: else:
target.write_text(render_org(data, previous), encoding="utf-8") target.write_text(render_org(data, previous), encoding="utf-8")
return page_to_dict(read_page(target)) return page_to_dict(read_page(target))
def image_dimensions(payload: bytes, ext: str) -> tuple[int, int] | None: 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: if ext == ".png" and payload.startswith(b"\x89PNG\r\n\x1a\n") and len(payload) >= 24:
width, height = struct.unpack(">II", payload[16:24]) width, height = struct.unpack(">II", payload[16:24])
return width, height return width, height
if ext == ".gif" and payload[:6] in {b"GIF87a", b"GIF89a"} and len(payload) >= 10: if ext == ".gif" and payload[:6] in {b"GIF87a", b"GIF89a"} and len(payload) >= 10:
width, height = struct.unpack("<HH", payload[6:10]) width, height = struct.unpack("<HH", payload[6:10])
return width, height return width, height
if ext in {".jpg", ".jpeg"} and payload.startswith(b"\xff\xd8"): if ext in {".jpg", ".jpeg"} and payload.startswith(b"\xff\xd8"):
i = 2 i = 2
while i + 9 < len(payload): while i + 9 < len(payload):
if payload[i] != 0xFF: if payload[i] != 0xFF:
i += 1 i += 1
continue continue
marker = payload[i + 1] marker = payload[i + 1]
i += 2 i += 2
if marker in {0xD8, 0xD9}: if marker in {0xD8, 0xD9}:
continue continue
if i + 2 > len(payload): if i + 2 > len(payload):
break break
size = int.from_bytes(payload[i:i + 2], "big") size = int.from_bytes(payload[i:i + 2], "big")
if size < 2: if size < 2:
break break
if marker in {0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF}: if marker in {0xC0, 0xC1, 0xC2, 0xC3, 0xC5, 0xC6, 0xC7, 0xC9, 0xCA, 0xCB, 0xCD, 0xCE, 0xCF}:
if i + 7 <= len(payload): if i + 7 <= len(payload):
height = int.from_bytes(payload[i + 3:i + 5], "big") height = int.from_bytes(payload[i + 3:i + 5], "big")
width = int.from_bytes(payload[i + 5:i + 7], "big") width = int.from_bytes(payload[i + 5:i + 7], "big")
return width, height return width, height
break break
i += size i += size
return None return None
def relative_asset_path(page_path: str, asset_path: str) -> str: 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 = safe_relative_path(page_path) if page_path else ROOT / "index.org"
page_rel = page.relative_to(ROOT).as_posix() page_rel = page.relative_to(ROOT).as_posix()
page_output_dir = posixpath.dirname(page_rel) page_output_dir = posixpath.dirname(page_rel)
return posixpath.relpath(asset_path, page_output_dir or ".") 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: 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}" absolute_url = f"https://zainezq.com/{asset_path}"
relative_url = relative_asset_path(page_path, asset_path) relative_url = relative_asset_path(page_path, asset_path)
dims = image_dimensions(payload, ext) dims = image_dimensions(payload, ext)
width, height = dims if dims else (1920, 1080) width, height = dims if dims else (1920, 1080)
alt = html_escape(filename) alt = html_escape(filename)
return ( return (
f'<a href="{absolute_url}" data-img="{absolute_url}" data-alt="{alt}" ' f'<a href="{absolute_url}" data-img="{absolute_url}" data-alt="{alt}" '
f'data-width="{width}" data-height="{height}">' f'data-width="{width}" data-height="{height}">'
f'<img src="{relative_url}" alt="{alt}" style="cursor: zoom-in;"></a>' f'<img src="{relative_url}" alt="{alt}" style="cursor: zoom-in;"></a>'
) )
def attachment_image_dir(page_path: str) -> Path: def attachment_image_dir(page_path: str) -> Path:
page = safe_relative_path(page_path) if page_path else ROOT / "index.org" page = safe_relative_path(page_path) if page_path else ROOT / "index.org"
if page.suffix == ".md" and page.is_relative_to(LIMA_DIR): if page.suffix == ".md" and page.is_relative_to(LIMA_DIR):
return HZONE_ASSETS_DIR return HZONE_ASSETS_DIR
if page.is_relative_to(POSTS_DIR): if page.is_relative_to(POSTS_DIR):
rel = page.relative_to(POSTS_DIR) rel = page.relative_to(POSTS_DIR)
section = rel.parts[0] if len(rel.parts) > 1 else "posts" section = rel.parts[0] if len(rel.parts) > 1 else "posts"
return IMAGE_ASSETS_DIR / slugify(section) return IMAGE_ASSETS_DIR / slugify(section)
if page.is_relative_to(BLOGS_DIR): if page.is_relative_to(BLOGS_DIR):
return IMAGE_ASSETS_DIR / "blogs" return IMAGE_ASSETS_DIR / "blogs"
rel = page.relative_to(ROOT) rel = page.relative_to(ROOT)
if len(rel.parts) > 1: if len(rel.parts) > 1:
return IMAGE_ASSETS_DIR / slugify(rel.parts[0]) return IMAGE_ASSETS_DIR / slugify(rel.parts[0])
return IMAGE_ASSETS_DIR / "pages" return IMAGE_ASSETS_DIR / "pages"
def save_upload(filename: str, payload: bytes, page_path: str = "") -> dict[str, str]: def save_upload(filename: str, payload: bytes, page_path: str = "") -> dict[str, str]:
original = Path(filename or "attachment").name original = Path(filename or "attachment").name
ext = Path(original).suffix.lower() ext = Path(original).suffix.lower()
if ext not in ALLOWED_UPLOAD_EXTENSIONS: if ext not in ALLOWED_UPLOAD_EXTENSIONS:
raise ValueError("Only common image files can be uploaded.") raise ValueError("Only common image files can be uploaded.")
now = datetime.now() now = datetime.now()
target_dir = attachment_image_dir(page_path) target_dir = attachment_image_dir(page_path)
target_dir.mkdir(parents=True, exist_ok=True) target_dir.mkdir(parents=True, exist_ok=True)
stem = slugify(Path(original).stem) stem = slugify(Path(original).stem)
prefix = "" if re.match(r"^\d{4}-\d{2}-\d{2}-", stem) else f"{now.strftime('%Y-%m-%d')}-" 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}" target = target_dir / f"{prefix}{stem}{ext}"
counter = 2 counter = 2
while target.exists(): while target.exists():
target = target_dir / f"{prefix}{stem}-{counter}{ext}" target = target_dir / f"{prefix}{stem}-{counter}{ext}"
counter += 1 counter += 1
target.write_bytes(payload) target.write_bytes(payload)
rel = target.relative_to(ROOT).as_posix() rel = target.relative_to(ROOT).as_posix()
absolute_url = f"https://zainezq.com/{rel}" absolute_url = f"https://zainezq.com/{rel}"
relative_url = relative_asset_path(page_path, rel) relative_url = relative_asset_path(page_path, rel)
is_markdown = page_path.endswith(".md") is_markdown = page_path.endswith(".md")
insert_text = f"![{target.name}]({absolute_url})" if is_markdown else f"[[{relative_url}]]" insert_text = f"![{target.name}]({absolute_url})" if is_markdown else f"[[{relative_url}]]"
return { return {
"url": absolute_url, "url": absolute_url,
"relativeUrl": relative_url, "relativeUrl": relative_url,
"path": rel, "path": rel,
"markdown": insert_text, "markdown": insert_text,
"insertText": insert_text, "insertText": insert_text,
"filename": target.name, "filename": target.name,
} }
def parse_upload_form(content_type: str, body: bytes) -> tuple[str, bytes, str]: def parse_upload_form(content_type: str, body: bytes) -> tuple[str, bytes, str]:
if not content_type.lower().startswith("multipart/form-data"): if not content_type.lower().startswith("multipart/form-data"):
raise ValueError("Uploads must use multipart/form-data.") raise ValueError("Uploads must use multipart/form-data.")
message = BytesParser(policy=email_default_policy).parsebytes( 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 f"Content-Type: {content_type}\r\nMIME-Version: 1.0\r\n\r\n".encode("utf-8") + body
) )
if not message.is_multipart(): if not message.is_multipart():
raise ValueError("Upload form is not multipart.") raise ValueError("Upload form is not multipart.")
filename = "" filename = ""
payload = b"" payload = b""
page_path = "" page_path = ""
for part in message.iter_parts(): for part in message.iter_parts():
name = part.get_param("name", header="content-disposition") name = part.get_param("name", header="content-disposition")
if name == "attachment": if name == "attachment":
filename = part.get_filename() or "" filename = part.get_filename() or ""
payload = part.get_payload(decode=True) or b"" payload = part.get_payload(decode=True) or b""
elif name == "pagePath": elif name == "pagePath":
raw_value = part.get_payload(decode=True) or b"" raw_value = part.get_payload(decode=True) or b""
page_path = raw_value.decode(part.get_content_charset() or "utf-8", errors="replace") page_path = raw_value.decode(part.get_content_charset() or "utf-8", errors="replace")
if not filename: if not filename:
raise ValueError("No attachment was uploaded.") raise ValueError("No attachment was uploaded.")
if not payload: if not payload:
raise ValueError("Attachment is empty.") raise ValueError("Attachment is empty.")
return filename, payload, page_path return filename, payload, page_path

File diff suppressed because it is too large Load Diff

View File

@@ -1,34 +1,34 @@
"""Shared data models.""" """Shared data models."""
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
@dataclass @dataclass
class OrgPage: class OrgPage:
path: str path: str
page_type: str page_type: str
title: str title: str
slug: str slug: str
tags: list[str] tags: list[str]
content: str content: str
date: str date: str
comments: bool comments: bool
options: str options: str
wip: str | None = None wip: str | None = None
@dataclass @dataclass
class ContentPage: class ContentPage:
path: str path: str
page_type: str page_type: str
title: str title: str
slug: str slug: str
tags: list[str] tags: list[str]
content: str content: str
date: str date: str
comments: bool comments: bool
options: str options: str
format: str format: str
wip: str | None = None wip: str | None = None

File diff suppressed because it is too large Load Diff

View File

@@ -1,57 +1,57 @@
"""Small formatting and parsing helpers.""" """Small formatting and parsing helpers."""
from __future__ import annotations from __future__ import annotations
import re import re
from datetime import datetime from datetime import datetime
from typing import Any from typing import Any
def slugify(value: str) -> str: def slugify(value: str) -> str:
slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-")
return slug or "untitled" return slug or "untitled"
def normalise_tags(value: Any) -> list[str]: def normalise_tags(value: Any) -> list[str]:
if isinstance(value, str): if isinstance(value, str):
raw = re.split(r"[,:\s]+", value) raw = re.split(r"[,:\s]+", value)
elif isinstance(value, list): elif isinstance(value, list):
raw = [str(item) for item in value] raw = [str(item) for item in value]
else: else:
raw = [] raw = []
tags = [] tags = []
for tag in raw: for tag in raw:
if not tag.strip(): if not tag.strip():
continue continue
clean = slugify(tag) clean = slugify(tag)
if clean and clean not in tags: if clean and clean not in tags:
tags.append(clean) tags.append(clean)
return tags return tags
def org_date(dt: datetime) -> str: def org_date(dt: datetime) -> str:
return dt.strftime("<%Y-%m-%d %a %H:%M>") return dt.strftime("<%Y-%m-%d %a %H:%M>")
def parse_org_datetime(value: str | None) -> datetime | None: def parse_org_datetime(value: str | None) -> datetime | None:
if not value: if not value:
return None return None
match = re.search(r"<(\d{4})-(\d{2})-(\d{2})(?:\s+\w+)?(?:\s+(\d{2}):(\d{2}))?>", value) match = re.search(r"<(\d{4})-(\d{2})-(\d{2})(?:\s+\w+)?(?:\s+(\d{2}):(\d{2}))?>", value)
if not match: if not match:
return None return None
year, month, day, hour, minute = match.groups() year, month, day, hour, minute = match.groups()
return datetime( return datetime(
int(year), int(year),
int(month), int(month),
int(day), int(day),
int(hour or 12), int(hour or 12),
int(minute or 0), int(minute or 0),
) )
def html_escape(value: str) -> str: def html_escape(value: str) -> str:
return ( return (
value.replace("&", "&amp;") value.replace("&", "&amp;")
.replace("<", "&lt;") .replace("<", "&lt;")
.replace(">", "&gt;") .replace(">", "&gt;")
.replace('"', "&quot;") .replace('"', "&quot;")
) )

View File

@@ -1,140 +1,140 @@
"""HTTP handler and server entry point.""" """HTTP handler and server entry point."""
from __future__ import annotations from __future__ import annotations
import json import json
import os import os
import sys import sys
import time import time
from email.utils import formatdate from email.utils import formatdate
from http import HTTPStatus from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Any from typing import Any
from urllib.parse import parse_qs, urlparse from urllib.parse import parse_qs, urlparse
from .build import BUILD_QUEUE, queue_build, queue_hidden_build from .build import BUILD_QUEUE, queue_build, queue_hidden_build
from .config import ROOT 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 .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 .hidden import load_hidden_store, save_hidden_store
from .templates import APP_HTML, HIDDEN_APP_HTML from .templates import APP_HTML, HIDDEN_APP_HTML
class Handler(BaseHTTPRequestHandler): class Handler(BaseHTTPRequestHandler):
server_version = "OrgAuthoring/1.0" server_version = "OrgAuthoring/1.0"
def log_message(self, fmt: str, *args: Any) -> None: def log_message(self, fmt: str, *args: Any) -> None:
sys.stderr.write("%s - %s\n" % (formatdate(time.time()), fmt % args)) sys.stderr.write("%s - %s\n" % (formatdate(time.time()), fmt % args))
def send_json(self, data: Any, status: HTTPStatus = HTTPStatus.OK) -> None: def send_json(self, data: Any, status: HTTPStatus = HTTPStatus.OK) -> None:
body = json.dumps(data).encode("utf-8") body = json.dumps(data).encode("utf-8")
self.send_response(status) self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(body))) self.send_header("Content-Length", str(len(body)))
self.send_header("Cache-Control", "no-store") self.send_header("Cache-Control", "no-store")
self.end_headers() self.end_headers()
self.wfile.write(body) self.wfile.write(body)
def send_error(self, code: int, message: str | None = None, explain: str | None = None) -> None: def send_error(self, code: int, message: str | None = None, explain: str | None = None) -> None:
if urlparse(self.path).path.startswith("/api/"): if urlparse(self.path).path.startswith("/api/"):
status = HTTPStatus(code) status = HTTPStatus(code)
self.send_json({"error": message or status.phrase}, status) self.send_json({"error": message or status.phrase}, status)
return return
super().send_error(code, message, explain) super().send_error(code, message, explain)
def do_GET(self) -> None: def do_GET(self) -> None:
parsed = urlparse(self.path) parsed = urlparse(self.path)
if parsed.path == "/": if parsed.path == "/":
body = APP_HTML.encode("utf-8") body = APP_HTML.encode("utf-8")
self.send_response(HTTPStatus.OK) self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body))) self.send_header("Content-Length", str(len(body)))
self.end_headers() self.end_headers()
self.wfile.write(body) self.wfile.write(body)
return return
if parsed.path == "/hidden": if parsed.path == "/hidden":
body = HIDDEN_APP_HTML.encode("utf-8") body = HIDDEN_APP_HTML.encode("utf-8")
self.send_response(HTTPStatus.OK) self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(body))) self.send_header("Content-Length", str(len(body)))
self.end_headers() self.end_headers()
self.wfile.write(body) self.wfile.write(body)
return return
if parsed.path == "/api/pages": if parsed.path == "/api/pages":
try: try:
self.send_json(list_pages()) self.send_json(list_pages())
except Exception as exc: except Exception as exc:
self.send_json({"error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) self.send_json({"error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR)
return return
if parsed.path == "/api/diagnostics": if parsed.path == "/api/diagnostics":
try: try:
self.send_json(server_diagnostics()) self.send_json(server_diagnostics())
except Exception as exc: except Exception as exc:
self.send_json({"error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) self.send_json({"error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR)
return return
if parsed.path == "/api/page": if parsed.path == "/api/page":
query = parse_qs(parsed.query) query = parse_qs(parsed.query)
try: try:
path = safe_relative_path(query.get("path", [""])[0]) path = safe_relative_path(query.get("path", [""])[0])
self.send_json(page_to_dict(read_page(path))) self.send_json(page_to_dict(read_page(path)))
except Exception as exc: except Exception as exc:
self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
return return
if parsed.path == "/api/build": if parsed.path == "/api/build":
self.send_json(BUILD_QUEUE.snapshot()) self.send_json(BUILD_QUEUE.snapshot())
return return
if parsed.path == "/api/hidden": if parsed.path == "/api/hidden":
try: try:
self.send_json(load_hidden_store()) self.send_json(load_hidden_store())
except Exception as exc: except Exception as exc:
self.send_json({"error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR) self.send_json({"error": str(exc)}, HTTPStatus.INTERNAL_SERVER_ERROR)
return return
self.send_error(HTTPStatus.NOT_FOUND) self.send_error(HTTPStatus.NOT_FOUND)
def do_POST(self) -> None: def do_POST(self) -> None:
if self.path == "/api/upload": if self.path == "/api/upload":
try: try:
length = int(self.headers.get("Content-Length", "0")) length = int(self.headers.get("Content-Length", "0"))
filename, payload, page_path = parse_upload_form( filename, payload, page_path = parse_upload_form(
self.headers.get("Content-Type", ""), self.headers.get("Content-Type", ""),
self.rfile.read(length), self.rfile.read(length),
) )
self.send_json(save_upload(filename, payload, page_path)) self.send_json(save_upload(filename, payload, page_path))
except Exception as exc: except Exception as exc:
self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
return return
if self.path == "/api/hidden": if self.path == "/api/hidden":
try: try:
length = int(self.headers.get("Content-Length", "0")) length = int(self.headers.get("Content-Length", "0"))
data = json.loads(self.rfile.read(length).decode("utf-8")) data = json.loads(self.rfile.read(length).decode("utf-8"))
saved = save_hidden_store(data) saved = save_hidden_store(data)
saved["queuedBuild"] = queue_hidden_build() saved["queuedBuild"] = queue_hidden_build()
self.send_json(saved) self.send_json(saved)
except Exception as exc: except Exception as exc:
self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
return return
if self.path != "/api/page": if self.path != "/api/page":
self.send_error(HTTPStatus.NOT_FOUND) self.send_error(HTTPStatus.NOT_FOUND)
return return
try: try:
length = int(self.headers.get("Content-Length", "0")) length = int(self.headers.get("Content-Length", "0"))
data = json.loads(self.rfile.read(length).decode("utf-8")) data = json.loads(self.rfile.read(length).decode("utf-8"))
saved = save_page(data) saved = save_page(data)
saved["queuedBuild"] = queue_build(saved) saved["queuedBuild"] = queue_build(saved)
self.send_json(saved) self.send_json(saved)
except Exception as exc: except Exception as exc:
self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST) self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
def main() -> None: def main() -> None:
port = int(os.environ.get("AUTHOR_PORT", "8765")) port = int(os.environ.get("AUTHOR_PORT", "8765"))
server = ThreadingHTTPServer(("127.0.0.1", port), Handler) server = ThreadingHTTPServer(("127.0.0.1", port), Handler)
page_count = len(list_pages()) page_count = len(list_pages())
print(f"Authoring UI running at http://127.0.0.1:{port}") print(f"Authoring UI running at http://127.0.0.1:{port}")
print(f"Content root: {ROOT}") print(f"Content root: {ROOT}")
print(f"Editable pages: {page_count}") print(f"Editable pages: {page_count}")
if page_count == 0: if page_count == 0:
print("WARNING: no editable pages were found. Check AUTHOR_CONTENT_ROOT/AUTHOR_ROOT and the launch directory.") print("WARNING: no editable pages were found. Check AUTHOR_CONTENT_ROOT/AUTHOR_ROOT and the launch directory.")
print("Press Ctrl-C to stop.") print("Press Ctrl-C to stop.")
try: try:
server.serve_forever() server.serve_forever()
except KeyboardInterrupt: except KeyboardInterrupt:
pass pass

View File

@@ -1,371 +1,371 @@
import os import os
import sys 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
sys.path.insert(0, str(Path(__file__).resolve().parents[1])) sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import authoring_service.build as build_server import authoring_service.build as build_server
import authoring_service.config as config_server import authoring_service.config as config_server
import authoring_service.content as server import authoring_service.content as server
import authoring_service.utils as utils_server import authoring_service.utils as utils_server
class AuthoringServerTestCase(unittest.TestCase): class AuthoringServerTestCase(unittest.TestCase):
def setUp(self): def setUp(self):
self.tmp = tempfile.TemporaryDirectory() self.tmp = tempfile.TemporaryDirectory()
self.root = Path(self.tmp.name) self.root = Path(self.tmp.name)
self.blogs = self.root / "blogs" self.blogs = self.root / "blogs"
self.posts = self.root / "posts" self.posts = self.root / "posts"
self.lima = self.root / "lima" self.lima = self.root / "lima"
self.home = self.root / "home" self.home = self.root / "home"
self.tags = self.root / "tags" self.tags = self.root / "tags"
self.hzone = self.root / "assets" / "images" / "hzone" self.hzone = self.root / "assets" / "images" / "hzone"
self.blogs.mkdir() self.blogs.mkdir()
self.posts.mkdir() self.posts.mkdir()
self.lima.mkdir() self.lima.mkdir()
self.home.mkdir() self.home.mkdir()
self.tags.mkdir() self.tags.mkdir()
patches = { patches = {
"ROOT": self.root, "ROOT": self.root,
"BLOGS_DIR": self.blogs, "BLOGS_DIR": self.blogs,
"POSTS_DIR": self.posts, "POSTS_DIR": self.posts,
"LIMA_DIR": self.lima, "LIMA_DIR": self.lima,
"IMAGE_ASSETS_DIR": self.root / "assets" / "images", "IMAGE_ASSETS_DIR": self.root / "assets" / "images",
"HZONE_ASSETS_DIR": self.hzone, "HZONE_ASSETS_DIR": self.hzone,
} }
self.patchers = [mock.patch.object(server, name, value) for name, value in patches.items()] self.patchers = [mock.patch.object(server, name, value) for name, value in patches.items()]
for patcher in self.patchers: for patcher in self.patchers:
patcher.start() patcher.start()
def tearDown(self): def tearDown(self):
for patcher in reversed(self.patchers): for patcher in reversed(self.patchers):
patcher.stop() patcher.stop()
self.tmp.cleanup() self.tmp.cleanup()
class UtilityTests(AuthoringServerTestCase): class UtilityTests(AuthoringServerTestCase):
def test_resolve_root_ignores_invalid_author_root(self): def test_resolve_root_ignores_invalid_author_root(self):
wrong_root = self.root / "empty" wrong_root = self.root / "empty"
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(config_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(config_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(utils_server.slugify("Hello, Org Web!"), "hello-org-web") self.assertEqual(utils_server.slugify("Hello, Org Web!"), "hello-org-web")
self.assertEqual(utils_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(
utils_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(
utils_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(
utils_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(utils_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(
server.safe_relative_path("blogs/example.org"), server.safe_relative_path("blogs/example.org"),
self.blogs / "example.org", self.blogs / "example.org",
) )
self.assertEqual( self.assertEqual(
server.safe_relative_path("posts/career/example.org"), server.safe_relative_path("posts/career/example.org"),
self.posts / "career" / "example.org", self.posts / "career" / "example.org",
) )
self.assertEqual( self.assertEqual(
server.safe_relative_path("lima/index.md"), server.safe_relative_path("lima/index.md"),
self.lima / "index.md", self.lima / "index.md",
) )
self.assertEqual( self.assertEqual(
server.safe_relative_path("home/notes.org"), server.safe_relative_path("home/notes.org"),
self.home / "notes.org", self.home / "notes.org",
) )
def test_safe_relative_path_rejects_escapes_and_wrong_locations(self): def test_safe_relative_path_rejects_escapes_and_wrong_locations(self):
for path in ("../secret.org", "/tmp/secret.org", "sitemap.org", "assets/style.org", "tags/life.org"): for path in ("../secret.org", "/tmp/secret.org", "sitemap.org", "assets/style.org", "tags/life.org"):
with self.subTest(path=path): with self.subTest(path=path):
with self.assertRaises(ValueError): with self.assertRaises(ValueError):
server.safe_relative_path(path) server.safe_relative_path(path)
def test_image_dimensions_detects_png_and_gif(self): def test_image_dimensions_detects_png_and_gif(self):
png = b"\x89PNG\r\n\x1a\n" + b"\x00" * 8 + (640).to_bytes(4, "big") + (480).to_bytes(4, "big") png = b"\x89PNG\r\n\x1a\n" + b"\x00" * 8 + (640).to_bytes(4, "big") + (480).to_bytes(4, "big")
gif = b"GIF89a" + (320).to_bytes(2, "little") + (200).to_bytes(2, "little") gif = b"GIF89a" + (320).to_bytes(2, "little") + (200).to_bytes(2, "little")
self.assertEqual(server.image_dimensions(png, ".png"), (640, 480)) self.assertEqual(server.image_dimensions(png, ".png"), (640, 480))
self.assertEqual(server.image_dimensions(gif, ".gif"), (320, 200)) self.assertEqual(server.image_dimensions(gif, ".gif"), (320, 200))
self.assertIsNone(server.image_dimensions(b"not an image", ".png")) self.assertIsNone(server.image_dimensions(b"not an image", ".png"))
def test_parse_upload_form_reads_attachment_and_page_path(self): def test_parse_upload_form_reads_attachment_and_page_path(self):
boundary = "----authoring-test" boundary = "----authoring-test"
content_type = f"multipart/form-data; boundary={boundary}" content_type = f"multipart/form-data; boundary={boundary}"
body = ( body = (
f"--{boundary}\r\n" f"--{boundary}\r\n"
'Content-Disposition: form-data; name="pagePath"\r\n\r\n' 'Content-Disposition: form-data; name="pagePath"\r\n\r\n'
"lima/index.md\r\n" "lima/index.md\r\n"
f"--{boundary}\r\n" f"--{boundary}\r\n"
'Content-Disposition: form-data; name="attachment"; filename="photo.png"\r\n' 'Content-Disposition: form-data; name="attachment"; filename="photo.png"\r\n'
"Content-Type: image/png\r\n\r\n" "Content-Type: image/png\r\n\r\n"
).encode("utf-8") + b"image bytes\r\n" + f"--{boundary}--\r\n".encode("utf-8") ).encode("utf-8") + b"image bytes\r\n" + f"--{boundary}--\r\n".encode("utf-8")
filename, payload, page_path = server.parse_upload_form(content_type, body) filename, payload, page_path = server.parse_upload_form(content_type, body)
self.assertEqual(filename, "photo.png") self.assertEqual(filename, "photo.png")
self.assertEqual(payload, b"image bytes") self.assertEqual(payload, b"image bytes")
self.assertEqual(page_path, "lima/index.md") self.assertEqual(page_path, "lima/index.md")
class PageRenderingTests(AuthoringServerTestCase): class PageRenderingTests(AuthoringServerTestCase):
def test_render_org_writes_metadata_and_body(self): def test_render_org_writes_metadata_and_body(self):
rendered = server.render_org( rendered = server.render_org(
{ {
"title": "A New Note", "title": "A New Note",
"slug": "Custom Slug", "slug": "Custom Slug",
"tags": ["Life", "life", "Review"], "tags": ["Life", "life", "Review"],
"content": "Body text", "content": "Body text",
"date": "<2026-05-07 Thu 10:30>", "date": "<2026-05-07 Thu 10:30>",
"comments": False, "comments": False,
"wip": "draft", "wip": "draft",
}, },
previous=None, previous=None,
) )
self.assertIn("#+TITLE: A New Note", rendered) self.assertIn("#+TITLE: A New Note", rendered)
self.assertIn("#+DATE: <2026-05-07 Thu 10:30>", rendered) self.assertIn("#+DATE: <2026-05-07 Thu 10:30>", rendered)
self.assertIn("#+filetags: :life:review:", rendered) self.assertIn("#+filetags: :life:review:", rendered)
self.assertIn("#+COMMENTS: ", rendered) self.assertIn("#+COMMENTS: ", rendered)
self.assertIn("#+SLUG: custom-slug", rendered) self.assertIn("#+SLUG: custom-slug", rendered)
self.assertIn("#+WIP: draft", rendered) self.assertIn("#+WIP: draft", rendered)
self.assertTrue(rendered.endswith("Body text\n")) self.assertTrue(rendered.endswith("Body text\n"))
def test_render_markdown_adds_or_replaces_title_heading(self): def test_render_markdown_adds_or_replaces_title_heading(self):
self.assertEqual( self.assertEqual(
server.render_markdown({"title": "Family Update", "content": "Body"}), server.render_markdown({"title": "Family Update", "content": "Body"}),
"# Family Update\n\nBody\n", "# Family Update\n\nBody\n",
) )
self.assertEqual( self.assertEqual(
server.render_markdown({"title": "New Title", "content": "## Old\n\nBody"}), server.render_markdown({"title": "New Title", "content": "## Old\n\nBody"}),
"# New Title\n\nBody\n", "# New Title\n\nBody\n",
) )
def test_save_page_creates_blog_and_round_trips_content(self): def test_save_page_creates_blog_and_round_trips_content(self):
saved = server.save_page( saved = server.save_page(
{ {
"pageType": "blog", "pageType": "blog",
"title": "Test Post", "title": "Test Post",
"slug": "test-post", "slug": "test-post",
"date": "<2026-05-07 Thu 09:00>", "date": "<2026-05-07 Thu 09:00>",
"tags": "test, blog", "tags": "test, blog",
"content": "The body", "content": "The body",
"comments": True, "comments": True,
} }
) )
self.assertEqual(saved["path"], "blogs/2026/05-may/test-post.org") self.assertEqual(saved["path"], "blogs/2026/05-may/test-post.org")
self.assertEqual(saved["title"], "Test Post") self.assertEqual(saved["title"], "Test Post")
self.assertEqual(saved["tags"], ["test", "blog"]) self.assertEqual(saved["tags"], ["test", "blog"])
self.assertEqual(saved["content"], "The body") self.assertEqual(saved["content"], "The body")
def test_save_page_creates_lima_markdown(self): def test_save_page_creates_lima_markdown(self):
saved = server.save_page( saved = server.save_page(
{ {
"pageType": "lima", "pageType": "lima",
"title": "Lima Entry", "title": "Lima Entry",
"slug": "lima-entry", "slug": "lima-entry",
"content": "Some markdown", "content": "Some markdown",
} }
) )
path = self.lima / "lima-entry.md" path = self.lima / "lima-entry.md"
self.assertEqual(saved["path"], "lima/lima-entry.md") self.assertEqual(saved["path"], "lima/lima-entry.md")
self.assertEqual(path.read_text(encoding="utf-8"), "# Lima Entry\n\nSome markdown\n") self.assertEqual(path.read_text(encoding="utf-8"), "# Lima Entry\n\nSome markdown\n")
def test_list_pages_excludes_generated_and_sync_conflict_files(self): def test_list_pages_excludes_generated_and_sync_conflict_files(self):
(self.blogs / "keep.org").write_text("#+TITLE: Keep\n#+DATE: <2026-05-07 Thu 12:00>\n", encoding="utf-8") (self.blogs / "keep.org").write_text("#+TITLE: Keep\n#+DATE: <2026-05-07 Thu 12:00>\n", encoding="utf-8")
(self.posts / "posts-list.org").write_text("#+TITLE: Generated\n", encoding="utf-8") (self.posts / "posts-list.org").write_text("#+TITLE: Generated\n", encoding="utf-8")
(self.blogs / "note.sync-conflict-1.org").write_text("#+TITLE: Conflict\n", encoding="utf-8") (self.blogs / "note.sync-conflict-1.org").write_text("#+TITLE: Conflict\n", encoding="utf-8")
(self.lima / "index.md").write_text("# Lima Home\n", encoding="utf-8") (self.lima / "index.md").write_text("# Lima Home\n", encoding="utf-8")
(self.home / "notes.org").write_text("#+TITLE: Notes\n", encoding="utf-8") (self.home / "notes.org").write_text("#+TITLE: Notes\n", encoding="utf-8")
(self.tags / "life.org").write_text("#+TITLE: Tag\n", encoding="utf-8") (self.tags / "life.org").write_text("#+TITLE: Tag\n", encoding="utf-8")
paths = [page["path"] for page in server.list_pages()] paths = [page["path"] for page in server.list_pages()]
self.assertEqual(set(paths), {"blogs/keep.org", "home/notes.org", "lima/index.md"}) self.assertEqual(set(paths), {"blogs/keep.org", "home/notes.org", "lima/index.md"})
def test_list_pages_skips_files_that_fail_to_read(self): def test_list_pages_skips_files_that_fail_to_read(self):
good = self.blogs / "keep.org" good = self.blogs / "keep.org"
bad = self.blogs / "bad.org" bad = self.blogs / "bad.org"
good.write_text("#+TITLE: Keep\n#+DATE: <2026-05-07 Thu 12:00>\n", encoding="utf-8") good.write_text("#+TITLE: Keep\n#+DATE: <2026-05-07 Thu 12:00>\n", encoding="utf-8")
bad.write_text("#+TITLE: Bad\n", encoding="utf-8") bad.write_text("#+TITLE: Bad\n", encoding="utf-8")
original_read_page = server.read_page original_read_page = server.read_page
def read_page(path): def read_page(path):
if path == bad: if path == bad:
raise OSError("file disappeared") raise OSError("file disappeared")
return original_read_page(path) return original_read_page(path)
with mock.patch.object(server, "read_page", side_effect=read_page): with mock.patch.object(server, "read_page", side_effect=read_page):
paths = [page["path"] for page in server.list_pages()] paths = [page["path"] for page in server.list_pages()]
self.assertEqual(paths, ["blogs/keep.org"]) self.assertEqual(paths, ["blogs/keep.org"])
def test_server_diagnostics_reports_root_and_page_count(self): def test_server_diagnostics_reports_root_and_page_count(self):
(self.blogs / "keep.org").write_text("#+TITLE: Keep\n#+DATE: <2026-05-07 Thu 12:00>\n", encoding="utf-8") (self.blogs / "keep.org").write_text("#+TITLE: Keep\n#+DATE: <2026-05-07 Thu 12:00>\n", encoding="utf-8")
diagnostics = server.server_diagnostics() diagnostics = server.server_diagnostics()
self.assertEqual(diagnostics["root"], self.root.as_posix()) self.assertEqual(diagnostics["root"], self.root.as_posix())
self.assertEqual(diagnostics["pageCount"], 1) self.assertEqual(diagnostics["pageCount"], 1)
self.assertEqual(diagnostics["firstPage"], "blogs/keep.org") self.assertEqual(diagnostics["firstPage"], "blogs/keep.org")
def test_relative_asset_path_is_calculated_from_lima_page_directory(self): def test_relative_asset_path_is_calculated_from_lima_page_directory(self):
self.assertEqual( self.assertEqual(
server.relative_asset_path("lima/family/update.md", "assets/images/hzone/pic.png"), server.relative_asset_path("lima/family/update.md", "assets/images/hzone/pic.png"),
"../../assets/images/hzone/pic.png", "../../assets/images/hzone/pic.png",
) )
self.assertEqual( self.assertEqual(
server.relative_asset_path("blogs/example.org", "assets/images/hzone/pic.png"), server.relative_asset_path("blogs/example.org", "assets/images/hzone/pic.png"),
"../assets/images/hzone/pic.png", "../assets/images/hzone/pic.png",
) )
def test_save_upload_stores_post_images_by_section_and_returns_org_link(self): def test_save_upload_stores_post_images_by_section_and_returns_org_link(self):
with mock.patch.object(server, "datetime") as datetime_mock: with mock.patch.object(server, "datetime") as datetime_mock:
datetime_mock.now.return_value = datetime(2026, 5, 7, 9, 30) datetime_mock.now.return_value = datetime(2026, 5, 7, 9, 30)
saved = server.save_upload( saved = server.save_upload(
"Stress In Workplace.png", "Stress In Workplace.png",
b"image bytes", b"image bytes",
"posts/career/management-of-self.org", "posts/career/management-of-self.org",
) )
target = self.root / "assets" / "images" / "career" / "2026-05-07-stress-in-workplace.png" target = self.root / "assets" / "images" / "career" / "2026-05-07-stress-in-workplace.png"
self.assertEqual(target.read_bytes(), b"image bytes") self.assertEqual(target.read_bytes(), b"image bytes")
self.assertEqual(saved["path"], "assets/images/career/2026-05-07-stress-in-workplace.png") self.assertEqual(saved["path"], "assets/images/career/2026-05-07-stress-in-workplace.png")
self.assertEqual(saved["relativeUrl"], "../../assets/images/career/2026-05-07-stress-in-workplace.png") self.assertEqual(saved["relativeUrl"], "../../assets/images/career/2026-05-07-stress-in-workplace.png")
self.assertEqual(saved["insertText"], "[[../../assets/images/career/2026-05-07-stress-in-workplace.png]]") self.assertEqual(saved["insertText"], "[[../../assets/images/career/2026-05-07-stress-in-workplace.png]]")
def test_save_upload_keeps_blog_images_in_blog_folder(self): def test_save_upload_keeps_blog_images_in_blog_folder(self):
with mock.patch.object(server, "datetime") as datetime_mock: with mock.patch.object(server, "datetime") as datetime_mock:
datetime_mock.now.return_value = datetime(2026, 5, 7, 9, 30) datetime_mock.now.return_value = datetime(2026, 5, 7, 9, 30)
saved = server.save_upload( saved = server.save_upload(
"Lunch.jpg", "Lunch.jpg",
b"image bytes", b"image bytes",
"blogs/2026/05-may/lunch.org", "blogs/2026/05-may/lunch.org",
) )
self.assertEqual(saved["path"], "assets/images/blogs/2026-05-07-lunch.jpg") self.assertEqual(saved["path"], "assets/images/blogs/2026-05-07-lunch.jpg")
self.assertEqual(saved["insertText"], "[[../../../assets/images/blogs/2026-05-07-lunch.jpg]]") self.assertEqual(saved["insertText"], "[[../../../assets/images/blogs/2026-05-07-lunch.jpg]]")
def test_save_upload_inserts_absolute_public_url_for_markdown(self): def test_save_upload_inserts_absolute_public_url_for_markdown(self):
with mock.patch.object(server, "datetime") as datetime_mock: with mock.patch.object(server, "datetime") as datetime_mock:
datetime_mock.now.return_value = datetime(2026, 5, 9, 9, 30) datetime_mock.now.return_value = datetime(2026, 5, 9, 9, 30)
saved = server.save_upload( saved = server.save_upload(
"Screen Shot.png", "Screen Shot.png",
b"image bytes", b"image bytes",
"lima/index.md", "lima/index.md",
) )
self.assertEqual(saved["path"], "assets/images/hzone/2026-05-09-screen-shot.png") self.assertEqual(saved["path"], "assets/images/hzone/2026-05-09-screen-shot.png")
self.assertEqual(saved["relativeUrl"], "../assets/images/hzone/2026-05-09-screen-shot.png") self.assertEqual(saved["relativeUrl"], "../assets/images/hzone/2026-05-09-screen-shot.png")
self.assertEqual( self.assertEqual(
saved["insertText"], saved["insertText"],
"![2026-05-09-screen-shot.png](https://zainezq.com/assets/images/hzone/2026-05-09-screen-shot.png)", "![2026-05-09-screen-shot.png](https://zainezq.com/assets/images/hzone/2026-05-09-screen-shot.png)",
) )
def test_save_upload_rejects_disallowed_extensions(self): def test_save_upload_rejects_disallowed_extensions(self):
with self.assertRaises(ValueError): with self.assertRaises(ValueError):
server.save_upload("shell.php", b"<?php") server.save_upload("shell.php", b"<?php")
with self.assertRaises(ValueError): with self.assertRaises(ValueError):
server.save_upload("movie.mp4", b"video") server.save_upload("movie.mp4", b"video")
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 = build_server.BuildQueue() queue = build_server.BuildQueue()
job = build_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
job.message = "Done" job.message = "Done"
job.log = "build log" job.log = "build log"
queue._recent.append(job) queue._recent.append(job)
snapshot = queue.snapshot() snapshot = queue.snapshot()
self.assertFalse(snapshot["running"]) self.assertFalse(snapshot["running"])
self.assertEqual(snapshot["message"], "Done") self.assertEqual(snapshot["message"], "Done")
self.assertEqual(snapshot["recent"][0]["status"], "done") self.assertEqual(snapshot["recent"][0]["status"], "done")
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 = build_server.BuildQueue() queue = build_server.BuildQueue()
with mock.patch.object(build_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 = build_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 = build_server.BuildQueue() queue = build_server.BuildQueue()
with mock.patch.object(build_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 = build_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")
self.assertEqual(queued["path"], "assets/content/hidden-details.json") self.assertEqual(queued["path"], "assets/content/hidden-details.json")
self.assertEqual(queued["title"], "Hidden Memory Observatory") self.assertEqual(queued["title"], "Hidden Memory Observatory")
def test_run_build_commands_streams_output_to_callback(self): def test_run_build_commands_streams_output_to_callback(self):
root = Path(tempfile.mkdtemp()) root = Path(tempfile.mkdtemp())
try: try:
venv_python = root / ".venv" / "bin" / "python" venv_python = root / ".venv" / "bin" / "python"
venv_python.parent.mkdir(parents=True) venv_python.parent.mkdir(parents=True)
venv_python.write_text("", encoding="utf-8") venv_python.write_text("", encoding="utf-8")
seen = [] seen = []
class FakeProcess: class FakeProcess:
def __init__(self, lines, return_code=0): def __init__(self, lines, return_code=0):
self.stdout = iter(lines) self.stdout = iter(lines)
self.return_code = return_code self.return_code = return_code
def wait(self): def wait(self):
return self.return_code return self.return_code
processes = [ processes = [
FakeProcess(["emacs line 1\n", "emacs line 2\n"]), FakeProcess(["emacs line 1\n", "emacs line 2\n"]),
FakeProcess(["index line\n"]), FakeProcess(["index line\n"]),
] ]
with mock.patch.object(build_server, "ROOT", root), mock.patch.object(build_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 = build_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)
self.assertIn("emacs line 1\n", seen) self.assertIn("emacs line 1\n", seen)
self.assertIn("index line\n", seen) self.assertIn("index line\n", seen)
self.assertEqual(log, "".join(seen)) self.assertEqual(log, "".join(seen))
finally: finally:
for path in sorted(root.rglob("*"), reverse=True): for path in sorted(root.rglob("*"), reverse=True):
if path.is_file(): if path.is_file():
path.unlink() path.unlink()
else: else:
path.rmdir() path.rmdir()
root.rmdir() root.rmdir()
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()

View File

@@ -1,18 +1,18 @@
[Unit] [Unit]
Description=Org web authoring UI Description=Org web authoring UI
After=network.target After=network.target
[Service] [Service]
Type=simple Type=simple
WorkingDirectory=/home/zaine/master-folder/org-platform/authoring-service WorkingDirectory=/home/zaine/master-folder/org-platform/authoring-service
Environment=AUTHOR_CONTENT_ROOT=/home/zaine/master-folder/org-platform/org_web Environment=AUTHOR_CONTENT_ROOT=/home/zaine/master-folder/org-platform/org_web
Environment=AUTHOR_HIDDEN_BACKUP_DIR=/home/zaine/master-folder/org-platform/authoring-service/backups/hidden-details Environment=AUTHOR_HIDDEN_BACKUP_DIR=/home/zaine/master-folder/org-platform/authoring-service/backups/hidden-details
Environment=AUTHOR_PORT=8765 Environment=AUTHOR_PORT=8765
Environment=PYTHONUNBUFFERED=1 Environment=PYTHONUNBUFFERED=1
ExecStartPre=/home/zaine/master-folder/org-platform/authoring-service/.venv/bin/python -c "import authoring_server as s; pages=s.list_pages(); print(f'Authoring root: {s.ROOT}'); print(f'Editable pages: {len(pages)}'); raise SystemExit(0 if pages else 1)" ExecStartPre=/home/zaine/master-folder/org-platform/authoring-service/.venv/bin/python -c "import authoring_server as s; pages=s.list_pages(); print(f'Authoring root: {s.ROOT}'); print(f'Editable pages: {len(pages)}'); raise SystemExit(0 if pages else 1)"
ExecStart=/home/zaine/master-folder/org-platform/authoring-service/.venv/bin/python -u /home/zaine/master-folder/org-platform/authoring-service/authoring_server.py ExecStart=/home/zaine/master-folder/org-platform/authoring-service/.venv/bin/python -u /home/zaine/master-folder/org-platform/authoring-service/authoring_server.py
Restart=on-failure Restart=on-failure
RestartSec=5 RestartSec=5
[Install] [Install]
WantedBy=default.target WantedBy=default.target