Compare commits

6 Commits

Author SHA1 Message Date
19a63069b2 Improve mobile navigation and responsive layout
All checks were successful
Build Authoring Service / build (push) Successful in 9s
2026-07-23 23:08:27 +01:00
1a49838b58 updates
All checks were successful
Build Authoring Service / build (push) Successful in 8s
2026-07-14 12:38:20 +01:00
2f0009df56 adding more things
All checks were successful
Build Authoring Service / build (push) Successful in 9s
2026-07-10 10:28:47 +01:00
8329c00fcc testing
All checks were successful
Build Authoring Service / build (push) Successful in 8s
2026-07-09 17:06:10 +01:00
4dd619d02c Merge pull request 'Improvements' (#1) from zq/work/constellation-fix into main
All checks were successful
Build Authoring Service / build (push) Successful in 9s
Reviewed-on: #1
2026-07-09 16:36:26 +01:00
Zaine
e9e3954f31 Improvements 2026-07-09 16:34:28 +01:00
20 changed files with 7332 additions and 6480 deletions

View File

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

10
.gitignore vendored
View File

@@ -1,5 +1,5 @@
.venv/
__pycache__/
*.py[cod]
.pytest_cache/
backups/
.venv/
__pycache__/
*.py[cod]
.pytest_cache/
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
VENV := .venv
PY := $(VENV)/bin/python
PIP := $(VENV)/bin/pip
AUTHOR_PORT := 8765
AUTHOR_URL := http://127.0.0.1:$(AUTHOR_PORT)
AUTHOR_PAGES_CHECK := /tmp/authoring-server-pages.json
AUTHOR_SERVICE := org-web-authoring.service
AUTHOR_UNIT_SRC := systemd/user/$(AUTHOR_SERVICE)
SYSTEMD_USER_DIR := $(HOME)/.config/systemd/user
AUTHOR_UNIT_DST := $(SYSTEMD_USER_DIR)/$(AUTHOR_SERVICE)
USER_ID := $(shell id -u)
SYSTEMD_RUNTIME_DIR ?= /run/user/$(USER_ID)
SYSTEMCTL_USER := env XDG_RUNTIME_DIR=$(SYSTEMD_RUNTIME_DIR) systemctl --user
JOURNALCTL_USER := env XDG_RUNTIME_DIR=$(SYSTEMD_RUNTIME_DIR) journalctl --user
all: author-restart
test: $(VENV)
PYTHONPATH=src $(PY) -m unittest discover -s src/tests -p 'test_authoring_server.py' -v
$(VENV):
@echo "Generating virtual environment"
python3 -m venv $(VENV)
$(PIP) install -r requirements.txt
author: author-start
author-install: $(VENV)
@echo "Installing $(AUTHOR_SERVICE) for the current user..."
@mkdir -p "$(SYSTEMD_USER_DIR)"
install -m 0644 "$(AUTHOR_UNIT_SRC)" "$(AUTHOR_UNIT_DST)"
$(SYSTEMCTL_USER) daemon-reload
$(SYSTEMCTL_USER) enable "$(AUTHOR_SERVICE)"
author-start: author-install author-stop-legacy
@echo "Starting authoring UI with systemd..."
$(SYSTEMCTL_USER) start "$(AUTHOR_SERVICE)"
@$(MAKE) --no-print-directory author-health
author-stop:
@echo "Stopping authoring UI with systemd..."
$(SYSTEMCTL_USER) stop "$(AUTHOR_SERVICE)" || true
@$(MAKE) --no-print-directory author-stop-legacy
author-stop-legacy:
@echo "Stopping any legacy process on authoring port $(AUTHOR_PORT)..."
@fuser -k $(AUTHOR_PORT)/tcp >/dev/null 2>&1 || true
author-restart: author-install author-stop-legacy
@echo "Restarting authoring UI with systemd..."
$(SYSTEMCTL_USER) restart "$(AUTHOR_SERVICE)"
@$(MAKE) --no-print-directory author-health
author-health:
@echo "Waiting for authoring UI and page list to respond..."
@for i in 1 2 3 4 5 6 7 8 9 10; do \
if ! $(SYSTEMCTL_USER) is-active --quiet "$(AUTHOR_SERVICE)"; then \
echo "Authoring UI service is not active. Service status:"; \
$(SYSTEMCTL_USER) status --no-pager "$(AUTHOR_SERVICE)" || true; \
echo "Recent logs:"; \
$(JOURNALCTL_USER) -u "$(AUTHOR_SERVICE)" -n 80 --no-pager || true; \
exit 1; \
fi; \
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 is running on $(AUTHOR_URL)/"; \
exit 0; \
fi; \
sleep 1; \
done; \
echo "Authoring UI failed to respond. Service status:"; \
$(SYSTEMCTL_USER) status --no-pager "$(AUTHOR_SERVICE)" || true; \
echo "Recent logs:"; \
$(JOURNALCTL_USER) -u "$(AUTHOR_SERVICE)" -n 80 --no-pager || true; \
exit 1
author-status:
@$(SYSTEMCTL_USER) status --no-pager "$(AUTHOR_SERVICE)" || true
@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
author-diagnostics:
@curl -fsS "$(AUTHOR_URL)/api/diagnostics" || true
author-logs:
$(JOURNALCTL_USER) -u "$(AUTHOR_SERVICE)" -n 100 --no-pager
clean-venv:
@echo "Cleaning virtual environment..."
rm -rf $(VENV)
help:
@echo "Available targets:"
@echo " make test - Run authoring server tests"
@echo " make author - Install and start the authoring UI service"
@echo " make author-install - Install and enable the authoring UI service"
@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"
.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
PY := $(VENV)/bin/python
PIP := $(VENV)/bin/pip
AUTHOR_PORT := 8765
AUTHOR_URL := http://127.0.0.1:$(AUTHOR_PORT)
AUTHOR_PAGES_CHECK := /tmp/authoring-server-pages.json
AUTHOR_SERVICE := org-web-authoring.service
AUTHOR_UNIT_SRC := systemd/user/$(AUTHOR_SERVICE)
SYSTEMD_USER_DIR := $(HOME)/.config/systemd/user
AUTHOR_UNIT_DST := $(SYSTEMD_USER_DIR)/$(AUTHOR_SERVICE)
USER_ID := $(shell id -u)
SYSTEMD_RUNTIME_DIR ?= /run/user/$(USER_ID)
SYSTEMCTL_USER := env XDG_RUNTIME_DIR=$(SYSTEMD_RUNTIME_DIR) systemctl --user
JOURNALCTL_USER := env XDG_RUNTIME_DIR=$(SYSTEMD_RUNTIME_DIR) journalctl --user
all: author-restart
test: $(VENV)
PYTHONPATH=src $(PY) -m unittest discover -s src/tests -p 'test_authoring_server.py' -v
$(VENV):
@echo "Generating virtual environment"
python3 -m venv $(VENV)
$(PIP) install -r requirements.txt
author: author-start
author-install: $(VENV)
@echo "Installing $(AUTHOR_SERVICE) for the current user..."
@mkdir -p "$(SYSTEMD_USER_DIR)"
install -m 0644 "$(AUTHOR_UNIT_SRC)" "$(AUTHOR_UNIT_DST)"
$(SYSTEMCTL_USER) daemon-reload
$(SYSTEMCTL_USER) enable "$(AUTHOR_SERVICE)"
author-start: author-install author-stop-legacy
@echo "Starting authoring UI with systemd..."
$(SYSTEMCTL_USER) start "$(AUTHOR_SERVICE)"
@$(MAKE) --no-print-directory author-health
author-dev:
@echo "Starting authoring UI without systemd..."
$(PY) authoring_server.py
@$(MAKE) --no-print-directory author-health
author-stop:
@echo "Stopping authoring UI with systemd..."
$(SYSTEMCTL_USER) stop "$(AUTHOR_SERVICE)" || true
@$(MAKE) --no-print-directory author-stop-legacy
author-stop-legacy:
@echo "Stopping any legacy process on authoring port $(AUTHOR_PORT)..."
@fuser -k $(AUTHOR_PORT)/tcp >/dev/null 2>&1 || true
author-restart: author-install author-stop-legacy
@echo "Restarting authoring UI with systemd..."
$(SYSTEMCTL_USER) restart "$(AUTHOR_SERVICE)"
@$(MAKE) --no-print-directory author-health
author-health:
@echo "Waiting for authoring UI and page list to respond..."
@for i in 1 2 3 4 5 6 7 8 9 10; do \
if ! $(SYSTEMCTL_USER) is-active --quiet "$(AUTHOR_SERVICE)"; then \
echo "Authoring UI service is not active. Service status:"; \
$(SYSTEMCTL_USER) status --no-pager "$(AUTHOR_SERVICE)" || true; \
echo "Recent logs:"; \
$(JOURNALCTL_USER) -u "$(AUTHOR_SERVICE)" -n 80 --no-pager || true; \
exit 1; \
fi; \
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 is running on $(AUTHOR_URL)/"; \
exit 0; \
fi; \
sleep 1; \
done; \
echo "Authoring UI failed to respond. Service status:"; \
$(SYSTEMCTL_USER) status --no-pager "$(AUTHOR_SERVICE)" || true; \
echo "Recent logs:"; \
$(JOURNALCTL_USER) -u "$(AUTHOR_SERVICE)" -n 80 --no-pager || true; \
exit 1
author-status:
@$(SYSTEMCTL_USER) status --no-pager "$(AUTHOR_SERVICE)" || true
@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
author-diagnostics:
@curl -fsS "$(AUTHOR_URL)/api/diagnostics" || true
author-logs:
$(JOURNALCTL_USER) -u "$(AUTHOR_SERVICE)" -n 100 --no-pager
clean-venv:
@echo "Cleaning virtual environment..."
rm -rf $(VENV)
help:
@echo "Available targets:"
@echo " make test - Run authoring server tests"
@echo " make author - Install and start the authoring UI service"
@echo " make author-install - Install and enable the authoring UI service"
@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
Standalone local authoring UI for the Org website content in:
`/home/zaine/master-folder/org-platform/org_web`
Run:
```sh
make author
```
Then open `http://127.0.0.1:8765`.
The service is installed as the user systemd unit `org-web-authoring.service`.
Useful commands:
- `make test` runs the authoring server unit tests.
- `make author-install` installs and enables the user service.
- `make author-restart` restarts the service.
- `make author-status` shows service status and editable page count.
- `make author-diagnostics` shows runtime diagnostics.
- `make author-logs` shows recent service logs.
The service code runs from this repository. The editable content root is set by `AUTHOR_CONTENT_ROOT` in `systemd/user/org-web-authoring.service`; it currently points at the website repository. Hidden-details backups are stored in this repository under `backups/hidden-details`.
Code layout:
- `authoring_server.py` is the compatibility executable used by systemd.
- `src/authoring_service/` contains the implementation modules.
- `src/tests/` contains the unit tests.
- `docs/hidden-memory-architecture.md` documents the Fragment/Memory/Story model used by the Hidden Memory Observatory.
## Authoring service
Standalone local authoring UI for the Org website content in:
`/home/zaine/master-folder/org-platform/org_web`
Run:
```sh
make author
```
Then open `http://127.0.0.1:8765`.
The service is installed as the user systemd unit `org-web-authoring.service`.
Useful commands:
- `make test` runs the authoring server unit tests.
- `make author-install` installs and enables the user service.
- `make author-restart` restarts the service.
- `make author-status` shows service status and editable page count.
- `make author-diagnostics` shows runtime diagnostics.
- `make author-logs` shows recent service logs.
The service code runs from this repository. The editable content root is set by `AUTHOR_CONTENT_ROOT` in `systemd/user/org-web-authoring.service`; it currently points at the website repository. Hidden-details backups are stored in this repository under `backups/hidden-details`.
Code layout:
- `authoring_server.py` is the compatibility executable used by systemd.
- `src/authoring_service/` contains the implementation modules.
- `src/tests/` contains the unit tests.
- `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,
"generatedAt": "2026-06-17T20:42:16",
"generatedAt": "2026-07-09T16:33:00",
"entries": [],
"stories": []
}

View File

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

View File

@@ -1,82 +1,82 @@
# Hidden Memory Architecture
The hidden ecosystem has one friendly source of truth:
- `assets/content/hidden-details.json` in `org_web`
The live site still consumes generated constants in:
- `assets/scripts/hidden-details.js` in `org_web`
## 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 newer `contentClass` field answers: what is this thing conceptually?
Use these classes:
- `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.
- `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.
- `lore`: world material such as dreams, terminal logs, guestbook entries, or hidden archive objects.
- `system layer`: structural observatory material, such as the layer guide.
## Stories
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.
## 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.
## Observatory Roles
Use `observatoryRole` to keep rendering calm:
- `ambient`: small lights and flavor, usually fragments.
- `node`: substantial emotional points, usually memories or story material.
- `event`: triggered behavior, routes, search, keyboard, seasonal, and play content.
- `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.
## Examples
`lima left this page a little steadier than she found it.`
- Class: `fragment`
- Runtime kind: `hidden tooltip`
- Surfaces: `tooltip`
- Observatory role: `ambient`
`July 2022, on the way to uni induction, lima pops into my life`
- Class: `memory`
- Runtime kind: `journal entry`
- Surfaces: `story`, `observatory`
- Observatory role: `node`
`search query "lima" opens a hidden route`
- Class: `interaction`
- Runtime kind: `search route`
- Surfaces: `search`, `hidden route`
- Observatory role: `event`
`The story of love between two souls`
- Class: story record, not an entry class
- References: ordered entry ids in `nodes`
- Purpose: emotional route, not duplicated content
## Authoring Rule
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.
# Hidden Memory Architecture
The hidden ecosystem has one friendly source of truth:
- `assets/content/hidden-details.json` in `org_web`
The live site still consumes generated constants in:
- `assets/scripts/features/hidden-details.js` in `org_web`
## 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 newer `contentClass` field answers: what is this thing conceptually?
Use these classes:
- `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.
- `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.
- `lore`: world material such as dreams, terminal logs, guestbook entries, or hidden archive objects.
- `system layer`: structural observatory material, such as the layer guide.
## Stories
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.
## 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.
## Observatory Roles
Use `observatoryRole` to keep rendering calm:
- `ambient`: small lights and flavor, usually fragments.
- `node`: substantial emotional points, usually memories or story material.
- `event`: triggered behavior, routes, search, keyboard, seasonal, and play content.
- `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.
## Examples
`lima left this page a little steadier than she found it.`
- Class: `fragment`
- Runtime kind: `hidden tooltip`
- Surfaces: `tooltip`
- Observatory role: `ambient`
`July 2022, on the way to uni induction, lima pops into my life`
- Class: `memory`
- Runtime kind: `journal entry`
- Surfaces: `story`, `observatory`
- Observatory role: `node`
`search query "lima" opens a hidden route`
- Class: `interaction`
- Runtime kind: `search route`
- Surfaces: `search`, `hidden route`
- Observatory role: `event`
`The story of love between two souls`
- Class: story record, not an entry class
- References: ordered entry ids in `nodes`
- Purpose: emotional route, not duplicated content
## Authoring Rule
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.

View File

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

View File

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

View File

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

View File

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

View File

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

View File

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

File diff suppressed because it is too large Load Diff

View File

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

File diff suppressed because it is too large Load Diff

View File

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

View File

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

View File

@@ -1,44 +1,160 @@
import os
import sys
import tempfile
import unittest
from datetime import datetime
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import os
import sys
import tempfile
import unittest
from datetime import datetime
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
import authoring_service.build as build_server
import authoring_service.config as config_server
import authoring_service.content as server
import authoring_service.hidden as hidden_server
import authoring_service.utils as utils_server
class AuthoringServerTestCase(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.root = Path(self.tmp.name)
self.blogs = self.root / "blogs"
self.posts = self.root / "posts"
self.lima = self.root / "lima"
self.home = self.root / "home"
self.tags = self.root / "tags"
self.hzone = self.root / "assets" / "images" / "hzone"
self.blogs.mkdir()
self.posts.mkdir()
self.lima.mkdir()
self.home.mkdir()
self.tags.mkdir()
patches = {
"ROOT": self.root,
"BLOGS_DIR": self.blogs,
"POSTS_DIR": self.posts,
"LIMA_DIR": self.lima,
"IMAGE_ASSETS_DIR": self.root / "assets" / "images",
"HZONE_ASSETS_DIR": self.hzone,
}
self.patchers = [mock.patch.object(server, name, value) for name, value in patches.items()]
for patcher in self.patchers:
patcher.start()
def tearDown(self):
for patcher in reversed(self.patchers):
patcher.stop()
self.tmp.cleanup()
class UtilityTests(AuthoringServerTestCase):
def test_resolve_root_ignores_invalid_author_root(self):
wrong_root = self.root / "empty"
wrong_root.mkdir()
(self.root / "authoring_server.py").write_text("", encoding="utf-8")
with mock.patch.dict(os.environ, {"AUTHOR_ROOT": str(wrong_root), "GITHUB_WORKSPACE": ""}), mock.patch.object(config_server.Path, "cwd", return_value=self.root):
self.assertEqual(config_server.resolve_root(), self.root)
def test_slugify_normalises_text_and_keeps_fallback(self):
self.assertEqual(utils_server.slugify("Hello, Org Web!"), "hello-org-web")
self.assertEqual(utils_server.slugify(" "), "untitled")
def test_normalise_tags_accepts_strings_and_deduplicates(self):
self.assertEqual(
utils_server.normalise_tags("Life, review:Life Emacs"),
["life", "review", "emacs"],
)
def test_parse_org_datetime_handles_date_and_optional_time(self):
self.assertEqual(
utils_server.parse_org_datetime("<2026-05-07 Thu 14:35>"),
datetime(2026, 5, 7, 14, 35),
)
self.assertEqual(
utils_server.parse_org_datetime("<2026-05-07 Thu>"),
datetime(2026, 5, 7, 12, 0),
)
self.assertIsNone(utils_server.parse_org_datetime("2026-05-07"))
def test_safe_relative_path_allows_expected_content_roots(self):
self.assertEqual(
server.safe_relative_path("blogs/example.org"),
self.blogs / "example.org",
)
self.assertEqual(
server.safe_relative_path("posts/career/example.org"),
self.posts / "career" / "example.org",
)
self.assertEqual(
server.safe_relative_path("lima/index.md"),
self.lima / "index.md",
)
self.assertEqual(
server.safe_relative_path("home/notes.org"),
self.home / "notes.org",
)
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"):
with self.subTest(path=path):
with self.assertRaises(ValueError):
server.safe_relative_path(path)
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")
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(gif, ".gif"), (320, 200))
self.assertIsNone(server.image_dimensions(b"not an image", ".png"))
def test_parse_upload_form_reads_attachment_and_page_path(self):
boundary = "----authoring-test"
content_type = f"multipart/form-data; boundary={boundary}"
body = (
f"--{boundary}\r\n"
'Content-Disposition: form-data; name="pagePath"\r\n\r\n'
"lima/index.md\r\n"
f"--{boundary}\r\n"
'Content-Disposition: form-data; name="attachment"; filename="photo.png"\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")
filename, payload, page_path = server.parse_upload_form(content_type, body)
self.assertEqual(filename, "photo.png")
self.assertEqual(payload, b"image bytes")
self.assertEqual(page_path, "lima/index.md")
class AuthoringServerTestCase(unittest.TestCase):
class HiddenLifeArcTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.root = Path(self.tmp.name)
self.blogs = self.root / "blogs"
self.posts = self.root / "posts"
self.lima = self.root / "lima"
self.home = self.root / "home"
self.tags = self.root / "tags"
self.hzone = self.root / "assets" / "images" / "hzone"
self.blogs.mkdir()
self.posts.mkdir()
self.lima.mkdir()
self.home.mkdir()
self.tags.mkdir()
patches = {
"ROOT": self.root,
"BLOGS_DIR": self.blogs,
"POSTS_DIR": self.posts,
"LIMA_DIR": self.lima,
"IMAGE_ASSETS_DIR": self.root / "assets" / "images",
"HZONE_ASSETS_DIR": self.hzone,
}
self.patchers = [mock.patch.object(server, name, value) for name, value in patches.items()]
self.content_json = self.root / "assets" / "content" / "hidden-details.json"
self.hidden_js = self.root / "assets" / "scripts" / "features" / "hidden-details.js"
self.backup_dir = self.root / "backups" / "hidden-details"
self.content_json.parent.mkdir(parents=True)
self.hidden_js.parent.mkdir(parents=True)
self.hidden_js.write_text(
"(function(){\n"
" // -----------------------------\n"
" // EDITABLE CONTENT\n"
" const familyLayers = [];\n"
" // -----------------------------\n"
" // STATE HELPERS\n"
"})();\n",
encoding="utf-8",
)
self.content_json.write_text('{"schemaVersion":3,"entries":[],"stories":[],"lifeArcs":[]}', encoding="utf-8")
self.patchers = [
mock.patch.object(hidden_server, "ROOT", self.root),
mock.patch.object(hidden_server, "HIDDEN_CONTENT_JSON", self.content_json),
mock.patch.object(hidden_server, "HIDDEN_DETAILS_JS", self.hidden_js),
mock.patch.object(hidden_server, "HIDDEN_BACKUP_DIR", self.backup_dir),
]
for patcher in self.patchers:
patcher.start()
@@ -47,325 +163,277 @@ class AuthoringServerTestCase(unittest.TestCase):
patcher.stop()
self.tmp.cleanup()
def test_save_hidden_store_preserves_life_arcs(self):
saved = hidden_server.save_hidden_store({
"entries": [],
"stories": [],
"lifeArcs": [{
"title": "Gym Journey",
"domain": "fitness",
"themes": ["consistency", "strength"],
"visibility": "private",
"chapters": [{
"title": "Starting Again",
"timeframe": "Summer 2026",
"purpose": "becoming someone who returns",
"storyDraft": "The first victory was returning.",
"moments": [{
"date": "2026-07-09",
"kind": "milestone",
"text": "First week back in the gym.",
"feeling": "steady",
"metric": "3 sessions",
"sourcePointer": "Obsidian: Gym",
}],
}],
}],
})
class UtilityTests(AuthoringServerTestCase):
def test_resolve_root_ignores_invalid_author_root(self):
wrong_root = self.root / "empty"
wrong_root.mkdir()
(self.root / "authoring_server.py").write_text("", encoding="utf-8")
with mock.patch.dict(os.environ, {"AUTHOR_ROOT": str(wrong_root), "GITHUB_WORKSPACE": ""}), mock.patch.object(config_server.Path, "cwd", return_value=self.root):
self.assertEqual(config_server.resolve_root(), self.root)
def test_slugify_normalises_text_and_keeps_fallback(self):
self.assertEqual(utils_server.slugify("Hello, Org Web!"), "hello-org-web")
self.assertEqual(utils_server.slugify(" "), "untitled")
def test_normalise_tags_accepts_strings_and_deduplicates(self):
self.assertEqual(
utils_server.normalise_tags("Life, review:Life Emacs"),
["life", "review", "emacs"],
)
def test_parse_org_datetime_handles_date_and_optional_time(self):
self.assertEqual(
utils_server.parse_org_datetime("<2026-05-07 Thu 14:35>"),
datetime(2026, 5, 7, 14, 35),
)
self.assertEqual(
utils_server.parse_org_datetime("<2026-05-07 Thu>"),
datetime(2026, 5, 7, 12, 0),
)
self.assertIsNone(utils_server.parse_org_datetime("2026-05-07"))
def test_safe_relative_path_allows_expected_content_roots(self):
self.assertEqual(
server.safe_relative_path("blogs/example.org"),
self.blogs / "example.org",
)
self.assertEqual(
server.safe_relative_path("posts/career/example.org"),
self.posts / "career" / "example.org",
)
self.assertEqual(
server.safe_relative_path("lima/index.md"),
self.lima / "index.md",
)
self.assertEqual(
server.safe_relative_path("home/notes.org"),
self.home / "notes.org",
)
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"):
with self.subTest(path=path):
with self.assertRaises(ValueError):
server.safe_relative_path(path)
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")
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(gif, ".gif"), (320, 200))
self.assertIsNone(server.image_dimensions(b"not an image", ".png"))
def test_parse_upload_form_reads_attachment_and_page_path(self):
boundary = "----authoring-test"
content_type = f"multipart/form-data; boundary={boundary}"
body = (
f"--{boundary}\r\n"
'Content-Disposition: form-data; name="pagePath"\r\n\r\n'
"lima/index.md\r\n"
f"--{boundary}\r\n"
'Content-Disposition: form-data; name="attachment"; filename="photo.png"\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")
filename, payload, page_path = server.parse_upload_form(content_type, body)
self.assertEqual(filename, "photo.png")
self.assertEqual(payload, b"image bytes")
self.assertEqual(page_path, "lima/index.md")
self.assertEqual(saved["lifeArcs"][0]["title"], "Gym Journey")
self.assertEqual(saved["lifeArcs"][0]["chapters"][0]["moments"][0]["metric"], "3 sessions")
reloaded = hidden_server.load_hidden_store()
self.assertEqual(reloaded["lifeArcs"][0]["themes"], ["consistency", "strength"])
self.assertEqual(reloaded["lifeArcs"][0]["chapters"][0]["storyDraft"], "The first victory was returning.")
class PageRenderingTests(AuthoringServerTestCase):
def test_render_org_writes_metadata_and_body(self):
rendered = server.render_org(
{
"title": "A New Note",
"slug": "Custom Slug",
"tags": ["Life", "life", "Review"],
"content": "Body text",
"date": "<2026-05-07 Thu 10:30>",
"comments": False,
"wip": "draft",
},
previous=None,
)
self.assertIn("#+TITLE: A New Note", rendered)
self.assertIn("#+DATE: <2026-05-07 Thu 10:30>", rendered)
self.assertIn("#+filetags: :life:review:", rendered)
self.assertIn("#+COMMENTS: ", rendered)
self.assertIn("#+SLUG: custom-slug", rendered)
self.assertIn("#+WIP: draft", rendered)
self.assertTrue(rendered.endswith("Body text\n"))
def test_render_markdown_adds_or_replaces_title_heading(self):
self.assertEqual(
server.render_markdown({"title": "Family Update", "content": "Body"}),
"# Family Update\n\nBody\n",
)
self.assertEqual(
server.render_markdown({"title": "New Title", "content": "## Old\n\nBody"}),
"# New Title\n\nBody\n",
)
def test_save_page_creates_blog_and_round_trips_content(self):
saved = server.save_page(
{
"pageType": "blog",
"title": "Test Post",
"slug": "test-post",
"date": "<2026-05-07 Thu 09:00>",
"tags": "test, blog",
"content": "The body",
"comments": True,
}
)
self.assertEqual(saved["path"], "blogs/2026/05-may/test-post.org")
self.assertEqual(saved["title"], "Test Post")
self.assertEqual(saved["tags"], ["test", "blog"])
self.assertEqual(saved["content"], "The body")
def test_save_page_creates_lima_markdown(self):
saved = server.save_page(
{
"pageType": "lima",
"title": "Lima Entry",
"slug": "lima-entry",
"content": "Some markdown",
}
)
path = self.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")
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.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.lima / "index.md").write_text("# Lima Home\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")
paths = [page["path"] for page in server.list_pages()]
self.assertEqual(set(paths), {"blogs/keep.org", "home/notes.org", "lima/index.md"})
def test_list_pages_skips_files_that_fail_to_read(self):
good = self.blogs / "keep.org"
bad = self.blogs / "bad.org"
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")
original_read_page = server.read_page
def read_page(path):
if path == bad:
raise OSError("file disappeared")
return original_read_page(path)
with mock.patch.object(server, "read_page", side_effect=read_page):
paths = [page["path"] for page in server.list_pages()]
self.assertEqual(paths, ["blogs/keep.org"])
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")
diagnostics = server.server_diagnostics()
self.assertEqual(diagnostics["root"], self.root.as_posix())
self.assertEqual(diagnostics["pageCount"], 1)
self.assertEqual(diagnostics["firstPage"], "blogs/keep.org")
def test_relative_asset_path_is_calculated_from_lima_page_directory(self):
self.assertEqual(
server.relative_asset_path("lima/family/update.md", "assets/images/hzone/pic.png"),
"../../assets/images/hzone/pic.png",
)
self.assertEqual(
server.relative_asset_path("blogs/example.org", "assets/images/hzone/pic.png"),
"../assets/images/hzone/pic.png",
)
def test_save_upload_stores_post_images_by_section_and_returns_org_link(self):
with mock.patch.object(server, "datetime") as datetime_mock:
datetime_mock.now.return_value = datetime(2026, 5, 7, 9, 30)
saved = server.save_upload(
"Stress In Workplace.png",
b"image bytes",
"posts/career/management-of-self.org",
)
target = self.root / "assets" / "images" / "career" / "2026-05-07-stress-in-workplace.png"
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["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]]")
def test_save_upload_keeps_blog_images_in_blog_folder(self):
with mock.patch.object(server, "datetime") as datetime_mock:
datetime_mock.now.return_value = datetime(2026, 5, 7, 9, 30)
saved = server.save_upload(
"Lunch.jpg",
b"image bytes",
"blogs/2026/05-may/lunch.org",
)
self.assertEqual(saved["path"], "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):
with mock.patch.object(server, "datetime") as datetime_mock:
datetime_mock.now.return_value = datetime(2026, 5, 9, 9, 30)
saved = server.save_upload(
"Screen Shot.png",
b"image bytes",
"lima/index.md",
)
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["insertText"],
"![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):
with self.assertRaises(ValueError):
server.save_upload("shell.php", b"<?php")
with self.assertRaises(ValueError):
server.save_upload("movie.mp4", b"video")
class BuildQueueTests(unittest.TestCase):
def test_snapshot_reports_recent_completed_job(self):
queue = build_server.BuildQueue()
job = build_server.BuildJob(1, "blogs/post.org", "Post")
job.started_at = 1.0
job.finished_at = 2.0
job.ok = True
job.message = "Done"
job.log = "build log"
queue._recent.append(job)
snapshot = queue.snapshot()
self.assertFalse(snapshot["running"])
self.assertEqual(snapshot["message"], "Done")
self.assertEqual(snapshot["recent"][0]["status"], "done")
self.assertEqual(snapshot["log"], "build log")
def test_queue_build_enqueues_from_page_data(self):
queue = build_server.BuildQueue()
with mock.patch.object(build_server, "BUILD_QUEUE", queue), mock.patch.object(queue, "_run_worker"):
queued = build_server.queue_build({"path": "blogs/post.org", "title": "Post"})
self.assertEqual(queued["id"], 1)
self.assertEqual(queued["status"], "queued")
self.assertEqual(queued["path"], "blogs/post.org")
def test_queue_hidden_build_enqueues_hidden_asset_publish(self):
queue = build_server.BuildQueue()
with mock.patch.object(build_server, "BUILD_QUEUE", queue), mock.patch.object(queue, "_run_worker"):
queued = build_server.queue_hidden_build()
self.assertEqual(queued["id"], 1)
self.assertEqual(queued["status"], "queued")
self.assertEqual(queued["path"], "assets/content/hidden-details.json")
self.assertEqual(queued["title"], "Hidden Memory Observatory")
def test_run_build_commands_streams_output_to_callback(self):
root = Path(tempfile.mkdtemp())
try:
venv_python = root / ".venv" / "bin" / "python"
venv_python.parent.mkdir(parents=True)
venv_python.write_text("", encoding="utf-8")
seen = []
class FakeProcess:
def __init__(self, lines, return_code=0):
self.stdout = iter(lines)
self.return_code = return_code
def wait(self):
return self.return_code
processes = [
FakeProcess(["emacs line 1\n", "emacs line 2\n"]),
FakeProcess(["index line\n"]),
]
with mock.patch.object(build_server, "ROOT", root), mock.patch.object(build_server.subprocess, "Popen", side_effect=processes):
ok, message, log = build_server.run_build_commands(seen.append)
self.assertTrue(ok)
self.assertIn("Build complete", message)
self.assertIn("emacs line 1\n", seen)
self.assertIn("index line\n", seen)
self.assertEqual(log, "".join(seen))
finally:
for path in sorted(root.rglob("*"), reverse=True):
if path.is_file():
path.unlink()
else:
path.rmdir()
root.rmdir()
if __name__ == "__main__":
unittest.main()
def test_render_org_writes_metadata_and_body(self):
rendered = server.render_org(
{
"title": "A New Note",
"slug": "Custom Slug",
"tags": ["Life", "life", "Review"],
"content": "Body text",
"date": "<2026-05-07 Thu 10:30>",
"comments": False,
"wip": "draft",
},
previous=None,
)
self.assertIn("#+TITLE: A New Note", rendered)
self.assertIn("#+DATE: <2026-05-07 Thu 10:30>", rendered)
self.assertIn("#+filetags: :life:review:", rendered)
self.assertIn("#+COMMENTS: ", rendered)
self.assertIn("#+SLUG: custom-slug", rendered)
self.assertIn("#+WIP: draft", rendered)
self.assertTrue(rendered.endswith("Body text\n"))
def test_render_markdown_adds_or_replaces_title_heading(self):
self.assertEqual(
server.render_markdown({"title": "Family Update", "content": "Body"}),
"# Family Update\n\nBody\n",
)
self.assertEqual(
server.render_markdown({"title": "New Title", "content": "## Old\n\nBody"}),
"# New Title\n\nBody\n",
)
def test_save_page_creates_blog_and_round_trips_content(self):
saved = server.save_page(
{
"pageType": "blog",
"title": "Test Post",
"slug": "test-post",
"date": "<2026-05-07 Thu 09:00>",
"tags": "test, blog",
"content": "The body",
"comments": True,
}
)
self.assertEqual(saved["path"], "blogs/2026/05-may/test-post.org")
self.assertEqual(saved["title"], "Test Post")
self.assertEqual(saved["tags"], ["test", "blog"])
self.assertEqual(saved["content"], "The body")
def test_save_page_creates_lima_markdown(self):
saved = server.save_page(
{
"pageType": "lima",
"title": "Lima Entry",
"slug": "lima-entry",
"content": "Some markdown",
}
)
path = self.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")
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.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.lima / "index.md").write_text("# Lima Home\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")
paths = [page["path"] for page in server.list_pages()]
self.assertEqual(set(paths), {"blogs/keep.org", "home/notes.org", "lima/index.md"})
def test_list_pages_skips_files_that_fail_to_read(self):
good = self.blogs / "keep.org"
bad = self.blogs / "bad.org"
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")
original_read_page = server.read_page
def read_page(path):
if path == bad:
raise OSError("file disappeared")
return original_read_page(path)
with mock.patch.object(server, "read_page", side_effect=read_page):
paths = [page["path"] for page in server.list_pages()]
self.assertEqual(paths, ["blogs/keep.org"])
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")
diagnostics = server.server_diagnostics()
self.assertEqual(diagnostics["root"], self.root.as_posix())
self.assertEqual(diagnostics["pageCount"], 1)
self.assertEqual(diagnostics["firstPage"], "blogs/keep.org")
def test_relative_asset_path_is_calculated_from_lima_page_directory(self):
self.assertEqual(
server.relative_asset_path("lima/family/update.md", "assets/images/hzone/pic.png"),
"../../assets/images/hzone/pic.png",
)
self.assertEqual(
server.relative_asset_path("blogs/example.org", "assets/images/hzone/pic.png"),
"../assets/images/hzone/pic.png",
)
def test_save_upload_stores_post_images_by_section_and_returns_org_link(self):
with mock.patch.object(server, "datetime") as datetime_mock:
datetime_mock.now.return_value = datetime(2026, 5, 7, 9, 30)
saved = server.save_upload(
"Stress In Workplace.png",
b"image bytes",
"posts/career/management-of-self.org",
)
target = self.root / "assets" / "images" / "career" / "2026-05-07-stress-in-workplace.png"
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["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]]")
def test_save_upload_keeps_blog_images_in_blog_folder(self):
with mock.patch.object(server, "datetime") as datetime_mock:
datetime_mock.now.return_value = datetime(2026, 5, 7, 9, 30)
saved = server.save_upload(
"Lunch.jpg",
b"image bytes",
"blogs/2026/05-may/lunch.org",
)
self.assertEqual(saved["path"], "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):
with mock.patch.object(server, "datetime") as datetime_mock:
datetime_mock.now.return_value = datetime(2026, 5, 9, 9, 30)
saved = server.save_upload(
"Screen Shot.png",
b"image bytes",
"lima/index.md",
)
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["insertText"],
"![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):
with self.assertRaises(ValueError):
server.save_upload("shell.php", b"<?php")
with self.assertRaises(ValueError):
server.save_upload("movie.mp4", b"video")
class BuildQueueTests(unittest.TestCase):
def test_snapshot_reports_recent_completed_job(self):
queue = build_server.BuildQueue()
job = build_server.BuildJob(1, "blogs/post.org", "Post")
job.started_at = 1.0
job.finished_at = 2.0
job.ok = True
job.message = "Done"
job.log = "build log"
queue._recent.append(job)
snapshot = queue.snapshot()
self.assertFalse(snapshot["running"])
self.assertEqual(snapshot["message"], "Done")
self.assertEqual(snapshot["recent"][0]["status"], "done")
self.assertEqual(snapshot["log"], "build log")
def test_queue_build_enqueues_from_page_data(self):
queue = build_server.BuildQueue()
with mock.patch.object(build_server, "BUILD_QUEUE", queue), mock.patch.object(queue, "_run_worker"):
queued = build_server.queue_build({"path": "blogs/post.org", "title": "Post"})
self.assertEqual(queued["id"], 1)
self.assertEqual(queued["status"], "queued")
self.assertEqual(queued["path"], "blogs/post.org")
def test_queue_hidden_build_enqueues_hidden_asset_publish(self):
queue = build_server.BuildQueue()
with mock.patch.object(build_server, "BUILD_QUEUE", queue), mock.patch.object(queue, "_run_worker"):
queued = build_server.queue_hidden_build()
self.assertEqual(queued["id"], 1)
self.assertEqual(queued["status"], "queued")
self.assertEqual(queued["path"], "assets/content/hidden-details.json")
self.assertEqual(queued["title"], "Hidden Memory Observatory")
def test_run_build_commands_streams_output_to_callback(self):
root = Path(tempfile.mkdtemp())
try:
venv_python = root / ".venv" / "bin" / "python"
venv_python.parent.mkdir(parents=True)
venv_python.write_text("", encoding="utf-8")
seen = []
class FakeProcess:
def __init__(self, lines, return_code=0):
self.stdout = iter(lines)
self.return_code = return_code
def wait(self):
return self.return_code
processes = [
FakeProcess(["emacs line 1\n", "emacs line 2\n"]),
FakeProcess(["index line\n"]),
]
with mock.patch.object(build_server, "ROOT", root), mock.patch.object(build_server.subprocess, "Popen", side_effect=processes):
ok, message, log = build_server.run_build_commands(seen.append)
self.assertTrue(ok)
self.assertIn("Build complete", message)
self.assertIn("emacs line 1\n", seen)
self.assertIn("index line\n", seen)
self.assertEqual(log, "".join(seen))
finally:
for path in sorted(root.rglob("*"), reverse=True):
if path.is_file():
path.unlink()
else:
path.rmdir()
root.rmdir()
if __name__ == "__main__":
unittest.main()

View File

@@ -1,18 +1,18 @@
[Unit]
Description=Org web authoring UI
After=network.target
[Service]
Type=simple
WorkingDirectory=/home/zaine/master-folder/org-platform/authoring-service
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_PORT=8765
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)"
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
RestartSec=5
[Install]
WantedBy=default.target
[Unit]
Description=Org web authoring UI
After=network.target
[Service]
Type=simple
WorkingDirectory=/home/zaine/master-folder/org-platform/authoring-service
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_PORT=8765
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)"
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
RestartSec=5
[Install]
WantedBy=default.target