Add blog and article narration import
Deploy production / deploy (push) Successful in 4s

This commit is contained in:
2026-08-13 15:12:33 -07:00
parent a2e405cb96
commit 2fe6b689b0
8 changed files with 2672 additions and 0 deletions
+47
View File
@@ -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
File diff suppressed because it is too large Load Diff
+959
View File
@@ -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("<Double-1>", 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("<Control-s>", 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())
+91
View File
@@ -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
Binary file not shown.
+187
View File
@@ -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("Were Testing: A GUI!"), "were-testing-a-gui")
if __name__ == "__main__":
unittest.main()
+138
View File
@@ -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("Were Testing: A GUI!"), "were-testing-a-gui")
if __name__ == "__main__":
unittest.main()
+91
View File
@@ -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()