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

changeset 6942
2602857055c5
parent 6645
ad476851d7e0
child 7192
a22eee00b052
equal deleted inserted replaced
6941:f99d60d6b59b 6942:2602857055c5
1 # -*- coding: utf-8 -*-
2
3 # Copyright (c) 2018 - 2019 Detlev Offenbach <detlev@die-offenbachs.de>
4 #
5
6 """
7 Module implementing the IBM Watson translation engine.
8 """
9
10 from __future__ import unicode_literals
11 try:
12 str = unicode
13 except NameError:
14 pass
15
16 import json
17
18 from PyQt5.QtCore import QUrl, QByteArray, QTimer
19 from PyQt5.QtNetwork import QNetworkAccessManager, QNetworkRequest, \
20 QNetworkReply
21
22 from E5Gui import E5MessageBox
23
24 from E5Network.E5NetworkProxyFactory import proxyAuthenticationRequired
25
26 from Globals import qVersionTuple
27
28 from .TranslationEngine import TranslationEngine
29
30
31 class IbmWatsonEngine(TranslationEngine):
32 """
33 Class implementing the translation engine for the IBM Watson Language
34 Translator service.
35 """
36 # Documentation:
37 # https://www.ibm.com/watson/developercloud/language-translator
38 #
39 # Start page:
40 # https://www.ibm.com/watson/services/language-translator/
41
42 def __init__(self, plugin, parent=None):
43 """
44 Constructor
45
46 @param plugin reference to the plugin object
47 @type TranslatorPlugin
48 @param parent reference to the parent object
49 @type QObject
50 """
51 super(IbmWatsonEngine, self).__init__(plugin, parent)
52
53 self.__ui = parent
54
55 self.__networkManager = QNetworkAccessManager(self)
56 self.__networkManager.proxyAuthenticationRequired.connect(
57 proxyAuthenticationRequired)
58
59 self.__availableTranslations = {}
60 # dictionary of sets of available translations
61
62 self.__replies = []
63
64 QTimer.singleShot(0, self.__getTranslationModels)
65
66 def engineName(self):
67 """
68 Public method to return the name of the engine.
69
70 @return engine name
71 @rtype str
72 """
73 return "ibm_watson"
74
75 def supportedLanguages(self):
76 """
77 Public method to get the supported languages.
78
79 @return list of supported language codes
80 @rtype list of str
81 """
82 return list(self.__availableTranslations.keys())
83
84 def supportedTargetLanguages(self, original):
85 """
86 Public method to get a list of supported target languages for an
87 original language.
88
89 @param original original language
90 @type str
91 @return list of supported target languages for the given original
92 @rtype list of str
93 """
94 targets = self.__availableTranslations.get(original, set())
95 return list(targets)
96
97 def hasTTS(self):
98 """
99 Public method indicating the Text-to-Speech capability.
100
101 @return flag indicating the Text-to-Speech capability
102 @rtype bool
103 """
104 return False
105
106 def getTranslation(self, requestObject, text, originalLanguage,
107 translationLanguage):
108 """
109 Public method to translate the given text.
110
111 @param requestObject reference to the request object
112 @type TranslatorRequest
113 @param text text to be translated
114 @type str
115 @param originalLanguage language code of the original
116 @type str
117 @param translationLanguage language code of the translation
118 @type str
119 @return tuple of translated text and flag indicating success
120 @rtype tuple of (str, bool)
121 """
122 apiKey = self.plugin.getPreferences("IbmKey")
123 if not apiKey:
124 return self.tr("A valid IBM Watson Language Translator key is"
125 " required."), False
126 translatorUrl = self.plugin.getPreferences("IbmUrl")
127 if not translatorUrl:
128 return self.tr("A valid IBM Watson Language Translator URL is"
129 " required."), False
130
131 params = "?version=2018-05-01"
132 url = QUrl(translatorUrl + "/v3/translate" + params)
133
134 requestDict = {
135 "text": [text],
136 "source": originalLanguage,
137 "target": translationLanguage,
138 }
139 request = QByteArray(json.dumps(requestDict).encode("utf-8"))
140
141 extraHeaders = [
142 (b"Authorization",
143 b"Basic " + QByteArray(
144 b"apikey:" + apiKey.encode("utf-8")).toBase64())
145 ]
146
147 response, ok = requestObject.post(url, request, dataType="json",
148 extraHeaders=extraHeaders)
149 if ok:
150 try:
151 responseDict = json.loads(response)
152 except ValueError:
153 return self.tr("Invalid response received"), False
154
155 if "translations" not in responseDict:
156 return self.tr("No translation available."), False
157
158 result = ""
159 translations = responseDict["translations"]
160 for translation in translations:
161 result += translation["translation"]
162 if translation != translations[-1]:
163 result += "<br/>"
164 else:
165 result = response
166 return result, ok
167
168 def __adjustLanguageCode(self, code):
169 """
170 Private method to adjust a given language code.
171
172 @param code code to be adjusted
173 @type str
174 @return adjusted language code
175 @rtype str
176 """
177 if code == "zh":
178 return "zh-CN"
179 else:
180 return code
181
182 def __getTranslationModels(self):
183 """
184 Private method to get the translation models supported by IBM Watson
185 Language Translator.
186 """
187 apiKey = self.plugin.getPreferences("IbmKey")
188 if not apiKey:
189 E5MessageBox.critical(
190 self.__ui,
191 self.tr("Error Getting Available Translations"),
192 self.tr("A valid IBM Watson Language Translator key is"
193 " required.")
194 )
195 return
196 translatorUrl = self.plugin.getPreferences("IbmUrl")
197 if not translatorUrl:
198 E5MessageBox.critical(
199 self.__ui,
200 self.tr("Error Getting Available Translations"),
201 self.tr("A valid IBM Watson Language Translator URL is"
202 " required.")
203 )
204 return
205
206 params = "?version=2018-05-01"
207 url = QUrl(translatorUrl + "/v3/models" + params)
208
209 extraHeaders = [
210 (b"Authorization",
211 b"Basic " + QByteArray(
212 b"apikey:" + apiKey.encode("utf-8")).toBase64())
213 ]
214
215 request = QNetworkRequest(url)
216 if qVersionTuple() >= (5, 6, 0):
217 request.setAttribute(QNetworkRequest.FollowRedirectsAttribute,
218 True)
219 if extraHeaders:
220 for name, value in extraHeaders:
221 request.setRawHeader(name, value)
222 reply = self.__networkManager.get(request)
223 reply.finished.connect(
224 lambda: self.__getTranslationModelsReplyFinished(reply))
225 self.__replies.append(reply)
226
227 def __getTranslationModelsReplyFinished(self, reply):
228 """
229 Private slot handling the receipt of the available translations.
230
231 @param reply reference to the network reply object
232 @type QNetworkReply
233 """
234 if reply in self.__replies:
235 self.__replies.remove(reply)
236 reply.deleteLater()
237
238 if reply.error() != QNetworkReply.NoError:
239 errorStr = reply.errorString()
240 E5MessageBox.critical(
241 self.__ui,
242 self.tr("Error Getting Available Translations"),
243 self.tr("The server sent an error indication.\n"
244 "Error: {0}").format(errorStr)
245 )
246 return
247 else:
248 response = str(reply.readAll(), "utf-8", "replace")
249 try:
250 responseDict = json.loads(response)
251 except ValueError:
252 E5MessageBox.critical(
253 self.__ui,
254 self.tr("Error Getting Available Translations"),
255 self.tr("Invalid response received")
256 )
257 return
258
259 if "models" not in responseDict:
260 E5MessageBox.critical(
261 self.__ui,
262 self.tr("Error Getting Available Translations"),
263 self.tr("No translation available.")
264 )
265 return
266
267 for model in responseDict["models"]:
268 if model["status"] == "available":
269 source = self.__adjustLanguageCode(model["source"])
270 target = self.__adjustLanguageCode(model["target"])
271 if source not in self.__availableTranslations:
272 self.__availableTranslations[source] = set()
273 self.__availableTranslations[source].add(target)
274
275 self.availableTranslationsLoaded.emit()

eric ide

mercurial