core/audio_gen.py

# @title core/audio_gen.py
import os
import json
import httpx
from pathlib import Path
from typing import Dict, Any, Optional, Tuple
from utils.logger import SpotMakerLogger

class AudioGenerator:
    """
    Generator ścieżki audio dla głosu lektora używając Hailuo T2A API.
    """
    def __init__(self, api_key: str, group_id: str, output_dir: str, logger: SpotMakerLogger):
        self.api_key = api_key
        self.group_id = group_id
        self.logger = logger
        self.output_dir = Path(output_dir)
        self.output_dir.mkdir(parents=True, exist_ok=True)
        
        # Konfiguracja API
        self.base_url = "https://api.minimaxi.chat/v1/t2a_v2"
        self.model = "speech-01-hd"
        
    async def generate_voice_over(self, text: str) -> Tuple[str, float]:
        """
        Generuje ścieżkę dźwiękową dla tekstu lektora.
        
        Args:
            text: Tekst do przeczytania przez lektora
            
        Returns:
            Tuple[str, float]: Ścieżka do pliku audio i jego długość w sekundach
        """
        try:
            self.logger.log("audio_gen", "Rozpoczynam generowanie audio", "info")
            self.logger.update_progress("audio_gen", 10, "Przygotowuję tekst...")
            
            # Konstruowanie URL z GroupId
            url = f"{self.base_url}?GroupId={self.group_id}"
            
            # Przygotuj dane dla API zgodnie z dokumentacją
            headers = {
                "Content-Type": "application/json",
                "Authorization": f"Bearer {self.api_key}",
                "accept": "application/json, text/plain, */*"
            }
            
            data = {
                "model": self.model,
                "text": text,
                "stream": False,
                "subtitle_enable": True,
                "voice_setting": {
                    "voice_id": "male-qn-qingse",
                    "speed": 1.0,
                    "vol": 1.0,
                    "pitch": 0
                },
                "audio_setting": {
                    "sample_rate": 32000,
                    "bitrate": 128000,
                    "format": "mp3",
                    "channel": 1
                }
            }
            
            # Log request details
            self.logger.log("audio_gen", 
                          f"Wysyłam zapytanie do: {url}\nDane: {json.dumps(data, indent=2)}", 
                          "debug")
            
            self.logger.update_progress("audio_gen", 30, "Generuję audio...")
            
            # Wywołaj API
            async with httpx.AsyncClient() as client:
                response = await client.post(
                    url,
                    headers=headers,
                    json=data,
                    timeout=30.0
                )
                
                # Log response
                self.logger.log("audio_gen", 
                              f"Status odpowiedzi: {response.status_code}", 
                              "debug")
                
                if response.status_code != 200:
                    error_msg = f"Błąd API: {response.status_code} - {response.text}"
                    self.logger.log("audio_gen", error_msg, "error")
                    raise Exception(error_msg)
                
                response_data = response.json()
                
                # Sprawdź status_code z API
                if response_data.get('base_resp', {}).get('status_code') != 0:
                    error_msg = response_data.get('base_resp', {}).get('status_msg', 'Unknown API error')
                    raise Exception(f"API error: {error_msg}")

            self.logger.update_progress("audio_gen", 60, "Przetwarzam odpowiedź...")
            
            # Log response data
            self.logger.log("audio_gen", 
                      f"Otrzymana odpowiedź:\n{json.dumps(response_data, indent=2)}", 
                      "debug")
            
            try:
                # Pobierz audio z odpowiedzi zgodnie z dokumentacją
                audio_data = response_data['data']['audio']
                audio_length = response_data['extra_info']['audio_length'] / 1000.0  # konwersja na sekundy
                
                self.logger.log("audio_gen", 
                              f"Długość audio: {audio_length:.2f}s", 
                              "info")
                
                # Zapisz plik audio
                output_path = self.output_dir / "voice_over.mp3"
                with open(output_path, 'wb') as f:
                    f.write(bytes.fromhex(audio_data))
                    
                self.logger.update_progress("audio_gen", 100, "Zakończono generowanie audio")
                
                return str(output_path), audio_length
                
            except KeyError as e:
                error_msg = f"Błąd w strukturze odpowiedzi: {str(e)}"
                self.logger.log("audio_gen", error_msg, "error")
                raise Exception(error_msg)
            
        except Exception as e:
            self.logger.format_error("audio_gen", e)
            raise
            
    def get_status(self) -> Dict[str, Any]:
        """Zwraca aktualny status generatora."""
        return {
            'progress': self.logger.get_module_progress("audio_gen"),
            'status': self.logger.get_module_status("audio_gen")
        }