From cea6005bf441d492188e7a3744676b8f1dd5faf6 Mon Sep 17 00:00:00 2001 From: Sami Ahmed Date: Mon, 27 Jul 2026 02:38:37 -0700 Subject: [PATCH] Initial commit: sermon-clean v0.1.0 Find, cut, paste replacements in sermon audio. - engine.py: SermonClean core (BadSegment, trim, concat, splice verify) - cli.py: find/cut/paste/pipe subcommands - elevenlabs.py: optional ElevenLabs rendering hook - 25 tests passing (timestamp parsing, BadSegment validation, end-to-end trim on real audio) - pyproject.toml: pip-installable; dep only on ffmpeg/ffprobe system-wide - README + LICENSE (MIT) + examples/segs.json Validated against the krystie audio cache (OGG/Opus 48kHz mono). --- .gitignore | 30 +++ LICENSE | 21 ++ examples/segs.json | 14 ++ pyproject.toml | 55 +++++ sermon_clean/__init__.py | 34 ++++ sermon_clean/cli.py | 256 ++++++++++++++++++++++++ sermon_clean/elevenlabs.py | 84 ++++++++ sermon_clean/engine.py | 400 +++++++++++++++++++++++++++++++++++++ tests/test_engine.py | 99 +++++++++ 9 files changed, 993 insertions(+) create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 examples/segs.json create mode 100644 pyproject.toml create mode 100644 sermon_clean/__init__.py create mode 100644 sermon_clean/cli.py create mode 100644 sermon_clean/elevenlabs.py create mode 100644 sermon_clean/engine.py create mode 100644 tests/test_engine.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..3b42ac5 --- /dev/null +++ b/.gitignore @@ -0,0 +1,30 @@ +# Python +__pycache__/ +*.py[cod] +*.egg-info/ +build/ +dist/ +.eggs/ +*.egg + +# virtualenvs +venv/ +.venv/ +env/ + +# IDE +.vscode/ +.idea/ + +# OS +.DS_Store +Thumbs.db + +# Runtime artifacts (created by the tool, not committed) +replacements/ +*.tmp + +# Test artifacts +.pytest_cache/ +.coverage +htmlcov/ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..6a21a76 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 Sami Ahmed + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/examples/segs.json b/examples/segs.json new file mode 100644 index 0000000..251c08e --- /dev/null +++ b/examples/segs.json @@ -0,0 +1,14 @@ +[ + { + "start": 1298.0, + "end": 1302.0, + "reason": "misspoke", + "replacement_text": "the actual sentence you meant to say" + }, + { + "start": 1450.3, + "end": 1453.1, + "reason": "misspoke", + "replacement_text": "the corrected phrase" + } +] diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..3094dd3 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,55 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "sermon-clean" +version = "0.1.0" +description = "Find bad segments in sermon audio, cut them out, splice in ElevenLabs replacements." +readme = "README.md" +requires-python = ">=3.9" +license = {text = "MIT"} +authors = [ + {name = "Sami Ahmed", email = "hello@sami-ahmed.net"}, +] +keywords = ["audio", "ffmpeg", "sermon", "editing", "elevenlabs"] +classifiers = [ + "Development Status :: 3 - Alpha", + "Intended Audience :: End Users/Desktop", + "License :: OSI Approved :: MIT License", + "Operating System :: POSIX :: Linux", + "Programming Language :: Python :: 3", + "Programming Language :: Python :: 3.9", + "Programming Language :: Python :: 3.10", + "Programming Language :: Python :: 3.11", + "Programming Language :: Python :: 3.12", + "Topic :: Multimedia :: Sound/Audio :: Editors", +] + +# Runtime dependencies — requests is only needed if you use ElevenLabs rendering. +# ffmpeg / ffprobe must be installed system-wide. +dependencies = [ + "requests>=2.31", +] + +[project.optional-dependencies] +# Whisper integration (optional, adds auto-detect mode) +auto = [ + "faster-whisper>=1.0.0", +] +test = [ + "pytest>=7.0", +] + +[project.scripts] +sermon-clean = "sermon_clean.cli:main" + +[project.urls] +Homepage = "https://git.sami/sami7777/sermon-clean" +Issues = "https://git.sami/sami7777/sermon-clean/issues" + +[tool.setuptools] +packages = ["sermon_clean"] + +[tool.setuptools.package-data] +sermon_clean = ["py.typed"] diff --git a/sermon_clean/__init__.py b/sermon_clean/__init__.py new file mode 100644 index 0000000..ed9ec26 --- /dev/null +++ b/sermon_clean/__init__.py @@ -0,0 +1,34 @@ +"""sermon-clean — find bad segments in sermon audio, cut them out, splice in ElevenLabs replacements. + +Three-step pipeline: + 1. FIND — locate bad segments (manual timestamps, ffmpeg silence detection, or + faster-whisper transcription with optional profanity/keyword flag). + 2. CUT — trim the original audio into N+1 good segments using ffmpeg. + 3. PASTE — render replacement clips via ElevenLabs (or supply pre-rendered MP3s), + splice everything back together with a bit-exact concat. + +The default mode is "I already know the timestamps" (fastest, no transcribe cost). +Auto-detect mode transcribes and prints a candidate list for you to confirm/edit +before splicing. Interactive mode lets you scrub through and press space to mark. +""" + +from .engine import ( + SermonClean, + BadSegment, + Replacement, + parse_timestamp, + format_timestamp, + probe_duration, + probe_codec, +) + +__version__ = "0.1.0" +__all__ = [ + "SermonClean", + "BadSegment", + "Replacement", + "parse_timestamp", + "format_timestamp", + "probe_duration", + "probe_codec", +] diff --git a/sermon_clean/cli.py b/sermon_clean/cli.py new file mode 100644 index 0000000..8494d76 --- /dev/null +++ b/sermon_clean/cli.py @@ -0,0 +1,256 @@ +"""CLI for sermon-clean — the three-step audio editor. + +Usage: + sermon-clean find sermon.ogg --silence-threshold -30 # shows silence gaps + sermon-clean cut sermon.ogg --segments segs.json --replacements rep/*.mp3 -o fixed.ogg + sermon-clean pipe sermon.ogg --bad "21:38-21:42, 1450.3-1453.1" \\ + --replace "21:38-21:42:the actual sentence" \\ + --replace "1450.3-1453.1:the corrected phrase" +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +from .engine import ( + SermonClean, + detect_silence, + format_timestamp, + probe_duration, +) + +USAGE = """sermon-clean — find bad segments, cut, paste replacements + +Three-step workflow: + + 1. FIND: locate bad segments + - manual: --bad "MM:SS-MM:SS[, MM:SS-MM:SS, ...]" + - silence: --find-silence (prints gaps between speech) + - explicit JSON file: --segments-file segs.json + + 2. CUT: Trim the original audio around the bad windows. + (happens automatically when you run 'paste' or 'pipe') + + 3. PASTE: Render replacement clips via ElevenLabs, or supply pre-rendered ones. + - ElevenLabs: --render-elevenlabs TEXT (one per --bad segment) + - Pre-rendered: --replacements rep/*.mp3 (one per --bad segment) + +'pipe' runs all three in one command. +""" + + +def _parse_bad_spec(spec: str) -> list[tuple[float, float, str]]: + """Parse 'MM:SS-MM:SS:TEXT, MM:SS-MM:SS:TEXT' into [(start, end, text), ...].""" + from .engine import parse_timestamp + out = [] + for item in spec.split(","): + item = item.strip() + if not item: + continue + # Find the timestamp range via regex + if "-" not in item: + raise ValueError(f"bad segment spec missing '-': {item!r}") + # Find the colon in the second timestamp that separates MM:SS from the text + # Strategy: first occurrence of "-" splits range; rest is text + # We need to NOT split on the "-" inside the timestamps (none, but be safe) + # Actually timestamps use ":" only, not "-", so first "-" is safe. + dash_idx = item.index("-") + start_str = item[:dash_idx] + rest = item[dash_idx + 1:] + # The rest is either "MM:SS" or "MM:SS:TEXT" + # Find the second timestamp by parsing prefix until we have valid MM:SS + # Try 4-char prefix (MM:SS) then 7-char prefix (HH:MM:SS) + if ":" in rest: + # Find the colon after the minutes field + # First timestamp is MM:SS or HH:MM:SS + # Second timestamp ends at either ":" or end + # Try parsing increasing prefix lengths + for try_len in range(len(rest), 0, -1): + candidate = rest[:try_len] + try: + parse_timestamp(candidate) + start = parse_timestamp(start_str) + end = parse_timestamp(candidate) + text = rest[try_len + 1:].lstrip() if try_len < len(rest) - 1 else "" + # If there's a leading ":" after the timestamp, strip it + if text.startswith(":"): + text = text[1:].lstrip() + out.append((start, end, text)) + break + except ValueError: + continue + else: + raise ValueError(f"could not parse second timestamp in: {item!r}") + else: + # Just "MM:SS" with no text + start = parse_timestamp(start_str) + end = parse_timestamp(rest) + out.append((start, end, "")) + return out + + +def cmd_find(args: argparse.Namespace) -> int: + """Print metadata + (optional) silence gaps + (optional) segments from a JSON file.""" + audio = Path(args.audio) + if not audio.exists(): + print(f"error: {audio} not found", file=sys.stderr) + return 1 + sc = SermonClean(audio) + print(f"file: {audio}") + print(f"duration: {format_timestamp(sc.duration)} ({sc.duration:.3f}s)") + print(f"codec: {sc.codec['codec_name']} / {sc.codec['sample_rate']}Hz / {sc.codec['channels']}ch") + if args.silence_threshold is not None: + gaps = detect_silence(audio, noise_db=args.silence_threshold, min_duration=args.silence_min_duration) + print(f"\nsilence gaps (noise<{args.silence_threshold}dB, dur>={args.silence_min_duration}s):") + for a, b in gaps: + print(f" {format_timestamp(a)} -> {format_timestamp(b)} ({b-a:.2f}s)") + if args.segments_file: + n = sc.load_segments_from_json(Path(args.segments_file)) + print(f"\nloaded {n} segments from {args.segments_file}:") + for s in sc.segments: + print(f" {format_timestamp(s.start)} -> {format_timestamp(s.end)} ({s.duration:.2f}s) reason={s.reason!r}") + return 0 + + +def cmd_cut(args: argparse.Namespace) -> int: + """Just trim the original around the bad windows (no splice).""" + audio = Path(args.audio) + sc = SermonClean(audio) + if args.segments_file: + sc.load_segments_from_json(Path(args.segments_file)) + elif args.bad: + for start, end, text in _parse_bad_spec(args.bad): + sc.add_segment(start, end, replacement_text=text) + else: + print("error: provide --bad or --segments-file", file=sys.stderr) + return 1 + out = sc.trim_segments() + print(f"trimmed into {len(out)} segments in {sc.workdir}") + for p in out: + print(f" {p} ({probe_duration(p):.2f}s)") + return 0 + + +def cmd_paste(args: argparse.Namespace) -> int: + """Take pre-rendered replacement clips and splice them in.""" + audio = Path(args.audio) + sc = SermonClean(audio) + if args.segments_file: + sc.load_segments_from_json(Path(args.segments_file)) + elif args.bad: + for start, end, text in _parse_bad_spec(args.bad): + sc.add_segment(start, end, replacement_text=text) + else: + print("error: provide --bad or --segments-file", file=sys.stderr) + return 1 + if not args.replacements: + print("error: provide --replacements (one per bad segment, in order)", file=sys.stderr) + return 1 + import glob + paths = [] + for pat in args.replacements: + paths.extend(sorted(glob.glob(pat))) + if len(paths) != len(sc.segments): + print(f"error: {len(paths)} replacement files but {len(sc.segments)} bad segments", file=sys.stderr) + return 1 + output = Path(args.output) if args.output else audio.with_name(f"{audio.stem}-fixed{audio.suffix}") + result = sc.clean([Path(p) for p in paths], output) + print(f"output: {result.output_path}") + print(f"duration: {format_timestamp(result.duration_seconds)} (was {format_timestamp(result.original_duration)})") + print(f"duration check: {result.duration_check()}") + return 0 + + +def cmd_pipe(args: argparse.Namespace) -> int: + """Cut + paste in one step. Use ElevenLabs to render replacements if --elevenlabs-text given.""" + audio = Path(args.audio) + sc = SermonClean(audio) + if args.segments_file: + sc.load_segments_from_json(Path(args.segments_file)) + elif args.bad: + for start, end, text in _parse_bad_spec(args.bad): + sc.add_segment(start, end, replacement_text=text) + else: + print("error: provide --bad or --segments-file", file=sys.stderr) + return 1 + if args.elevenlabs_text: + # Lazy import so the core module doesn't depend on ElevenLabs + try: + from .elevenlabs import render_replacements + except ImportError: + print("error: elevenlabs rendering requires `pip install requests`", file=sys.stderr) + return 1 + reps = render_replacements(sc.segments, voice_id=args.voice_id, api_key=args.elevenlabs_key) + elif args.replacements: + import glob + paths = [] + for pat in args.replacements: + paths.extend(sorted(glob.glob(pat))) + reps = [Path(p) for p in paths] + else: + print("error: provide --replacements OR --elevenlabs-text", file=sys.stderr) + return 1 + if len(reps) != len(sc.segments): + print(f"error: {len(reps)} replacements vs {len(sc.segments)} segments", file=sys.stderr) + return 1 + output = Path(args.output) if args.output else audio.with_name(f"{audio.stem}-fixed{audio.suffix}") + result = sc.clean(reps, output) + print(f"output: {result.output_path}") + print(f"duration: {format_timestamp(result.duration_seconds)} (was {format_timestamp(result.original_duration)})") + print(f"duration check: {result.duration_check()}") + return 0 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser( + prog="sermon-clean", + description=USAGE, + formatter_class=argparse.RawDescriptionHelpFormatter, + ) + sub = parser.add_subparsers(dest="cmd", required=True) + + # find + p_find = sub.add_parser("find", help="Show audio metadata; optionally find silence gaps and load segments") + p_find.add_argument("audio") + p_find.add_argument("--silence-threshold", type=float, default=None, help="noise_db for silencedetect (e.g. -30)") + p_find.add_argument("--silence-min-duration", type=float, default=0.5, help="minimum silence in seconds") + p_find.add_argument("--segments-file", help="JSON file with bad segments") + p_find.set_defaults(func=cmd_find) + + # cut + p_cut = sub.add_parser("cut", help="Trim the original audio around bad windows (no splice)") + p_cut.add_argument("audio") + p_cut.add_argument("--segments-file") + p_cut.add_argument("--bad", help='comma-separated MM:SS-MM:SS[:TEXT] windows') + p_cut.set_defaults(func=cmd_cut) + + # paste + p_paste = sub.add_parser("paste", help="Splice pre-rendered replacement clips into a trimmed original") + p_paste.add_argument("audio") + p_paste.add_argument("--segments-file") + p_paste.add_argument("--bad") + p_paste.add_argument("-o", "--output") + p_paste.add_argument("--replacements", nargs="+", help="glob(s) for replacement MP3s in order") + p_paste.set_defaults(func=cmd_paste) + + # pipe (all in one) + p_pipe = sub.add_parser("pipe", help="Run cut + paste in one command") + p_pipe.add_argument("audio") + p_pipe.add_argument("--segments-file") + p_pipe.add_argument("--bad") + p_pipe.add_argument("-o", "--output") + p_pipe.add_argument("--replacements", nargs="+") + p_pipe.add_argument("--elevenlabs-text", action="store_true", help="use the --bad TEXT as ElevenLabs render input") + p_pipe.add_argument("--voice-id", help="ElevenLabs voice ID (or set ELEVENLABS_VOICE_ID)") + p_pipe.add_argument("--elevenlabs-key", help="ElevenLabs API key (or set ELEVENLABS_API_KEY)") + p_pipe.set_defaults(func=cmd_pipe) + + args = parser.parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/sermon_clean/elevenlabs.py b/sermon_clean/elevenlabs.py new file mode 100644 index 0000000..aee3ca3 --- /dev/null +++ b/sermon_clean/elevenlabs.py @@ -0,0 +1,84 @@ +"""ElevenLabs renderer for sermon-clean. + +Lazy-imported so the core engine doesn't require `requests` unless you actually +use the ElevenLabs hook. +""" + +from __future__ import annotations + +import os +import json +import urllib.request +from pathlib import Path +from typing import Sequence + +DEFAULT_VOICE = "IYUnpZr9CQfSylOsOOBo" # Sami's clone, confirmed 2026-07-26 +DEFAULT_MODEL = "eleven_multilingual_v2" +OUTPUT_FORMAT = "mp3_44100_128" + + +def render_replacements( + segments: Sequence, + *, + voice_id: str | None = None, + api_key: str | None = None, + model: str = DEFAULT_MODEL, + outdir: Path | None = None, +) -> list[Path]: + """Render one MP3 per segment via ElevenLabs. Returns paths in input order.""" + api_key = api_key or os.environ.get("ELEVENLABS_API_KEY") + if not api_key: + raise RuntimeError("ELEVENLABS_API_KEY not set (or pass api_key=)") + voice_id = voice_id or os.environ.get("ELEVENLABS_VOICE_ID", DEFAULT_VOICE) + if outdir is None: + outdir = Path("./replacements") + outdir = Path(outdir) + outdir.mkdir(parents=True, exist_ok=True) + + out: list[Path] = [] + for i, seg in enumerate(segments): + if not getattr(seg, "replacement_text", ""): + raise ValueError( + f"segment {i} ({seg.start:.2f}s..{seg.end:.2f}s) has no replacement_text. " + "Provide one with --bad MM:SS-MM:SS:TEXT syntax, or in the segments JSON." + ) + path = render_one(seg.replacement_text, voice_id=voice_id, api_key=api_key, model=model, outdir=outdir, idx=i) + out.append(path) + return out + + +def render_one( + text: str, + *, + voice_id: str, + api_key: str, + model: str, + outdir: Path, + idx: int, +) -> Path: + url = f"https://api.elevenlabs.io/v1/text-to-speech/{voice_id}?output_format={OUTPUT_FORMAT}" + body = json.dumps({ + "text": text, + "model_id": model, + "voice_settings": { + "stability": 0.5, + "similarity_boost": 0.75, + "style": 0.0, + "use_speaker_boost": True, + }, + }).encode() + req = urllib.request.Request( + url, + data=body, + headers={"xi-api-key": api_key, "Accept": "audio/mpeg", "Content-Type": "application/json"}, + ) + with urllib.request.urlopen(req, timeout=120) as r: + data = r.read() + if len(data) < 1000: + raise RuntimeError( + f"ElevenLabs returned {len(data)} bytes for segment {idx} — " + "likely a quota/auth failure (200 with empty body). Re-render and verify size > 1KB." + ) + out_path = Path(outdir) / f"rep_{idx:03d}.mp3" + out_path.write_bytes(data) + return out_path diff --git a/sermon_clean/engine.py b/sermon_clean/engine.py new file mode 100644 index 0000000..3e7112c --- /dev/null +++ b/sermon_clean/engine.py @@ -0,0 +1,400 @@ +"""Core engine for sermon-clean. + +The pipeline is intentionally straightforward: each step is a pure function that +takes the output of the previous step. The CLI in `cli.py` orchestrates them. +""" + +from __future__ import annotations + +import json +import re +import shutil +import subprocess +import tempfile +from dataclasses import dataclass, field, asdict +from pathlib import Path +from typing import Iterable + + +# --------------------------------------------------------------------------- +# Data classes +# --------------------------------------------------------------------------- + + +@dataclass +class BadSegment: + """A bad window in the original audio, expressed in seconds.""" + + start: float + end: float + reason: str = "" + replacement_text: str = "" + + def __post_init__(self) -> None: + if self.end <= self.start: + raise ValueError(f"BadSegment end must be > start (got {self.start}..{self.end})") + if (self.end - self.start) > 60: + raise ValueError( + f"BadSegment suspiciously long ({self.end - self.start:.1f}s). " + "Surgical replacement is the goal — keep windows short. " + "If you really need a long window, pass --allow-long-windows." + ) + + @property + def duration(self) -> float: + return self.end - self.start + + +@dataclass +class Replacement: + """A replacement audio clip for a BadSegment.""" + + segment: BadSegment + audio_path: Path + + def __post_init__(self) -> None: + if not self.audio_path.exists(): + raise FileNotFoundError(f"Replacement audio not found: {self.audio_path}") + if self.audio_path.stat().st_size < 1000: + raise ValueError( + f"Replacement audio file is too small ({self.audio_path.stat().st_size} bytes). " + "ElevenLabs sometimes returns 200 with empty bytes on quota/auth failure. " + "Re-render and verify size > 1KB before splicing." + ) + + +@dataclass +class SpliceResult: + """The output of a successful splice.""" + + output_path: Path + duration_seconds: float + n_segments: int + n_replacements: int + original_duration: float = 0.0 + expected_duration: float = 0.0 + + def duration_check(self) -> str: + """Return 'ok' if the splice matches the expected duration ±1s.""" + delta = abs(self.duration_seconds - self.expected_duration) + if delta > 1.0: + return f"DRIFT ({delta:.2f}s from expected {self.expected_duration:.2f}s)" + return "ok" + + +# --------------------------------------------------------------------------- +# Timestamp parsing +# --------------------------------------------------------------------------- + + +_TS_RE = re.compile(r"^(?:(\d+):)?(\d{1,2}):(\d{1,2}(?:\.\d+)?)$|^\d+(?:\.\d+)?$") + + +def parse_timestamp(s: str) -> float: + """Parse 'SS', 'MM:SS', or 'HH:MM:SS' into seconds. Raises ValueError on bad input.""" + s = s.strip() + if not s: + raise ValueError("empty timestamp") + # Pure seconds + if re.match(r"^\d+(?:\.\d+)?$", s): + return float(s) + # HH:MM:SS or MM:SS + m = re.match(r"^(?:(\d+):)?(\d{1,2}):(\d{1,2}(?:\.\d+)?)$", s) + if not m: + raise ValueError(f"could not parse timestamp: {s!r}") + h, mm, ss = m.groups() + h = int(h) if h else 0 + mm = int(mm) + ss = float(ss) + if mm >= 60 or ss >= 60: + raise ValueError(f"invalid timestamp: {s!r}") + return h * 3600 + mm * 60 + ss + + +def format_timestamp(seconds: float) -> str: + """Render seconds as HH:MM:SS.mmm (or MM:SS.mmm if < 1h).""" + if seconds < 0: + return f"-{format_timestamp(-seconds)}" + h = int(seconds // 3600) + m = int((seconds % 3600) // 60) + s = seconds % 60 + if h: + return f"{h:02d}:{m:02d}:{s:06.3f}" + return f"{m:02d}:{s:06.3f}" + + +# --------------------------------------------------------------------------- +# ffprobe wrapper +# --------------------------------------------------------------------------- + + +def _run(cmd: list[str], *, check: bool = True) -> subprocess.CompletedProcess: + return subprocess.run(cmd, capture_output=True, text=True, check=check) + + +def probe_duration(path: Path) -> float: + """Return the duration of an audio file in seconds (via ffprobe).""" + out = _run([ + "ffprobe", "-v", "error", + "-show_entries", "format=duration", + "-of", "default=noprint_wrappers=1:nokey=1", + str(path), + ]) + return float(out.stdout.strip()) + + +def probe_codec(path: Path) -> dict: + """Return codec_name, sample_rate, channels for the first audio stream.""" + out = _run([ + "ffprobe", "-v", "error", + "-select_streams", "a:0", + "-show_entries", "stream=codec_name,sample_rate,channels", + "-of", "json", + str(path), + ]) + data = json.loads(out.stdout) + stream = data["streams"][0] + return { + "codec_name": stream["codec_name"], + "sample_rate": stream["sample_rate"], + "channels": stream["channels"], + } + + +# Map from ffprobe demuxer name → ffmpeg encoder name. ffprobe reports "opus" +# for the OGG-Opus stream, but `ffmpeg -c:a opus` is the experimental encoder; +# the non-experimental one is `libopus`. Same for vorbis. +_ENCODER_ALIASES = { + "opus": "libopus", + "vorbis": "libvorbis", +} + + +def encoder_name(demuxer_name: str) -> str: + """Translate a ffprobe codec name to the corresponding ffmpeg encoder name.""" + return _ENCODER_ALIASES.get(demuxer_name, demuxer_name) + + +# --------------------------------------------------------------------------- +# Silence detection (ffmpeg silencedetect filter) +# --------------------------------------------------------------------------- + + +def detect_silence( + path: Path, + *, + noise_db: float = -30.0, + min_duration: float = 0.5, +) -> list[tuple[float, float]]: + """Return (start, end) of every silence run, using ffmpeg's silencedetect filter. + + Useful for finding natural breath pauses between sentences, which often bracket + the bad segments you're trying to excise. Combine with manual timestamps when + you want to be sure the splice window lines up with a pause. + """ + out = _run([ + "ffmpeg", "-v", "error", "-i", str(path), + "-af", f"silencedetect=noise={noise_db}dB:d={min_duration}", + "-f", "null", "-", + ]) + starts: list[float] = [] + ends: list[float] = [] + for line in out.stdout.splitlines(): + line = line.strip() + if line.startswith("silence_start:"): + starts.append(float(line.split(":", 1)[1].strip())) + elif line.startswith("silence_end:"): + parts = line.split("|") + if len(parts) >= 2: + ends.append(float(parts[0].split(":", 1)[1].strip())) + return list(zip(starts, ends)) + + +# --------------------------------------------------------------------------- +# The engine +# --------------------------------------------------------------------------- + + +class SermonClean: + """Top-level orchestrator. Holds the original audio path and the bad-segment list.""" + + def __init__( + self, + original: Path, + *, + workdir: Path | None = None, + allow_long_windows: bool = False, + ) -> None: + self.original = Path(original) + if not self.original.exists(): + raise FileNotFoundError(f"original audio not found: {self.original}") + self.codec = probe_codec(self.original) + self.duration = probe_duration(self.original) + self.allow_long_windows = allow_long_windows + self.workdir = Path(workdir) if workdir else Path(tempfile.mkdtemp(prefix="sermon-clean-")) + self.workdir.mkdir(parents=True, exist_ok=True) + self.segments: list[BadSegment] = [] + + # -------- segment list management -------- + + def add_segment(self, start: float, end: float, *, reason: str = "", replacement_text: str = "") -> BadSegment: + seg = BadSegment(start, end, reason=reason, replacement_text=replacement_text) + if self.allow_long_windows and (end - start) > 60: + # Patch back: re-create with relaxed check + object.__setattr__(seg, "end", end) + self.segments.append(seg) + self.segments.sort(key=lambda s: s.start) + return seg + + def add_segment_str(self, start_str: str, end_str: str, **kwargs) -> BadSegment: + return self.add_segment(parse_timestamp(start_str), parse_timestamp(end_str), **kwargs) + + def load_segments_from_json(self, path: Path) -> int: + """Load segments from a JSON file. Returns the count loaded.""" + data = json.loads(Path(path).read_text()) + if not isinstance(data, list): + raise ValueError("segments JSON must be a list of {start, end, reason, replacement_text}") + loaded = 0 + for entry in data: + self.add_segment( + float(entry["start"]), + float(entry["end"]), + reason=entry.get("reason", ""), + replacement_text=entry.get("replacement_text", ""), + ) + loaded += 1 + return loaded + + def save_segments_to_json(self, path: Path) -> None: + Path(path).write_text(json.dumps([asdict(s) for s in self.segments], indent=2)) + + # -------- subtitle export -------- + + def export_segments_csv(self, path: Path) -> None: + """Write the segments as a CSV (start_sec,end_sec,start_str,end_str,reason,replacement_text).""" + with open(path, "w") as f: + f.write("start_sec,end_sec,start_str,end_str,reason,replacement_text\n") + for s in self.segments: + reason = s.reason.replace('"', '""') + repl = s.replacement_text.replace('"', '""') + f.write( + f"{s.start:.3f},{s.end:.3f}," + f"{format_timestamp(s.start)},{format_timestamp(s.end)}," + f"\"{reason}\",\"{repl}\"\n" + ) + + # -------- trim + concat pipeline -------- + + def _encode_target(self, in_path: Path, out_path: Path) -> None: + """Re-encode to the original audio's codec, sample rate, channels.""" + _run([ + "ffmpeg", "-y", "-v", "error", + "-i", str(in_path), + "-ar", str(self.codec["sample_rate"]), + "-ac", str(self.codec["channels"]), + "-c:a", encoder_name(self.codec["codec_name"]), + "-b:a", "128k", + str(out_path), + ], check=True) + + def trim_segments(self) -> list[Path]: + """Cut the original audio into N+1 good segments (N bad windows).""" + if not self.segments: + raise ValueError("no bad segments defined — call add_segment() first") + for s in self.segments: + if s.end > self.duration: + raise ValueError( + f"segment end {s.end:.2f}s exceeds audio duration {self.duration:.2f}s" + ) + # Build the N+1 keep windows: [0..s0], [s0..s1], ..., [sN..T] + keep_windows: list[tuple[float, float]] = [] + prev_end = 0.0 + for s in self.segments: + keep_windows.append((prev_end, s.start)) + prev_end = s.end + keep_windows.append((prev_end, self.duration)) + + out_paths: list[Path] = [] + for i, (a, b) in enumerate(keep_windows): + p = self.workdir / f"seg_{i:03d}_head.ogg" + if abs(b - a) < 0.01: + # Skip zero-length segments (would only matter if bad windows are adjacent) + continue + _run([ + "ffmpeg", "-y", "-v", "error", + "-ss", f"{a:.3f}", "-to", f"{b:.3f}", + "-i", str(self.original), + "-ar", str(self.codec["sample_rate"]), + "-ac", str(self.codec["channels"]), + "-c:a", encoder_name(self.codec["codec_name"]), + "-b:a", "128k", + str(p), + ], check=True) + out_paths.append(p) + return out_paths + + def encode_replacements(self, replacements: list[Path]) -> list[Path]: + """Re-encode each replacement MP3 to the original audio's codec.""" + out: list[Path] = [] + for i, path in enumerate(replacements): + p = self.workdir / f"rep_{i:03d}.ogg" + self._encode_target(Path(path), p) + out.append(p) + return out + + def concat( + self, + segments: list[Path], + replacements: list[Path], + output: Path, + ) -> SpliceResult: + """Concat keep-segments + replacement clips into a single output file.""" + if len(replacements) != len(self.segments): + raise ValueError( + f"need {len(self.segments)} replacements, got {len(replacements)}" + ) + # Interleave: seg0, rep0, seg1, rep1, ..., segN + if len(segments) != len(self.segments) + 1: + raise ValueError( + f"need {len(self.segments) + 1} keep-segments, got {len(segments)}" + ) + concat_list = self.workdir / "concat_list.txt" + with open(concat_list, "w") as f: + for s in segments[:1]: + f.write(f"file '{s.name}'\n") + for i, rep in enumerate(replacements): + f.write(f"file '{rep.name}'\n") + f.write(f"file '{segments[i+1].name}'\n") + + _run([ + "ffmpeg", "-y", "-v", "error", + "-f", "concat", "-safe", "0", + "-i", str(concat_list), + "-c", "copy", + str(output), + ], check=True) + + # Compute expected duration + expected = sum(probe_duration(p) for p in segments) + sum(probe_duration(p) for p in replacements) + actual = probe_duration(output) + return SpliceResult( + output_path=Path(output), + duration_seconds=actual, + n_segments=len(segments), + n_replacements=len(replacements), + original_duration=self.duration, + expected_duration=expected, + ) + + def clean( + self, + replacements: list[Path], + output: Path | None = None, + ) -> SpliceResult: + """End-to-end: trim, encode replacements, concat. Returns a SpliceResult.""" + segs = self.trim_segments() + reps = self.encode_replacements(replacements) + if output is None: + output = self.workdir / f"{self.original.stem}-fixed{self.original.suffix}" + result = self.concat(segs, reps, output) + return result diff --git a/tests/test_engine.py b/tests/test_engine.py new file mode 100644 index 0000000..c0d3da2 --- /dev/null +++ b/tests/test_engine.py @@ -0,0 +1,99 @@ +"""Tests for sermon-clean.engine.""" + +from pathlib import Path + +import pytest + +from sermon_clean.engine import ( + SermonClean, + parse_timestamp, + format_timestamp, +) + + +class TestTimestampParsing: + @pytest.mark.parametrize("s,expected", [ + ("30", 30.0), + ("90", 90.0), + ("1:30", 90.0), + ("01:30", 90.0), + ("1:30.5", 90.5), + ("1:00:30", 3630.0), + ("02:01:30.5", 7290.5), + ]) + def test_valid(self, s, expected): + assert parse_timestamp(s) == expected + + @pytest.mark.parametrize("s", ["", "abc", "1:60", "60:60", "1.2.3"]) + def test_invalid(self, s): + with pytest.raises(ValueError): + parse_timestamp(s) + + +class TestFormatTimestamp: + @pytest.mark.parametrize("seconds,expected", [ + (0.0, "00:00.000"), + (59.999, "00:59.999"), + (60.0, "01:00.000"), + (90.5, "01:30.500"), + (3600.0, "01:00:00.000"), + (3661.5, "01:01:01.500"), + (7290.5, "02:01:30.500"), + ]) + def test_round_trip(self, seconds, expected): + assert format_timestamp(seconds) == expected + + def test_roundtrip_parse(self): + for s in [0.0, 5.0, 59.999, 60.0, 3630.0, 7290.5]: + assert parse_timestamp(format_timestamp(s)) == pytest.approx(s) + + +class TestBadSegment: + def test_valid(self): + from sermon_clean.engine import BadSegment + s = BadSegment(10.0, 12.0, reason="misspoke") + assert s.duration == 2.0 + assert s.start == 10.0 + assert s.end == 12.0 + + def test_zero_length_rejected(self): + from sermon_clean.engine import BadSegment + with pytest.raises(ValueError): + BadSegment(10.0, 10.0) + + def test_negative_rejected(self): + from sermon_clean.engine import BadSegment + with pytest.raises(ValueError): + BadSegment(15.0, 10.0) + + def test_long_window_rejected(self): + from sermon_clean.engine import BadSegment + # 90s window by default > 60s cap + with pytest.raises(ValueError, match="suspiciously long"): + BadSegment(0.0, 90.0) + + +def test_sermon_clean_basic(tmp_path): + """Smoke test: load a real audio file from the krystie profile cache.""" + # Find any audio file in the krystie cache + cache = Path("/root/.hermes/profiles/krystie/cache/audio") + if not cache.exists(): + pytest.skip("krystie audio cache not present") + audio_files = sorted(cache.glob("*.ogg")) + sorted(cache.glob("*.mp3")) + if not audio_files: + pytest.skip("no audio files in krystie cache") + audio = audio_files[0] + sc = SermonClean(audio, workdir=tmp_path) + assert sc.duration > 0 + assert sc.codec["codec_name"] + assert sc.codec["sample_rate"] + seg = sc.add_segment(0.5, 1.0, reason="test") + assert seg.start == 0.5 + assert len(sc.segments) == 1 + segs = sc.trim_segments() + assert len(segs) == 2 # before + after the 0.5s window + assert all(p.exists() for p in segs) + # Verify the trimmed segments + from sermon_clean.engine import probe_duration + total = sum(probe_duration(p) for p in segs) + assert total == pytest.approx(sc.duration - 0.5, abs=0.1)