Coverage for app/venv/lib/python3.14/site-packages/weblate/machinery/googlev3.py: 39%

1# Copyright © Michal Čihař <michal@weblate.org> 

2# 

3# SPDX-License-Identifier: GPL-3.0-or-later 

4 

5from __future__ import annotations 

6 

7import json 

8import operator 

9from typing import TYPE_CHECKING 

10 

11from django.utils.functional import cached_property 

12from google.api_core.exceptions import AlreadyExists, NotFound 

13from google.cloud import storage # type: ignore[attr-defined] 

14from google.cloud.translate_v3 import ( 

15 GcsSource, 

16 Glossary, 

17 GlossaryInputConfig, 

18 TranslateTextGlossaryConfig, 

19 TranslationServiceClient, 

20) 

21from google.oauth2 import service_account 

22 

23from .base import ( 

24 GlossaryAlreadyExistsError, 

25 GlossaryDoesNotExistError, 

26 GlossaryMachineTranslationMixin, 

27 XMLMachineTranslationMixin, 

28) 

29from .forms import GoogleV3MachineryForm 

30from .google import GoogleBaseTranslation 

31 

32if TYPE_CHECKING: 32 ↛ 33line 32 didn't jump to line 33 because the condition on line 32 was never true

33 from weblate.trans.models import Unit 

34 

35 from .base import ( 

36 DownloadTranslations, 

37 ) 

38 

39 

40class GoogleV3Translation( 

41 XMLMachineTranslationMixin, GoogleBaseTranslation, GlossaryMachineTranslationMixin 

42): 

43 """Google Translate API v3 machine translation support.""" 

44 

45 name = "Google Cloud Translation Advanced" 

46 max_score = 90 

47 settings_form = GoogleV3MachineryForm 

48 

49 # estimation, actual limit is 10.4 million (10,485,760) UTF-8 bytes 

50 glossary_count_limit = 1000 

51 

52 # Identifier must contain only lowercase letters, digits, or hyphens. 

53 glossary_name_format = ( 

54 "weblate__{project}__{source_language}__{target_language}__{checksum}" 

55 ) 

56 

57 @classmethod 

58 def get_identifier(cls) -> str: 

59 return "google-translate-api-v3" 

60 

61 @cached_property 

62 def client(self): 

63 credentials = service_account.Credentials.from_service_account_info( 

64 json.loads(self.settings["credentials"]) 

65 ) 

66 api_endpoint = "translate.googleapis.com" 

67 if self.settings["location"].startswith("europe-"): 

68 api_endpoint = "translate-eu.googleapis.com" 

69 elif self.settings["location"].startswith("us-"): 

70 api_endpoint = "translate-us.googleapis.com" 

71 return TranslationServiceClient( 

72 credentials=credentials, client_options={"api_endpoint": api_endpoint} 

73 ) 

74 

75 @cached_property 

76 def storage_client(self): 

77 credentials = service_account.Credentials.from_service_account_info( 

78 json.loads(self.settings["credentials"]) 

79 ) 

80 return storage.Client(credentials=credentials) 

81 

82 @cached_property 

83 def storage_bucket(self): 

84 return self.storage_client.get_bucket(self.settings["bucket_name"]) 

85 

86 @cached_property 

87 def parent(self) -> str: 

88 project = self.settings["project"] 

89 location = self.settings["location"] 

90 return f"projects/{project}/locations/{location}" 

91 

92 def download_languages(self): 

93 """List of supported languages.""" 

94 response = self.client.get_supported_languages(request={"parent": self.parent}) 

95 return [language.language_code for language in response.languages] 

96 

97 def download_translations( 

98 self, 

99 source_language, 

100 target_language, 

101 text: str, 

102 unit, 

103 user, 

104 threshold: int = 75, 

105 ) -> DownloadTranslations: 

106 """Download list of possible translations from a service.""" 

107 request = { 

108 "parent": self.parent, 

109 "contents": [text], 

110 "target_language_code": target_language, 

111 "source_language_code": source_language, 

112 "mime_type": "text/html", 

113 } 

