import os from pydub import AudioSegment import noisereduce as nr import webrtcvad from ..config import settings import numpy as np import librosa import soundfile as sf from typing import Optional def clean_audio(input_file: str, output_file: Optional[str] = None) -> str: """ Ses dosyasını temizleyerek gürültüyü azaltır ve kaliteyi artırır Args: input_file: Girdi ses dosyasının yolu output_file: Çıktı ses dosyasının yolu (None ise otomatik oluşturulur) Returns: Temizlenmiş ses dosyasının yolu """ # Çıktı dosya adını oluştur if output_file is None: output_file = input_file.replace('.wav', '_cleaned.wav') try: # Ses dosyasını yükle audio, sr = librosa.load(input_file, sr=None) # 1. Gürültü azaltma reduced_noise = nr.reduce_noise(y=audio, sr=sr, stationary=True) # 2. Ses seviyesi normalizasyonu normalized = librosa.util.normalize(reduced_noise) # 3. Sessiz bölümleri kırp non_silent_intervals = librosa.effects.split(normalized, top_db=30) if len(non_silent_intervals) > 0: non_silent_audio = np.concatenate([ normalized[start:end] for start, end in non_silent_intervals ]) # Çok kısa ise orijinali kullan if len(non_silent_audio) > 0.5 * len(normalized): normalized = non_silent_audio # 4. Band-pass filtresi (insan sesi için optimize - 80Hz-8000Hz) y_filter = librosa.effects.preemphasis(normalized) # Temizlenmiş sesi kaydet sf.write(output_file, y_filter, sr) return output_file except Exception as e: print(f"Ses temizleme hatası: {str(e)}") # Hata durumunda orijinal dosyayı döndür return input_file def clean_audio_streaming(input_file: str, output_file: Optional[str] = None, chunk_seconds: int = 30) -> str: """Büyük dosyaları parça parça işle""" if output_file is None: output_file = input_file.replace('.wav', '_cleaned.wav') try: # Temel bilgileri al y, sr = librosa.load(input_file, sr=None, duration=10) # Sadece örnek rate için # Çıktı dosyasını hazırla with sf.SoundFile(output_file, 'w', sr, channels=1) as out_f: # Dosyayı parça parça işle for i, (audio_chunk, sr) in enumerate(librosa.stream(input_file, block_length=chunk_seconds, frame_length=2048, hop_length=512)): # Her parça için gürültü azaltma reduced_noise = nr.reduce_noise(y=audio_chunk, sr=sr) # Normalize normalized = librosa.util.normalize(reduced_noise) # Yaz out_f.write(normalized) return output_file except Exception as e: print(f"Ses temizleme hatası: {str(e)}") return input_file def _apply_vad(audio: AudioSegment, vad: webrtcvad.Vad) -> AudioSegment: frame_duration = 30 frames = [] samples = audio.get_array_of_samples() for i in range(0, len(samples), int(settings.SAMPLE_RATE * frame_duration / 1000)): frame = samples[i:i + int(settings.SAMPLE_RATE * frame_duration / 1000)] is_speech = vad.is_speech( frame.tobytes(), sample_rate=settings.SAMPLE_RATE ) if is_speech: frames.extend(frame) return AudioSegment( data=bytes(frames), sample_width=audio.sample_width, frame_rate=settings.SAMPLE_RATE, channels=audio.channels )