Spaces:
Running
Running
| from pyannote.audio import Pipeline | |
| from config import settings | |
| _diar_pipeline = Pipeline.from_pretrained(settings.DIAR_MODEL) | |
| def diarize_segments(audio_path, segments): | |
| """ | |
| Konuşma segmentlerini konuşmacılara göre ayırır | |
| Parameters: | |
| ----------- | |
| audio_path : str | |
| Ses dosyasının tam yolu | |
| segments : list | |
| ASR tarafından oluşturulan zaman damgalı segmentler | |
| Returns: | |
| -------- | |
| list | |
| Konuşmacı bilgisi eklenmiş segmentler | |
| """ | |
| try: | |
| # Pyannote modelini yükle | |
| import os | |
| import torch | |
| from pyannote.audio import Pipeline | |
| # HuggingFace token kontrolü | |
| token = os.environ.get("HF_TOKEN") | |
| if not token: | |
| print("UYARI: HF_TOKEN bulunamadı, diyarizasyon atlanıyor.") | |
| # Tüm segmentleri Konuşmacı 1 olarak işaretle | |
| for seg in segments: | |
| seg["speaker"] = "Konuşmacı 1" | |
| return segments | |
| # Pyannote pipeline'ını oluştur | |
| pipeline = Pipeline.from_pretrained( | |
| "pyannote/speaker-diarization-3.0", | |
| use_auth_token=token | |
| ) | |
| # GPU varsa kullan | |
| if torch.cuda.is_available(): | |
| pipeline = pipeline.to(torch.device("cuda")) | |
| # Diyarizasyon yap | |
| diarization = pipeline(audio_path) | |
| # Konuşmacı etiketlerini oluştur (Hekim, Asistan, vs.) | |
| speaker_labels = {} | |
| # Segmentlere konuşmacı bilgisi ekle | |
| for seg in segments: | |
| seg_start = seg["start"] | |
| seg_end = seg["end"] | |
| # Bu segment için en çok konuşan kişiyi bul | |
| speaker_times = {} | |
| for turn, _, speaker in diarization.itertracks(yield_label=True): | |
| # Segment ve konuşma turu arasındaki çakışmayı hesapla | |
| overlap_start = max(seg_start, turn.start) | |
| overlap_end = min(seg_end, turn.end) | |
| if overlap_end > overlap_start: # Çakışma varsa | |
| duration = overlap_end - overlap_start | |
| speaker_times[speaker] = speaker_times.get(speaker, 0) + duration | |
| # En uzun konuşan kişiyi bul | |
| if speaker_times: | |
| max_speaker = max(speaker_times, key=speaker_times.get) | |
| # Konuşmacıyı etiketle | |
| if max_speaker not in speaker_labels: | |
| # Yeni konuşmacı tespit edildi | |
| # Pediatri bağlamında etiketler belirle (ilk konuşmacı genelde hekim) | |
| idx = len(speaker_labels) + 1 | |
| if idx == 1: | |
| label = "Hekim" | |
| elif idx == 2: | |
| label = "Asistan" | |
| elif idx == 3: | |
| label = "Ebeveyn" | |
| else: | |
| label = f"Konuşmacı {idx}" | |
| speaker_labels[max_speaker] = label | |
| seg["speaker"] = speaker_labels[max_speaker] | |
| else: | |
| # Diyarizasyon bilgisi yoksa | |
| seg["speaker"] = "Bilinmeyen Konuşmacı" | |
| return segments | |
| except Exception as e: | |
| # Hata durumunda basit etiketleme yap | |
| print(f"Diyarizasyon hatası: {str(e)}") | |
| for seg in segments: | |
| seg["speaker"] = "Konuşmacı" | |
| return segments | |