feat(magazine): EPUB 3 Media Overlay export with measured sentence timings; tolerate missing mutagen
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
4a20a12f66
commit
268ed391a3
@ -7,6 +7,7 @@ Follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) · versioned wi
|
||||
|
||||
## [Unreleased]
|
||||
### Added
|
||||
- **EPUB 3 with Media Overlays.** *EPUB with audio* in the contents toolbar (and `GET /api/magazine/{id}/epub`) packages the approved articles that have audio into a book whose text highlights as the audio plays. MP3 export now records each sentence's real start and end (`audio/<id>.timings.json`, no text), so new audio highlights exactly; older audio is placed by estimate and the book's description says so. Validated with W3C EPUBCheck 5.2.1 (0 errors, 0 warnings) on a real 12-minute two-article issue.
|
||||
- **Magazine reading view shows sidebars.** A sidebar was kept out of narration and therefore never shown at all. The article text endpoint now also returns `asides` and the reflowed view displays each one where it sits, labelled "Sidebar · not read aloud · p. N", as plain text that is neither a sentence nor highlighted. Only for pages the article alone occupies (on a shared page there is no telling which article a sidebar belongs to); not for advertisement pages; a sidebar you ticked into the narration stays narrated.
|
||||
- **Magazine review: extraction diff.** A region whose text you edited by hand is marked *Edited*, with *Show changes* (word-level comparison against the text as first recognised; removals struck through and marked "−", additions underlined and marked "+") and *Restore recognised text*. The first reading is kept in a new `ocr_text` field across repeated edits; re-reading the region or typing the original back clears it.
|
||||
- **"Continued from page N" now joins a feature's tail to its head.** Until now only the forward "continued on page N" reference linked pages. A tail that says "Fortsetzung von Seite 10" / "Continued from page 10" (found even when set as small furniture or sidebar text, on lines up to 160 characters) is attached to the single article on that page. Skipped when a contents entry anchors the tail, when the page is shared by several articles, or when the marker points forward or at an empty page. Articles gain a `continued_from` list.
|
||||
@ -17,6 +18,7 @@ Follows [Keep a Changelog](https://keepachangelog.com/en/1.0.0/) · versioned wi
|
||||
|
||||
## [1.55.0] — 2026-09-29
|
||||
### Fixed
|
||||
- **MP3 export no longer throws away a finished synthesis when ID3 tagging is unavailable, and article durations no longer read 0 without mutagen.** The deployed image lacked `mutagen` although requirements.txt lists it, so tagging raised after every sentence had been rendered and the contents showed no MP3 lengths. Tagging is now best-effort (a warning), and durations fall back to ffprobe. Rebuild the image to get the tags back.
|
||||
- Tests: the form-pair fixture used the retired prompt version `v3` as "current"; the phone-width case of the Characters sorting test is marked known-failing (sticky name column covers the rating stars, tracked in TODO.md).
|
||||
- **Character scene images.** The scene prompt now states the character's gender and the book's pre-industrial period as hard constraints and asks for the focal person's face, upper body and hands in the near foreground. A sheet whose image checker is unavailable is no longer saved over a previously validated image: generation stops with a clear error instead. The weapon-integrity wording ("one hilt and one blade") is only added for characters who actually carry a weapon (it was planted into every scene).
|
||||
- The character-generation status poll rides out a transient 502/503/504 (proxy or app restart) instead of reporting a false "lost job".
|
||||
|
||||
26
Manual.md
26
Manual.md
@ -338,6 +338,32 @@ Open an article and expand **Pronunciation & spelling** under the player.
|
||||
**Mobile / accessibility:** the lists are ordinary buttons with labels ("Replace X with Y"); marks are underlined, not colour-only.
|
||||
**Privacy:** the agent sends only the flagged word and its surrounding sentence to the LLM you configured in Settings; without an LLM nothing leaves the machine.
|
||||
|
||||
### EPUB with synchronised audio
|
||||
|
||||
**Purpose.** Take an issue to an e-reader or reading app that highlights the sentence being spoken (EPUB 3 *Media Overlays*).
|
||||
|
||||
**Prerequisites.** At least one approved article with generated audio. Only those are included, in contents order.
|
||||
|
||||
**Steps.** Open the magazine's contents and press **EPUB with audio**. The book downloads; open it in a reader that supports
|
||||
read-aloud overlays (for example Thorium Reader, Apple Books, or Readium-based apps).
|
||||
|
||||
**Expected result.** One chapter per article, a table of contents, the MP3s inside the file, and each sentence linked to its
|
||||
place in the audio. The file passes the W3C EPUBCheck (0 errors, 0 warnings on a real two-article, twelve-minute test).
|
||||
|
||||
**How exact is the highlighting.** Audio rendered **after 2026-09-30** records where every sentence starts and ends, so the
|
||||
highlight is exact. Older audio has no such record: its sentences are placed by estimate, proportionally to their length
|
||||
across the real duration, so the highlight can drift by a sentence or two. The book's description states which it is; to get
|
||||
exact timings for an older article, regenerate its MP3.
|
||||
|
||||
**Errors.** No approved article with audio gives "No approved article has audio yet". Audio whose length cannot be read is
|
||||
skipped. Sidebars and figures are not part of the book.
|
||||
|
||||
**Mobile / accessibility.** The book is reflowable text, so text size and contrast are set in the reader. A screen reader
|
||||
reads the text directly; the overlay is an addition, not a replacement.
|
||||
|
||||
**Offline.** The file is self-contained. **Privacy.** It contains the magazine's text and your generated audio, so share it only
|
||||
where the magazine's licence allows.
|
||||
|
||||
### Editing recognised text, and seeing what you changed
|
||||
|
||||
**Purpose.** Fix what OCR got wrong and always be able to see, and undo, what you changed.
|
||||
|
||||
5
TODO.md
5
TODO.md
@ -717,8 +717,9 @@ casting. The original PDF remains the visual source of truth while approved arti
|
||||
caption-capable model is reachable as the fallback; documented as a config-dependent limitation in Manual.md,
|
||||
not a code defect.)*
|
||||
- [~] Add word-level karaoke timing and EPUB 3 Media Overlay export.
|
||||
*(word-level highlighting done as a character-length estimate against the real measured audio duration, not true
|
||||
forced alignment; no EPUB 3 export)*
|
||||
*(EPUB 3 Media Overlay export done 2026-09-30, EPUBCheck-clean, with exact sentence timings for newly rendered audio; word-level
|
||||
highlighting is still a character-length estimate within each sentence, not forced alignment; the in-app MP3 player still
|
||||
aligns its transcript by estimate and could use the new timing records)*
|
||||
- [~] Add mobile/driving mode with large controls, automatic article advance, and hands-free resume.
|
||||
*(driving mode done: two oversized controls, reuses whatever is already playing/queued; not mobile-specific)*
|
||||
- [~] Add OCR-engine benchmark comparisons and per-page fallback selection.
|
||||
|
||||
@ -283,6 +283,26 @@ it can drift on a sentence with heavy punctuation-driven pauses. A checkbox
|
||||
(measured, not estimated — it just waits for `audio.onended`) always still
|
||||
works regardless.
|
||||
|
||||
### EPUB 3 export and sentence timings
|
||||
|
||||
MP3 export renders one TTS call per spoken sentence and joins them, so every sentence's position is known exactly.
|
||||
`synthesize_article_audio` now records it (`write_audio_timings`) in `audio/<id>.timings.json`: per narrated segment its
|
||||
character count and the `[start_ms, end_ms]` of each of its sentences, plus the total duration and whether the audio used
|
||||
interview voices. No text is stored. Older MP3s have no record.
|
||||
|
||||
`core/epub_export.py` (pure packaging, no layout or OCR code) builds the book. `plan_sentences` splits each narrated block
|
||||
with the same sentence rule the MP3 and the reader use and assigns times in three tiers, reported per article: **exact**
|
||||
(record present, segment count and every segment's character count match the current text, and the sentence counts agree),
|
||||
**anchored** (paragraph start/end measured, its sentences share that span by length), **estimated** (no usable record, changed
|
||||
text, or interview audio: each sentence's share of the real duration by spoken length). The package: `mimetype` stored first,
|
||||
`container.xml`, an OPF 3.0 with `media:duration` per overlay and the active class, a nav document, one XHTML + one SMIL + the
|
||||
stored MP3 per article. All text is XML-escaped. `GET /api/magazine/{id}/epub[?article=id]` includes approved articles that
|
||||
have audio, builds into a temp file (removed after sending) and returns `application/epub+zip`. Validated with W3C EPUBCheck
|
||||
5.2.1: 0 fatals, 0 errors, 0 warnings on a real two-article issue.
|
||||
|
||||
`mp3_duration_seconds` uses mutagen and falls back to ffprobe. ID3 tagging failing because mutagen is missing no longer
|
||||
discards a finished synthesis.
|
||||
|
||||
### Sidebars in the reading view
|
||||
|
||||
`article_asides(layout, article, all_articles)` lists the sidebar regions on an article's own pages that are not part of
|
||||
|
||||
223
core/epub_export.py
Normal file
223
core/epub_export.py
Normal file
@ -0,0 +1,223 @@
|
||||
"""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}
|
||||
@ -5524,18 +5524,23 @@ def synthesize_article_audio(doc_dir: Path, article: dict[str, Any], layout: dic
|
||||
voice_for = {speaker: (voice if i == 0 else interview_voice)
|
||||
for i, speaker in enumerate(speakers[:2])}
|
||||
utterances: list[tuple[str, str]] = []
|
||||
for segment in segments:
|
||||
utterance_segment: list[int] = [] # which segment each utterance came from, for the timing record
|
||||
for n, segment in enumerate(segments):
|
||||
spoken = segment.get("spoken") or prepare_text_for_tts(str(segment["text"]), language, glossary)
|
||||
segment_voice = voice_for.get(segment.get("speaker"), voice)
|
||||
utterances.extend((sentence, segment_voice)
|
||||
for sentence in split_sentences_for_tts(spoken))
|
||||
for sentence in split_sentences_for_tts(spoken):
|
||||
utterances.append((sentence, segment_voice))
|
||||
utterance_segment.append(n)
|
||||
total = len(utterances)
|
||||
|
||||
combined = AudioSegment.empty()
|
||||
spans: list[tuple[int, int]] = [] # [start_ms, end_ms) of every utterance in the final MP3
|
||||
for i, (sentence, sentence_voice) in enumerate(utterances):
|
||||
audio_bytes, media_type = _tts_request_audio(sentence, sentence_voice, settings)
|
||||
fmt = "wav" if audio_bytes[:4] == b"RIFF" or "wav" in media_type.lower() else None
|
||||
combined += AudioSegment.from_file(_io.BytesIO(audio_bytes), format=fmt)
|
||||
piece = AudioSegment.from_file(_io.BytesIO(audio_bytes), format=fmt)
|
||||
spans.append((len(combined), len(combined) + len(piece)))
|
||||
combined += piece
|
||||
if progress_cb:
|
||||
progress_cb(i + 1, total)
|
||||
|
||||
@ -5543,11 +5548,68 @@ def synthesize_article_audio(doc_dir: Path, article: dict[str, Any], layout: dic
|
||||
audio_dir.mkdir(parents=True, exist_ok=True)
|
||||
out_path = audio_dir / f"{article['id']}.mp3"
|
||||
combined.export(out_path, format="mp3", bitrate="128k")
|
||||
tag_article_mp3(out_path, article, full_text, album=album,
|
||||
cover_bytes=cover_bytes, cover_mime=cover_mime)
|
||||
try:
|
||||
tag_article_mp3(out_path, article, full_text, album=album,
|
||||
cover_bytes=cover_bytes, cover_mime=cover_mime)
|
||||
except ImportError:
|
||||
# Tags (title, artist, cover, lyrics) are a convenience; losing them must not throw away minutes of
|
||||
# synthesis. The audio and its timings are complete without them.
|
||||
logging.getLogger(__name__).warning("mutagen is not installed: %s was saved without ID3 tags", out_path.name)
|
||||
write_audio_timings(out_path, segments, utterance_segment, spans, interview=bool(interview_voice))
|
||||
return out_path
|
||||
|
||||
|
||||
def mp3_duration_seconds(path: Path) -> float | None:
|
||||
"""Length of an MP3 in seconds: mutagen when installed, else ffprobe, else None (never raises)."""
|
||||
try:
|
||||
from mutagen import File as MutagenFile
|
||||
parsed = MutagenFile(path)
|
||||
if parsed is not None and parsed.info is not None and parsed.info.length:
|
||||
return float(parsed.info.length)
|
||||
except Exception:
|
||||
pass
|
||||
try:
|
||||
out = subprocess.run(["ffprobe", "-v", "error", "-show_entries", "format=duration", "-of",
|
||||
"default=nw=1:nk=1", str(path)], capture_output=True, text=True, timeout=20, check=True).stdout
|
||||
value = float(out.strip())
|
||||
return value if value > 0 else None
|
||||
except (subprocess.SubprocessError, OSError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def write_audio_timings(mp3_path: Path, segments: Sequence[dict[str, Any]], utterance_segment: Sequence[int],
|
||||
spans: Sequence[tuple[int, int]], *, interview: bool) -> None:
|
||||
"""Record where every spoken sentence sits in the MP3 (`<id>.timings.json` beside it).
|
||||
|
||||
Measured, not estimated: the sentences were rendered one by one and joined, so each one's length is
|
||||
known exactly. Holds no text -- only, per narrated segment, its character count (to notice later that
|
||||
the text changed) and the [start, end] milliseconds of each of its sentences. Best-effort: a failed
|
||||
write only costs the EPUB its exact timings.
|
||||
"""
|
||||
per_segment: list[list[list[int]]] = [[] for _ in segments]
|
||||
for n, (start, end) in zip(utterance_segment, spans):
|
||||
per_segment[n].append([int(start), int(end)])
|
||||
record = {
|
||||
"version": 1, "interview": bool(interview), "duration_ms": int(spans[-1][1]) if spans else 0,
|
||||
"segments": [{"chars": len(str(seg.get("text") or "")), "times": times}
|
||||
for seg, times in zip(segments, per_segment)],
|
||||
}
|
||||
try:
|
||||
tmp = mp3_path.with_suffix(".timings.tmp")
|
||||
tmp.write_text(json.dumps(record), "utf-8")
|
||||
tmp.replace(mp3_path.with_suffix(".timings.json"))
|
||||
except OSError:
|
||||
logging.getLogger(__name__).warning("Could not write sentence timings for %s", mp3_path.name)
|
||||
|
||||
|
||||
def read_audio_timings(mp3_path: Path) -> dict[str, Any] | None:
|
||||
try:
|
||||
data = json.loads(mp3_path.with_suffix(".timings.json").read_text("utf-8"))
|
||||
except (OSError, ValueError):
|
||||
return None
|
||||
return data if isinstance(data, dict) and data.get("version") == 1 and isinstance(data.get("segments"), list) else None
|
||||
|
||||
|
||||
def tag_article_mp3(path: Path, article: dict[str, Any], lyrics: str, *,
|
||||
album: str = "", cover_bytes: bytes | None = None,
|
||||
cover_mime: str = "image/png") -> None:
|
||||
|
||||
@ -582,9 +582,8 @@ async def magazine_articles(doc_id: str):
|
||||
a["has_audio"] = audio_path.is_file()
|
||||
if a["has_audio"]:
|
||||
try:
|
||||
from mutagen import File as MutagenFile
|
||||
parsed = MutagenFile(audio_path)
|
||||
a["audio_seconds"] = round(float(parsed.info.length), 2) if parsed and parsed.info else 0
|
||||
length = mag.mp3_duration_seconds(audio_path)
|
||||
a["audio_seconds"] = round(length, 2) if length else 0
|
||||
except Exception:
|
||||
# Duration is presentational; a damaged/mid-write tag must not
|
||||
# make the whole contents endpoint fail. The browser will read
|
||||
@ -2951,6 +2950,62 @@ async def magazine_generate_batch_audio(doc_id: str, request: Request):
|
||||
return {"job_id": job_id, "count": len(article_ids)}
|
||||
|
||||
|
||||
@router.get("/api/magazine/{doc_id}/epub")
|
||||
async def magazine_epub(doc_id: str, article: str = ""):
|
||||
"""The issue (or one article) as an EPUB 3 with Media Overlays: text that highlights as the MP3 plays.
|
||||
|
||||
Only approved articles that already have audio are included, in listing order. Sentence timings are
|
||||
exact for audio rendered since timings were recorded and estimated for older audio (the book's
|
||||
description says which).
|
||||
"""
|
||||
from starlette.background import BackgroundTask
|
||||
from core import epub_export
|
||||
d, meta = _require_magazine(doc_id)
|
||||
layout = mag.load_layout(d)
|
||||
if not layout:
|
||||
raise HTTPException(404, "Not ingested yet")
|
||||
articles = mag.load_articles(d)
|
||||
if article:
|
||||
articles = [a for a in articles if a.get("id") == _article_id(article)]
|
||||
if not articles:
|
||||
raise HTTPException(404, "Article not found")
|
||||
parts = []
|
||||
for a in articles:
|
||||
mp3 = d / "audio" / f"{a['id']}.mp3"
|
||||
if a.get("status") != "approved" or not mp3.is_file():
|
||||
continue
|
||||
blocks = mag.article_text(layout, a, fuse=True, glossary=a.get("glossary"))
|
||||
timings = mag.read_audio_timings(mp3)
|
||||
duration_ms = (timings or {}).get("duration_ms") or 0
|
||||
if not duration_ms:
|
||||
length = mag.mp3_duration_seconds(mp3)
|
||||
if not length:
|
||||
continue # an unreadable MP3 cannot be given timings at all
|
||||
duration_ms = int(length * 1000)
|
||||
parts.append({"title": a.get("title") or "", "author": a.get("author") or "", "blocks": blocks,
|
||||
"mp3": mp3, "duration_ms": duration_ms, "timings": timings})
|
||||
if not parts:
|
||||
raise HTTPException(404, "No approved article has audio yet")
|
||||
language = mag.detect_article_language(b["text"] for p in parts for b in p["blocks"])
|
||||
import tempfile
|
||||
tmp = tempfile.NamedTemporaryFile(prefix="mag-epub-", suffix=".epub", delete=False)
|
||||
tmp.close()
|
||||
out = Path(tmp.name)
|
||||
try:
|
||||
await asyncio.to_thread(epub_export.build_epub, out, title=str(meta.get("title") or "Magazine"),
|
||||
issue=str(meta.get("issue") or ""), language=language or "de",
|
||||
articles=parts, identifier=f"urn:uuid:{doc_id[:8]}-{doc_id[8:12]}-{doc_id[12:16]}-{doc_id[16:20]}-{doc_id[20:32]}")
|
||||
except ValueError as exc:
|
||||
out.unlink(missing_ok=True)
|
||||
raise HTTPException(404, str(exc))
|
||||
except BaseException:
|
||||
out.unlink(missing_ok=True)
|
||||
raise
|
||||
filename = re.sub(r"[^A-Za-z0-9._ -]+", "_", str(meta.get("title") or "magazine")).strip(" ._")[:80] or "magazine"
|
||||
return FileResponse(out, media_type="application/epub+zip", filename=f"{filename}.epub",
|
||||
background=BackgroundTask(lambda: out.unlink(missing_ok=True)))
|
||||
|
||||
|
||||
@router.get("/api/magazine/{doc_id}/audio.zip")
|
||||
async def magazine_audio_zip(doc_id: str):
|
||||
d, meta = _require_magazine(doc_id)
|
||||
|
||||
@ -1325,6 +1325,13 @@ function magUpdateAudioZipLink() {
|
||||
const any = state.articles.some((a) => a.has_audio);
|
||||
link.hidden = !any;
|
||||
if (any) link.href = '/api/magazine/' + state.doc.id + '/audio.zip';
|
||||
// The EPUB carries only approved articles that have audio.
|
||||
const epub = $('mag-epub-link');
|
||||
if (epub) {
|
||||
const ready = state.articles.some((a) => a.has_audio && a.status === 'approved');
|
||||
epub.hidden = !ready;
|
||||
if (ready) epub.href = '/api/magazine/' + state.doc.id + '/epub';
|
||||
}
|
||||
}
|
||||
|
||||
// ── Review (Stage D) ─────────────────────────────────────────────────────────
|
||||
|
||||
@ -138,6 +138,10 @@
|
||||
<a class="btn-sm" id="mag-audio-zip-link" href="#" hidden download>
|
||||
<span class="mdi mdi-folder-zip-outline"></span> Download all (ZIP)
|
||||
</a>
|
||||
<a class="btn-sm" id="mag-epub-link" href="#" hidden download
|
||||
title="An EPUB 3 book whose text highlights while the audio plays (Media Overlays). Approved articles that already have audio.">
|
||||
<span class="mdi mdi-book-open-page-variant-outline" aria-hidden="true"></span> EPUB with audio
|
||||
</a>
|
||||
<span class="mag-audio-export-progress-wrap" hidden id="mag-audio-export-progress-wrap">
|
||||
<progress id="mag-audio-export-progress" max="1" value="0"></progress>
|
||||
<span id="mag-audio-export-count" class="muted"></span>
|
||||
|
||||
@ -1534,3 +1534,24 @@ test('sidebars appear in the reading view in place, labelled, and are not senten
|
||||
await asides.nth(1).click();
|
||||
await expect(page.locator('#mag-article [data-i].is-current')).toHaveCount(1);
|
||||
});
|
||||
|
||||
test('the EPUB link appears only when an approved article has audio', async ({ page }) => {
|
||||
await mockBackend(page, { articleStatus: 'approved', hasAudio: true });
|
||||
await page.goto('/', { waitUntil: 'domcontentloaded' });
|
||||
await page.waitForSelector('#app-skeleton', { state: 'detached' });
|
||||
await navTo(page, 's-magazine');
|
||||
await page.locator('#mag-library-list .mag-card').first().click();
|
||||
const link = page.getByRole('link', { name: /EPUB with audio/ });
|
||||
await expect(link).toBeVisible();
|
||||
await expect(link).toHaveAttribute('href', `/api/magazine/${DOC_ID}/epub`);
|
||||
});
|
||||
|
||||
test('no EPUB link for an approved article without audio', async ({ page }) => {
|
||||
await mockBackend(page, { articleStatus: 'approved', hasAudio: false });
|
||||
await page.goto('/', { waitUntil: 'domcontentloaded' });
|
||||
await page.waitForSelector('#app-skeleton', { state: 'detached' });
|
||||
await navTo(page, 's-magazine');
|
||||
await page.locator('#mag-library-list .mag-card').first().click();
|
||||
await page.waitForSelector('.mag-toc-row');
|
||||
await expect(page.locator('#mag-epub-link')).toBeHidden();
|
||||
});
|
||||
|
||||
300
tests/test_epub_export.py
Normal file
300
tests/test_epub_export.py
Normal file
@ -0,0 +1,300 @@
|
||||
"""EPUB 3 Media Overlay export: package structure, timing tiers, escaping, and the timing record."""
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import tempfile
|
||||
import unittest
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from xml.etree import ElementTree as ET
|
||||
|
||||
os.environ.setdefault("CONFIG_DIR", tempfile.mkdtemp(prefix="ttvc-test-config-"))
|
||||
|
||||
from core import epub_export as ee
|
||||
from core import magazine as mag
|
||||
|
||||
NS = {"o": "http://www.idpf.org/2007/opf", "s": "http://www.w3.org/ns/SMIL", "x": "http://www.w3.org/1999/xhtml"}
|
||||
|
||||
|
||||
def blocks():
|
||||
return [
|
||||
{"type": "title", "text": "Die Zukunft der Batterien"},
|
||||
{"type": "byline", "text": "Von Jane Smith"},
|
||||
{"type": "body", "text": "Erster Satz. Zweiter Satz! Dritter Satz?"},
|
||||
{"type": "body", "text": "Ein einzelner Absatz."},
|
||||
]
|
||||
|
||||
|
||||
def exact_timings(bl=None):
|
||||
bl = bl or blocks()
|
||||
times, t = [], 0
|
||||
segs = []
|
||||
for b in bl:
|
||||
n = len(ee.split_sentences(b["text"]))
|
||||
seg = []
|
||||
for _ in range(n):
|
||||
seg.append([t, t + 1000]); t += 1000
|
||||
segs.append({"chars": len(b["text"]), "times": seg})
|
||||
return {"version": 1, "interview": False, "duration_ms": t, "segments": segs}
|
||||
|
||||
|
||||
class PlanTests(unittest.TestCase):
|
||||
def test_exact_uses_the_recorded_times(self):
|
||||
units, q = ee.plan_sentences(blocks(), 99999, exact_timings())
|
||||
self.assertEqual(q, "exact")
|
||||
self.assertEqual(len(units), 1 + 1 + 3 + 1)
|
||||
self.assertEqual((units[0]["start"], units[0]["end"]), (0, 1000))
|
||||
self.assertEqual(units[-1]["end"], 6000)
|
||||
|
||||
def test_a_sentence_count_mismatch_keeps_the_paragraph_anchored(self):
|
||||
t = exact_timings()
|
||||
t["segments"][2]["times"] = [[2000, 5000]] # recorded as one utterance, text splits into three
|
||||
units, q = ee.plan_sentences(blocks(), 99999, t)
|
||||
self.assertEqual(q, "anchored")
|
||||
para = [u for u in units if u["block"] == 2]
|
||||
self.assertAlmostEqual(para[0]["start"], 2000)
|
||||
self.assertAlmostEqual(para[-1]["end"], 5000)
|
||||
self.assertTrue(all(a["end"] <= b["start"] + 1e-6 for a, b in zip(para, para[1:])))
|
||||
|
||||
def test_changed_text_falls_back_to_an_estimate_over_the_real_duration(self):
|
||||
t = exact_timings()
|
||||
t["segments"][3]["chars"] += 1 # text was edited after the audio was rendered
|
||||
units, q = ee.plan_sentences(blocks(), 10000, t)
|
||||
self.assertEqual(q, "estimated")
|
||||
self.assertAlmostEqual(units[-1]["end"], 10000, places=3)
|
||||
self.assertTrue(all(a["end"] <= b["start"] + 1e-6 for a, b in zip(units, units[1:])))
|
||||
|
||||
def test_no_record_and_interview_audio_are_estimated(self):
|
||||
self.assertEqual(ee.plan_sentences(blocks(), 5000, None)[1], "estimated")
|
||||
t = exact_timings(); t["interview"] = True
|
||||
self.assertEqual(ee.plan_sentences(blocks(), 5000, t)[1], "estimated")
|
||||
|
||||
def test_empty_blocks_yield_nothing(self):
|
||||
self.assertEqual(ee.plan_sentences([{"type": "body", "text": " "}], 1000, None), ([], "estimated"))
|
||||
|
||||
def test_clock_format(self):
|
||||
self.assertEqual(ee.clock(0), "0:00:00.000")
|
||||
self.assertEqual(ee.clock(3_723_456), "1:02:03.456")
|
||||
self.assertEqual(ee.clock(-5), "0:00:00.000")
|
||||
|
||||
|
||||
class BuildTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.dir = Path(tempfile.mkdtemp(prefix="ttvc-epub-"))
|
||||
self.mp3 = self.dir / "a.mp3"
|
||||
self.mp3.write_bytes(b"ID3fakeaudio" * 50)
|
||||
self.out = self.dir / "book.epub"
|
||||
|
||||
def article(self, **kw):
|
||||
base = {"title": "Batterien & <Zukunft>", "author": "J", "blocks": blocks(), "mp3": self.mp3,
|
||||
"duration_ms": 6000, "timings": exact_timings()}
|
||||
base.update(kw)
|
||||
return base
|
||||
|
||||
def build(self, arts=None, **kw):
|
||||
return ee.build_epub(self.out, title=kw.pop("title", "c't"), issue=kw.pop("issue", "9/2026"),
|
||||
language="de", articles=arts or [self.article()], identifier="urn:uuid:test", **kw)
|
||||
|
||||
def test_mimetype_is_first_and_stored_and_every_xml_part_is_well_formed(self):
|
||||
info = self.build()
|
||||
with zipfile.ZipFile(self.out) as z:
|
||||
first = z.infolist()[0]
|
||||
self.assertEqual((first.filename, first.compress_type), ("mimetype", zipfile.ZIP_STORED))
|
||||
self.assertEqual(z.read("mimetype"), b"application/epub+zip")
|
||||
for name in z.namelist():
|
||||
if name.endswith((".xhtml", ".smil", ".opf", ".xml")):
|
||||
ET.fromstring(z.read(name)) # raises on malformed XML
|
||||
self.assertEqual(info["quality"], {"Batterien & <Zukunft>": "exact"})
|
||||
|
||||
def test_manifest_references_exist_ids_are_unique_and_overlay_is_linked(self):
|
||||
self.build()
|
||||
with zipfile.ZipFile(self.out) as z:
|
||||
opf = ET.fromstring(z.read("OEBPS/content.opf"))
|
||||
items = opf.findall(".//o:manifest/o:item", NS)
|
||||
ids = [i.get("id") for i in items]
|
||||
self.assertEqual(len(ids), len(set(ids)))
|
||||
for i in items:
|
||||
self.assertIn("OEBPS/" + i.get("href"), z.namelist())
|
||||
xhtml = next(i for i in items if i.get("media-type") == "application/xhtml+xml" and i.get("id") == "a001")
|
||||
self.assertEqual(xhtml.get("media-overlay"), "a001smil")
|
||||
spine = [r.get("idref") for r in opf.findall(".//o:spine/o:itemref", NS)]
|
||||
self.assertEqual(spine, ["a001"])
|
||||
props = {m.get("property") for m in opf.findall(".//o:metadata/o:meta", NS)}
|
||||
self.assertTrue({"media:duration", "media:active-class", "dcterms:modified"} <= props)
|
||||
|
||||
def test_every_smil_text_target_exists_and_clips_are_ordered(self):
|
||||
self.build()
|
||||
with zipfile.ZipFile(self.out) as z:
|
||||
page = ET.fromstring(z.read("OEBPS/a001.xhtml"))
|
||||
span_ids = {e.get("id") for e in page.iter("{%s}span" % NS["x"])}
|
||||
smil = ET.fromstring(z.read("OEBPS/a001.smil"))
|
||||
pars = smil.findall(".//s:par", NS)
|
||||
self.assertEqual(len(pars), 6)
|
||||
last_end = -1.0
|
||||
for p in pars:
|
||||
self.assertIn(p.find("s:text", NS).get("src").split("#")[1], span_ids)
|
||||
a = p.find("s:audio", NS)
|
||||
def secs(v):
|
||||
h, m, s = v.split(":"); return int(h) * 3600 + int(m) * 60 + float(s)
|
||||
b, e = secs(a.get("clipBegin")), secs(a.get("clipEnd"))
|
||||
self.assertLess(b, e)
|
||||
self.assertGreaterEqual(b, last_end - 1e-9)
|
||||
last_end = e
|
||||
self.assertEqual(a.get("src"), "audio/a001.mp3")
|
||||
|
||||
def test_hostile_text_is_escaped_not_interpreted(self):
|
||||
bl = [{"type": "title", "text": "T"}, {"type": "body", "text": "Ein <script>alert(1)</script> & \"Test\"."}]
|
||||
self.build([self.article(blocks=bl, timings=None, title="x</title><b>")])
|
||||
with zipfile.ZipFile(self.out) as z:
|
||||
raw = z.read("OEBPS/a001.xhtml").decode()
|
||||
self.assertNotIn("<script>", raw)
|
||||
self.assertNotIn("<b>", raw)
|
||||
ET.fromstring(raw)
|
||||
|
||||
def test_headings_and_classes_follow_block_types_and_audio_is_stored_not_recompressed(self):
|
||||
self.build()
|
||||
with zipfile.ZipFile(self.out) as z:
|
||||
page = z.read("OEBPS/a001.xhtml").decode()
|
||||
self.assertIn("<h1>", page)
|
||||
self.assertIn('<p class="byline">', page)
|
||||
self.assertEqual(z.getinfo("OEBPS/audio/a001.mp3").compress_type, zipfile.ZIP_STORED)
|
||||
self.assertEqual(z.read("OEBPS/audio/a001.mp3"), self.mp3.read_bytes())
|
||||
|
||||
def test_several_articles_keep_order_and_an_estimated_one_is_declared(self):
|
||||
arts = [self.article(title="Erster"), self.article(title="Zweiter", timings=None)]
|
||||
info = self.build(arts)
|
||||
self.assertEqual(info["quality"], {"Erster": "exact", "Zweiter": "estimated"})
|
||||
with zipfile.ZipFile(self.out) as z:
|
||||
opf = z.read("OEBPS/content.opf").decode()
|
||||
self.assertIn("estimated", re.search(r"<dc:description>(.*?)</dc:description>", opf).group(1))
|
||||
nav = z.read("OEBPS/nav.xhtml").decode()
|
||||
self.assertLess(nav.index("Erster"), nav.index("Zweiter"))
|
||||
|
||||
def test_nothing_to_package_is_an_error_and_leaves_no_file(self):
|
||||
with self.assertRaises(ValueError):
|
||||
ee.build_epub(self.out, title="t", issue="", language="de", articles=[])
|
||||
with self.assertRaises(ValueError):
|
||||
self.build([self.article(blocks=[{"type": "body", "text": " "}])])
|
||||
self.assertEqual([p.name for p in self.dir.iterdir() if p.suffix in (".epub", ".part")], [])
|
||||
|
||||
def test_a_missing_audio_file_leaves_no_partial_book(self):
|
||||
with self.assertRaises(OSError):
|
||||
self.build([self.article(mp3=self.dir / "missing.mp3")])
|
||||
self.assertEqual([p.name for p in self.dir.iterdir() if p.suffix in (".epub", ".part")], [])
|
||||
|
||||
|
||||
class TimingRecordTests(unittest.TestCase):
|
||||
def test_write_and_read_round_trip_without_any_text(self):
|
||||
d = Path(tempfile.mkdtemp(prefix="ttvc-tm-"))
|
||||
mp3 = d / "article-001.mp3"
|
||||
mp3.write_bytes(b"x")
|
||||
segs = [{"text": "Erster Satz. Zweiter."}, {"text": "Dritter."}]
|
||||
mag.write_audio_timings(mp3, segs, [0, 0, 1], [(0, 900), (900, 1500), (1500, 2100)], interview=False)
|
||||
rec = mag.read_audio_timings(mp3)
|
||||
self.assertEqual(rec["duration_ms"], 2100)
|
||||
self.assertEqual(rec["segments"], [{"chars": 21, "times": [[0, 900], [900, 1500]]}, {"chars": 8, "times": [[1500, 2100]]}])
|
||||
self.assertNotIn("Satz", (d / "article-001.timings.json").read_text())
|
||||
|
||||
def test_missing_or_corrupt_record_reads_as_none(self):
|
||||
d = Path(tempfile.mkdtemp(prefix="ttvc-tm-"))
|
||||
mp3 = d / "x.mp3"
|
||||
self.assertIsNone(mag.read_audio_timings(mp3))
|
||||
(d / "x.timings.json").write_text("not json")
|
||||
self.assertIsNone(mag.read_audio_timings(mp3))
|
||||
(d / "x.timings.json").write_text(json.dumps({"version": 2, "segments": []}))
|
||||
self.assertIsNone(mag.read_audio_timings(mp3))
|
||||
|
||||
|
||||
class Mp3DurationTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
from unittest import mock
|
||||
self.mock = mock
|
||||
self.mp3 = Path(tempfile.mkdtemp(prefix="ttvc-dur-")) / "a.mp3"
|
||||
self.mp3.write_bytes(b"not really audio")
|
||||
|
||||
def test_falls_back_to_ffprobe_when_mutagen_is_missing(self):
|
||||
import subprocess, sys
|
||||
done = subprocess.CompletedProcess([], 0, stdout="12.5\n")
|
||||
with self.mock.patch.dict(sys.modules, {"mutagen": None}), \
|
||||
self.mock.patch.object(mag.subprocess, "run", return_value=done):
|
||||
self.assertEqual(mag.mp3_duration_seconds(self.mp3), 12.5)
|
||||
|
||||
def test_unreadable_audio_is_none_not_an_exception(self):
|
||||
import subprocess, sys
|
||||
with self.mock.patch.dict(sys.modules, {"mutagen": None}), \
|
||||
self.mock.patch.object(mag.subprocess, "run", side_effect=subprocess.CalledProcessError(1, "ffprobe")):
|
||||
self.assertIsNone(mag.mp3_duration_seconds(self.mp3))
|
||||
with self.mock.patch.dict(sys.modules, {"mutagen": None}), \
|
||||
self.mock.patch.object(mag.subprocess, "run", side_effect=FileNotFoundError()):
|
||||
self.assertIsNone(mag.mp3_duration_seconds(self.mp3))
|
||||
|
||||
def test_garbage_or_zero_output_is_none(self):
|
||||
import subprocess, sys
|
||||
for out in ("N/A\n", "0\n", ""):
|
||||
with self.mock.patch.dict(sys.modules, {"mutagen": None}), \
|
||||
self.mock.patch.object(mag.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, stdout=out)):
|
||||
self.assertIsNone(mag.mp3_duration_seconds(self.mp3))
|
||||
|
||||
|
||||
class EndpointTests(unittest.TestCase):
|
||||
def setUp(self):
|
||||
from fastapi import FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
from unittest import mock
|
||||
import routes.magazine as routes
|
||||
self.lib = Path(tempfile.mkdtemp(prefix="ttvc-lib-"))
|
||||
p = mock.patch.object(routes, "_LIB", self.lib)
|
||||
p.start(); self.addCleanup(p.stop)
|
||||
app = FastAPI(); app.include_router(routes.router)
|
||||
self.client = TestClient(app)
|
||||
self.doc = self.client.post("/api/magazine/docs", json={"title": "Test Mag", "issue": "1/2026"}).json()["id"]
|
||||
self.d = self.lib / self.doc
|
||||
|
||||
def seed(self, approved=True, with_audio=True):
|
||||
layout = {"page_count": 1, "pages": [{"page": 1, "width": 595, "height": 842}], "images": [], "blocks": [
|
||||
{"id": "t", "page": 1, "bbox": [0, 0, 1, 1], "type": "title", "text": "Kopf", "reading_order": 0},
|
||||
{"id": "b", "page": 1, "bbox": [0, 2, 1, 1], "type": "body", "text": "Ein Satz. Noch ein Satz.", "reading_order": 1}]}
|
||||
(self.d / "layout.json").write_text(json.dumps(layout))
|
||||
(self.d / "articles.json").write_text(json.dumps({"articles": [{"id": "article-001", "title": "Kopf", "pages": [1],
|
||||
"blocks": ["t", "b"], "images": [], "status": "approved" if approved else "detected", "confidence": .9}]}))
|
||||
import shutil
|
||||
shutil.rmtree(self.d / "audio", ignore_errors=True)
|
||||
if with_audio:
|
||||
(self.d / "audio").mkdir()
|
||||
(self.d / "audio" / "article-001.mp3").write_bytes(b"ID3" + b"0" * 200)
|
||||
(self.d / "audio" / "article-001.timings.json").write_text(json.dumps(
|
||||
{"version": 1, "interview": False, "duration_ms": 3000, "segments": [
|
||||
{"chars": 4, "times": [[0, 500]]}, {"chars": 24, "times": [[500, 1700], [1700, 3000]]}]}))
|
||||
|
||||
def test_downloads_a_valid_epub(self):
|
||||
self.seed()
|
||||
r = self.client.get(f"/api/magazine/{self.doc}/epub")
|
||||
self.assertEqual(r.status_code, 200, r.text[:200])
|
||||
self.assertEqual(r.headers["content-type"], "application/epub+zip")
|
||||
self.assertIn("Test%20Mag.epub", r.headers["content-disposition"])
|
||||
import io
|
||||
with zipfile.ZipFile(io.BytesIO(r.content)) as z:
|
||||
self.assertEqual(z.namelist()[0], "mimetype")
|
||||
self.assertIn("exact", z.read("OEBPS/content.opf").decode())
|
||||
|
||||
def test_unapproved_or_silent_articles_are_not_packaged(self):
|
||||
self.seed(approved=False)
|
||||
self.assertEqual(self.client.get(f"/api/magazine/{self.doc}/epub").status_code, 404)
|
||||
self.seed(with_audio=False)
|
||||
self.assertEqual(self.client.get(f"/api/magazine/{self.doc}/epub").status_code, 404)
|
||||
|
||||
def test_unknown_article_and_bad_document(self):
|
||||
self.seed()
|
||||
self.assertEqual(self.client.get(f"/api/magazine/{self.doc}/epub?article=nope").status_code, 404)
|
||||
self.assertIn(self.client.get("/api/magazine/xx/epub").status_code, (400, 404))
|
||||
|
||||
def test_the_temporary_file_is_removed_after_sending(self):
|
||||
import glob
|
||||
self.seed()
|
||||
before = set(glob.glob(os.path.join(tempfile.gettempdir(), "mag-epub-*")))
|
||||
self.client.get(f"/api/magazine/{self.doc}/epub")
|
||||
self.assertEqual(set(glob.glob(os.path.join(tempfile.gettempdir(), "mag-epub-*"))) - before, set())
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Loading…
Reference in New Issue
Block a user