Coverage for app/venv/lib/python3.14/site-packages/weblate/memory/models.py: 25%
221 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
7import json
8import math
9import os
10import re
11from typing import TYPE_CHECKING, BinaryIO
13from django.conf import settings
14from django.db import models
15from django.db.models import Q, Value
16from django.db.models.functions import MD5
17from django.utils.encoding import force_str
18from django.utils.translation import gettext, gettext_lazy, pgettext
19from translate.misc.xml_helpers import getXMLlang, getXMLspace
20from translate.storage.tmx import tmxfile
21from weblate_schemas import load_schema
23from weblate.lang.models import Language
24from weblate.memory.utils import (
25 CATEGORY_FILE,
26 CATEGORY_PRIVATE_OFFSET,
27 CATEGORY_SHARED,
28 CATEGORY_USER_OFFSET,
29 is_valid_memory_entry,
30)
31from weblate.utils.db import adjust_similarity_threshold, using_postgresql
32from weblate.utils.errors import report_error
34if TYPE_CHECKING: 34 ↛ 35line 34 didn't jump to line 35 because the condition on line 34 was never true
35 from weblate.auth.models import AuthenticatedHttpRequest, User
36 from weblate.trans.models import Project
38NON_WORD_RE = re.compile(r"\W")
40SUPPORTED_FORMATS = (
41 "json",
42 "tmx",
43 "xliff",
44 "po",
45 "csv",
46)
49class MemoryImportError(Exception):
50 pass
53def get_node_data(unit, node):
54 """
55 Return XML unit text.
57 Generic implementation of LISAUnit.gettarget.
58 """
59 # The language should be present as xml:lang, but in some
60 # cases it's there only as lang
61 return (
62 getXMLlang(node) or node.get("lang"),
63 unit.getNodeText(node, getXMLspace(unit.xmlelement, "preserve")),
64 )
67class MemoryQuerySet(models.QuerySet):
68 def filter_type(
69 self,
70 *,
71 user: User | None = None,
72 project: Project | None = None,
73 use_shared: bool = False,
74 from_file: bool = False,
75 ):
76 base = self
77 if "memory_db" in settings.DATABASES:
78 base = base.using("memory_db")
79 query = Q()
80 if from_file:
81 query |= Q(from_file=from_file)
82 if use_shared:
83 query |= Q(shared=use_shared)
84 if project:
85 query |= Q(project=project)
86 if user:
87 query |= Q(user=user)
88 return base.filter(query)
90 def filter(self, *args, **kwargs):
91 if using_postgresql(): 91 ↛ 103line 91 didn't jump to line 103 because the condition on line 91 was always true
92 # Use MD5 for filtering to utilize MD5 index,
93 # MariaDB does not support that, but has partial
94 # index on text fields created manually
95 for field in ("source", "target", "origin"):
96 if field in kwargs: 96 ↛ 97line 96 didn't jump to line 97 because the condition on line 96 was never true
97 kwargs[f"{field}__md5"] = MD5(Value(kwargs.pop(field)))
98 in_field = f"{field}__in"
99 if in_field in kwargs: 99 ↛ 100line 99 didn't jump to line 100 because the condition on line 99 was never true
100 kwargs[f"{field}__md5__in"] = [
101 MD5(Value(value)) for value in kwargs.pop(in_field)
102 ]
103 return super().filter(*args, **kwargs)
105 def threshold_to_similarity(self, text: str, threshold: int) -> float:
106 """
107 Convert machinery threshold into PostgreSQL similarity threshold.
109 Machinery threshold typical values:
111 - 75 machinery
112 - 80 automatic translation (default value)
113 - 10 search
115 PostgreSQL similarity threshold needs to be higher to avoid too slow
116 queries.
118 We exclude non-word characters while calculating this as those are
119 excluded in the trigram matching.
120 """
121 # Highest similarity we want to get
122 high = 0.985
123 # Limit the number of decimals to avoid too frequent flipping of the setting
124 # inside PostgreSQL
125 decimals = 3
127 # Maps threshold to a minimal score, approximately:
128 # 10 => 0.7
129 # 75 => 0.95
130 # 80 => 0.96
131 # 100 => 1.0
132 base = 0.127264 * math.log(24.282 * threshold)
133 if base >= high:
134 return min(round(base, decimals), 1.0)
136 # Allow up to +20% boost based on length
137 maximum = min(base * 1.2, high)
139 # Measure the length of alphanumeric characters in the text
140 max_length = 2000
141 length = min(max(1, len(NON_WORD_RE.sub("", text))), max_length)
143 # Apply boost based on square root of length so that it grows faster
144 # for shorter strings
145 boost = (maximum - base) * math.sqrt(length) / math.sqrt(max_length)
147 # Cap result into reasonable limits
148 return max(0.6, min(1.0, round(base + boost, decimals)))
150 def lookup(
151 self,
152 source_language,
153 target_language,
154 text: str,
155 user,
156 project,
157 use_shared,
158 threshold: int = 75,
159 ):
160 # Adjust similarity based on string length to get more relevant matches
161 # for long strings
162 adjust_similarity_threshold(self.threshold_to_similarity(text, threshold))
164 # Actual database query
165 return (
166 self.prefetch_project()
167 .filter_type(
168 # Type filtering
169 user=user,
170 project=project,
171 use_shared=use_shared,
172 from_file=True,
173 )
174 .filter(
175 # Full-text search on source
176 source__search=text,
177 # Language filtering
178 source_language=source_language,
179 target_language=target_language,
180 )[:50]
181 )
183 def prefetch_lang(self):
184 return self.prefetch_related("source_language", "target_language")
186 def prefetch_project(self):
187 return self.select_related("project")
190class MemoryManager(models.Manager):
191 def import_file(
192 self,
193 request: AuthenticatedHttpRequest | None,
194 fileobj: BinaryIO,
195 langmap: dict[str, str] | None = None,
196 source_language: Language | str | None = None,
197 target_language: Language | str | None = None,
198 **kwargs,
199 ):
200 kwargs.update(
201 {
202 "from_file": True,
203 "status": Memory.STATUS_ACTIVE,
204 }
205 )
206 origin = os.path.basename(fileobj.name).lower()
207 name, extension = os.path.splitext(origin)
209 if extension.lower().strip(".") not in SUPPORTED_FORMATS:
210 raise MemoryImportError(
211 gettext("Unsupported file extension: %s") % extension
212 )
214 if len(name) > 25:
215 origin = f"{name[:25]}...{extension}"
217 if extension == ".tmx":
218 result = self.import_tmx(request, fileobj, origin, langmap, **kwargs)
219 elif extension == ".json":
220 result = self.import_json(request, fileobj, origin, **kwargs)
221 else:
222 result = self.import_other_format(
223 request,
224 fileobj,
225 origin,
226 source_language,
227 target_language,
228 **kwargs,
229 )
231 if not result:
232 raise MemoryImportError(
233 gettext("No valid entries found in the uploaded file!")
234 )
235 return result
237 def import_json(
238 self,
239 request: AuthenticatedHttpRequest | None,
240 fileobj: BinaryIO,
241 origin: str | None = None,
242 **kwargs,
243 ) -> int:
244 # Lazily import as this is expensive
245 from jsonschema import validate
246 from jsonschema.exceptions import ValidationError
248 content = fileobj.read()
249 try:
250 data = json.loads(force_str(content))
251 except json.JSONDecodeError as error:
252 report_error("Could not parse memory")
253 raise MemoryImportError(
254 gettext("Could not parse JSON file: %s") % error
255 ) from error
256 try:
257 validate(data, load_schema("weblate-memory.schema.json"))
258 except ValidationError as error:
259 report_error("Could not validate memory")
260 raise MemoryImportError(
261 gettext("Could not parse JSON file: %s") % error
262 ) from error
263 found = 0
264 lang_cache: dict[str, Language] = {}
265 for entry in data:
266 try:
267 self.update_entry(
268 source_language=Language.objects.get_by_code(
269 entry["source_language"], lang_cache
270 ),
271 target_language=Language.objects.get_by_code(
272 entry["target_language"], lang_cache
273 ),
274 source=entry["source"],
275 target=entry["target"],
276 origin=origin,
277 context=entry.get("context", ""),
278 **kwargs,
279 )
280 found += 1
281 except Language.DoesNotExist:
282 continue
283 return found
285 def import_tmx(
286 self,
287 request: AuthenticatedHttpRequest | None,
288 fileobj: BinaryIO,
289 origin: str | None = None,
290 langmap: dict[str, str] | None = None,
291 **kwargs,
292 ) -> int:
293 try:
294 storage = tmxfile.parsefile(fileobj)
295 except (SyntaxError, AssertionError) as error:
296 report_error("Could not parse")
297 raise MemoryImportError(
298 gettext("Could not parse TMX file: %s") % error
299 ) from error
300 header = next(
301 storage.document.getroot().iterchildren(storage.namespaced("header"))
302 )
303 lang_cache: dict[str, Language] = {}
304 srclang = header.get("srclang")
305 if not srclang:
306 raise MemoryImportError(
307 gettext("Source language not defined in the TMX file!")
308 )
309 try:
310 source_language = Language.objects.get_by_code(srclang, lang_cache, langmap)
311 except Language.DoesNotExist as error:
312 raise MemoryImportError(
313 gettext("Could not find language %s!") % srclang
314 ) from error
316 found = 0
317 for unit in storage.units:
318 # Parse translations (translate-toolkit does not care about
319 # languages here, it just picks first and second XML elements)
320 translations = {}
321 for node in unit.getlanguageNodes():
322 lang_code, text = get_node_data(unit, node)
323 if not lang_code or not text:
324 continue
325 try:
326 language = Language.objects.get_by_code(
327 lang_code, lang_cache, langmap
328 )
329 except Language.DoesNotExist as error:
330 raise MemoryImportError(
331 gettext("Could not find language %s!") % header.get("srclang")
332 ) from error
333 translations[language.code] = text
335 try:
336 source = translations.pop(source_language.code)
337 except KeyError:
338 # Skip if source language is not present
339 continue
341 for lang, text in translations.items():
342 self.update_entry(
343 source_language=source_language,
344 target_language=Language.objects.get_by_code(
345 lang, lang_cache, langmap
346 ),
347 source=source,
348 target=text,
349 origin=origin,
350 **kwargs,
351 )
352 found += 1
353 return found
355 def import_other_format(
356 self,
357 request: AuthenticatedHttpRequest | None,
358 fileobj: BinaryIO,
359 origin: str,
360 source_language: Language | str | None = None,
361 target_language: Language | str | None = None,
362 **kwargs,
363 ) -> int:
364 """
365 Import memory from other formats.
367 This is a generic function to import memories from other formats.
368 It currently supports all formats supported by `try_load` from
369 `weblate.formats.auto`.
371 """
372 from weblate.formats.auto import try_load
374 lang_cache: dict[str, Language] = {}
375 try:
376 storage = try_load(origin, fileobj.read(), None, None)
377 except Exception as error:
378 report_error("Could not parse memory")
379 raise MemoryImportError(gettext("Unsupported file!")) from error
381 if storage.monolingual is True:
382 raise MemoryImportError(
383 gettext("Monolingual format not supported for memory upload")
384 )
386 def get_language(language: Language | str | None) -> Language:
387 """Get a language object based on the given code."""
388 if isinstance(language, Language):
389 return language
391 if not language:
392 raise MemoryImportError(
393 gettext("Missing source or target language in file!")
394 )
395 try:
396 return Language.objects.get_by_code(language, lang_cache)
397 except Language.DoesNotExist as error:
398 raise MemoryImportError(
399 gettext("Could not find language %s!") % language
400 ) from error
402 source_language = get_language(storage.source_language or source_language)
403 target_language = get_language(storage.language_code or target_language)
405 count = 0
406 for _unused, unit in storage.iterate_merge("", only_translated=True):
407 self.update_entry(
408 source_language=source_language,
409 target_language=target_language,
410 source=unit.source,
411 target=unit.target,
412 origin=origin,
413 context=unit.context,
414 **kwargs,
415 )
416 count += 1
417 return count
419 def update_entry(self, **kwargs) -> None:
420 if not is_valid_memory_entry(**kwargs): # pylint: disable=missing-kwoa
421 return
422 if not self.filter(**kwargs).exists():
423 self.create(**kwargs)
426class Memory(models.Model):
427 # Status choices for the memory entry
428 STATUS_PENDING = 0
429 STATUS_ACTIVE = 1
430 STATUS_CHOICES = (
431 (STATUS_PENDING, gettext_lazy("Pending")),
432 (STATUS_ACTIVE, gettext_lazy("Active")),
433 )
435 source_language = models.ForeignKey(
436 "lang.Language",
437 on_delete=models.deletion.CASCADE,
438 related_name="memory_source_set",
439 )
440 target_language = models.ForeignKey(
441 "lang.Language",
442 on_delete=models.deletion.CASCADE,
443 related_name="memory_target_set",
444 )
445 source = models.TextField()
446 target = models.TextField()
447 origin = models.TextField()
448 context = models.TextField(default="", blank=True)
449 user = models.ForeignKey(
450 settings.AUTH_USER_MODEL,
451 on_delete=models.deletion.CASCADE,
452 null=True,
453 blank=True,
454 default=None,
455 )
456 project = models.ForeignKey(
457 "trans.Project",
458 on_delete=models.deletion.CASCADE,
459 null=True,
460 blank=True,
461 default=None,
462 )
463 from_file = models.BooleanField(default=False)
464 shared = models.BooleanField(default=False)
465 status = models.IntegerField(
466 choices=STATUS_CHOICES,
467 default=STATUS_PENDING,
468 )
470 objects = MemoryManager.from_queryset(MemoryQuerySet)()
472 class Meta:
473 verbose_name = "Translation memory entry"
474 verbose_name_plural = "Translation memory entries"
475 indexes = [ # noqa: RUF012
476 # Additional indexes are created manually in the migration for full text search
477 # Use MD5 to index text fields, applied in MemoryQuerySet.filter
478 models.Index(
479 MD5("origin"),
480 MD5("source"),
481 MD5("target"),
482 "source_language",
483 "target_language",
484 name="memory_md5_index",
485 ),
486 # Partial index for to optimize lookup for file based entries
487 # MySQL/MariaDB does not supports condition and uses full index instead.
488 models.Index(
489 "from_file",
490 condition=Q(from_file=True),
491 name="memory_from_file",
492 ),
493 ]
495 def __str__(self) -> str:
496 return f"Memory: {self.source_language}:{self.target_language}"
498 def get_origin_display(self):
499 if self.project:
500 text = pgettext("Translation memory category", "Project: {}")
501 elif self.user:
502 text = pgettext("Translation memory category", "Personal: {}")
503 elif self.shared:
504 text = pgettext("Translation memory category", "Shared: {}")
505 elif self.from_file:
506 text = pgettext("Translation memory category", "File: {}")
507 else:
508 text = "Unknown: {}"
509 return text.format(self.origin)
511 def get_category(self):
512 if self.from_file:
513 return CATEGORY_FILE
514 if self.shared:
515 return CATEGORY_SHARED
516 if self.project_id:
517 return CATEGORY_PRIVATE_OFFSET + self.project_id
518 if self.user_id:
519 return CATEGORY_USER_OFFSET + self.user_id
520 return 0
522 def as_dict(self):
523 """Convert to dict suitable for JSON export."""
524 return {
525 "source": self.source,
526 "context": self.context,
527 "target": self.target,
528 "source_language": self.source_language.code,
529 "target_language": self.target_language.code,
530 "origin": self.origin,
531 "category": self.get_category(),
532 "status": self.status,
533 }