denoiser.py

"""
Denoiser — Cleanvoice SDK v2 (cleanvoice-sdk)

FIXES APPLIED:
  1. Always convert input to 16kHz mono WAV before uploading to Cleanvoice.
     Root cause of HTTP 400: Cleanvoice rejects .aac, .m4a, .ogg, .opus, .flac
     at the signed upload URL step — only .mp3 and .wav are reliably accepted.
  2. SDK install check at startup — fails loudly with actionable message.
  3. API key read from CLEANVOICE_API_KEY env var only — no hardcoded key.
  4. Full traceback always printed on failure — never swallowed silently.
  5. Correct SDK v2 parameter names (fillers=, stutters=, long_silences=,
     studio_sound="nightly") — v1 names were silently ignored.
  6. output_path= verified; falls back to result.audio.download() if needed.

Install: pip install cleanvoice-sdk
Requires: CLEANVOICE_API_KEY environment variable.
"""

import os
import re
import subprocess
import tempfile
import traceback
import logging

logger = logging.getLogger(__name__)

# ── API key — env var only, never hardcoded ───────────────────────────────────
CLEANVOICE_API_KEY = os.environ.get("CLEANVOICE_API_KEY", "").strip()

# Filler words for transcript text-side cleanup
FILLER_PATTERN = r'\b(um|uh|hmm|basically|like|ante|ane|matlab|yani|acha|right|okay)\b'

# Formats Cleanvoice reliably accepts for upload (tested)
# Everything else must be converted to WAV first
CLEANVOICE_SAFE_FORMATS = {".mp3", ".wav"}

# ── SDK import check at module load time ──────────────────────────────────────
_SDK_AVAILABLE = False
_CleanvoiceClass = None
try:
    from cleanvoice import Cleanvoice as _CleanvoiceClass
    _SDK_AVAILABLE = True
except ImportError:
    pass


