sermon-clean/sermon_clean/cli.py

"""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_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_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)

    # 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)

    # 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())