Coverage for app/venv/lib/python3.14/site-packages/weblate/trans/autotranslate.py: 14%
152 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
5from __future__ import annotations
7from typing import TYPE_CHECKING, Literal
9from celery import current_task
10from django.conf import settings
11from django.core.exceptions import PermissionDenied
12from django.db import transaction
13from django.db.models import Value
14from django.db.models.functions import MD5, Lower
15from django.utils.translation import gettext, ngettext
17from weblate.machinery.base import (
18 MachineTranslationError,
19)
20from weblate.machinery.models import MACHINERY
21from weblate.trans.actions import ActionEvents
22from weblate.trans.models import Component, Suggestion, Unit
23from weblate.trans.util import split_plural
24from weblate.utils.state import (
25 STATE_APPROVED,
26 STATE_FUZZY,
27 STATE_READONLY,
28 STATE_TRANSLATED,
29)
31if TYPE_CHECKING: 31 ↛ 32line 31 didn't jump to line 32 because the condition on line 31 was never true
32 from weblate.auth.models import User
33 from weblate.machinery.base import (
34 BatchMachineTranslation,
35 UnitMemoryResultDict,
36 )
37 from weblate.trans.models import Translation
38 from weblate.utils.state import (
39 StringState,
40 )
43class AutoTranslate:
44 def __init__(
45 self,
46 *,
47 user: User | None,
48 translation: Translation,
49 q: str,
50 mode: str,
51 component_wide: bool = False,
52 unit_ids: list[int] | None = None,
53 ) -> None:
54 self.user: User | None = user
55 self.translation: Translation = translation
56 translation.component.batch_checks = True
58 self.unit_ids: list[int] | None = unit_ids
60 self.q: str = q
61 self.mode: str = mode
62 self.updated = 0
63 self.progress_steps = 0
64 self.progress_base = 0
65 self.target_state = STATE_TRANSLATED
66 if mode == "fuzzy":
67 self.target_state = STATE_FUZZY
68 elif mode == "approved" and translation.enable_review:
69 self.target_state = STATE_APPROVED
70 self.component_wide: bool = component_wide
72 def get_units(self):
73 units = self.translation.unit_set.exclude(state=STATE_READONLY)
74 if self.unit_ids is not None:
75 units = units.filter(pk__in=self.unit_ids)
76 if self.mode == "suggest":
77 units = units.filter(suggestion__isnull=True)
78 return units.search(self.q, parser="unit")
80 def set_progress(self, current) -> None:
81 if current_task and current_task.request.id and self.progress_steps:
82 current_task.update_state(
83 state="PROGRESS",
84 meta={
85 "progress": 100 * current // self.progress_steps,
86 "translation": self.translation.pk,
87 },
88 )
90 def update(
91 self, unit: Unit, state: StringState, target: list[str], user=None
92 ) -> None:
93 if isinstance(target, str):
94 target = [target]
95 max_length = unit.get_max_length()
96 if self.mode == "suggest" or any(len(item) > max_length for item in target):
97 suggestion = Suggestion.objects.add(
98 unit, target, request=None, vote=False, user=user, raise_exception=False
99 )
100 if suggestion:
101 self.updated += 1
102 else:
103 unit.is_batch_update = True
104 unit.translate(
105 user or self.user,
106 target,
107 state,
108 change_action=ActionEvents.AUTO,
109 propagate=False,
110 )
111 self.updated += 1
113 def post_process(self) -> None:
114 if self.updated > 0:
115 self.translation.log_info("finalizing automatic translation")
116 if not self.component_wide:
117 self.translation.component.update_source_checks()
118 self.translation.component.run_batched_checks()
119 self.translation.invalidate_cache()
120 if self.user:
121 self.user.profile.increase_count("translated", self.updated)
123 @transaction.atomic
124 def process_others(self, source: int | None) -> None:
125 """Perform automatic translation based on other components."""
126 kwargs = {
127 "translation__plural": self.translation.plural,
128 "state__gte": STATE_TRANSLATED,
129 }
130 source_language = self.translation.component.source_language
131 exclude = {}
132 if source:
133 component = Component.objects.get(id=source)
135 if (
136 not component.project.contribute_shared_tm
137 and component.project != self.translation.component.project
138 ):
139 msg = "Project has disabled contribution to shared translation memory."
140 raise PermissionDenied(msg)
141 if component.source_language != source_language:
142 msg = "Component have different source languages."
143 raise PermissionDenied(msg)
144 kwargs["translation__component"] = component
145 else:
146 project = self.translation.component.project
147 kwargs["translation__component__project"] = project
148 kwargs["translation__component__source_language"] = source_language
149 exclude["translation"] = self.translation
150 sources = Unit.objects.filter(**kwargs)
152 # Use memory_db for the query in case it exists. This is supposed
153 # to be a read-only replica for offloading expensive translation
154 # queries.
155 if "memory_db" in settings.DATABASES:
156 sources = sources.using("memory_db")
158 if exclude:
159 sources = sources.exclude(**exclude)
161 # Get source MD5s
162 source_md5s = list(
163 self.get_units()
164 .annotate(source__lower__md5=MD5(Lower("source")))
165 .values_list("source__lower__md5", flat=True)
166 )
168 # Fetch available translations
169 translations = {
170 source: split_plural(target)
171 for source, target in sources.filter(
172 source__lower__md5__in=source_md5s
173 ).values_list("source", "target")
174 }
176 # Fetch translated unit IDs
177 # Cannot use get_units() directly as SELECT FOR UPDATE cannot be used with JOIN
178 unit_ids = list(
179 self.get_units()
180 .filter(
181 source__lower__md5__in=[
182 MD5(Lower(Value(translation))) for translation in translations
183 ]
184 )
185 .values_list("id", flat=True)
186 )
187 units = (
188 Unit.objects.filter(pk__in=unit_ids)
189 .prefetch()
190 .prefetch_bulk()
191 .select_for_update()
192 )
193 self.progress_steps = len(units)
195 for pos, unit in enumerate(units):
196 # Get update
197 try:
198 target = translations[unit.source]
199 except KeyError:
200 # Happens due to case-insensitive lookup
201 continue
203 self.set_progress(pos)
205 # No save if translation is same or unit does not exist
206 if unit.state == self.target_state and unit.target == target:
207 continue
208 # Copy translation
209 self.update(unit, self.target_state, target)
211 self.post_process()
213 def fetch_mt(
214 self, engines_list: list[str], threshold: int
215 ) -> dict[int, UnitMemoryResultDict]:
216 """Get the translations."""
217 units: list[Unit] = list(self.get_units().select_related("source_unit"))
218 num_units = len(units)
220 machinery_settings = self.translation.component.project.get_machinery_settings()
222 engines: list[BatchMachineTranslation] = sorted(
223 (
224 MACHINERY[engine](setting)
225 for engine, setting in machinery_settings.items()
226 if engine in MACHINERY and engine in engines_list
227 ),
228 key=lambda engine: engine.get_rank(),
229 reverse=True,
230 )
232 self.progress_base = len(engines) * num_units
233 # Estimate number of strings to translate, this is adjusted in process_mt
234 self.progress_steps = self.progress_base + num_units
236 for pos, translation_service in enumerate(engines):
237 batch_size = translation_service.batch_size
238 self.translation.log_info(
239 "fetching translations for %d units from %s, %d per request",
240 num_units,
241 translation_service.name,
242 batch_size,
243 )
245 for batch_start in range(0, num_units, batch_size):
246 self.set_progress(pos * num_units + batch_start)
247 try:
248 translation_service.batch_translate(
249 units[batch_start : batch_start + batch_size],
250 self.user,
251 threshold=threshold,
252 )
253 except MachineTranslationError as error:
254 # Ignore errors here to complete fetching
255 self.translation.log_error(
256 "failed automatic translation: %s", error
257 )
259 self.set_progress(self.progress_base)
260 return {
261 unit.id: unit.machinery
262 for unit in units
263 if unit.machinery and any(unit.machinery["quality"])
264 }
266 def process_mt(self, engines: list[str], threshold: int) -> None:
267 """Perform automatic translation based on machine translation."""
268 translations = self.fetch_mt(engines, int(threshold))
270 # Adjust total number to show correct progress
271 self.progress_steps = self.progress_base + len(translations)
273 with transaction.atomic():
274 # Perform the translation
275 self.translation.log_info("updating %d strings", len(translations))
276 for pos, unit in enumerate(
277 self.translation.unit_set.filter(id__in=translations.keys())
278 .prefetch_bulk()
279 .select_for_update()
280 ):
281 translation: UnitMemoryResultDict = translations[unit.pk]
282 # Use first existing origin for user
283 # (there can be blanks for missing plurals)
284 user: User | None = None
285 for origin in translation["origin"]:
286 if origin is not None:
287 user = origin.user
288 break
289 # Copy translation
290 self.update(
291 unit,
292 self.target_state,
293 translation["translation"],
294 user=user,
295 )
296 self.set_progress(self.progress_base + pos + 1)
298 self.post_process()
300 def perform(
301 self,
302 *,
303 auto_source: Literal["mt", "others"],
304 engines: list[str],
305 threshold: int,
306 source: int | None,
307 ) -> str:
308 translation = self.translation
309 translation.log_info(
310 "starting automatic translation (%s) %s: %s: %s",
311 self.mode,
312 current_task.request.id if current_task and current_task.request.id else "",
313 auto_source,
314 ", ".join(engines) if engines else source,
315 )
316 try:
317 if auto_source == "mt":
318 self.process_mt(engines, threshold)
319 else:
320 self.process_others(source)
321 except (MachineTranslationError, Component.DoesNotExist) as error:
322 translation.log_error("failed automatic translation: %s", error)
323 return gettext("Automatic translation failed: %s") % error
325 translation.log_info("completed automatic translation")
327 if self.updated == 0:
328 return gettext("Automatic translation completed, no strings were updated.")
329 return (
330 ngettext(
331 "Automatic translation completed, %d string was updated.",
332 "Automatic translation completed, %d strings were updated.",
333 self.updated,
334 )
335 % self.updated
336 )