286 lines
11 KiB
Python
286 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Read-only local package-preview prototype for a Scrivener project.
|
|
|
|
This deliberately has no model call and no Scrivener write path. It is a
|
|
small standard-library server so the first workflow test has no dependency
|
|
installation step.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
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"
|
|
|
|
|
|
@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()
|
|
|
|
|
|
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")
|
|
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)}
|
|
|
|
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", ""),
|
|
}
|
|
)
|
|
|
|
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],
|
|
}
|
|
|
|
|
|
def section(name: str, value: str) -> str:
|
|
return f"### {name}\n{value.strip() or '(none)'}"
|
|
|
|
|
|
def build_package(project: ScrivenerReader, payload: dict) -> str:
|
|
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"
|
|
|
|
|
|
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": "read-only 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:
|
|
if urlparse(self.path).path != "/api/package-preview":
|
|
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"))
|
|
self.send_json({"package": build_package(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 read-only 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"Read-only authoring prototype: http://127.0.0.1:{args.port}")
|
|
print(f"Project: {project}")
|
|
server.serve_forever()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|