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