"""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.edition.fonts import decoded_fonts
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):
from src.edition.localize import audit_edition
edition = audit_edition(translated)
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)
if edition:
issues.extend({'kind': error, 'detail': 'Inspect edition cover/metadata'} for error in edition['issues'])
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, 'edition': edition, '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()}
fonts = {}
try:
for key, archive in archives.items():
opf_path = package_path(archive)
root = etree.fromstring(archive.read(opf_path), etree.XMLParser(resolve_entities=False, no_network=True))
fonts[key] = decoded_fonts(archive, opf_path, root)
except Exception:
for archive in archives.values(): archive.close()
raise
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 = fonts[prefix].get(name)
if data is None: 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
from src.edition.localize import read_package, edition_record, audit_edition
edition = edition_record(read_package(translated)[1])
added_cover_pages = set()
if edition and not audit_edition(translated)['issues']:
added_cover_pages = {b['file'] for b in edition['cover_bindings'] if b['target'] == edition['cover_path']}
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 after['documents'] or (name not in before['documents'] and name not in added_cover_pages):
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'):
if label == 'source' and name not in before['documents']: continue
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'{html.escape(file)} — {book} {page}'
if x else 'No corresponding viewport')
(out/'review.html').write_text('Headless EPUB QA'
''
'
Headless EPUB QA
Generated evidence; visual and semantic review remain pending. See report.json.