114 glossary_path: str | None = None 

115 if self.settings.get("bucket_name"): 

116 glossary_path = self.get_glossary_id(source_language, target_language, unit) 

117 request["glossary_config"] = TranslateTextGlossaryConfig( 

118 glossary=glossary_path 

119 ) 

120 

121 response = self.client.translate_text(request) 

122 

123 response_translations = ( 

124 response.glossary_translations if glossary_path else response.translations 

125 ) 

126 

127 yield { 

128 "text": response_translations[0].translated_text, 

129 "quality": self.max_score, 

130 "service": self.name, 

131 "source": text, 

132 } 

133 

134 def format_replacement( 

135 self, h_start: int, h_end: int, h_text: str, h_kind: Unit | None 

136 ) -> str: 

137 """Generate a single replacement.""" 

138 return f'<span translate="no" id="{h_start}">{self.escape_text(h_text)}</span>' 

139 

140 def cleanup_text(self, text, unit): 

141 text, replacements = super().cleanup_text(text, unit) 

142 

143 # Sanitize newlines 

144 replacement = '<br translate="no">' 

145 replacements[replacement] = "\n" 

146 

147 return text.replace("\n", replacement), replacements 

148 

149 def list_glossaries(self) -> dict[str, str]: 

150 """Return dictionary with the name/id of the glossary as the key and value.""" 

151 return { 

152 glossary.display_name: glossary.display_name 

153 for glossary in self.client.list_glossaries(parent=self.parent) 

154 } 

155 

156 def create_glossary( 

157 self, source_language: str, target_language: str, name: str, tsv: str 

158 ) -> None: 

159 """ 

160 Create glossary in the service. 

161 

162 - Uploads the TSV file to gcs bucket 

163 - Creates the glossary in the service 

164 """ 

165 # upload tsv to storage bucket 

166 glossary_bucket_file = self.storage_bucket.blob(f"{name}.tsv") 

167 glossary_bucket_file.upload_from_string( 

168 tsv, content_type="text/tab-separated-values" 

169 ) 

170 # create glossary 

171 bucket_name = self.settings["bucket_name"] 

172 gcs_source = GcsSource(input_uri=f"gs://{bucket_name}/{name}.tsv") 

173 input_config = GlossaryInputConfig(gcs_source=gcs_source) 

174 

175 glossary = Glossary( 

176 name=self.get_glossary_resource_path(name), 

177 language_pair=Glossary.LanguageCodePair( 

178 source_language_code=source_language, 

179 target_language_code=target_language, 

180 ), 

181 input_config=input_config, 

182 ) 

183 try: 

184 self.client.create_glossary(parent=self.parent, glossary=glossary) 

185 except AlreadyExists as error: 

186 raise GlossaryAlreadyExistsError from error 

187 

188 def delete_glossary(self, glossary_name: str) -> None: 

189 """Delete the glossary in service and storage bucket.""" 

190 try: 

191 self.client.delete_glossary( 

192 name=self.get_glossary_resource_path(glossary_name) 

193 ) 

194 except NotFound as error: 

195 raise GlossaryDoesNotExistError from error 

196 finally: 

197 try: 

198 # delete tsv from storage bucket 

199 glossary_bucket_file = self.storage_bucket.blob(f"{glossary_name}.tsv") 

200 glossary_bucket_file.delete() 

201 except NotFound: 

202 pass 

203 

204 def delete_oldest_glossary(self) -> None: 

205 """Delete the oldest glossary if any.""" 

206 glossaries = sorted( 

207 self.client.list_glossaries(parent=self.parent), 

208 key=operator.attrgetter("submit_time"), 

209 ) 

210 if glossaries: 

211 self.delete_glossary(glossaries[0].display_name) 

212 

213 def get_glossary_resource_path(self, glossary_name: str): 

214 """Return the resource path used by the Translation API.""" 

215 return self.client.glossary_path( 

216 self.settings["project"], self.settings["location"], glossary_name 

217 )