"""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_silence_index(args: argparse.Namespace) -> int: """Print a tabular index of silence runs with timestamps.""" from .waveform import render_silence_index audio = Path(args.audio) if not audio.exists(): print(f"error: {audio} not found", file=sys.stderr) return 1 print(render_silence_index( audio, threshold_db=args.silence_threshold, min_duration=args.silence_min_duration, )) return 0 def cmd_multiband(args: argparse.Namespace) -> int: """Print a multi-band ASCII waveform. Each row is N seconds labeled.""" from .waveform import render_multiband_waveform audio = Path(args.audio) if not audio.exists(): print(f"error: {audio} not found", file=sys.stderr) return 1 print(render_multiband_waveform( audio, width=args.width, band_seconds=args.band_seconds, )) return 0 def cmd_normalize(args: argparse.Namespace) -> int: """Apply EBU R128 loudness normalization.""" from .processing import normalize_loudness audio = Path(args.audio) if not audio.exists(): print(f"error: {audio} not found", file=sys.stderr) return 1 output = Path(args.output) if args.output else audio.with_name(f"{audio.stem}-normalized{audio.suffix}") result = normalize_loudness( audio, output, target_lufs=args.target_lufs, true_peak_db=args.true_peak, loudness_range_lu=args.loudness_range, ) print(f"normalized: {result['output_path']}") print(f"target: {result['target_lufs']} LUFS, true peak {result['true_peak_db']}dB") print(f"measured input loudness: {result['measured_input_i']} LUFS") print(f"applied gain offset: {result['applied_offset']} dB") return 0 def cmd_silence_stats(args: argparse.Namespace) -> int: """Quantitative summary of silence distribution (count, mean, longest).""" from .processing import silence_stats import json audio = Path(args.audio) if not audio.exists(): print(f"error: {audio} not found", file=sys.stderr) return 1 stats = silence_stats( audio, threshold_db=args.silence_threshold, min_duration_seconds=args.silence_min_duration, ) if args.json: print(json.dumps(stats.to_dict(), indent=2)) return 0 # Human-readable summary print(f"file: {stats.audio_path}") print(f"duration: {stats.duration_seconds:.2f}s") print(f"threshold: {stats.threshold_db}dB, min duration: {stats.min_duration_seconds}s") print() print(f"silence runs: {stats.n_silence_runs}") print(f"total silence: {stats.total_silence_seconds:.2f}s ({stats.silence_fraction*100:.1f}% of audio)") print(f"mean silence: {stats.mean_silence_seconds:.2f}s") print(f"median silence: {stats.median_silence_seconds:.2f}s") print(f"longest silence: {stats.longest_silence_seconds:.2f}s " f"@ {stats.longest_silence_range[0]:.1f}-{stats.longest_silence_range[1]:.1f}s") print(f"silence density: {stats.silence_per_minute:.2f} runs/min") return 0 def cmd_threshold_tune(args: argparse.Namespace) -> int: """Auto-pick the silence threshold for this audio based on expected pause density.""" from .processing import threshold_tune import json audio = Path(args.audio) if not audio.exists(): print(f"error: {audio} not found", file=sys.stderr) return 1 candidates = [float(x) for x in args.candidates.split(",")] result = threshold_tune( audio, candidates=candidates, target_silence_per_minute=args.target_spm, target_min_duration=args.min_duration, ) if args.json: print(json.dumps(result, indent=2)) return 0 print(f"target: {result['target_silence_per_minute']:.1f} silence runs/min") print(f"picked threshold: {result['picked_threshold_db']}dB " f"(gives {result['picked_silence_per_minute']:.2f} runs/min)") print() print(f"{'threshold_dB':>12} {'runs':>6} {'runs/min':>10} {'distance':>10}") for row in result["candidates"]: marker = " ← picked" if row["threshold_db"] == result["picked_threshold_db"] else "" print(f"{row['threshold_db']:>12.0f} {row['n_runs']:>6} {row['silence_per_minute']:>10.2f} " f"{row['distance_from_target']:>10.2f}{marker}") return 0 def cmd_denoise(args: argparse.Namespace) -> int: """Apply light FFT-based noise reduction (good for tape hiss, mic preamp noise).""" from .denoise import denoise audio = Path(args.audio) if not audio.exists(): print(f"error: {audio} not found", file=sys.stderr) return 1 output = Path(args.output) if args.output else audio.with_name(f"{audio.stem}-denoised{audio.suffix}") result = denoise( audio, output, noise_reduction_db=args.noise_reduction, noise_floor_db=args.noise_floor, ) print(f"denoised: {result.output_path}") print(f"noise reduction: {result.noise_reduction_db}dB (assumed floor: {result.noise_floor_db}dB)") return 0 def cmd_batch(args: argparse.Namespace) -> int: """Run a sermon-clean subcommand across many files.""" from .batch import run_batch output_dir = Path(args.output_dir) result = run_batch( args.subcommand, [Path(p) for p in args.sources], output_dir, output_suffix=args.suffix or "-fixed", output_extension=args.extension, # Don't forward --suffix or --extension to the subcommand — they're # batch-only knobs that control output naming, not subcommand args. extra_args=[], ) print(result.summary()) return 0 if result.n_failed == 0 else 1 def cmd_slices(args: argparse.Namespace) -> int: """Extract overlapping audio slices for manual review. Useful when you don't have exact timestamps and want to scrub through the audio in a media player, marking bad segments as you go. """ from .slices import extract_slices audio = Path(args.audio) out = Path(args.output_dir) paths = extract_slices( audio, out, slice_seconds=args.slice_seconds, overlap_seconds=args.overlap_seconds, ) print(f"extracted {len(paths)} slices into {out}") for p in paths: print(f" {p.name}") print() print(f"play them in any media player, then mark bad-segment start/end times") print(f"in your audio player's display, then build a segments JSON and run:") print(f" sermon-clean auto {audio} --bad-words ...") return 0 def cmd_scan(args: argparse.Namespace) -> int: """Print an ASCII waveform + silence marks for the audio. No whisper, fast.""" audio = Path(args.audio) if not audio.exists(): print(f"error: {audio} not found", file=sys.stderr) return 1 from .waveform import render_ascii_waveform, render_silence_marks print(f"=== waveform: {audio} ===") print(render_ascii_waveform(audio, width=args.width)) print() print(f"=== silence runs (noise<{args.silence_threshold}dB, dur>={args.silence_min_duration}s) ===") print(render_silence_marks( audio, threshold_db=args.silence_threshold, min_duration_seconds=args.silence_min_duration, width=args.width, )) return 0 def cmd_auto(args: argparse.Namespace) -> int: """Transcribe audio with faster-whisper + suggest bad-word segments.""" audio = Path(args.audio) if not audio.exists(): print(f"error: {audio} not found", file=sys.stderr) return 1 try: from .transcribe import ( find_bad_words, suggest_replacements, export_to_segments_json, ) except ImportError as e: print(f"error: --auto requires the [auto] extra: {e}", file=sys.stderr) print("install with: pip install 'sermon-clean[auto]'", file=sys.stderr) return 1 bad_words = [w.strip() for w in args.bad_words.split(",") if w.strip()] print(f"transcribing {audio} with model={args.model}, scanning for: {bad_words}...") hits = find_bad_words( audio, bad_words, model_name=args.model, confidence_threshold=args.confidence_threshold, ) print(f"found {len(hits)} candidate bad-word hits") if not hits: print("no bad words detected — nothing to write") return 0 # Get transcript context for replacement suggestions from .transcribe import transcribe words = transcribe(audio, model_name=args.model) hits = suggest_replacements(hits, transcript_context=words, max_window=args.context_window) segs = export_to_segments_json(hits, include_suggested_only=args.include_suggested_only) out_path = Path(args.output) out_path.write_text(json.dumps(segs, indent=2)) print(f"wrote {len(segs)} segments to {out_path}") print("\nreview and edit the JSON before running 'paste' or 'pipe':") print(f" sermon-clean paste {audio} --segments-file {out_path} --replacements 'replacements/*.mp3'") 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) # silence-index (tabular list of silence runs with timestamps) p_si = sub.add_parser("silence-index", aliases=["si"], help="Tabular index of silence runs with timestamps") p_si.add_argument("audio") p_si.add_argument("--silence-threshold", type=float, default=-35.0, help="noise_db threshold") p_si.add_argument("--silence-min-duration", type=float, default=0.3, help="minimum silence in seconds") p_si.set_defaults(func=cmd_silence_index) # multiband (multi-row ASCII waveform with band labels) p_mb = sub.add_parser("multiband", aliases=["mb"], help="Multi-band ASCII waveform with band-start labels") p_mb.add_argument("audio") p_mb.add_argument("--width", type=int, default=60, help="waveform width in columns per band") p_mb.add_argument("--band-seconds", type=float, default=60.0, help="seconds of audio per band (one row)") p_mb.set_defaults(func=cmd_multiband) # slices (extract overlapping audio chunks for manual review) p_slices = sub.add_parser("slices", help="Extract overlapping audio slices for manual review") p_slices.add_argument("audio") p_slices.add_argument("--output-dir", default="./slices", help="directory to write slices") p_slices.add_argument("--slice-seconds", type=float, default=5.0, help="length of each slice") p_slices.add_argument("--overlap-seconds", type=float, default=1.0, help="overlap between adjacent slices") p_slices.set_defaults(func=cmd_slices) # normalize (EBU R128 loudness normalization) p_norm = sub.add_parser("normalize", aliases=["norm"], help="Apply EBU R128 loudness normalization") p_norm.add_argument("audio") p_norm.add_argument("-o", "--output", help="output file (default: -normalized)") p_norm.add_argument("--target-lufs", type=float, default=-16.0, help="target integrated loudness (LUFS, default -16 for podcast/YouTube)") p_norm.add_argument("--true-peak", type=float, default=-1.5, help="true peak ceiling in dB (default -1.5)") p_norm.add_argument("--loudness-range", type=float, default=11.0, help="loudness range target in LU (default 11)") p_norm.set_defaults(func=cmd_normalize) # silence-stats (quantitative summary of silence distribution) p_ss = sub.add_parser("silence-stats", aliases=["ss"], help="Quantitative summary of silence distribution") p_ss.add_argument("audio") p_ss.add_argument("--silence-threshold", type=float, default=-35.0, help="noise_db threshold") p_ss.add_argument("--silence-min-duration", type=float, default=0.3, help="minimum silence duration in seconds") p_ss.add_argument("--json", action="store_true", help="output JSON instead of human-readable") p_ss.set_defaults(func=cmd_silence_stats) # threshold-tune (auto-pick silence threshold for this audio) p_tt = sub.add_parser("threshold-tune", aliases=["tt"], help="Auto-pick the silence threshold based on expected pause density") p_tt.add_argument("audio") p_tt.add_argument("--candidates", default="-25,-30,-35,-40,-45,-50", help="comma-separated threshold candidates in dB (default: -25,-30,-35,-40,-45,-50)") p_tt.add_argument("--target-spm", type=float, default=4.0, help="target silence runs per minute (default 4.0 — typical sermon pause rate)") p_tt.add_argument("--min-duration", type=float, default=0.3, help="minimum silence duration in seconds") p_tt.add_argument("--json", action="store_true", help="output JSON instead of human-readable") p_tt.set_defaults(func=cmd_threshold_tune) # denoise (light FFT-based noise reduction) p_dn = sub.add_parser("denoise", help="Apply light FFT-based noise reduction (afftdn)") p_dn.add_argument("audio") p_dn.add_argument("-o", "--output", help="output file (default: -denoised)") p_dn.add_argument("--noise-reduction", type=float, default=12.0, help="noise attenuation in dB (default 12; higher = more aggressive)") p_dn.add_argument("--noise-floor", type=float, default=-50.0, help="expected noise floor in dB (default -50)") p_dn.set_defaults(func=cmd_denoise) # batch (run any subcommand across many files) p_batch = sub.add_parser("batch", help="Run a sermon-clean subcommand across many files") p_batch.add_argument("subcommand", help="which sermon-clean subcommand to run (normalize, scan, etc.)") p_batch.add_argument("sources", nargs="+", help="input files or glob patterns (e.g. 'sermons/*.ogg')") p_batch.add_argument("--output-dir", required=True, help="directory to write outputs") p_batch.add_argument("--suffix", default="-fixed", help="filename suffix for outputs (default '-fixed')") p_batch.add_argument("--extension", help="extension override (e.g. '.wav'); defaults to input extension") p_batch.set_defaults(func=cmd_batch) # scan (ASCII waveform + silence marks, no whisper) p_scan = sub.add_parser("scan", help="Print an ASCII waveform + silence marks for the audio") p_scan.add_argument("audio") p_scan.add_argument("--width", type=int, default=80, help="waveform width in columns") p_scan.add_argument("--silence-threshold", type=float, default=-40.0, help="noise_db for silence detection") p_scan.add_argument("--silence-min-duration", type=float, default=0.5, help="minimum silence duration in seconds") p_scan.set_defaults(func=cmd_scan) # auto (find bad words via faster-whisper) p_auto = sub.add_parser("auto", help="Transcribe audio + flag bad words, write a candidate segments JSON") p_auto.add_argument("audio") p_auto.add_argument("--bad-words", required=True, help="comma-separated list of bad words to find (e.g. 'fuck,shit,damn')") p_auto.add_argument("--model", default="tiny.en", help="whisper model name (tiny.en, base.en, small.en, ...)") p_auto.add_argument("--output", "-o", required=True, help="output JSON file for the suggested segments") p_auto.add_argument("--context-window", type=float, default=3.0, help="seconds before/after to look for context words") p_auto.add_argument("--confidence-threshold", type=float, default=0.30, help="minimum word probability (whisper)") p_auto.add_argument("--include-suggested-only", action="store_true", default=True, help="skip matches with no suggested replacement") p_auto.set_defaults(func=cmd_auto) # 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())