593 lines
23 KiB
Python
593 lines
23 KiB
Python
"""events.json + scaffold.json — the timing layer between alignment and render.
|
|
|
|
The render stage is split in two:
|
|
|
|
build aligns manuscript markers to the transcript, then writes
|
|
events.json — one editable entry per marker: {type, id, time, mapping}
|
|
scaffold.json — the compiled timeline (transcription + narration
|
|
schedule + merged events + total_duration)
|
|
render loads the scaffold's event times (honouring any hand-edits to
|
|
events.json) and renders WITHOUT re-aligning.
|
|
|
|
events.json is the authoritative, GUI-editable seam: a future gnommoweb
|
|
review-stage timeline can read it, let the user drag slide-marker / video-start
|
|
events around, and write it straight back. Because it stores absolute
|
|
final-timeline times with stable ids, edits always win over the fuzzy aligner.
|
|
|
|
mapping values:
|
|
exact a confident fuzzy match to the spoken words
|
|
interpolated placed by the aligner's repair/fallback path, or filled in
|
|
between neighbours because the phrase wasn't found at all
|
|
manual set by a human / the GUI; never recomputed by build
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
from .models import CAMERA_PRESETS
|
|
from .transformer import MarkerTiming, resolve_video_presentation
|
|
|
|
# Per-occurrence presentation fields ALWAYS materialized onto video events (atomic
|
|
# events.json, GUI-ready). Round-tripped as overrides so a stored value drives render.
|
|
_PRESENTATION_KEYS = ("cutout", "layer", "end_on", "take", "object-fit", "object-position")
|
|
# `volume` is materialized SPARSELY — only when actually overridden (inline/GUI/manual),
|
|
# so the videos.json value keeps flowing as the default and only a real override pins it.
|
|
# Round-tripping still carries it whenever present in the event dict.
|
|
_EVENT_OVERRIDE_KEYS = _PRESENTATION_KEYS + ("volume",)
|
|
|
|
EVENTS_FILE = "events.json"
|
|
SCAFFOLD_FILE = "scaffold.json"
|
|
TRANSCRIBED_FILE = "manuscript_transcribed.txt"
|
|
|
|
MAPPING_EXACT = "exact"
|
|
MAPPING_INTERPOLATED = "interpolated"
|
|
MAPPING_MANUAL = "manual"
|
|
|
|
# Confidence at/above this counts as an exact match (mirrors the aligner default).
|
|
_EXACT_THRESHOLD = 0.6
|
|
|
|
|
|
# ── marker classification ─────────────────────────────────────────────────────
|
|
|
|
# Shorthand prefixes that denote a video/narration trigger (mirrors transformer).
|
|
_VIDEO_MARKER_PREFIXES = (
|
|
"video:", "narration:",
|
|
"vft:", "vfb:", "vfm:", "vf2t:", "vf2b:", "vf2m:",
|
|
"vst:", "vsb:", "vsm:",
|
|
"vftp:", "vfbp:", "vfmp:", "vf2tp:", "vf2bp:", "vf2mp:",
|
|
"vstp:", "vsbp:", "vsmp:",
|
|
)
|
|
|
|
|
|
def _ci_contains(d: dict, key: str) -> bool:
|
|
"""Case-insensitive membership: the render lowercases video/audio handles
|
|
(e.g. the marker vst:UnconstrainedLight resolves to videos.json's
|
|
unconstrainedlight), so classification must match case-insensitively too."""
|
|
if not d:
|
|
return False
|
|
if key in d:
|
|
return True
|
|
lk = key.lower()
|
|
return any(k.lower() == lk for k in d)
|
|
|
|
|
|
def marker_type(marker_id: str, slides: dict, videos: dict, audio: dict) -> str:
|
|
"""Classify a marker id as slide / video / audio / camera / other.
|
|
|
|
Prefix-aware and case-insensitive. Video markers carry a shorthand prefix
|
|
(vst:, vfb:, video:, …) and their handle is stored lowercased in videos.json,
|
|
so a marker like `vst:UnconstrainedLight` is a video even though videos.json
|
|
only has `unconstrainedlight`.
|
|
"""
|
|
if _ci_contains(slides, marker_id):
|
|
return "slide"
|
|
if marker_id.startswith(_VIDEO_MARKER_PREFIXES) or _ci_contains(videos, marker_id):
|
|
return "video"
|
|
if marker_id.startswith("audio:") or _ci_contains(audio, marker_id):
|
|
return "audio"
|
|
if marker_id.startswith("A") and len(marker_id) > 1:
|
|
aid = marker_id[1:]
|
|
if aid.isdigit() or _ci_contains(audio, aid):
|
|
return "audio"
|
|
if _ci_contains(CAMERA_PRESETS, marker_id):
|
|
return "camera"
|
|
if marker_id.startswith("end:"):
|
|
return "end"
|
|
return "other"
|
|
|
|
|
|
# Pause-variant video marker prefixes (mirrors transformer). These freeze the
|
|
# narration, so they're what makes narration_time and final_time diverge.
|
|
_PAUSE_MARKER_PREFIXES = (
|
|
"vftp:", "vfbp:", "vfmp:", "vf2tp:", "vf2bp:", "vf2mp:",
|
|
"vstp:", "vsbp:", "vsmp:",
|
|
)
|
|
|
|
|
|
def _lookup_video(marker_id: str, videos: dict):
|
|
"""Case-insensitive videos.json lookup for a video marker (prefix stripped)."""
|
|
if not videos:
|
|
return None
|
|
handle = marker_id.split(":", 1)[1].lower() if ":" in marker_id else marker_id.lower()
|
|
return videos.get(handle) or next(
|
|
(v for k, v in videos.items() if k.lower() == handle), None
|
|
)
|
|
|
|
|
|
def _pause_duration(marker_id: str, videos: dict) -> float:
|
|
"""Seconds a pause-variant video marker freezes the narration for, else 0."""
|
|
if not marker_id.startswith(_PAUSE_MARKER_PREFIXES):
|
|
return 0.0
|
|
handle = marker_id.split(":", 1)[1].lower() if ":" in marker_id else marker_id.lower()
|
|
vs = None
|
|
if videos:
|
|
vs = videos.get(handle) or next(
|
|
(v for k, v in videos.items() if k.lower() == handle), None
|
|
)
|
|
return float(getattr(vs, "pause_narration", 0.0) or 0.0) if vs else 0.0
|
|
|
|
|
|
# ── derive events from aligner output ─────────────────────────────────────────
|
|
|
|
def derive_events(
|
|
marker_timings: list[MarkerTiming],
|
|
slides: dict,
|
|
videos: dict,
|
|
audio: dict,
|
|
) -> list[dict]:
|
|
"""Turn ordered MarkerTiming objects into three-clock event dicts.
|
|
|
|
Each event carries two clocks plus a tweak:
|
|
narration_time — aligner-owned position on the narration/transcript timeline
|
|
(interpolated for markers Whisper couldn't place). Read-only.
|
|
adjustment — a human tweak in seconds layered on narration_time. 0 here;
|
|
preserved across rebuilds by merge_events. The ONLY editable knob.
|
|
final_time — filled by compute_final_times: (narration_time + adjustment)
|
|
shifted forward by every pause before it. What the viewer sees.
|
|
Pause-narration videos also carry pause_duration so the offset is self-describing.
|
|
"""
|
|
events: list[dict] = []
|
|
for t in marker_timings:
|
|
placed = t.timestamp is not None and t.timestamp >= 0
|
|
after_prev = (t.context or "").startswith("(after previous)")
|
|
mapping = (
|
|
MAPPING_EXACT
|
|
if (placed and t.confidence >= _EXACT_THRESHOLD and not after_prev)
|
|
else MAPPING_INTERPOLATED
|
|
)
|
|
etype = marker_type(t.marker_id, slides, videos, audio)
|
|
e = {
|
|
"type": etype,
|
|
"id": t.marker_id,
|
|
"narration_time": round(t.timestamp, 3) if placed else None,
|
|
"adjustment": 0.0,
|
|
"final_time": None,
|
|
"mapping": mapping,
|
|
"confidence": round(t.confidence, 3),
|
|
"context": (t.context or "")[:80],
|
|
}
|
|
# Materialize per-occurrence presentation onto video events so events.json is
|
|
# atomic (each occurrence self-contained, GUI-editable) rather than depending on
|
|
# the handle's videos.json entry. Resolved from the shorthand prefix + any prior
|
|
# override the marker carried.
|
|
if etype == "video":
|
|
vs = _lookup_video(t.marker_id, videos)
|
|
if vs is not None:
|
|
pres = resolve_video_presentation(
|
|
t.marker_id,
|
|
vs,
|
|
t.overrides,
|
|
default_end_on=(
|
|
None if t.marker_id.startswith("narration:") else "next_video"
|
|
),
|
|
)
|
|
e["handle"] = pres["handle"]
|
|
e["cutout"] = pres["cutout"]
|
|
e["layer"] = pres["layer"]
|
|
e["end_on"] = pres["end_on"]
|
|
if pres["take"] is not None:
|
|
e["take"] = pres["take"]
|
|
# volume is sparse: written only when actually overridden, so the
|
|
# videos.json default keeps flowing until someone pins it here.
|
|
if t.overrides and "volume" in t.overrides:
|
|
e["volume"] = pres["volume"]
|
|
# object-fit/position are sparse too: materialize only when non-default
|
|
# so plain center-cover videos keep a clean events.json.
|
|
if pres["object_fit"] != "cover":
|
|
e["object-fit"] = pres["object_fit"]
|
|
if pres["object_position"] != "center":
|
|
e["object-position"] = pres["object_position"]
|
|
pd = _pause_duration(t.marker_id, videos)
|
|
if pd:
|
|
e["pause_duration"] = pd
|
|
events.append(e)
|
|
|
|
_interpolate_narration(events)
|
|
return events
|
|
|
|
|
|
def derive_narration_events(
|
|
narration_schedule: list,
|
|
narration_videos: list,
|
|
pauses: list,
|
|
total_duration: float,
|
|
) -> list[dict]:
|
|
"""Representation-only events mirroring the always-visible talking-head track.
|
|
|
|
One event per narration segment (or a single event for legacy single-file
|
|
narration), shaped like a `video` event — handle, cutout, layer, source_file,
|
|
skip/take, timing — so a GUI can render and lay out every track uniformly
|
|
instead of treating the narration backbone as invisible.
|
|
|
|
These are NOT read back into the render: events_to_marker_timings skips
|
|
`type == "narration"`, because narration timing is owned by the aligner and the
|
|
schedule (it is the clock, and it is multi-file). So this is a read-only mirror,
|
|
added purely for the editing surface — it never changes what renders.
|
|
"""
|
|
# Cutout/layer come from the resolved narration video source (defaults match the
|
|
# talking-head convention used by import/transformer).
|
|
cutout = "talkinghead"
|
|
layer = "below"
|
|
if narration_videos:
|
|
_vs = narration_videos[0][1]
|
|
cutout = getattr(_vs, "cutout", None) or cutout
|
|
layer = getattr(_vs, "layer", None) or layer
|
|
|
|
pause_list = [
|
|
(float(p.narration_time), float(p.duration)) for p in (pauses or [])
|
|
]
|
|
|
|
def _final(offset: float) -> float:
|
|
# Same shift compute_final_times applies: push forward by every pause at or
|
|
# before this point on the narration timeline.
|
|
return round(offset + sum(d for pn, d in pause_list if pn <= offset), 3)
|
|
|
|
def _base(seg_id, source_name, offset, duration, skip, take):
|
|
return {
|
|
"type": "narration",
|
|
"id": seg_id,
|
|
"narration_time": round(offset, 3),
|
|
"adjustment": 0.0,
|
|
"final_time": _final(offset),
|
|
"mapping": MAPPING_EXACT,
|
|
"confidence": 1.0,
|
|
"context": "(talking-head narration)",
|
|
"handle": seg_id,
|
|
"source_file": source_name,
|
|
"cutout": cutout,
|
|
"layer": layer,
|
|
"always_visible": True,
|
|
"end_on": "next_video",
|
|
"skip": round(skip or 0.0, 3),
|
|
"take": (round(take, 3) if take is not None else None),
|
|
"duration": round(duration, 3),
|
|
}
|
|
|
|
events: list[dict] = []
|
|
if narration_schedule:
|
|
for seg in narration_schedule:
|
|
src = getattr(seg.source_path, "name", None) or str(seg.source_path)
|
|
events.append(
|
|
_base(seg.seg_id, src, seg.offset, seg.duration, seg.skip, seg.take)
|
|
)
|
|
elif narration_videos:
|
|
_id, _vs, _ = narration_videos[0]
|
|
events.append(
|
|
_base(
|
|
_id,
|
|
getattr(_vs, "source_file", "") or _id,
|
|
0.0,
|
|
total_duration,
|
|
getattr(_vs, "skip", 0.0),
|
|
getattr(_vs, "take", None),
|
|
)
|
|
)
|
|
return events
|
|
|
|
|
|
def _interpolate_narration(events: list[dict]) -> None:
|
|
"""Fill `narration_time: None` entries by linear interpolation between placed
|
|
neighbours. Head/tail runs spread at +1s steps from the nearest known time (or
|
|
0.0). Mutates events in place."""
|
|
n = len(events)
|
|
i = 0
|
|
while i < n:
|
|
if events[i]["narration_time"] is not None:
|
|
i += 1
|
|
continue
|
|
j = i
|
|
while j < n and events[j]["narration_time"] is None:
|
|
j += 1
|
|
prev_time = events[i - 1]["narration_time"] if i > 0 else None
|
|
next_time = events[j]["narration_time"] if j < n else None
|
|
gap = j - i
|
|
if prev_time is not None and next_time is not None:
|
|
step = (next_time - prev_time) / (gap + 1)
|
|
for k in range(gap):
|
|
events[i + k]["narration_time"] = round(prev_time + step * (k + 1), 3)
|
|
elif prev_time is not None: # trailing run
|
|
for k in range(gap):
|
|
events[i + k]["narration_time"] = round(prev_time + 1.0 * (k + 1), 3)
|
|
elif next_time is not None: # leading run
|
|
base = max(0.0, next_time - gap)
|
|
for k in range(gap):
|
|
events[i + k]["narration_time"] = round(base + 1.0 * k, 3)
|
|
else: # nothing placed at all
|
|
for k in range(gap):
|
|
events[i + k]["narration_time"] = round(1.0 * (i + k), 3)
|
|
i = j
|
|
|
|
|
|
# ── final_time = narration + adjustment + pause offsets ───────────────────────
|
|
|
|
def compute_final_times(events: list[dict]) -> None:
|
|
"""Fill each event's final_time from its (narration_time + adjustment), shifted
|
|
forward by every pause whose own (narration + adjustment) is at or before it.
|
|
|
|
This is the single, clean definition of the narration→final mapping; the render
|
|
is fed the same effective narration times so its output matches these values.
|
|
Mutates events in place.
|
|
"""
|
|
def eff(e):
|
|
n = e.get("narration_time")
|
|
return None if n is None else n + e.get("adjustment", 0.0)
|
|
|
|
pauses = [
|
|
(eff(e), e.get("pause_duration", 0.0), id(e))
|
|
for e in events
|
|
if e.get("pause_duration") and eff(e) is not None
|
|
]
|
|
for e in events:
|
|
ee = eff(e)
|
|
if ee is None:
|
|
e["final_time"] = None
|
|
continue
|
|
shift = sum(dur for (pn, dur, oid) in pauses if pn <= ee and oid != id(e))
|
|
e["final_time"] = round(ee + shift, 3)
|
|
|
|
|
|
# ── re-interpolation on rebuild ───────────────────────────────────────────────
|
|
|
|
def reinterpolate_events(events: list[dict]) -> None:
|
|
"""Recompute every `interpolated` event's narration_time between its anchors.
|
|
|
|
Anchors are the events the aligner placed (mapping `exact`). Interpolated events
|
|
carry no real spoken cue, so their narration_time is only ever a guess spread
|
|
between anchors — recomputed each build. Adjustments are untouched (they layer
|
|
on top). Mutates events in place; call compute_final_times afterwards.
|
|
"""
|
|
for e in events:
|
|
if e.get("mapping") == MAPPING_INTERPOLATED:
|
|
e["narration_time"] = None
|
|
_interpolate_narration(events)
|
|
|
|
|
|
# ── merge: human tweaks survive re-align ──────────────────────────────────────
|
|
|
|
def pair_events_by_ordinal(fresh: list[dict], existing: Optional[list[dict]]):
|
|
"""Yield (fresh_event, matching_existing_event_or_None) pairs.
|
|
|
|
Both lists are in deterministic manuscript order, so the Nth occurrence of a
|
|
given id in `fresh` corresponds to the Nth occurrence in `existing`. Keying by
|
|
id alone would collapse legitimately-repeated markers (e.g. the same video
|
|
reused at two points in the script) onto a single old entry and cross-
|
|
contaminate their timings/tweaks.
|
|
"""
|
|
from collections import defaultdict
|
|
|
|
buckets: dict = defaultdict(list)
|
|
for e in existing or []:
|
|
buckets[e.get("id")].append(e)
|
|
counters: dict = defaultdict(int)
|
|
for e in fresh:
|
|
mid = e.get("id")
|
|
idx = counters[mid]
|
|
counters[mid] += 1
|
|
bucket = buckets.get(mid, [])
|
|
yield e, (bucket[idx] if idx < len(bucket) else None)
|
|
|
|
|
|
def merge_events(fresh: list[dict], existing: Optional[list[dict]]) -> list[dict]:
|
|
"""Carry each id's `adjustment` (the human tweak) from the existing events.json
|
|
onto the freshly-derived events. narration_time and mapping come from `fresh`
|
|
(aligner-owned); the tweak is relative so it stays meaningful even when the
|
|
alignment shifts. New ids are added; ids gone from the manuscript are dropped.
|
|
|
|
Matching is by ordinal (Nth occurrence to Nth occurrence), not by id, so a
|
|
marker reused several times keeps each occurrence's own tweak.
|
|
"""
|
|
if not existing:
|
|
return fresh
|
|
for e, old in pair_events_by_ordinal(fresh, existing):
|
|
if old is not None and old.get("adjustment"):
|
|
e["adjustment"] = old["adjustment"]
|
|
return fresh
|
|
|
|
|
|
# ── events <-> marker timings ─────────────────────────────────────────────────
|
|
|
|
def events_to_marker_timings(events: list[dict]) -> list[MarkerTiming]:
|
|
"""Reconstruct MarkerTiming objects for build_render_plan override.
|
|
|
|
Emits the EFFECTIVE narration time (narration_time + adjustment). The render's
|
|
pause pass then shifts these into final time, reproducing each event's stored
|
|
final_time exactly — so events.json is the single source of the render's timing.
|
|
"""
|
|
timings: list[MarkerTiming] = []
|
|
for e in events:
|
|
# Representation-only base-track events (derive_narration_events) are a
|
|
# read-only mirror of the narration backbone — never fed back as markers.
|
|
if e.get("type") == "narration":
|
|
continue
|
|
n = e.get("narration_time")
|
|
eff = (n + e.get("adjustment", 0.0)) if n is not None else -1.0
|
|
# Carry any stored presentation as overrides so render honors the atomic event
|
|
# (e.g. a GUI edit) over the shorthand/videos.json default. Absent on old events.
|
|
overrides = {k: e[k] for k in _EVENT_OVERRIDE_KEYS if k in e} or None
|
|
timings.append(
|
|
MarkerTiming(
|
|
marker_id=e["id"],
|
|
timestamp=eff,
|
|
context=e.get("context", ""),
|
|
confidence=float(e.get("confidence", 1.0)),
|
|
overrides=overrides,
|
|
)
|
|
)
|
|
return timings
|
|
|
|
|
|
# ── serialization ─────────────────────────────────────────────────────────────
|
|
|
|
def read_events(project_path: Path) -> Optional[list[dict]]:
|
|
p = project_path / EVENTS_FILE
|
|
if not p.exists():
|
|
return None
|
|
try:
|
|
return json.loads(p.read_text(encoding="utf-8"))
|
|
except (json.JSONDecodeError, OSError):
|
|
return None
|
|
|
|
|
|
def write_events(project_path: Path, events: list[dict]) -> Path:
|
|
p = project_path / EVENTS_FILE
|
|
p.write_text(json.dumps(events, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
|
|
return p
|
|
|
|
|
|
def read_scaffold(project_path: Path) -> Optional[dict]:
|
|
p = project_path / SCAFFOLD_FILE
|
|
if not p.exists():
|
|
return None
|
|
try:
|
|
return json.loads(p.read_text(encoding="utf-8"))
|
|
except (json.JSONDecodeError, OSError):
|
|
return None
|
|
|
|
|
|
def write_scaffold(
|
|
project_path: Path,
|
|
events: list[dict],
|
|
transcription: list,
|
|
narration_schedule: list,
|
|
total_duration: float,
|
|
) -> Path:
|
|
"""Write the compiled timeline document render consumes.
|
|
|
|
Stores times + ids (not heavy slide/video objects — those are re-resolved
|
|
from disk at render). Includes the transcription so the future timeline GUI
|
|
can show the spoken words under the draggable markers.
|
|
"""
|
|
scaffold = {
|
|
"version": 1,
|
|
"total_duration": round(total_duration, 3),
|
|
"narration": [
|
|
{
|
|
"seg_id": s.seg_id,
|
|
"source_path": str(s.source_path),
|
|
"skip": round(s.skip, 3),
|
|
"take": (round(s.take, 3) if s.take is not None else None),
|
|
"duration": round(s.duration, 3),
|
|
"offset": round(s.offset, 3),
|
|
}
|
|
for s in (narration_schedule or [])
|
|
],
|
|
"events": events,
|
|
"transcription": [
|
|
{"word": w.word, "start": round(w.start, 3), "end": round(w.end, 3)}
|
|
for w in (transcription or [])
|
|
],
|
|
}
|
|
p = project_path / SCAFFOLD_FILE
|
|
p.write_text(json.dumps(scaffold, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
|
|
return p
|
|
|
|
|
|
# ── transcribed manuscript (for diffing against the script) ───────────────────
|
|
|
|
def _w_start(w) -> float:
|
|
return w["start"] if isinstance(w, dict) else w.start
|
|
|
|
|
|
def _w_word(w) -> str:
|
|
return w["word"] if isinstance(w, dict) else w.word
|
|
|
|
|
|
def _fmt_marker(e: dict) -> str:
|
|
"""Render a marker line, mirroring manuscript.txt so the two files diff cleanly."""
|
|
mid = e["id"]
|
|
t = e.get("type")
|
|
if t == "video":
|
|
return f"[video:{mid}]"
|
|
if t == "audio":
|
|
return f"[audio:{mid}]"
|
|
return f"[{mid}]" # slide / camera / other
|
|
|
|
|
|
def build_transcribed_manuscript(events: list[dict], transcription: list) -> str:
|
|
"""Interleave the aligned markers into the ACTUAL spoken transcript.
|
|
|
|
Walks the transcript along the timeline and drops each marker (at its aligned
|
|
time) on its own line followed by the words spoken until the next marker —
|
|
the same shape as manuscript.txt, but with what was really said. Diff the two
|
|
to see exactly where the recording drifted from the script.
|
|
"""
|
|
# The transcript is the NARRATION timeline, so place markers by their effective
|
|
# narration time (narration_time + adjustment), not final_time.
|
|
def _eff(e):
|
|
nt = e.get("narration_time")
|
|
return None if nt is None else nt + e.get("adjustment", 0.0)
|
|
|
|
evs = sorted(
|
|
[(_eff(e), e) for e in events if _eff(e) is not None],
|
|
key=lambda pair: pair[0],
|
|
)
|
|
words = transcription or []
|
|
n = len(words)
|
|
|
|
lines: list[str] = [
|
|
"# Auto-generated from the spoken transcript with aligned markers.",
|
|
"# Diff against manuscript.txt to see where the recording drifts from the script.",
|
|
"",
|
|
]
|
|
buf: list[str] = []
|
|
|
|
def _flush() -> None:
|
|
if buf:
|
|
lines.append(" ".join(buf))
|
|
buf.clear()
|
|
|
|
wi = 0
|
|
for t, e in evs:
|
|
while wi < n and _w_start(words[wi]) < t:
|
|
buf.append(_w_word(words[wi]))
|
|
wi += 1
|
|
_flush()
|
|
if lines and lines[-1] != "":
|
|
lines.append("")
|
|
lines.append(_fmt_marker(e))
|
|
while wi < n:
|
|
buf.append(_w_word(words[wi]))
|
|
wi += 1
|
|
_flush()
|
|
|
|
return "\n".join(lines).rstrip() + "\n"
|
|
|
|
|
|
def write_transcribed_manuscript(
|
|
project_path: Path, events: list[dict], transcription: list
|
|
) -> Path:
|
|
p = project_path / TRANSCRIBED_FILE
|
|
p.write_text(build_transcribed_manuscript(events, transcription), encoding="utf-8")
|
|
return p
|
|
|
|
|
|
def mapping_summary(events: list[dict]) -> dict:
|
|
"""Count events by mapping for a concise build/render summary."""
|
|
out = {MAPPING_EXACT: 0, MAPPING_INTERPOLATED: 0, MAPPING_MANUAL: 0}
|
|
for e in events:
|
|
out[e.get("mapping", MAPPING_INTERPOLATED)] = out.get(e.get("mapping", MAPPING_INTERPOLATED), 0) + 1
|
|
return out
|