963 lines
50 KiB
Python
963 lines
50 KiB
Python
#!/usr/bin/env python3
|
|
"""Local authoring-system prototype for a Scrivener project.
|
|
|
|
It previews and sends packages, and makes carefully scoped, lock-guarded
|
|
writes for accepted drafts. It remains a standard-library server.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import datetime as dt
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import urllib.error
|
|
import urllib.request
|
|
import uuid
|
|
import xml.etree.ElementTree as ET
|
|
from dataclasses import dataclass, field
|
|
from http import HTTPStatus
|
|
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
APP_DIR = Path(__file__).resolve().parent
|
|
DEFAULT_PROJECT = APP_DIR.parent / "hapagirlsauthoringsystemv1.scriv"
|
|
OPENROUTER_MODEL = "anthropic/claude-sonnet-4.6"
|
|
|
|
|
|
@dataclass
|
|
class Item:
|
|
uuid: str
|
|
title: str
|
|
item_type: str
|
|
children: list["Item"] = field(default_factory=list)
|
|
parent_title: str = ""
|
|
|
|
|
|
def rtf_to_text(path: Path) -> str:
|
|
"""Extract readable text; use macOS textutil when available, otherwise fall back."""
|
|
if shutil.which("textutil"):
|
|
result = subprocess.run(
|
|
["textutil", "-convert", "txt", "-stdout", str(path)],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if result.returncode == 0:
|
|
return result.stdout.strip()
|
|
|
|
raw = path.read_text(encoding="latin-1", errors="replace")
|
|
raw = re.sub(r"\\'[0-9a-fA-F]{2}", lambda m: bytes.fromhex(m.group()[2:]).decode("latin-1"), raw)
|
|
raw = re.sub(r"\\u-?\d+\??", "", raw)
|
|
raw = raw.replace("\\par", "\n").replace("\\line", "\n").replace("\\tab", "\t")
|
|
raw = re.sub(r"\\[a-zA-Z]+-?\d* ?", "", raw)
|
|
raw = raw.replace("{", "").replace("}", "").replace("\\", "")
|
|
return re.sub(r"\n{3,}", "\n\n", raw).strip()
|
|
|
|
|
|
def text_to_rtf(value: str) -> str:
|
|
"""Create deliberately simple RTF suitable for Scrivener text documents."""
|
|
escaped = []
|
|
for character in value.replace("\r\n", "\n").replace("\r", "\n"):
|
|
if character == "\n":
|
|
escaped.append("\\par\n")
|
|
elif character in "\\{}":
|
|
escaped.append("\\" + character)
|
|
elif ord(character) > 127:
|
|
codepoint = ord(character)
|
|
if codepoint > 32767:
|
|
codepoint -= 65536
|
|
escaped.append(f"\\u{codepoint}?")
|
|
else:
|
|
escaped.append(character)
|
|
return "{\\rtf1\\ansi\\deff0{\\fonttbl{\\f0\\fnil Helvetica;}}\n\\f0\\fs24 " + "".join(escaped) + "\n}"
|
|
|
|
|
|
class ScrivenerReader:
|
|
def __init__(self, project: Path):
|
|
self.project = project
|
|
self.bindings_path = project / "Files" / "authoring-system-bindings.json"
|
|
project_files = list(project.glob("*.scrivx"))
|
|
if len(project_files) != 1:
|
|
raise ValueError("Project must contain exactly one .scrivx file.")
|
|
self.project_file = project_files[0]
|
|
self.root_items = self._load_binder()
|
|
self.by_uuid = {item.uuid: item for item in self.walk()}
|
|
|
|
def binding_metadata(self) -> dict:
|
|
if not self.bindings_path.exists():
|
|
return {}
|
|
try:
|
|
data = json.loads(self.bindings_path.read_text(encoding="utf-8"))
|
|
return data if isinstance(data, dict) else {}
|
|
except (json.JSONDecodeError, OSError):
|
|
return {}
|
|
|
|
def bindings(self) -> dict[str, str]:
|
|
bindings = self.binding_metadata().get("scene_to_brief", {})
|
|
return {str(scene_uuid): str(brief_uuid) for scene_uuid, brief_uuid in bindings.items()} if isinstance(bindings, dict) else {}
|
|
|
|
def thread_predecessors(self) -> dict[str, str | None]:
|
|
predecessors = self.binding_metadata().get("scene_predecessors", {})
|
|
if not isinstance(predecessors, dict):
|
|
return {}
|
|
return {str(scene_uuid): (str(predecessor) if predecessor is not None else None) for scene_uuid, predecessor in predecessors.items()}
|
|
|
|
def _load_binder(self) -> list[Item]:
|
|
root = ET.parse(self.project_file).getroot()
|
|
binder = root.find("Binder")
|
|
if binder is None:
|
|
raise ValueError("Scrivener project has no Binder.")
|
|
|
|
def parse(element: ET.Element, parent_title: str = "") -> Item:
|
|
item = Item(
|
|
uuid=element.attrib["UUID"],
|
|
title=(element.findtext("Title") or "").strip(),
|
|
item_type=element.attrib.get("Type", ""),
|
|
parent_title=parent_title,
|
|
)
|
|
children = element.find("Children")
|
|
if children is not None:
|
|
item.children = [parse(child, item.title) for child in children.findall("BinderItem")]
|
|
return item
|
|
|
|
return [parse(item) for item in binder.findall("BinderItem")]
|
|
|
|
def walk(self, items: list[Item] | None = None):
|
|
for item in self.root_items if items is None else items:
|
|
yield item
|
|
yield from self.walk(item.children)
|
|
|
|
def top_level(self, title: str) -> Item | None:
|
|
return next((item for item in self.root_items if item.title == title), None)
|
|
|
|
@staticmethod
|
|
def child(item: Item | None, title: str) -> Item | None:
|
|
if item is None:
|
|
return None
|
|
return next((candidate for candidate in item.children if candidate.title == title), None)
|
|
|
|
def content(self, item: Item | None) -> str:
|
|
if item is None:
|
|
return ""
|
|
path = self.project / "Files" / "Data" / item.uuid / "content.rtf"
|
|
return rtf_to_text(path) if path.exists() else ""
|
|
|
|
@staticmethod
|
|
def records(folder: Item | None, reader: "ScrivenerReader") -> list[dict]:
|
|
if folder is None:
|
|
return []
|
|
return [
|
|
{"uuid": item.uuid, "title": item.title, "text": reader.content(item)}
|
|
for item in folder.children
|
|
if item.item_type == "Text"
|
|
]
|
|
|
|
def manuscript_scenes(self) -> list[dict]:
|
|
manuscript = self.top_level("Manuscript")
|
|
scenes: list[dict] = []
|
|
if manuscript is None:
|
|
return scenes
|
|
for chapter in manuscript.children:
|
|
for item in chapter.children:
|
|
if item.item_type == "Text":
|
|
scenes.append(
|
|
{
|
|
"uuid": item.uuid,
|
|
"title": item.title,
|
|
"chapter": chapter.title,
|
|
"text": self.content(item),
|
|
"state": "canonical" if self.content(item).strip() else "planned",
|
|
}
|
|
)
|
|
return scenes
|
|
|
|
def manuscript_chapters(self) -> list[dict]:
|
|
manuscript = self.top_level("Manuscript")
|
|
if manuscript is None:
|
|
return []
|
|
return [{"uuid": item.uuid, "title": item.title} for item in manuscript.children if item.item_type == "Folder"]
|
|
|
|
@staticmethod
|
|
def lexicon_entries(text: str) -> list[dict]:
|
|
"""Read author-maintained ``Term — definition`` paragraphs from Master Lexicon."""
|
|
entries: list[dict] = []
|
|
entry_lines: list[str] = []
|
|
|
|
def save_entry() -> None:
|
|
if not entry_lines:
|
|
return
|
|
entry = "\n".join(entry_lines).strip()
|
|
title, _, definition = entry.partition(" — ")
|
|
if title.strip() and definition.strip():
|
|
entries.append({"title": title.strip(), "text": entry})
|
|
|
|
for raw_line in text.splitlines():
|
|
line = raw_line.strip()
|
|
if " — " in line:
|
|
save_entry()
|
|
entry_lines = [line]
|
|
elif not line:
|
|
save_entry()
|
|
entry_lines = []
|
|
elif entry_lines:
|
|
entry_lines.append(line)
|
|
save_entry()
|
|
return entries
|
|
|
|
def summary(self) -> dict:
|
|
authoring = self.top_level("Authoring System")
|
|
setup = self.child(authoring, "Book Setup")
|
|
continuity = self.child(authoring, "Continuity")
|
|
briefs_folder = self.child(authoring, "Scene Briefs")
|
|
summaries_folder = self.child(continuity, "Scene Summaries")
|
|
archive = self.child(authoring, "Drafting Archive")
|
|
accepted_folder = self.child(archive, "Accepted for Editing")
|
|
snapshots_folder = self.child(archive, "Package Snapshots")
|
|
style = self.child(setup, "STYLE")
|
|
instructions = self.child(setup, "Frontier-Model Instructions")
|
|
lexicon = self.child(continuity, "Master Lexicon")
|
|
world_notes = self.child(continuity, "World Notes")
|
|
manuscript = self.manuscript_scenes()
|
|
briefs = self.records(briefs_folder, self)
|
|
brief_by_title = {brief["title"]: brief for brief in briefs}
|
|
brief_by_uuid = {brief["uuid"]: brief for brief in briefs}
|
|
bindings = self.bindings()
|
|
predecessors = self.thread_predecessors()
|
|
summary_by_title = {record["title"]: record for record in self.records(summaries_folder, self)}
|
|
character_by_title = {character["title"]: character["uuid"] for character in self.records(self.top_level("Characters"), self)}
|
|
working_by_title: dict[str, dict] = {}
|
|
for record in self.records(accepted_folder, self):
|
|
match = re.match(r"^(.*?) — working copy — ", record["title"])
|
|
if match:
|
|
working_by_title[match.group(1)] = record
|
|
package_context: dict[str, dict] = {}
|
|
for snapshot in self.records(snapshots_folder, self):
|
|
scene_title = re.search(r"^Scene:\s*(.+)$", snapshot["text"], re.MULTILINE)
|
|
if not scene_title:
|
|
continue
|
|
title = scene_title.group(1).strip()
|
|
text = snapshot["text"]
|
|
uuid_line = re.search(r"^Character UUIDs:\s*(.*)$", text, re.MULTILINE)
|
|
if uuid_line:
|
|
character_uuids = [value for value in uuid_line.group(1).split(",") if value]
|
|
else:
|
|
character_block = self._package_section(text, "CHARACTERS")
|
|
character_uuids = [character_by_title[line.strip()] for line in re.findall(r"^(.+?)\n", character_block) if line.strip() in character_by_title]
|
|
notes = self._package_section(text, "NOTES")
|
|
place_uuids: list[str] = []
|
|
for place in self.records(self.top_level("Places"), self):
|
|
place_block = f"{place['title']} — {place['text'].strip()}"
|
|
if place_block and place_block in notes:
|
|
place_uuids.append(place["uuid"])
|
|
notes = notes.replace(place_block, "").strip()
|
|
package_context[title] = {
|
|
"character_uuids": character_uuids,
|
|
"place_uuids": place_uuids,
|
|
"lexicon": self._package_section(text, "LEXICON"),
|
|
"story": self._package_section(text, "STORY"),
|
|
"recall": self._package_section(text, "RECALL"),
|
|
"previous": self._package_section(text, "PREVIOUS"),
|
|
"notes": notes,
|
|
"directive": self._package_section(text, "DIRECTIVE"),
|
|
"exact_package": text.split("\n\n", 1)[1] if "\n\n" in text else text,
|
|
}
|
|
|
|
paired = []
|
|
bound_brief_uuids: set[str] = set()
|
|
for scene in manuscript:
|
|
brief = brief_by_uuid.get(bindings.get(scene["uuid"], ""))
|
|
binding_state = "stable" if brief else ""
|
|
if brief is None:
|
|
brief = brief_by_title.get(scene["title"])
|
|
binding_state = "legacy-title-match" if brief else "unpaired"
|
|
if brief:
|
|
bound_brief_uuids.add(brief["uuid"])
|
|
paired.append(
|
|
{
|
|
**scene,
|
|
"brief": (brief or {}).get("text", ""),
|
|
"brief_uuid": (brief or {}).get("uuid"),
|
|
"binding_state": binding_state,
|
|
"summary": summary_by_title.get(scene["title"], {}).get("text", ""),
|
|
"working_copy": working_by_title.get(scene["title"]),
|
|
"package_context": package_context.get(scene["title"], {"character_uuids": [], "place_uuids": [], "lexicon": "", "story": "", "recall": "", "previous": "", "notes": "", "directive": "", "exact_package": ""}),
|
|
"predecessor_uuid": predecessors.get(scene["uuid"]),
|
|
"thread_state": "new-thread" if scene["uuid"] in predecessors and predecessors[scene["uuid"]] is None else "continues" if predecessors.get(scene["uuid"]) else "unassigned",
|
|
}
|
|
)
|
|
|
|
return {
|
|
"project": str(self.project),
|
|
"locked": (self.project / "Files" / "user.lock").exists(),
|
|
"style": self.content(style),
|
|
"frontier_instructions": self.content(instructions),
|
|
"master_lexicon": self.content(lexicon),
|
|
"lexicon_entries": self.lexicon_entries(self.content(lexicon)),
|
|
"world_notes": self.content(world_notes),
|
|
"scenes": paired,
|
|
"chapters": self.manuscript_chapters(),
|
|
"characters": self.records(self.top_level("Characters"), self),
|
|
"places": self.records(self.top_level("Places"), self),
|
|
"unmatched_briefs": [brief for brief in briefs if brief["uuid"] not in bound_brief_uuids],
|
|
"unmatched_scenes": [scene for scene in paired if not scene["brief_uuid"]],
|
|
"binding_counts": {
|
|
"stable": sum(scene["binding_state"] == "stable" for scene in paired),
|
|
"legacy": sum(scene["binding_state"] == "legacy-title-match" for scene in paired),
|
|
"unpaired": sum(scene["binding_state"] == "unpaired" for scene in paired),
|
|
},
|
|
}
|
|
|
|
@staticmethod
|
|
def _package_section(package: str, name: str) -> str:
|
|
match = re.search(rf"^### {re.escape(name)}\n(.*?)(?=^### |\Z)", package, re.MULTILINE | re.DOTALL)
|
|
return match.group(1).strip() if match else ""
|
|
|
|
|
|
class ScrivenerWriter:
|
|
"""The small, guarded subset of Scrivener writes needed by this prototype."""
|
|
|
|
def __init__(self, project: Path):
|
|
self.project = project
|
|
self.reader = ScrivenerReader(project)
|
|
|
|
def _assert_unlocked(self) -> None:
|
|
if (self.project / "Files" / "user.lock").exists():
|
|
raise ValueError("Scrivener is open (Files/user.lock exists). Close the project before writing.")
|
|
|
|
def _write_bindings(self, bindings: dict[str, str], predecessors: dict[str, str | None] | None = None) -> None:
|
|
payload = self.reader.binding_metadata()
|
|
payload.update({
|
|
"version": 1,
|
|
"updated_at": dt.datetime.now().astimezone().isoformat(timespec="seconds"),
|
|
"scene_to_brief": bindings,
|
|
})
|
|
if predecessors is not None:
|
|
payload["scene_predecessors"] = predecessors
|
|
temporary = self.reader.bindings_path.with_suffix(".tmp")
|
|
temporary.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
|
|
os.replace(temporary, self.reader.bindings_path)
|
|
|
|
def assign_thread_predecessor(self, scene_uuid: str, predecessor_uuid: str | None) -> None:
|
|
"""Store the author's explicit narrative-thread relationship for a scene."""
|
|
self._assert_unlocked()
|
|
scenes = {scene["uuid"]: scene for scene in self.reader.summary()["scenes"]}
|
|
if scene_uuid not in scenes:
|
|
raise ValueError("The selected Manuscript scene no longer exists.")
|
|
if predecessor_uuid is not None:
|
|
if predecessor_uuid not in scenes or not scenes[predecessor_uuid]["text"].strip():
|
|
raise ValueError("A narrative predecessor must be a canonical Manuscript scene.")
|
|
if predecessor_uuid == scene_uuid:
|
|
raise ValueError("A scene cannot continue from itself.")
|
|
predecessors = self.reader.thread_predecessors()
|
|
cursor = predecessor_uuid
|
|
while cursor is not None:
|
|
if cursor == scene_uuid:
|
|
raise ValueError("That narrative-thread link would create a cycle.")
|
|
cursor = predecessors.get(cursor)
|
|
predecessors[scene_uuid] = predecessor_uuid
|
|
self._write_bindings(self.reader.bindings(), predecessors)
|
|
self.reader = ScrivenerReader(self.project)
|
|
|
|
def reconcile_scene_bindings(self) -> dict:
|
|
"""Persist currently resolved scene/BRIEF pairs without relying on their titles."""
|
|
self._assert_unlocked()
|
|
data = self.reader.summary()
|
|
briefs_folder = self.reader.child(self.reader.top_level("Authoring System"), "Scene Briefs")
|
|
valid_brief_uuids = {item.uuid for item in (briefs_folder.children if briefs_folder else []) if item.item_type == "Text"}
|
|
bindings = {scene_uuid: brief_uuid for scene_uuid, brief_uuid in self.reader.bindings().items() if scene_uuid in self.reader.by_uuid and brief_uuid in valid_brief_uuids}
|
|
for scene in data["scenes"]:
|
|
if scene["brief_uuid"]:
|
|
bindings[scene["uuid"]] = scene["brief_uuid"]
|
|
self._write_bindings(bindings)
|
|
self.reader = ScrivenerReader(self.project)
|
|
return self.reader.summary()["binding_counts"]
|
|
|
|
def _tree_and_binder(self) -> tuple[ET.ElementTree, ET.Element]:
|
|
tree = ET.parse(self.reader.project_file)
|
|
binder = tree.getroot().find("Binder")
|
|
if binder is None:
|
|
raise ValueError("Scrivener project has no Binder.")
|
|
return tree, binder
|
|
|
|
@staticmethod
|
|
def _find_child(element: ET.Element, title: str) -> ET.Element | None:
|
|
children = element.find("Children")
|
|
if children is None:
|
|
return None
|
|
return next((item for item in children.findall("BinderItem") if (item.findtext("Title") or "").strip() == title), None)
|
|
|
|
def _folder(self, binder: ET.Element, path: list[str]) -> ET.Element:
|
|
parent: ET.Element | None = None
|
|
children = binder
|
|
for title in path:
|
|
parent = next((item for item in children.findall("BinderItem") if (item.findtext("Title") or "").strip() == title), None)
|
|
if parent is None:
|
|
raise ValueError("Required Scrivener folder is missing: " + " → ".join(path))
|
|
children = parent.find("Children")
|
|
if children is None and title != path[-1]:
|
|
raise ValueError("Required Scrivener folder has no children: " + title)
|
|
assert parent is not None
|
|
return parent
|
|
|
|
@staticmethod
|
|
def _element_by_uuid(parent: ET.Element, item_uuid: str) -> ET.Element | None:
|
|
for item in parent.findall("BinderItem"):
|
|
if item.attrib.get("UUID") == item_uuid:
|
|
return item
|
|
children = item.find("Children")
|
|
if children is not None:
|
|
found = ScrivenerWriter._element_by_uuid(children, item_uuid)
|
|
if found is not None:
|
|
return found
|
|
return None
|
|
|
|
def _backup_binder(self) -> None:
|
|
backup_dir = self.project / "Files" / "authoring-backups"
|
|
backup_dir.mkdir(parents=True, exist_ok=True)
|
|
stamp = dt.datetime.now().strftime("%Y%m%d-%H%M%S")
|
|
shutil.copy2(self.reader.project_file, backup_dir / f"{stamp}.scrivx")
|
|
checksum = self.project / "Files" / "Data" / "docs.checksum"
|
|
if checksum.exists():
|
|
shutil.copy2(checksum, backup_dir / f"{stamp}.docs.checksum")
|
|
|
|
def _update_checksum(self, uuid: str, content: bytes) -> None:
|
|
checksum_path = self.project / "Files" / "Data" / "docs.checksum"
|
|
digest = hashlib.sha1(content).hexdigest()
|
|
key = f"{uuid.lower()}/content.rtf"
|
|
lines = checksum_path.read_text(encoding="utf-8", errors="replace").splitlines() if checksum_path.exists() else []
|
|
updated = False
|
|
output = []
|
|
for line in lines:
|
|
if line.startswith(key + "="):
|
|
output.append(f"{key}={digest}")
|
|
updated = True
|
|
else:
|
|
output.append(line)
|
|
if not updated:
|
|
output.append(f"{key}={digest}")
|
|
checksum_path.write_text("\n".join(output) + "\n", encoding="utf-8")
|
|
|
|
def _write_rtf(self, uuid: str, text: str) -> None:
|
|
content = text_to_rtf(text).encode("utf-8")
|
|
data_dir = self.project / "Files" / "Data" / uuid
|
|
data_dir.mkdir(parents=True, exist_ok=False)
|
|
(data_dir / "content.rtf").write_bytes(content)
|
|
self._update_checksum(uuid, content)
|
|
|
|
def _add_text(self, folder: ET.Element, title: str, text: str) -> str:
|
|
children = folder.find("Children")
|
|
if children is None:
|
|
children = ET.SubElement(folder, "Children")
|
|
doc_uuid = str(uuid.uuid4()).upper()
|
|
stamp = dt.datetime.now().astimezone().isoformat(timespec="seconds")
|
|
item = ET.SubElement(children, "BinderItem", {"UUID": doc_uuid, "Type": "Text", "Created": stamp, "Modified": stamp})
|
|
ET.SubElement(item, "Title").text = title
|
|
metadata = ET.SubElement(item, "MetaData")
|
|
ET.SubElement(metadata, "IncludeInCompile").text = "No"
|
|
settings = ET.SubElement(item, "TextSettings")
|
|
ET.SubElement(settings, "TextSelection").text = "0,0"
|
|
self._write_rtf(doc_uuid, text)
|
|
return doc_uuid
|
|
|
|
def _save_tree(self, tree: ET.ElementTree) -> None:
|
|
tree.write(self.reader.project_file, encoding="UTF-8", xml_declaration=True)
|
|
|
|
def save_brief(self, scene_uuid: str, text: str) -> dict:
|
|
"""Update an existing BRIEF or create and bind one for a planned scene."""
|
|
self._assert_unlocked()
|
|
scene = self.reader.by_uuid.get(scene_uuid)
|
|
if scene is None:
|
|
raise ValueError("The selected Manuscript scene no longer exists.")
|
|
paired = next((item for item in self.reader.summary()["scenes"] if item["uuid"] == scene_uuid), None)
|
|
if paired and paired["brief_uuid"]:
|
|
brief_uuid = paired["brief_uuid"]
|
|
path = self.project / "Files" / "Data" / brief_uuid / "content.rtf"
|
|
content = text_to_rtf(text).encode("utf-8")
|
|
path.write_bytes(content)
|
|
self._update_checksum(brief_uuid, content)
|
|
bindings = self.reader.bindings()
|
|
bindings[scene_uuid] = brief_uuid
|
|
self._write_bindings(bindings)
|
|
self.reader = ScrivenerReader(self.project)
|
|
return {"created": False, "brief_uuid": brief_uuid}
|
|
self._backup_binder()
|
|
tree, binder = self._tree_and_binder()
|
|
briefs_folder = self._folder(binder, ["Authoring System", "Scene Briefs"])
|
|
brief_uuid = self._add_text(briefs_folder, scene.title, text)
|
|
self._save_tree(tree)
|
|
bindings = self.reader.bindings()
|
|
bindings[scene_uuid] = brief_uuid
|
|
self._write_bindings(bindings)
|
|
self.reader = ScrivenerReader(self.project)
|
|
return {"created": True, "brief_uuid": brief_uuid}
|
|
|
|
def create_planned_scene(self, chapter_uuid: str, title: str) -> dict:
|
|
"""Create a blank Manuscript scene and its blank, stably bound BRIEF."""
|
|
self._assert_unlocked()
|
|
manuscript = self.reader.top_level("Manuscript")
|
|
chapter = self.reader.by_uuid.get(chapter_uuid)
|
|
if manuscript is None or chapter is None or chapter_uuid not in {item.uuid for item in manuscript.children}:
|
|
raise ValueError("Choose a current Manuscript chapter.")
|
|
title = title.strip()
|
|
existing_titles = {scene["title"] for scene in self.reader.summary()["scenes"]}
|
|
if not title:
|
|
numbers = [int(match.group(1)) for scene_title in existing_titles if (match := re.fullmatch(r"Scene\s+(\d+)", scene_title, re.IGNORECASE))]
|
|
title = f"Scene {max(numbers, default=0) + 1}"
|
|
if title in existing_titles:
|
|
raise ValueError(f"A Manuscript scene named {title!r} already exists.")
|
|
self._backup_binder()
|
|
tree, binder = self._tree_and_binder()
|
|
chapter_element = self._element_by_uuid(binder, chapter_uuid)
|
|
if chapter_element is None:
|
|
raise ValueError("The selected chapter no longer exists.")
|
|
briefs_folder = self._folder(binder, ["Authoring System", "Scene Briefs"])
|
|
scene_uuid = self._add_text(chapter_element, title, "")
|
|
brief_uuid = self._add_text(briefs_folder, title, "")
|
|
self._save_tree(tree)
|
|
bindings = self.reader.bindings()
|
|
bindings[scene_uuid] = brief_uuid
|
|
self._write_bindings(bindings)
|
|
self.reader = ScrivenerReader(self.project)
|
|
return {"scene_uuid": scene_uuid, "brief_uuid": brief_uuid, "title": title}
|
|
|
|
def accept_for_editing(self, scene_uuid: str, draft_text: str, package: str, model: str, character_uuids: list[str]) -> dict:
|
|
self._assert_unlocked()
|
|
if not draft_text.strip() or not package.strip():
|
|
raise ValueError("A non-empty draft and package are required for acceptance.")
|
|
scene = self.reader.by_uuid.get(scene_uuid)
|
|
if scene is None:
|
|
raise ValueError("The selected scene no longer exists.")
|
|
self._backup_binder()
|
|
tree, binder = self._tree_and_binder()
|
|
accepted = self._folder(binder, ["Authoring System", "Drafting Archive", "Accepted for Editing"])
|
|
snapshots = self._folder(binder, ["Authoring System", "Drafting Archive", "Package Snapshots"])
|
|
stamp = dt.datetime.now().strftime("%Y-%m-%d %H%M")
|
|
initial_uuid = self._add_text(accepted, f"{scene.title} — initial draft — {stamp}", draft_text)
|
|
working_uuid = self._add_text(accepted, f"{scene.title} — working copy — {stamp}", draft_text)
|
|
package_header = f"Scene: {scene.title}\nModel: {model}\nCharacter UUIDs: {','.join(character_uuids)}\nAccepted for editing: {dt.datetime.now().astimezone().isoformat(timespec='seconds')}\n\n"
|
|
package_uuid = self._add_text(snapshots, f"{scene.title} — package — {stamp}", package_header + package)
|
|
self._save_tree(tree)
|
|
self.reader = ScrivenerReader(self.project)
|
|
return {"initial_uuid": initial_uuid, "working_uuid": working_uuid, "package_uuid": package_uuid}
|
|
|
|
def save_working_copy(self, working_uuid: str, text: str) -> None:
|
|
self._assert_unlocked()
|
|
item = self.reader.by_uuid.get(working_uuid)
|
|
if item is None or item.item_type != "Text":
|
|
raise ValueError("The working-copy archive record no longer exists.")
|
|
path = self.project / "Files" / "Data" / working_uuid / "content.rtf"
|
|
if not path.exists():
|
|
raise ValueError("The working-copy archive record has no text file.")
|
|
content = text_to_rtf(text).encode("utf-8")
|
|
path.write_bytes(content)
|
|
self._update_checksum(working_uuid, content)
|
|
|
|
def accept_into_book(self, scene_uuid: str, text: str, predecessor_uuid: str | None) -> None:
|
|
self._assert_unlocked()
|
|
item = self.reader.by_uuid.get(scene_uuid)
|
|
if item is None or item.item_type != "Text":
|
|
raise ValueError("The selected Manuscript scene no longer exists.")
|
|
manuscript = self.reader.top_level("Manuscript")
|
|
if manuscript is None or not any(candidate.uuid == scene_uuid for candidate in self.reader.walk(manuscript.children)):
|
|
raise ValueError("The selected item is not a Manuscript scene.")
|
|
path = self.project / "Files" / "Data" / scene_uuid / "content.rtf"
|
|
content = text_to_rtf(text).encode("utf-8")
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(content)
|
|
self._update_checksum(scene_uuid, content)
|
|
self.assign_thread_predecessor(scene_uuid, predecessor_uuid)
|
|
|
|
def save_scene_summary(self, scene_uuid: str, summary: str) -> str:
|
|
"""Create or update the one author-approved dense summary for a scene."""
|
|
self._assert_unlocked()
|
|
scene = self.reader.by_uuid.get(scene_uuid)
|
|
if scene is None or not summary.strip():
|
|
raise ValueError("A scene and non-empty summary are required.")
|
|
tree, binder = self._tree_and_binder()
|
|
summaries = self._folder(binder, ["Authoring System", "Continuity", "Scene Summaries"])
|
|
existing = self._find_child(summaries, scene.title)
|
|
if existing is not None:
|
|
existing_uuid = existing.attrib.get("UUID", "")
|
|
path = self.project / "Files" / "Data" / existing_uuid / "content.rtf"
|
|
if not existing_uuid or not path.exists():
|
|
raise ValueError("The existing scene summary record is incomplete.")
|
|
content = text_to_rtf(summary).encode("utf-8")
|
|
path.write_bytes(content)
|
|
self._update_checksum(existing_uuid, content)
|
|
return existing_uuid
|
|
self._backup_binder()
|
|
summary_uuid = self._add_text(summaries, scene.title, summary)
|
|
self._save_tree(tree)
|
|
self.reader = ScrivenerReader(self.project)
|
|
return summary_uuid
|
|
|
|
def _append_to_item(self, item: Item, addition: str) -> bool:
|
|
path = self.project / "Files" / "Data" / item.uuid / "content.rtf"
|
|
existing = self.reader.content(item)
|
|
addition = addition.strip()
|
|
if not addition or addition in existing:
|
|
return False
|
|
content = text_to_rtf((existing.rstrip() + "\n\n" + addition).strip() + "\n").encode("utf-8")
|
|
path.write_bytes(content)
|
|
self._update_checksum(item.uuid, content)
|
|
return True
|
|
|
|
def _replace_item(self, item: Item, text: str) -> bool:
|
|
text = text.strip()
|
|
if not text:
|
|
raise ValueError("A Character record cannot be replaced with empty text.")
|
|
path = self.project / "Files" / "Data" / item.uuid / "content.rtf"
|
|
if self.reader.content(item).strip() == text:
|
|
return False
|
|
content = text_to_rtf(text + "\n").encode("utf-8")
|
|
path.write_bytes(content)
|
|
self._update_checksum(item.uuid, content)
|
|
return True
|
|
|
|
def save_continuity_updates(self, scene_uuid: str, character_updates: list[dict], new_characters: list[dict], lexicon_additions: str, world_note_additions: str) -> dict:
|
|
self._assert_unlocked()
|
|
scene = self.reader.by_uuid.get(scene_uuid)
|
|
if scene is None:
|
|
raise ValueError("The selected scene no longer exists.")
|
|
characters_folder = self.reader.top_level("Characters")
|
|
characters = {item.uuid: item for item in (characters_folder.children if characters_folder else []) if item.item_type == "Text"}
|
|
saved_characters = []
|
|
for update in character_updates:
|
|
item = characters.get(update.get("uuid", ""))
|
|
if item is None:
|
|
raise ValueError("A proposed character update does not target a current Character record.")
|
|
if self._replace_item(item, update.get("text", "")):
|
|
saved_characters.append(item.title)
|
|
stamp = f"Continuity update from {scene.title}:"
|
|
authoring = self.reader.top_level("Authoring System")
|
|
continuity = self.reader.child(authoring, "Continuity")
|
|
lexicon = self.reader.child(continuity, "Master Lexicon")
|
|
world_notes = self.reader.child(continuity, "World Notes")
|
|
lexicon_saved = self._append_to_item(lexicon, f"{stamp}\n{lexicon_additions.strip()}") if lexicon and lexicon_additions.strip() else False
|
|
world_saved = self._append_to_item(world_notes, f"{stamp}\n{world_note_additions.strip()}") if world_notes and world_note_additions.strip() else False
|
|
existing_titles = {item.title.casefold() for item in characters.values()}
|
|
created_characters = []
|
|
candidates = []
|
|
for character in new_characters:
|
|
name = str(character.get("name", "")).strip()
|
|
description = str(character.get("text", "")).strip()
|
|
if name and description and name.casefold() not in existing_titles:
|
|
candidates.append((name, description))
|
|
existing_titles.add(name.casefold())
|
|
if candidates:
|
|
self._backup_binder()
|
|
tree, binder = self._tree_and_binder()
|
|
characters_folder = self._folder(binder, ["Characters"])
|
|
for name, description in candidates:
|
|
self._add_text(characters_folder, name, description)
|
|
created_characters.append(name)
|
|
self._save_tree(tree)
|
|
self.reader = ScrivenerReader(self.project)
|
|
return {"characters": saved_characters, "new_characters": created_characters, "lexicon": lexicon_saved, "world_notes": world_saved}
|
|
|
|
|
|
def section(name: str, value: str) -> str:
|
|
return f"### {name}\n{value.strip() or '(none)'}"
|
|
|
|
|
|
def preflight(project: ScrivenerReader, payload: dict) -> list[str]:
|
|
data = project.summary()
|
|
scene = next((item for item in data["scenes"] if item["uuid"] == payload.get("scene_uuid")), None)
|
|
if scene is None or not scene["brief"].strip():
|
|
raise ValueError("Choose a scene with a paired BRIEF.")
|
|
if not data["style"].strip():
|
|
raise ValueError("STYLE is empty.")
|
|
directive = payload.get("directive", "").strip()
|
|
if not directive:
|
|
raise ValueError("DIRECTIVE is required.")
|
|
if not re.search(r"\b(maximum|max|not exceed)\b", directive, re.IGNORECASE):
|
|
raise ValueError("DIRECTIVE needs a word maximum.")
|
|
return ["No scene-specific NOTES or Place records selected; verify that no physical or time continuity constraint is needed."] if not payload.get("notes", "").strip() and not payload.get("place_uuids", []) else []
|
|
|
|
|
|
def build_package(project: ScrivenerReader, payload: dict) -> str:
|
|
preflight(project, payload)
|
|
data = project.summary()
|
|
scene = next((scene for scene in data["scenes"] if scene["uuid"] == payload.get("scene_uuid")), None)
|
|
if scene is None:
|
|
raise ValueError("Choose a paired Manuscript scene.")
|
|
if not scene["brief"].strip():
|
|
raise ValueError(f"{scene['title']} has no paired BRIEF.")
|
|
|
|
selected_characters = set(payload.get("character_uuids", []))
|
|
character_blocks = []
|
|
for character in data["characters"]:
|
|
if character["uuid"] in selected_characters:
|
|
character_blocks.append(f"{character['title']}\n{character['text'].strip()}")
|
|
|
|
selected_places = set(payload.get("place_uuids", []))
|
|
place_blocks = []
|
|
for place in data["places"]:
|
|
if place["uuid"] in selected_places:
|
|
place_blocks.append(f"{place['title']} — {place['text'].strip()}")
|
|
|
|
notes = "\n".join(block for block in ["\n".join(place_blocks), payload.get("notes", "").strip()] if block)
|
|
story = payload.get("story", "").strip()
|
|
if not story:
|
|
story = "(none - new storyline)" if payload.get("new_thread", True) else "(none)"
|
|
|
|
return "\n\n".join(
|
|
[
|
|
section("STYLE", data["style"]),
|
|
section("LEXICON", payload.get("lexicon", "")),
|
|
section("STORY", story),
|
|
section("CHARACTERS", "\n\n".join(character_blocks)),
|
|
section("RECALL", payload.get("recall", "")),
|
|
section("PREVIOUS", payload.get("previous", "")),
|
|
section("NOTES", notes),
|
|
section("BRIEF", scene["brief"]),
|
|
section("DIRECTIVE", payload.get("directive", "")),
|
|
"### SCENE",
|
|
]
|
|
) + "\n"
|
|
|
|
|
|
def draft(project: ScrivenerReader, payload: dict) -> dict:
|
|
key = os.environ.get("OPENROUTER_API_KEY")
|
|
if not key:
|
|
raise ValueError("OPENROUTER_API_KEY is not set in this server's Terminal session.")
|
|
if not payload.get("send_confirmation"):
|
|
raise ValueError("Confirm the OpenRouter send before generating.")
|
|
system = project.summary()["frontier_instructions"].strip()
|
|
package = build_package(project, payload)
|
|
body = json.dumps({"model": OPENROUTER_MODEL, "messages": [{"role": "system", "content": system}, {"role": "user", "content": package}], "max_tokens": int(payload.get("max_tokens", 3000))}).encode()
|
|
request = urllib.request.Request("https://openrouter.ai/api/v1/chat/completions", data=body, headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"}, method="POST")
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=300) as response:
|
|
result = json.loads(response.read().decode())
|
|
except urllib.error.HTTPError as error:
|
|
raise ValueError(f"OpenRouter returned HTTP {error.code}: {error.read().decode(errors='replace')}") from error
|
|
text = result.get("choices", [{}])[0].get("message", {}).get("content")
|
|
if not isinstance(text, str):
|
|
raise ValueError("OpenRouter response did not contain draft text.")
|
|
return {"draft": text, "usage": result.get("usage", {}), "model": result.get("model", OPENROUTER_MODEL)}
|
|
|
|
|
|
def propose_summary(project: ScrivenerReader, payload: dict) -> dict:
|
|
key = os.environ.get("OPENROUTER_API_KEY")
|
|
if not key:
|
|
raise ValueError("OPENROUTER_API_KEY is not set in this server's Terminal session.")
|
|
if not payload.get("send_confirmation"):
|
|
raise ValueError("Confirm the OpenRouter send before generating a summary.")
|
|
scene = project.by_uuid.get(payload.get("scene_uuid", ""))
|
|
text = payload.get("text", "").strip()
|
|
if scene is None or not text:
|
|
raise ValueError("A canonical scene is required to generate a summary.")
|
|
system = "You create continuity records for a fiction-authoring system. Return exactly one dense, factual sentence that captures the scene's viewpoint, material plot movement, character or relationship changes, discoveries, and unresolved state. Do not add a title, explanation, or invented facts."
|
|
user = f"Scene label: {scene.title}\n\nCanonical scene:\n{text}"
|
|
body = json.dumps({"model": OPENROUTER_MODEL, "messages": [{"role": "system", "content": system}, {"role": "user", "content": user}], "max_tokens": 500}).encode()
|
|
request = urllib.request.Request("https://openrouter.ai/api/v1/chat/completions", data=body, headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"}, method="POST")
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=300) as response:
|
|
result = json.loads(response.read().decode())
|
|
except urllib.error.HTTPError as error:
|
|
raise ValueError(f"OpenRouter returned HTTP {error.code}: {error.read().decode(errors='replace')}") from error
|
|
summary = result.get("choices", [{}])[0].get("message", {}).get("content")
|
|
if not isinstance(summary, str) or not summary.strip():
|
|
raise ValueError("OpenRouter response did not contain a scene summary.")
|
|
return {"summary": summary.strip(), "model": result.get("model", OPENROUTER_MODEL)}
|
|
|
|
|
|
def propose_continuity_updates(project: ScrivenerReader, payload: dict) -> dict:
|
|
key = os.environ.get("OPENROUTER_API_KEY")
|
|
if not key:
|
|
raise ValueError("OPENROUTER_API_KEY is not set in this server's Terminal session.")
|
|
if not payload.get("send_confirmation"):
|
|
raise ValueError("Confirm the OpenRouter send before generating continuity proposals.")
|
|
scene = project.by_uuid.get(payload.get("scene_uuid", ""))
|
|
text = payload.get("text", "").strip()
|
|
if scene is None or not text:
|
|
raise ValueError("A canonical scene is required to propose continuity updates.")
|
|
data = project.summary()
|
|
context = {"characters": data["characters"], "master_lexicon": data["master_lexicon"], "world_notes": data["world_notes"]}
|
|
system = "You propose factual continuity updates for a fiction-authoring system. Return one strict JSON object and nothing else: no Markdown, explanation, or code fences. It must contain exactly these keys: character_updates (array of objects with character_uuid and text), new_characters (array of objects with name and text), lexicon_additions (string), world_note_additions (string). Suggest only information newly established or materially changed by the canonical scene and absent from the supplied records. Keep every proposed text field to at most two concise sentences and return no more than eight total proposed records across all arrays. Character text must be a concise factual update, without a heading. Use a supplied UUID only when updating an existing character. Put a newly introduced named person in new_characters, never in lexicon_additions; name is the character's display name and text is an initial concise factual Character record. Do not create a new character for a passing unnamed person. Use empty arrays or empty strings when there is no justified addition."
|
|
user = "Existing continuity records:\n" + json.dumps(context, ensure_ascii=False) + f"\n\nCanonical scene ({scene.title}):\n{text}"
|
|
body = json.dumps({"model": OPENROUTER_MODEL, "messages": [{"role": "system", "content": system}, {"role": "user", "content": user}], "max_tokens": 2500, "temperature": 0, "response_format": {"type": "json_object"}}).encode()
|
|
request = urllib.request.Request("https://openrouter.ai/api/v1/chat/completions", data=body, headers={"Authorization": f"Bearer {key}", "Content-Type": "application/json"}, method="POST")
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=300) as response:
|
|
result = json.loads(response.read().decode())
|
|
except urllib.error.HTTPError as error:
|
|
raise ValueError(f"OpenRouter returned HTTP {error.code}: {error.read().decode(errors='replace')}") from error
|
|
choice = result.get("choices", [{}])[0]
|
|
if choice.get("finish_reason") == "length":
|
|
raise ValueError("OpenRouter stopped the continuity proposal at its output limit; nothing was saved. Please retry after restarting with the updated proposal budget.")
|
|
raw = choice.get("message", {}).get("content")
|
|
if not isinstance(raw, str):
|
|
raise ValueError("OpenRouter response did not contain continuity proposals.")
|
|
raw = re.sub(r"^```(?:json)?\s*|\s*```$", "", raw.strip(), flags=re.IGNORECASE)
|
|
# Frontier models occasionally add a one-line introduction despite the JSON-only
|
|
# instruction. Accept one complete object embedded in that formatting, but do
|
|
# not attempt to repair malformed JSON.
|
|
first_brace = raw.find("{")
|
|
last_brace = raw.rfind("}")
|
|
if first_brace >= 0 and last_brace > first_brace:
|
|
raw = raw[first_brace:last_brace + 1]
|
|
# Some otherwise valid model responses use Markdown-style escapes for
|
|
# underscores or double-escape quotes inside a JSON string. Neither is
|
|
# valid JSON; normalize only these two unambiguous formatting mistakes.
|
|
raw = re.sub(r"\\+_", "_", raw)
|
|
raw = re.sub(r'(?:\\){2,}(?=")', r"\\", raw)
|
|
try:
|
|
proposals = json.loads(raw)
|
|
except json.JSONDecodeError as error:
|
|
start = max(0, error.pos - 120)
|
|
end = min(len(raw), error.pos + 180)
|
|
context_excerpt = re.sub(r"\s+", " ", raw[start:end]).strip()
|
|
raise ValueError(f"OpenRouter returned proposals in an unexpected format near character {error.pos}: {context_excerpt!r}") from error
|
|
if not isinstance(proposals, dict):
|
|
raise ValueError("OpenRouter returned proposals in an unexpected format; please try again.")
|
|
valid_uuids = {character["uuid"] for character in data["characters"]}
|
|
updates = [
|
|
{"uuid": update.get("character_uuid", ""), "text": str(update.get("text", "")).strip()}
|
|
for update in proposals.get("character_updates", []) if isinstance(update, dict) and update.get("character_uuid") in valid_uuids and str(update.get("text", "")).strip()
|
|
]
|
|
existing_names = {character["title"].casefold() for character in data["characters"]}
|
|
new_characters = []
|
|
for character in proposals.get("new_characters", []):
|
|
if not isinstance(character, dict):
|
|
continue
|
|
name = str(character.get("name", "")).strip()
|
|
text = str(character.get("text", "")).strip()
|
|
if name and text and name.casefold() not in existing_names:
|
|
new_characters.append({"name": name, "text": text})
|
|
existing_names.add(name.casefold())
|
|
return {"character_updates": updates, "new_characters": new_characters, "lexicon_additions": str(proposals.get("lexicon_additions", "")).strip(), "world_note_additions": str(proposals.get("world_note_additions", "")).strip(), "model": result.get("model", OPENROUTER_MODEL)}
|
|
|
|
|
|
class AppHandler(SimpleHTTPRequestHandler):
|
|
reader: ScrivenerReader
|
|
|
|
def __init__(self, *args: object, **kwargs: object) -> None:
|
|
super().__init__(*args, directory=str(APP_DIR), **kwargs)
|
|
|
|
def send_json(self, value: object, status: HTTPStatus = HTTPStatus.OK) -> None:
|
|
body = json.dumps(value, ensure_ascii=False).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def do_GET(self) -> None:
|
|
path = urlparse(self.path).path
|
|
if path == "/api/project":
|
|
self.send_json(self.reader.summary())
|
|
return
|
|
if path == "/api/health":
|
|
self.send_json({"status": "ok", "mode": "local prototype"})
|
|
return
|
|
if path == "/":
|
|
self.path = "/static/index.html"
|
|
elif path.startswith("/static/"):
|
|
self.path = path
|
|
else:
|
|
self.send_error(HTTPStatus.NOT_FOUND)
|
|
return
|
|
super().do_GET()
|
|
|
|
def do_POST(self) -> None:
|
|
path = urlparse(self.path).path
|
|
if path not in {"/api/package-preview", "/api/draft", "/api/accept-for-editing", "/api/save-working-copy", "/api/accept-into-book", "/api/propose-summary", "/api/save-summary", "/api/propose-continuity-updates", "/api/save-continuity-updates", "/api/reconcile-bindings", "/api/save-brief", "/api/save-thread-link", "/api/create-scene"}:
|
|
self.send_error(HTTPStatus.NOT_FOUND)
|
|
return
|
|
try:
|
|
length = int(self.headers.get("Content-Length", "0"))
|
|
payload = json.loads(self.rfile.read(length).decode("utf-8"))
|
|
if path == "/api/draft":
|
|
self.send_json(draft(self.reader, payload))
|
|
elif path == "/api/accept-for-editing":
|
|
if not payload.get("accept_confirmation"):
|
|
raise ValueError("Confirm acceptance before creating archive records.")
|
|
package = build_package(self.reader, payload)
|
|
writer = ScrivenerWriter(self.reader.project)
|
|
result = writer.accept_for_editing(payload.get("scene_uuid", ""), payload.get("draft", ""), package, payload.get("model", OPENROUTER_MODEL), payload.get("character_uuids", []))
|
|
AppHandler.reader = writer.reader
|
|
self.send_json(result)
|
|
elif path == "/api/save-working-copy":
|
|
writer = ScrivenerWriter(self.reader.project)
|
|
writer.save_working_copy(payload.get("working_uuid", ""), payload.get("text", ""))
|
|
self.send_json({"saved": True})
|
|
elif path == "/api/accept-into-book":
|
|
if not payload.get("final_confirmation"):
|
|
raise ValueError("Confirm final acceptance before writing canonical Manuscript text.")
|
|
writer = ScrivenerWriter(self.reader.project)
|
|
writer.accept_into_book(payload.get("scene_uuid", ""), payload.get("text", ""), payload.get("predecessor_uuid"))
|
|
AppHandler.reader = writer.reader
|
|
self.send_json({"accepted": True})
|
|
elif path == "/api/propose-summary":
|
|
self.send_json(propose_summary(self.reader, payload))
|
|
elif path == "/api/save-summary":
|
|
if not payload.get("summary_confirmation"):
|
|
raise ValueError("Confirm the summary before saving it to Scrivener.")
|
|
writer = ScrivenerWriter(self.reader.project)
|
|
summary_uuid = writer.save_scene_summary(payload.get("scene_uuid", ""), payload.get("summary", ""))
|
|
AppHandler.reader = writer.reader
|
|
self.send_json({"saved": True, "summary_uuid": summary_uuid})
|
|
elif path == "/api/propose-continuity-updates":
|
|
self.send_json(propose_continuity_updates(self.reader, payload))
|
|
elif path == "/api/save-continuity-updates":
|
|
if not payload.get("updates_confirmation"):
|
|
raise ValueError("Confirm the proposed continuity updates before saving them to Scrivener.")
|
|
writer = ScrivenerWriter(self.reader.project)
|
|
result = writer.save_continuity_updates(payload.get("scene_uuid", ""), payload.get("character_updates", []), payload.get("new_characters", []), payload.get("lexicon_additions", ""), payload.get("world_note_additions", ""))
|
|
AppHandler.reader = ScrivenerReader(self.reader.project)
|
|
self.send_json(result)
|
|
elif path == "/api/reconcile-bindings":
|
|
writer = ScrivenerWriter(self.reader.project)
|
|
counts = writer.reconcile_scene_bindings()
|
|
AppHandler.reader = writer.reader
|
|
self.send_json({"binding_counts": counts})
|
|
elif path == "/api/save-brief":
|
|
writer = ScrivenerWriter(self.reader.project)
|
|
result = writer.save_brief(payload.get("scene_uuid", ""), payload.get("text", ""))
|
|
AppHandler.reader = writer.reader
|
|
self.send_json(result)
|
|
elif path == "/api/save-thread-link":
|
|
writer = ScrivenerWriter(self.reader.project)
|
|
writer.assign_thread_predecessor(payload.get("scene_uuid", ""), payload.get("predecessor_uuid"))
|
|
AppHandler.reader = writer.reader
|
|
self.send_json({"saved": True})
|
|
elif path == "/api/create-scene":
|
|
writer = ScrivenerWriter(self.reader.project)
|
|
result = writer.create_planned_scene(payload.get("chapter_uuid", ""), payload.get("title", ""))
|
|
AppHandler.reader = writer.reader
|
|
self.send_json(result)
|
|
else:
|
|
self.send_json({"package": build_package(self.reader, payload), "warnings": preflight(self.reader, payload)})
|
|
except (ValueError, json.JSONDecodeError) as error:
|
|
self.send_json({"error": str(error)}, HTTPStatus.BAD_REQUEST)
|
|
|
|
def log_message(self, format: str, *args: object) -> None:
|
|
print("[authoring prototype] " + format % args)
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Run the local authoring-system prototype.")
|
|
parser.add_argument("--project", type=Path, default=DEFAULT_PROJECT)
|
|
parser.add_argument("--port", type=int, default=8765)
|
|
args = parser.parse_args()
|
|
project = args.project.expanduser().resolve()
|
|
if not project.is_dir():
|
|
parser.error(f"Project not found: {project}")
|
|
AppHandler.reader = ScrivenerReader(project)
|
|
server = ThreadingHTTPServer(("127.0.0.1", args.port), AppHandler)
|
|
print(f"Local authoring prototype: http://127.0.0.1:{args.port}")
|
|
print(f"Project: {project}")
|
|
server.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|