The longform marker dialect (# heading / [voice:] / [pause] / SSML-lite) was parsed by three independent code paths that already disagreed (client vs server on [pause] units, [voice:] empty, H1-only chapters). This lands the single canonical Python parser; the JS port + cross-impl test follow in slice B. - New backend/services/longform_parser.py — parse_script_to_spans(text, *, default_voice, default_speed) + _parse_chapter_body (the reusable voice→pause →SSML layering the JS twin mirrors). Moves the H1/voice regexes verbatim from audiobook.py (already CodeQL-cleared), reuses parse_pause_markers + ssml_lite unchanged. Coerces None→"" and normalizes CRLF/CR→LF at entry (cross-platform parity so Windows-authored scripts never carry a stray \r). Adds default_speed plumbing (inline SSML speed overrides the per-line default). - audiobook.py: parse_audiobook_script is now a thin wrapper that wraps the canonical span dicts in Span/Chapter/AudiobookPlan — public return type and .to_dict() shape unchanged, all four router call sites untouched. Deleted _parse_spans / _HEADING_RE / _VOICE_RE and the now-dead `import re` + parse_pause_markers import. - tests/fixtures/longform_parser_cases.json — 78-case golden corpus (≥40 required) covering §A–I: H1-only chapters (H2–H6 + `# ` no-title → body), the full pause dialect incl. the NO-MATCH boundary, banker's-rounding ties ([pause 0.5]→0, [pause 1.5]→2), [voice:] empty→default, [voice:[nested]] literal, SSML nesting/spell/unknown-tag, speed override, CRLF, combined precedence. Generated from actual parser output (the truth the JS port must match). - tests/test_longform_parser.py — parametrized over the corpus + None-input + ReDoS-linearity (5000× repeats < 1 s). 130 passed (corpus + test_audiobook + test_pause_markers + test_ssml_lite all green); CJK guard green.
This commit is contained in:
Palash Debnath
@@ -22,26 +22,9 @@ ingestion, the streaming synth job + UI are deferred follow-ups.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Callable, Optional
|
||||
|
||||
from omnivoice.utils.text import parse_pause_markers
|
||||
|
||||
# A Markdown H1 (``# Title``) starts a new chapter. Deeper headings stay in the
|
||||
# body as ordinary text (narrated, not chapter breaks). The title capture
|
||||
# starts with ``\S`` (a non-space) so the leading ``[ \t]+`` and the title's
|
||||
# ``.*`` can't both match the same whitespace run — that overlap is what makes
|
||||
# ``[ \t]+(.+)`` polynomial-time on adversarial tabs (ReDoS). Stripped in code.
|
||||
_HEADING_RE = re.compile(r"^[ \t]*#[ \t]+(\S.*)$", re.MULTILINE)
|
||||
# ``[voice:NAME]`` switches the active narrator for the text that follows. The
|
||||
# content class excludes BOTH brackets (``[^\]\[]``) so a run of nested
|
||||
# ``[voice:`` prefixes can't create overlapping match attempts across
|
||||
# ``finditer`` (the source of the polynomial-time ReDoS). A voice name never
|
||||
# contains a bracket; the value is stripped in code.
|
||||
_VOICE_RE = re.compile(r"\[voice:([^\]\[]*)\]")
|
||||
|
||||
|
||||
@dataclass
|
||||
class Span:
|
||||
"""One contiguous run of text in a single voice, plus trailing silence.
|
||||
@@ -90,73 +73,20 @@ class AudiobookPlan:
|
||||
}
|
||||
|
||||
|
||||
def _parse_spans(body: str, default_voice: Optional[str]) -> list[Span]:
|
||||
"""Split a chapter body into voice-tagged, pause-aware spans."""
|
||||
spans: list[Span] = []
|
||||
cur_voice = default_voice
|
||||
runs: list[tuple[Optional[str], str]] = []
|
||||
last = 0
|
||||
for m in _VOICE_RE.finditer(body):
|
||||
if m.start() > last:
|
||||
runs.append((cur_voice, body[last:m.start()]))
|
||||
cur_voice = (m.group(1).strip() or default_voice)
|
||||
last = m.end()
|
||||
runs.append((cur_voice, body[last:]))
|
||||
|
||||
from services.ssml_lite import parse_ssml_lite, spell_out
|
||||
|
||||
for voice, run_text in runs:
|
||||
# Marker precedence (outer → inner): [voice:] (run split above) →
|
||||
# [pause] (shared dialect) → SSML-lite prosody ([slow]/[fast]/[emphasis]
|
||||
# /[spell]) within the text. The trailing pause attaches to the LAST
|
||||
# SSML segment of the run.
|
||||
for span_text, pause_ms in parse_pause_markers(run_text):
|
||||
t = span_text.strip()
|
||||
if not t and pause_ms == 0:
|
||||
continue # pure whitespace between markers — nothing to render
|
||||
rendered: list[tuple[str, Optional[float]]] = []
|
||||
for seg in (parse_ssml_lite(t) if t else []):
|
||||
st = (spell_out(seg["text"]) if seg["spell"] else seg["text"]).strip()
|
||||
if st:
|
||||
rendered.append((st, seg["speed"]))
|
||||
if not rendered:
|
||||
# Only-markers / empty text but a real pause → carry the silence.
|
||||
if pause_ms > 0:
|
||||
spans.append(Span(voice_id=voice, text="", pause_ms_after=pause_ms))
|
||||
continue
|
||||
for j, (st, sp) in enumerate(rendered):
|
||||
spans.append(Span(
|
||||
voice_id=voice, text=st, speed=sp,
|
||||
pause_ms_after=pause_ms if j == len(rendered) - 1 else 0,
|
||||
))
|
||||
return spans
|
||||
|
||||
|
||||
def parse_audiobook_script(text: str, *, default_voice: Optional[str] = None) -> AudiobookPlan:
|
||||
"""Parse a chapter-delimited script into an :class:`AudiobookPlan`.
|
||||
|
||||
``# Heading`` lines delimit chapters; text before the first heading becomes
|
||||
an untitled lead-in chapter. Chapters with no renderable spans are dropped.
|
||||
Thin wrapper over the canonical :func:`services.longform_parser.
|
||||
parse_script_to_spans` (the single grammar source of truth, #27); wraps its
|
||||
span dicts in the ``Span``/``Chapter``/``AudiobookPlan`` dataclasses so the
|
||||
four router call sites and ``.to_dict()`` shape are unchanged.
|
||||
"""
|
||||
text = text or ""
|
||||
matches = list(_HEADING_RE.finditer(text))
|
||||
if not matches:
|
||||
raw = [(None, text)]
|
||||
else:
|
||||
raw = []
|
||||
intro = text[:matches[0].start()]
|
||||
if intro.strip():
|
||||
raw.append((None, intro))
|
||||
for i, m in enumerate(matches):
|
||||
end = matches[i + 1].start() if i + 1 < len(matches) else len(text)
|
||||
raw.append((m.group(1).strip(), text[m.end():end]))
|
||||
from services.longform_parser import parse_script_to_spans
|
||||
|
||||
chapters: list[Chapter] = []
|
||||
for title, body in raw:
|
||||
spans = _parse_spans(body, default_voice)
|
||||
if not spans:
|
||||
continue
|
||||
chapters.append(Chapter(title=title or f"Chapter {len(chapters) + 1}", spans=spans))
|
||||
chapters = [
|
||||
Chapter(title=c["title"], spans=[Span(**s) for s in c["spans"]])
|
||||
for c in parse_script_to_spans(text, default_voice=default_voice)
|
||||
]
|
||||
return AudiobookPlan(chapters=chapters)
|
||||
|
||||
|
||||
|
||||
@@ -0,0 +1,143 @@
|
||||
"""Canonical longform marker parser (#27) — the single source of grammar truth.
|
||||
|
||||
The longform marker dialect (``# heading``, ``[voice:NAME]``, ``[pause …]``,
|
||||
``[slow]/[fast]/[emphasis]/[spell]``) was parsed by three independent code
|
||||
paths that disagreed (client/server/regex-level). This module is the one
|
||||
canonical Python parser; ``frontend/src/utils/longformParser.js`` is its
|
||||
mechanically-mirrored JS twin, and ``tests/fixtures/longform_parser_cases.json``
|
||||
is the shared golden corpus asserted byte-for-byte against both.
|
||||
|
||||
Pure text→plan, import-light (no torch). Grammar precedence (outer→inner):
|
||||
|
||||
# chapter → [voice:] → [pause] → SSML-lite → [spell]
|
||||
|
||||
It reuses the existing pause dialect (``omnivoice.utils.text.parse_pause_markers``)
|
||||
and SSML-lite (``services.ssml_lite``) verbatim so those modules stay the single
|
||||
home of their sub-grammars.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Optional
|
||||
|
||||
from omnivoice.utils.text import parse_pause_markers
|
||||
|
||||
# A Markdown H1 (``# Title``) starts a new chapter. Deeper headings (``##``…)
|
||||
# stay in the body as ordinary text. The title capture starts with ``\S`` (a
|
||||
# non-space) so the leading ``[ \t]+`` and the title's ``.*`` can't both match
|
||||
# the same whitespace run — that overlap is what makes ``[ \t]+(.+)``
|
||||
# polynomial-time on adversarial tabs (ReDoS). Moved verbatim from
|
||||
# audiobook.py (already CodeQL-cleared). Stripped in code.
|
||||
_HEADING_RE = re.compile(r"^[ \t]*#[ \t]+(\S.*)$", re.MULTILINE)
|
||||
# ``[voice:NAME]`` switches the active narrator. The content class excludes BOTH
|
||||
# brackets (``[^\]\[]``) so nested ``[voice:`` prefixes can't create overlapping
|
||||
# match attempts across ``finditer`` (the ReDoS source). A voice name never
|
||||
# contains a bracket; the value is stripped in code. Empty → default voice.
|
||||
_VOICE_RE = re.compile(r"\[voice:([^\]\[]*)\]")
|
||||
|
||||
|
||||
def _normalize(text: Optional[str]) -> str:
|
||||
"""Coerce None→'' and normalize CRLF/CR→LF so ``$`` (re.MULTILINE) and span
|
||||
text never carry a stray ``\\r`` on Windows-authored scripts — a
|
||||
cross-platform default-behaviour divergence the JS twin mirrors exactly."""
|
||||
if not text:
|
||||
return ""
|
||||
return text.replace("\r\n", "\n").replace("\r", "\n")
|
||||
|
||||
|
||||
def _parse_chapter_body(
|
||||
body: str,
|
||||
*,
|
||||
default_voice: Optional[str] = None,
|
||||
default_speed: Optional[float] = None,
|
||||
) -> list[dict]:
|
||||
"""Voice→pause→SSML layering for ONE chapter body (no chapter split).
|
||||
|
||||
Returns a list of span dicts ``{voice_id, text, pause_ms_after, speed}``.
|
||||
A ``#`` inside ``body`` is NOT treated as a heading here — that is the
|
||||
caller's (chapter-split) concern. The JS twin (``parseChapterBody``) is what
|
||||
``storyToSpans`` calls per spoken track."""
|
||||
spans: list[dict] = []
|
||||
cur_voice = default_voice
|
||||
runs: list[tuple[Optional[str], str]] = []
|
||||
last = 0
|
||||
for m in _VOICE_RE.finditer(body):
|
||||
if m.start() > last:
|
||||
runs.append((cur_voice, body[last:m.start()]))
|
||||
cur_voice = (m.group(1).strip() or default_voice)
|
||||
last = m.end()
|
||||
runs.append((cur_voice, body[last:]))
|
||||
|
||||
from services.ssml_lite import parse_ssml_lite, spell_out
|
||||
|
||||
for voice, run_text in runs:
|
||||
for span_text, pause_ms in parse_pause_markers(run_text):
|
||||
t = span_text.strip()
|
||||
if not t and pause_ms == 0:
|
||||
continue # pure whitespace between markers — nothing to render
|
||||
rendered: list[tuple[str, Optional[float]]] = []
|
||||
for seg in (parse_ssml_lite(t) if t else []):
|
||||
st = (spell_out(seg["text"]) if seg["spell"] else seg["text"]).strip()
|
||||
if st:
|
||||
# Inline SSML speed overrides the per-line default; a plain
|
||||
# segment inherits default_speed.
|
||||
sp = seg["speed"] if seg["speed"] is not None else default_speed
|
||||
rendered.append((st, sp))
|
||||
if not rendered:
|
||||
# Only-markers / empty text but a real pause → carry the silence.
|
||||
if pause_ms > 0:
|
||||
spans.append({"voice_id": voice, "text": "",
|
||||
"pause_ms_after": pause_ms, "speed": None})
|
||||
continue
|
||||
for j, (st, sp) in enumerate(rendered):
|
||||
spans.append({
|
||||
"voice_id": voice, "text": st,
|
||||
"pause_ms_after": pause_ms if j == len(rendered) - 1 else 0,
|
||||
"speed": sp,
|
||||
})
|
||||
return spans
|
||||
|
||||
|
||||
def parse_script_to_spans(
|
||||
text: Optional[str],
|
||||
*,
|
||||
default_voice: Optional[str] = None,
|
||||
default_speed: Optional[float] = None,
|
||||
) -> list[dict]:
|
||||
"""Parse a chapter-delimited script into ``[{"title", "spans": [...]}, …]``.
|
||||
|
||||
span dict == ``{"voice_id": str|None, "text": str, "pause_ms_after": int,
|
||||
"speed": float|None}`` (key order matches ``Span.to_dict()``).
|
||||
|
||||
Contract:
|
||||
* None / "" / whitespace-only input → ``[]``.
|
||||
* CRLF/CR normalized to LF at entry (cross-platform parity).
|
||||
* H1 (``# <non-space>…``) opens a chapter; ``##``…``######`` and ``# ``
|
||||
(no ``\\S`` title) are body.
|
||||
* Each chapter body resets the active voice to ``default_voice``.
|
||||
* A span is dropped iff its text is empty AND pause_ms_after == 0.
|
||||
* Chapters with no surviving spans are dropped; untitled bodies are
|
||||
numbered ``Chapter {kept_so_far + 1}`` (post-drop numbering).
|
||||
"""
|
||||
text = _normalize(text)
|
||||
matches = list(_HEADING_RE.finditer(text))
|
||||
if not matches:
|
||||
raw = [(None, text)]
|
||||
else:
|
||||
raw = []
|
||||
intro = text[:matches[0].start()]
|
||||
if intro.strip():
|
||||
raw.append((None, intro))
|
||||
for i, m in enumerate(matches):
|
||||
end = matches[i + 1].start() if i + 1 < len(matches) else len(text)
|
||||
raw.append((m.group(1).strip(), text[m.end():end]))
|
||||
|
||||
chapters: list[dict] = []
|
||||
for title, body in raw:
|
||||
spans = _parse_chapter_body(body, default_voice=default_voice,
|
||||
default_speed=default_speed)
|
||||
if not spans:
|
||||
continue
|
||||
chapters.append({"title": title or f"Chapter {len(chapters) + 1}",
|
||||
"spans": spans})
|
||||
return chapters
|
||||
+1477
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,48 @@
|
||||
"""Canonical longform parser (#27) — pytest side of the cross-impl golden corpus.
|
||||
|
||||
This suite and ``frontend/src/test/longformParser.test.js`` load the SAME JSON
|
||||
(`tests/fixtures/longform_parser_cases.json`) and assert both impls produce it
|
||||
byte-for-byte. A divergence cannot pass both suites — the side that drifts fails
|
||||
its own assertion against the shared truth.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import pathlib
|
||||
|
||||
import pytest
|
||||
|
||||
from services.longform_parser import parse_script_to_spans
|
||||
|
||||
_FIXTURE = pathlib.Path(__file__).parent / "fixtures" / "longform_parser_cases.json"
|
||||
_CASES = json.loads(_FIXTURE.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("case", _CASES, ids=[c["name"] for c in _CASES])
|
||||
def test_corpus(case):
|
||||
got = parse_script_to_spans(
|
||||
case["input"],
|
||||
default_voice=case["default_voice"],
|
||||
default_speed=case.get("default_speed"),
|
||||
)
|
||||
assert got == case["expected"]
|
||||
|
||||
|
||||
def test_none_input_returns_empty():
|
||||
# The wrapper coerces, but the parser itself must not raise on None.
|
||||
assert parse_script_to_spans(None) == []
|
||||
|
||||
|
||||
def test_corpus_has_enough_cases():
|
||||
# The spec mandates ≥40 cases covering §A–I.
|
||||
assert len(_CASES) >= 40
|
||||
|
||||
|
||||
def test_pathological_inputs_are_linear():
|
||||
# ReDoS guard: adversarial repeats must finish fast (mirrors the JS suite).
|
||||
import time
|
||||
for blob in ("[slow]" * 5000, "[pause" * 5000, "[voice:" * 5000,
|
||||
"# \n" * 5000, "[a]" * 5000):
|
||||
t0 = time.perf_counter()
|
||||
parse_script_to_spans(blob, default_voice="v")
|
||||
assert time.perf_counter() - t0 < 1.0
|
||||
Reference in New Issue
Block a user