kai/epub_bilingual_translator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add EPUB translation skill and headless full-book QA with Luna validation

Browse Source
This commit is contained in:
Kai Tan
2026-10-08 12:18:06 +08:00
parent cab13568df
commit 8297e6d2c8
12 changed files with 1016 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/qa/headless.py
+401
View File
@@ -0,0 +1,401 @@
"""Desktop-independent EPUB evidence: semantic links, coverage and optional Chromium."""
import collections
import hashlib
import html
import json
import mimetypes
import posixpath
import re
import threading
import unicodedata
import zipfile
from contextlib import contextmanager
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import quote, unquote, urlsplit
from lxml import etree
from src.common.html import parse_html
from src.assembly.package_writer import package_path
from src.assembly.format_restorer import FormatRestorer
from src.common.data_model import ManifestEntry
from src.qa.epub_audit import audit_epub, compare_audits
def destination(document, href):
url = urlsplit(href)
if url.scheme or url.netloc:
return None
path = posixpath.normpath(posixpath.join(posixpath.dirname(document), unquote(url.path))) if url.path else document
return path + ('#' + unquote(url.fragment) if url.fragment else '')
def book_inventory(path):
with zipfile.ZipFile(path) as z:
opf_path = package_path(z)
root = etree.fromstring(z.read(opf_path), etree.XMLParser(resolve_entities=False, no_network=True))
items = root.xpath('//*[local-name()="manifest"]/*')
resources = {}
docs = {}
toc = []
label_variants = []
for item in items:
name = destination(opf_path, item.get('href', ''))
if not name or name not in z.namelist():
continue
resources[item.get('id')] = name
if item.get('media-type') == 'application/xhtml+xml':
soup = parse_html(z.read(name).decode('utf-8', errors='replace'))
docs[name] = soup
from src.preprocessing.navigation import is_toc
if is_toc(soup, name, item.get('properties', '')):
label_variants.extend({'file': name, 'label': a.get_text(' ', strip=True),
'target': destination(name, a['href'])} for a in soup.find_all('a', href=True))
for nav in soup.find_all('nav'):
if 'toc' not in str(nav.get('epub:type', '')).split() and nav.get('role') != 'doc-toc':
continue
for a in nav.find_all('a', href=True):
toc.append({'file': name, 'label': a.get_text(' ', strip=True),
'target': destination(name, a['href']),
'depth': len(a.find_parents('ol'))})
if not toc:
for item in items:
if item.get('media-type') != 'application/x-dtbncx+xml':
continue
name = resources.get(item.get('id'))
if not name:
continue
ncx = etree.fromstring(z.read(name), etree.XMLParser(resolve_entities=False, no_network=True))
for node in ncx.xpath('//*[local-name()="navMap"]//*[local-name()="navPoint"]'):
labels = node.xpath('./*[local-name()="navLabel"]/*[local-name()="text"]')
hrefs = node.xpath('./*[local-name()="content"]/@src')
if hrefs:
toc.append({'file': name, 'label': ''.join(labels[0].itertext()) if labels else '',
'target': destination(name, hrefs[0]),
'depth': len(node.xpath('ancestor::*[local-name()="navPoint"]')) + 1})
for item in items:
if item.get('media-type') == 'application/x-dtbncx+xml' and resources.get(item.get('id')):
name = resources[item.get('id')]
ncx = etree.fromstring(z.read(name), etree.XMLParser(resolve_entities=False, no_network=True))
for node in ncx.xpath('//*[local-name()="navMap"]//*[local-name()="navPoint"]'):
labels = node.xpath('./*[local-name()="navLabel"]/*[local-name()="text"]')
hrefs = node.xpath('./*[local-name()="content"]/@src')
if hrefs and labels:
label_variants.append({'file': name, 'label': ''.join(labels[0].itertext()), 'target': destination(name, hrefs[0])})
spine = [resources[r] for r in root.xpath('//*[local-name()="spine"]/*/@idref') if r in resources]
binaries = {n: hashlib.sha256(z.read(n)).hexdigest() for n in z.namelist()
if n.lower().endswith(('.png', '.jpg', '.jpeg', '.gif', '.svg', '.ttf', '.otf', '.woff', '.woff2'))}
return {'documents': docs, 'toc': toc, 'spine': spine, 'binaries': binaries,
'package_base': posixpath.dirname(opf_path), 'label_variants': label_variants}
def plain_key(text):
return ''.join(c for c in text if not c.isspace() and not unicodedata.category(c).startswith('P')).casefold()
def is_noteref(node):
return (node.get('role') == 'doc-noteref' or 'noteref' in str(node.get('epub:type', '')).split()
or bool(re.search(r'(?:foot|end)note[-_]?(?:ref|reference)', ' '.join(node.get('class', [])), re.I)))
def notes_report(inventory):
refs = []
docs = inventory['documents']
for name, soup in docs.items():
for a in soup.find_all('a', href=True):
if not is_noteref(a):
continue
target = destination(name, a['href'])
record = {'file': name, 'href': a['href'], 'target': target}
if target is None:
record['classification'] = 'external_note'
else:
file, _, fragment = target.partition('#')
doc = docs.get(file)
node = doc.find(id=fragment) if doc and fragment else doc.body if doc else None
if node is None:
record['classification'] = 'missing_target'
else:
semantic = any(re.search(r'footnote|endnote|doc-note',
str(n.get('role', '')) + ' ' + str(n.get('epub:type', '')) + ' ' + ' '.join(n.get('class', [])), re.I)
for n in [node, *node.parents] if getattr(n, 'attrs', None))
urls = [link['href'] for link in doc.find_all('a', href=True)
if urlsplit(link['href']).scheme in {'http', 'https'} and re.search(r'notes?', link['href'], re.I)]
record['classification'] = 'embedded_note' if semantic else 'external_notes_gateway' if urls else 'non_note_target'
if urls:
record['external_urls'] = sorted(set(urls))
refs.append(record)
counts = collections.Counter(r['classification'] for r in refs)
groups = collections.Counter(r['target'] for r in refs)
findings = []
for target, count in groups.items():
matching = [r for r in refs if r['target'] == target]
if count >= 3 and any(r['classification'] in {'non_note_target', 'external_notes_gateway'} for r in matching):
findings.append({'kind': 'concentrated_note_gateway', 'target': target, 'count': count,
'classification': matching[0]['classification'],
'detail': 'References lead to a shared non-note page; verify external note policy against source.'})
return {'counts': dict(counts), 'findings': findings, 'references': refs}
def compare_navigation(before, after):
a = [(n['target'], n['depth']) for n in before['toc']]
b = [(n['target'], n['depth']) for n in after['toc']]
return [] if a == b else [{'kind': 'navigation_changed', 'detail': 'TOC destinations, order or hierarchy differ from source'}]
def navigation_markup(inventory):
findings = []
for name, doc in inventory['documents'].items():
for nav in doc.find_all('nav'):
if 'toc' not in str(nav.get('epub:type', '')).split() and nav.get('role') != 'doc-toc':
continue
for a in nav.find_all('a', href=True):
if a.parent.name != 'li':
findings.append({'kind': 'nav_link_not_direct_li', 'file': name, 'detail': a['href']})
return findings
def coverage_report(manifest, source, structure=None):
entries = [ManifestEntry.model_validate(e) for e in json.loads(Path(manifest).read_text())]
ids = [e.entry_id for e in entries]
failures = []
restorer = FormatRestorer()
for e in entries:
errors = list(e.validation_errors)
if not e.translated_text or not e.translated_html:
errors.append('missing translation/restored HTML')
elif restorer.validate(e.translated_text, e.placeholders, e.original_text):
errors.append('placeholder validation failed')
elif not restorer.restore(e.translated_text, e.placeholders, original_text=e.original_text)[1]:
errors.append('restoration failed')
if errors:
failures.append({'entry_id': e.entry_id, 'errors': errors})
provenance = 'unverified'
missing_ids = []; extra_ids = []; mismatched_ids = []
if structure:
data = json.loads(Path(structure).read_text())
provenance = 'matched' if data.get('source_sha256') == hashlib.sha256(Path(source).read_bytes()).hexdigest() else 'mismatch'
if provenance == 'matched':
from src.common.data_model import BookStructure
from src.preprocessing.text_extractor import FineGrainedExtractor
from src.preprocessing.navigation import prepare_navigation
book = BookStructure.model_validate(data)
expected = FineGrainedExtractor(complete_book=True).extract(book)
prepare_navigation(source, book, expected)
originals = {e.entry_id: e for e in expected}
actual = {e.entry_id: e for e in entries}
missing_ids = sorted(originals.keys() - actual.keys())
extra_ids = sorted(actual.keys() - originals.keys())
mismatched_ids = [key for key in originals.keys() & actual.keys()
if (originals[key].original_text, originals[key].placeholders) !=
(actual[key].original_text, actual[key].placeholders)]
duplicates = [i for i, n in collections.Counter(ids).items() if n > 1]
return {'total': len(entries), 'accepted': len(entries) - len(failures), 'failed': failures,
'duplicate_ids': duplicates, 'source_provenance': provenance,
'missing_ids': missing_ids, 'extra_ids': extra_ids, 'mismatched_ids': mismatched_ids,
'scope': 'complete-book extraction' if structure and provenance == 'matched' else 'provided manifest only'}
def audit_pair(source, translated, manifest=None, structure=None):
before = book_inventory(source)
after = book_inventory(translated)
sa, ta = audit_epub(source), audit_epub(translated)
issues = compare_audits(sa, ta) + compare_navigation(before, after)
baseline_nav = navigation_markup(before)
issues.extend(f for f in navigation_markup(after) if f not in baseline_nav)
for name, digest in before['binaries'].items():
if after['binaries'].get(name) != digest:
issues.append({'kind': 'binary_resource_changed', 'file': name, 'detail': 'Image/font missing or changed'})
sn, tn = notes_report(before), notes_report(after)
source_refs = {(r['file'], r['href'], r['classification']) for r in sn['references']}
for ref in tn['references']:
if ref['classification'] in {'missing_target', 'non_note_target', 'external_notes_gateway'} and (ref['file'], ref['href'], ref['classification']) not in source_refs:
issues.append({'kind': 'new_note_destination_problem', **ref})
coverage = coverage_report(manifest, source, structure) if manifest else None
if coverage and (not coverage['total'] or coverage['failed'] or coverage['duplicate_ids'] or
coverage['missing_ids'] or coverage['extra_ids'] or coverage['mismatched_ids'] or coverage['source_provenance'] == 'mismatch'):
issues.append({'kind': 'incomplete_or_mismatched_manifest', 'detail': 'Inspect coverage'})
if manifest:
entries = [ManifestEntry.model_validate(x) for x in json.loads(Path(manifest).read_text())]
records = json.loads(Path(structure).read_text()).get('navigation_titles', []) if structure else []
record_by_id = {r['entry_id']: r for r in records}
for e in entries:
if e.file_path == '__navigation__' or not e.translated_text:
continue
name = e.file_path if e.file_path in after['documents'] else posixpath.normpath(posixpath.join(after['package_base'], e.file_path))
doc = after['documents'].get(name)
node = doc.find(id=e.element_id) if doc else None
translated_node = doc.find(attrs={'data-translation-for': e.element_id}) if doc else None
if not translated_node and not (node and 'translation' in node.get('class', [])):
issues.append({'kind': 'translation_not_backfilled', 'file': e.file_path, 'entry_id': e.entry_id})
else:
from src.preprocessing.navigation import normalize_heading_html
expected, valid = FormatRestorer().restore(e.translated_text, e.placeholders, original_text=e.original_text)
if valid:
if e.entry_id in record_by_id:
expected = normalize_heading_html(record_by_id[e.entry_id], expected)
expected_text = re.sub(r'\s+', '', parse_html(expected).get_text())
actual_text = re.sub(r'\s+', '', (translated_node or node).get_text())
if expected_text != actual_text:
issues.append({'kind': 'translation_content_mismatch', 'file': e.file_path, 'entry_id': e.entry_id})
if records:
from src.common.data_model import BookStructure
from src.preprocessing.navigation import navigation_labels
expected_labels = navigation_labels(BookStructure.model_validate(json.loads(Path(structure).read_text())), entries)
for target, label in expected_labels.items():
file, _, frag = target.partition('#')
archive_file = file if file in after['documents'] else posixpath.normpath(posixpath.join(after['package_base'], file))
key = archive_file + ('#' + frag if frag else '')
for variant in after['label_variants']:
if variant['target'] == key and not plain_key(variant['label']).endswith(plain_key(label)):
issues.append({'kind': 'navigation_title_mismatch', **variant, 'expected': label})
return {'source': str(source), 'translated': str(translated),
'source_sha256': hashlib.sha256(Path(source).read_bytes()).hexdigest(),
'translated_sha256': hashlib.sha256(Path(translated).read_bytes()).hexdigest(),
'source_audit': sa, 'translated_audit': ta, 'new_issues': issues,
'coverage': coverage, 'toc_items': len(after['toc']),
'unchanged_binary_resources': len(before['binaries']),
'notes': {'source': sn, 'translated': tn}, 'visual_review': 'pending',
'render_status': 'not_requested', 'semantic_review': 'pending'}, before, after
@contextmanager
def archive_server(paths):
"""Serve allowlisted ZIP members on loopback, without extracting executable book data."""
archives = {key: zipfile.ZipFile(path) for key, path in paths.items()}
class Handler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass
def do_GET(self):
prefix, _, name = unquote(urlsplit(self.path).path).lstrip('/').partition('/')
archive = archives.get(prefix)
if not archive or name not in archive.namelist() or name.endswith('/'):
self.send_error(404); return
data = archive.read(name)
mime = 'text/html' if name.lower().endswith(('.xhtml', '.html', '.htm')) else mimetypes.guess_type(name)[0] or 'application/octet-stream'
self.send_response(200)
self.send_header('Content-Type', mime + ('; charset=utf-8' if mime.startswith('text/') else ''))
self.send_header('Content-Security-Policy', "default-src 'none'; style-src 'self' 'unsafe-inline'; img-src 'self' data:; font-src 'self' data:; script-src 'none'; object-src 'none'")
self.end_headers(); self.wfile.write(data)
server = ThreadingHTTPServer(('127.0.0.1', 0), Handler)
thread = threading.Thread(target=server.serve_forever, daemon=True); thread.start()
try:
yield f'http://127.0.0.1:{server.server_port}'
finally:
server.shutdown(); server.server_close(); thread.join()
for archive in archives.values(): archive.close()
LAYOUT_JS = """() => {
const findings=[];
for(const e of document.querySelectorAll('img, p, h1, h2, h3, h4, table, [data-translation-ordinal]')) {
const r=e.getBoundingClientRect(), s=getComputedStyle(e);
if(!r.width || !r.height) continue;
const id=e.dataset.translationFor || e.id || e.tagName + ':' + [...e.parentElement.children].indexOf(e);
if(r.right > innerWidth+3 || r.left < -3) findings.push({kind:'horizontal_overflow',element:id});
if(e.tagName==='IMG' && !e.naturalWidth) findings.push({kind:'broken_image',element:id});
if(e.dataset.translationRole==='body' && Math.abs(parseFloat(s.textIndent)/parseFloat(s.fontSize)-2)>.1)
findings.push({kind:'chinese_indent',element:id});
if(e.dataset.translationOrdinal && Math.abs(parseFloat(s.fontSize)/parseFloat(getComputedStyle(e.parentElement).fontSize)-1)>.1)
findings.push({kind:'ordinal_font_size',element:id});
}
return findings;
}"""
def render_pair(source, translated, before, after, out, chapters=None, max_documents=4, pages=2, width=900, height=1200):
from playwright.sync_api import sync_playwright
candidates = [n for n in after['spine'] if n in after['documents']]
risky = []
for predicate in (lambda n: re.search(r'part|chapter|ch\d', n, re.I),
lambda n: any(is_noteref(a) for a in after['documents'][n].find_all('a')),
lambda n: after['documents'][n].find('img'),
lambda n: re.search(r'note|credit|index|bibli', n, re.I)):
match = next((n for n in candidates if n not in risky and predicate(n)), None)
if match: risky.append(match)
selected = chapters or (risky + [n for n in candidates if n not in risky])[:max_documents]
if not selected:
raise ValueError('No documents selected for rendering')
for name in selected:
if name not in before['documents'] or name not in after['documents']:
raise ValueError(f'Document absent from source or translation: {name}')
evidence = []; layout = []; clicks = []; browser_version = None
with archive_server({'source': source, 'translated': translated}) as base, sync_playwright() as p:
browser = p.chromium.launch(headless=True)
browser_version = browser.version
try:
context = browser.new_context(viewport={'width': width, 'height': height}, java_script_enabled=False, service_workers='block')
context.route('**/*', lambda route: route.continue_() if route.request.url.startswith(base + '/') else route.abort())
page = context.new_page()
for index, name in enumerate(selected):
for label in ('source', 'translated'):
url = base + '/' + label + '/' + quote(name)
page.goto(url, wait_until='networkidle', timeout=30000)
page.evaluate('() => document.fonts.ready')
findings = page.evaluate(LAYOUT_JS)
layout.extend({'book': label, 'file': name, **f} for f in findings)
total_height = page.evaluate('() => document.documentElement.scrollHeight')
for n in range(min(pages, max(1, (total_height + height - 1)//height))):
page.evaluate('(y) => scrollTo(0,y)', n * height)
png = out / f'{index:03d}-{label}-{n+1}.png'
page.screenshot(path=str(png))
evidence.append({'file': name, 'book': label, 'page': n+1, 'image': png.name})
# Check one real internal note navigation per selected document.
soup = (before if label == 'source' else after)['documents'][name]
links = soup.find_all('a')
for i, a in enumerate(links):
if not a.get('href') or not is_noteref(a) or destination(name, a['href']) is None:
continue
page.goto(url, wait_until='networkidle')
page.locator('a').nth(i).click(timeout=5000)
expected = destination(name, a['href'])
target, _, fragment = expected.partition('#')
actual = unquote(urlsplit(page.url).path).removeprefix('/' + label + '/')
found = not fragment or page.evaluate('(id) => !!document.getElementById(id)', fragment)
classification = next(r['classification'] for r in notes_report(before if label == 'source' else after)['references']
if r['file'] == name and r['href'] == a['href'])
reached = actual == target and found
click = {'book': label, 'file': name, 'href': a['href'], 'target': expected,
'destination_reached': reached, 'classification': classification,
'embedded_note_reached': reached and classification == 'embedded_note'}
if reached and classification == 'embedded_note':
target_doc = (before if label == 'source' else after)['documents'][target]
target_node = target_doc.find(id=fragment) if fragment else target_doc.body
block = target_node.parent if target_node.name == 'a' else target_node
back = next((link for link in block.find_all('a', href=True)
if (destination(target, link['href']) or '').partition('#')[0] == name), None)
if back:
link_index = target_doc.find_all('a').index(back)
page.locator('a').nth(link_index).click(timeout=5000)
back_dest = destination(target, back['href'])
back_file, _, back_fragment = back_dest.partition('#')
actual_back = unquote(urlsplit(page.url).path).removeprefix('/' + label + '/')
click['backlink_reached'] = actual_back == back_file and (not back_fragment or page.evaluate('(id) => !!document.getElementById(id)', back_fragment))
else:
click['backlink_reached'] = None
clicks.append(click)
break
context.close()
finally:
browser.close()
source_flags = {(f['file'], f['kind'], f['element']) for f in layout if f['book'] == 'source'}
return {'engine': 'Chromium headless continuous XHTML; not reader-specific pagination',
'browser_version': browser_version,
'viewport': [width, height], 'sampled_documents': selected, 'images': evidence,
'layout_findings': layout,
'new_layout_findings': [f for f in layout if f['book'] == 'translated' and (f['file'], f['kind'], f['element']) not in source_flags],
'note_clicks': clicks, 'visual_review': 'pending'}
def write_review(out, report):
images = report.get('render', {}).get('images', [])
indexed = {(x['file'], x['page'], x['book']): x for x in images}
rows = ''
for file in dict.fromkeys(x['file'] for x in images):
for page in sorted({x['page'] for x in images if x['file'] == file}):
for book in ('source', 'translated'):
x = indexed.get((file, page, book))
rows += (f'<figure><img loading="lazy" src="{html.escape(x["image"])}"><figcaption>{html.escape(file)} — {book} {page}</figcaption></figure>'
if x else '<figure>No corresponding viewport</figure>')
(out/'review.html').write_text('<!doctype html><meta charset="utf-8"><title>Headless EPUB QA</title>'
'<style>main{display:grid;grid-template-columns:1fr 1fr;gap:1em}img{width:100%}</style>'
'<h1>Headless EPUB QA</h1><p>Generated evidence; visual and semantic review remain pending. See report.json.</p><main>'+rows+'</main>')