eric7/Plugins/UiExtensionPlugins/Translator/TranslatorEngines/IbmWatsonEngine.py

# -*- coding: utf-8 -*-

# Copyright (c) 2018 - 2021 Detlev Offenbach <detlev@die-offenbachs.de>
#

"""
Module implementing the IBM Watson translation engine.
"""

import json

from PyQt6.QtCore import QUrl, QByteArray, QTimer
from PyQt6.QtNetwork import (
    QNetworkAccessManager, QNetworkRequest, QNetworkReply
)

from E5Gui import EricMessageBox

from EricNetwork.EricNetworkProxyFactory import proxyAuthenticationRequired

from .TranslationEngine import TranslationEngine


class IbmWatsonEngine(TranslationEngine):
    """
    Class implementing the translation engine for the IBM Watson Language
    Translator service.
    """
    # Documentation:
    # https://www.ibm.com/watson/developercloud/language-translator
    #
    # Start page:
    # https://www.ibm.com/watson/services/language-translator/
    
    def __init__(self, plugin, parent=None):
        """
        Constructor
        
        @param plugin reference to the plugin object
        @type TranslatorPlugin
        @param parent reference to the parent object
        @type QObject
        """
        super().__init__(plugin, parent)
        
        self.__ui = parent
        
        self.__networkManager = QNetworkAccessManager(self)
        self.__networkManager.proxyAuthenticationRequired.connect(
            proxyAuthenticationRequired)
        
        self.__availableTranslations = {}
        # dictionary of sets of available translations
        
        self.__replies = []
        
        QTimer.singleShot(0, self.__getTranslationModels)
    
    def engineName(self):
        """
        Public method to return the name of the engine.
        
        @return engine name
        @rtype str
        """
        return "ibm_watson"
    
    def supportedLanguages(self):
        """
        Public method to get the supported languages.
        
        @return list of supported language codes
        @rtype list of str
        """
        return list(self.__availableTranslations.keys())
    
    def supportedTargetLanguages(self, original):
        """
        Public method to get a list of supported target languages for an
        original language.
        
        @param original original language
        @type str
        @return list of supported target languages for the given original
        @rtype list of str
        """
        targets = self.__availableTranslations.get(original, set())
        return list(targets)
    
    def hasTTS(self):
        """
        Public method indicating the Text-to-Speech capability.
        
        @return flag indicating the Text-to-Speech capability
        @rtype bool
        """
        return False
    
    def getTranslation(self, requestObject, text, originalLanguage,
                       translationLanguage):
        """
        Public method to translate the given text.
        
        @param requestObject reference to the request object
        @type TranslatorRequest
        @param text text to be translated
        @type str
        @param originalLanguage language code of the original
        @type str
        @param translationLanguage language code of the translation
        @type str
        @return tuple of translated text and flag indicating success
        @rtype tuple of (str, bool)
        """
        apiKey = self.plugin.getPreferences("IbmKey")
        if not apiKey:
            return self.tr("IBM Watson: A valid Language Translator key is"
                           " required."), False
        translatorUrl = self.plugin.getPreferences("IbmUrl")
        if not translatorUrl:
            return self.tr("IBM Watson: A valid Language Translator URL is"
                           " required."), False
        
        params = "?version=2018-05-01"
        url = QUrl(translatorUrl + "/v3/translate" + params)
        
        requestDict = {
            "text": [text],
            "source": originalLanguage,
            "target": translationLanguage,
        }
        request = QByteArray(json.dumps(requestDict).encode("utf-8"))
        
        extraHeaders = [
            (b"Authorization",
             b"Basic " + QByteArray(
                 b"apikey:" + apiKey.encode("utf-8")).toBase64())
        ]
        
        response, ok = requestObject.post(url, request, dataType="json",
                                          extraHeaders=extraHeaders)
        if ok:
            try:
                responseDict = json.loads(response)
            except ValueError:
                return self.tr("IBM Watson: Invalid response received"), False
            
            if "translations" not in responseDict:
                return self.tr("IBM Watson: No translation available."), False
            
            result = ""
            translations = responseDict["translations"]
            for translation in translations:
                result += translation["translation"]
                if translation != translations[-1]:
                    result += "<br/>"
        else:
            result = response
        return result, ok
    
    def __adjustLanguageCode(self, code):
        """
        Private method to adjust a given language code.
        
        @param code code to be adjusted
        @type str
        @return adjusted language code
        @rtype str
        """
        if code == "zh":
            return "zh-CN"
        else:
            return code
    
    def __getTranslationModels(self):
        """
        Private method to get the translation models supported by IBM Watson
        Language Translator.
        """
        apiKey = self.plugin.getPreferences("IbmKey")
        if not apiKey:
            EricMessageBox.critical(
                self.__ui,
                self.tr("Error Getting Available Translations"),
                self.tr("IBM Watson: A valid Language Translator key is"
                        " required.")
            )
            return
        translatorUrl = self.plugin.getPreferences("IbmUrl")
        if not translatorUrl:
            EricMessageBox.critical(
                self.__ui,
                self.tr("Error Getting Available Translations"),
                self.tr("IBM Watson: A valid Language Translator URL is"
                        " required.")
            )
            return
        
        params = "?version=2018-05-01"
        url = QUrl(translatorUrl + "/v3/models" + params)
        
        extraHeaders = [
            (b"Authorization",
             b"Basic " + QByteArray(
                 b"apikey:" + apiKey.encode("utf-8")).toBase64())
        ]
        
        request = QNetworkRequest(url)
        request.setAttribute(
            QNetworkRequest.Attribute.FollowRedirectsAttribute, True)
        if extraHeaders:
            for name, value in extraHeaders:
                request.setRawHeader(name, value)
        reply = self.__networkManager.get(request)
        reply.finished.connect(
            lambda: self.__getTranslationModelsReplyFinished(reply))
        self.__replies.append(reply)
    
    def __getTranslationModelsReplyFinished(self, reply):
        """
        Private slot handling the receipt of the available translations.
        
        @param reply reference to the network reply object
        @type QNetworkReply
        """
        if reply in self.__replies:
            self.__replies.remove(reply)
            reply.deleteLater()
            
            if reply.error() != QNetworkReply.NetworkError.NoError:
                errorStr = reply.errorString()
                EricMessageBox.critical(
                    self.__ui,
                    self.tr("Error Getting Available Translations"),
                    self.tr("IBM Watson: The server sent an error indication."
                            "\n Error: {0}").format(errorStr)
                )
                return
            else:
                response = str(reply.readAll(), "utf-8", "replace")
                try:
                    responseDict = json.loads(response)
                except ValueError:
                    EricMessageBox.critical(
                        self.__ui,
                        self.tr("Error Getting Available Translations"),
                        self.tr("IBM Watson: Invalid response received")
                    )
                    return
                
                if "models" not in responseDict:
                    EricMessageBox.critical(
                        self.__ui,
                        self.tr("Error Getting Available Translations"),
                        self.tr("IBM Watson: No translation available.")
                    )
                    return
                
                for model in responseDict["models"]:
                    if model["status"] == "available":
                        source = self.__adjustLanguageCode(model["source"])
                        target = self.__adjustLanguageCode(model["target"])
                        if source not in self.__availableTranslations:
                            self.__availableTranslations[source] = set()
                        self.__availableTranslations[source].add(target)
                
                self.availableTranslationsLoaded.emit()