xin61
1345 lines
48 KiB
Python
1345 lines
48 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import copy
|
|
import datetime as dt
|
|
import hashlib
|
|
import html
|
|
import json
|
|
import os
|
|
import posixpath
|
|
import re
|
|
import shutil
|
|
import signal
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import urllib.parse
|
|
import urllib.request
|
|
import webbrowser
|
|
import zipfile
|
|
from html.parser import HTMLParser
|
|
from pathlib import Path
|
|
from typing import Any
|
|
import xml.etree.ElementTree as ET
|
|
|
|
from flask import Flask, jsonify, request, send_from_directory
|
|
|
|
|
|
P_RE = re.compile(r"(<p\b[^>]*>)(.*?)(</p>)", re.S | re.I)
|
|
TAG_RE = re.compile(r"<[^>]+>")
|
|
JP_RE = re.compile(r"[\u3040-\u30ff\u3400-\u9fff々〆〤]")
|
|
KANA_RE = re.compile(r"[\u3040-\u30ff]")
|
|
GRAY_RE = re.compile(
|
|
r"(?:color\s*:\s*(?:#?808080|gray|grey|rgb\(\s*128\s*,\s*128\s*,\s*128\s*\))|class\s*=\s*['\"][^'\"]*(?:gray|grey|jp|source)[^'\"]*['\"])",
|
|
re.I,
|
|
)
|
|
|
|
DEFAULT_REVIEW_ROOT = "epub_review_sessions"
|
|
STATE_VERSION = 1
|
|
UPLOAD_DIR_NAME = "uploads"
|
|
|
|
|
|
def now_iso() -> str:
|
|
return dt.datetime.now(dt.timezone.utc).astimezone().isoformat(timespec="seconds")
|
|
|
|
|
|
def safe_slug(value: str, max_len: int = 80) -> str:
|
|
value = re.sub(r'[\\/:*?"<>|]+', "_", value).strip().strip(".")
|
|
value = re.sub(r"\s+", "_", value)
|
|
return value[:max_len] or "epub"
|
|
|
|
|
|
def find_free_port(start: int, host: str = "127.0.0.1") -> int:
|
|
for port in range(start, start + 100):
|
|
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
|
|
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
try:
|
|
sock.bind((host, port))
|
|
except OSError:
|
|
continue
|
|
return port
|
|
raise RuntimeError(f"no free port found from {start}")
|
|
|
|
|
|
def display_host(host: str) -> str:
|
|
return "localhost" if host in {"127.0.0.1", "0.0.0.0", "::"} else host
|
|
|
|
|
|
def read_json(path: Path, default: Any) -> Any:
|
|
if not path.exists():
|
|
return copy.deepcopy(default)
|
|
return json.loads(path.read_text(encoding="utf-8-sig"))
|
|
|
|
|
|
def write_json(path: Path, data: Any) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
|
|
|
|
|
def append_jsonl(path: Path, record: dict[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
with path.open("a", encoding="utf-8", newline="\n") as fh:
|
|
fh.write(json.dumps(record, ensure_ascii=False) + "\n")
|
|
|
|
|
|
def strip_tags(fragment: str) -> str:
|
|
fragment = re.sub(r"<rt\b[^>]*>.*?</rt>", "", fragment, flags=re.S | re.I)
|
|
return html.unescape(TAG_RE.sub("", fragment)).strip()
|
|
|
|
|
|
def compact_text(value: str) -> str:
|
|
return re.sub(r"\s+", " ", html.unescape(value or "")).strip()
|
|
|
|
|
|
def file_sort_key(value: str) -> list[Any]:
|
|
parts = re.split(r"(\d+)", value.lower())
|
|
return [int(part) if part.isdigit() else part for part in parts]
|
|
|
|
|
|
def entry_dir(entry_name: str) -> str:
|
|
directory = posixpath.dirname(entry_name)
|
|
return f"{directory}/" if directory else ""
|
|
|
|
|
|
def normalize_href(base_dir: str, href: str) -> str:
|
|
href = urllib.parse.unquote((href or "").split("#", 1)[0])
|
|
if not href:
|
|
return ""
|
|
return posixpath.normpath(posixpath.join(base_dir, href)).lstrip("./")
|
|
|
|
|
|
def read_xml_root(path: Path) -> ET.Element | None:
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
return ET.fromstring(path.read_text(encoding="utf-8", errors="replace"))
|
|
except ET.ParseError:
|
|
try:
|
|
return ET.fromstring(path.read_bytes())
|
|
except ET.ParseError:
|
|
return None
|
|
|
|
|
|
def local_name(tag: str) -> str:
|
|
return tag.rsplit("}", 1)[-1].lower()
|
|
|
|
|
|
def element_text(elem: ET.Element) -> str:
|
|
return compact_text("".join(elem.itertext()))
|
|
|
|
|
|
def is_japaneseish_fragment(fragment: str) -> bool:
|
|
return bool(KANA_RE.search(fragment or ""))
|
|
|
|
|
|
def is_japaneseish(fragment: str) -> bool:
|
|
text = strip_tags(fragment)
|
|
return bool(KANA_RE.search(text))
|
|
|
|
|
|
def is_gray_source(attrs: str, inner: str) -> bool:
|
|
if GRAY_RE.search(attrs):
|
|
return True
|
|
return is_japaneseish(inner)
|
|
|
|
|
|
class HtmlSanitizer(HTMLParser):
|
|
allowed_tags = {
|
|
"a",
|
|
"b",
|
|
"br",
|
|
"code",
|
|
"em",
|
|
"i",
|
|
"mark",
|
|
"rb",
|
|
"rp",
|
|
"rt",
|
|
"ruby",
|
|
"small",
|
|
"span",
|
|
"strong",
|
|
"sub",
|
|
"sup",
|
|
}
|
|
allowed_attrs = {
|
|
"a": {"href", "title"},
|
|
"mark": {"class", "data-review-mark"},
|
|
"span": {"class", "style", "data-review-mark"},
|
|
"ruby": {"class"},
|
|
"rt": {"class"},
|
|
"rb": {"class"},
|
|
"rp": {"class"},
|
|
}
|
|
void_tags = {"br"}
|
|
style_safe = re.compile(r"^(?:\s*(?:color|background-color|font-weight|font-style)\s*:\s*[-#(),.%\w\s]+\s*;?)*$", re.I)
|
|
|
|
def __init__(self) -> None:
|
|
super().__init__(convert_charrefs=False)
|
|
self.parts: list[str] = []
|
|
|
|
def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
|
|
tag = tag.lower()
|
|
if tag not in self.allowed_tags:
|
|
return
|
|
safe_attrs = []
|
|
for key, value in attrs:
|
|
key = key.lower()
|
|
value = value or ""
|
|
if key not in self.allowed_attrs.get(tag, set()):
|
|
continue
|
|
if key.startswith("on"):
|
|
continue
|
|
if key == "href" and re.match(r"\s*(?:javascript|data):", value, re.I):
|
|
continue
|
|
if key == "style" and not self.style_safe.fullmatch(value):
|
|
continue
|
|
safe_attrs.append(f'{key}="{html.escape(value, quote=True)}"')
|
|
attr_text = (" " + " ".join(safe_attrs)) if safe_attrs else ""
|
|
if tag in self.void_tags:
|
|
self.parts.append(f"<{tag}{attr_text}/>")
|
|
else:
|
|
self.parts.append(f"<{tag}{attr_text}>")
|
|
|
|
def handle_endtag(self, tag: str) -> None:
|
|
tag = tag.lower()
|
|
if tag in self.allowed_tags and tag not in self.void_tags:
|
|
self.parts.append(f"</{tag}>")
|
|
|
|
def handle_data(self, data: str) -> None:
|
|
self.parts.append(html.escape(data, quote=False))
|
|
|
|
def handle_entityref(self, name: str) -> None:
|
|
self.parts.append(f"&{name};")
|
|
|
|
def handle_charref(self, name: str) -> None:
|
|
self.parts.append(f"&#{name};")
|
|
|
|
|
|
def sanitize_cn_html(value: str) -> str:
|
|
value = re.sub(r"(?is)<\s*(script|style|iframe|object|embed)\b.*?</\s*\1\s*>", "", value)
|
|
sanitizer = HtmlSanitizer()
|
|
sanitizer.feed(value)
|
|
sanitizer.close()
|
|
return "".join(sanitizer.parts).strip()
|
|
|
|
|
|
def session_root_for(review_root: Path, epub_path: Path) -> Path:
|
|
digest = hashlib.sha1(str(epub_path.resolve()).encode("utf-8")).hexdigest()[:10]
|
|
return review_root / f"{safe_slug(epub_path.stem)}_{digest}"
|
|
|
|
|
|
def app_manifest_path(review_root: Path) -> Path:
|
|
return review_root / "app_state.json"
|
|
|
|
|
|
def session_manifest_path(session_root: Path) -> Path:
|
|
return session_root / "review_state" / "session.json"
|
|
|
|
|
|
def session_public_id(session_root: Path) -> str:
|
|
return session_root.name
|
|
|
|
|
|
def session_root_from_id(review_root: Path, session_id: str) -> Path:
|
|
safe_id = safe_slug(session_id, max_len=160)
|
|
if safe_id != session_id:
|
|
raise ValueError("invalid session id")
|
|
root = (review_root / session_id).resolve()
|
|
try:
|
|
root.relative_to(review_root.resolve())
|
|
except ValueError as exc:
|
|
raise ValueError("invalid session id") from exc
|
|
return root
|
|
|
|
|
|
def write_session_manifest(session_root: Path, epub_path: Path, source_name: str = "") -> None:
|
|
existing = read_json(session_manifest_path(session_root), {})
|
|
record = {
|
|
"id": session_public_id(session_root),
|
|
"source_name": source_name or epub_path.name,
|
|
"source_epub": str(epub_path),
|
|
"session_root": str(session_root),
|
|
"created_at": existing.get("created_at") or now_iso(),
|
|
"updated_at": now_iso(),
|
|
}
|
|
write_json(session_manifest_path(session_root), record)
|
|
|
|
|
|
def update_app_manifest(review_root: Path, session_root: Path | None) -> None:
|
|
data = read_json(app_manifest_path(review_root), {})
|
|
data["active_session_id"] = session_public_id(session_root) if session_root else ""
|
|
data["updated_at"] = now_iso()
|
|
write_json(app_manifest_path(review_root), data)
|
|
|
|
|
|
def session_epub_path(session_root: Path) -> Path:
|
|
manifest = read_json(session_manifest_path(session_root), {})
|
|
source_epub = manifest.get("source_epub") or read_json(
|
|
session_root / "review_state" / "state.json", {}
|
|
).get("source_epub")
|
|
if source_epub:
|
|
path = Path(source_epub)
|
|
if path.exists():
|
|
return path
|
|
fallback = session_root / "source.epub"
|
|
if fallback.exists():
|
|
return fallback
|
|
return Path(source_epub or fallback)
|
|
|
|
|
|
def summarize_session(session_root: Path) -> dict[str, Any]:
|
|
manifest = read_json(session_manifest_path(session_root), {})
|
|
state = read_json(session_root / "review_state" / "state.json", {})
|
|
rows = merge_rows(session_root) if (session_root / "review_state" / "rows.json").exists() else []
|
|
touched = [row for row in rows if rowTouched_py(row)]
|
|
source_epub = manifest.get("source_epub") or state.get("source_epub", "")
|
|
source_name = manifest.get("source_name") or (Path(source_epub).name if source_epub else session_root.name)
|
|
exports_dir = session_root / "exports"
|
|
latest_export = ""
|
|
if exports_dir.exists():
|
|
exports = sorted(exports_dir.glob("*.epub"), key=lambda p: p.stat().st_mtime, reverse=True)
|
|
latest_export = str(exports[0]) if exports else ""
|
|
updated_at = state.get("updated_at") or manifest.get("updated_at") or ""
|
|
return {
|
|
"id": session_public_id(session_root),
|
|
"source_name": source_name,
|
|
"source_epub": source_epub,
|
|
"session_root": str(session_root),
|
|
"row_count": len(rows),
|
|
"touched_count": len(touched),
|
|
"marked_count": sum(1 for row in rows if row.get("marked")),
|
|
"updated_at": updated_at,
|
|
"created_at": state.get("created_at") or manifest.get("created_at") or "",
|
|
"feedback_md": str(session_root / "feedback" / "feedback_for_codex.md"),
|
|
"feedback_jsonl": str(session_root / "feedback" / "translation_feedback.jsonl"),
|
|
"latest_export": latest_export,
|
|
}
|
|
|
|
|
|
def rowTouched_py(row: dict[str, Any]) -> bool:
|
|
return bool(
|
|
row.get("edited")
|
|
or row.get("marked")
|
|
or row.get("comment")
|
|
or row.get("learn_note")
|
|
or row.get("issue_type")
|
|
or row.get("severity")
|
|
or row.get("tags")
|
|
)
|
|
|
|
|
|
def list_review_sessions(review_root: Path) -> list[dict[str, Any]]:
|
|
if not review_root.exists():
|
|
return []
|
|
sessions = []
|
|
for child in review_root.iterdir():
|
|
if not child.is_dir() or child.name == UPLOAD_DIR_NAME:
|
|
continue
|
|
if not (child / "review_state" / "state.json").exists():
|
|
continue
|
|
try:
|
|
sessions.append(summarize_session(child))
|
|
except Exception:
|
|
continue
|
|
sessions.sort(key=lambda item: item.get("updated_at") or item.get("created_at") or "", reverse=True)
|
|
return sessions
|
|
|
|
|
|
def validate_epub_file(path: Path) -> None:
|
|
if path.suffix.lower() != ".epub":
|
|
raise ValueError("只支持 .epub 文件")
|
|
if not zipfile.is_zipfile(path):
|
|
raise ValueError("文件不是有效的 EPUB/ZIP 包")
|
|
|
|
|
|
def copy_upload_to_review_root(file_storage: Any, review_root: Path) -> Path:
|
|
filename = safe_slug(Path(file_storage.filename or "uploaded.epub").name, max_len=140)
|
|
if not filename.lower().endswith(".epub"):
|
|
filename = f"{filename}.epub"
|
|
upload_dir = review_root / UPLOAD_DIR_NAME
|
|
upload_dir.mkdir(parents=True, exist_ok=True)
|
|
stem = safe_slug(Path(filename).stem, max_len=100)
|
|
target = upload_dir / f"{dt.datetime.now().strftime('%Y%m%d_%H%M%S')}_{stem}.epub"
|
|
try:
|
|
file_storage.save(target)
|
|
validate_epub_file(target)
|
|
return target
|
|
except Exception:
|
|
target.unlink(missing_ok=True)
|
|
raise
|
|
|
|
|
|
def extracted_path(session_root: Path, entry_name: str) -> Path:
|
|
root = (session_root / "extracted").resolve()
|
|
target = (root / entry_name).resolve()
|
|
try:
|
|
target.relative_to(root)
|
|
except ValueError as exc:
|
|
raise ValueError(f"unsafe epub entry path: {entry_name}") from exc
|
|
return target
|
|
|
|
|
|
def extract_epub(epub_path: Path, session_root: Path, reset: bool = False) -> None:
|
|
extracted = session_root / "extracted"
|
|
if reset and session_root.exists():
|
|
shutil.rmtree(session_root)
|
|
if extracted.exists():
|
|
return
|
|
session_root.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(epub_path, session_root / "source.epub")
|
|
extracted.mkdir(parents=True, exist_ok=True)
|
|
entries = []
|
|
with zipfile.ZipFile(epub_path, "r") as zf:
|
|
for info in zf.infolist():
|
|
entries.append(
|
|
{
|
|
"filename": info.filename,
|
|
"compress_type": int(info.compress_type),
|
|
"date_time": list(info.date_time),
|
|
"is_dir": info.is_dir(),
|
|
}
|
|
)
|
|
target = extracted_path(session_root, info.filename)
|
|
if info.is_dir():
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
continue
|
|
target.parent.mkdir(parents=True, exist_ok=True)
|
|
target.write_bytes(zf.read(info.filename))
|
|
write_json(session_root / "review_state" / "epub_entries.json", entries)
|
|
|
|
|
|
def find_opf_entry(session_root: Path) -> str:
|
|
extracted = session_root / "extracted"
|
|
container = extracted / "META-INF" / "container.xml"
|
|
root = read_xml_root(container)
|
|
if root is not None:
|
|
for elem in root.iter():
|
|
if local_name(elem.tag) == "rootfile":
|
|
full_path = elem.attrib.get("full-path") or elem.attrib.get("full_path")
|
|
if full_path:
|
|
return posixpath.normpath(full_path).lstrip("./")
|
|
entries = read_json(session_root / "review_state" / "epub_entries.json", [])
|
|
for item in entries:
|
|
name = item.get("filename", "")
|
|
if not item.get("is_dir") and name.lower().endswith(".opf"):
|
|
return name
|
|
return ""
|
|
|
|
|
|
def parse_opf_metadata(session_root: Path) -> dict[str, Any]:
|
|
opf_entry = find_opf_entry(session_root)
|
|
if not opf_entry:
|
|
return {"opf_entry": "", "opf_dir": "", "spine": []}
|
|
opf_path = extracted_path(session_root, opf_entry)
|
|
root = read_xml_root(opf_path)
|
|
if root is None:
|
|
return {"opf_entry": opf_entry, "opf_dir": entry_dir(opf_entry), "spine": []}
|
|
|
|
opf_dir = entry_dir(opf_entry)
|
|
manifest: dict[str, dict[str, str]] = {}
|
|
spine_idrefs: list[str] = []
|
|
for elem in root.iter():
|
|
name = local_name(elem.tag)
|
|
if name == "item":
|
|
item_id = elem.attrib.get("id", "")
|
|
href = elem.attrib.get("href", "")
|
|
if item_id and href:
|
|
manifest[item_id] = {
|
|
"href": normalize_href(opf_dir, href),
|
|
"media_type": elem.attrib.get("media-type", ""),
|
|
"properties": elem.attrib.get("properties", ""),
|
|
}
|
|
elif name == "itemref":
|
|
idref = elem.attrib.get("idref", "")
|
|
if idref:
|
|
spine_idrefs.append(idref)
|
|
|
|
spine = []
|
|
for idref in spine_idrefs:
|
|
item = manifest.get(idref)
|
|
if not item:
|
|
continue
|
|
href = item["href"]
|
|
if href.lower().endswith((".xhtml", ".html", ".htm")):
|
|
spine.append(href)
|
|
return {
|
|
"opf_entry": opf_entry,
|
|
"opf_dir": opf_dir,
|
|
"manifest": manifest,
|
|
"spine": spine,
|
|
}
|
|
|
|
|
|
def html_entries_in_reading_order(session_root: Path) -> list[str]:
|
|
entries = read_json(session_root / "review_state" / "epub_entries.json", [])
|
|
all_html = [
|
|
item["filename"]
|
|
for item in entries
|
|
if not item.get("is_dir")
|
|
and item["filename"].lower().endswith((".xhtml", ".html", ".htm"))
|
|
]
|
|
opf = parse_opf_metadata(session_root)
|
|
ordered: list[str] = []
|
|
for item in opf.get("spine", []):
|
|
if item in all_html and item not in ordered:
|
|
ordered.append(item)
|
|
for item in sorted(all_html, key=file_sort_key):
|
|
basename = posixpath.basename(item).lower()
|
|
if item in ordered or basename in {"nav.xhtml", "toc.xhtml"}:
|
|
continue
|
|
ordered.append(item)
|
|
return ordered
|
|
|
|
|
|
def extract_heading_from_html(text: str) -> str:
|
|
for pattern in (
|
|
r"<h[1-6]\b[^>]*>(.*?)</h[1-6]>",
|
|
r"<title\b[^>]*>(.*?)</title>",
|
|
):
|
|
match = re.search(pattern, text, flags=re.S | re.I)
|
|
if not match:
|
|
continue
|
|
title = strip_tags(match.group(1))
|
|
if title:
|
|
return title
|
|
return ""
|
|
|
|
|
|
def read_document_title(session_root: Path, entry_name: str) -> str:
|
|
path = extracted_path(session_root, entry_name)
|
|
if not path.exists():
|
|
return ""
|
|
text = path.read_text(encoding="utf-8", errors="replace")
|
|
return extract_heading_from_html(text)
|
|
|
|
|
|
def parse_nav_toc(session_root: Path) -> list[dict[str, Any]]:
|
|
entries = read_json(session_root / "review_state" / "epub_entries.json", [])
|
|
nav_entries = [
|
|
item["filename"]
|
|
for item in entries
|
|
if not item.get("is_dir")
|
|
and posixpath.basename(item["filename"]).lower() in {"nav.xhtml", "toc.xhtml", "toc.html", "toc.htm"}
|
|
]
|
|
opf = parse_opf_metadata(session_root)
|
|
for manifest_item in opf.get("manifest", {}).values():
|
|
if "nav" in manifest_item.get("properties", "").split():
|
|
href = manifest_item.get("href", "")
|
|
if href and href not in nav_entries:
|
|
nav_entries.insert(0, href)
|
|
|
|
for entry_name in nav_entries:
|
|
path = extracted_path(session_root, entry_name)
|
|
root = read_xml_root(path)
|
|
if root is None:
|
|
continue
|
|
base_dir = entry_dir(entry_name)
|
|
toc_root: ET.Element | None = None
|
|
for elem in root.iter():
|
|
if local_name(elem.tag) != "nav":
|
|
continue
|
|
nav_type = elem.attrib.get("{http://www.idpf.org/2007/ops}type", "") or elem.attrib.get("epub:type", "")
|
|
if "toc" in nav_type.split() or toc_root is None:
|
|
toc_root = elem
|
|
if "toc" in nav_type.split():
|
|
break
|
|
if toc_root is None:
|
|
continue
|
|
top_ol = next((child for child in list(toc_root) if local_name(child.tag) == "ol"), None)
|
|
if top_ol is None:
|
|
top_ol = next((elem for elem in toc_root.iter() if local_name(elem.tag) == "ol"), None)
|
|
if top_ol is None:
|
|
continue
|
|
items = parse_nav_ol(top_ol, base_dir)
|
|
if items:
|
|
return items
|
|
return []
|
|
|
|
|
|
def parse_nav_ol(ol: ET.Element, base_dir: str) -> list[dict[str, Any]]:
|
|
items: list[dict[str, Any]] = []
|
|
for li in list(ol):
|
|
if local_name(li.tag) != "li":
|
|
continue
|
|
label = ""
|
|
href = ""
|
|
children: list[dict[str, Any]] = []
|
|
for child in list(li):
|
|
name = local_name(child.tag)
|
|
if name == "a":
|
|
label = element_text(child)
|
|
href = normalize_href(base_dir, child.attrib.get("href", ""))
|
|
elif name == "span" and not label:
|
|
label = element_text(child)
|
|
elif name == "ol":
|
|
children = parse_nav_ol(child, base_dir)
|
|
if not label:
|
|
label = element_text(li)
|
|
items.append({"title": label, "href": href, "children": children})
|
|
return items
|
|
|
|
|
|
def flatten_nav_items(items: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
flattened: list[dict[str, Any]] = []
|
|
|
|
def walk(nodes: list[dict[str, Any]], depth: int) -> None:
|
|
for node in nodes:
|
|
href = node.get("href", "")
|
|
if href:
|
|
flattened.append({"title": node.get("title") or href, "href": href, "depth": depth})
|
|
walk(node.get("children", []), depth + 1)
|
|
|
|
walk(items, 0)
|
|
return flattened
|
|
|
|
|
|
def build_structure(session_root: Path) -> dict[str, Any]:
|
|
rows = merge_rows(session_root)
|
|
rows_by_file: dict[str, list[dict[str, Any]]] = {}
|
|
for row in rows:
|
|
rows_by_file.setdefault(row["file"], []).append(row)
|
|
|
|
ordered_files = [file for file in html_entries_in_reading_order(session_root) if file in rows_by_file]
|
|
for file in rows_by_file:
|
|
if file not in ordered_files:
|
|
ordered_files.append(file)
|
|
|
|
nav_tree = parse_nav_toc(session_root)
|
|
nav_flat = flatten_nav_items(nav_tree)
|
|
nav_titles: dict[str, str] = {}
|
|
for item in nav_flat:
|
|
href = item.get("href", "")
|
|
title = item.get("title", "")
|
|
if href and title and href not in nav_titles:
|
|
nav_titles[href] = title
|
|
|
|
part_serial = 0
|
|
|
|
def file_counts(file_rows: list[dict[str, Any]]) -> tuple[int, int]:
|
|
touched_count = sum(
|
|
1
|
|
for row in file_rows
|
|
if row["edited"]
|
|
or row["marked"]
|
|
or row["comment"]
|
|
or row["learn_note"]
|
|
or row["issue_type"]
|
|
or row["severity"]
|
|
or row["tags"]
|
|
)
|
|
marked_count = sum(1 for row in file_rows if row["marked"])
|
|
return touched_count, marked_count
|
|
|
|
def make_part(entry_name: str, title_override: str = "") -> dict[str, Any]:
|
|
nonlocal part_serial
|
|
part_serial += 1
|
|
file_rows = rows_by_file[entry_name]
|
|
title = (
|
|
title_override
|
|
or nav_titles.get(entry_name)
|
|
or read_document_title(session_root, entry_name)
|
|
or Path(entry_name).stem
|
|
)
|
|
touched_count, marked_count = file_counts(file_rows)
|
|
return {
|
|
"id": f"PT{part_serial:04d}",
|
|
"title": title,
|
|
"file": entry_name,
|
|
"file_label": Path(entry_name).name,
|
|
"row_ids": [row["id"] for row in file_rows],
|
|
"row_count": len(file_rows),
|
|
"touched_count": touched_count,
|
|
"marked_count": marked_count,
|
|
"first_row_id": file_rows[0]["id"] if file_rows else "",
|
|
"last_row_id": file_rows[-1]["id"] if file_rows else "",
|
|
}
|
|
|
|
def chapter_from_parts(chapter_id: str, title: str, parts: list[dict[str, Any]]) -> dict[str, Any]:
|
|
return {
|
|
"id": chapter_id,
|
|
"title": title,
|
|
"row_count": sum(part["row_count"] for part in parts),
|
|
"touched_count": sum(part["touched_count"] for part in parts),
|
|
"marked_count": sum(part["marked_count"] for part in parts),
|
|
"parts": parts,
|
|
}
|
|
|
|
def flatten_node_files(node: dict[str, Any]) -> list[dict[str, str]]:
|
|
items: list[dict[str, str]] = []
|
|
href = node.get("href", "")
|
|
if href in rows_by_file:
|
|
items.append({"href": href, "title": node.get("title") or ""})
|
|
for child in node.get("children", []):
|
|
items.extend(flatten_node_files(child))
|
|
return items
|
|
|
|
chapters: list[dict[str, Any]] = []
|
|
assigned_files: set[str] = set()
|
|
chapter_serial = 0
|
|
|
|
chapter_nodes = nav_tree
|
|
while (
|
|
len(chapter_nodes) == 1
|
|
and chapter_nodes[0].get("href") not in rows_by_file
|
|
and chapter_nodes[0].get("children")
|
|
):
|
|
chapter_nodes = chapter_nodes[0]["children"]
|
|
|
|
for node in chapter_nodes:
|
|
node_files = []
|
|
seen_in_node: set[str] = set()
|
|
for item in flatten_node_files(node):
|
|
href = item["href"]
|
|
if href in assigned_files or href in seen_in_node:
|
|
continue
|
|
seen_in_node.add(href)
|
|
node_files.append(item)
|
|
if not node_files:
|
|
continue
|
|
chapter_serial += 1
|
|
parts = [make_part(item["href"], item.get("title", "")) for item in node_files]
|
|
assigned_files.update(item["href"] for item in node_files)
|
|
chapter_title = node.get("title") or parts[0]["title"]
|
|
chapters.append(chapter_from_parts(f"C{chapter_serial:04d}", chapter_title, parts))
|
|
|
|
for entry_name in ordered_files:
|
|
if entry_name in assigned_files:
|
|
continue
|
|
chapter_serial += 1
|
|
part = make_part(entry_name)
|
|
chapters.append(chapter_from_parts(f"C{chapter_serial:04d}", part["title"], [part]))
|
|
|
|
return {
|
|
"chapters": chapters,
|
|
"chapter_count": len(chapters),
|
|
"part_count": sum(len(chapter["parts"]) for chapter in chapters),
|
|
"row_count": len(rows),
|
|
}
|
|
|
|
|
|
def build_rows(session_root: Path) -> list[dict[str, Any]]:
|
|
html_entries = html_entries_in_reading_order(session_root)
|
|
rows: list[dict[str, Any]] = []
|
|
serial = 0
|
|
for file_order, entry_name in enumerate(html_entries, start=1):
|
|
path = extracted_path(session_root, entry_name)
|
|
if not path.exists():
|
|
continue
|
|
document_title = read_document_title(session_root, entry_name) or Path(entry_name).stem
|
|
text = path.read_text(encoding="utf-8", errors="replace")
|
|
matches = list(P_RE.finditer(text))
|
|
i = 0
|
|
file_serial = 0
|
|
while i < len(matches) - 1:
|
|
open_tag = matches[i].group(1)
|
|
attrs = open_tag[2:-1]
|
|
jp_inner = matches[i].group(2)
|
|
cn_inner = matches[i + 1].group(2)
|
|
next_open = matches[i + 1].group(1)
|
|
next_attrs = next_open[2:-1]
|
|
if is_gray_source(attrs, jp_inner) and not is_gray_source(next_attrs, cn_inner):
|
|
serial += 1
|
|
file_serial += 1
|
|
rows.append(
|
|
{
|
|
"id": f"R{serial:05d}",
|
|
"file": entry_name,
|
|
"file_order": file_order,
|
|
"file_row_index": file_serial,
|
|
"file_label": Path(entry_name).name,
|
|
"document_title": document_title,
|
|
"ja_p_index": i,
|
|
"cn_p_index": i + 1,
|
|
"jp_html": jp_inner,
|
|
"jp_text": strip_tags(jp_inner),
|
|
"cn_html": cn_inner,
|
|
"cn_text": strip_tags(cn_inner),
|
|
}
|
|
)
|
|
i += 2
|
|
else:
|
|
i += 1
|
|
write_json(session_root / "review_state" / "rows.json", rows)
|
|
return rows
|
|
|
|
|
|
def ensure_state(session_root: Path, epub_path: Path, reset: bool = False) -> dict[str, Any]:
|
|
extract_epub(epub_path, session_root, reset=reset)
|
|
rows_path = session_root / "review_state" / "rows.json"
|
|
rows = read_json(rows_path, None)
|
|
if rows is None:
|
|
rows = build_rows(session_root)
|
|
state_path = session_root / "review_state" / "state.json"
|
|
state = read_json(
|
|
state_path,
|
|
{
|
|
"version": STATE_VERSION,
|
|
"created_at": now_iso(),
|
|
"updated_at": now_iso(),
|
|
"source_epub": str(epub_path),
|
|
"edits": {},
|
|
},
|
|
)
|
|
state.setdefault("edits", {})
|
|
state["source_epub"] = str(epub_path)
|
|
write_json(state_path, state)
|
|
write_session_manifest(session_root, epub_path)
|
|
return state
|
|
|
|
|
|
def merge_rows(session_root: Path) -> list[dict[str, Any]]:
|
|
rows = read_json(session_root / "review_state" / "rows.json", [])
|
|
state = read_json(session_root / "review_state" / "state.json", {"edits": {}})
|
|
edits = state.get("edits", {})
|
|
merged = []
|
|
for row in rows:
|
|
item = dict(row)
|
|
edit = edits.get(row["id"], {})
|
|
item["current_html"] = edit.get("current_html", row["cn_html"])
|
|
item["marked"] = bool(edit.get("marked", False))
|
|
item["issue_type"] = edit.get("issue_type", "")
|
|
item["severity"] = edit.get("severity", "")
|
|
item["comment"] = edit.get("comment", "")
|
|
item["learn_note"] = edit.get("learn_note", "")
|
|
item["tags"] = edit.get("tags", "")
|
|
item["edited"] = item["current_html"] != row["cn_html"]
|
|
item["updated_at"] = edit.get("updated_at", "")
|
|
merged.append(item)
|
|
return merged
|
|
|
|
|
|
def write_feedback_summary(session_root: Path) -> Path:
|
|
rows = merge_rows(session_root)
|
|
marked = [
|
|
row
|
|
for row in rows
|
|
if row["edited"]
|
|
or row["marked"]
|
|
or row["comment"]
|
|
or row["learn_note"]
|
|
or row["issue_type"]
|
|
or row["severity"]
|
|
or row["tags"]
|
|
]
|
|
out = session_root / "feedback" / "feedback_for_codex.md"
|
|
lines = [
|
|
"# 双语 EPUB 审校反馈",
|
|
"",
|
|
f"- 更新时间：{now_iso()}",
|
|
f"- 有记录段落：{len(marked)}",
|
|
"",
|
|
"这些记录用于之后优化翻译、润色、术语和角色口吻。`修改前` 来自初始双语 EPUB，`修改后` 来自网页编辑器当前保存内容。",
|
|
"",
|
|
]
|
|
for row in marked:
|
|
lines.extend(
|
|
[
|
|
f"## {row['id']} · {row['file_label']} · P{row['cn_p_index']}",
|
|
"",
|
|
f"- 文件：`{row['file']}`",
|
|
f"- 问题类型：{row['issue_type'] or '未填写'}",
|
|
f"- 严重程度：{row['severity'] or '未填写'}",
|
|
f"- 标签：{row['tags'] or '无'}",
|
|
f"- 备注：{row['comment'] or '无'}",
|
|
f"- 可沉淀规则：{row['learn_note'] or '无'}",
|
|
"",
|
|
"日文原文：",
|
|
"",
|
|
row["jp_text"],
|
|
"",
|
|
"修改前：",
|
|
"",
|
|
strip_tags(row["cn_html"]),
|
|
"",
|
|
"修改后：",
|
|
"",
|
|
strip_tags(row["current_html"]),
|
|
"",
|
|
]
|
|
)
|
|
out.parent.mkdir(parents=True, exist_ok=True)
|
|
out.write_text("\n".join(lines), encoding="utf-8")
|
|
return out
|
|
|
|
|
|
def apply_edits_to_xhtml(session_root: Path) -> list[str]:
|
|
rows = merge_rows(session_root)
|
|
edited_rows = [row for row in rows if row["edited"]]
|
|
by_file: dict[str, list[dict[str, Any]]] = {}
|
|
for row in edited_rows:
|
|
by_file.setdefault(row["file"], []).append(row)
|
|
modified = []
|
|
for entry_name, items in by_file.items():
|
|
path = extracted_path(session_root, entry_name)
|
|
text = path.read_text(encoding="utf-8", errors="replace")
|
|
matches = list(P_RE.finditer(text))
|
|
replace_by_index = {int(row["cn_p_index"]): row["current_html"] for row in items}
|
|
parts: list[str] = []
|
|
last = 0
|
|
for index, match in enumerate(matches):
|
|
parts.append(text[last:match.start()])
|
|
if index in replace_by_index:
|
|
parts.append(match.group(1))
|
|
parts.append(replace_by_index[index])
|
|
parts.append(match.group(3))
|
|
else:
|
|
parts.append(match.group(0))
|
|
last = match.end()
|
|
parts.append(text[last:])
|
|
path.write_text("".join(parts), encoding="utf-8")
|
|
modified.append(entry_name)
|
|
write_feedback_summary(session_root)
|
|
return modified
|
|
|
|
|
|
def export_epub(session_root: Path) -> Path:
|
|
apply_edits_to_xhtml(session_root)
|
|
entries = read_json(session_root / "review_state" / "epub_entries.json", [])
|
|
manifest = read_json(session_manifest_path(session_root), {})
|
|
state_source = read_json(session_root / "review_state" / "state.json", {}).get("source_epub", "reviewed.epub")
|
|
source_name = manifest.get("source_name") or Path(state_source).name
|
|
ts = dt.datetime.now().strftime("%Y%m%d_%H%M%S")
|
|
out_dir = session_root / "exports"
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
out_path = out_dir / f"{safe_slug(Path(source_name).stem)}_网页审校版_{ts}.epub"
|
|
with zipfile.ZipFile(out_path, "w") as zf:
|
|
names_written: set[str] = set()
|
|
mimetype = extracted_path(session_root, "mimetype")
|
|
if mimetype.exists():
|
|
zf.write(mimetype, "mimetype", compress_type=zipfile.ZIP_STORED)
|
|
names_written.add("mimetype")
|
|
for entry in entries:
|
|
name = entry["filename"]
|
|
if name in names_written:
|
|
continue
|
|
path = extracted_path(session_root, name)
|
|
if not path.exists():
|
|
continue
|
|
compress = zipfile.ZIP_STORED if name == "mimetype" else zipfile.ZIP_DEFLATED
|
|
if entry.get("is_dir"):
|
|
info = zipfile.ZipInfo(name)
|
|
zf.writestr(info, b"")
|
|
else:
|
|
zf.write(path, name, compress_type=compress)
|
|
names_written.add(name)
|
|
append_jsonl(
|
|
session_root / "feedback" / "translation_feedback.jsonl",
|
|
{"ts": now_iso(), "event": "export_epub", "output": str(out_path)},
|
|
)
|
|
return out_path
|
|
|
|
|
|
def create_app(review_root: Path, initial_epub_path: Path | None = None, reset: bool = False) -> Flask:
|
|
app = Flask(
|
|
__name__,
|
|
root_path=str(Path(__file__).resolve().parent),
|
|
static_folder="static",
|
|
static_url_path="/static",
|
|
)
|
|
review_root.mkdir(parents=True, exist_ok=True)
|
|
app.config["REVIEW_ROOT"] = review_root
|
|
app.config["SESSION_ROOT"] = None
|
|
app.config["EPUB_PATH"] = None
|
|
app.config["LOCK"] = threading.Lock()
|
|
|
|
def set_active_session(session_root: Path, epub_path: Path | None = None) -> None:
|
|
app.config["SESSION_ROOT"] = session_root
|
|
app.config["EPUB_PATH"] = epub_path or session_epub_path(session_root)
|
|
update_app_manifest(review_root, session_root)
|
|
|
|
def active_session() -> tuple[Path | None, Path | None]:
|
|
session_root = app.config.get("SESSION_ROOT")
|
|
epub_path = app.config.get("EPUB_PATH")
|
|
if isinstance(session_root, Path):
|
|
return session_root, epub_path if isinstance(epub_path, Path) else session_epub_path(session_root)
|
|
return None, None
|
|
|
|
def no_active_response():
|
|
return jsonify(
|
|
{
|
|
"has_session": False,
|
|
"error": "尚未打开 EPUB。请先上传中日双语 EPUB，或从已有会话中选择一个。",
|
|
"review_root": str(review_root),
|
|
}
|
|
), 409
|
|
|
|
if initial_epub_path is not None:
|
|
session_root = session_root_for(review_root, initial_epub_path)
|
|
ensure_state(session_root, initial_epub_path, reset=reset)
|
|
set_active_session(session_root, initial_epub_path)
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return send_from_directory(app.static_folder, "index.html")
|
|
|
|
@app.route("/api/session")
|
|
def api_session():
|
|
session_root, epub_path = active_session()
|
|
if session_root is None or epub_path is None:
|
|
return jsonify(
|
|
{
|
|
"has_session": False,
|
|
"review_root": str(review_root),
|
|
"sessions": list_review_sessions(review_root),
|
|
}
|
|
)
|
|
rows = merge_rows(session_root)
|
|
touched = [row for row in rows if rowTouched_py(row)]
|
|
manifest = read_json(session_manifest_path(session_root), {})
|
|
return jsonify(
|
|
{
|
|
"has_session": True,
|
|
"id": session_public_id(session_root),
|
|
"source_name": manifest.get("source_name") or epub_path.name,
|
|
"source_epub": str(epub_path),
|
|
"session_root": str(session_root),
|
|
"row_count": len(rows),
|
|
"touched_count": len(touched),
|
|
"marked_count": sum(1 for row in rows if row.get("marked")),
|
|
"feedback_md": str(session_root / "feedback" / "feedback_for_codex.md"),
|
|
"feedback_jsonl": str(session_root / "feedback" / "translation_feedback.jsonl"),
|
|
}
|
|
)
|
|
|
|
@app.route("/api/sessions")
|
|
def api_sessions():
|
|
session_root, _epub_path = active_session()
|
|
return jsonify(
|
|
{
|
|
"review_root": str(review_root),
|
|
"active_session_id": session_public_id(session_root) if session_root else "",
|
|
"sessions": list_review_sessions(review_root),
|
|
}
|
|
)
|
|
|
|
@app.route("/api/upload", methods=["POST"])
|
|
def api_upload():
|
|
file_storage = request.files.get("epub")
|
|
if file_storage is None or not file_storage.filename:
|
|
return jsonify({"error": "请选择一个 .epub 文件"}), 400
|
|
reset_upload = str(request.form.get("reset", "")).lower() in {"1", "true", "yes", "on"}
|
|
try:
|
|
epub_path = copy_upload_to_review_root(file_storage, review_root)
|
|
session_root = session_root_for(review_root, epub_path)
|
|
with app.config["LOCK"]:
|
|
if reset_upload and session_root.exists():
|
|
shutil.rmtree(session_root)
|
|
ensure_state(session_root, epub_path, reset=reset_upload)
|
|
write_session_manifest(session_root, epub_path, source_name=file_storage.filename)
|
|
set_active_session(session_root, epub_path)
|
|
summary = summarize_session(session_root)
|
|
summary["has_session"] = True
|
|
return jsonify(summary)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 400
|
|
except zipfile.BadZipFile:
|
|
return jsonify({"error": "文件不是有效的 EPUB/ZIP 包"}), 400
|
|
except Exception as exc:
|
|
return jsonify({"error": f"EPUB 解析失败：{exc}"}), 500
|
|
|
|
@app.route("/api/session/open", methods=["POST"])
|
|
def api_open_session():
|
|
data = request.get_json(silent=True) or {}
|
|
session_id = str(data.get("session_id", ""))
|
|
if not session_id:
|
|
return jsonify({"error": "缺少 session_id"}), 400
|
|
try:
|
|
session_root = session_root_from_id(review_root, session_id)
|
|
except ValueError as exc:
|
|
return jsonify({"error": str(exc)}), 400
|
|
if not (session_root / "review_state" / "state.json").exists():
|
|
return jsonify({"error": "会话不存在或不完整"}), 404
|
|
with app.config["LOCK"]:
|
|
set_active_session(session_root)
|
|
summary = summarize_session(session_root)
|
|
summary["has_session"] = True
|
|
return jsonify(summary)
|
|
|
|
@app.route("/api/rows")
|
|
def api_rows():
|
|
session_root, _epub_path = active_session()
|
|
if session_root is None:
|
|
return no_active_response()
|
|
return jsonify({"rows": merge_rows(session_root)})
|
|
|
|
@app.route("/api/structure")
|
|
def api_structure():
|
|
session_root, _epub_path = active_session()
|
|
if session_root is None:
|
|
return no_active_response()
|
|
return jsonify(build_structure(session_root))
|
|
|
|
@app.route("/api/row/<row_id>", methods=["POST"])
|
|
def api_save_row(row_id: str):
|
|
session_root, _epub_path = active_session()
|
|
if session_root is None:
|
|
return no_active_response()
|
|
data = request.get_json(silent=True) or {}
|
|
rows = read_json(session_root / "review_state" / "rows.json", [])
|
|
base = next((row for row in rows if row["id"] == row_id), None)
|
|
if base is None:
|
|
return jsonify({"error": "row not found"}), 404
|
|
current_html = sanitize_cn_html(str(data.get("current_html", "")))
|
|
edit = {
|
|
"current_html": current_html,
|
|
"marked": bool(data.get("marked", False)),
|
|
"issue_type": str(data.get("issue_type", ""))[:80],
|
|
"severity": str(data.get("severity", ""))[:40],
|
|
"comment": str(data.get("comment", ""))[:4000],
|
|
"learn_note": str(data.get("learn_note", ""))[:4000],
|
|
"tags": str(data.get("tags", ""))[:500],
|
|
"updated_at": now_iso(),
|
|
}
|
|
with app.config["LOCK"]:
|
|
state_path = session_root / "review_state" / "state.json"
|
|
state = read_json(state_path, {"edits": {}})
|
|
state.setdefault("edits", {})[row_id] = edit
|
|
state["updated_at"] = now_iso()
|
|
write_json(state_path, state)
|
|
append_jsonl(
|
|
session_root / "feedback" / "translation_feedback.jsonl",
|
|
{
|
|
"ts": edit["updated_at"],
|
|
"event": "row_saved",
|
|
"row_id": row_id,
|
|
"file": base["file"],
|
|
"cn_p_index": base["cn_p_index"],
|
|
"edited": current_html != base["cn_html"],
|
|
"marked": edit["marked"],
|
|
"issue_type": edit["issue_type"],
|
|
"severity": edit["severity"],
|
|
"comment": edit["comment"],
|
|
"learn_note": edit["learn_note"],
|
|
"tags": edit["tags"],
|
|
"jp_text": base["jp_text"],
|
|
"before_cn_text": strip_tags(base["cn_html"]),
|
|
"after_cn_text": strip_tags(current_html),
|
|
"before_cn_html": base["cn_html"],
|
|
"after_cn_html": current_html,
|
|
},
|
|
)
|
|
return jsonify({"status": "ok", "row_id": row_id, "updated_at": edit["updated_at"]})
|
|
|
|
@app.route("/api/apply", methods=["POST"])
|
|
def api_apply():
|
|
session_root, _epub_path = active_session()
|
|
if session_root is None:
|
|
return no_active_response()
|
|
with app.config["LOCK"]:
|
|
modified = apply_edits_to_xhtml(session_root)
|
|
return jsonify(
|
|
{
|
|
"status": "ok",
|
|
"modified_files": modified,
|
|
"feedback_md": str(session_root / "feedback" / "feedback_for_codex.md"),
|
|
}
|
|
)
|
|
|
|
@app.route("/api/export", methods=["POST"])
|
|
def api_export():
|
|
session_root, _epub_path = active_session()
|
|
if session_root is None:
|
|
return no_active_response()
|
|
with app.config["LOCK"]:
|
|
out_path = export_epub(session_root)
|
|
return jsonify(
|
|
{
|
|
"status": "ok",
|
|
"output": str(out_path),
|
|
"download_url": f"/downloads/export/{urllib.parse.quote(out_path.name)}",
|
|
}
|
|
)
|
|
|
|
@app.route("/api/feedback")
|
|
def api_feedback():
|
|
session_root, _epub_path = active_session()
|
|
if session_root is None:
|
|
return no_active_response()
|
|
path = write_feedback_summary(session_root)
|
|
return jsonify(
|
|
{
|
|
"status": "ok",
|
|
"feedback_md": str(path),
|
|
"feedback_jsonl": str(session_root / "feedback" / "translation_feedback.jsonl"),
|
|
"feedback_md_url": "/downloads/feedback/feedback_for_codex.md",
|
|
"feedback_jsonl_url": "/downloads/feedback/translation_feedback.jsonl",
|
|
}
|
|
)
|
|
|
|
@app.route("/downloads/export/<path:filename>")
|
|
def download_export(filename: str):
|
|
session_root, _epub_path = active_session()
|
|
if session_root is None:
|
|
return no_active_response()
|
|
return send_from_directory(session_root / "exports", filename, as_attachment=True)
|
|
|
|
@app.route("/downloads/feedback/<path:filename>")
|
|
def download_feedback(filename: str):
|
|
session_root, _epub_path = active_session()
|
|
if session_root is None:
|
|
return no_active_response()
|
|
if filename not in {"feedback_for_codex.md", "translation_feedback.jsonl"}:
|
|
return jsonify({"error": "unknown feedback file"}), 404
|
|
return send_from_directory(session_root / "feedback", filename, as_attachment=True)
|
|
|
|
@app.route("/api/shutdown", methods=["POST"])
|
|
def api_shutdown():
|
|
def stop() -> None:
|
|
time.sleep(0.3)
|
|
os._exit(0)
|
|
|
|
threading.Thread(target=stop, daemon=True).start()
|
|
return jsonify({"status": "ok"})
|
|
|
|
return app
|
|
|
|
|
|
def wait_until_ready(url: str, proc: subprocess.Popen, timeout: int = 15) -> bool:
|
|
deadline = time.time() + timeout
|
|
while time.time() < deadline:
|
|
if proc.poll() is not None:
|
|
return False
|
|
try:
|
|
with urllib.request.urlopen(f"{url}/api/session", timeout=1) as response:
|
|
if response.status == 200:
|
|
return True
|
|
except OSError:
|
|
time.sleep(0.25)
|
|
return False
|
|
|
|
|
|
def run_daemon(args: argparse.Namespace) -> int:
|
|
bind_host = display_host(args.host)
|
|
port = find_free_port(args.port, bind_host)
|
|
review_root = Path(args.review_root).resolve()
|
|
review_root.mkdir(parents=True, exist_ok=True)
|
|
if args.epub:
|
|
epub_path = Path(args.epub).resolve()
|
|
if not epub_path.exists():
|
|
print(f"EPUB not found: {epub_path}", file=sys.stderr)
|
|
return 1
|
|
try:
|
|
validate_epub_file(epub_path)
|
|
except ValueError as exc:
|
|
print(str(exc), file=sys.stderr)
|
|
return 1
|
|
session_root = session_root_for(review_root, epub_path)
|
|
if args.reset and session_root.exists():
|
|
shutil.rmtree(session_root)
|
|
session_root.mkdir(parents=True, exist_ok=True)
|
|
log_path = session_root / "server.log"
|
|
else:
|
|
epub_path = None
|
|
log_path = review_root / "server.log"
|
|
cmd = [
|
|
sys.executable,
|
|
str(Path(__file__).resolve()),
|
|
"--review-root",
|
|
str(review_root),
|
|
"--host",
|
|
args.host,
|
|
"--port",
|
|
str(port),
|
|
]
|
|
if epub_path is not None:
|
|
cmd.insert(2, str(epub_path))
|
|
cmd.append("--no-browser")
|
|
if args.reset:
|
|
cmd.append("--reset")
|
|
creationflags = 0
|
|
popen_kwargs: dict[str, Any] = {}
|
|
if os.name == "nt":
|
|
creationflags = subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.DETACHED_PROCESS
|
|
else:
|
|
popen_kwargs["start_new_session"] = True
|
|
with log_path.open("a", encoding="utf-8") as log:
|
|
proc = subprocess.Popen(
|
|
cmd,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=log,
|
|
stderr=subprocess.STDOUT,
|
|
creationflags=creationflags,
|
|
**popen_kwargs,
|
|
)
|
|
url = f"http://{display_host(args.host)}:{port}"
|
|
if not wait_until_ready(url, proc):
|
|
print(f"review editor failed to start; see {log_path}", file=sys.stderr)
|
|
return 1
|
|
if not args.no_browser:
|
|
try:
|
|
webbrowser.open(url)
|
|
except Exception:
|
|
pass
|
|
print(url)
|
|
if epub_path is not None:
|
|
print(f"session: {session_root}")
|
|
else:
|
|
print(f"review root: {review_root}")
|
|
print(f"log: {log_path}")
|
|
return 0
|
|
|
|
|
|
def build_parser() -> argparse.ArgumentParser:
|
|
parser = argparse.ArgumentParser(description="Open a bilingual EPUB in a browser review/editor.")
|
|
parser.add_argument("epub", nargs="?", help="Optional path to a bilingual EPUB. Omit it to upload EPUBs in the browser.")
|
|
parser.add_argument("--review-root", default=DEFAULT_REVIEW_ROOT, help="Directory for review sessions.")
|
|
parser.add_argument("--host", default="127.0.0.1", help="Host to bind. Use 0.0.0.0 for trusted LAN/server access.")
|
|
parser.add_argument("--port", type=int, default=5177, help="Port, auto-advances if busy.")
|
|
parser.add_argument("--daemon", action="store_true", help="Start in background and print URL.")
|
|
parser.add_argument("--no-browser", action="store_true", help="Do not auto-open the browser.")
|
|
parser.add_argument("--reset", action="store_true", help="Re-extract the EPUB and discard prior review state.")
|
|
return parser
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
args = build_parser().parse_args(argv)
|
|
epub_path = Path(args.epub).resolve() if args.epub else None
|
|
if epub_path is not None:
|
|
if not epub_path.exists():
|
|
print(f"EPUB not found: {epub_path}", file=sys.stderr)
|
|
return 1
|
|
try:
|
|
validate_epub_file(epub_path)
|
|
except ValueError as exc:
|
|
print(str(exc), file=sys.stderr)
|
|
return 1
|
|
if args.daemon:
|
|
return run_daemon(args)
|
|
|
|
review_root = Path(args.review_root).resolve()
|
|
app = create_app(review_root, epub_path, reset=args.reset)
|
|
bind_host = display_host(args.host)
|
|
port = find_free_port(args.port, bind_host)
|
|
url = f"http://{display_host(args.host)}:{port}"
|
|
if not args.no_browser:
|
|
try:
|
|
webbrowser.open(url)
|
|
except Exception:
|
|
pass
|
|
|
|
def on_sigterm(_signum: int, _frame: Any) -> None:
|
|
sys.exit(0)
|
|
|
|
try:
|
|
signal.signal(signal.SIGTERM, on_sigterm)
|
|
except Exception:
|
|
pass
|
|
print(f"EPUB review editor: {url}")
|
|
print(f"Review root: {review_root}")
|
|
session_root = app.config.get("SESSION_ROOT")
|
|
if isinstance(session_root, Path):
|
|
print(f"Session: {session_root}")
|
|
print(f"Rows: {len(read_json(session_root / 'review_state' / 'rows.json', []))}")
|
|
else:
|
|
print("No EPUB loaded yet. Upload one in the browser.")
|
|
app.run(host=args.host, port=port, debug=False, threaded=True)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|