xin61
1027 lines
36 KiB
Python
1027 lines
36 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import atexit
|
|
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
|
|
|
|
|
|
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) -> 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(("127.0.0.1", port))
|
|
except OSError:
|
|
continue
|
|
return port
|
|
raise RuntimeError(f"no free port found from {start}")
|
|
|
|
|
|
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 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", {})
|
|
write_json(state_path, state)
|
|
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", [])
|
|
source_epub = Path(read_json(session_root / "review_state" / "state.json", {}).get("source_epub", "reviewed.epub"))
|
|
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(source_epub.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(session_root: Path, epub_path: Path) -> Flask:
|
|
app = Flask(
|
|
__name__,
|
|
root_path=str(Path(__file__).resolve().parent),
|
|
static_folder="static",
|
|
static_url_path="/static",
|
|
)
|
|
app.config["SESSION_ROOT"] = session_root
|
|
app.config["EPUB_PATH"] = epub_path
|
|
app.config["LOCK"] = threading.Lock()
|
|
|
|
@app.route("/")
|
|
def index():
|
|
return send_from_directory(app.static_folder, "index.html")
|
|
|
|
@app.route("/api/session")
|
|
def api_session():
|
|
rows = merge_rows(session_root)
|
|
touched = [row for row in rows if row["edited"] or row["marked"] or row["comment"] or row["learn_note"]]
|
|
return jsonify(
|
|
{
|
|
"source_epub": str(epub_path),
|
|
"session_root": str(session_root),
|
|
"row_count": len(rows),
|
|
"touched_count": len(touched),
|
|
"feedback_md": str(session_root / "feedback" / "feedback_for_codex.md"),
|
|
"feedback_jsonl": str(session_root / "feedback" / "translation_feedback.jsonl"),
|
|
}
|
|
)
|
|
|
|
@app.route("/api/rows")
|
|
def api_rows():
|
|
return jsonify({"rows": merge_rows(session_root)})
|
|
|
|
@app.route("/api/structure")
|
|
def api_structure():
|
|
return jsonify(build_structure(session_root))
|
|
|
|
@app.route("/api/row/<row_id>", methods=["POST"])
|
|
def api_save_row(row_id: str):
|
|
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():
|
|
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():
|
|
with app.config["LOCK"]:
|
|
out_path = export_epub(session_root)
|
|
return jsonify({"status": "ok", "output": str(out_path)})
|
|
|
|
@app.route("/api/feedback")
|
|
def api_feedback():
|
|
path = write_feedback_summary(session_root)
|
|
return jsonify(
|
|
{
|
|
"status": "ok",
|
|
"feedback_md": str(path),
|
|
"feedback_jsonl": str(session_root / "feedback" / "translation_feedback.jsonl"),
|
|
}
|
|
)
|
|
|
|
@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:
|
|
port = find_free_port(args.port)
|
|
review_root = Path(args.review_root).resolve()
|
|
epub_path = Path(args.epub).resolve()
|
|
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"
|
|
cmd = [
|
|
sys.executable,
|
|
str(Path(__file__).resolve()),
|
|
str(epub_path),
|
|
"--review-root",
|
|
str(review_root),
|
|
"--port",
|
|
str(port),
|
|
]
|
|
if args.no_browser:
|
|
cmd.append("--no-browser")
|
|
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://localhost:{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)
|
|
print(f"session: {session_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", help="Path to the bilingual EPUB.")
|
|
parser.add_argument("--review-root", default=DEFAULT_REVIEW_ROOT, help="Directory for review sessions.")
|
|
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 not epub_path.exists():
|
|
print(f"EPUB not found: {epub_path}", file=sys.stderr)
|
|
return 1
|
|
if args.daemon:
|
|
return run_daemon(args)
|
|
|
|
review_root = Path(args.review_root).resolve()
|
|
session_root = session_root_for(review_root, epub_path)
|
|
ensure_state(session_root, epub_path, reset=args.reset)
|
|
|
|
app = create_app(session_root, epub_path)
|
|
port = find_free_port(args.port)
|
|
url = f"http://localhost:{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"Session: {session_root}")
|
|
print(f"Rows: {len(read_json(session_root / 'review_state' / 'rows.json', []))}")
|
|
app.run(host="127.0.0.1", port=port, debug=False, threaded=True)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|