Plugins/UiExtensionPlugins/Translator/TranslatorEngines/IbmWatsonEngine.py

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

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

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

from __future__ import unicode_literals
try:
    str = unicode
except NameError:
    pass

import json

from PyQt5.QtCore import QUrl, QByteArray

from E5Gui import E5MessageBox

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(IbmWatsonEngine, self).__init__(plugin, parent)
        
        self.__ui = parent
        
        self.__availableTranslations = {}
        # dictionary of sets of available translations
        
        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("A valid IBM Watson Language Translator key is"
                           " required."), False
        translatorUrl = self.plugin.getPreferences("IbmUrl")
        if not translatorUrl:
            return self.tr("A valid IBM Watson 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("Invalid response received"), False
            
            if "translations" not in responseDict:
                return self.tr("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:
            E5MessageBox.critical(
                self.__ui,
                self.tr("Error Getting Available Translations"),
                self.tr("A valid IBM Watson Language Translator key is"
                        " required.")
            )
            return
        translatorUrl = self.plugin.getPreferences("IbmUrl")
        if not translatorUrl:
            E5MessageBox.critical(
                self.__ui,
                self.tr("Error Getting Available Translations"),
                self.tr("A valid IBM Watson 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())
        ]
        
        from ..TranslatorRequest import TranslatorRequest
        requestObject = TranslatorRequest(self)
        response, ok = requestObject.get(url, extraHeaders=extraHeaders)
        if ok:
            response = str(response, "utf-8", "replace")
            try:
                responseDict = json.loads(response)
            except ValueError:
                E5MessageBox.critical(
                    self.__ui,
                    self.tr("Error Getting Available Translations"),
                    self.tr("Invalid response received")
                )
                return
            
            if "models" not in responseDict:
                E5MessageBox.critical(
                    self.__ui,
                    self.tr("Error Getting Available Translations"),
                    self.tr("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)