src/tools/audio.py

#
# SPDX-FileCopyrightText: Hadad <hadad@linuxmail.org>
# SPDX-License-Identifier: Apache-2.0
#

import aiohttp  # Import aiohttp library to handle asynchronous HTTP requests and responses
from urllib.parse import quote  # Import quote function to safely encode URL query parameters
from src.utils.ip_generator import generate_ip  # Import custom utility to generate random IP addresses for request headers
from src.utils.tools import initialize_tools  # Import initialize_tools function to set up necessary API tools and retrieve tokens

# Define a class to handle audio generation tasks asynchronously
class AudioGeneration:
    """
    This class provides functionality to generate audio content based on textual instructions asynchronously.
    It encapsulates all necessary steps including preparing the request, encoding parameters, initializing tools,
    making HTTP requests to the audio generation service, and handling the response.
    
    The design leverages asynchronous programming to efficiently manage network I/O without blocking the main thread.
    It uses external utilities for IP generation to simulate diverse client requests and for tool initialization to
    obtain API endpoints and authorization tokens securely.
    
    This class is intended for integration in larger applications requiring dynamic audio content creation from text,
    supporting scalability and responsiveness in asynchronous environments.
    """
    @staticmethod  # Decorator indicating this method does not depend on instance state
    # Asynchronous method to create an audio based on given instructions and parameters
    async def create_audio(generate_audio_instruction: str) -> str:
        """
        Asynchronously generates an audio URL from a given textual instruction describing the desired audio content.
        
        This method performs the following steps:
        - Encodes the input text to ensure it is safe for URL transmission
        - Initializes necessary tools and retrieves the audio generation endpoint and authorization token
        - Constructs the request URL and query parameters specifying the audio model and voice characteristics
        - Sets up HTTP headers including authorization and IP address to simulate client diversity
        - Opens an asynchronous HTTP session and sends a GET request to the audio generation service
        - Checks the response status and content type to verify successful audio generation
        - Returns the URL of the generated audio if successful, otherwise raises an exception
        
        The method is designed to be non-blocking and suitable for use in asynchronous workflows.
        
        Args:
            generate_audio_instruction (str): A textual description or instruction for the audio to be generated
        
        Returns:
            str: The URL pointing to the generated audio resource
        
        Raises:
            Exception: If the HTTP response indicates failure or the content is not audio
        """
        # Encode the textual instruction to be safely included in a URL path segment
        generate_audio_instruct = quote(generate_audio_instruction)
        
        # Initialize external tools and retrieve the audio generation endpoint and authorization token
        _, _, audio_tool, poll_token = initialize_tools()
        
        # Construct the full URL by appending the encoded instruction to the audio tool base URL
        url = f"{audio_tool}/{generate_audio_instruct}"
        
        # Define query parameters specifying the audio generation model and voice style to be used
        params = {
            "model": "openai-audio",  # Specify the audio generation model identifier
            "voice": "echo"  # Specify the voice style or effect to apply to the generated audio
        }

        # Prepare HTTP headers
        headers = {
            "Authorization": f"Bearer {poll_token}",  # Bearer token for API authorization
            "X-Forwarded-For": generate_ip()  # Random IP address for request header to simulate client origin
        }

        # Create an asynchronous HTTP client session to manage connections efficiently
        async with aiohttp.ClientSession() as session:
            """
            This block manages the lifecycle of the HTTP client session, ensuring proper connection handling,
            resource cleanup, and support for asynchronous request execution to avoid blocking the event loop.
            """
            # Send an asynchronous GET request to the audio generation service with specified URL, parameters, and headers
            async with session.get(
                url,  # The fully constructed URL including the encoded instruction
                params=params,  # Query parameters dictating model and voice options
                headers=headers  # Authorization and client identity headers
            ) as resp:
                """
                This context manages the HTTP response object, allowing inspection of status, headers,
                and content in an asynchronous manner, suitable for streaming or large payloads.
                """
                # Verify that the response status is HTTP 200 OK and the content type indicates an audio MPEG file
                if resp.status == 200 and 'audio/mpeg' in resp.headers.get('Content-Type', ''):
                    # Return the final URL from which the generated audio can be accessed or downloaded
                    return str(resp.url)
                # If the response is not successful or content type is unexpected, raise an exception
                raise Exception()