xin61
1952 lines
74 KiB
Python
1952 lines
74 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.error
|
||
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, make_response, request, send_from_directory
|
||
|
||
try:
|
||
from version import VERSION_RULES, version
|
||
except ImportError:
|
||
VERSION_RULES = {
|
||
"PATCH": "修复 bug 或兼容性小修",
|
||
"MINOR": "新增向后兼容能力、API 字段或可选配置",
|
||
"MAJOR": "破坏兼容的 API、配置或行为变更",
|
||
}
|
||
version = "0.4.1"
|
||
|
||
|
||
P_RE = re.compile(r"(<p\b[^>]*>)(.*?)(</p>)", re.S | re.I)
|
||
TAG_RE = re.compile(r"<[^>]+>")
|
||
IMG_TAG_RE = re.compile(r"<(?:img|image)\b[^>]*>", re.S | re.I)
|
||
ATTR_RE = re.compile(r"([:\w-]+)\s*=\s*(['\"])(.*?)\2", re.S)
|
||
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,
|
||
)
|
||
EXTERNAL_URI_RE = re.compile(r"^(?:[a-z][a-z0-9+.-]*:|//)", re.I)
|
||
IMAGE_EXTENSIONS = {".apng", ".avif", ".bmp", ".gif", ".jpeg", ".jpg", ".png", ".svg", ".webp"}
|
||
RASTER_IMAGE_EXTENSIONS = {".apng", ".avif", ".bmp", ".gif", ".jpeg", ".jpg", ".png", ".webp"}
|
||
|
||
DEFAULT_REVIEW_ROOT = "epub_review_sessions"
|
||
STATE_VERSION = 1
|
||
UPLOAD_DIR_NAME = "uploads"
|
||
DEFAULT_GPT_BASE_URL = "https://api.openai.com/v1"
|
||
DEFAULT_GPT_MODEL = "gpt-4o-mini"
|
||
|
||
|
||
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 normalize_href_with_fragment(base_dir: str, href: str) -> tuple[str, str]:
|
||
href = urllib.parse.unquote(href or "")
|
||
path, fragment = (href.split("#", 1) + [""])[:2] if "#" in href else (href, "")
|
||
path = posixpath.normpath(posixpath.join(base_dir, path)).lstrip("./") if path else ""
|
||
return path, fragment
|
||
|
||
|
||
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 normalize_gpt_base_url(value: str) -> str:
|
||
value = (value or DEFAULT_GPT_BASE_URL).strip().rstrip("/")
|
||
if not value:
|
||
value = DEFAULT_GPT_BASE_URL
|
||
parsed = urllib.parse.urlparse(value)
|
||
if parsed.scheme not in {"http", "https"} or not parsed.netloc:
|
||
raise ValueError("Base URL 必须是 http(s) 地址")
|
||
return value
|
||
|
||
|
||
def read_gpt_config(review_root: Path) -> dict[str, Any]:
|
||
data = read_json(gpt_config_path(review_root), {})
|
||
api_key = str(data.get("api_key") or os.environ.get("OPENAI_API_KEY") or "").strip()
|
||
base_url = str(data.get("base_url") or os.environ.get("OPENAI_BASE_URL") or DEFAULT_GPT_BASE_URL).strip()
|
||
model = str(data.get("model") or os.environ.get("OPENAI_MODEL") or DEFAULT_GPT_MODEL).strip()
|
||
return {
|
||
"base_url": normalize_gpt_base_url(base_url),
|
||
"model": model or DEFAULT_GPT_MODEL,
|
||
"api_key": api_key,
|
||
"configured": bool(api_key),
|
||
"key_source": "config" if str(data.get("api_key") or "").strip() else ("env" if os.environ.get("OPENAI_API_KEY") else ""),
|
||
"updated_at": data.get("updated_at", ""),
|
||
}
|
||
|
||
|
||
def public_gpt_config(review_root: Path) -> dict[str, Any]:
|
||
config = read_gpt_config(review_root)
|
||
return {
|
||
"configured": config["configured"],
|
||
"base_url": config["base_url"],
|
||
"model": config["model"],
|
||
"key_source": config["key_source"],
|
||
"updated_at": config.get("updated_at", ""),
|
||
}
|
||
|
||
|
||
def save_gpt_config(review_root: Path, data: dict[str, Any]) -> dict[str, Any]:
|
||
base_url = normalize_gpt_base_url(str(data.get("base_url") or DEFAULT_GPT_BASE_URL))
|
||
model = str(data.get("model") or DEFAULT_GPT_MODEL).strip() or DEFAULT_GPT_MODEL
|
||
api_key = str(data.get("api_key") or "").strip()
|
||
keep_existing = bool(data.get("keep_existing", False))
|
||
existing = read_json(gpt_config_path(review_root), {})
|
||
if keep_existing and not api_key:
|
||
api_key = str(existing.get("api_key") or "").strip()
|
||
record = {
|
||
"base_url": base_url,
|
||
"model": model,
|
||
"api_key": api_key,
|
||
"updated_at": now_iso(),
|
||
}
|
||
write_json(gpt_config_path(review_root), record)
|
||
return public_gpt_config(review_root)
|
||
|
||
|
||
def compact_glossary_for_prompt(row: dict[str, Any], limit: int = 36) -> list[str]:
|
||
project_root = Path(__file__).resolve().parents[2]
|
||
glossary_path = project_root / "TRPG玩家在异世界打造最强角色" / "mingcibiao.json"
|
||
if not glossary_path.exists():
|
||
return ["Falchion/ファルシオン：大砍刀，不要译成偃月刀。"]
|
||
try:
|
||
glossary = read_json(glossary_path, {})
|
||
except Exception:
|
||
glossary = {}
|
||
source = f"{row.get('jp_text', '')}\n{row.get('jp_html', '')}"
|
||
selected: list[str] = []
|
||
for key, value in glossary.items():
|
||
if not isinstance(key, str) or not isinstance(value, str):
|
||
continue
|
||
clean_value = value.split("#", 1)[0].strip()
|
||
if not clean_value:
|
||
continue
|
||
if key and key in source:
|
||
selected.append(f"{key} => {clean_value}")
|
||
if len(selected) >= limit:
|
||
break
|
||
falchion_rule = "Falchion/ファルシオン => 大砍刀，不要译成偃月刀。"
|
||
if not any("Falchion" in item or "ファルシオン" in item for item in selected):
|
||
selected.append(falchion_rule)
|
||
return selected[:limit]
|
||
|
||
|
||
def build_retranslate_messages(row: dict[str, Any], instruction: str = "") -> list[dict[str, str]]:
|
||
glossary_lines = "\n".join(f"- {item}" for item in compact_glossary_for_prompt(row))
|
||
system_prompt = (
|
||
"你是日语轻小说到简体中文的审校重翻助手。请只输出当前这一段的中文译文 HTML 片段，"
|
||
"不要解释，不要输出 Markdown。译文要自然流畅，符合简中轻小说文风；意义优先，不贴日语语序硬译。"
|
||
"不得合并或拆分段落。保留必要的 HTML 内联标签，尤其是有意义的 ruby："
|
||
"必须使用 <ruby><rb>中文</rb><rt>读音/外语</rt></ruby>，不要改成括号。"
|
||
"纯读音 ruby 可自然译入中文。术语优先服从给定术语表。"
|
||
)
|
||
user_parts = [
|
||
"请重翻以下单段。",
|
||
"",
|
||
"日文原文 HTML：",
|
||
row.get("jp_html") or row.get("jp_text") or "",
|
||
"",
|
||
"当前中文译文 HTML：",
|
||
row.get("current_html") or row.get("cn_html") or row.get("cn_text") or "",
|
||
]
|
||
if glossary_lines:
|
||
user_parts.extend(["", "相关术语：", glossary_lines])
|
||
if instruction.strip():
|
||
user_parts.extend(["", "本次额外要求：", instruction.strip()[:1200]])
|
||
return [
|
||
{"role": "system", "content": system_prompt},
|
||
{"role": "user", "content": "\n".join(user_parts)},
|
||
]
|
||
|
||
|
||
def call_openai_compatible_chat(
|
||
config: dict[str, Any],
|
||
messages: list[dict[str, str]],
|
||
temperature: float = 0.2,
|
||
timeout: int = 90,
|
||
) -> str:
|
||
api_key = str(config.get("api_key") or "").strip()
|
||
if not api_key:
|
||
raise ValueError("尚未配置 GPT API Key")
|
||
base_url = normalize_gpt_base_url(str(config.get("base_url") or DEFAULT_GPT_BASE_URL))
|
||
url = f"{base_url}/chat/completions"
|
||
payload = {
|
||
"model": str(config.get("model") or DEFAULT_GPT_MODEL),
|
||
"messages": messages,
|
||
"temperature": temperature,
|
||
}
|
||
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
||
req = urllib.request.Request(
|
||
url,
|
||
data=body,
|
||
headers={
|
||
"Content-Type": "application/json",
|
||
"Authorization": f"Bearer {api_key}",
|
||
},
|
||
method="POST",
|
||
)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=timeout) as response:
|
||
result = json.loads(response.read().decode("utf-8", errors="replace"))
|
||
except urllib.error.HTTPError as exc:
|
||
detail = exc.read().decode("utf-8", errors="replace")[:1200]
|
||
raise RuntimeError(f"GPT 请求失败：HTTP {exc.code} {detail}") from exc
|
||
except urllib.error.URLError as exc:
|
||
raise RuntimeError(f"GPT 请求失败：{exc.reason}") from exc
|
||
choices = result.get("choices") or []
|
||
if not choices:
|
||
raise RuntimeError("GPT 返回为空")
|
||
content = ((choices[0].get("message") or {}).get("content") or "").strip()
|
||
if content.startswith("```"):
|
||
content = re.sub(r"^```(?:html|xml)?\s*", "", content, flags=re.I).strip()
|
||
content = re.sub(r"\s*```$", "", content).strip()
|
||
if not content:
|
||
raise RuntimeError("GPT 没有返回译文")
|
||
return content
|
||
|
||
|
||
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 gpt_config_path(review_root: Path) -> Path:
|
||
return review_root / "gpt_config.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=240)
|
||
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),
|
||
"version": version,
|
||
"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 = compact_text(strip_tags(match.group(1)))
|
||
if title:
|
||
return title
|
||
return ""
|
||
|
||
|
||
def extract_body_heading_from_html(text: str) -> str:
|
||
match = re.search(r"<h[1-6]\b[^>]*>(.*?)</h[1-6]>", text, flags=re.S | re.I)
|
||
if not match:
|
||
return ""
|
||
return compact_text(strip_tags(match.group(1)))
|
||
|
||
|
||
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 read_body_heading(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_body_heading_from_html(text)
|
||
|
||
|
||
def first_content_text(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")
|
||
for match in P_RE.finditer(text):
|
||
inner = match.group(2)
|
||
if re.search(r"<img\b", inner, flags=re.I):
|
||
continue
|
||
label = compact_text(strip_tags(inner))
|
||
if label:
|
||
return label
|
||
return ""
|
||
|
||
|
||
def tag_attrs(tag: str) -> dict[str, str]:
|
||
attrs: dict[str, str] = {}
|
||
for key, _quote, value in ATTR_RE.findall(tag):
|
||
attrs[key.lower()] = html.unescape(value or "")
|
||
return attrs
|
||
|
||
|
||
def part_display_label(entry_name: str, fallback_index: int) -> str:
|
||
stem = Path(entry_name).stem
|
||
for pattern in (r"part[-_]?(\d+)$", r"p[-_]?(\d+)$", r"^(\d+)$"):
|
||
match = re.search(pattern, stem, flags=re.I)
|
||
if match:
|
||
return f"part{int(match.group(1)):04d}"
|
||
if stem and stem.lower() not in {"titlepage", "cover"}:
|
||
return stem
|
||
return f"part{fallback_index:04d}"
|
||
|
||
|
||
def is_image_asset(entry_name: str) -> bool:
|
||
suffix = posixpath.splitext(entry_name.split("?", 1)[0])[1].lower()
|
||
return suffix in IMAGE_EXTENSIONS
|
||
|
||
|
||
def is_raster_image_asset(entry_name: str) -> bool:
|
||
suffix = posixpath.splitext(entry_name.split("?", 1)[0])[1].lower()
|
||
return suffix in RASTER_IMAGE_EXTENSIONS
|
||
|
||
|
||
def asset_url(entry_name: str) -> str:
|
||
return f"/api/asset/{urllib.parse.quote(entry_name, safe='/')}"
|
||
|
||
|
||
def extract_images_from_html(session_root: Path, entry_name: str) -> list[dict[str, Any]]:
|
||
path = extracted_path(session_root, entry_name)
|
||
if not path.exists():
|
||
return []
|
||
text = path.read_text(encoding="utf-8", errors="replace")
|
||
images: list[dict[str, Any]] = []
|
||
for index, match in enumerate(IMG_TAG_RE.finditer(text), start=1):
|
||
attrs = tag_attrs(match.group(0))
|
||
href = attrs.get("src") or attrs.get("href") or attrs.get("xlink:href")
|
||
if not href or EXTERNAL_URI_RE.match(href):
|
||
continue
|
||
image_entry = normalize_href(entry_dir(entry_name), href)
|
||
if not image_entry or not is_raster_image_asset(image_entry):
|
||
continue
|
||
images.append(
|
||
{
|
||
"index": index,
|
||
"file": entry_name,
|
||
"file_label": Path(entry_name).name,
|
||
"asset_path": image_entry,
|
||
"asset_name": Path(image_entry).name,
|
||
"asset_url": asset_url(image_entry),
|
||
"alt": compact_text(attrs.get("alt") or attrs.get("title") or ""),
|
||
}
|
||
)
|
||
return images
|
||
|
||
|
||
def collect_asset_paths_from_structure(structure: dict[str, Any]) -> set[str]:
|
||
paths: set[str] = set()
|
||
for chapter in structure.get("chapters", []):
|
||
for image in chapter.get("images", []):
|
||
asset_path = image.get("asset_path", "")
|
||
if asset_path:
|
||
paths.add(asset_path)
|
||
return paths
|
||
|
||
|
||
def nav_target_key(base_dir: str, href: str) -> tuple[str, str]:
|
||
path, fragment = normalize_href_with_fragment(base_dir, href)
|
||
return path, fragment
|
||
|
||
|
||
def parse_internal_toc(session_root: Path) -> list[dict[str, Any]]:
|
||
ordered_files = html_entries_in_reading_order(session_root)
|
||
ordered_set = set(ordered_files)
|
||
for entry_name in ordered_files:
|
||
path = extracted_path(session_root, entry_name)
|
||
if not path.exists():
|
||
continue
|
||
text = path.read_text(encoding="utf-8", errors="replace")
|
||
if not re.search(r"<a\b[^>]*href=", text, flags=re.I):
|
||
continue
|
||
basename = posixpath.basename(entry_name).lower()
|
||
heading = extract_heading_from_html(text)
|
||
is_toc_like_name = basename in {"nav.xhtml", "toc.xhtml", "toc.html", "toc.htm"} or "toc" in basename
|
||
is_toc_like_title = bool(re.search(r"(目录|目次|contents?)", heading, flags=re.I))
|
||
matches = list(P_RE.finditer(text))
|
||
base_dir = entry_dir(entry_name)
|
||
candidate_items: list[dict[str, Any]] = []
|
||
i = 0
|
||
while i < len(matches):
|
||
source_inner = matches[i].group(2)
|
||
link_match = re.search(r"<a\b[^>]*href=['\"]([^'\"]+)['\"][^>]*>(.*?)</a>", source_inner, flags=re.S | re.I)
|
||
if not link_match:
|
||
i += 1
|
||
continue
|
||
target_file, target_fragment = nav_target_key(base_dir, link_match.group(1))
|
||
if not target_file:
|
||
i += 1
|
||
continue
|
||
if target_file not in ordered_set:
|
||
i += 1
|
||
continue
|
||
source_title = compact_text(strip_tags(link_match.group(2))) or compact_text(strip_tags(source_inner))
|
||
title = source_title
|
||
if i + 1 < len(matches):
|
||
next_inner = matches[i + 1].group(2)
|
||
if not is_gray_source(matches[i + 1].group(1)[2:-1], next_inner):
|
||
next_title = compact_text(strip_tags(next_inner))
|
||
if next_title:
|
||
title = next_title
|
||
i += 1
|
||
if title:
|
||
candidate_items.append(
|
||
{
|
||
"title": title,
|
||
"href": target_file,
|
||
"fragment": target_fragment,
|
||
"children": [],
|
||
"source_title": source_title,
|
||
}
|
||
)
|
||
i += 1
|
||
if candidate_items:
|
||
internal_targets = sum(1 for item in candidate_items if item["href"] in ordered_set and item["href"] != entry_name)
|
||
link_density = internal_targets / max(1, len(matches))
|
||
if is_toc_like_name or is_toc_like_title or (len(candidate_items) >= 4 and link_density >= 0.35):
|
||
return candidate_items
|
||
return []
|
||
|
||
|
||
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 = ""
|
||
fragment = ""
|
||
children: list[dict[str, Any]] = []
|
||
for child in list(li):
|
||
name = local_name(child.tag)
|
||
if name == "a":
|
||
label = element_text(child)
|
||
href, fragment = normalize_href_with_fragment(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, "fragment": fragment, "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)
|
||
|
||
all_ordered_files = html_entries_in_reading_order(session_root)
|
||
image_by_file = {
|
||
entry_name: images
|
||
for entry_name in all_ordered_files
|
||
if (images := extract_images_from_html(session_root, entry_name))
|
||
}
|
||
ordered_files = [file for file in all_ordered_files if file in rows_by_file]
|
||
for file in rows_by_file:
|
||
if file not in ordered_files:
|
||
ordered_files.append(file)
|
||
content_files = [file for file in all_ordered_files if file in rows_by_file or file in image_by_file]
|
||
for file in sorted(set(rows_by_file) | set(image_by_file), key=file_sort_key):
|
||
if file not in content_files:
|
||
content_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
|
||
|
||
start_titles: dict[str, str] = {}
|
||
chapter_specs: list[dict[str, Any]] = []
|
||
toc_items = parse_internal_toc(session_root)
|
||
ordered_index = {entry_name: index for index, entry_name in enumerate(all_ordered_files)}
|
||
content_file_set = set(content_files)
|
||
|
||
def first_content_file_at_or_after(entry_name: str) -> str:
|
||
start_pos = ordered_index.get(entry_name, -1)
|
||
if start_pos < 0:
|
||
return ""
|
||
for candidate in all_ordered_files[start_pos:]:
|
||
if candidate in content_file_set:
|
||
return candidate
|
||
return ""
|
||
|
||
for item in toc_items:
|
||
href = item.get("href", "")
|
||
if href not in ordered_index:
|
||
continue
|
||
start = first_content_file_at_or_after(href)
|
||
if start and item.get("title") and start not in start_titles:
|
||
start_titles[start] = item["title"]
|
||
|
||
for entry_name in ordered_files:
|
||
heading = read_body_heading(session_root, entry_name)
|
||
if heading and entry_name not in start_titles and len(strip_tags(heading)) <= 80:
|
||
start_titles[entry_name] = heading
|
||
|
||
if start_titles:
|
||
starts = sorted(start_titles, key=lambda name: ordered_index.get(name, 10**9))
|
||
for index, start in enumerate(starts):
|
||
start_pos = ordered_index.get(start, -1)
|
||
next_pos = ordered_index.get(starts[index + 1], len(all_ordered_files)) if index + 1 < len(starts) else len(all_ordered_files)
|
||
if start_pos < 0:
|
||
continue
|
||
files = [
|
||
entry_name
|
||
for entry_name in all_ordered_files[start_pos:next_pos]
|
||
if entry_name in rows_by_file or entry_name in image_by_file
|
||
]
|
||
if files:
|
||
chapter_specs.append({"title": start_titles.get(start, ""), "files": files, "start": files[0]})
|
||
|
||
grouped_files = {entry_name for spec in chapter_specs for entry_name in spec["files"]}
|
||
for entry_name in content_files:
|
||
if entry_name not in grouped_files:
|
||
chapter_specs.append({"title": "", "files": [entry_name], "start": entry_name})
|
||
chapter_specs.sort(key=lambda spec: ordered_index.get(spec["start"], 10**9))
|
||
|
||
part_serial = 0
|
||
image_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 = part_display_label(entry_name, part_serial - 1)
|
||
touched_count, marked_count = file_counts(file_rows)
|
||
return {
|
||
"id": f"PT{part_serial:04d}",
|
||
"kind": "text",
|
||
"title": title,
|
||
"source_title": title_override
|
||
or nav_titles.get(entry_name)
|
||
or read_body_heading(session_root, entry_name)
|
||
or first_content_text(session_root, entry_name)
|
||
or read_document_title(session_root, entry_name)
|
||
or Path(entry_name).stem,
|
||
"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 make_image_chapter(
|
||
entry_name: str,
|
||
image: dict[str, Any],
|
||
title_override: str = "",
|
||
image_index: int = 1,
|
||
image_total: int = 1,
|
||
) -> dict[str, Any]:
|
||
nonlocal chapter_serial, image_serial
|
||
chapter_serial += 1
|
||
image_serial += 1
|
||
suffix = f"-{image_index:02d}" if image_total > 1 else ""
|
||
label = part_display_label(entry_name, image_serial - 1)
|
||
default_title = "封面" if Path(entry_name).stem.lower() in {"titlepage", "cover"} else f"插图 {label}{suffix}"
|
||
base_title = title_override or nav_titles.get(entry_name) or read_body_heading(session_root, entry_name)
|
||
if title_override and base_title:
|
||
title = f"{base_title} · 插图{suffix}"
|
||
else:
|
||
title = base_title or default_title
|
||
image_item = {
|
||
**image,
|
||
"id": f"IMG{image_serial:04d}-{image['index']:02d}",
|
||
}
|
||
return {
|
||
"id": f"C{chapter_serial:04d}",
|
||
"kind": "image",
|
||
"title": title,
|
||
"file": entry_name,
|
||
"file_label": Path(entry_name).name,
|
||
"images": [image_item],
|
||
"row_count": 0,
|
||
"touched_count": 0,
|
||
"marked_count": 0,
|
||
"parts": [],
|
||
"_sort": ordered_index.get(entry_name, 10**9) + image_index / 100,
|
||
}
|
||
|
||
def chapter_from_parts(chapter_id: str, title: str, parts: list[dict[str, Any]]) -> dict[str, Any]:
|
||
return {
|
||
"id": chapter_id,
|
||
"kind": "text",
|
||
"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,
|
||
"_sort": min((ordered_index.get(part["file"], 10**9) for part in parts), default=10**9),
|
||
}
|
||
|
||
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 or href in image_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
|
||
|
||
def append_file_as_chapter(entry_name: str, title_override: str = "") -> None:
|
||
nonlocal chapter_serial
|
||
if entry_name in image_by_file:
|
||
images = image_by_file.get(entry_name, [])
|
||
for image_index, image in enumerate(images, start=1):
|
||
chapters.append(make_image_chapter(entry_name, image, title_override, image_index, len(images)))
|
||
return
|
||
if entry_name in rows_by_file:
|
||
chapter_serial += 1
|
||
part = make_part(entry_name, title_override)
|
||
chapters.append(chapter_from_parts(f"C{chapter_serial:04d}", title_override or part["source_title"] or part["title"], [part]))
|
||
|
||
def append_grouped_files(title: str, files: list[str]) -> None:
|
||
nonlocal chapter_serial
|
||
text_parts: list[dict[str, Any]] = []
|
||
text_title = title
|
||
|
||
def flush_text_parts() -> None:
|
||
nonlocal chapter_serial, text_parts, text_title
|
||
if not text_parts:
|
||
return
|
||
chapter_serial += 1
|
||
chapters.append(chapter_from_parts(f"C{chapter_serial:04d}", text_title or text_parts[0]["source_title"] or text_parts[0]["title"], text_parts))
|
||
text_parts = []
|
||
text_title = ""
|
||
|
||
for file_index, entry_name in enumerate(files):
|
||
if entry_name in image_by_file:
|
||
flush_text_parts()
|
||
images = image_by_file.get(entry_name, [])
|
||
for image_index, image in enumerate(images, start=1):
|
||
image_title = title if file_index == 0 and not text_parts else ""
|
||
chapters.append(make_image_chapter(entry_name, image, image_title, image_index, len(images)))
|
||
elif entry_name in rows_by_file:
|
||
text_parts.append(make_part(entry_name))
|
||
flush_text_parts()
|
||
|
||
for spec in chapter_specs:
|
||
node_files = [entry_name for entry_name in spec.get("files", []) if entry_name not in assigned_files]
|
||
if not node_files:
|
||
continue
|
||
if len(node_files) == 1 and node_files[0] in image_by_file:
|
||
append_file_as_chapter(node_files[0], spec.get("title") or "")
|
||
else:
|
||
append_grouped_files(spec.get("title") or "", node_files)
|
||
assigned_files.update(node_files)
|
||
|
||
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"]
|
||
|
||
if not chapters:
|
||
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
|
||
append_grouped_files(node.get("title") or "", [item["href"] for item in node_files])
|
||
assigned_files.update(item["href"] for item in node_files)
|
||
|
||
for entry_name in content_files:
|
||
if entry_name in assigned_files:
|
||
continue
|
||
append_file_as_chapter(entry_name)
|
||
|
||
chapters.sort(key=lambda chapter: chapter.get("_sort", 10**9))
|
||
for index, chapter in enumerate(chapters, start=1):
|
||
chapter["id"] = f"C{index:04d}"
|
||
chapter.pop("_sort", None)
|
||
|
||
return {
|
||
"chapters": chapters,
|
||
"chapter_count": len(chapters),
|
||
"text_chapter_count": sum(1 for chapter in chapters if chapter.get("kind") != "image"),
|
||
"image_chapter_count": sum(1 for chapter in chapters if chapter.get("kind") == "image"),
|
||
"part_count": sum(len(chapter["parts"]) for chapter in chapters),
|
||
"image_count": sum(len(chapter.get("images", [])) for chapter in chapters),
|
||
"asset_count": len(collect_asset_paths_from_structure({"chapters": 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["APP_VERSION"] = version
|
||
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)
|
||
active_id = read_json(app_manifest_path(review_root), {}).get("active_session_id", "")
|
||
if active_id:
|
||
try:
|
||
manifest_root = session_root_from_id(review_root, active_id)
|
||
except ValueError:
|
||
return None, None
|
||
if (manifest_root / "review_state" / "state.json").exists():
|
||
set_active_session(manifest_root)
|
||
return app.config["SESSION_ROOT"], app.config["EPUB_PATH"]
|
||
return None, None
|
||
|
||
def no_active_response():
|
||
return jsonify(
|
||
{
|
||
"has_session": False,
|
||
"version": version,
|
||
"version_rules": VERSION_RULES,
|
||
"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,
|
||
"version": version,
|
||
"version_rules": VERSION_RULES,
|
||
"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,
|
||
"version": version,
|
||
"version_rules": VERSION_RULES,
|
||
"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/version")
|
||
def api_version():
|
||
return jsonify({"version": version, "version_rules": VERSION_RULES})
|
||
|
||
@app.route("/api/sessions")
|
||
def api_sessions():
|
||
session_root, _epub_path = active_session()
|
||
return jsonify(
|
||
{
|
||
"review_root": str(review_root),
|
||
"version": version,
|
||
"version_rules": VERSION_RULES,
|
||
"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/asset/<path:entry_name>")
|
||
def api_asset(entry_name: str):
|
||
session_root, _epub_path = active_session()
|
||
if session_root is None:
|
||
return no_active_response()
|
||
entry_name = urllib.parse.unquote(entry_name)
|
||
structure = build_structure(session_root)
|
||
if entry_name not in collect_asset_paths_from_structure(structure):
|
||
return jsonify({"error": "asset not allowed"}), 404
|
||
if not is_raster_image_asset(entry_name):
|
||
return jsonify({"error": "asset type not allowed"}), 415
|
||
try:
|
||
path = extracted_path(session_root, entry_name)
|
||
except ValueError as exc:
|
||
return jsonify({"error": str(exc)}), 400
|
||
if not path.exists() or not path.is_file():
|
||
return jsonify({"error": "asset not found"}), 404
|
||
response = make_response(send_from_directory(path.parent, path.name))
|
||
response.headers["X-Content-Type-Options"] = "nosniff"
|
||
response.headers["Content-Security-Policy"] = "default-src 'none'; img-src 'self' data:; style-src 'none'; script-src 'none'; object-src 'none'"
|
||
return response
|
||
|
||
@app.route("/api/gpt/config", methods=["GET", "POST"])
|
||
def api_gpt_config():
|
||
if request.method == "GET":
|
||
return jsonify(public_gpt_config(review_root))
|
||
data = request.get_json(silent=True) or {}
|
||
try:
|
||
return jsonify(save_gpt_config(review_root, data))
|
||
except ValueError as exc:
|
||
return jsonify({"error": str(exc)}), 400
|
||
|
||
@app.route("/api/row/<row_id>/retranslate", methods=["POST"])
|
||
def api_retranslate_row(row_id: str):
|
||
session_root, _epub_path = active_session()
|
||
if session_root is None:
|
||
return no_active_response()
|
||
rows = merge_rows(session_root)
|
||
row = next((item for item in rows if item["id"] == row_id), None)
|
||
if row is None:
|
||
return jsonify({"error": "row not found"}), 404
|
||
data = request.get_json(silent=True) or {}
|
||
try:
|
||
config = read_gpt_config(review_root)
|
||
requested_base_url = str(data.get("base_url") or "").strip()
|
||
requested_model = str(data.get("model") or "").strip()
|
||
if requested_base_url and normalize_gpt_base_url(requested_base_url) != config["base_url"]:
|
||
return jsonify({"error": "重翻请求不能临时覆盖 Base URL。请先保存 API 设置，避免把已保存密钥发往未确认端点。"}), 400
|
||
if requested_model and requested_model != config["model"]:
|
||
return jsonify({"error": "重翻请求不能临时覆盖模型。请先保存 API 设置后再重翻。"}), 400
|
||
instruction = str(data.get("instruction") or "")
|
||
raw = call_openai_compatible_chat(config, build_retranslate_messages(row, instruction))
|
||
candidate_html = sanitize_cn_html(raw)
|
||
if not candidate_html:
|
||
return jsonify({"error": "GPT 返回内容为空或无法作为译文保存"}), 502
|
||
candidate_text = strip_tags(candidate_html)
|
||
record = {
|
||
"ts": now_iso(),
|
||
"event": "row_retranslated",
|
||
"row_id": row_id,
|
||
"file": row["file"],
|
||
"model": config["model"],
|
||
"base_url": config["base_url"],
|
||
"jp_text": row["jp_text"],
|
||
"before_cn_text": strip_tags(row.get("current_html") or row.get("cn_html") or ""),
|
||
"candidate_cn_text": candidate_text,
|
||
}
|
||
append_jsonl(session_root / "feedback" / "translation_feedback.jsonl", record)
|
||
return jsonify(
|
||
{
|
||
"status": "ok",
|
||
"row_id": row_id,
|
||
"candidate_html": candidate_html,
|
||
"candidate_text": candidate_text,
|
||
"model": config["model"],
|
||
"updated_at": record["ts"],
|
||
}
|
||
)
|
||
except ValueError as exc:
|
||
return jsonify({"error": str(exc)}), 400
|
||
except RuntimeError as exc:
|
||
return jsonify({"error": str(exc)}), 502
|
||
|
||
@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())
|