denoiser.py

import os
import requests
import tempfile
import logging
import re
import subprocess

logger = logging.getLogger(__name__)

class Denoiser:
    def __init__(self):
        # FIX: read API key from environment instead of hard-coding it in source
        self.api_key = os.environ.get("CLEANVOICE_API_KEY", "")
        print("[Denoiser] ☁️ WhatsApp-Ready Engine Loaded")

    def process(self, audio_path: str, out_dir: str, **kwargs) -> dict:
        # Step A: Convert WhatsApp/Any format to standard WAV using FFmpeg
        standard_input = os.path.join(out_dir, "clean_input_pre.wav")
        # FIX: check=True so a failed conversion raises immediately with a clear error
        # instead of silently passing a missing file to the Cleanvoice API
        result = subprocess.run([
            "ffmpeg", "-y", "-i", audio_path,
            "-ar", "16000", "-ac", "1", standard_input
        ], capture_output=True)
        if result.returncode != 0:
            raise RuntimeError(
                f"[Denoiser] ffmpeg conversion failed (code {result.returncode}): "
                f"{result.stderr.decode(errors='replace').strip()}"
            )

        try:
            from cleanvoice import Cleanvoice
        except ImportError:
            raise RuntimeError("Please add cleanvoice-sdk to your requirements.txt")

        cv = Cleanvoice({'api_key': self.api_key})
        
        # Step B: Process through Cleanvoice API
        result = cv.process(standard_input, remove_noise=True, studio_sound=True, **kwargs)
        
        # Step C: Download processed audio
        audio_data = requests.get(result.audio.url).content
        final_wav = os.path.join(out_dir, "processed_internal.wav")
        with open(final_wav, "wb") as f:
            f.write(audio_data)
            
        return {'audio_path': final_wav}

    def clean_transcript_fillers(self, transcript: str) -> str:
        words = transcript.split()
        fillers = {"um", "umm", "uh", "basically", "like", "ante", "ane"}
        return " ".join([w for w in words if re.sub(r'[^a-z]', '', w.lower()) not in fillers])

    def _remove_fillers(self, audio, sr, segments): return audio, 0
    def _remove_stutters(self, audio, sr, segments): return audio, 0