class Denoiser:
    def __init__(self):
        # ── Check 1: SDK installed? ───────────────────────────────────
        if not _SDK_AVAILABLE:
            print(
                "[Denoiser] ❌ cleanvoice-sdk is NOT installed.\n"
                "           Fix: pip install cleanvoice-sdk\n"
                "           Denoising will be skipped until this is resolved."
            )
        else:
            print("[Denoiser] ✅ cleanvoice-sdk found")

        # ── Check 2: API key set? ─────────────────────────────────────
        if not CLEANVOICE_API_KEY:
            print(
                "[Denoiser] ❌ CLEANVOICE_API_KEY is not set.\n"
                "           Fix: export CLEANVOICE_API_KEY=your_key_here\n"
                "           Or add it as a HuggingFace Space Secret.\n"
                "           Denoising will be skipped until this is resolved."
            )
        else:
            masked = CLEANVOICE_API_KEY[:4] + "..." + CLEANVOICE_API_KEY[-4:]
            print(f"[Denoiser] ✅ API key found ({masked})")

        if _SDK_AVAILABLE and CLEANVOICE_API_KEY:
            print("[Denoiser] ✅ Ready — Cleanvoice SDK v2 will process audio")

    # ══════════════════════════════════════════════════════════════════
    # PUBLIC
    # ══════════════════════════════════════════════════════════════════
    def process(self, audio_path: str, out_dir: str,
                remove_fillers: bool = True,
                remove_stutters: bool = True,
                remove_silences: bool = True,
                **kwargs) -> dict:
        """
        Process audio using Cleanvoice SDK v2.
        Returns {'audio_path': str, 'stats': dict}.
        On failure, returns original audio_path so the rest of the pipeline
        (transcription, translation) can still run.
        """
        stats = {"noise_method": "none"}

        # ── Pre-flight checks ─────────────────────────────────────────
        if not _SDK_AVAILABLE:
            msg = "Skipped — cleanvoice-sdk not installed (run: pip install cleanvoice-sdk)"
            print(f"[Denoiser] ⚠️  {msg}")
            stats["noise_method"] = msg
            return {"audio_path": audio_path, "stats": stats}

        if not CLEANVOICE_API_KEY:
            msg = "Skipped — CLEANVOICE_API_KEY not set"
            print(f"[Denoiser] ⚠️  {msg}")
            stats["noise_method"] = msg
            return {"audio_path": audio_path, "stats": stats}

        # ── Step 1: Convert to WAV if needed ─────────────────────────
        # Cleanvoice returns HTTP 400 for .aac, .m4a, .ogg, .opus, .flac
        # at the signed upload step. Always convert anything that isn't
        # .mp3 or .wav to a clean 16kHz mono WAV before uploading.
        upload_path, tmp_wav = self._ensure_wav(audio_path, out_dir)

        # ── Step 2: Cleanvoice SDK v2 call ────────────────────────────
        try:
            client = _CleanvoiceClass(api_key=CLEANVOICE_API_KEY)

            ext       = os.path.splitext(upload_path)[1] or ".wav"
            clean_out = os.path.join(out_dir, f"cleanvoice_output{ext}")

            file_kb = os.path.getsize(upload_path) / 1024
            print(f"[Denoiser] Sending to Cleanvoice v2: {upload_path} ({file_kb:.1f} KB)")

            # SDK v2 correct parameter names
            # WRONG (v1, silently broken):
            #   remove_fillers=, remove_stutters=, remove_silences=, studio_sound=True
            # CORRECT (v2):
            #   fillers=,        stutters=,        long_silences=,   studio_sound="nightly"
            result = client.process(
                upload_path,
                remove_noise=True,
                studio_sound="nightly",         # must be "nightly", NOT True
                normalize=True,
                fillers=remove_fillers,         # was: remove_fillers= (v1, ignored)
                stutters=remove_stutters,       # was: remove_stutters= (v1, ignored)
                long_silences=remove_silences,  # was: remove_silences= (v1, ignored)
                mouth_sounds=True,
                breath=True,
                output_path=clean_out,
            )

            # output_path= is honoured in most SDK builds, but not all —
            # fall back to result.audio.download() if the file wasn't written
            if not os.path.exists(clean_out):
                print("[Denoiser] output_path not written — trying result.audio.download()")
                result.audio.download(clean_out)

            if not os.path.exists(clean_out) or os.path.getsize(clean_out) == 0:
                raise RuntimeError("Cleanvoice returned an empty or missing output file")

            # Clean up temp WAV if we created one
            if tmp_wav and os.path.exists(tmp_wav):
                os.unlink(tmp_wav)

            out_kb = os.path.getsize(clean_out) / 1024
            print(f"[Denoiser] ✅ Cleanvoice v2 done → {clean_out} ({out_kb:.1f} KB)")

            stats.update({
                "noise_method":     "Cleanvoice SDK v2 ✅",
                "fillers_removed":  "yes" if remove_fillers  else "no",
                "stutters_removed": "yes" if remove_stutters else "no",
                "silences_removed": "yes" if remove_silences else "no",
            })
            return {"audio_path": clean_out, "stats": stats}

        except Exception as e:
            # Clean up temp WAV on failure too
            if tmp_wav and os.path.exists(tmp_wav):
                try:
                    os.unlink(tmp_wav)
                except Exception:
                    pass

            # Print the FULL traceback — never hide it
            print("\n[Denoiser] ❌ Cleanvoice v2 FAILED — full traceback below:")
            traceback.print_exc()
            print(f"[Denoiser] ❌ Error: {type(e).__name__}: {e}\n")
            logger.error("[Denoiser] Cleanvoice v2 failed", exc_info=True)

            stats["noise_method"] = f"Cleanvoice FAILED ({type(e).__name__}): {str(e)[:200]}"
            return {"audio_path": audio_path, "stats": stats}

    # ══════════════════════════════════════════════════════════════════
    # FORMAT CONVERSION
    # ══════════════════════════════════════════════════════════════════
    def _ensure_wav(self, audio_path: str, out_dir: str):
        """
        Converts audio to 16kHz mono WAV if it's not already in a format
        Cleanvoice reliably accepts (.mp3 or .wav).

        Returns (path_to_upload, temp_wav_path_or_None).
        Caller must delete temp_wav_path after the API call.
        """
        ext = os.path.splitext(audio_path)[1].lower()
        if ext in CLEANVOICE_SAFE_FORMATS:
            print(f"[Denoiser] Format {ext} — no conversion needed")
            return audio_path, None

        # Convert to 16kHz mono WAV — safest format for Cleanvoice upload
        wav_path = os.path.join(out_dir, "cleanvoice_input.wav")
        print(f"[Denoiser] Converting {ext} → WAV for Cleanvoice upload...")
        try:
            r = subprocess.run([
                "ffmpeg", "-y", "-i", audio_path,
                "-vn",                  # strip video if present
                "-ar", "16000",         # 16kHz — Cleanvoice's preferred rate
                "-ac", "1",             # mono
                "-acodec", "pcm_s16le", # uncompressed PCM — always accepted
                wav_path
            ], capture_output=True)

            if r.returncode != 0 or not os.path.exists(wav_path):
                stderr = r.stderr.decode("utf-8", errors="replace")
                raise RuntimeError(f"ffmpeg conversion failed: {stderr[-300:]}")

            kb = os.path.getsize(wav_path) / 1024
            print(f"[Denoiser] ✅ Converted to WAV: {wav_path} ({kb:.1f} KB)")
            return wav_path, wav_path

        except Exception as e:
            print(f"[Denoiser] ⚠️  WAV conversion failed ({e}) — trying original file")
            return audio_path, None

    # ══════════════════════════════════════════════════════════════════
    # TEXT-SIDE FILLER CLEANUP
    # ══════════════════════════════════════════════════════════════════
    def clean_transcript_fillers(self, text: str) -> str:
        """Remove filler words from transcript text."""
        cleaned = re.sub(FILLER_PATTERN, '', text, flags=re.IGNORECASE)
        cleaned = re.sub(r'\s+', ' ', cleaned).strip()
        return cleaned

    # Kept for backwards compatibility
    def _remove_fillers(self, a, s, seg):
        return a, 0

    def _remove_stutters(self, a, s, seg):
        return a, 0