224 lines
11 KiB
Python
224 lines
11 KiB
Python
"""EPUB 3 with Media Overlays: an issue's narrated articles as a book that highlights while it plays.
|
||
|
||
Pure packaging: text blocks, the article's MP3 and (when it exists) the sentence timings recorded at
|
||
synthesis go in; a .epub file comes out. No network, no layout or OCR code, so it is testable on its own.
|
||
|
||
Timing quality is reported per article, honestly:
|
||
exact every sentence's start/end was measured when the MP3 was rendered (audio/<id>.timings.json)
|
||
anchored the paragraph's start/end were measured; sentences inside it share that span by length
|
||
estimated no usable record (an MP3 from before timings were kept, or the text changed since): every
|
||
sentence's share of the MP3's real duration is proportional to its spoken length
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import uuid
|
||
import zipfile
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from typing import Any, Sequence
|
||
from xml.sax.saxutils import escape, quoteattr
|
||
|
||
_SENTENCE_SPLIT_RE = re.compile(r"(?<=[.!?…])\s+(?=[A-ZÀ-Ü»„\"'‘])") # same rule the MP3 and the reader use
|
||
_HEADING_TYPES = {"title": "h1", "heading": "h2"}
|
||
_CLASS_TYPES = {"deck": "deck", "byline": "byline", "caption": "caption", "figure_description": "figure"}
|
||
_ACTIVE_CLASS = "-epub-media-overlay-active"
|
||
MAX_ARTICLES = 200
|
||
|
||
|
||
def split_sentences(text: str) -> list[str]:
|
||
return [p.strip() for p in _SENTENCE_SPLIT_RE.split(text or "") if p.strip()]
|
||
|
||
|
||
def clock(ms: float) -> str:
|
||
"""SMIL clock value h:mm:ss.mmm."""
|
||
ms = max(0, int(round(ms)))
|
||
h, rem = divmod(ms, 3_600_000)
|
||
m, rem = divmod(rem, 60_000)
|
||
return f"{h}:{m:02d}:{rem // 1000:02d}.{rem % 1000:03d}"
|
||
|
||
|
||
def _weight(text: str) -> float:
|
||
words = text.split()
|
||
return sum(max(1, len(re.sub(r"[.,!?;:\"'()«»„“”]", "", w))) + 0.6 for w in words) or 1.0
|
||
|
||
|
||
def plan_sentences(blocks: Sequence[dict[str, Any]], duration_ms: float,
|
||
timings: dict[str, Any] | None) -> tuple[list[dict[str, Any]], str]:
|
||
"""Split narrated blocks into sentences with [start, end] milliseconds; returns (units, quality).
|
||
|
||
Each unit: {block: index, text, start, end}. Blocks with no text are skipped.
|
||
"""
|
||
narrated = [(i, b) for i, b in enumerate(blocks) if (b.get("text") or "").strip()]
|
||
units: list[dict[str, Any]] = []
|
||
for i, b in narrated:
|
||
for s in split_sentences(b["text"]):
|
||
units.append({"block": i, "text": s, "spoken": s})
|
||
if not units:
|
||
return [], "estimated"
|
||
|
||
segs = (timings or {}).get("segments") if timings and not timings.get("interview") else None
|
||
usable = isinstance(segs, list) and len(segs) == len(narrated) and all(
|
||
isinstance(s, dict) and s.get("chars") == len(str(b["text"])) and isinstance(s.get("times"), list)
|
||
and s["times"] for (_, b), s in zip(narrated, segs))
|
||
if usable:
|
||
quality = "exact"
|
||
for (i, b), seg in zip(narrated, segs):
|
||
mine = [u for u in units if u["block"] == i]
|
||
times = seg["times"]
|
||
if len(times) == len(mine):
|
||
for u, (s0, s1) in zip(mine, times):
|
||
u["start"], u["end"] = float(s0), float(s1)
|
||
else: # counts disagree: keep the paragraph exact, split by length
|
||
quality = "anchored" if quality == "exact" else quality
|
||
a, z = float(times[0][0]), float(times[-1][1])
|
||
total = sum(_weight(u["text"]) for u in mine)
|
||
t = a
|
||
for u in mine:
|
||
share = (z - a) * _weight(u["text"]) / total
|
||
u["start"], u["end"] = t, t + share
|
||
t += share
|
||
return units, quality
|
||
|
||
total = sum(_weight(u["text"]) for u in units)
|
||
t = 0.0
|
||
for u in units:
|
||
share = duration_ms * _weight(u["text"]) / total
|
||
u["start"], u["end"] = t, t + share
|
||
t += share
|
||
return units, "estimated"
|
||
|
||
|
||
def _xhtml(title: str, lang: str, body: str) -> str:
|
||
return (
|
||
'<?xml version="1.0" encoding="utf-8"?>\n<!DOCTYPE html>\n'
|
||
f'<html xmlns="http://www.w3.org/1999/xhtml" xmlns:epub="http://www.idpf.org/2007/ops" '
|
||
f'xml:lang={quoteattr(lang)} lang={quoteattr(lang)}>\n<head><meta charset="utf-8"/>'
|
||
f'<title>{escape(title)}</title><link rel="stylesheet" type="text/css" href="style.css"/></head>\n'
|
||
f'<body>\n{body}\n</body>\n</html>\n')
|
||
|
||
|
||
_CSS = f""".deck {{ font-style: italic; }}
|
||
.byline {{ color: #555; }}
|
||
.caption, .figure {{ font-size: .9em; color: #444; }}
|
||
.{_ACTIVE_CLASS} {{ background: #ffe08a; color: #000; }}
|
||
"""
|
||
|
||
|
||
def build_epub(dest: Path, *, title: str, issue: str, language: str,
|
||
articles: Sequence[dict[str, Any]], identifier: str | None = None) -> dict[str, Any]:
|
||
"""Write the EPUB to `dest`; returns {articles, duration_ms, quality: {article title: tier}}.
|
||
|
||
Each article: {"title", "author", "blocks": [article_text dicts], "mp3": Path, "duration_ms", "timings"}.
|
||
Raises ValueError when there is nothing to package. The file is written to a temporary name and moved
|
||
into place, so a failure never leaves a half-written book.
|
||
"""
|
||
if not articles:
|
||
raise ValueError("No article has audio to package")
|
||
if len(articles) > MAX_ARTICLES:
|
||
raise ValueError(f"At most {MAX_ARTICLES} articles per book")
|
||
lang = re.sub(r"[^A-Za-z-]", "", language or "") or "de"
|
||
uid = identifier or f"urn:uuid:{uuid.uuid4()}"
|
||
modified = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||
|
||
manifest: list[str] = []
|
||
spine: list[str] = []
|
||
nav_items: list[str] = []
|
||
meta_durations: list[str] = []
|
||
files: list[tuple[str, bytes | Path, bool]] = [] # (path in package, content, compress?)
|
||
quality: dict[str, str] = {}
|
||
total_ms = 0.0
|
||
|
||
for n, art in enumerate(articles, 1):
|
||
key = f"a{n:03d}"
|
||
units, tier = plan_sentences(art["blocks"], float(art["duration_ms"]), art.get("timings"))
|
||
if not units:
|
||
continue
|
||
quality[str(art.get("title") or key)] = tier
|
||
|
||
body: list[str] = ['<section epub:type="chapter">']
|
||
pars: list[str] = []
|
||
current, open_tag = -1, ""
|
||
for k, u in enumerate(units, 1):
|
||
if u["block"] != current:
|
||
if open_tag:
|
||
body.append(f"</{open_tag}>")
|
||
btype = str(art["blocks"][u["block"]].get("type") or "body")
|
||
open_tag = _HEADING_TYPES.get(btype, "p")
|
||
cls = _CLASS_TYPES.get(btype)
|
||
body.append(f"<{open_tag}" + (f' class="{cls}"' if cls else "") + ">")
|
||
current = u["block"]
|
||
sid = f"{key}s{k:04d}"
|
||
body.append(f'<span id="{sid}">{escape(u["text"])}</span> ')
|
||
start = u["start"]
|
||
end = max(u["end"], start + 1)
|
||
pars.append(f'<par id="{key}p{k:04d}"><text src="{key}.xhtml#{sid}"/>'
|
||
f'<audio src="audio/{key}.mp3" clipBegin="{clock(start)}" clipEnd="{clock(end)}"/></par>')
|
||
body.append(f"</{open_tag}>")
|
||
body.append("</section>")
|
||
|
||
smil = ('<?xml version="1.0" encoding="utf-8"?>\n'
|
||
'<smil xmlns="http://www.w3.org/ns/SMIL" xmlns:epub="http://www.idpf.org/2007/ops" version="3.0">\n'
|
||
f'<body><seq id="{key}seq" epub:textref="{key}.xhtml" epub:type="bodymatter chapter">\n'
|
||
+ "\n".join(pars) + "\n</seq></body>\n</smil>\n")
|
||
dur = float(art["duration_ms"])
|
||
total_ms += dur
|
||
a_title = str(art.get("title") or f"Article {n}")
|
||
files.append((f"OEBPS/{key}.xhtml", _xhtml(a_title, lang, "\n".join(body)).encode("utf-8"), True))
|
||
files.append((f"OEBPS/{key}.smil", smil.encode("utf-8"), True))
|
||
files.append((f"OEBPS/audio/{key}.mp3", Path(art["mp3"]), False))
|
||
manifest += [
|
||
f'<item id="{key}" href="{key}.xhtml" media-type="application/xhtml+xml" media-overlay="{key}smil"/>',
|
||
f'<item id="{key}smil" href="{key}.smil" media-type="application/smil+xml"/>',
|
||
f'<item id="{key}audio" href="audio/{key}.mp3" media-type="audio/mpeg"/>',
|
||
]
|
||
spine.append(f'<itemref idref="{key}"/>')
|
||
nav_items.append(f'<li><a href="{key}.xhtml">{escape(a_title)}</a></li>')
|
||
meta_durations.append(f'<meta property="media:duration" refines="#{key}smil">{clock(dur)}</meta>')
|
||
|
||
if not spine:
|
||
raise ValueError("No article has narratable text")
|
||
|
||
tiers = sorted(set(quality.values()))
|
||
note = ("Sentence timings: " + ", ".join(tiers)
|
||
+ (" (estimated where the audio predates timing records)" if "estimated" in tiers else ""))
|
||
full_title = f"{title} — {issue}" if issue else title
|
||
opf = ('<?xml version="1.0" encoding="utf-8"?>\n'
|
||
f'<package xmlns="http://www.idpf.org/2007/opf" version="3.0" unique-identifier="bookid" xml:lang="{lang}">\n'
|
||
'<metadata xmlns:dc="http://purl.org/dc/elements/1.1/">\n'
|
||
f'<dc:identifier id="bookid">{escape(uid)}</dc:identifier>\n<dc:title>{escape(full_title)}</dc:title>\n'
|
||
f'<dc:language>{lang}</dc:language>\n<dc:description>{escape(note)}</dc:description>\n'
|
||
f'<meta property="dcterms:modified">{modified}</meta>\n'
|
||
+ "\n".join(meta_durations) + f'\n<meta property="media:duration">{clock(total_ms)}</meta>\n'
|
||
f'<meta property="media:active-class">{_ACTIVE_CLASS}</meta>\n</metadata>\n<manifest>\n'
|
||
'<item id="nav" href="nav.xhtml" media-type="application/xhtml+xml" properties="nav"/>\n'
|
||
'<item id="css" href="style.css" media-type="text/css"/>\n' + "\n".join(manifest) +
|
||
'\n</manifest>\n<spine>\n' + "\n".join(spine) + '\n</spine>\n</package>\n')
|
||
nav = _xhtml(full_title, lang,
|
||
'<nav epub:type="toc" id="toc"><h1>' + escape(full_title) + '</h1><ol>\n'
|
||
+ "\n".join(nav_items) + '\n</ol></nav>')
|
||
container = ('<?xml version="1.0" encoding="utf-8"?>\n'
|
||
'<container version="1.0" xmlns="urn:oasis:names:tc:opendocument:xmlns:container">'
|
||
'<rootfiles><rootfile full-path="OEBPS/content.opf" media-type="application/oebps-package+xml"/>'
|
||
'</rootfiles></container>\n')
|
||
|
||
tmp = dest.with_suffix(".part")
|
||
try:
|
||
with zipfile.ZipFile(tmp, "w") as z:
|
||
z.writestr(zipfile.ZipInfo("mimetype"), "application/epub+zip", compress_type=zipfile.ZIP_STORED)
|
||
z.writestr("META-INF/container.xml", container, compress_type=zipfile.ZIP_DEFLATED)
|
||
z.writestr("OEBPS/content.opf", opf, compress_type=zipfile.ZIP_DEFLATED)
|
||
z.writestr("OEBPS/nav.xhtml", nav, compress_type=zipfile.ZIP_DEFLATED)
|
||
z.writestr("OEBPS/style.css", _CSS, compress_type=zipfile.ZIP_DEFLATED)
|
||
for name, content, compress in files:
|
||
method = zipfile.ZIP_DEFLATED if compress else zipfile.ZIP_STORED
|
||
if isinstance(content, Path):
|
||
z.write(content, name, compress_type=method)
|
||
else:
|
||
z.writestr(name, content, compress_type=method)
|
||
tmp.replace(dest)
|
||
except BaseException:
|
||
tmp.unlink(missing_ok=True)
|
||
raise
|
||
return {"articles": len(spine), "duration_ms": int(total_ms), "quality": quality}
|