diff --git a/README.md b/README.md index b00952e..fce752a 100644 --- a/README.md +++ b/README.md @@ -289,6 +289,53 @@ test -f build/index.html The project currently has no third-party Lektor packages or plugins. +### Local blog editor + +For a focused desktop form that creates and edits records under `content/blog/`, +run: + +```bash +python scripts/blog_editor.py +``` + +The editor uses Tkinter from the Python standard library. It reads the approved +tag choices from `content/tags/`, writes normal `contents.lr` records to the +working tree, and can delete a loaded entry after explicit confirmation. Entry +deletion removes its complete record directory, including attachments, so the +confirmation identifies additional files before proceeding. The editor does +not commit, push, build, or deploy changes. Review saved or deleted records with +the routine content workflow above. + +For a saved entry, **Choose Narration WAV…** converts a selected WAV using +`ffmpeg` from `PATH`, or the repository-provided `support/ffmpeg.exe` when it +is not installed system-wide, and writes the entry-local `narration.mp3` +attachment that the existing narration player recognizes. It uses a fixed +44.1 kHz mono, 96 kbps MP3 preset; replacing or removing narration requires +confirmation. If `ffmpeg` is unavailable or conversion fails, the existing +narration attachment is left unchanged. + +### Local article editor + +For a focused desktop form that creates and edits records under +`content/articles/`, run: + +```bash +python scripts/article_editor.py +``` + +The article editor follows the existing cover convention: a first Markdown +image referencing an article-local `cover-image.png` or `cover-image.jpg`. +Choose a PNG, JPG, or JPEG in the editor and it is copied beside `contents.lr` +with that canonical name. Replacing or removing a cover requires confirmation. +The editor does not commit, push, build, or deploy changes; review all working +tree changes with the routine content workflow above. + +The article editor also has the same saved-entry narration control: it converts +a selected WAV to the article-local `narration.mp3` attachment using `ffmpeg` +from `PATH` or `support/ffmpeg.exe`, with the fixed 44.1 kHz mono, 96 kbps MP3 +preset. Replacement and removal require confirmation; failed conversion leaves +any existing narration unchanged. + ## Production filesystem and permissions ```text diff --git a/scripts/article_editor.py b/scripts/article_editor.py new file mode 100644 index 0000000..62b413e --- /dev/null +++ b/scripts/article_editor.py @@ -0,0 +1,1159 @@ +#!/usr/bin/env python3 +"""Small, article-specific desktop editor for Labyricorn Lektor records.""" + +from __future__ import annotations + +import argparse +import configparser +from dataclasses import dataclass +from datetime import date as current_date +import os +from pathlib import Path +import re +import shutil +import tempfile +import unicodedata +from urllib.parse import urlsplit + +from narration_import import NarrationImportError, import_wav, narration_path, remove_narration + + +SLUG_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") +SEPARATOR_RE = re.compile(r"(?m)(^---[ \t]*(?:\r\n|\n|$))") +FIELD_RE = re.compile(r"([A-Za-z_][A-Za-z0-9_-]*):(.*)") +FIELD_ORDER = ( + "_model", + "title", + "date", + "updated", + "author", + "tags", + "kicker", + "summary", + "external_url", + "published_urls", + "body", +) +EXPECTED_MODEL_FIELDS = { + "title": "string", + "date": "date", + "updated": "date", + "author": "string", + "tags": "checkboxes", + "kicker": "select", + "summary": "text", + "body": "markdown", + "external_url": "url", + "published_urls": "strings", +} +IMAGE_SUFFIXES = {".png", ".jpg", ".jpeg"} +COVER_IMAGE_RE = re.compile( + r"\A(?P\s*)!\[(?P[^\]]*)\]\((?Pcover-image\.(?:png|jpe?g))\)(?P\r?\n(?:\r?\n)?)", + re.IGNORECASE, +) + + +class ArticleEditorError(Exception): + """An error that can be shown directly to an editor user.""" + + +@dataclass(frozen=True) +class Tag: + slug: str + title: str + + +@dataclass +class LrDocument: + blocks: list[str] + delimiters: list[str] + newline: str + source: str + + @classmethod + def parse(cls, text: str, source: str) -> "LrDocument": + parts = SEPARATOR_RE.split(text) + blocks = [parts[0]] + delimiters: list[str] = [] + for index in range(1, len(parts), 2): + delimiters.append(parts[index]) + blocks.append(parts[index + 1]) + + if not blocks or any(not block.strip() for block in blocks): + raise ArticleEditorError(f"{source}: empty or malformed record block") + newline = "\r\n" if "\r\n" in text else "\n" + document = cls(blocks, delimiters, newline, source) + document._field_indexes() + return document + + def render(self) -> str: + pieces = [self.blocks[0]] + for delimiter, block in zip(self.delimiters, self.blocks[1:]): + pieces.extend((delimiter, block)) + return "".join(pieces) + + def _field_indexes(self) -> dict[str, int]: + indexes: dict[str, int] = {} + for index, block in enumerate(self.blocks): + key, _value = self._parse_block(block) + if key in indexes: + raise ArticleEditorError(f"{self.source}: duplicate field {key!r}") + indexes[key] = index + return indexes + + def _parse_block(self, block: str) -> tuple[str, str]: + line_end = block.find("\n") + if line_end < 0: + first_line = block + remainder = "" + else: + first_line = block[:line_end].rstrip("\r") + remainder = block[line_end + 1 :] + match = FIELD_RE.fullmatch(first_line) + if match is None: + raise ArticleEditorError(f"{self.source}: malformed field header {first_line!r}") + key, inline = match.groups() + if inline.startswith(" "): + inline = inline[1:] + if inline: + value = inline + if remainder.rstrip("\r\n"): + value += "\n" + remainder.rstrip("\r\n") + else: + value = remainder.rstrip("\r\n") + return key, value + + def get(self, key: str, default: str = "") -> str: + index = self._field_indexes().get(key) + if index is None: + return default + return self._parse_block(self.blocks[index])[1] + + def set_field(self, key: str, value: str, style: str = "scalar") -> None: + indexes = self._field_indexes() + current_index = indexes.get(key) + if current_index is not None and self.get(key) == value: + return + rendered = self._render_field(key, value, style) + if current_index is not None: + self.blocks[current_index] = rendered + return + + desired_order = FIELD_ORDER.index(key) + insertion_index = len(self.blocks) + for candidate, index in indexes.items(): + if candidate in FIELD_ORDER and FIELD_ORDER.index(candidate) > desired_order: + insertion_index = min(insertion_index, index) + self.blocks.insert(insertion_index, rendered) + delimiter = f"---{self.newline}" + if insertion_index == 0: + self.delimiters.insert(0, delimiter) + else: + self.delimiters.insert(insertion_index - 1, delimiter) + + def remove_field(self, key: str) -> None: + index = self._field_indexes().get(key) + if index is None: + return + if index == 0: + raise ArticleEditorError(f"{self.source}: cannot remove the first record field") + del self.blocks[index] + del self.delimiters[index - 1] + + def _render_field(self, key: str, value: str, style: str) -> str: + if style == "published_urls": + lines = [line.strip() for line in value.splitlines() if line.strip()] + suffix = self.newline.join(lines) + if suffix: + suffix += self.newline + return f"{key}:{self.newline}{self.newline}{suffix}" + if style == "multiline" or "\n" in value: + suffix = value.rstrip("\r\n") + if suffix: + suffix += self.newline + return f"{key}:{self.newline}{suffix}" + return f"{key}: {value}{self.newline}" + + +@dataclass +class ArticleEntry: + slug: str + path: Path + document: LrDocument + original_bytes: bytes + + @property + def title(self) -> str: + return self.document.get("title") + + @property + def publication_date(self) -> str: + return self.document.get("date") + + +@dataclass(frozen=True) +class CoverImage: + name: str + alt: str + path: Path + + +class ArticleRepository: + def __init__(self, site_root: Path): + self.site_root = site_root.resolve() + self.article_root = (self.site_root / "content" / "articles").resolve() + self.tags_root = (self.site_root / "content" / "tags").resolve() + self._validate_layout() + self.tags = self._load_tags() + + def _validate_layout(self) -> None: + project_files = list(self.site_root.glob("*.lektorproject")) + if len(project_files) != 1: + raise ArticleEditorError( + f"{self.site_root}: expected exactly one .lektorproject file" + ) + if not self.article_root.is_dir() or not (self.article_root / "contents.lr").is_file(): + raise ArticleEditorError(f"missing Lektor article section at {self.article_root}") + if not self.tags_root.is_dir(): + raise ArticleEditorError(f"missing controlled tag records at {self.tags_root}") + + model_path = self.site_root / "models" / "entry.ini" + parser = configparser.ConfigParser(interpolation=None) + try: + with model_path.open("r", encoding="utf-8") as model_file: + parser.read_file(model_file) + except (OSError, UnicodeError, configparser.Error) as exc: + raise ArticleEditorError(f"cannot read article entry model {model_path}: {exc}") from exc + + actual_fields = { + section.removeprefix("fields."): parser.get(section, "type", fallback="") + for section in parser.sections() + if section.startswith("fields.") + } + if actual_fields != EXPECTED_MODEL_FIELDS: + raise ArticleEditorError( + "models/entry.ini no longer matches the fields supported by this " + "article-specific editor" + ) + if parser.get("fields.tags", "source", fallback="") != "site.query('/tags')": + raise ArticleEditorError("the article tag field is not sourced from content/tags") + choices = [ + choice.strip() + for choice in parser.get("fields.kicker", "choices", fallback="").split(",") + ] + if "Article" not in choices: + raise ArticleEditorError("the entry model no longer permits the Article kicker") + + def _read_document(self, path: Path) -> tuple[LrDocument, bytes]: + try: + raw = path.read_bytes() + text = raw.decode("utf-8") + except (OSError, UnicodeError) as exc: + raise ArticleEditorError(f"cannot read {path}: {exc}") from exc + return LrDocument.parse(text, str(path)), raw + + def _load_tags(self) -> list[Tag]: + tags: list[Tag] = [] + for directory in sorted(self.tags_root.iterdir(), key=lambda item: item.name): + if not directory.is_dir(): + continue + slug = directory.name + if not SLUG_RE.fullmatch(slug): + raise ArticleEditorError(f"invalid controlled tag directory name: {slug}") + record_path = directory / "contents.lr" + if not record_path.is_file(): + raise ArticleEditorError(f"controlled tag is missing contents.lr: {directory}") + self._ensure_within(record_path, self.tags_root) + document, _raw = self._read_document(record_path) + if document.get("_model") != "tag" or not document.get("title").strip(): + raise ArticleEditorError(f"malformed controlled tag record: {record_path}") + tags.append(Tag(slug, document.get("title").strip())) + if not tags: + raise ArticleEditorError("content/tags contains no controlled tag records") + return tags + + @staticmethod + def _ensure_within(path: Path, parent: Path) -> Path: + resolved = path.resolve() + if not resolved.is_relative_to(parent.resolve()): + raise ArticleEditorError(f"unsafe path outside {parent}: {path}") + return resolved + + def list_entries(self) -> list[ArticleEntry]: + entries: list[ArticleEntry] = [] + for directory in self.article_root.iterdir(): + if not directory.is_dir(): + continue + record_path = directory / "contents.lr" + if not record_path.is_file(): + continue + entries.append(self.load_entry(directory.name)) + entries.sort(key=lambda entry: (entry.publication_date, entry.title), reverse=True) + return entries + + def load_entry(self, slug: str) -> ArticleEntry: + self._validate_slug(slug) + path = self._ensure_within(self.article_root / slug / "contents.lr", self.article_root) + if not path.is_file(): + raise ArticleEditorError(f"article entry does not exist: {slug}") + document, raw = self._read_document(path) + self._validate_loaded_entry(document, slug) + return ArticleEntry(slug, path, document, raw) + + def _validate_loaded_entry(self, document: LrDocument, slug: str) -> None: + if document.get("_model") != "entry": + raise ArticleEditorError(f"content/articles/{slug} is not an entry record") + if document.get("kicker") != "Article": + raise ArticleEditorError(f"content/articles/{slug} does not use the Article kicker") + self.validate_values(self.values_for(document)) + + @staticmethod + def _validate_slug(slug: str) -> None: + if not SLUG_RE.fullmatch(slug): + raise ArticleEditorError( + "slug must contain lowercase letters, numbers, and single hyphens only" + ) + + @staticmethod + def _validate_iso_date(value: str, label: str, required: bool) -> str: + value = value.strip() + if not value and not required: + return "" + if not value: + raise ArticleEditorError(f"{label} is required") + try: + parsed = current_date.fromisoformat(value) + except ValueError as exc: + raise ArticleEditorError(f"{label} must use YYYY-MM-DD") from exc + if parsed.isoformat() != value: + raise ArticleEditorError(f"{label} must use YYYY-MM-DD") + return value + + @staticmethod + def _validate_url(value: str, label: str) -> str: + value = value.strip() + if not value: + return "" + parsed = urlsplit(value) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise ArticleEditorError(f"{label} must be a complete HTTP or HTTPS URL") + return value + + def validate_values(self, values: dict[str, str]) -> dict[str, str]: + cleaned = {key: value for key, value in values.items()} + for required in ("title", "summary", "body"): + cleaned[required] = cleaned.get(required, "") + if not cleaned[required].strip(): + raise ArticleEditorError(f"{required.replace('_', ' ').title()} is required") + cleaned["title"] = cleaned["title"].strip() + cleaned["author"] = cleaned.get("author", "").strip() + cleaned["date"] = self._validate_iso_date( + cleaned.get("date", ""), "Publication date", True + ) + cleaned["updated"] = self._validate_iso_date( + cleaned.get("updated", ""), "Updated date", False + ) + cleaned["kicker"] = cleaned.get("kicker", "").strip() + if cleaned["kicker"] != "Article": + raise ArticleEditorError("Kicker must be Article") + + approved_tags = {tag.slug for tag in self.tags} + selected_tags = [ + tag.strip() + for tag in re.split(r"[,\n]", cleaned.get("tags", "")) + if tag.strip() + ] + unknown_tags = sorted(set(selected_tags) - approved_tags) + if unknown_tags: + raise ArticleEditorError( + "unapproved tag slug(s): " + ", ".join(unknown_tags) + ) + cleaned["tags"] = ", ".join(dict.fromkeys(selected_tags)) + cleaned["external_url"] = self._validate_url( + cleaned.get("external_url", ""), "External URL" + ) + published = [ + self._validate_url(line, "Each published URL") + for line in cleaned.get("published_urls", "").splitlines() + if line.strip() + ] + cleaned["published_urls"] = "\n".join(published) + cleaned["title"] = cleaned["title"].replace("\r", "").replace("\n", " ") + cleaned["author"] = cleaned["author"].replace("\r", "").replace("\n", " ") + cleaned["summary"] = cleaned["summary"].replace("\r\n", "\n") + cleaned["body"] = cleaned["body"].replace("\r\n", "\n") + return cleaned + + @staticmethod + def values_for(document: LrDocument) -> dict[str, str]: + tags = ", ".join( + part.strip() + for part in re.split(r"[,\n]", document.get("tags")) + if part.strip() + ) + published_urls = "\n".join( + line.strip() + for line in document.get("published_urls").splitlines() + if line.strip() + ) + return { + "title": document.get("title"), + "date": document.get("date"), + "updated": document.get("updated"), + "author": document.get("author"), + "tags": tags, + "kicker": document.get("kicker"), + "summary": document.get("summary"), + "body": document.get("body"), + "external_url": document.get("external_url"), + "published_urls": published_urls, + } + + def create_entry(self, slug: str, values: dict[str, str]) -> ArticleEntry: + self._validate_slug(slug) + values = self.validate_values(values) + directory = self.article_root / slug + destination = directory / "contents.lr" + self._ensure_within(destination, self.article_root) + if directory.exists(): + raise ArticleEditorError(f"content/articles/{slug} already exists") + + document = LrDocument.parse( + "_model: entry\n---\ntitle: placeholder\n", str(destination) + ) + self._apply_values(document, values, creating=True) + payload = document.render().encode("utf-8") + try: + directory.mkdir() + with destination.open("xb") as output: + output.write(payload) + output.flush() + os.fsync(output.fileno()) + except (OSError, FileExistsError) as exc: + try: + if directory.is_dir() and not any(directory.iterdir()): + directory.rmdir() + except OSError: + pass + raise ArticleEditorError(f"cannot create {destination}: {exc}") from exc + return self.load_entry(slug) + + def save_entry(self, entry: ArticleEntry, values: dict[str, str]) -> ArticleEntry: + values = self.validate_values(values) + destination = self._ensure_within(entry.path, self.article_root) + temporary: Path | None = None + try: + current = destination.read_bytes() + except OSError as exc: + raise ArticleEditorError(f"cannot re-read {destination}: {exc}") from exc + if current != entry.original_bytes: + raise ArticleEditorError( + "the entry changed on disk after it was loaded; reload it before saving" + ) + + self._apply_values(entry.document, values, creating=False) + payload = entry.document.render().encode("utf-8") + if payload == current: + return entry + try: + with tempfile.NamedTemporaryFile( + mode="wb", prefix=".contents.lr.", dir=destination.parent, delete=False + ) as output: + temporary = Path(output.name) + output.write(payload) + output.flush() + os.fsync(output.fileno()) + os.chmod(temporary, destination.stat().st_mode) + os.replace(temporary, destination) + except OSError as exc: + try: + if temporary is not None: + temporary.unlink(missing_ok=True) + except OSError: + pass + raise ArticleEditorError(f"cannot save {destination}: {exc}") from exc + return self.load_entry(entry.slug) + + def delete_entry(self, entry: ArticleEntry) -> None: + self._validate_slug(entry.slug) + directory = self._ensure_within(entry.path.parent, self.article_root) + destination = self._ensure_within(entry.path, self.article_root) + if directory.parent != self.article_root or directory.name != entry.slug: + raise ArticleEditorError(f"unsafe article entry directory: {directory}") + if directory.is_symlink() or getattr(directory, "is_junction", lambda: False)(): + raise ArticleEditorError(f"refusing to delete linked directory: {directory}") + try: + current = destination.read_bytes() + except OSError as exc: + raise ArticleEditorError(f"cannot re-read {destination}: {exc}") from exc + if current != entry.original_bytes: + raise ArticleEditorError( + "the entry changed on disk after it was loaded; reload it before deleting" + ) + for child in directory.rglob("*"): + if child.is_symlink() or getattr(child, "is_junction", lambda: False)(): + raise ArticleEditorError( + f"refusing to delete an entry containing a linked path: {child}" + ) + try: + shutil.rmtree(directory) + except OSError as exc: + raise ArticleEditorError(f"cannot delete {directory}: {exc}") from exc + + def narration_file(self, entry: ArticleEntry) -> Path: + return self._ensure_within(narration_path(entry.path.parent), self.article_root) + + def import_narration(self, entry: ArticleEntry, source: Path) -> Path: + directory = self._ensure_within(entry.path.parent, self.article_root) + if not entry.path.is_file(): + raise ArticleEditorError(f"article entry does not exist: {entry.slug}") + try: + return import_wav( + directory, source, fallback_ffmpeg=self.site_root / "support" / "ffmpeg.exe" + ) + except NarrationImportError as exc: + raise ArticleEditorError(str(exc)) from exc + + def remove_narration(self, entry: ArticleEntry) -> bool: + directory = self._ensure_within(entry.path.parent, self.article_root) + try: + return remove_narration(directory) + except NarrationImportError as exc: + raise ArticleEditorError(str(exc)) from exc + + def cover_image(self, entry: ArticleEntry) -> CoverImage | None: + match = COVER_IMAGE_RE.match(entry.document.get("body")) + if match is None: + return None + path = self._ensure_within(entry.path.parent / match["name"], entry.path.parent) + if not path.is_file(): + raise ArticleEditorError( + f"article cover image is referenced but missing: {path.name}" + ) + return CoverImage(match["name"], match["alt"], path) + + @staticmethod + def cover_body(body: str, image_name: str, alt: str) -> str: + match = COVER_IMAGE_RE.match(body) + if match is not None: + return f"![{alt}]({image_name})\n\n" + body[match.end() :] + return f"![{alt}]({image_name})\n\n{body.lstrip()}" + + @staticmethod + def remove_cover_body(body: str) -> str: + match = COVER_IMAGE_RE.match(body) + if match is None: + return body + return body[match.end() :] + + def add_or_replace_cover( + self, entry: ArticleEntry, source: Path, alt: str, replace: bool + ) -> tuple[ArticleEntry, CoverImage]: + source = source.resolve() + if not source.is_file(): + raise ArticleEditorError(f"selected image does not exist: {source}") + suffix = source.suffix.lower() + if suffix not in IMAGE_SUFFIXES: + raise ArticleEditorError("cover image must be a PNG, JPG, or JPEG file") + destination = self._ensure_within( + entry.path.parent / f"cover-image{suffix}", entry.path.parent + ) + current_cover = self.cover_image(entry) + if destination.exists() and not replace: + raise ArticleEditorError( + f"{destination.name} already exists; confirm replacement before overwriting it" + ) + if current_cover is not None and current_cover.path != destination and not replace: + raise ArticleEditorError( + "the article already has a cover image; confirm replacement before changing it" + ) + + temporary: Path | None = None + try: + with source.open("rb") as image_input, tempfile.NamedTemporaryFile( + mode="wb", prefix=".cover-image.", dir=entry.path.parent, delete=False + ) as output: + temporary = Path(output.name) + shutil.copyfileobj(image_input, output) + output.flush() + os.fsync(output.fileno()) + os.replace(temporary, destination) + except OSError as exc: + if temporary is not None: + temporary.unlink(missing_ok=True) + raise ArticleEditorError(f"cannot copy cover image: {exc}") from exc + + values = self.values_for(entry.document) + values["body"] = self.cover_body(values["body"], destination.name, alt.strip()) + try: + saved = self.save_entry(entry, values) + except ArticleEditorError: + raise + if current_cover is not None and current_cover.path != destination: + try: + current_cover.path.unlink() + except OSError as exc: + raise ArticleEditorError( + f"new cover saved but could not remove old cover {current_cover.path.name}: {exc}" + ) from exc + return saved, CoverImage(destination.name, alt.strip(), destination) + + def remove_cover(self, entry: ArticleEntry, remove_file: bool) -> ArticleEntry: + cover = self.cover_image(entry) + if cover is None: + raise ArticleEditorError("the loaded article has no managed cover image") + values = self.values_for(entry.document) + values["body"] = self.remove_cover_body(values["body"]) + saved = self.save_entry(entry, values) + if remove_file: + try: + cover.path.unlink() + except OSError as exc: + raise ArticleEditorError( + f"cover reference removed but could not remove {cover.path.name}: {exc}" + ) from exc + return saved + + @staticmethod + def _apply_values( + document: LrDocument, values: dict[str, str], creating: bool + ) -> None: + existing = ArticleRepository.values_for(document) if not creating else {} + styles = { + "summary": "multiline" if "\n" in values["summary"] else "scalar", + "body": "multiline", + "published_urls": "published_urls", + } + optional = {"updated", "tags", "external_url"} + for key in FIELD_ORDER[1:]: + value = values.get(key, "") + if not creating and existing.get(key, "") == value: + continue + if key in optional and not value: + document.remove_field(key) + continue + if key == "published_urls" or value or creating: + document.set_field(key, value, styles.get(key, "scalar")) + + +def suggest_slug(title: str) -> str: + normalized = unicodedata.normalize("NFKD", title).encode("ascii", "ignore").decode() + return re.sub(r"[^a-z0-9]+", "-", normalized.lower()).strip("-") + + +def run_gui(repository: ArticleRepository) -> None: + try: + import tkinter as tk + from tkinter import filedialog, messagebox, ttk + except ImportError as exc: + raise ArticleEditorError( + "Tkinter is not available in this Python installation" + ) from exc + + class ArticleEditorApp: + def __init__(self) -> None: + self.root = tk.Tk() + self.root.title("Labyricorn Article Editor") + self.root.geometry("1240x860") + self.root.minsize(980, 680) + self.current_entry: ArticleEntry | None = None + self.mode = "new" + self.snapshot: dict[str, str] | None = None + + pane = ttk.Panedwindow(self.root, orient=tk.HORIZONTAL) + pane.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) + browser = ttk.Frame(pane, padding=5) + form = ttk.Frame(pane, padding=5) + pane.add(browser, weight=1) + pane.add(form, weight=3) + + columns = ("date", "title", "slug") + self.tree = ttk.Treeview(browser, columns=columns, show="headings") + self.tree.heading("date", text="Date") + self.tree.heading("title", text="Title") + self.tree.heading("slug", text="Slug") + self.tree.column("date", width=90, stretch=False) + self.tree.column("title", width=210) + self.tree.column("slug", width=180) + tree_scroll = ttk.Scrollbar(browser, orient=tk.VERTICAL, command=self.tree.yview) + self.tree.configure(yscrollcommand=tree_scroll.set) + self.tree.grid(row=0, column=0, columnspan=3, sticky="nsew") + tree_scroll.grid(row=0, column=3, sticky="ns") + browser.rowconfigure(0, weight=1) + browser.columnconfigure(0, weight=1) + ttk.Button(browser, text="New", command=self.new_entry).grid( + row=1, column=0, sticky="ew", pady=(8, 0) + ) + ttk.Button(browser, text="Edit selected", command=self.edit_selected).grid( + row=1, column=1, sticky="ew", padx=5, pady=(8, 0) + ) + ttk.Button(browser, text="Refresh", command=self.refresh).grid( + row=1, column=2, sticky="ew", pady=(8, 0) + ) + self.tree.bind("", lambda _event: self.edit_selected()) + + self.title_var = tk.StringVar() + self.slug_var = tk.StringVar() + self.date_var = tk.StringVar() + self.updated_var = tk.StringVar() + self.author_var = tk.StringVar() + self.kicker_var = tk.StringVar(value="Article") + self.external_url_var = tk.StringVar() + self.status_var = tk.StringVar() + self.cover_var = tk.StringVar(value="No cover image") + self.pending_cover: Path | None = None + self.remove_cover_requested = False + + form.columnconfigure(1, weight=1) + form.columnconfigure(3, weight=1) + form.rowconfigure(10, weight=1) + ttk.Label(form, text="Title").grid(row=0, column=0, sticky="w") + ttk.Entry(form, textvariable=self.title_var).grid( + row=0, column=1, columnspan=3, sticky="ew", pady=3 + ) + ttk.Label(form, text="Slug").grid(row=1, column=0, sticky="w") + self.slug_entry = ttk.Entry(form, textvariable=self.slug_var) + self.slug_entry.grid(row=1, column=1, columnspan=2, sticky="ew", pady=3) + ttk.Button(form, text="Suggest", command=self.fill_suggested_slug).grid( + row=1, column=3, sticky="e", padx=(5, 0) + ) + ttk.Label(form, text="Publication date").grid(row=2, column=0, sticky="w") + ttk.Entry(form, textvariable=self.date_var, width=16).grid( + row=2, column=1, sticky="ew", pady=3 + ) + ttk.Label(form, text="Updated date").grid( + row=2, column=2, sticky="w", padx=(12, 0) + ) + ttk.Entry(form, textvariable=self.updated_var, width=16).grid( + row=2, column=3, sticky="ew", pady=3 + ) + ttk.Label(form, text="Author").grid(row=3, column=0, sticky="w") + ttk.Entry(form, textvariable=self.author_var).grid( + row=3, column=1, sticky="ew", pady=3 + ) + ttk.Label(form, text="Type").grid(row=3, column=2, sticky="w", padx=(12, 0)) + ttk.Combobox( + form, textvariable=self.kicker_var, values=("Article",), state="readonly" + ).grid(row=3, column=3, sticky="ew", pady=3) + ttk.Label(form, text="External URL").grid(row=4, column=0, sticky="w") + ttk.Entry(form, textvariable=self.external_url_var).grid( + row=4, column=1, columnspan=3, sticky="ew", pady=3 + ) + + ttk.Label(form, text="Cover image").grid(row=5, column=0, sticky="w") + cover_frame = ttk.Frame(form) + cover_frame.grid(row=5, column=1, columnspan=3, sticky="ew", pady=3) + cover_frame.columnconfigure(0, weight=1) + ttk.Label(cover_frame, textvariable=self.cover_var).grid(row=0, column=0, sticky="w") + ttk.Button(cover_frame, text="Choose…", command=self.choose_cover).grid( + row=0, column=1, padx=(8, 0) + ) + self.remove_cover_button = ttk.Button( + cover_frame, text="Remove", command=self.request_cover_removal, state="disabled" + ) + self.remove_cover_button.grid(row=0, column=2, padx=(8, 0)) + ttk.Label(form, text="Narration").grid(row=6, column=0, sticky="w") + narration_frame = ttk.Frame(form) + narration_frame.grid(row=6, column=1, columnspan=3, sticky="ew", pady=3) + narration_frame.columnconfigure(0, weight=1) + self.narration_var = tk.StringVar(value="Narration: None") + ttk.Label(narration_frame, textvariable=self.narration_var).grid(row=0, column=0, sticky="w") + self.choose_narration_button = ttk.Button( + narration_frame, text="Choose Narration WAV…", command=self.choose_narration, state="disabled" + ) + self.choose_narration_button.grid(row=0, column=1, padx=(8, 0)) + self.remove_narration_button = ttk.Button( + narration_frame, text="Remove", command=self.remove_current_narration, state="disabled" + ) + self.remove_narration_button.grid(row=0, column=2, padx=(8, 0)) + ttk.Label(form, text="Summary").grid(row=7, column=0, sticky="nw", pady=(5, 0)) + self.summary_text = tk.Text(form, height=3, wrap="word", undo=True) + self.summary_text.grid(row=7, column=1, columnspan=3, sticky="nsew", pady=3) + ttk.Label(form, text="Published URLs\n(one per line)").grid( + row=8, column=0, sticky="nw", pady=(5, 0) + ) + self.published_text = tk.Text(form, height=3, wrap="none", undo=True) + self.published_text.grid(row=8, column=1, columnspan=3, sticky="nsew", pady=3) + ttk.Label(form, text="Controlled tags").grid( + row=9, column=0, sticky="nw", pady=(5, 0) + ) + tag_frame = ttk.Frame(form) + tag_frame.grid(row=9, column=1, columnspan=3, sticky="nsew", pady=3) + tag_frame.columnconfigure(0, weight=1) + self.tag_list = tk.Listbox( + tag_frame, height=6, selectmode=tk.EXTENDED, exportselection=False + ) + tag_scroll = ttk.Scrollbar(tag_frame, orient=tk.VERTICAL, command=self.tag_list.yview) + self.tag_list.configure(yscrollcommand=tag_scroll.set) + self.tag_list.grid(row=0, column=0, sticky="nsew") + tag_scroll.grid(row=0, column=1, sticky="ns") + for tag in repository.tags: + self.tag_list.insert(tk.END, f"{tag.title} ({tag.slug})") + + ttk.Label(form, text="Body (Markdown)").grid( + row=10, column=0, sticky="nw", pady=(5, 0) + ) + body_frame = ttk.Frame(form) + body_frame.grid(row=10, column=1, columnspan=3, sticky="nsew", pady=3) + body_frame.rowconfigure(0, weight=1) + body_frame.columnconfigure(0, weight=1) + self.body_text = tk.Text(body_frame, wrap="word", undo=True) + body_scroll = ttk.Scrollbar(body_frame, orient=tk.VERTICAL, command=self.body_text.yview) + self.body_text.configure(yscrollcommand=body_scroll.set) + self.body_text.grid(row=0, column=0, sticky="nsew") + body_scroll.grid(row=0, column=1, sticky="ns") + + footer = ttk.Frame(form) + footer.grid(row=11, column=0, columnspan=4, sticky="ew", pady=(8, 0)) + footer.columnconfigure(0, weight=1) + ttk.Label(footer, textvariable=self.status_var).grid(row=0, column=0, sticky="w") + self.delete_button = ttk.Button( + footer, text="Delete entry", command=self.delete_current, state="disabled" + ) + self.delete_button.grid(row=0, column=1, sticky="e", padx=(0, 8)) + ttk.Button(footer, text="Save entry", command=self.save).grid( + row=0, column=2, sticky="e" + ) + + self.root.bind("", lambda _event: self.save()) + self.root.protocol("WM_DELETE_WINDOW", self.close) + self.refresh(preserve_selection=False) + self.new_entry(confirm=False) + + def collect_values(self) -> dict[str, str]: + selected = [repository.tags[index].slug for index in self.tag_list.curselection()] + return { + "title": self.title_var.get(), + "date": self.date_var.get(), + "updated": self.updated_var.get(), + "author": self.author_var.get(), + "tags": ", ".join(selected), + "kicker": self.kicker_var.get(), + "summary": self.summary_text.get("1.0", "end-1c"), + "body": self.body_text.get("1.0", "end-1c"), + "external_url": self.external_url_var.get(), + "published_urls": self.published_text.get("1.0", "end-1c"), + } + + def set_values(self, values: dict[str, str]) -> None: + self.title_var.set(values.get("title", "")) + self.date_var.set(values.get("date", "")) + self.updated_var.set(values.get("updated", "")) + self.author_var.set(values.get("author", "")) + self.kicker_var.set("Article") + self.external_url_var.set(values.get("external_url", "")) + for widget, key in ( + (self.summary_text, "summary"), + (self.published_text, "published_urls"), + (self.body_text, "body"), + ): + widget.delete("1.0", tk.END) + widget.insert("1.0", values.get(key, "")) + selected = { + part.strip() for part in values.get("tags", "").split(",") if part.strip() + } + self.tag_list.selection_clear(0, tk.END) + for index, tag in enumerate(repository.tags): + if tag.slug in selected: + self.tag_list.selection_set(index) + + def has_unsaved_changes(self) -> bool: + return self.snapshot is not None and self.collect_values() != self.snapshot + + def may_discard(self) -> bool: + return not self.has_unsaved_changes() or messagebox.askyesno( + "Discard changes?", "Discard the unsaved changes in the current form?" + ) + + def refresh(self, preserve_selection: bool = True) -> None: + selected = self.tree.selection()[0] if preserve_selection and self.tree.selection() else None + self.tree.delete(*self.tree.get_children()) + try: + entries = repository.list_entries() + except ArticleEditorError as exc: + messagebox.showerror("Cannot load article entries", str(exc)) + return + for entry in entries: + self.tree.insert( + "", tk.END, iid=entry.slug, values=(entry.publication_date, entry.title, entry.slug) + ) + if selected and self.tree.exists(selected): + self.tree.selection_set(selected) + self.status_var.set(f"{len(entries)} article entries") + + def new_entry(self, confirm: bool = True) -> None: + if confirm and not self.may_discard(): + return + self.mode = "new" + self.current_entry = None + self.delete_button.configure(state="disabled") + self.slug_entry.configure(state="normal") + self.slug_var.set("") + self.set_values( + { + "date": current_date.today().isoformat(), + "author": "Christopher Chambers", + "kicker": "Article", + } + ) + self.snapshot = self.collect_values() + self.status_var.set("Creating a new article entry") + self.pending_cover = None + self.remove_cover_requested = False + self.cover_var.set("No cover image") + self.remove_cover_button.configure(state="disabled") + self.narration_var.set("Narration: save this entry before adding narration") + self.choose_narration_button.configure(state="disabled") + self.remove_narration_button.configure(state="disabled") + + def edit_selected(self) -> None: + selection = self.tree.selection() + if not selection: + messagebox.showinfo("Select an entry", "Select a article entry to edit.") + return + if not self.may_discard(): + return + try: + entry = repository.load_entry(selection[0]) + except ArticleEditorError as exc: + messagebox.showerror("Cannot load entry", str(exc)) + return + self.mode = "edit" + self.current_entry = entry + self.delete_button.configure(state="normal") + self.slug_var.set(entry.slug) + self.slug_entry.configure(state="readonly") + self.set_values(repository.values_for(entry.document)) + self.pending_cover = None + self.remove_cover_requested = False + try: + cover = repository.cover_image(entry) + except ArticleEditorError as exc: + messagebox.showerror("Cannot load cover image", str(exc)) + return + if cover is None: + self.cover_var.set("No cover image") + self.remove_cover_button.configure(state="disabled") + else: + self.cover_var.set(f"{cover.name} — {cover.alt or 'no alt text'}") + self.remove_cover_button.configure(state="normal") + self.update_narration_status() + self.snapshot = self.collect_values() + self.status_var.set(f"Editing content/articles/{entry.slug}/contents.lr") + + def fill_suggested_slug(self) -> None: + if self.mode != "new": + return + self.slug_var.set(suggest_slug(self.title_var.get())) + + def update_narration_status(self) -> None: + if self.current_entry is None: + self.narration_var.set("Narration: None") + self.choose_narration_button.configure(state="disabled") + self.remove_narration_button.configure(state="disabled") + return + narration = repository.narration_file(self.current_entry) + exists = narration.is_file() + self.narration_var.set("Narration: narration.mp3" if exists else "Narration: None") + self.choose_narration_button.configure(state="normal") + self.remove_narration_button.configure(state="normal" if exists else "disabled") + + def choose_narration(self) -> None: + if self.current_entry is None: + return + selected = filedialog.askopenfilename( + title="Choose narration WAV file", + filetypes=[("WAV audio", "*.wav"), ("All files", "*.*")], + ) + if not selected: + return + destination = repository.narration_file(self.current_entry) + if destination.is_file() and not messagebox.askyesno( + "Replace narration?", + "Replace the existing narration.mp3 for this article entry?", + icon="warning", + ): + return + try: + repository.import_narration(self.current_entry, Path(selected)) + except ArticleEditorError as exc: + messagebox.showerror("Cannot import narration", str(exc)) + return + self.update_narration_status() + self.status_var.set(f"Imported narration.mp3 for {self.current_entry.slug}") + + def remove_current_narration(self) -> None: + if self.current_entry is None: + return + if not messagebox.askyesno( + "Remove narration?", + "Remove narration.mp3 from this article entry?", + icon="warning", + ): + return + try: + repository.remove_narration(self.current_entry) + except ArticleEditorError as exc: + messagebox.showerror("Cannot remove narration", str(exc)) + return + self.update_narration_status() + self.status_var.set(f"Removed narration.mp3 from {self.current_entry.slug}") + + def choose_cover(self) -> None: + selected = filedialog.askopenfilename( + title="Choose article cover image", + filetypes=[("Images", "*.png *.jpg *.jpeg"), ("All files", "*.*")], + ) + if not selected: + return + source = Path(selected) + if source.suffix.lower() not in IMAGE_SUFFIXES: + messagebox.showerror("Unsupported cover image", "Choose a PNG, JPG, or JPEG file.") + return + has_cover = self.current_entry is not None and repository.cover_image(self.current_entry) + if has_cover or self.pending_cover is not None: + if not messagebox.askyesno( + "Replace cover image?", + "Replacing the cover will overwrite the existing canonical cover file " + "when saved. Continue?", + icon="warning", + ): + return + self.pending_cover = source + self.remove_cover_requested = False + self.cover_var.set(f"Will copy {source.name} when saved") + self.remove_cover_button.configure(state="normal") + + def request_cover_removal(self) -> None: + if self.pending_cover is not None: + self.pending_cover = None + self.cover_var.set("No cover image") + self.remove_cover_button.configure(state="disabled") + return + if self.current_entry is None: + return + try: + cover = repository.cover_image(self.current_entry) + except ArticleEditorError as exc: + messagebox.showerror("Cannot inspect cover image", str(exc)) + return + if cover is None: + return + if not messagebox.askyesno( + "Remove cover image?", + f"Remove the {cover.name} reference and delete that article-local file?", + icon="warning", + ): + return + self.remove_cover_requested = True + self.cover_var.set(f"Will remove {cover.name} when saved") + + def save(self) -> None: + try: + values = self.collect_values() + if self.mode == "new": + entry = repository.create_entry(self.slug_var.get().strip(), values) + else: + if self.current_entry is None: + raise ArticleEditorError("no article entry is loaded") + entry = repository.save_entry(self.current_entry, values) + if self.pending_cover is not None: + entry, _cover = repository.add_or_replace_cover( + entry, self.pending_cover, "Article cover image", replace=True + ) + elif self.remove_cover_requested: + entry = repository.remove_cover(entry, remove_file=True) + except ArticleEditorError as exc: + messagebox.showerror("Cannot save entry", str(exc)) + return + self.mode = "edit" + self.current_entry = entry + self.delete_button.configure(state="normal") + self.slug_var.set(entry.slug) + self.slug_entry.configure(state="readonly") + self.set_values(repository.values_for(entry.document)) + self.pending_cover = None + self.remove_cover_requested = False + cover = repository.cover_image(entry) + if cover is None: + self.cover_var.set("No cover image") + self.remove_cover_button.configure(state="disabled") + else: + self.cover_var.set(f"{cover.name} — {cover.alt or 'no alt text'}") + self.remove_cover_button.configure(state="normal") + self.update_narration_status() + self.snapshot = self.collect_values() + self.refresh(preserve_selection=False) + self.tree.selection_set(entry.slug) + self.tree.see(entry.slug) + self.status_var.set(f"Saved content/articles/{entry.slug}/contents.lr") + + def delete_current(self) -> None: + if self.mode != "edit" or self.current_entry is None: + return + entry = self.current_entry + try: + additional_files = sum( + 1 + for path in entry.path.parent.rglob("*") + if path.is_file() and path != entry.path + ) + except OSError as exc: + messagebox.showerror("Cannot inspect entry", str(exc)) + return + attachment_note = ( + f" It also contains {additional_files} additional file(s)." + if additional_files + else "" + ) + unsaved_note = ( + " Unsaved form changes will be discarded." if self.has_unsaved_changes() else "" + ) + confirmed = messagebox.askyesno( + "Delete article entry?", + f"Delete {entry.title!r} ({entry.slug}) from the working tree?" + f"{attachment_note}{unsaved_note}\n\n" + "This removes the complete entry directory. The deletion can be reviewed " + "and recovered with Git until it is committed.", + icon="warning", + ) + if not confirmed: + return + try: + repository.delete_entry(entry) + except ArticleEditorError as exc: + messagebox.showerror("Cannot delete entry", str(exc)) + return + deleted_slug = entry.slug + self.snapshot = None + self.refresh(preserve_selection=False) + self.new_entry(confirm=False) + self.status_var.set(f"Deleted content/articles/{deleted_slug} from the working tree") + + def close(self) -> None: + if self.may_discard(): + self.root.destroy() + + try: + app = ArticleEditorApp() + except tk.TclError as exc: + raise ArticleEditorError(f"cannot open the Tkinter window: {exc}") from exc + app.root.mainloop() + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--site-root", type=Path, default=Path(__file__).resolve().parents[1] + ) + args = parser.parse_args() + try: + repository = ArticleRepository(args.site_root) + run_gui(repository) + except ArticleEditorError as exc: + parser.exit(1, f"article-editor: ERROR: {exc}\n") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/blog_editor.py b/scripts/blog_editor.py new file mode 100644 index 0000000..721d31a --- /dev/null +++ b/scripts/blog_editor.py @@ -0,0 +1,959 @@ +#!/usr/bin/env python3 +"""Small, blog-specific desktop editor for Labyricorn Lektor records.""" + +from __future__ import annotations + +import argparse +import configparser +from dataclasses import dataclass +from datetime import date as current_date +import os +from pathlib import Path +import re +import shutil +import tempfile +import unicodedata +from urllib.parse import urlsplit + +from narration_import import NarrationImportError, import_wav, narration_path, remove_narration + + +SLUG_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$") +SEPARATOR_RE = re.compile(r"(?m)(^---[ \t]*(?:\r\n|\n|$))") +FIELD_RE = re.compile(r"([A-Za-z_][A-Za-z0-9_-]*):(.*)") +FIELD_ORDER = ( + "_model", + "title", + "date", + "updated", + "author", + "tags", + "kicker", + "summary", + "external_url", + "published_urls", + "body", +) +EXPECTED_MODEL_FIELDS = { + "title": "string", + "date": "date", + "updated": "date", + "author": "string", + "tags": "checkboxes", + "kicker": "select", + "summary": "text", + "body": "markdown", + "external_url": "url", + "published_urls": "strings", +} + + +class BlogEditorError(Exception): + """An error that can be shown directly to an editor user.""" + + +@dataclass(frozen=True) +class Tag: + slug: str + title: str + + +@dataclass +class LrDocument: + blocks: list[str] + delimiters: list[str] + newline: str + source: str + + @classmethod + def parse(cls, text: str, source: str) -> "LrDocument": + parts = SEPARATOR_RE.split(text) + blocks = [parts[0]] + delimiters: list[str] = [] + for index in range(1, len(parts), 2): + delimiters.append(parts[index]) + blocks.append(parts[index + 1]) + + if not blocks or any(not block.strip() for block in blocks): + raise BlogEditorError(f"{source}: empty or malformed record block") + newline = "\r\n" if "\r\n" in text else "\n" + document = cls(blocks, delimiters, newline, source) + document._field_indexes() + return document + + def render(self) -> str: + pieces = [self.blocks[0]] + for delimiter, block in zip(self.delimiters, self.blocks[1:]): + pieces.extend((delimiter, block)) + return "".join(pieces) + + def _field_indexes(self) -> dict[str, int]: + indexes: dict[str, int] = {} + for index, block in enumerate(self.blocks): + key, _value = self._parse_block(block) + if key in indexes: + raise BlogEditorError(f"{self.source}: duplicate field {key!r}") + indexes[key] = index + return indexes + + def _parse_block(self, block: str) -> tuple[str, str]: + line_end = block.find("\n") + if line_end < 0: + first_line = block + remainder = "" + else: + first_line = block[:line_end].rstrip("\r") + remainder = block[line_end + 1 :] + match = FIELD_RE.fullmatch(first_line) + if match is None: + raise BlogEditorError(f"{self.source}: malformed field header {first_line!r}") + key, inline = match.groups() + if inline.startswith(" "): + inline = inline[1:] + if inline: + value = inline + if remainder.rstrip("\r\n"): + value += "\n" + remainder.rstrip("\r\n") + else: + value = remainder.rstrip("\r\n") + return key, value + + def get(self, key: str, default: str = "") -> str: + index = self._field_indexes().get(key) + if index is None: + return default + return self._parse_block(self.blocks[index])[1] + + def set_field(self, key: str, value: str, style: str = "scalar") -> None: + indexes = self._field_indexes() + current_index = indexes.get(key) + if current_index is not None and self.get(key) == value: + return + rendered = self._render_field(key, value, style) + if current_index is not None: + self.blocks[current_index] = rendered + return + + desired_order = FIELD_ORDER.index(key) + insertion_index = len(self.blocks) + for candidate, index in indexes.items(): + if candidate in FIELD_ORDER and FIELD_ORDER.index(candidate) > desired_order: + insertion_index = min(insertion_index, index) + self.blocks.insert(insertion_index, rendered) + delimiter = f"---{self.newline}" + if insertion_index == 0: + self.delimiters.insert(0, delimiter) + else: + self.delimiters.insert(insertion_index - 1, delimiter) + + def remove_field(self, key: str) -> None: + index = self._field_indexes().get(key) + if index is None: + return + if index == 0: + raise BlogEditorError(f"{self.source}: cannot remove the first record field") + del self.blocks[index] + del self.delimiters[index - 1] + + def _render_field(self, key: str, value: str, style: str) -> str: + if style == "published_urls": + lines = [line.strip() for line in value.splitlines() if line.strip()] + suffix = self.newline.join(lines) + if suffix: + suffix += self.newline + return f"{key}:{self.newline}{self.newline}{suffix}" + if style == "multiline" or "\n" in value: + suffix = value.rstrip("\r\n") + if suffix: + suffix += self.newline + return f"{key}:{self.newline}{suffix}" + return f"{key}: {value}{self.newline}" + + +@dataclass +class BlogEntry: + slug: str + path: Path + document: LrDocument + original_bytes: bytes + + @property + def title(self) -> str: + return self.document.get("title") + + @property + def publication_date(self) -> str: + return self.document.get("date") + + +class BlogRepository: + def __init__(self, site_root: Path): + self.site_root = site_root.resolve() + self.blog_root = (self.site_root / "content" / "blog").resolve() + self.tags_root = (self.site_root / "content" / "tags").resolve() + self._validate_layout() + self.tags = self._load_tags() + + def _validate_layout(self) -> None: + project_files = list(self.site_root.glob("*.lektorproject")) + if len(project_files) != 1: + raise BlogEditorError( + f"{self.site_root}: expected exactly one .lektorproject file" + ) + if not self.blog_root.is_dir() or not (self.blog_root / "contents.lr").is_file(): + raise BlogEditorError(f"missing Lektor blog section at {self.blog_root}") + if not self.tags_root.is_dir(): + raise BlogEditorError(f"missing controlled tag records at {self.tags_root}") + + model_path = self.site_root / "models" / "entry.ini" + parser = configparser.ConfigParser(interpolation=None) + try: + with model_path.open("r", encoding="utf-8") as model_file: + parser.read_file(model_file) + except (OSError, UnicodeError, configparser.Error) as exc: + raise BlogEditorError(f"cannot read blog entry model {model_path}: {exc}") from exc + + actual_fields = { + section.removeprefix("fields."): parser.get(section, "type", fallback="") + for section in parser.sections() + if section.startswith("fields.") + } + if actual_fields != EXPECTED_MODEL_FIELDS: + raise BlogEditorError( + "models/entry.ini no longer matches the fields supported by this " + "blog-specific editor" + ) + if parser.get("fields.tags", "source", fallback="") != "site.query('/tags')": + raise BlogEditorError("the blog tag field is not sourced from content/tags") + choices = [ + choice.strip() + for choice in parser.get("fields.kicker", "choices", fallback="").split(",") + ] + if "Blog" not in choices: + raise BlogEditorError("the entry model no longer permits the Blog kicker") + + def _read_document(self, path: Path) -> tuple[LrDocument, bytes]: + try: + raw = path.read_bytes() + text = raw.decode("utf-8") + except (OSError, UnicodeError) as exc: + raise BlogEditorError(f"cannot read {path}: {exc}") from exc + return LrDocument.parse(text, str(path)), raw + + def _load_tags(self) -> list[Tag]: + tags: list[Tag] = [] + for directory in sorted(self.tags_root.iterdir(), key=lambda item: item.name): + if not directory.is_dir(): + continue + slug = directory.name + if not SLUG_RE.fullmatch(slug): + raise BlogEditorError(f"invalid controlled tag directory name: {slug}") + record_path = directory / "contents.lr" + if not record_path.is_file(): + raise BlogEditorError(f"controlled tag is missing contents.lr: {directory}") + self._ensure_within(record_path, self.tags_root) + document, _raw = self._read_document(record_path) + if document.get("_model") != "tag" or not document.get("title").strip(): + raise BlogEditorError(f"malformed controlled tag record: {record_path}") + tags.append(Tag(slug, document.get("title").strip())) + if not tags: + raise BlogEditorError("content/tags contains no controlled tag records") + return tags + + @staticmethod + def _ensure_within(path: Path, parent: Path) -> Path: + resolved = path.resolve() + if not resolved.is_relative_to(parent.resolve()): + raise BlogEditorError(f"unsafe path outside {parent}: {path}") + return resolved + + def list_entries(self) -> list[BlogEntry]: + entries: list[BlogEntry] = [] + for directory in self.blog_root.iterdir(): + if not directory.is_dir(): + continue + record_path = directory / "contents.lr" + if not record_path.is_file(): + continue + entries.append(self.load_entry(directory.name)) + entries.sort(key=lambda entry: (entry.publication_date, entry.title), reverse=True) + return entries + + def load_entry(self, slug: str) -> BlogEntry: + self._validate_slug(slug) + path = self._ensure_within(self.blog_root / slug / "contents.lr", self.blog_root) + if not path.is_file(): + raise BlogEditorError(f"blog entry does not exist: {slug}") + document, raw = self._read_document(path) + self._validate_loaded_entry(document, slug) + return BlogEntry(slug, path, document, raw) + + def _validate_loaded_entry(self, document: LrDocument, slug: str) -> None: + if document.get("_model") != "entry": + raise BlogEditorError(f"content/blog/{slug} is not an entry record") + if document.get("kicker") != "Blog": + raise BlogEditorError(f"content/blog/{slug} does not use the Blog kicker") + self.validate_values(self.values_for(document)) + + @staticmethod + def _validate_slug(slug: str) -> None: + if not SLUG_RE.fullmatch(slug): + raise BlogEditorError( + "slug must contain lowercase letters, numbers, and single hyphens only" + ) + + @staticmethod + def _validate_iso_date(value: str, label: str, required: bool) -> str: + value = value.strip() + if not value and not required: + return "" + if not value: + raise BlogEditorError(f"{label} is required") + try: + parsed = current_date.fromisoformat(value) + except ValueError as exc: + raise BlogEditorError(f"{label} must use YYYY-MM-DD") from exc + if parsed.isoformat() != value: + raise BlogEditorError(f"{label} must use YYYY-MM-DD") + return value + + @staticmethod + def _validate_url(value: str, label: str) -> str: + value = value.strip() + if not value: + return "" + parsed = urlsplit(value) + if parsed.scheme not in {"http", "https"} or not parsed.netloc: + raise BlogEditorError(f"{label} must be a complete HTTP or HTTPS URL") + return value + + def validate_values(self, values: dict[str, str]) -> dict[str, str]: + cleaned = {key: value for key, value in values.items()} + for required in ("title", "author", "summary", "body"): + cleaned[required] = cleaned.get(required, "") + if not cleaned[required].strip(): + raise BlogEditorError(f"{required.replace('_', ' ').title()} is required") + cleaned["title"] = cleaned["title"].strip() + cleaned["author"] = cleaned["author"].strip() + cleaned["date"] = self._validate_iso_date( + cleaned.get("date", ""), "Publication date", True + ) + cleaned["updated"] = self._validate_iso_date( + cleaned.get("updated", ""), "Updated date", False + ) + cleaned["kicker"] = cleaned.get("kicker", "").strip() + if cleaned["kicker"] != "Blog": + raise BlogEditorError("Kicker must be Blog") + + approved_tags = {tag.slug for tag in self.tags} + selected_tags = [ + tag.strip() + for tag in re.split(r"[,\n]", cleaned.get("tags", "")) + if tag.strip() + ] + unknown_tags = sorted(set(selected_tags) - approved_tags) + if unknown_tags: + raise BlogEditorError( + "unapproved tag slug(s): " + ", ".join(unknown_tags) + ) + cleaned["tags"] = ", ".join(dict.fromkeys(selected_tags)) + cleaned["external_url"] = self._validate_url( + cleaned.get("external_url", ""), "External URL" + ) + published = [ + self._validate_url(line, "Each published URL") + for line in cleaned.get("published_urls", "").splitlines() + if line.strip() + ] + cleaned["published_urls"] = "\n".join(published) + cleaned["title"] = cleaned["title"].replace("\r", "").replace("\n", " ") + cleaned["author"] = cleaned["author"].replace("\r", "").replace("\n", " ") + cleaned["summary"] = cleaned["summary"].replace("\r\n", "\n") + cleaned["body"] = cleaned["body"].replace("\r\n", "\n") + return cleaned + + @staticmethod + def values_for(document: LrDocument) -> dict[str, str]: + tags = ", ".join( + part.strip() + for part in re.split(r"[,\n]", document.get("tags")) + if part.strip() + ) + published_urls = "\n".join( + line.strip() + for line in document.get("published_urls").splitlines() + if line.strip() + ) + return { + "title": document.get("title"), + "date": document.get("date"), + "updated": document.get("updated"), + "author": document.get("author"), + "tags": tags, + "kicker": document.get("kicker"), + "summary": document.get("summary"), + "body": document.get("body"), + "external_url": document.get("external_url"), + "published_urls": published_urls, + } + + def create_entry(self, slug: str, values: dict[str, str]) -> BlogEntry: + self._validate_slug(slug) + values = self.validate_values(values) + directory = self.blog_root / slug + destination = directory / "contents.lr" + self._ensure_within(destination, self.blog_root) + if directory.exists(): + raise BlogEditorError(f"content/blog/{slug} already exists") + + document = LrDocument.parse( + "_model: entry\n---\ntitle: placeholder\n", str(destination) + ) + self._apply_values(document, values, creating=True) + payload = document.render().encode("utf-8") + try: + directory.mkdir() + with destination.open("xb") as output: + output.write(payload) + output.flush() + os.fsync(output.fileno()) + except (OSError, FileExistsError) as exc: + try: + if directory.is_dir() and not any(directory.iterdir()): + directory.rmdir() + except OSError: + pass + raise BlogEditorError(f"cannot create {destination}: {exc}") from exc + return self.load_entry(slug) + + def save_entry(self, entry: BlogEntry, values: dict[str, str]) -> BlogEntry: + values = self.validate_values(values) + destination = self._ensure_within(entry.path, self.blog_root) + temporary: Path | None = None + try: + current = destination.read_bytes() + except OSError as exc: + raise BlogEditorError(f"cannot re-read {destination}: {exc}") from exc + if current != entry.original_bytes: + raise BlogEditorError( + "the entry changed on disk after it was loaded; reload it before saving" + ) + + self._apply_values(entry.document, values, creating=False) + payload = entry.document.render().encode("utf-8") + if payload == current: + return entry + try: + with tempfile.NamedTemporaryFile( + mode="wb", prefix=".contents.lr.", dir=destination.parent, delete=False + ) as output: + temporary = Path(output.name) + output.write(payload) + output.flush() + os.fsync(output.fileno()) + os.chmod(temporary, destination.stat().st_mode) + os.replace(temporary, destination) + except OSError as exc: + try: + if temporary is not None: + temporary.unlink(missing_ok=True) + except OSError: + pass + raise BlogEditorError(f"cannot save {destination}: {exc}") from exc + return self.load_entry(entry.slug) + + def delete_entry(self, entry: BlogEntry) -> None: + self._validate_slug(entry.slug) + directory = self._ensure_within(entry.path.parent, self.blog_root) + destination = self._ensure_within(entry.path, self.blog_root) + if directory.parent != self.blog_root or directory.name != entry.slug: + raise BlogEditorError(f"unsafe blog entry directory: {directory}") + if directory.is_symlink() or getattr(directory, "is_junction", lambda: False)(): + raise BlogEditorError(f"refusing to delete linked directory: {directory}") + try: + current = destination.read_bytes() + except OSError as exc: + raise BlogEditorError(f"cannot re-read {destination}: {exc}") from exc + if current != entry.original_bytes: + raise BlogEditorError( + "the entry changed on disk after it was loaded; reload it before deleting" + ) + for child in directory.rglob("*"): + if child.is_symlink() or getattr(child, "is_junction", lambda: False)(): + raise BlogEditorError( + f"refusing to delete an entry containing a linked path: {child}" + ) + try: + shutil.rmtree(directory) + except OSError as exc: + raise BlogEditorError(f"cannot delete {directory}: {exc}") from exc + + def narration_file(self, entry: BlogEntry) -> Path: + return self._ensure_within(narration_path(entry.path.parent), self.blog_root) + + def import_narration(self, entry: BlogEntry, source: Path) -> Path: + directory = self._ensure_within(entry.path.parent, self.blog_root) + if not entry.path.is_file(): + raise BlogEditorError(f"blog entry does not exist: {entry.slug}") + try: + return import_wav( + directory, source, fallback_ffmpeg=self.site_root / "support" / "ffmpeg.exe" + ) + except NarrationImportError as exc: + raise BlogEditorError(str(exc)) from exc + + def remove_narration(self, entry: BlogEntry) -> bool: + directory = self._ensure_within(entry.path.parent, self.blog_root) + try: + return remove_narration(directory) + except NarrationImportError as exc: + raise BlogEditorError(str(exc)) from exc + + @staticmethod + def _apply_values( + document: LrDocument, values: dict[str, str], creating: bool + ) -> None: + existing = BlogRepository.values_for(document) if not creating else {} + styles = { + "summary": "multiline" if "\n" in values["summary"] else "scalar", + "body": "multiline", + "published_urls": "published_urls", + } + optional = {"updated", "tags", "external_url"} + for key in FIELD_ORDER[1:]: + value = values.get(key, "") + if not creating and existing.get(key, "") == value: + continue + if key in optional and not value: + document.remove_field(key) + continue + if key == "published_urls" or value or creating: + document.set_field(key, value, styles.get(key, "scalar")) + + +def suggest_slug(title: str) -> str: + normalized = unicodedata.normalize("NFKD", title).encode("ascii", "ignore").decode() + return re.sub(r"[^a-z0-9]+", "-", normalized.lower()).strip("-") + + +def run_gui(repository: BlogRepository) -> None: + try: + import tkinter as tk + from tkinter import filedialog, messagebox, ttk + except ImportError as exc: + raise BlogEditorError( + "Tkinter is not available in this Python installation" + ) from exc + + class BlogEditorApp: + def __init__(self) -> None: + self.root = tk.Tk() + self.root.title("Labyricorn Blog Editor") + self.root.geometry("1240x860") + self.root.minsize(980, 680) + self.current_entry: BlogEntry | None = None + self.mode = "new" + self.snapshot: dict[str, str] | None = None + + pane = ttk.Panedwindow(self.root, orient=tk.HORIZONTAL) + pane.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) + browser = ttk.Frame(pane, padding=5) + form = ttk.Frame(pane, padding=5) + pane.add(browser, weight=1) + pane.add(form, weight=3) + + columns = ("date", "title", "slug") + self.tree = ttk.Treeview(browser, columns=columns, show="headings") + self.tree.heading("date", text="Date") + self.tree.heading("title", text="Title") + self.tree.heading("slug", text="Slug") + self.tree.column("date", width=90, stretch=False) + self.tree.column("title", width=210) + self.tree.column("slug", width=180) + tree_scroll = ttk.Scrollbar(browser, orient=tk.VERTICAL, command=self.tree.yview) + self.tree.configure(yscrollcommand=tree_scroll.set) + self.tree.grid(row=0, column=0, columnspan=3, sticky="nsew") + tree_scroll.grid(row=0, column=3, sticky="ns") + browser.rowconfigure(0, weight=1) + browser.columnconfigure(0, weight=1) + ttk.Button(browser, text="New", command=self.new_entry).grid( + row=1, column=0, sticky="ew", pady=(8, 0) + ) + ttk.Button(browser, text="Edit selected", command=self.edit_selected).grid( + row=1, column=1, sticky="ew", padx=5, pady=(8, 0) + ) + ttk.Button(browser, text="Refresh", command=self.refresh).grid( + row=1, column=2, sticky="ew", pady=(8, 0) + ) + self.tree.bind("", lambda _event: self.edit_selected()) + + self.title_var = tk.StringVar() + self.slug_var = tk.StringVar() + self.date_var = tk.StringVar() + self.updated_var = tk.StringVar() + self.author_var = tk.StringVar() + self.kicker_var = tk.StringVar(value="Blog") + self.external_url_var = tk.StringVar() + self.status_var = tk.StringVar() + + form.columnconfigure(1, weight=1) + form.columnconfigure(3, weight=1) + form.rowconfigure(9, weight=1) + ttk.Label(form, text="Title").grid(row=0, column=0, sticky="w") + ttk.Entry(form, textvariable=self.title_var).grid( + row=0, column=1, columnspan=3, sticky="ew", pady=3 + ) + ttk.Label(form, text="Slug").grid(row=1, column=0, sticky="w") + self.slug_entry = ttk.Entry(form, textvariable=self.slug_var) + self.slug_entry.grid(row=1, column=1, columnspan=2, sticky="ew", pady=3) + ttk.Button(form, text="Suggest", command=self.fill_suggested_slug).grid( + row=1, column=3, sticky="e", padx=(5, 0) + ) + ttk.Label(form, text="Publication date").grid(row=2, column=0, sticky="w") + ttk.Entry(form, textvariable=self.date_var, width=16).grid( + row=2, column=1, sticky="ew", pady=3 + ) + ttk.Label(form, text="Updated date").grid( + row=2, column=2, sticky="w", padx=(12, 0) + ) + ttk.Entry(form, textvariable=self.updated_var, width=16).grid( + row=2, column=3, sticky="ew", pady=3 + ) + ttk.Label(form, text="Author").grid(row=3, column=0, sticky="w") + ttk.Entry(form, textvariable=self.author_var).grid( + row=3, column=1, sticky="ew", pady=3 + ) + ttk.Label(form, text="Type").grid(row=3, column=2, sticky="w", padx=(12, 0)) + ttk.Combobox( + form, textvariable=self.kicker_var, values=("Blog",), state="readonly" + ).grid(row=3, column=3, sticky="ew", pady=3) + ttk.Label(form, text="External URL").grid(row=4, column=0, sticky="w") + ttk.Entry(form, textvariable=self.external_url_var).grid( + row=4, column=1, columnspan=3, sticky="ew", pady=3 + ) + + ttk.Label(form, text="Narration").grid(row=5, column=0, sticky="w") + narration_frame = ttk.Frame(form) + narration_frame.grid(row=5, column=1, columnspan=3, sticky="ew", pady=3) + narration_frame.columnconfigure(0, weight=1) + self.narration_var = tk.StringVar(value="Narration: None") + ttk.Label(narration_frame, textvariable=self.narration_var).grid(row=0, column=0, sticky="w") + self.choose_narration_button = ttk.Button( + narration_frame, text="Choose Narration WAV…", command=self.choose_narration, state="disabled" + ) + self.choose_narration_button.grid(row=0, column=1, padx=(8, 0)) + self.remove_narration_button = ttk.Button( + narration_frame, text="Remove", command=self.remove_current_narration, state="disabled" + ) + self.remove_narration_button.grid(row=0, column=2, padx=(8, 0)) + + ttk.Label(form, text="Summary").grid(row=6, column=0, sticky="nw", pady=(5, 0)) + self.summary_text = tk.Text(form, height=3, wrap="word", undo=True) + self.summary_text.grid(row=6, column=1, columnspan=3, sticky="nsew", pady=3) + ttk.Label(form, text="Published URLs\n(one per line)").grid( + row=7, column=0, sticky="nw", pady=(5, 0) + ) + self.published_text = tk.Text(form, height=3, wrap="none", undo=True) + self.published_text.grid(row=7, column=1, columnspan=3, sticky="nsew", pady=3) + ttk.Label(form, text="Controlled tags").grid( + row=8, column=0, sticky="nw", pady=(5, 0) + ) + tag_frame = ttk.Frame(form) + tag_frame.grid(row=8, column=1, columnspan=3, sticky="nsew", pady=3) + tag_frame.columnconfigure(0, weight=1) + self.tag_list = tk.Listbox( + tag_frame, height=6, selectmode=tk.EXTENDED, exportselection=False + ) + tag_scroll = ttk.Scrollbar(tag_frame, orient=tk.VERTICAL, command=self.tag_list.yview) + self.tag_list.configure(yscrollcommand=tag_scroll.set) + self.tag_list.grid(row=0, column=0, sticky="nsew") + tag_scroll.grid(row=0, column=1, sticky="ns") + for tag in repository.tags: + self.tag_list.insert(tk.END, f"{tag.title} ({tag.slug})") + + ttk.Label(form, text="Body (Markdown)").grid( + row=9, column=0, sticky="nw", pady=(5, 0) + ) + body_frame = ttk.Frame(form) + body_frame.grid(row=9, column=1, columnspan=3, sticky="nsew", pady=3) + body_frame.rowconfigure(0, weight=1) + body_frame.columnconfigure(0, weight=1) + self.body_text = tk.Text(body_frame, wrap="word", undo=True) + body_scroll = ttk.Scrollbar(body_frame, orient=tk.VERTICAL, command=self.body_text.yview) + self.body_text.configure(yscrollcommand=body_scroll.set) + self.body_text.grid(row=0, column=0, sticky="nsew") + body_scroll.grid(row=0, column=1, sticky="ns") + + footer = ttk.Frame(form) + footer.grid(row=10, column=0, columnspan=4, sticky="ew", pady=(8, 0)) + footer.columnconfigure(0, weight=1) + ttk.Label(footer, textvariable=self.status_var).grid(row=0, column=0, sticky="w") + self.delete_button = ttk.Button( + footer, text="Delete entry", command=self.delete_current, state="disabled" + ) + self.delete_button.grid(row=0, column=1, sticky="e", padx=(0, 8)) + ttk.Button(footer, text="Save entry", command=self.save).grid( + row=0, column=2, sticky="e" + ) + + self.root.bind("", lambda _event: self.save()) + self.root.protocol("WM_DELETE_WINDOW", self.close) + self.refresh(preserve_selection=False) + self.new_entry(confirm=False) + + def collect_values(self) -> dict[str, str]: + selected = [repository.tags[index].slug for index in self.tag_list.curselection()] + return { + "title": self.title_var.get(), + "date": self.date_var.get(), + "updated": self.updated_var.get(), + "author": self.author_var.get(), + "tags": ", ".join(selected), + "kicker": self.kicker_var.get(), + "summary": self.summary_text.get("1.0", "end-1c"), + "body": self.body_text.get("1.0", "end-1c"), + "external_url": self.external_url_var.get(), + "published_urls": self.published_text.get("1.0", "end-1c"), + } + + def set_values(self, values: dict[str, str]) -> None: + self.title_var.set(values.get("title", "")) + self.date_var.set(values.get("date", "")) + self.updated_var.set(values.get("updated", "")) + self.author_var.set(values.get("author", "")) + self.kicker_var.set("Blog") + self.external_url_var.set(values.get("external_url", "")) + for widget, key in ( + (self.summary_text, "summary"), + (self.published_text, "published_urls"), + (self.body_text, "body"), + ): + widget.delete("1.0", tk.END) + widget.insert("1.0", values.get(key, "")) + selected = { + part.strip() for part in values.get("tags", "").split(",") if part.strip() + } + self.tag_list.selection_clear(0, tk.END) + for index, tag in enumerate(repository.tags): + if tag.slug in selected: + self.tag_list.selection_set(index) + + def has_unsaved_changes(self) -> bool: + return self.snapshot is not None and self.collect_values() != self.snapshot + + def may_discard(self) -> bool: + return not self.has_unsaved_changes() or messagebox.askyesno( + "Discard changes?", "Discard the unsaved changes in the current form?" + ) + + def refresh(self, preserve_selection: bool = True) -> None: + selected = self.tree.selection()[0] if preserve_selection and self.tree.selection() else None + self.tree.delete(*self.tree.get_children()) + try: + entries = repository.list_entries() + except BlogEditorError as exc: + messagebox.showerror("Cannot load blog entries", str(exc)) + return + for entry in entries: + self.tree.insert( + "", tk.END, iid=entry.slug, values=(entry.publication_date, entry.title, entry.slug) + ) + if selected and self.tree.exists(selected): + self.tree.selection_set(selected) + self.status_var.set(f"{len(entries)} blog entries") + + def new_entry(self, confirm: bool = True) -> None: + if confirm and not self.may_discard(): + return + self.mode = "new" + self.current_entry = None + self.delete_button.configure(state="disabled") + self.slug_entry.configure(state="normal") + self.slug_var.set("") + self.set_values( + { + "date": current_date.today().isoformat(), + "author": "Christopher Chambers", + "kicker": "Blog", + } + ) + self.snapshot = self.collect_values() + self.status_var.set("Creating a new blog entry") + self.narration_var.set("Narration: save this entry before adding narration") + self.choose_narration_button.configure(state="disabled") + self.remove_narration_button.configure(state="disabled") + + def edit_selected(self) -> None: + selection = self.tree.selection() + if not selection: + messagebox.showinfo("Select an entry", "Select a blog entry to edit.") + return + if not self.may_discard(): + return + try: + entry = repository.load_entry(selection[0]) + except BlogEditorError as exc: + messagebox.showerror("Cannot load entry", str(exc)) + return + self.mode = "edit" + self.current_entry = entry + self.delete_button.configure(state="normal") + self.slug_var.set(entry.slug) + self.slug_entry.configure(state="readonly") + self.set_values(repository.values_for(entry.document)) + self.update_narration_status() + self.snapshot = self.collect_values() + self.status_var.set(f"Editing content/blog/{entry.slug}/contents.lr") + + def fill_suggested_slug(self) -> None: + if self.mode != "new": + return + self.slug_var.set(suggest_slug(self.title_var.get())) + + def update_narration_status(self) -> None: + if self.current_entry is None: + self.narration_var.set("Narration: None") + self.choose_narration_button.configure(state="disabled") + self.remove_narration_button.configure(state="disabled") + return + narration = repository.narration_file(self.current_entry) + exists = narration.is_file() + self.narration_var.set("Narration: narration.mp3" if exists else "Narration: None") + self.choose_narration_button.configure(state="normal") + self.remove_narration_button.configure(state="normal" if exists else "disabled") + + def choose_narration(self) -> None: + if self.current_entry is None: + return + selected = filedialog.askopenfilename( + title="Choose narration WAV file", + filetypes=[("WAV audio", "*.wav"), ("All files", "*.*")], + ) + if not selected: + return + destination = repository.narration_file(self.current_entry) + if destination.is_file() and not messagebox.askyesno( + "Replace narration?", + "Replace the existing narration.mp3 for this blog entry?", + icon="warning", + ): + return + try: + repository.import_narration(self.current_entry, Path(selected)) + except BlogEditorError as exc: + messagebox.showerror("Cannot import narration", str(exc)) + return + self.update_narration_status() + self.status_var.set(f"Imported narration.mp3 for {self.current_entry.slug}") + + def remove_current_narration(self) -> None: + if self.current_entry is None: + return + if not messagebox.askyesno( + "Remove narration?", + "Remove narration.mp3 from this blog entry?", + icon="warning", + ): + return + try: + repository.remove_narration(self.current_entry) + except BlogEditorError as exc: + messagebox.showerror("Cannot remove narration", str(exc)) + return + self.update_narration_status() + self.status_var.set(f"Removed narration.mp3 from {self.current_entry.slug}") + + def save(self) -> None: + try: + values = self.collect_values() + if self.mode == "new": + entry = repository.create_entry(self.slug_var.get().strip(), values) + else: + if self.current_entry is None: + raise BlogEditorError("no blog entry is loaded") + entry = repository.save_entry(self.current_entry, values) + except BlogEditorError as exc: + messagebox.showerror("Cannot save entry", str(exc)) + return + self.mode = "edit" + self.current_entry = entry + self.delete_button.configure(state="normal") + self.slug_var.set(entry.slug) + self.slug_entry.configure(state="readonly") + self.set_values(repository.values_for(entry.document)) + self.update_narration_status() + self.snapshot = self.collect_values() + self.refresh(preserve_selection=False) + self.tree.selection_set(entry.slug) + self.tree.see(entry.slug) + self.status_var.set(f"Saved content/blog/{entry.slug}/contents.lr") + + def delete_current(self) -> None: + if self.mode != "edit" or self.current_entry is None: + return + entry = self.current_entry + try: + additional_files = sum( + 1 + for path in entry.path.parent.rglob("*") + if path.is_file() and path != entry.path + ) + except OSError as exc: + messagebox.showerror("Cannot inspect entry", str(exc)) + return + attachment_note = ( + f" It also contains {additional_files} additional file(s)." + if additional_files + else "" + ) + unsaved_note = ( + " Unsaved form changes will be discarded." if self.has_unsaved_changes() else "" + ) + confirmed = messagebox.askyesno( + "Delete blog entry?", + f"Delete {entry.title!r} ({entry.slug}) from the working tree?" + f"{attachment_note}{unsaved_note}\n\n" + "This removes the complete entry directory. The deletion can be reviewed " + "and recovered with Git until it is committed.", + icon="warning", + ) + if not confirmed: + return + try: + repository.delete_entry(entry) + except BlogEditorError as exc: + messagebox.showerror("Cannot delete entry", str(exc)) + return + deleted_slug = entry.slug + self.snapshot = None + self.refresh(preserve_selection=False) + self.new_entry(confirm=False) + self.status_var.set(f"Deleted content/blog/{deleted_slug} from the working tree") + + def close(self) -> None: + if self.may_discard(): + self.root.destroy() + + try: + app = BlogEditorApp() + except tk.TclError as exc: + raise BlogEditorError(f"cannot open the Tkinter window: {exc}") from exc + app.root.mainloop() + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--site-root", type=Path, default=Path(__file__).resolve().parents[1] + ) + args = parser.parse_args() + try: + repository = BlogRepository(args.site_root) + run_gui(repository) + except BlogEditorError as exc: + parser.exit(1, f"blog-editor: ERROR: {exc}\n") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/narration_import.py b/scripts/narration_import.py new file mode 100644 index 0000000..81e1c58 --- /dev/null +++ b/scripts/narration_import.py @@ -0,0 +1,91 @@ +"""Narrow WAV-to-MP3 narration attachment support for Labyricorn editors.""" + +from __future__ import annotations + +import os +from pathlib import Path +import shutil +import subprocess +import tempfile + + +NARRATION_FILENAME = "narration.mp3" + + +class NarrationImportError(Exception): + """Raised when a narration attachment cannot be safely changed.""" + + +def narration_path(entry_directory: Path) -> Path: + """Return the one supported narration attachment path for an entry.""" + return entry_directory / NARRATION_FILENAME + + +def import_wav( + entry_directory: Path, + source: Path, + *, + ffmpeg: str | Path | None = None, + fallback_ffmpeg: Path | None = None, +) -> Path: + """Convert *source* WAV to an atomically replaced narration attachment.""" + source = source.expanduser() + if source.suffix.lower() != ".wav": + raise NarrationImportError("choose a WAV (.wav) file for narration") + if not source.is_file(): + raise NarrationImportError(f"narration source is not a regular file: {source}") + if not entry_directory.is_dir(): + raise NarrationImportError(f"entry directory does not exist: {entry_directory}") + + executable = str(ffmpeg) if ffmpeg else shutil.which("ffmpeg") + if not executable and fallback_ffmpeg is not None and fallback_ffmpeg.is_file(): + executable = str(fallback_ffmpeg) + if not executable: + raise NarrationImportError( + "ffmpeg is required to convert WAV narration to narration.mp3; add it to PATH " + "or provide support/ffmpeg.exe" + ) + + destination = narration_path(entry_directory) + descriptor, temporary_name = tempfile.mkstemp( + prefix=".narration-", suffix=".mp3", dir=entry_directory + ) + os.close(descriptor) + temporary = Path(temporary_name) + try: + command = [ + executable, "-y", "-i", str(source), "-vn", "-ar", "44100", "-ac", "1", + "-c:a", "libmp3lame", "-b:a", "96k", str(temporary), + ] + run_options: dict[str, object] = {"capture_output": True, "text": True} + if os.name == "nt": + run_options["creationflags"] = subprocess.CREATE_NO_WINDOW + completed = subprocess.run(command, **run_options) + if completed.returncode: + detail = (completed.stderr or completed.stdout or "ffmpeg returned an error").strip() + raise NarrationImportError(f"ffmpeg could not convert the narration WAV: {detail}") + if not temporary.is_file() or temporary.stat().st_size == 0: + raise NarrationImportError("ffmpeg completed without producing a usable MP3 file") + os.replace(temporary, destination) + except OSError as exc: + raise NarrationImportError(f"cannot convert narration WAV: {exc}") from exc + finally: + try: + temporary.unlink(missing_ok=True) + except OSError: + pass + return destination + + +def remove_narration(entry_directory: Path) -> bool: + """Remove only the canonical narration attachment, if it exists.""" + destination = narration_path(entry_directory) + if not destination.exists(): + return False + if not destination.is_file(): + raise NarrationImportError(f"narration attachment is not a regular file: {destination}") + try: + destination.unlink() + except OSError as exc: + raise NarrationImportError(f"cannot remove narration attachment: {exc}") from exc + return True diff --git a/support/ffmpeg.exe b/support/ffmpeg.exe new file mode 100644 index 0000000..2099dce Binary files /dev/null and b/support/ffmpeg.exe differ diff --git a/tests/test_article_editor.py b/tests/test_article_editor.py new file mode 100644 index 0000000..56d114c --- /dev/null +++ b/tests/test_article_editor.py @@ -0,0 +1,187 @@ +from __future__ import annotations + +from pathlib import Path +import shutil +import sys +import unittest +import uuid + + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) + +from article_editor import ArticleEditorError, ArticleRepository, suggest_slug # noqa: E402 + + +class ArticleEditorTests(unittest.TestCase): + def setUp(self) -> None: + cache_root = Path(__file__).resolve().parents[1] / ".cache" + cache_root.mkdir(exist_ok=True) + self.site_root = cache_root / f"article-editor-test-{uuid.uuid4().hex}" + (self.site_root / "content" / "articles").mkdir(parents=True) + (self.site_root / "content" / "tags" / "writing").mkdir(parents=True) + (self.site_root / "models").mkdir() + (self.site_root / "Test.lektorproject").write_text( + "[project]\nname = Test\n", encoding="utf-8" + ) + (self.site_root / "content" / "articles" / "contents.lr").write_text( + "_model: section\n---\ntitle: Articles\n", encoding="utf-8" + ) + (self.site_root / "content" / "tags" / "writing" / "contents.lr").write_text( + "_model: tag\n---\ntitle: Writing\n---\nsummary: Writing.\n", + encoding="utf-8", + ) + source_model = Path(__file__).resolve().parents[1] / "models" / "entry.ini" + shutil.copy2(source_model, self.site_root / "models" / "entry.ini") + self.repository = ArticleRepository(self.site_root) + + def tearDown(self) -> None: + shutil.rmtree(self.site_root) + + @staticmethod + def values(**overrides: str) -> dict[str, str]: + values = { + "title": "Test Entry", + "date": "2026-08-13", + "updated": "", + "author": "Test Author", + "tags": "writing", + "kicker": "Article", + "summary": "A test summary.", + "body": "A **Markdown** body.", + "external_url": "", + "published_urls": "https://example.test/post", + } + values.update(overrides) + return values + + def test_current_site_article_records_and_controlled_tags_load(self) -> None: + repository = ArticleRepository(Path(__file__).resolve().parents[1]) + entries = repository.list_entries() + self.assertGreaterEqual(len(entries), 7) + self.assertGreaterEqual(len(repository.tags), 45) + self.assertTrue(all(entry.document.get("kicker") == "Article" for entry in entries)) + for entry in entries: + with self.subTest(slug=entry.slug): + self.assertEqual( + entry.document.render().encode("utf-8"), entry.original_bytes + ) + + def test_create_uses_existing_record_conventions_and_refuses_collision(self) -> None: + entry = self.repository.create_entry( + "test-entry", self.values(body=" preserved indentation\n\nBody text.") + ) + rendered = entry.path.read_text(encoding="utf-8") + self.assertIn("_model: entry\n---\ntitle: Test Entry\n", rendered) + self.assertIn("tags: writing\n", rendered) + self.assertIn( + "published_urls:\n\nhttps://example.test/post\n---\nbody:\n", + rendered, + ) + with self.assertRaises(ArticleEditorError): + self.repository.create_entry("test-entry", self.values()) + + def test_edit_preserves_unmanaged_and_unchanged_blocks_exactly(self) -> None: + entry = self.repository.create_entry( + "test-entry", self.values(body=" preserved indentation\n\nBody text.") + ) + original = entry.path.read_text(encoding="utf-8") + marker = "---\neditor_note:\nKeep this unknown field exactly. \n" + original = original.replace("---\nbody:\n", marker + "---\nbody:\n") + entry.path.write_text(original, encoding="utf-8") + + loaded = self.repository.load_entry("test-entry") + values = self.repository.values_for(loaded.document) + values["summary"] = "An edited summary." + saved = self.repository.save_entry(loaded, values) + rendered = saved.path.read_text(encoding="utf-8") + + self.assertIn(marker, rendered) + self.assertIn("summary: An edited summary.\n", rendered) + self.assertIn("body:\n preserved indentation\n\nBody text.\n", rendered) + self.assertIn("published_urls:\n\nhttps://example.test/post\n", rendered) + + def test_save_detects_an_external_change(self) -> None: + entry = self.repository.create_entry("test-entry", self.values()) + with entry.path.open("a", encoding="utf-8") as output: + output.write("\n") + with self.assertRaisesRegex(ArticleEditorError, "changed on disk"): + self.repository.save_entry(entry, self.values(summary="Changed")) + + def test_delete_removes_only_the_loaded_entry_directory(self) -> None: + entry = self.repository.create_entry("test-entry", self.values()) + attachment = entry.path.parent / "narration.mp3" + attachment.write_bytes(b"disposable attachment") + neighbor = self.repository.create_entry("neighbor-entry", self.values()) + + self.repository.delete_entry(entry) + + self.assertFalse(entry.path.parent.exists()) + self.assertTrue(neighbor.path.is_file()) + + def test_delete_detects_an_external_change(self) -> None: + entry = self.repository.create_entry("test-entry", self.values()) + with entry.path.open("a", encoding="utf-8") as output: + output.write("\n") + with self.assertRaisesRegex(ArticleEditorError, "changed on disk"): + self.repository.delete_entry(entry) + self.assertTrue(entry.path.is_file()) + + def test_unapproved_tag_is_rejected(self) -> None: + with self.assertRaisesRegex(ArticleEditorError, "unapproved tag"): + self.repository.create_entry("test-entry", self.values(tags="invented")) + + def test_cover_image_is_copied_renamed_referenced_and_reopened(self) -> None: + entry = self.repository.create_entry("test-entry", self.values()) + source = self.site_root / "source.png" + source.write_bytes(b"disposable png bytes") + + saved, cover = self.repository.add_or_replace_cover( + entry, source, "Test cover", replace=False + ) + + self.assertEqual(cover.name, "cover-image.png") + self.assertEqual(cover.path.read_bytes(), source.read_bytes()) + self.assertTrue(saved.document.get("body").startswith("![Test cover](cover-image.png)\n\n")) + reopened = self.repository.load_entry("test-entry") + self.assertEqual(self.repository.cover_image(reopened), cover) + + def test_cover_replacement_requires_confirmation_and_removes_old_cover(self) -> None: + entry = self.repository.create_entry("test-entry", self.values()) + first = self.site_root / "first.png" + second = self.site_root / "second.jpg" + first.write_bytes(b"first") + second.write_bytes(b"second") + saved, _cover = self.repository.add_or_replace_cover( + entry, first, "First", replace=False + ) + with self.assertRaisesRegex(ArticleEditorError, "confirm replacement"): + self.repository.add_or_replace_cover(saved, second, "Second", replace=False) + + replaced, cover = self.repository.add_or_replace_cover( + saved, second, "Second", replace=True + ) + self.assertEqual(cover.name, "cover-image.jpg") + self.assertFalse((replaced.path.parent / "cover-image.png").exists()) + self.assertTrue((replaced.path.parent / "cover-image.jpg").is_file()) + self.assertTrue(replaced.document.get("body").startswith("![Second](cover-image.jpg)\n\n")) + + def test_remove_cover_removes_reference_and_article_local_file(self) -> None: + entry = self.repository.create_entry("test-entry", self.values()) + source = self.site_root / "source.png" + source.write_bytes(b"cover") + saved, cover = self.repository.add_or_replace_cover( + entry, source, "Test cover", replace=False + ) + + without_cover = self.repository.remove_cover(saved, remove_file=True) + + self.assertIsNone(self.repository.cover_image(without_cover)) + self.assertFalse(cover.path.exists()) + self.assertEqual(without_cover.document.get("body"), self.values()["body"]) + + def test_slug_suggestion_matches_repository_style(self) -> None: + self.assertEqual(suggest_slug("We’re Testing: A GUI!"), "were-testing-a-gui") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_blog_editor.py b/tests/test_blog_editor.py new file mode 100644 index 0000000..20d412a --- /dev/null +++ b/tests/test_blog_editor.py @@ -0,0 +1,138 @@ +from __future__ import annotations + +from pathlib import Path +import shutil +import sys +import unittest +import uuid + + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) + +from blog_editor import BlogEditorError, BlogRepository, suggest_slug # noqa: E402 + + +class BlogEditorTests(unittest.TestCase): + def setUp(self) -> None: + cache_root = Path(__file__).resolve().parents[1] / ".cache" + cache_root.mkdir(exist_ok=True) + self.site_root = cache_root / f"blog-editor-test-{uuid.uuid4().hex}" + (self.site_root / "content" / "blog").mkdir(parents=True) + (self.site_root / "content" / "tags" / "writing").mkdir(parents=True) + (self.site_root / "models").mkdir() + (self.site_root / "Test.lektorproject").write_text( + "[project]\nname = Test\n", encoding="utf-8" + ) + (self.site_root / "content" / "blog" / "contents.lr").write_text( + "_model: section\n---\ntitle: Blog\n", encoding="utf-8" + ) + (self.site_root / "content" / "tags" / "writing" / "contents.lr").write_text( + "_model: tag\n---\ntitle: Writing\n---\nsummary: Writing.\n", + encoding="utf-8", + ) + source_model = Path(__file__).resolve().parents[1] / "models" / "entry.ini" + shutil.copy2(source_model, self.site_root / "models" / "entry.ini") + self.repository = BlogRepository(self.site_root) + + def tearDown(self) -> None: + shutil.rmtree(self.site_root) + + @staticmethod + def values(**overrides: str) -> dict[str, str]: + values = { + "title": "Test Entry", + "date": "2026-08-13", + "updated": "", + "author": "Test Author", + "tags": "writing", + "kicker": "Blog", + "summary": "A test summary.", + "body": "A **Markdown** body.", + "external_url": "", + "published_urls": "https://example.test/post", + } + values.update(overrides) + return values + + def test_current_site_blog_records_and_controlled_tags_load(self) -> None: + repository = BlogRepository(Path(__file__).resolve().parents[1]) + entries = repository.list_entries() + self.assertGreaterEqual(len(entries), 7) + self.assertGreaterEqual(len(repository.tags), 45) + self.assertTrue(all(entry.document.get("kicker") == "Blog" for entry in entries)) + for entry in entries: + with self.subTest(slug=entry.slug): + self.assertEqual( + entry.document.render().encode("utf-8"), entry.original_bytes + ) + + def test_create_uses_existing_record_conventions_and_refuses_collision(self) -> None: + entry = self.repository.create_entry( + "test-entry", self.values(body=" preserved indentation\n\nBody text.") + ) + rendered = entry.path.read_text(encoding="utf-8") + self.assertIn("_model: entry\n---\ntitle: Test Entry\n", rendered) + self.assertIn("tags: writing\n", rendered) + self.assertIn( + "published_urls:\n\nhttps://example.test/post\n---\nbody:\n", + rendered, + ) + with self.assertRaises(BlogEditorError): + self.repository.create_entry("test-entry", self.values()) + + def test_edit_preserves_unmanaged_and_unchanged_blocks_exactly(self) -> None: + entry = self.repository.create_entry( + "test-entry", self.values(body=" preserved indentation\n\nBody text.") + ) + original = entry.path.read_text(encoding="utf-8") + marker = "---\neditor_note:\nKeep this unknown field exactly. \n" + original = original.replace("---\nbody:\n", marker + "---\nbody:\n") + entry.path.write_text(original, encoding="utf-8") + + loaded = self.repository.load_entry("test-entry") + values = self.repository.values_for(loaded.document) + values["summary"] = "An edited summary." + saved = self.repository.save_entry(loaded, values) + rendered = saved.path.read_text(encoding="utf-8") + + self.assertIn(marker, rendered) + self.assertIn("summary: An edited summary.\n", rendered) + self.assertIn("body:\n preserved indentation\n\nBody text.\n", rendered) + self.assertIn("published_urls:\n\nhttps://example.test/post\n", rendered) + + def test_save_detects_an_external_change(self) -> None: + entry = self.repository.create_entry("test-entry", self.values()) + with entry.path.open("a", encoding="utf-8") as output: + output.write("\n") + with self.assertRaisesRegex(BlogEditorError, "changed on disk"): + self.repository.save_entry(entry, self.values(summary="Changed")) + + def test_delete_removes_only_the_loaded_entry_directory(self) -> None: + entry = self.repository.create_entry("test-entry", self.values()) + attachment = entry.path.parent / "narration.mp3" + attachment.write_bytes(b"disposable attachment") + neighbor = self.repository.create_entry("neighbor-entry", self.values()) + + self.repository.delete_entry(entry) + + self.assertFalse(entry.path.parent.exists()) + self.assertTrue(neighbor.path.is_file()) + + def test_delete_detects_an_external_change(self) -> None: + entry = self.repository.create_entry("test-entry", self.values()) + with entry.path.open("a", encoding="utf-8") as output: + output.write("\n") + with self.assertRaisesRegex(BlogEditorError, "changed on disk"): + self.repository.delete_entry(entry) + self.assertTrue(entry.path.is_file()) + + def test_unapproved_tag_is_rejected(self) -> None: + with self.assertRaisesRegex(BlogEditorError, "unapproved tag"): + self.repository.create_entry("test-entry", self.values(tags="invented")) + + def test_slug_suggestion_matches_repository_style(self) -> None: + self.assertEqual(suggest_slug("We’re Testing: A GUI!"), "were-testing-a-gui") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_narration_import.py b/tests/test_narration_import.py new file mode 100644 index 0000000..8284c49 --- /dev/null +++ b/tests/test_narration_import.py @@ -0,0 +1,91 @@ +from __future__ import annotations + +from pathlib import Path +import shutil +import subprocess +import sys +import unittest +import uuid +from unittest import mock + + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) + +from narration_import import NarrationImportError, import_wav, remove_narration # noqa: E402 + + +class NarrationImportTests(unittest.TestCase): + def setUp(self) -> None: + cache_root = Path(__file__).resolve().parents[1] / ".cache" + cache_root.mkdir(exist_ok=True) + self.directory = cache_root / f"narration-import-test-{uuid.uuid4().hex}" + self.directory.mkdir() + self.source = self.directory / "source.wav" + self.source.write_bytes(b"disposable wav data") + + def tearDown(self) -> None: + shutil.rmtree(self.directory) + + def test_success_uses_fixed_preset_and_replaces_only_after_temp_output(self) -> None: + destination = self.directory / "narration.mp3" + destination.write_bytes(b"old narration") + + def successful_ffmpeg(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]: + Path(command[-1]).write_bytes(b"converted mp3") + return subprocess.CompletedProcess(command, 0, "", "") + + with mock.patch("narration_import.subprocess.run", side_effect=successful_ffmpeg) as run: + result = import_wav(self.directory, self.source, ffmpeg="ffmpeg") + + self.assertEqual(result, destination) + self.assertEqual(destination.read_bytes(), b"converted mp3") + command = run.call_args.args[0] + self.assertEqual( + command[1:-1], + ["-y", "-i", str(self.source), "-vn", "-ar", "44100", "-ac", "1", + "-c:a", "libmp3lame", "-b:a", "96k"], + ) + self.assertFalse(any(self.directory.glob(".narration-*.mp3"))) + + def test_failed_conversion_preserves_existing_narration(self) -> None: + destination = self.directory / "narration.mp3" + destination.write_bytes(b"old narration") + failed = subprocess.CompletedProcess(["ffmpeg"], 1, "", "bad WAV") + + with mock.patch("narration_import.subprocess.run", return_value=failed): + with self.assertRaisesRegex(NarrationImportError, "bad WAV"): + import_wav(self.directory, self.source, ffmpeg="ffmpeg") + + self.assertEqual(destination.read_bytes(), b"old narration") + self.assertFalse(any(self.directory.glob(".narration-*.mp3"))) + + def test_missing_ffmpeg_and_removal_are_safe(self) -> None: + with mock.patch("narration_import.shutil.which", return_value=None): + with self.assertRaisesRegex(NarrationImportError, "ffmpeg is required"): + import_wav(self.directory, self.source) + unrelated = self.directory / "other-attachment.txt" + unrelated.write_text("keep", encoding="utf-8") + narration = self.directory / "narration.mp3" + narration.write_bytes(b"narration") + self.assertTrue(remove_narration(self.directory)) + self.assertFalse(narration.exists()) + self.assertEqual(unrelated.read_text(encoding="utf-8"), "keep") + + def test_repository_fallback_is_used_when_ffmpeg_is_not_on_path(self) -> None: + fallback = self.directory / "ffmpeg.exe" + fallback.write_bytes(b"placeholder executable") + + def successful_ffmpeg(command: list[str], **_kwargs: object) -> subprocess.CompletedProcess[str]: + Path(command[-1]).write_bytes(b"converted mp3") + return subprocess.CompletedProcess(command, 0, "", "") + + with mock.patch("narration_import.shutil.which", return_value=None), mock.patch( + "narration_import.subprocess.run", side_effect=successful_ffmpeg + ) as run: + import_wav(self.directory, self.source, fallback_ffmpeg=fallback) + + self.assertEqual(run.call_args.args[0][0], str(fallback)) + + +if __name__ == "__main__": + unittest.main()