Update core/audio_gen.py

core/audio_gen.py

@@ -0,0 +1,139 @@
+# @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")
+        }