totxt

totxt_gigaam.py

from pyannote.audio import Model
from pyannote.audio.pipelines import VoiceActivityDetection
from pyannote.core import Segment, Timeline
from datetime import timedelta
import gigaam
import tempfile
import os
from pydub import AudioSegment
import warnings
import json

# Игнорируем предупреждения для чистоты вывода
warnings.filterwarnings('ignore')

def split_audio_segments(audio_path, max_duration=25.0, 
                        hf_token="YOUR_HUGGING_FACE_TOKEN"):
    """
    Разбивает аудиофайл на сегменты длиной не более max_duration секунд,
    распознает каждый сегмент и сохраняет результаты в JSON
    """
    
    # Загружаем основное аудио для извлечения сегментов
    print("Загрузка аудиофайла...")
    try:
        full_audio = AudioSegment.from_wav(audio_path)
    except Exception as e:
        print(f"Ошибка загрузки аудиофайла: {e}")
        return []
    
    # 1. Загрузка модели и создание пайплайна
    print("Загрузка модели VAD...")
    try:
        model = Model.from_pretrained(
            "pyannote/segmentation", 
            use_auth_token=hf_token
        )
        
        # Создание пайплайна для обнаружения речевых активностей
        pipeline = VoiceActivityDetection(segmentation=model)
        
        # Гиперпараметры для VAD
        HYPER_PARAMETERS = {
            "onset": 0.1,      # порог активации начала речи
            "offset": 0,     # порог деактивации конца речи
            "min_duration_on": 0,   # минимальная длительность речевого сегмента
            "min_duration_off": 0,  # минимальная длительность неречевого сегмента
        }
        
        pipeline.instantiate(HYPER_PARAMETERS)
        
        # Обработка аудиофайла
        print("Обработка аудиофайла VAD...")
        vad_result = pipeline(audio_path)
        
    except Exception as e:
        print(f"Ошибка при работе с VAD: {e}")
        # Если VAD не работает, создаем сегменты через равные промежутки
        print("Используем равномерное разделение...")
        vad_result = None
    
    # 4. Разделение на фрагменты
    print("Разделение на фрагменты...")
    
    def split_segment(segment, max_duration):
        """Разделяет сегмент на подсегменты длиной не более max_duration секунд"""
        segments = []
        start = segment.start
        end = segment.end
        
        if (end - start) <= max_duration:
            return [segment]
        
        while start < end:
            segment_end = min(start + max_duration, end)
            segments.append(Segment(start, segment_end))
            start = segment_end
            
        return segments
    
    timeline = Timeline()
    
    if vad_result:
        # Используем результаты VAD
        for segment in vad_result.get_timeline():
            sub_segments = split_segment(segment, max_duration)
            for sub_seg in sub_segments:
                timeline.add(sub_seg)
    else:
        # Равномерное разделение, если VAD не сработал
        duration_seconds = len(full_audio) / 1000.0  # pydub uses milliseconds
        start = 0.0
        while start < duration_seconds:
            end = min(start + max_duration, duration_seconds)
            timeline.add(Segment(start, end))
            start = end
    
    # 5. Загрузка модели распознавания речи
    print("Загрузка модели распознавания речи...")
    try:
        model_name = "rnnt" # Options: "v2_ctc" or "ctc", "v2_rnnt" or "rnnt", "v1_ctc", "v1_rnnt"
        asr_model = gigaam.load_model(model_name)
        print(f"Модель {model_name} загружена успешно")
    except Exception as e:
        print(f"Ошибка загрузки модели распознавания: {e}")
        return []
    
    # 6. Обработка каждого сегмента
    print("Обработка сегментов...")
    
    sorted_segments = sorted(timeline, key=lambda x: x.start)
    results = []
    
    with tempfile.TemporaryDirectory() as temp_dir:
        for i, segment in enumerate(sorted_segments, 1):
            start_sec = segment.start
            end_sec = segment.end
            duration = segment.duration
            
            # Извлекаем сегмент из аудио
            start_ms = int(start_sec * 1000)
            end_ms = int(end_sec * 1000)
            audio_segment = full_audio[start_ms:end_ms]
            
            # Сохраняем временный файл
            temp_file = os.path.join(temp_dir, f"segment_{i}.wav")
            audio_segment.export(temp_file, format="wav")
            
            # Распознаем речь
            try:
                print(f"Распознавание сегмента {i}/{len(sorted_segments)}...")
                transcription = asr_model.transcribe(temp_file)
                text = transcription.strip() if transcription else "[нет речи]"
            except Exception as e:
                print(f"Ошибка распознавания сегмента {i}: {e}")
                text = "[ошибка распознавания]"
            
            # Форматируем время для JSON
            start_time_str = str(timedelta(seconds=start_sec)).split(".")[0]
            end_time_str = str(timedelta(seconds=end_sec)).split(".")[0]
            
            # Сохраняем результат в формате для JSON
            results.append({
                'index': i,
                'start_seconds': round(start_sec, 2),
                'end_seconds': round(end_sec, 2),
                'duration_seconds': round(duration, 2),
                'start_time': start_time_str,
                'end_time': end_time_str,
                'text': text
            })
            
            # Вывод прогресса
            print(f"[{i:3d}] {start_time_str} - {end_time_str}: {text[:50]}...")
    
    # 7. Сохранение результатов в JSON
    print("Сохранение результатов в JSON...")
    
    # Создаем структуру данных для JSON
    output_data = {
        "audio_file": audio_path,
        "total_segments": len(results),
        "max_duration_seconds": max_duration,
        "segments": results
    }
        
    # 9. Вывод результатов в консоль
    print("\n" + "="*100)
    print("РЕЗУЛЬТАТЫ РАСПОЗНАВАНИЯ:")
    print("="*100)
    print(f"{'№':<3} {'Начало':<8} {'Конец':<8} {'Длит.':<6} {'Время начала':<12} {'Время конца':<12} {'Текст (первые 50 символов)'}")
    print("-"*100)
    
    for result in results:
        text_preview = result['text'][:50] + "..." if len(result['text']) > 50 else result['text']
        print(f"{result['index']:<3} {result['start_seconds']:<8.2f} {result['end_seconds']:<8.2f} "
              f"{result['duration_seconds']:<6.2f} {result['start_time']:<12} {result['end_time']:<12} {text_preview}")
    
    print("="*100)
    print(f"Всего сегментов: {len(results)}")
    # print(f"JSON результаты сохранены в файл: {output_file}")
    # print(f"Текстовые результаты сохранены в файл: {txt_output_file}")
    
    return output_data

# Пример использования
if __name__ == "__main__":
    # Замените на ваш реальный токен Hugging Face
    HF_TOKEN = "key"
    
    # Замените на путь к вашему аудиофайлу
    AUDIO_FILE = "../1.wav"
    
    # Выполняем сегментацию и распознавание
    results = split_audio_segments(
        audio_path=AUDIO_FILE,
        output_file="audio_segments.json",  # Теперь сохраняем в JSON
        max_duration=25.0,
        hf_token=HF_TOKEN
    )
    
    # Вывод JSON структуры для демонстрации
    print("\nСтруктура JSON результата:")
    print(json.dumps(results[:2], ensure_ascii=False, indent=2))  # Показываем первые 2 элемента