""" title: LibreTranslate Action description: Advanced translation capabilities using LibreTranslate API with retry mechanisms and language detection author: @iamg30 author_url: https://openwebui.com/u/iamg30/ forked from https://openwebui.com/f/jthesse/libretranslate_action funding_url: https://github.com/open-webui version: 0.1.0 license: MIT icon_url: data:image/svg+xml;base64,PHN2ZyB4bWxucz0iaHR0cDovL3d3dy53My5vcmcvMjAwMC9zdmciIHdpZHRoPSIxZW0iIGhlaWdodD0iMWVtIiB2aWV3Qm94PSIwIDAgMjQgMjQiPjxwYXRoIGZpbGw9ImN1cnJlbnRDb2xvciIgZD0ibTEyLjg3IDE1LjA3bC0yLjU0LTIuNTFsLjAzLS4wM0ExNy41IDE3LjUgMCAwIDAgMTQuMDcgNkgxN1Y0aC03VjJIOHYySDF2MmgxMS4xN0MxMS41IDcuOTIgMTAuNDQgOS43NSA5IDExLjM1QzguMDcgMTAuMzIgNy4zIDkuMTkgNi42OSA4aC0yYy43MyAxLjYzIDEuNzMgMy4xNyAyLjk4IDQuNTZsLTUuMDkgNS4wMkw0IDE5bDUtNWwzLjExIDMuMTF6TTE4LjUgMTBoLTJMMTIgMjJoMmwxLjEyLTNoNC43NUwyMSAyMmgyem0tMi42MiA3bDEuNjItNC4zM0wxOS4xMiAxN3oiLz48L3N2Zz4= """ from pydantic import BaseModel, Field from typing import Optional, List, Dict, Any from tenacity import retry, stop_after_attempt, wait_fixed, retry_if_exception_type import requests import asyncio import time import logging from open_webui.utils.misc import get_last_assistant_message # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger("LibreTranslateAction") # Pre-defined list of supported languages SUPPORTED_LANGUAGES = { "sq": "Albanian", "ar": "Arabic", "az": "Azerbaijani", "bn": "Bengali", "bg": "Bulgarian", "ca": "Catalan, Valencian", "zh": "Chinese", "cs": "Czech", "da": "Danish", "nl": "Dutch, Flemish", "en": "English", "eo": "Esperanto", "et": "Estonian", "fi": "Finnish", "fr": "French", "de": "German", "el": "Greek", "he": "Hebrew", "hi": "Hindi", "hu": "Hungarian", "id": "Indonesian", "ga": "Irish", "it": "Italian", "ja": "Japanese", "ko": "Korean", "lv": "Latvian", "lt": "Lithuanian", "ms": "Malay", "fa": "Persian", "pl": "Polish", "pt": "Portuguese", "ro": "Romanian, Moldavian, Moldovan", "ru": "Russian", "sk": "Slovak", "sl": "Slovenian", "es": "Spanish, Castilian", "sv": "Swedish", "tl": "Tagalog", "th": "Thai", "tr": "Turkish", "uk": "Ukrainian", "ur": "Urdu", } class Action: class Valves(BaseModel): LIBRE_TRANSLATE_URL: str = Field( default="http://localhost:5000", description="The URL of the LibreTranslate instance.", ) API_KEY: str = Field( default="", description="API key for LibreTranslate (if required).", ) ENABLE_LANGUAGE_DETECTION: bool = Field( default=True, description="Automatically detect source language when set to 'auto'.", ) MAX_RETRY_ATTEMPTS: int = Field( default=3, description="Maximum number of retry attempts for API calls.", ) RETRY_DELAY_SECONDS: int = Field( default=2, description="Delay between retry attempts in seconds.", ) class UserValves(BaseModel): SOURCE_LANGUAGE: str = Field( default="auto", description="User-specific source language for assistant messages", ) TARGET_LANGUAGE: str = Field( default="de", description="User-specific target language for assistant messages", ) def __init__(self): self.valves = self.Valves() self.api_languages = [] async def get_supported_languages(self): """Fetch supported languages from the LibreTranslate API.""" try: response = requests.get(f"{self.valves.LIBRE_TRANSLATE_URL}/languages") if response.status_code == 200: self.api_languages = response.json() return self.api_languages else: logger.error( f"Failed to get supported languages: {response.status_code} - {response.text}" ) return [] except Exception as e: logger.error(f"Error fetching supported languages: {str(e)}") return [] async def detect_language(self, text: str) -> str: """Detect the language of the provided text.""" if not text or not self.valves.ENABLE_LANGUAGE_DETECTION: return "auto" payload = {"q": text[:1000]} # Use first 1000 chars for detection if self.valves.API_KEY: payload["api_key"] = self.valves.API_KEY try: response = requests.post( f"{self.valves.LIBRE_TRANSLATE_URL}/detect", json=payload ) if response.status_code == 200: detections = response.json() if detections and len(detections) > 0: return detections[0]["language"] return "auto" except Exception as e: logger.error(f"Language detection error: {str(e)}") return "auto" async def translate_text( self, text: str, source: str, target: str ) -> Dict[str, Any]: """Translate text with retry logic.""" if not text: raise ValueError("Empty text provided for translation") # Define retry function with fixed parameters @retry( stop=stop_after_attempt(self.valves.MAX_RETRY_ATTEMPTS), wait=wait_fixed(self.valves.RETRY_DELAY_SECONDS), retry=retry_if_exception_type( (requests.exceptions.RequestException, ConnectionError) ), reraise=True, ) def do_translate(): payload = { "q": text, "source": source, "target": target, } if self.valves.API_KEY: payload["api_key"] = self.valves.API_KEY response = requests.post( f"{self.valves.LIBRE_TRANSLATE_URL}/translate", json=payload ) if response.status_code != 200: error_msg = ( f"Translation error: {response.status_code} - {response.text}" ) logger.error(error_msg) raise requests.exceptions.RequestException(error_msg) return response.json() return do_translate() def format_language_name(self, language_code: str) -> str: """Get the human-readable name for a language code.""" if language_code == "auto": return "Auto-detect" if language_code in SUPPORTED_LANGUAGES: return f"{SUPPORTED_LANGUAGES[language_code]} ({language_code})" # Check if we have data from the API for lang in self.api_languages: if lang.get("code") == language_code: return f"{lang.get('name')} ({language_code})" return f"Unknown language ({language_code})" def is_language_supported(self, language_code: str) -> bool: """Check if a language is supported.""" if language_code == "auto": return True # Check predefined list if language_code in SUPPORTED_LANGUAGES: return True # Check API data if available for lang in self.api_languages: if lang.get("code") == language_code: return True return False async def action( self, body: dict, __user__=None, __event_emitter__=None, __event_call__=None, ) -> Optional[dict]: logger.info(f"Starting LibreTranslate action") # Handle missing event emitter if not __event_emitter__: logger.error("Event emitter not provided") return # Validate LibreTranslate URL if not self.valves.LIBRE_TRANSLATE_URL: await __event_emitter__( { "type": "status", "data": { "status": "error", "description": "Please set your LibreTranslate URL", "done": True, }, } ) return # Get user-specific settings source = __user__["valves"].SOURCE_LANGUAGE target = __user__["valves"].TARGET_LANGUAGE # Validate language codes if not self.is_language_supported(target): await __event_emitter__( { "type": "status", "data": { "status": "error", "description": f"Target language '{target}' is not supported", "done": True, }, } ) return if source != "auto" and not self.is_language_supported(source): await __event_emitter__( { "type": "status", "data": { "status": "error", "description": f"Source language '{source}' is not supported", "done": True, }, } ) return # Get the last assistant message try: messages = body["messages"] assistant_message = get_last_assistant_message(messages) if not assistant_message: await __event_emitter__( { "type": "status", "data": { "status": "error", "description": "No assistant message found to translate", "done": True, }, } ) return logger.info( f"Retrieved assistant message of length: {len(assistant_message)}" ) except Exception as e: await __event_emitter__( { "type": "status", "data": { "status": "error", "description": f"Error retrieving assistant message: {str(e)}", "done": True, }, } ) return # Try to get languages from API for more accurate validation await self.get_supported_languages() # Detect language if source is set to auto and detection is enabled actual_source = source if source == "auto" and self.valves.ENABLE_LANGUAGE_DETECTION: await __event_emitter__( { "type": "status", "data": { "description": "Detecting source language...", "done": False, }, } ) detected_lang = await self.detect_language(assistant_message) if detected_lang != "auto": actual_source = detected_lang logger.info(f"Detected source language: {actual_source}") # Format language names source_name = self.format_language_name(actual_source) target_name = self.format_language_name(target) # Update status for translation await __event_emitter__( { "type": "status", "data": { "description": f"Translating from {source_name} to {target_name}...", "done": False, }, } ) # Perform translation with retry logic try: data = await self.translate_text(assistant_message, actual_source, target) # Format the translated text content = f""" --- ### Translation ({target_name}) {data["translatedText"]} """ # Send the translated message await __event_emitter__({"type": "message", "data": {"content": content}}) await __event_emitter__( { "type": "status", "data": {"description": "Translation complete", "done": True}, } ) logger.info("Translation successfully completed and delivered") except Exception as e: logger.error(f"Translation failed: {str(e)}") await __event_emitter__( { "type": "status", "data": { "status": "error", "description": f"Translation error: {str(e)}", "done": True, }, } )