Spaces:
Running
Running
| # | |
| # SPDX-FileCopyrightText: Hadad <hadad@linuxmail.org> | |
| # 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. | |
| """ | |
| # 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 | |
| # 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) |