All Voice Lab MCP Server

import logging
import os
import random
import time

import requests
from mcp.types import TextContent

from .base import get_client


def text_translation_tool(
    file_path: str,
    target_lang: str,
    source_lang: str = "auto",
    output_dir: str = None
) -> TextContent:
    all_voice_lab = get_client()
    output_dir = all_voice_lab.get_output_path(output_dir)
    max_polling_time = 600
    polling_interval = 10
    logging.info(
        f"Tool called: text_translation, file_path: {file_path}, target_lang: {target_lang}, source_lang: {source_lang}")
    logging.info(f"Max polling time: {max_polling_time}s, Polling interval: {polling_interval}s")

    
    if not file_path:
        logging.warning("File path parameter is empty")
        return TextContent(
            type="text",
            text="file_path parameter cannot be empty"
        )

    
    if not os.path.exists(file_path):
        logging.warning(f"File does not exist: {file_path}")
        return TextContent(
            type="text",
            text=f"File does not exist: {file_path}"
        )

    
    _, file_extension = os.path.splitext(file_path)
    file_extension = file_extension.lower()
    if file_extension not in [".txt", ".srt"]:
        logging.warning(f"Unsupported file format: {file_extension}")
        return TextContent(
            type="text",
            text=f"Unsupported file format. Only TXT and SRT formats are supported."
        )

    
    max_size_bytes = 10 * 1024 * 1024  # 10MB in bytes
    file_size = os.path.getsize(file_path)
    if file_size > max_size_bytes:
        logging.warning(f"File size exceeds limit: {file_size} bytes, max allowed: {max_size_bytes} bytes")
        return TextContent(
            type="text",
            text=f"File size exceeds the maximum limit of 10MB. Please use a smaller file."
        )

    try:
        if all_voice_lab is None:
            logging.error("all_voice_lab client is not initialized.")
            return TextContent(type="text",
                               text="Error: AllVoiceLab client not initialized. Please check server setup.")

        
        logging.info("Starting text translation process")
        project_id = all_voice_lab.text_translation(
            file_path=file_path,
            target_lang=target_lang,
            source_lang=source_lang
        )
        logging.info(f"Text translation task submitted. Project ID: {project_id}")

        
        logging.info(f"Starting to poll translation status for Project ID: {project_id}")
        start_time = time.time()
        completed = False

        
        while time.time() - start_time < max_polling_time:
            try:
                
                translation_result = all_voice_lab.get_text_translation_result(project_id)
                if translation_result is None:
                    logging.warning(f"Failed to get translation result for Project ID: {project_id}")
                    time.sleep(polling_interval)
                    continue

                logging.info(f"Translation status: {translation_result.status} for Project ID: {project_id}")

                
                if translation_result.status.lower() == "success":
                    logging.info(f"Text translation completed for Project ID: {project_id}")
                    completed = True

                    
                    if translation_result.result and translation_result.result.startswith("http"):
                        try:
                            
                            os.makedirs(output_dir, exist_ok=True)

                            
                            # Get original filename without extension
                            original_filename = os.path.splitext(os.path.basename(file_path))[0]
                            timestamp = int(time.time())
                            random_suffix = ''.join(random.choices('abcdefghijklmnopqrstuvwxyz0123456789', k=6))
                            filename = f"{original_filename}_translation_{timestamp}_{random_suffix}.txt"
                            file_path = os.path.join(output_dir, filename)

                            
                            logging.info(f"Downloading translation result from URL: {translation_result.result}")
                            response = requests.get(translation_result.result, timeout=30)
                            response.raise_for_status()

                            
                            with open(file_path, 'wb') as f:
                                f.write(response.content)

                            
                            with open(file_path, 'r', encoding='utf-8') as f:
                                translated_text = f.read()

                            logging.info(f"Translation result downloaded and saved to: {file_path}")

                            
                            result_text = f"Translation completed successfully.\n\n"
                            result_text += f"Source Language: {translation_result.source_lang}\n"
                            result_text += f"Target Language: {translation_result.target_lang}\n\n"
                            result_text += f"Translated Text:\n{translated_text}\n\n"
                            result_text += f"Result file saved at: {file_path}"

                            return TextContent(
                                type="text",
                                text=result_text
                            )
                        except Exception as e:
                            logging.error(f"Error downloading translation result: {str(e)}")
                            result_text = f"Translation completed, but failed to download result: {str(e)}\n\n"
                            result_text += f"Source Language: {translation_result.source_lang}\n"
                            result_text += f"Target Language: {translation_result.target_lang}\n\n"
                            result_text += f"Result URL: {translation_result.result}"

                            return TextContent(
                                type="text",
                                text=result_text
                            )
                    else:
                        
                        result_text = f"Translation completed successfully.\n\n"
                        result_text += f"Source Language: {translation_result.source_lang}\n"
                        result_text += f"Target Language: {translation_result.target_lang}\n\n"
                        result_text += f"No result URL available."

                        return TextContent(
                            type="text",
                            text=result_text
                        )
                elif translation_result.status.lower() == "failed":
                    logging.error(
                        f"Translation failed for Project ID: {project_id}")
                    return TextContent(
                        type="text",
                        text=f"Translation failed"
                    )

            except Exception as e:
                logging.error(f"Error while polling translation status: {str(e)}")

            
            time.sleep(polling_interval)

        
        if not completed:
            logging.warning(f"Translation not completed within {max_polling_time} seconds for Project ID: {project_id}")
            return TextContent(
                type="text",
                text=f"Translation is still in progress. Please check the status later using the project ID: {project_id}"
            )

    except Exception as e:
        logging.error(f"Text translation failed: {str(e)}")
        return TextContent(
            type="text",
            text=f"Translation failed: {str(e)}"
        )