sami7777/sermon-clean
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Initial commit: sermon-clean v0.1.0

Browse Source 
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).
This commit is contained in:
Sami Ahmed
2026-07-27 02:38:37 -07:00
parent 0cd16e6dc2
commit cea6005bf4
9 changed files with 993 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+30
View File
@@ -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/
LICENSE
+21
View File
@@ -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.
examples/segs.json
+14
View File
@@ -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"
}
]
pyproject.toml
+55
View File
@@ -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"]
sermon_clean/__init__.py
+34
View File
@@ -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",
]
sermon_clean/cli.py
+256
View File
@@ -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())
sermon_clean/elevenlabs.py
+84
View File
@@ -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
sermon_clean/engine.py
+400
View File
@@ -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
tests/test_engine.py
+99
View File
@@ -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)