#!/usr/bin/env python3 """Local authoring UI for the Org published website.""" from __future__ import annotations import json import os import re import subprocess import sys import threading import time from dataclasses import dataclass from datetime import datetime from email.utils import formatdate from http import HTTPStatus from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any from urllib.parse import parse_qs, urlparse ROOT = Path(__file__).resolve().parent BLOGS_DIR = ROOT / "blogs" POSTS_DIR = ROOT / "posts" GENERATED_ORG_NAMES = { "blogs-list.org", "posts-list.org", "career-list.org", "sitemap.org", "recently-updated.org", "wip.org", } MONTH_NAMES = [ "january", "february", "march", "april", "may", "june", "july", "august", "september", "october", "november", "december", ] def slugify(value: str) -> str: slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") return slug or "untitled" def normalise_tags(value: Any) -> list[str]: if isinstance(value, str): raw = re.split(r"[,:\s]+", value) elif isinstance(value, list): raw = [str(item) for item in value] else: raw = [] tags = [] for tag in raw: if not tag.strip(): continue clean = slugify(tag) if clean and clean not in tags: tags.append(clean) return tags def org_date(dt: datetime) -> str: return dt.strftime("<%Y-%m-%d %a %H:%M>") def parse_org_datetime(value: str | None) -> datetime | None: if not value: return None match = re.search(r"<(\d{4})-(\d{2})-(\d{2})(?:\s+\w+)?(?:\s+(\d{2}):(\d{2}))?>", value) if not match: return None year, month, day, hour, minute = match.groups() return datetime( int(year), int(month), int(day), int(hour or 12), int(minute or 0), ) def html_escape(value: str) -> str: return ( value.replace("&", "&") .replace("<", "<") .replace(">", ">") .replace('"', """) ) @dataclass class OrgPage: path: str page_type: str title: str slug: str tags: list[str] content: str date: str comments: bool options: str wip: str | None = None class BuildState: def __init__(self) -> None: self._lock = threading.Lock() self.running = False self.started_at: float | None = None self.finished_at: float | None = None self.ok: bool | None = None self.message = "No build has run yet." self.log = "" def snapshot(self) -> dict[str, Any]: with self._lock: return { "running": self.running, "startedAt": self.started_at, "finishedAt": self.finished_at, "ok": self.ok, "message": self.message, "log": self.log[-12000:], } def start(self) -> bool: with self._lock: if self.running: return False self.running = True self.started_at = time.time() self.finished_at = None self.ok = None self.message = "Build started. The editor is locked until publishing finishes." self.log = "" return True def finish(self, ok: bool, message: str, log: str) -> None: with self._lock: self.running = False self.finished_at = time.time() self.ok = ok self.message = message self.log = log BUILD_STATE = BuildState() 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.suffix != ".org": raise ValueError("Only .org files can be edited.") if not (full.is_relative_to(BLOGS_DIR) or full.is_relative_to(POSTS_DIR)): raise ValueError("Only blogs/ and posts/ files can be edited here.") return full def read_page(path: Path) -> OrgPage: 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() page_type = "blog" if path.is_relative_to(BLOGS_DIR) else "post" slug = meta.get("SLUG") or path.stem tags = normalise_tags(meta.get("FILETAGS", "")) return OrgPage( 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"), wip=meta.get("WIP"), ) def page_to_dict(page: OrgPage) -> 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, "wip": page.wip or "", } def list_pages() -> list[dict[str, Any]]: pages = [] for base, page_type in ((BLOGS_DIR, "blog"), (POSTS_DIR, "post")): if not base.exists(): continue for path in sorted(base.rglob("*.org")): if path.name in GENERATED_ORG_NAMES or "sync-conflict" in path.name: continue try: page = read_page(path) except UnicodeDecodeError: continue parsed = parse_org_datetime(page.date) pages.append( { "path": page.path, "pageType": page_type, "title": page.title, "slug": page.slug, "tags": page.tags, "date": page.date, "timestamp": parsed.timestamp() if parsed else path.stat().st_mtime, } ) 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") 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" raise ValueError("pageType must be blog or post.") def render_org(data: dict[str, Any], previous: OrgPage | 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.") target.write_text(render_org(data, previous), encoding="utf-8") return page_to_dict(read_page(target)) def run_build() -> None: 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 = [] ok = True for command in commands: combined.append(f"$ {' '.join(command)}\n") proc = subprocess.run( command, cwd=ROOT, text=True, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, ) combined.append(proc.stdout) if proc.returncode != 0: ok = False combined.append(f"\nCommand exited with {proc.returncode}.\n") break message = "Build complete. The site output and search index were regenerated." if ok else "Build failed. Check the log below." BUILD_STATE.finish(ok, message, "".join(combined)) def trigger_build() -> None: if not BUILD_STATE.start(): return thread = threading.Thread(target=run_build, daemon=True) thread.start() APP_HTML = r""" Org Site Authoring

New page

Build log


    

  

  


"""


class Handler(BaseHTTPRequestHandler):
    server_version = "OrgAuthoring/1.0"

    def log_message(self, fmt: str, *args: Any) -> None:
        sys.stderr.write("%s - %s\n" % (formatdate(time.time()), fmt % args))

    def send_json(self, data: Any, status: HTTPStatus = HTTPStatus.OK) -> None:
        body = json.dumps(data).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(body)))
        self.send_header("Cache-Control", "no-store")
        self.end_headers()
        self.wfile.write(body)

    def do_GET(self) -> None:
        parsed = urlparse(self.path)
        if parsed.path == "/":
            body = APP_HTML.encode("utf-8")
            self.send_response(HTTPStatus.OK)
            self.send_header("Content-Type", "text/html; charset=utf-8")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)
            return
        if parsed.path == "/api/pages":
            self.send_json(list_pages())
            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_STATE.snapshot())
            return
        self.send_error(HTTPStatus.NOT_FOUND)

    def do_POST(self) -> None:
        if self.path != "/api/page":
            self.send_error(HTTPStatus.NOT_FOUND)
            return
        if BUILD_STATE.snapshot()["running"]:
            self.send_json({"error": "A build is already running. Wait for it to finish before editing."}, HTTPStatus.CONFLICT)
            return
        try:
            length = int(self.headers.get("Content-Length", "0"))
            data = json.loads(self.rfile.read(length).decode("utf-8"))
            saved = save_page(data)
            trigger_build()
            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)
    print(f"Authoring UI running at http://127.0.0.1:{port}")
    print("Press Ctrl-C to stop.")
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass


if __name__ == "__main__":
    main()