pdf2epub/scripts/translate.py

#!/usr/bin/env python3
"""Перевод XHTML от pdf2html.py через сторонний book_translator (репозиторий не
модифицируется — используется его штатный CLI и формат файлов).

Схема: наш XHTML -> extracted/chapter_NNN.json + metadata.json ->
03_translate_parallel.py --all -> translations/chapter_NNN_translated.json ->
наш XHTML с сохранённой разметкой.

Курсив/полужирный переживают внешний переводчик в виде маркеров ⟦i⟧…⟦/i⟧:
модель их обычно не трогает. Каждый абзац проверяется на баланс маркеров; при
поломке абзац сохраняется переведённым, но без внутренней разметки.
"""
import argparse
import html
import json
import re
import subprocess
import sys
from pathlib import Path

BLOCK = re.compile(r"^<(p|h1|h2)([^>]*)>(.*)</\1>$")
TAG = re.compile(r"</?([ib])>")
MARK = re.compile(r"⟦(/?)([ib])⟧")
# картинки и пустые блоки не переводим
SKIP = re.compile(r"<img\b")


def detect_language(text):
    """Грубая эвристика: доля кириллицы против латиницы."""
    cyr = sum(1 for c in text if "Ѐ" <= c <= "ӿ")
    lat = sum(1 for c in text if c.isascii() and c.isalpha())
    total = cyr + lat
    if total < 200:
        return "unknown", 0.0
    share = cyr / total
    return ("ru" if share > 0.5 else "en"), share


def to_marks(s):
    """Инлайновые теги -> маркеры, HTML-сущности -> обычные символы."""
    return html.unescape(TAG.sub(lambda m: "⟦%s⟧" % m.group(0)[1:-1], s))


def from_marks(s):
    """Маркеры -> теги. Возвращает (html, ok): ok=False если разметка разъехалась."""
    escaped = html.escape(s, quote=False)
    depth = {"i": 0, "b": 0}
    ok = True
    for slash, tag in MARK.findall(escaped):
        depth[tag] += -1 if slash else 1
        if depth[tag] < 0:
            ok = False
    if any(v != 0 for v in depth.values()):
        ok = False
    if not ok:
        return MARK.sub("", escaped), False
    return MARK.sub(lambda m: "<%s%s>" % (m.group(1), m.group(2)), escaped), True


def parse_blocks(path):
    """[(строка_исходника, tag, attrs, inner)] — только переводимые блоки."""
    out = []
    for i, line in enumerate(path.read_text(encoding="utf-8").splitlines()):
        m = BLOCK.match(line.strip())
        if m and not SKIP.search(m.group(3)) and m.group(3).strip():
            out.append((i, m.group(1), m.group(2), m.group(3)))
    return out


def split_chapters(blocks):
    """Разбивка по h1 — их переводчик держит контекст в пределах главы."""
    chapters, cur = [], []
    for b in blocks:
        if b[1] == "h1" and cur:
            chapters.append(cur)
            cur = []
        cur.append(b)
    if cur:
        chapters.append(cur)
    return chapters


