# # SPDX-FileCopyrightText: Hadad # SPDX-License-Identifier: Apache-2.0 # import asyncio # Import asyncio to enable asynchronous programming, including async/await syntax and event loop management import httpx # Import httpx library to perform asynchronous HTTP requests with advanced features like connection pooling and timeout control import aiohttp # Import aiohttp library to provide an alternative asynchronous HTTP client for flexible request handling from urllib.parse import quote # Import quote function to safely encode strings for URL usage, escaping special characters from src.utils.ip_generator import generate_ip # Import a custom utility function to generate random IP addresses from config import auth # Import authentication credentials or configuration from the config module (not used directly here but imported for completeness) from src.utils.tools import initialize_tools # Import utility function to initialize and retrieve service endpoints or tool URLs # Define a class named AudioGeneration to encapsulate functionalities related to generating audio content class AudioGeneration: # This class provides methods to create audio files based on text instructions and voice parameters """ This class provides methods to generate audio files from text instructions asynchronously. It supports retrying requests until successful audio generation is confirmed. """ @staticmethod # This method does not depend on class instance state and can be called directly on the class async def create_audio(generate_audio_instruction: str, voice: str = "echo") -> str: """ Asynchronously generate an audio file URL by sending a request to an audio generation service. The method continuously retries until it receives a successful HTTP 200 response with audio content. Args: generate_audio_instruction (str): The text instruction or content to convert into audio. voice (str, optional): The voice style or effect to apply on the generated audio. Defaults to "echo". Returns: str: The URL of the generated audio file upon successful retrieval. Raises: Exception: Currently, the method retries indefinitely and does not raise exceptions on failure. """ # Encode the text instruction to make it safe for URL path inclusion by escaping special characters generate_audio_instruct = quote(generate_audio_instruction) # Initialize tools and extract the audio generation service endpoint (third element in the returned tuple) _, _, audio_tool = initialize_tools() # Construct the full URL by appending the encoded instruction to the base audio tool URL url = f"{audio_tool}/{generate_audio_instruct}" # Define query parameters specifying the audio generation model and voice effect params = { "model": "openai-audio", # Specify the model used by the audio generation service "voice": voice # Specify the voice style or effect for the generated audio } # Create an aiohttp asynchronous HTTP client session with no timeout to allow long-running requests async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=None)) as session: # Enter an infinite loop to retry the request until success criteria are met while True: # Generate a random IP address to spoof the client's origin in the request headers headers = { "X-Forwarded-For": generate_ip() # Set the X-Forwarded-For header to a random IP address } try: # Perform an asynchronous GET request to the audio generation service with URL, parameters, and headers async with session.get(url, params=params, headers=headers) as resp: # Check if the response status code is 200 (OK) and content type indicates MPEG audio stream content_type = resp.headers.get('Content-Type', '') if resp.status == 200 and 'audio/mpeg' in content_type: # Return the final URL of the generated audio resource as a string return str(resp.url) else: # If the response is not successful or content type is unexpected, wait before retrying await asyncio.sleep(15) # Pause for 15 seconds to avoid overwhelming the server except aiohttp.ClientError: # Catch network-related errors such as connection issues and wait before retrying await asyncio.sleep(15) # Pause for 15 seconds before retrying after an exception @staticmethod # Provide an alternative implementation using httpx for flexibility or fallback async def create_audio_httpx(generate_audio_instruction: str, voice: str = "echo") -> str: """ Alternative asynchronous method to generate audio using httpx client. This method also retries indefinitely until a successful response with audio content is received. Args: generate_audio_instruction (str): The text instruction to convert into audio. voice (str, optional): Voice style or effect. Defaults to "echo". Returns: str: URL of the generated audio file. """ # Encode instruction for safe URL usage generate_audio_instruct = quote(generate_audio_instruction) # Initialize tools and get audio generation endpoint _, _, audio_tool = initialize_tools() # Construct request URL url = f"{audio_tool}/{generate_audio_instruct}" # Define query parameters for the request params = { "model": "openai-audio", "voice": voice } # Create an asynchronous HTTP client with no timeout limit async with httpx.AsyncClient(timeout=None) as client: # Retry loop until success while True: # Define HTTP headers for the request, including random IP address to simulate different client origins headers = { "X-Forwarded-For": generate_ip() # Generate and set a random IP address for the request header } try: # Send GET request asynchronously resp = await client.get(url, params=params, headers=headers) # Check for successful response with audio content type if resp.status_code == 200 and 'audio/mpeg' in resp.headers.get('Content-Type', ''): # Return the URL of generated audio return str(resp.url) else: # Wait before retrying on failure await asyncio.sleep(15) except httpx.RequestError: # Handle network errors and wait before retrying await asyncio.sleep(15)