Coverage for app/venv/lib/python3.14/site-packages/weblate/machinery/modernmt.py: 29%
88 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 07:15 +0000
1# Copyright © Michal Čihař <michal@weblate.org>
2#
3# SPDX-License-Identifier: GPL-3.0-or-later
4from __future__ import annotations
6import csv
7import json
8import os
9import tempfile
10from typing import TYPE_CHECKING, ClassVar
12from dateutil.parser import isoparse
13from requests.exceptions import HTTPError
15import weblate.utils.version
17from .base import (
18 GlossaryDoesNotExistError,
19 GlossaryMachineTranslationMixin,
20 MachineTranslationError,
21)
22from .forms import ModernMTMachineryForm
24if TYPE_CHECKING: 24 ↛ 25line 24 didn't jump to line 25 because the condition on line 24 was never true
25 from .base import (
26 DownloadTranslations,
27 )
30class ModernMTTranslation(GlossaryMachineTranslationMixin):
31 """ModernMT machine translation support."""
33 name = "ModernMT"
34 max_score = 90
35 settings_form = ModernMTMachineryForm
37 language_map: ClassVar[dict[str, str]] = {
38 "fa": "pes",
39 "pt": "pt-PT",
40 "sr": "sr-Cyrl",
41 "zh_Hant": "zh-TW",
42 "zh_Hans": "zh-CN",
43 }
44 glossary_count_limit = 1000
46 def map_language_code(self, code):
47 """Convert language to service specific code."""
48 return super().map_language_code(code).replace("_", "-").split("@")[0]
50 def get_headers(self) -> dict[str, str]:
51 """Add authentication headers to request."""
52 return {
53 "MMT-ApiKey": self.settings["key"],
54 "MMT-Platform": "Weblate",
55 "MMT-PlatformVersion": weblate.utils.version.VERSION,
56 }
58 def is_supported(self, source_language, target_language):
59 """Check whether given language combination is supported."""
60 return (source_language, target_language) in self.supported_languages
62 def check_failure(self, response) -> None:
63 super().check_failure(response)
64 payload = response.json()
66 if "error" in payload:
67 raise MachineTranslationError(payload["error"]["message"])
69 def download_languages(self):
70 """List of supported languages."""
71 response = self.request("get", self.get_api_url("languages"))
72 payload = response.json()
74 for source_language, target_languages in payload["data"].items():
75 yield from (
76 (source_language, target_language)
77 for target_language in target_languages
78 )
80 def download_translations(
81 self,
82 source_language,
83 target_language,
84 text: str,
85 unit,
86 user,
87 threshold: int = 75,
88 ) -> DownloadTranslations:
89 """Download list of possible translations from a service."""
90 params = {"q": text, "source": source_language, "target": target_language}
91 glossary_id: str | None = self.get_glossary_id(
92 source_language, target_language, unit
93 )
95 if glossary_id:
96 params["glossaries"] = glossary_id
98 if context_vector := self.settings.get("context_vector"):
99 params["context_vector"] = context_vector
101 response = self.request(
102 "get",
103 self.get_api_url("translate"),
104 params=params,
105 )
106 payload = response.json()
108 yield {
109 "text": payload["data"]["translation"],
110 "quality": self.max_score,
111 "service": self.name,
112 "source": text,
113 }
115 def get_error_message(self, exc):
116 if hasattr(exc, "read"):
117 content = exc.read()
118 try:
119 data = json.loads(content)
120 except json.JSONDecodeError:
121 data = {}
123 try:
124 return data["error"]["message"]
125 except KeyError:
126 pass
128 return super().get_error_message(exc)
130 def is_glossary_supported(self, source_language: str, target_language: str) -> bool:
131 """Check whether given languages pair is supported by service glossaries."""
132 return self.is_supported(source_language, target_language)
134 def list_glossaries(self) -> dict[str, str]:
135 """List all glossaries from service."""
136 response = self.request("get", self.get_api_url("memories"))
137 return {
138 glossary["name"]: glossary["id"]
139 for glossary in response.json()["data"]
140 if self.match_name_format(glossary["name"])
141 }
143 def delete_glossary(self, glossary_id: str) -> None:
144 """Delete single glossary."""
145 try:
146 self.request("delete", self.get_api_url("memories", str(glossary_id)))
147 except HTTPError as error:
148 if error.response.status_code == 404:
149 raise GlossaryDoesNotExistError from error
151 def delete_oldest_glossary(self) -> None:
152 """Delete oldest glossary if any."""
153 response = self.request("get", self.get_api_url("memories"))
154 glossaries: list[dict] = sorted(
155 [
156 glossary
157 for glossary in response.json()["data"]
158 if self.match_name_format(glossary["name"])
159 ],
160 key=lambda glossary: isoparse(glossary["creationDate"]),
161 )
162 if glossaries:
163 self.delete_glossary(glossaries[0]["id"])
165 def create_glossary(
166 self, source_language: str, target_language: str, name: str, tsv: str
167 ) -> None:
168 """
169 Create glossary in service.
171 Create a memory with the given name and the populate with tsv content.
172 """
173 # ModernMT gracefully handles glossaries with duplicate name by updating the existing one
174 response = self.request(
175 "post",
176 self.get_api_url("memories"),
177 data={"name": name},
178 )
179 glossary_id: int = response.json()["data"]["id"]
181 temp_filename = ""
182 with tempfile.NamedTemporaryFile(
183 suffix=".csv", mode="w", encoding="utf-8", delete=False
184 ) as file_content:
185 reader = csv.reader(tsv.splitlines(), delimiter="\t")
186 writer = csv.writer(file_content)
187 writer.writerow([source_language, target_language]) # mandatory header
188 writer.writerows(reader)
189 temp_filename = file_content.name
191 try:
192 with open(temp_filename, "rb") as file_content:
193 response = self.request(
194 "post",
195 self.get_api_url("memories", str(glossary_id), "glossary"),
196 data={
197 "type": "unidirectional",
198 },
199 files={
200 "csv": file_content,
201 },
202 )
203 finally:
204 if os.path.exists(temp_filename):
205 os.unlink(temp_filename)