def write_input(chapters, extracted):
    extracted.mkdir(parents=True, exist_ok=True)
    index = []
    for n, ch in enumerate(chapters):
        paragraphs = [to_marks(b[3]) for b in ch]
        title = re.sub("<[^>]+>", "", ch[0][3])[:120] if ch[0][1] == "h1" else "Часть %d" % n
        words = sum(len(p.split()) for p in paragraphs)
        data = {
            "number": n,
            "title": title,
            "paragraphs": paragraphs,
            "paragraph_count": len(paragraphs),
            "word_count": words,
            "char_count": sum(len(p) for p in paragraphs),
            "source_format": "xhtml",
            "extraction_method": "pdf2html-bridge",
        }
        (extracted / ("chapter_%03d.json" % n)).write_text(
            json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
        index.append({"number": n, "title": title,
                      "paragraph_count": len(paragraphs), "word_count": words})
    (extracted / "metadata.json").write_text(json.dumps(
        {"total_chapters": len(chapters),
         "total_words": sum(c["word_count"] for c in index),
         "chapters": index, "source_format": "xhtml"},
        ensure_ascii=False, indent=2), encoding="utf-8")
    return index


def run_translator(repo, workdir, extracted, workers):
    script = repo / "03_translate_parallel.py"
    if not script.exists():
        sys.exit("не найден %s — укажи --repo на клон book_translator" % script)
    cmd = [sys.executable, str(script), "--all",
           "--workers", str(workers), "--extracted-dir", str(extracted)]
    print("запуск:", " ".join(cmd), "(cwd=%s)" % workdir)
    r = subprocess.run(cmd, cwd=workdir)
    if r.returncode != 0:
        sys.exit("book_translator завершился с кодом %d" % r.returncode)


def rebuild(src, chapters, workdir, out):
    lines = src.read_text(encoding="utf-8").splitlines()
    broken = missing = translated = 0
    for n, ch in enumerate(chapters):
        f = workdir / "translations" / ("chapter_%03d_translated.json" % n)
        if not f.exists():
            missing += len(ch)
            continue
        paragraphs = json.loads(f.read_text(encoding="utf-8"))["paragraphs"]
        if len(paragraphs) != len(ch):
            print("глава %d: %d абзацев вместо %d, оставлен оригинал"
                  % (n, len(paragraphs), len(ch)))
            missing += len(ch)
            continue
        for (idx, tag, attrs, _), text in zip(ch, paragraphs):
            body, ok = from_marks(text)
            broken += not ok
            translated += 1
            lines[idx] = "<%s%s>%s</%s>" % (tag, attrs, body, tag)
    out.write_text("\n".join(lines) + "\n", encoding="utf-8")
    print("переведено блоков: %d, без перевода: %d, разметка потеряна в %d"
          % (translated, missing, broken))
    return missing == 0


def verify_output(out, target):
    """Совпадение числа абзацев ещё не значит, что перевод состоялся: внешний
    скрипт при ошибке API молча подставляет оригинал. Проверяем язык результата.
    """
    text = re.sub("<[^>]+>", "", out.read_text(encoding="utf-8"))
    lang, share = detect_language(text)
    stub = text.count("[UNTRANSLATED]")
    print("результат: %s (кириллица %.0f%%), заглушек [UNTRANSLATED]: %d"
          % (lang, share * 100, stub))
    if stub:
        print("ВНИМАНИЕ: внешний переводчик вернул заглушки — проверь его логи и ключ API")
        return False
    if lang != target:
        print("ВНИМАНИЕ: результат не на языке %s — перевода фактически не было" % target)
        return False
    return True


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("source", type=Path, help="XHTML от pdf2html.py")
    ap.add_argument("output", type=Path, help="куда писать переведённый XHTML")
    ap.add_argument("--repo", type=Path, required=True,
                    help="клон vetermanve/book_translator (не модифицируется)")
    ap.add_argument("--workdir", type=Path, required=True,
                    help="рабочий каталог: extracted/ и translations/")
    ap.add_argument("--workers", type=int, default=12)
    ap.add_argument("--target", default="ru", help="целевой язык (по умолчанию ru)")
    ap.add_argument("--force", action="store_true",
                    help="переводить, даже если книга уже на целевом языке")
    ap.add_argument("--check-only", action="store_true",
                    help="только проверить язык и объём, ничего не переводить")
    args = ap.parse_args()

    blocks = parse_blocks(args.source)
    if not blocks:
        sys.exit("в %s нет блоков <p>/<h1>/<h2> — это не вывод pdf2html.py" % args.source)
    plain = " ".join(re.sub("<[^>]+>", "", b[3]) for b in blocks)
    lang, share = detect_language(plain)
    chapters = split_chapters(blocks)
    print("блоков: %d, глав: %d, знаков: %d" % (len(blocks), len(chapters), len(plain)))
    print("язык источника: %s (кириллица %.0f%%)" % (lang, share * 100))

    if lang == args.target and not args.force:
        print("книга уже на языке %s — перевод не требуется (--force чтобы всё равно)"
              % args.target)
        return
    if lang == "unknown":
        print("язык определить не удалось — продолжаю")
    if args.check_only:
        return

    args.workdir.mkdir(parents=True, exist_ok=True)
    extracted = (args.workdir / "extracted").resolve()
    write_input(chapters, extracted)
    run_translator(args.repo.resolve(), args.workdir.resolve(), extracted, args.workers)
    ok = rebuild(args.source, chapters, args.workdir, args.output)
    if not verify_output(args.output, args.target):
        sys.exit("перевод не состоялся — упаковывать этот файл нельзя")
    if not ok:
        print("часть блоков осталась на языке оригинала — см. счётчик выше")


if __name__ == "__main__":
    main()