667 lines
34 KiB
Python
667 lines
34 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
|
|
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 _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 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")
|
|
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}
|
|
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)}
|
|
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]
|
|
package_context[title] = {"character_uuids": character_uuids, "notes": self._package_section(text, "NOTES")}
|
|
|
|
paired = []
|
|
for scene in manuscript:
|
|
paired.append(
|
|
{
|
|
**scene,
|
|
"brief": brief_by_title.get(scene["title"], {}).get("text", ""),
|
|
"brief_uuid": brief_by_title.get(scene["title"], {}).get("uuid"),
|
|
"summary": summary_by_title.get(scene["title"], {}).get("text", ""),
|
|
"package_context": package_context.get(scene["title"], {"character_uuids": [], "notes": ""}),
|
|
}
|
|
)
|
|
|
|
manuscript_titles = {scene["title"] for scene in manuscript}
|
|
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),
|
|
"world_notes": self.content(world_notes),
|
|
"scenes": paired,
|
|
"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["title"] not in manuscript_titles],
|
|
"unmatched_scenes": [scene for scene in manuscript if scene["title"] not in brief_by_title],
|
|
}
|
|
|
|
@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 _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
|
|
|
|
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 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) -> 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)
|
|
|
|
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 save_continuity_updates(self, scene_uuid: str, character_updates: 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.")
|
|
stamp = f"Continuity update from {scene.title}:"
|
|
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._append_to_item(item, f"{stamp}\n{update.get('text', '').strip()}"):
|
|
saved_characters.append(item.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
|
|
return {"characters": saved_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 JSON only, with exactly these keys: character_updates (array of objects with character_uuid 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. Character text must be a concise factual update, without a heading. Use a supplied UUID only when updating an existing character; do not create characters here. 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": 1200}).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
|
|
raw = result.get("choices", [{}])[0].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)
|
|
try:
|
|
proposals = json.loads(raw)
|
|
except json.JSONDecodeError as error:
|
|
raise ValueError("OpenRouter returned proposals in an unexpected format; please try again.") from error
|
|
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()
|
|
]
|
|
return {"character_updates": updates, "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"}:
|
|
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", ""))
|
|
AppHandler.reader = ScrivenerReader(self.reader.project)
|
|
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("lexicon_additions", ""), payload.get("world_note_additions", ""))
|
|
AppHandler.reader = ScrivenerReader(self.reader.project)
|
|
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()
|