# доходят до результата нетронутыми. Здесь вырезается только из проверок
# языка — иначе английский код утянул бы долю кириллицы ниже порога приёмки.
PRE = re.compile(r"", re.S)
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 = dict.fromkeys(INLINE, 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)
for stale in extracted.glob("chapter_*.json"):
stale.unlink() # главы прошлого прогона иначе уедут в перевод
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 claim_workdir(workdir, source, blocks):
"""Рабочий каталог принадлежит одной книге. Имена chapter_NNN.json у всех
книг одинаковы, а внешний репозиторий пропускает главы, помеченные
готовыми, — без этой привязки вторая книга молча соберётся из перевода
первой."""
digest = hashlib.sha256(
"\n".join(b[3] for b in blocks).encode("utf-8")).hexdigest()[:16]
claim = workdir / "bridge_source.json"
now = {"source": source.name, "blocks": len(blocks), "sha256": digest}
if claim.exists():
was = json.loads(claim.read_text(encoding="utf-8"))
if was.get("sha256") != digest:
sys.exit("рабочий каталог занят другой книгой (%s, блоков %s). "
"Возьми чистый --workdir." % (was.get("source"), was.get("blocks")))
else:
claim.write_text(json.dumps(now, ensure_ascii=False, indent=2),
encoding="utf-8")
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)
# Доля от длины оригинала, ниже которой перевод считается мусором. Совпадение
# числа абзацев ничего не гарантирует: модель иногда возвращает на длинный абзац
# огрызок вида «, v», и прежние ворота такое пропускали.
MIN_LEN_SHARE = 0.25
def rebuild(src, chapters, workdir, out):
lines = src.read_text(encoding="utf-8").splitlines()
broken = missing = translated = junk = untouched = 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, original), text in zip(ch, paragraphs):
plain_src = re.sub("<[^>]+>", "", original)
if len(plain_src) > 120 and len(text) < max(20, len(plain_src) * MIN_LEN_SHARE):
junk += 1
lines[idx] = "<%s%s>%s" % (tag, attrs, original, tag)
continue # оставляем оригинал: английский абзац лучше огрызка
# Абзац, вернувшийся на языке оригинала. Доля кириллицы по всему
# документу такое не ловит: полтора процента в ней тонут, а на
# странице это заметный кусок английского текста.
if len(plain_src) > 150 and detect_language(re.sub("<[^>]+>", "", text))[0] \
== detect_language(plain_src)[0] != "unknown":
untouched += 1
body, ok = from_marks(text)
broken += not ok
translated += 1
lines[idx] = "<%s%s>%s" % (tag, attrs, body, tag)
out.write_text("\n".join(lines) + "\n", encoding="utf-8")
print("переведено блоков: %d, без перевода: %d, разметка потеряна в %d, "
"огрызков заменено оригиналом: %d, осталось на языке оригинала: %d"
% (translated, missing, broken, junk, untouched))
return missing == 0 and junk * 200 <= translated
def verify_output(out, target):
"""Совпадение числа абзацев ещё не значит, что перевод состоялся: внешний
скрипт при ошибке API молча подставляет оригинал. Проверяем язык результата.
"""
text = re.sub("<[^>]+>", "", PRE.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 нет блоков /
/ — это не вывод pdf2html.py" % args.source)
plain = " ".join(re.sub("<[^>]+>", "", b[3]) for b in blocks)
lang, share = detect_language(plain)
chapters = split_chapters(blocks)
pre_n = len(PRE.findall(args.source.read_text(encoding="utf-8")))
print("блоков: %d, глав: %d, знаков: %d, листингов (не переводятся): %d"
% (len(blocks), len(chapters), len(plain), pre_n))
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)
claim_workdir(args.workdir, args.source, blocks)
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()