Coverage for app/venv/lib/python3.14/site-packages/weblate/trans/models/unit.py: 43%
989 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 operator
8import re
9from functools import partial, reduce
10from typing import TYPE_CHECKING, Any, Literal, TypedDict
12import sentry_sdk
13from django.conf import settings
14from django.core.cache import cache
15from django.db import Error as DjangoDatabaseError
16from django.db import models, transaction
17from django.db.models import Count, Max, Q, Sum, Value
18from django.db.models.functions import MD5, Length, Lower
19from django.utils import timezone
20from django.utils.functional import cached_property
21from django.utils.translation import gettext, gettext_lazy, ngettext
22from pyparsing import ParseException
24from weblate.auth.results import PermissionResult
25from weblate.checks.flags import Flags
26from weblate.checks.models import CHECKS, Check
27from weblate.formats.helpers import CONTROLCHARS
28from weblate.memory.tasks import handle_unit_translation_change
29from weblate.memory.utils import is_valid_memory_entry
30from weblate.trans.actions import ActionEvents
31from weblate.trans.autofixes import fix_target
32from weblate.trans.mixins import LoggerMixin
33from weblate.trans.models.category import Category
34from weblate.trans.models.change import Change
35from weblate.trans.models.comment import Comment
36from weblate.trans.models.pending import PendingUnitChange
37from weblate.trans.models.project import Project
38from weblate.trans.models.suggestion import Suggestion
39from weblate.trans.models.variant import Variant
40from weblate.trans.signals import unit_post_sync, unit_pre_create
41from weblate.trans.util import (
42 count_words,
43 get_distinct_translations,
44 is_plural,
45 is_unused_string,
46 join_plural,
47 split_plural,
48)
49from weblate.trans.validators import validate_check_flags
50from weblate.utils import messages
51from weblate.utils.db import using_postgresql, verify_in_transaction
52from weblate.utils.errors import report_error
53from weblate.utils.hash import calculate_hash, hash_to_checksum
54from weblate.utils.state import (
55 STATE_APPROVED,
56 STATE_EMPTY,
57 STATE_FUZZY,
58 STATE_READONLY,
59 STATE_TRANSLATED,
60 StringState,
61)
63if TYPE_CHECKING: 63 ↛ 64line 63 didn't jump to line 64 because the condition on line 63 was never true
64 from collections.abc import Generator, Iterable
65 from datetime import datetime
67 from weblate.auth.models import AuthenticatedHttpRequest, User
68 from weblate.formats.base import TranslationUnit
69 from weblate.machinery.base import UnitMemoryResultDict
72NEWLINES = re.compile(r"\r\n|\r|\n")
75def fill_in_source_translation(units: Iterable[Unit]) -> None:
76 """
77 Inject source translation intro component from the source unit.
79 This materializes the query.
81 This assumes prefetch_source() was called before on the query.
82 """
83 for unit in units:
84 unit.translation.component.source_translation = unit.source_unit.translation
87class UnitQuerySet(models.QuerySet["Unit"]):
88 def prefetch(self):
89 from weblate.trans.models import Component
91 return self.prefetch_related(
92 "translation",
93 "translation__language",
94 "translation__plural",
95 models.Prefetch(
96 "translation__component", queryset=Component.objects.defer_huge()
97 ),
98 "translation__component__category",
99 "translation__component__category__project",
100 "translation__component__category__category",
101 "translation__component__category__category__project",
102 "translation__component__category__category__category",
103 "translation__component__category__category__category__project",
104 "translation__component__project",
105 "translation__component__source_language",
106 )
108 def prefetch_source(self):
109 from weblate.trans.models import Component
111 return self.prefetch_related(
112 "source_unit",
113 "source_unit__translation",
114 "source_unit__translation__language",
115 "source_unit__translation__plural",
116 models.Prefetch(
117 "source_unit__translation__component",
118 queryset=Component.objects.defer_huge(),
119 ),
120 "source_unit__translation__component__source_language",
121 "source_unit__translation__component__project",
122 )
124 def fill_in_source_translation(self):
125 """
126 Inject source translation intro component from the source unit.
128 This materializes the query.
130 This assumes prefetch_source() was called before on the query.
131 """
132 fill_in_source_translation(self)
133 return self
135 def prefetch_all_checks(self):
136 return self.prefetch_related(
137 "source_unit",
138 models.Prefetch(
139 "check_set",
140 to_attr="all_checks",
141 ),
142 )
144 def count_screenshots(self):
145 return self.annotate(Count("screenshots"))
147 def prefetch_full(self):
148 return (
149 self.prefetch_all_checks()
150 .prefetch_source()
151 .prefetch_related(
152 "labels",
153 models.Prefetch(
154 "suggestion_set",
155 queryset=Suggestion.objects.order(),
156 to_attr="suggestions",
157 ),
158 models.Prefetch(
159 "comment_set",
160 queryset=Comment.objects.filter(resolved=False),
161 to_attr="unresolved_comments",
162 ),
163 )
164 )
166 def prefetch_bulk(self):
167 """Prefetch useful for bulk editing."""
168 return self.prefetch_full().prefetch_related("defined_variants")
170 def search(self, query, **context) -> UnitQuerySet:
171 """High level wrapper for searching."""
172 from weblate.utils.search import parse_query
174 filters, annotations = parse_query(query, **context)
175 result = self.annotate(**annotations).filter(filters)
176 return result.distinct()
178 def same(self, unit: Unit, exclude: bool = True) -> UnitQuerySet:
179 """Get units with same source within same project."""
180 translation = unit.translation
181 component = translation.component
182 result = self.filter(
183 source__lower__md5=MD5(Lower(Value(unit.source))),
184 context__lower__md5=MD5(Lower(Value(unit.context))),
185 source=unit.source,
186 context=unit.context,
187 translation__component__project_id=component.project_id,
188 translation__language_id=translation.language_id,
189 translation__component__source_language_id=component.source_language_id,
190 )
191 if exclude:
192 result = result.exclude(pk=unit.id)
193 return result
195 def same_target(self, unit: Unit, target: str | None = None) -> UnitQuerySet:
196 if target is None:
197 target = unit.target
198 if not target or not any(split_plural(target)):
199 return self.none()
200 translation = unit.translation
201 component = translation.component
202 result = self.filter(
203 state__gte=STATE_TRANSLATED,
204 target__lower__md5=MD5(Lower(Value(target))),
205 target=target,
206 translation__component__project_id=component.project_id,
207 translation__language_id=translation.language_id,
208 translation__component__source_language_id=component.source_language_id,
209 translation__component__allow_translation_propagation=True,
210 translation__plural_id=translation.plural_id,
211 translation__plural__number__gt=1,
212 ).exclude(source=unit.source)
213 if not unit.translation.language.is_case_sensitive():
214 result = result.exclude(source__lower__md5=MD5(Lower(Value(unit.source))))
215 return result
217 def order_by_request(self, form_data, obj) -> UnitQuerySet:
218 sort_list_request = form_data.get("sort_by", "").split(",")
219 available_sort_choices = [
220 "priority",
221 "position",
222 "context",
223 "num_words",
224 "labels",
225 "timestamp",
226 "last_updated",
227 "source",
228 "target",
229 "location",
230 "component",
231 ]
232 countable_sort_choices: dict[str, dict[str, Any]] = {
233 "num_comments": {"order_by": "comment__count", "filter": None},
234 "num_failing_checks": {
235 "order_by": "check__count",
236 "filter": Q(check__dismissed=False),
237 },
238 }
239 sort_list = []
240 for choice in sort_list_request:
241 unsigned_choice = choice.replace("-", "")
242 if unsigned_choice in countable_sort_choices:
243 return self.order_by_count(
244 choice.replace(
245 unsigned_choice,
246 countable_sort_choices[unsigned_choice]["order_by"],
247 ),
248 countable_sort_choices[unsigned_choice]["filter"],
249 )
250 if unsigned_choice in available_sort_choices:
251 if unsigned_choice == "component":
252 sign = "-" if choice[0] == "-" else ""
253 sort_list.extend(
254 [
255 sign + "translation__component__priority",
256 sign + "translation__component__is_glossary",
257 sign + "translation__component__name",
258 ]
259 )
260 continue
262 if unsigned_choice == "labels":
263 choice = choice.replace("labels", "max_labels_name")
264 sort_list.append(choice)
265 if not sort_list:
266 if hasattr(obj, "component") and obj.component.is_glossary:
267 sort_list = ["source"]
268 elif isinstance(obj, (Project, Category)):
269 sort_list = [
270 "translation__component__priority",
271 "translation__component__is_glossary",
272 "translation__component__name",
273 "-priority",
274 ]
275 else:
276 sort_list = ["-priority", "position"]
277 if "max_labels_name" in sort_list or "-max_labels_name" in sort_list:
278 return self.annotate(max_labels_name=Max("labels__name")).order_by(
279 *sort_list
280 )
281 return self.order_by(*sort_list)
283 def order_by_count(self, choice: str, count_filter) -> UnitQuerySet:
284 model = choice.split("__", 1)[0].replace("-", "")
285 annotation_name = choice.replace("-", "")
286 return self.annotate(
287 **{annotation_name: Count(model, filter=count_filter)}
288 ).order_by(choice)
290 @cached_property
291 def source_context_lookup(self):
292 return {(unit.context, unit.source): unit for unit in self}
294 @cached_property
295 def source_lookup(self) -> dict[str, Unit]:
296 return {unit.source: unit for unit in self}
298 def get_unit(self, ttunit: TranslationUnit) -> Unit:
299 """
300 Find unit matching translate-toolkit unit.
302 This is used for import, so kind of fuzzy matching is expected.
303 """
304 source = ttunit.source
305 context = ttunit.context
307 contexts = [context]
308 # Special case for XLIFF, strip file
309 if "///" in context:
310 contexts.append(context.split("///", 1)[1])
312 # Try with empty context if exact context is not found, useful for importing
313 # monolingual to bilingual
314 if context:
315 contexts.append("")
317 # Lookups based on context
318 for match in contexts:
319 try:
320 return self.source_context_lookup[match, source]
321 except KeyError:
322 continue
324 # Fallback to source string only lookup
325 try:
326 return self.source_lookup[source]
327 except KeyError:
328 msg = "No matching unit found!"
329 raise Unit.DoesNotExist(msg) from None
331 def order(self):
332 return self.order_by("-priority", "position")
334 def filter_access(self, user: User):
335 result = self
336 if user.needs_project_filter: 336 ↛ 337line 336 didn't jump to line 337 because the condition on line 336 was never true
337 result = result.filter(
338 translation__component__project__in=user.allowed_projects
339 )
340 if user.needs_component_restrictions_filter: 340 ↛ 341line 340 didn't jump to line 341 because the condition on line 340 was never true
341 result = result.filter(
342 Q(translation__component__restricted=False)
343 | Q(translation__component_id__in=user.component_permissions)
344 )
345 return result
347 def get_ordered(self, ids):
348 """Return list of units ordered by ID."""
349 return sorted(self.filter(id__in=ids), key=lambda unit: ids.index(unit.id))
351 def select_for_update(self) -> UnitQuerySet: # type: ignore[override]
352 if using_postgresql(): 352 ↛ 356line 352 didn't jump to line 356 because the condition on line 352 was always true
353 # Use weaker locking and limit locking to Unit table only
354 return super().select_for_update(no_key=True, of=("self",))
355 # Discard any select_related to avoid locking additional tables
356 return super().select_for_update().select_related(None)
358 def annotate_stats(self):
359 return self.annotate(
360 strings=Count("pk"), words=Sum("num_words"), chars=Sum(Length("source"))
361 )
364class LabelsField(models.ManyToManyField):
365 def save_form_data(self, instance, data) -> None:
366 from weblate.trans.models.label import TRANSLATION_LABELS
368 super().save_form_data(instance, data)
370 # Delete translation labels when not checked
371 new_labels = {label.name for label in data}
372 through = getattr(instance, self.attname).through.objects
373 for label in TRANSLATION_LABELS:
374 if label not in new_labels:
375 through.filter(unit__source_unit=instance, label__name=label).delete()
378class OldUnit(TypedDict):
379 state: StringState
380 source: str
381 target: str
382 context: str
383 extra_flags: str
384 explanation: str
387class UnitAttributesDict(TypedDict):
388 location: str
389 explanation: str
390 source_explanation: str
391 flags: Flags
392 source: str
393 target: str
394 context: str
395 note: str
396 previous_source: str
397 unit: TranslationUnit
398 created: bool
399 pos: int
400 id_hash: int
403class Unit(models.Model, LoggerMixin):
404 translation = models.ForeignKey(
405 "trans.Translation", on_delete=models.deletion.CASCADE, db_index=False
406 )
407 id_hash = models.BigIntegerField()
408 location = models.TextField(default="", blank=True)
409 context = models.TextField(default="", blank=True)
410 note = models.TextField(default="", blank=True)
411 flags = models.TextField(default="", blank=True)
412 source = models.TextField()
413 previous_source = models.TextField(default="", blank=True)
414 target = models.TextField(default="", blank=True)
415 state = models.IntegerField(default=STATE_EMPTY, choices=StringState.choices)
416 # Stores string state ignoring Weblate originated read-only state
417 original_state = models.IntegerField(
418 default=STATE_EMPTY, choices=StringState.choices
419 )
420 details = models.JSONField(default=dict)
422 position = models.IntegerField()
424 num_words = models.IntegerField(default=0)
426 priority = models.IntegerField(default=100)
428 timestamp = models.DateTimeField(auto_now_add=True)
429 last_updated = models.DateTimeField(auto_now=True)
431 extra_flags = models.TextField(
432 verbose_name=gettext_lazy("Translation flags"),
433 default="",
434 help_text=gettext_lazy(
435 "Additional comma-separated flags to influence Weblate behavior."
436 ),
437 validators=[validate_check_flags],
438 blank=True,
439 )
440 explanation = models.TextField(
441 verbose_name=gettext_lazy("Explanation"),
442 default="",
443 blank=True,
444 help_text=gettext_lazy(
445 "Additional explanation to clarify meaning or usage of the string."
446 ),
447 )
448 variant = models.ForeignKey(
449 "trans.Variant",
450 on_delete=models.deletion.SET_NULL,
451 blank=True,
452 null=True,
453 default=None,
454 )
455 labels = LabelsField("Label", verbose_name=gettext_lazy("Labels"), blank=True)
457 # The type annotation hides that field can be None because
458 # save() updates it to non-None immediately.
459 source_unit: Unit = models.ForeignKey(
460 "trans.Unit", on_delete=models.deletion.CASCADE, blank=True, null=True
461 ) # type: ignore[assignment]
463 objects = UnitQuerySet.as_manager()
465 class Meta:
466 app_label = "trans"
467 unique_together = [("translation", "id_hash")] # noqa: RUF012
468 verbose_name = "string"
469 verbose_name_plural = "strings"
470 indexes = [ # noqa: RUF012
471 models.Index(
472 MD5(Lower("source")), "translation", name="trans_unit_source_md5"
473 ),
474 models.Index(
475 MD5(Lower("target")), "translation", name="trans_unit_target_md5"
476 ),
477 models.Index(
478 MD5(Lower("context")), "translation", name="trans_unit_context_md5"
479 ),
480 ]
482 def __str__(self) -> str:
483 source = self.get_source_plurals()[0]
484 if self.translation.is_template: 484 ↛ 485line 484 didn't jump to line 485 because the condition on line 484 was never true
485 name = self.context
486 elif self.context: 486 ↛ 487line 486 didn't jump to line 487 because the condition on line 486 was never true
487 name = f"[{self.context}] {source}"
488 else:
489 name = source
490 return f"{self.pk}: {name}"
492 def __init__(self, *args, **kwargs) -> None:
493 super().__init__(*args, **kwargs)
494 self.is_batch_update = False
495 self.source_updated = False
496 self.check_cache: dict[str, Any] = {}
497 self.trigger_update_variants = True
498 self.fixups: list[str] = []
499 # Data for machinery integration
500 self.machinery: UnitMemoryResultDict = {}
501 # PluralMapper integration
502 self.plural_map: list[str] = []
503 # Data for glossary integration
504 self.glossary_terms: list[Unit] | None = None
505 self.glossary_positions: tuple[tuple[int, int], ...] = ()
506 # Project backup integration
507 self.import_data: dict[str, Any] = {}
508 # Store original attributes for change tracking
509 self.old_unit: OldUnit
510 # Unit attributes used when parsing
511 self.unit_attributes: UnitAttributesDict | None = None
512 # To handle pending update for enforced checks
513 self.pending_unit_change: PendingUnitChange | None = None
514 # Avoid loading self-referencing source unit from the database
515 # Skip this when deferred fields are present to avoid database access
516 if (
517 self.id
518 and not self.get_deferred_fields()
519 and self.source_unit_id == self.id
520 ):
521 self.source_unit = self
522 if "state" in self.__dict__ and "source" in self.__dict__: 522 ↛ exitline 522 didn't return from function '__init__' because the condition on line 522 was always true
523 # Avoid storing if .only() was used to fetch the query (eg. in stats)
524 self.store_old_unit(self)
526 def save( # type: ignore[override]
527 self,
528 *,
529 same_content: bool = False,
530 run_checks: bool = True,
531 force_propagate_checks: bool = False,
532 force_insert: bool = False,
533 force_update: bool = False,
534 only_save: bool = False,
535 sync_terminology: bool = True,
536 using=None,
537 update_fields: list[str] | None = None,
538 ) -> None:
539 """
540 Save the unit.
542 Wrapper around save to run checks or update fulltext.
543 """
544 # Store number of words
545 if not same_content or not self.num_words:
546 self.num_words = count_words(
547 self.source, self.translation.component.source_language
548 )
549 if update_fields and "num_words" not in update_fields:
550 update_fields.append("num_words")
552 # Update last_updated timestamp
553 if update_fields and "last_updated" not in update_fields:
554 update_fields.append("last_updated")
556 # Actually save the unit
557 super().save(
558 force_insert=force_insert,
559 force_update=force_update,
560 using=using,
561 update_fields=update_fields,
562 )
564 # Set source_unit for source units, this needs to be done after
565 # having a primary key
566 if self.is_source and not self.source_unit_id:
567 self.source_unit = self
568 # Avoid using save() for recursion
569 Unit.objects.filter(pk=self.pk).update(source_unit=self)
571 if only_save:
572 return
574 # Update checks if content or fuzzy flag has changed
575 if run_checks: 575 ↛ 577line 575 didn't jump to line 577 because the condition on line 575 was always true
576 self.run_checks(force_propagate=force_propagate_checks)
577 if self.is_source:
578 self.source_unit_save()
580 # Update manual variants
581 if (
582 self.old_unit["extra_flags"] != self.extra_flags
583 or self.context != self.old_unit["context"]
584 or force_insert
585 ):
586 self.update_variants()
588 # Update terminology
589 if sync_terminology: 589 ↛ exitline 589 didn't return from function 'save' because the condition on line 589 was always true
590 self.sync_terminology()
592 def get_absolute_url(self) -> str:
593 return f"{self.translation.get_translate_url()}?checksum={self.checksum}"
595 def get_url_path(self):
596 return (*self.translation.get_url_path(), str(self.pk))
598 def invalidate_checks_cache(self) -> None:
599 self.check_cache = {}
600 for key in ["propagated_units"]:
601 if key in self.__dict__:
602 del self.__dict__[key]
604 def store_old_unit(self, unit) -> None:
605 self.old_unit = {
606 "state": unit.state,
607 "source": unit.source,
608 "target": unit.target,
609 "context": unit.context,
610 "extra_flags": unit.extra_flags,
611 "explanation": unit.explanation,
612 }
614 def store_disk_state(self) -> None:
615 """
616 Store a snapshot of unit state before pending changes are created.
618 Only stores the state if it doesn't already exist (first pending change).
619 """
620 if self.old_unit is None: 620 ↛ 621line 620 didn't jump to line 621 because the condition on line 620 was never true
621 msg = "`store_old_unit` should be called before saving disk state."
622 raise ValueError(msg)
623 if "disk_state" not in self.details:
624 self.details["disk_state"] = {
625 "target": self.old_unit["target"],
626 "state": self.old_unit["state"],
627 "explanation": self.old_unit["explanation"],
628 }
629 self.save(same_content=True, only_save=True, update_fields=["details"])
631 def clear_disk_state(self) -> None:
632 """
633 Clear the disk_state snapshot from the details field.
635 This should be called after all pending changes for this unit have been committed.
636 """
637 if "disk_state" in self.details: 637 ↛ 638line 637 didn't jump to line 638 because the condition on line 637 was never true
638 del self.details["disk_state"]
639 self.save(same_content=True, only_save=True, update_fields=["details"])
641 def get_comparison_state(self) -> dict[str, Any]:
642 """
643 Get the fields that can differ in database and disk to compare against during check_sync.
645 If disk_state exists, returns it. Otherwise, returns the current unit state.
646 This allows comparing file contents against the state before pending changes.
648 Returns:
649 Dictionary containing unit state fields.
651 """
652 if "disk_state" in self.details:
653 return self.details["disk_state"]
655 return {
656 "target": self.target,
657 "state": self.state,
658 "explanation": self.explanation,
659 }
661 @property
662 def approved(self) -> bool:
663 return self.state == STATE_APPROVED
665 @property
666 def translated(self) -> bool:
667 return self.state >= STATE_TRANSLATED
669 @property
670 def readonly(self) -> bool:
671 return self.state == STATE_READONLY
673 @property
674 def fuzzy(self) -> bool:
675 return self.state == STATE_FUZZY
677 @property
678 def has_failing_check(self) -> bool:
679 return bool(self.active_checks)
681 @property
682 def has_comment(self) -> bool:
683 # Use bool here as unresolved_comments might be list
684 # or a queryset (from prefetch)
685 return bool(self.unresolved_comments)
687 @property
688 def has_suggestion(self) -> bool:
689 return bool(self.suggestions)
691 def source_unit_save(self) -> None:
692 # Run checks, update state and priority if flags changed
693 # or running bulk edit
694 if (
695 self.old_unit["extra_flags"] != self.extra_flags
696 or self.state != self.old_unit["state"]
697 ):
698 # We can not exclude current unit here as we need to trigger
699 # the updates below
700 for unit in self.unit_set.select_for_update().prefetch().prefetch_bulk():
701 # Share component instance for locking and possible bulk updates
702 unit.translation.component = self.translation.component
703 unit.update_state()
704 unit.update_priority()
705 unit.run_checks()
706 if not self.is_batch_update: 706 ↛ 707line 706 didn't jump to line 707 because the condition on line 706 was never true
707 self.translation.component.invalidate_cache()
709 def sync_terminology(self) -> None:
710 try:
711 unit_flags = Flags(self.flags)
712 except ParseException:
713 unit_flags = None
714 new_flags = Flags(self.extra_flags, unit_flags)
716 if "terminology" in new_flags: 716 ↛ 717line 716 didn't jump to line 717 because the condition on line 716 was never true
717 self.translation.component.schedule_sync_terminology()
719 def update_variants(self) -> None:
720 variants = self.defined_variants.all()
721 component = self.translation.component
722 flags = self.all_flags
723 new_variant = None
724 remove = False
726 if not flags.has_value("variant"): 726 ↛ 729line 726 didn't jump to line 729 because the condition on line 726 was always true
727 remove = bool(variants)
728 else:
729 new_variant = flags.get_value("variant")
730 if any(variant.key != new_variant for variant in variants):
731 remove = True
733 # Delete stale variant
734 if remove: 734 ↛ 735line 734 didn't jump to line 735 because the condition on line 734 was never true
735 for variant in variants:
736 variant.defining_units.remove(self)
737 if variant.defining_units.count() == 0:
738 variant.delete()
739 else:
740 variant.unit_set.filter(id_hash=self.id_hash).update(variant=None)
742 # Add new variant
743 if new_variant: 743 ↛ 744line 743 didn't jump to line 744 because the condition on line 743 was never true
744 variant = Variant.objects.get_or_create(
745 key=new_variant, component=component
746 )[0]
747 variant.defining_units.add(self)
749 # Update variant links
750 if ( 750 ↛ 758line 750 didn't jump to line 758 because the condition on line 750 was never true
751 remove
752 or new_variant
753 or (
754 component.variant_regex
755 and re.findall(component.variant_regex, self.context)
756 )
757 ):
758 if self.trigger_update_variants:
759 component.update_variants()
760 else:
761 component.needs_variants_update = True
763 def get_unit_state(
764 self,
765 unit,
766 flags: Flags | str | None,
767 string_changed: bool = False,
768 disk_unit_state: StringState | None = None,
769 ) -> StringState:
770 """Calculate translated and fuzzy status."""
771 # Read-only from the file format
772 if unit.is_readonly(): 772 ↛ 773line 772 didn't jump to line 773 because the condition on line 772 was never true
773 return STATE_READONLY
775 if flags is not None:
776 # Read-only from the source
777 if ( 777 ↛ 782line 777 didn't jump to line 782 because the condition on line 777 was never true
778 not self.is_source
779 and self.source_unit.state < STATE_TRANSLATED
780 and self.translation.component.intermediate
781 ):
782 return STATE_READONLY
784 # Read-only from flags
785 if "read-only" in self.get_all_flags(flags):
786 return STATE_READONLY
788 # We need to keep approved/fuzzy state for formats which do not
789 # support saving it
790 is_existing_fuzzy_state = self.fuzzy or disk_unit_state == STATE_FUZZY
791 if unit.is_fuzzy(is_existing_fuzzy_state and not string_changed): 791 ↛ 792line 791 didn't jump to line 792 because the condition on line 791 was never true
792 return STATE_FUZZY
794 if not unit.is_translated():
795 return STATE_EMPTY
797 is_existing_approved_state = self.approved or disk_unit_state == STATE_APPROVED
798 if ( 798 ↛ 802line 798 didn't jump to line 802 because the condition on line 798 was never true
799 unit.is_approved(is_existing_approved_state and not string_changed)
800 and self.translation.enable_review
801 ):
802 return STATE_APPROVED
804 return STATE_TRANSLATED
806 @staticmethod
807 def check_valid(texts) -> None:
808 for text in texts:
809 if any(char in text for char in CONTROLCHARS): 809 ↛ 810line 809 didn't jump to line 810 because the condition on line 809 was never true
810 raise ValueError(
811 gettext("String contains control character: %s") % repr(text)
812 )
814 def update_source_unit(
815 self, component, source, context, pos, note, location, flags: Flags, explanation
816 ) -> None:
817 source_unit = component.get_source(
818 self.id_hash,
819 create={
820 "source": source,
821 "target": source,
822 "context": context,
823 "position": pos,
824 "note": note,
825 "location": location,
826 "explanation": explanation,
827 "flags": flags.format(),
828 },
829 )
830 same_flags = flags == Flags(source_unit.flags)
831 if ( 831 ↛ 841line 831 didn't jump to line 841 because the condition on line 831 was never true
832 not source_unit.source_updated
833 and not source_unit.translation.filename
834 and (
835 pos != source_unit.position
836 or location != source_unit.location
837 or not same_flags
838 or note != source_unit.note
839 )
840 ):
841 source_unit.position = pos
842 source_unit.source_updated = True
843 source_unit.location = location
844 source_unit.explanation = explanation
845 source_unit.flags = flags.format()
846 source_unit.note = note
847 source_unit.save(
848 update_fields=["position", "location", "explanation", "flags", "note"],
849 same_content=True,
850 run_checks=False,
851 only_save=same_flags,
852 )
853 self.source_unit = source_unit
855 def store_unit_attributes(
856 self,
857 *,
858 unit: TranslationUnit,
859 pos: int,
860 created: bool,
861 id_hash: int,
862 ) -> UnitAttributesDict:
863 """Get unit attributes."""
864 if self.unit_attributes is None: 864 ↛ 909line 864 didn't jump to line 909 because the condition on line 864 was always true
865 translation = self.translation
866 component = translation.component
867 try:
868 location = unit.locations
869 if self.translation.component.file_format_cls.supports_explanation: 869 ↛ 870line 869 didn't jump to line 870 because the condition on line 869 was never true
870 explanation = unit.explanation
871 source_explanation = unit.source_explanation
872 else:
873 explanation = self.explanation
874 source_explanation = "" if created else self.source_unit.explanation
875 flags = Flags(unit.flags)
876 source = unit.source
877 self.check_valid(split_plural(source))
878 if not translation.is_template and translation.is_source:
879 # Load target from source string for bilingual source translations
880 target = source
881 else:
882 target = unit.target
883 self.check_valid(split_plural(target))
884 context = unit.context
885 self.check_valid([context])
886 note = unit.notes
887 previous_source = unit.previous_source
888 except DjangoDatabaseError:
889 raise
890 except Exception as error:
891 report_error("Unit update error", project=component.project)
892 translation.component.handle_parse_error(error, translation)
894 self.unit_attributes = {
895 "location": location,
896 "explanation": explanation,
897 "source_explanation": source_explanation,
898 "flags": flags,
899 "source": source,
900 "target": target,
901 "context": context,
902 "note": note,
903 "previous_source": previous_source,
904 "unit": unit,
905 "created": created,
906 "pos": pos,
907 "id_hash": id_hash,
908 }
909 return self.unit_attributes
911 def update_from_unit( # noqa: C901,PLR0914
912 self,
913 *,
914 user: User | None = None,
915 author: User | None = None,
916 ) -> None:
917 """Update Unit from ttkit unit."""
918 translation = self.translation
919 component = translation.component
920 self.is_batch_update = True
921 self.trigger_update_variants = False
922 self.source_updated = True
924 pending = False
926 # Get unit attributes
927 if self.unit_attributes is None: 927 ↛ 928line 927 didn't jump to line 928 because the condition on line 927 was never true
928 msg = "store_unit_attributes has to be called first"
929 raise ValueError(msg)
930 unit_attributes = self.unit_attributes
931 location = unit_attributes["location"]
932 explanation = unit_attributes["explanation"]
933 source_explanation = unit_attributes["source_explanation"]
934 flags = unit_attributes["flags"]
935 source = unit_attributes["source"]
936 target = unit_attributes["target"]
937 context = unit_attributes["context"]
938 note = unit_attributes["note"]
939 source_change = previous_source = unit_attributes["previous_source"]
940 unit = unit_attributes["unit"]
941 created = unit_attributes["created"]
942 pos = unit_attributes["pos"]
944 # Should not be needed again
945 self.unit_attributes = None
947 # Ensure we track source string for bilingual, this can not use
948 # Unit.is_source as that depends on source_unit attribute, which
949 # we set here
950 old_source_unit = self.source_unit
951 if not translation.is_source:
952 self.update_source_unit(
953 component,
954 source,
955 context,
956 pos,
957 note,
958 location,
959 flags,
960 source_explanation,
961 )
963 # Get comparison state (disk_state if exists, otherwise current state)
964 comparison_state = self.get_comparison_state()
966 # Has source/target changed
967 same_source = source == self.source and context == self.context
968 same_target = target == comparison_state["target"]
970 # Calculate state
971 state = self.get_unit_state(
972 unit,
973 flags,
974 string_changed=not same_source or not same_target,
975 disk_unit_state=comparison_state["state"],
976 )
977 original_state = self.get_unit_state(unit, None)
979 # Monolingual files handling (without target change)
980 if ( 980 ↛ 986line 980 didn't jump to line 986 because the condition on line 980 was never true
981 not created
982 and state != STATE_READONLY
983 and unit.template is not None
984 and same_target
985 ):
986 if not same_source and state in {STATE_TRANSLATED, STATE_APPROVED}:
987 if (
988 self.previous_source == source
989 and comparison_state["state"] == STATE_FUZZY
990 ):
991 # Source change was reverted
992 source_change = self.source
993 previous_source = ""
994 state = STATE_TRANSLATED
995 else:
996 # Store previous source and fuzzy flag for monolingual
997 if not previous_source:
998 source_change = previous_source = self.source
999 # Keep prevoious source if already set in case source
1000 # changes multiple times
1001 if self.previous_source:
1002 previous_source = self.previous_source
1003 state = STATE_FUZZY
1004 pending = True
1005 elif (
1006 comparison_state["state"] == STATE_FUZZY
1007 and state == STATE_FUZZY
1008 and not previous_source
1009 ):
1010 # Avoid losing previous source of fuzzy strings
1011 previous_source = self.previous_source
1013 # Update checks on fuzzy update or on content change
1014 same_state = state == comparison_state["state"] and flags == Flags(self.flags)
1015 same_metadata = (
1016 location == self.location
1017 and explanation == comparison_state["explanation"]
1018 and note == self.note
1019 and pos == self.position
1020 and not pending
1021 )
1022 same_data = (
1023 not created
1024 and same_source
1025 and same_target
1026 and same_state
1027 and flags == Flags(self.flags)
1028 and previous_source == self.previous_source
1029 and self.source_unit == old_source_unit
1030 and old_source_unit is not None
1031 )
1033 # Conditionally check original state changes if it would be used. It is not
1034 # properly tracked in PendingUnitChange, so this would not work for units
1035 # with pending changes. But there shouldn't be any uncommitable pending changes
1036 # for read-only units.
1037 if STATE_READONLY in {state, self.state, comparison_state["state"]}:
1038 same_data &= original_state == self.original_state
1040 # Check if we actually need to change anything
1041 if same_data and same_metadata:
1042 return
1044 # Store updated values
1045 self.original_state = original_state
1046 self.position = pos
1047 self.location = location
1048 self.explanation = explanation
1049 self.flags = flags.format()
1050 self.source = source
1051 self.target = target
1052 self.state = state
1053 self.context = context
1054 self.note = note
1055 self.previous_source = previous_source
1056 self.update_priority(save=False)
1058 # Metadata update only, these do not trigger any actions in Weblate and
1059 # are display only
1060 if same_data and not same_metadata: 1060 ↛ 1061line 1060 didn't jump to line 1061 because the condition on line 1060 was never true
1061 self.save(
1062 same_content=True,
1063 only_save=True,
1064 update_fields=["location", "explanation", "note", "position"],
1065 )
1066 return
1068 # Sanitize number of plurals
1069 if self.is_plural and not component.file_format_cls.has_multiple_strings: 1069 ↛ 1070line 1069 didn't jump to line 1070 because the condition on line 1069 was never true
1070 self.target = join_plural(self.get_target_plurals())
1072 if created: 1072 ↛ 1075line 1072 didn't jump to line 1075 because the condition on line 1072 was always true
1073 unit_pre_create.send(sender=self.__class__, unit=self)
1075 if not created and not same_target: 1075 ↛ 1076line 1075 didn't jump to line 1076 because the condition on line 1075 was never true
1076 unit_post_sync.send(sender=self.__class__, unit=self, updated_attr="target")
1078 # Save into database
1079 self.save(
1080 force_insert=created,
1081 same_content=same_source and same_target,
1082 run_checks=not same_source or not same_target or not same_state,
1083 )
1084 self.clear_disk_state()
1085 PendingUnitChange.objects.filter(unit=self).delete()
1087 if pending: 1087 ↛ 1088line 1087 didn't jump to line 1088 because the condition on line 1087 was never true
1088 PendingUnitChange.store_unit_change(unit=self)
1089 # Track updated sources for source checks
1090 if translation.is_template: 1090 ↛ 1091line 1090 didn't jump to line 1091 because the condition on line 1090 was never true
1091 component.updated_sources[self.id] = self
1092 # Indicate source string change
1093 if not same_source and source_change: 1093 ↛ 1094line 1093 didn't jump to line 1094 because the condition on line 1093 was never true
1094 translation.update_changes.append(
1095 self.generate_change(
1096 user,
1097 author,
1098 ActionEvents.SOURCE_CHANGE,
1099 check_new=False,
1100 old=source_change,
1101 target=self.source,
1102 save=False,
1103 )
1104 )
1105 # Track VCS change
1106 if not same_data: 1106 ↛ 1120line 1106 didn't jump to line 1120 because the condition on line 1106 was always true
1107 translation.update_changes.append(
1108 self.generate_change(
1109 user,
1110 author,
1111 change_action=translation.create_unit_change_action
1112 if created
1113 else translation.update_unit_change_action,
1114 check_new=False,
1115 save=False,
1116 )
1117 )
1119 # Update translation memory if needed
1120 if created or not same_source or not same_target: 1120 ↛ exitline 1120 didn't return from function 'update_from_unit' because the condition on line 1120 was always true
1121 self.update_translation_memory(needs_user_check=False)
1123 def update_state(self) -> None:
1124 """
1125 Update state based on flags.
1127 Mark read-only strings:
1129 * Flagged with 'read-only'
1130 * Where source string is untranslated
1131 """
1132 if "read-only" in self.all_flags or ( 1132 ↛ 1145line 1132 didn't jump to line 1145 because the condition on line 1132 was always true
1133 not self.is_source
1134 and self.source_unit.state < STATE_TRANSLATED
1135 and self.translation.component.intermediate
1136 ):
1137 if not self.readonly: 1137 ↛ 1138line 1137 didn't jump to line 1138 because the condition on line 1137 was never true
1138 self.original_state = self.state
1139 self.state = STATE_READONLY
1140 self.save(
1141 same_content=True,
1142 run_checks=False,
1143 update_fields=["state", "original_state"],
1144 )
1145 elif self.readonly and self.state != self.original_state:
1146 self.state = self.original_state
1147 self.save(same_content=True, run_checks=False, update_fields=["state"])
1149 def update_priority(self, save: bool = True) -> None:
1150 if self.all_flags.has_value("priority"): 1150 ↛ 1151line 1150 didn't jump to line 1151 because the condition on line 1150 was never true
1151 priority = self.all_flags.get_value("priority")
1152 else:
1153 priority = 100
1154 if self.priority != priority: 1154 ↛ 1155line 1154 didn't jump to line 1155 because the condition on line 1154 was never true
1155 self.priority = priority
1156 if save:
1157 self.save(
1158 same_content=True, run_checks=False, update_fields=["priority"]
1159 )
1161 @cached_property
1162 def is_plural(self) -> bool:
1163 """Check whether message is plural."""
1164 return is_plural(self.source) or is_plural(self.target)
1166 @cached_property
1167 def is_source(self) -> bool:
1168 return self.source_unit_id is None or self.source_unit_id == self.id
1170 def get_source_plurals(self) -> list[str]:
1171 """Return source plurals in array."""
1172 return split_plural(self.source)
1174 @cached_property
1175 def source_string(self) -> str:
1176 """
1177 Return a single source string.
1179 In most cases it's singular with exception of unused singulars
1180 generated by some frameworks.
1181 """
1182 plurals = self.get_source_plurals()
1183 singular = plurals[0]
1184 if len(plurals) == 1 or not is_unused_string(singular):
1185 return singular
1186 return plurals[1]
1188 def adjust_plurals(
1189 self, values: list[str], plurals: int | None = None
1190 ) -> list[str]:
1191 if not self.is_plural: 1191 ↛ 1193line 1191 didn't jump to line 1193 because the condition on line 1191 was always true
1192 plurals = 1
1193 elif plurals is None:
1194 plurals = self.translation.plural.number
1196 # Check if we have expected number of them
1197 if len(values) == plurals:
1198 return values
1200 # Pad with empty translations
1201 while len(values) < plurals:
1202 values.append("")
1204 # Delete extra plurals
1205 while len(values) > plurals:
1206 del values[-1]
1208 return values
1210 def get_target_plurals(self, plurals: int | None = None) -> list[str]:
1211 """Return target plurals in array."""
1212 # Split plurals
1213 ret = split_plural(self.target)
1215 if not self.translation.component.is_multivalue: 1215 ↛ 1218line 1215 didn't jump to line 1218 because the condition on line 1215 was always true
1216 ret = self.adjust_plurals(ret, plurals=plurals)
1218 return ret
1220 def propagate(
1221 self, user: User | None, change_action=None, author=None, request=None
1222 ) -> bool:
1223 """Propagate current translation to all others."""
1224 warnings: list[str] = []
1226 with sentry_sdk.start_span(op="unit.propagate", name=f"{self.pk}"):
1227 to_update: list[Unit] = []
1228 units = self.propagated_units.exclude(
1229 target=self.target, state=self.state
1230 ).select_for_update()
1231 for unit in units:
1232 if user is not None and not (
1233 denied := user.has_perm("unit.edit", unit)
1234 ):
1235 component = unit.translation.component
1236 if request and isinstance(denied, PermissionResult):
1237 warnings.append(
1238 gettext(
1239 "String could not be propagated to %(component)s: %(reason)s"
1240 )
1241 % {"component": component, "reason": denied.reason},
1242 )
1243 continue
1245 # Update unit attributes for the current instance, the database is bulk updated later
1246 unit.store_old_unit(unit)
1247 unit.target = self.target
1248 unit.state = self.state
1250 PendingUnitChange.store_unit_change(
1251 unit=unit,
1252 author=author,
1253 )
1255 to_update.append(unit)
1257 unit.update_translation_memory(user)
1259 if warnings:
1260 if len(warnings) > 10:
1261 messages.warning(
1262 request,
1263 ngettext(
1264 "String could not be propagated to %d component.",
1265 "String could not be propagated to %d components.",
1266 len(warnings),
1267 )
1268 % len(warnings),
1269 )
1270 else:
1271 for warning in warnings:
1272 messages.warning(request, warning)
1273 if not to_update:
1274 return False
1276 # Bulk update units
1277 Unit.objects.filter(pk__in=(unit.pk for unit in to_update)).update(
1278 target=self.target,
1279 state=self.state,
1280 original_state=self.state,
1281 last_updated=self.last_updated,
1282 )
1284 # Postprocess changes and generate change objects
1285 changes = [
1286 unit.post_save(
1287 user,
1288 user,
1289 change_action=ActionEvents.PROPAGATED_EDIT,
1290 check_new=False,
1291 save=False,
1292 )
1293 for unit in to_update
1294 ]
1296 # Bulk create changes
1297 Change.objects.bulk_create(changes)
1299 # Update user stats
1300 if user is not None:
1301 user.profile.increase_count("translated", len(to_update))
1303 return True
1305 def save_backend(
1306 self,
1307 user: User | None,
1308 propagate: bool = True,
1309 change_action=None,
1310 author: User | None = None,
1311 run_checks: bool = True,
1312 request=None,
1313 ) -> bool:
1314 """
1315 Store unit to backend.
1317 Optional user parameters defines authorship of a change.
1319 This should be always called in a transaction with updated unit
1320 locked for update.
1321 """
1322 verify_in_transaction()
1323 # For case when authorship specified, use user
1324 author = author or user
1326 # Propagate to other projects
1327 # This has to be done before changing source for template
1328 was_propagated = False
1329 if propagate:
1330 was_propagated = self.propagate(
1331 user, change_action, author=author, request=request
1332 )
1334 # Return if there was no change
1335 # We have to explicitly check for fuzzy flag change on monolingual
1336 # files, where we handle it ourselves without storing to backend
1337 if (
1338 self.old_unit["state"] == self.state
1339 and self.old_unit["target"] == self.target
1340 and self.old_unit["explanation"] == self.explanation
1341 and not was_propagated
1342 ):
1343 return False
1345 update_fields = ["target", "state", "original_state", "explanation"]
1346 if self.is_source and not self.translation.component.intermediate:
1347 self.source = self.target
1348 update_fields.extend(["source"])
1350 # Unit is pending for write
1351 self.pending_unit_change = PendingUnitChange.store_unit_change(
1352 unit=self,
1353 author=author,
1354 )
1356 # Update translated flag (not fuzzy and at least one translation)
1357 translation = any(self.get_target_plurals())
1358 if self.state >= STATE_TRANSLATED and not translation:
1359 self.state = STATE_EMPTY
1360 elif self.state == STATE_EMPTY and translation:
1361 self.state = STATE_TRANSLATED
1362 self.original_state = self.state
1364 # Save updated unit to database, skip running checks
1365 self.save(
1366 update_fields=update_fields,
1367 run_checks=run_checks,
1368 force_propagate_checks=was_propagated,
1369 )
1371 # Generate change and process it
1372 self.post_save(user or author, author, change_action)
1374 # Update related source strings if working on a template
1375 if self.translation.is_template and self.old_unit["target"] != self.target:
1376 self.update_source_units(self.old_unit["target"], user or author, author)
1378 return True
1380 def post_save(
1381 self,
1382 user: User | None,
1383 author: User | None,
1384 change_action: int | None,
1385 *,
1386 save: bool = True,
1387 check_new: bool = True,
1388 ) -> Change:
1389 # Generate Change object for this change
1390 change = self.generate_change(
1391 user or author, author, change_action, save=save, check_new=check_new
1392 )
1394 if change.action not in {
1395 ActionEvents.UPLOAD,
1396 ActionEvents.AUTO,
1397 ActionEvents.BULK_EDIT,
1398 }:
1399 old_translated = self.translation.stats.translated
1401 # Update translation stats
1402 self.translation.invalidate_cache()
1404 # Postpone completed translation detection for translated strings
1405 if self.state >= STATE_TRANSLATED:
1406 transaction.on_commit(
1407 partial(
1408 self.translation.detect_completed_translation,
1409 change,
1410 old_translated,
1411 )
1412 )
1414 # Update user stats
1415 if save and change.author and not change.author.is_anonymous:
1416 change.author.profile.increase_count("translated")
1417 return change
1419 def update_source_units(
1420 self, previous_source: str, user: User | None, author: User | None
1421 ) -> None:
1422 """
1423 Update source for units within same component.
1425 This is needed when editing template translation for monolingual formats.
1426 """
1427 with sentry_sdk.start_span(op="unit.update_source_units", name=f"{self.pk}"):
1428 changes = []
1430 # Find relevant units
1431 for unit in self.unit_set.exclude(id=self.id).prefetch().prefetch_bulk():
1432 # Update source and number of words
1433 unit.source = self.target
1434 unit.num_words = self.num_words
1435 # Find reverted units
1436 if (
1437 unit.state == STATE_FUZZY
1438 and unit.previous_source == self.target
1439 and unit.target
1440 ):
1441 # Unset fuzzy on reverted
1442 unit.original_state = unit.state = STATE_TRANSLATED
1443 PendingUnitChange.store_unit_change(
1444 unit=unit,
1445 author=author,
1446 )
1447 unit.previous_source = ""
1448 elif (
1449 unit.original_state == STATE_FUZZY
1450 and unit.previous_source == self.target
1451 and unit.target
1452 ):
1453 # Unset fuzzy on reverted
1454 unit.original_state = STATE_TRANSLATED
1455 unit.previous_source = ""
1456 elif unit.state >= STATE_TRANSLATED and unit.target:
1457 # Set fuzzy on changed
1458 unit.original_state = STATE_FUZZY
1459 if unit.state < STATE_READONLY:
1460 unit.state = STATE_FUZZY
1461 PendingUnitChange.store_unit_change(
1462 unit=unit,
1463 author=author,
1464 )
1465 unit.previous_source = previous_source
1467 # Save unit
1468 unit.save()
1469 # Generate change
1470 changes.append(
1471 unit.generate_change(
1472 user,
1473 author,
1474 ActionEvents.SOURCE_CHANGE,
1475 check_new=False,
1476 old=previous_source,
1477 target=self.target,
1478 save=False,
1479 )
1480 )
1481 if changes:
1482 # Bulk create changes
1483 Change.objects.bulk_create(changes)
1484 # Invalidate stats
1485 self.translation.component.invalidate_cache()
1487 def generate_change(
1488 self,
1489 user: User | None,
1490 author: User | None,
1491 change_action: int | None,
1492 *,
1493 check_new: bool = True,
1494 save: bool = True,
1495 old: str | None = None,
1496 target: str | None = None,
1497 ) -> Change:
1498 """Create Change entry for saving unit."""
1499 # Notify about new contributor
1500 if ( 1500 ↛ 1507line 1500 didn't jump to line 1507 because the condition on line 1500 was never true
1501 check_new
1502 and not self.is_batch_update
1503 and user is not None
1504 and not user.is_bot
1505 and not self.translation.change_set.filter(user=user).exists()
1506 ):
1507 self.change_set.create(
1508 unit=self,
1509 action=ActionEvents.NEW_CONTRIBUTOR,
1510 user=user,
1511 author=author,
1512 )
1514 # Action type to store
1515 if change_action is not None: 1515 ↛ 1517line 1515 didn't jump to line 1517 because the condition on line 1515 was always true
1516 action = change_action
1517 elif self.state == STATE_FUZZY:
1518 action = ActionEvents.MARKED_EDIT
1519 elif self.old_unit["state"] >= STATE_FUZZY:
1520 if self.state == STATE_APPROVED:
1521 action = ActionEvents.APPROVE
1522 else:
1523 action = ActionEvents.CHANGE
1524 elif self.state == STATE_APPROVED:
1525 action = ActionEvents.APPROVE
1526 else:
1527 action = ActionEvents.NEW
1529 # Create change object
1530 change = Change(
1531 unit=self,
1532 action=action,
1533 user=user,
1534 author=author,
1535 target=self.target if target is None else target,
1536 old=self.old_unit["target"] if old is None else old,
1537 details={
1538 "state": self.state,
1539 "old_state": self.old_unit["state"],
1540 "source": self.source,
1541 "context": self.context,
1542 },
1543 )
1544 if save: 1544 ↛ 1545line 1544 didn't jump to line 1545 because the condition on line 1544 was never true
1545 change.save(force_insert=True)
1546 return change
1548 @cached_property
1549 def suggestions(self) -> models.QuerySet[Suggestion]:
1550 """Return all suggestions for this unit."""
1551 return self.suggestion_set.order()
1553 @cached_property
1554 def all_checks(self) -> models.QuerySet[Check]:
1555 result = self.check_set.all()
1556 # Force fetching
1557 list(result)
1558 return result
1560 def clear_checks_cache(self) -> None:
1561 if "all_checks" in self.__dict__: 1561 ↛ exitline 1561 didn't return from function 'clear_checks_cache' because the condition on line 1561 was always true
1562 del self.__dict__["all_checks"]
1564 @property
1565 def all_checks_names(self) -> set[str]:
1566 return {check.name for check in self.all_checks}
1568 @property
1569 def dismissed_checks(self) -> list[Check]:
1570 return [check for check in self.all_checks if check.dismissed]
1572 @property
1573 def active_checks(self) -> list[Check]:
1574 """Return all active (not ignored) checks for this unit."""
1575 return [check for check in self.all_checks if not check.dismissed]
1577 @cached_property
1578 def all_comments(self) -> models.QuerySet[Comment]:
1579 """Return list of target comments."""
1580 if self.is_source:
1581 # Add all comments on translation on source string comment
1582 query = Q(unit__source_unit=self)
1583 else:
1584 # Add source string comments for translation unit
1585 query = Q(unit__in=(self, self.source_unit))
1586 return Comment.objects.filter(query).prefetch_related("unit", "user").order()
1588 @cached_property
1589 def unresolved_comments(self) -> list[Comment]:
1590 return [
1591 comment
1592 for comment in self.all_comments
1593 if not comment.resolved and comment.unit_id == self.id
1594 ]
1596 def run_checks( # noqa: C901
1597 self, *, force_propagate: bool = False, skip_propagate: bool = False
1598 ) -> None:
1599 """Update checks for this unit."""
1600 src = self.get_source_plurals()
1601 tgt = self.get_target_plurals()
1603 old_checks = self.all_checks_names
1604 create = []
1606 args: tuple[list[str], Unit] | tuple[list[str], list[str], Unit]
1607 if self.is_source:
1608 checks = CHECKS.source
1609 meth = "check_source"
1610 args = src, self
1611 else:
1612 checks = {} if self.readonly else CHECKS.target
1613 meth = "check_target"
1614 args = src, tgt, self
1615 if self.translation.component.is_glossary:
1616 checks = CHECKS.glossary
1617 meth = "check_target"
1618 args = src, tgt, self
1620 # Initial propagation setup
1621 propagation: set[Literal["source", "target"]] = set()
1622 if force_propagate: 1622 ↛ 1623line 1622 didn't jump to line 1623 because the condition on line 1622 was never true
1623 propagation.add("source")
1625 # Run all checks
1626 for check, check_obj in checks.items():
1627 # Does the check fire?
1628 if getattr(check_obj, meth)(*args): 1628 ↛ 1629line 1628 didn't jump to line 1629 because the condition on line 1628 was never true
1629 if check in old_checks:
1630 # We already have this check
1631 old_checks.remove(check)
1632 # Propagation is handled later in this method
1633 else:
1634 # Create new check
1635 create.append(Check(unit=self, dismissed=False, name=check))
1636 if check_obj.propagates and not skip_propagate:
1637 propagation.add(check_obj.propagates)
1639 if create: 1639 ↛ 1640line 1639 didn't jump to line 1640 because the condition on line 1639 was never true
1640 Check.objects.bulk_create(create, batch_size=500, ignore_conflicts=True)
1642 # Delete no longer failing checks
1643 if old_checks: 1643 ↛ 1644line 1643 didn't jump to line 1644 because the condition on line 1643 was never true
1644 Check.objects.filter(unit=self, name__in=old_checks).delete()
1645 if not skip_propagate:
1646 for check_name in old_checks:
1647 try:
1648 check_obj = CHECKS[check_name]
1649 except KeyError:
1650 # Skip disabled/removed checks
1651 continue
1652 if check_obj.propagates:
1653 if check_obj.propagates == "source":
1654 propagated_units = self.propagated_units
1655 values = set(
1656 propagated_units.values_list("target", flat=True)
1657 )
1658 elif check_obj.propagates == "target":
1659 propagated_units = Unit.objects.same_target(
1660 self, self.old_unit["target"]
1661 )
1662 values = set(
1663 propagated_units.values_list("source", flat=True)
1664 )
1665 else:
1666 message = f"Unsupported propagation: {check_obj.propagates}"
1667 raise ValueError(message)
1669 if len(values) == 1:
1670 for other in propagated_units:
1671 other.check_set.filter(name=check_name).delete()
1672 if (
1673 other.translation != self.translation
1674 or other.source != self.source
1675 ):
1676 other.translation.invalidate_cache()
1677 other.clear_checks_cache()
1679 # Propagate checks which need it (for example consistency)
1680 if propagation: 1680 ↛ 1681line 1680 didn't jump to line 1681 because the condition on line 1680 was never true
1681 querymap: dict[Literal["source", "target"], UnitQuerySet] = {
1682 "source": self.propagated_units,
1683 "target": Unit.objects.same_target(self),
1684 }
1685 propagated_units: UnitQuerySet = reduce(
1686 operator.or_, (querymap[item] for item in propagation)
1687 )
1688 propagated_units = propagated_units.distinct().prefetch_all_checks()
1690 for unit in propagated_units:
1691 try:
1692 unit.run_checks(force_propagate=False, skip_propagate=True)
1693 except Unit.DoesNotExist:
1694 # This can happen in some corner cases like changing
1695 # source language of a project - the source language is
1696 # changed first and then components are updated. But
1697 # not all are yet updated and this spans across them.
1698 continue
1700 # Trigger source checks on target check update (multiple failing checks)
1701 if (create or old_checks) and not self.is_source: 1701 ↛ 1702line 1701 didn't jump to line 1702 because the condition on line 1701 was never true
1702 if self.is_batch_update:
1703 self.translation.component.updated_sources[self.source_unit.id] = (
1704 self.source_unit
1705 )
1706 else:
1707 self.source_unit.run_checks()
1709 # This is always preset as it is used in top of this method
1710 self.clear_checks_cache()
1712 if not self.is_batch_update and (create or old_checks): 1712 ↛ 1713line 1712 didn't jump to line 1713 because the condition on line 1712 was never true
1713 self.translation.invalidate_cache()
1715 def nearby(self, count: int) -> models.QuerySet[Unit]:
1716 """Return list of nearby messages based on location."""
1717 if self.position == 0:
1718 return Unit.objects.none()
1719 with sentry_sdk.start_span(op="unit.nearby", name=f"{self.pk}"):
1720 # Limiting the query is needed to avoid issues when unit
1721 # position is not properly populated
1722 result = (
1723 self.translation.unit_set.prefetch_full()
1724 .order_by("position")
1725 .filter(
1726 position__gte=self.position - count,
1727 position__lte=self.position + count,
1728 )[: ((2 * count) + 1)]
1729 )
1730 # Force materializing the query
1731 return result.fill_in_source_translation()
1733 def nearby_keys(self, count: int) -> Iterable[Unit]:
1734 # Do not show nearby keys on bilingual
1735 if not self.translation.component.has_template():
1736 return []
1737 with sentry_sdk.start_span(op="unit.nearby_keys", name=f"{self.pk}"):
1738 key = self.translation.keys_cache_key
1739 key_list = cache.get(key)
1740 unit_set = self.translation.unit_set
1741 if key_list is None or self.pk not in key_list:
1742 key_list = list(
1743 unit_set.order_by("context").values_list("id", flat=True)
1744 )
1745 cache.set(key, key_list)
1746 offset = key_list.index(self.pk)
1747 nearby = key_list[max(offset - count, 0) : offset + count]
1748 return (
1749 unit_set.filter(id__in=nearby)
1750 .prefetch_full()
1751 .order_by("context")
1752 .fill_in_source_translation()
1753 )
1755 def variants(self) -> Iterable[Unit]:
1756 if not self.variant:
1757 return []
1758 return (
1759 self.variant.unit_set.filter(translation=self.translation)
1760 .prefetch()
1761 .prefetch_full()
1762 .order_by("context")
1763 )
1765 @transaction.atomic
1766 def translate(
1767 self,
1768 user: User | None,
1769 new_target: str | list[str],
1770 new_state: StringState,
1771 *,
1772 change_action: int | None = None,
1773 propagate: bool = True,
1774 author: User | None = None,
1775 request: AuthenticatedHttpRequest | None = None,
1776 add_alternative: bool = False,
1777 ) -> bool:
1778 """
1779 Store new translation of a unit.
1781 Propagation is currently disabled on import.
1782 """
1783 component = self.translation.component
1785 # Force flushing checks cache
1786 self.invalidate_checks_cache()
1788 # Fetch current copy from database and lock it for update
1789 old_unit = Unit.objects.select_for_update().get(pk=self.pk)
1790 self.store_old_unit(old_unit)
1792 # Handle simple string units
1793 new_target_list = [new_target] if isinstance(new_target, str) else new_target
1795 # Handle managing alternative translations
1796 if add_alternative:
1797 new_target_list.append("")
1798 elif component.is_multivalue:
1799 new_target_list = [target for target in new_target_list if target]
1800 if not new_target_list:
1801 new_target_list = [""]
1803 if not component.is_multivalue:
1804 new_target_list = self.adjust_plurals(new_target_list)
1806 # Apply autofixes
1807 if not self.translation.is_template:
1808 new_target_list, self.fixups = fix_target(new_target_list, self)
1810 # Update unit and save it
1811 self.target = join_plural(new_target_list)
1812 not_empty = any(new_target_list)
1814 # Newlines fixup
1815 if "dos-eol" in self.all_flags:
1816 self.target = NEWLINES.sub("\r\n", self.target)
1818 # Update string state
1819 if not_empty:
1820 self.state = new_state
1821 else:
1822 self.state = STATE_EMPTY
1824 # Update original state unless we are updating read-only strings. This
1825 # does never happen directly, but FillReadOnlyAddon does this.
1826 if new_state != STATE_READONLY:
1827 self.original_state = self.state
1829 # Save to the database
1830 saved = self.save_backend(
1831 user,
1832 change_action=change_action,
1833 propagate=propagate,
1834 author=author,
1835 request=request,
1836 )
1838 # Enforced checks can revert the state to needs editing (fuzzy)
1839 if (
1840 self.state >= STATE_TRANSLATED
1841 and component.enforced_checks
1842 and self.all_checks_names & set(component.enforced_checks)
1843 ):
1844 self.state = self.original_state = STATE_FUZZY
1845 self.save(
1846 run_checks=False,
1847 same_content=True,
1848 update_fields=["state", "original_state"],
1849 )
1850 self.generate_change(
1851 user or author, author, ActionEvents.ENFORCED_CHECK, check_new=False
1852 )
1853 if self.pending_unit_change is not None:
1854 # Update PendingUnitChange if there is one
1855 self.pending_unit_change.state = STATE_FUZZY
1856 self.pending_unit_change.save(update_fields=["state"])
1857 elif saved:
1858 # There should be a pending unit if saved
1859 msg = "Updating unit, but pending unit change is not set!"
1860 raise ValueError(msg)
1861 else:
1862 # Generate pending unit change otherwise
1863 PendingUnitChange.store_unit_change(unit=self, author=author)
1864 # Indicate as saved
1865 saved = True
1867 self.update_translation_memory(user)
1869 if change_action == ActionEvents.AUTO:
1870 self.labels.add(component.project.automatically_translated_label)
1871 else:
1872 self.labels.through.objects.filter(
1873 unit=self, label__name="Automatically translated"
1874 ).delete()
1876 return saved
1878 def get_all_flags(self, override: Flags | str | None = None) -> Flags:
1879 """Return union of own and component flags."""
1880 # Validate flags from the unit to avoid crash
1881 try:
1882 unit_flags = Flags(override or self.flags)
1883 except ParseException:
1884 unit_flags = None
1886 # Ordering is important here as that defines overriding
1887 return Flags(
1888 # Base on translation + component flags
1889 self.translation.all_flags,
1890 # Apply unit flags from the file format
1891 unit_flags,
1892 # The source_unit is None before saving the object for the first time
1893 getattr(self.source_unit, "extra_flags", ""),
1894 # This unit flag overrides
1895 self.extra_flags,
1896 )
1898 @cached_property
1899 def all_flags(self) -> Flags:
1900 return self.get_all_flags()
1902 def get_unit_flags(self) -> Flags:
1903 return Flags(self.extra_flags)
1905 @cached_property
1906 def edit_mode(self) -> str:
1907 """Return syntax highlighting mode for Prismjs."""
1908 flags = self.all_flags
1909 if "icu-message-format" in flags:
1910 return "icu-message-format"
1911 if "rst-text" in flags:
1912 return "rest"
1913 if "md-text" in flags:
1914 return "markdown"
1915 if "xml-text" in flags:
1916 return "xml"
1917 if "safe-html" in flags:
1918 return "html"
1919 return "none"
1921 def get_secondary_units(self, user: User) -> list[Unit]:
1922 """Return list of secondary units."""
1923 translation = self.translation
1924 component = translation.component
1925 secondary_langs: set[int] = user.profile.secondary_language_ids
1927 # Add project/component secondary languages
1928 if component.secondary_language_id:
1929 secondary_langs.add(component.secondary_language_id)
1930 elif component.project.secondary_language_id:
1931 secondary_langs.add(component.project.secondary_language_id)
1933 # Remove current source and target language
1934 secondary_langs -= {translation.language_id, component.source_language_id}
1936 if not secondary_langs:
1937 return []
1938 result = get_distinct_translations(
1939 self.source_unit.unit_set.filter(
1940 Q(translation__language__in=secondary_langs)
1941 & Q(state__gte=STATE_TRANSLATED)
1942 & Q(state__lt=STATE_READONLY)
1943 & ~Q(target__lower__md5=MD5(Value("")))
1944 & ~Q(pk=self.pk)
1945 ).select_related(
1946 "source_unit",
1947 "translation__language",
1948 "translation__plural",
1949 )
1950 )
1951 # Avoid fetching component again from the database
1952 for unit in result:
1953 unit.translation.component = component
1954 return result
1956 @property
1957 def checksum(self):
1958 """
1959 Return unique hex identifier.
1961 It's unsigned representation of id_hash in hex.
1962 """
1963 return hash_to_checksum(self.id_hash)
1965 @cached_property
1966 def propagated_units(self) -> UnitQuerySet:
1967 return (
1968 Unit.objects.same(self)
1969 .prefetch()
1970 .filter(
1971 translation__component__allow_translation_propagation=True,
1972 translation__plural_id=self.translation.plural_id,
1973 )
1974 )
1976 def get_max_length(self):
1977 """Return maximal translation length."""
1978 # Fallback to reasonably big value
1979 fallback = 10000
1981 # Not yet saved unit
1982 if not self.pk:
1983 return fallback
1984 # Flag defines length
1985 if self.all_flags.has_value("max-length"):
1986 return self.all_flags.get_value("max-length")
1987 # Avoid limiting source strings
1988 if self.is_source and not self.translation.component.intermediate:
1989 return fallback
1990 # Base length on source string
1991 if settings.LIMIT_TRANSLATION_LENGTH_BY_SOURCE_LENGTH:
1992 return max(100, len(self.get_source_plurals()[0]) * 10)
1994 return fallback
1996 def get_target_hash(self) -> int:
1997 return calculate_hash(self.target)
1999 @cached_property
2000 def content_hash(self) -> int:
2001 return calculate_hash(self.source, self.context)
2003 @cached_property
2004 def recent_content_changes(self):
2005 """
2006 Content changes for a unit ordered by timestamp.
2008 Can be prefetched using prefetch_recent_content_changes.
2009 """
2010 return self.change_set.content().select_related("author").order_by("-timestamp")
2012 def get_last_content_change(self, silent: bool = False) -> tuple[User, datetime]:
2013 """
2014 Get last content change metadata.
2016 Used when committing pending changes, needs to handle and report inconsistencies
2017 from past releases.
2018 """
2019 from weblate.auth.models import get_anonymous
2021 try:
2022 change = self.recent_content_changes[0]
2023 except IndexError:
2024 return get_anonymous(), timezone.now()
2025 return change.author or get_anonymous(), change.timestamp
2027 def get_locations(self) -> Generator[tuple[str, str, str], None, None]:
2028 """Return list of location filenames."""
2029 for location in self.location.split(","):
2030 location = location.strip()
2031 if not location:
2032 continue
2033 location_parts = location.split(":")
2034 if len(location_parts) == 2:
2035 filename, line = location_parts
2036 else:
2037 filename = location_parts[0]
2038 line = "0"
2039 yield location, filename, line
2041 @cached_property
2042 def all_labels(self):
2043 from weblate.trans.models import Label
2045 if self.is_source:
2046 return self.labels.all()
2047 return Label.objects.filter(
2048 unit__id__in=(self.id, self.source_unit_id)
2049 ).distinct()
2051 def get_flag_actions(self):
2052 flags = self.all_flags
2053 translation = self.translation
2054 component = translation.component
2055 result = []
2056 if self.is_source:
2057 if "read-only" in flags:
2058 if (
2059 "read-only" not in translation.all_flags
2060 and "read-only" not in component.all_flags
2061 ):
2062 result.append(
2063 ("removeflag", "read-only", gettext("Unmark as read-only"))
2064 )
2065 else:
2066 result.append(("addflag", "read-only", gettext("Mark as read-only")))
2067 if component.is_glossary:
2068 if "read-only" in self.source_unit.get_unit_flags():
2069 result.append(
2070 ("removeflag", "read-only", gettext("Unmark as untranslatable"))
2071 )
2072 else:
2073 result.append(
2074 ("addflag", "read-only", gettext("Mark as untranslatable"))
2075 )
2076 if "forbidden" in flags:
2077 result.append(
2078 (
2079 "removeflag",
2080 "forbidden",
2081 gettext("Unmark as forbidden translation"),
2082 )
2083 )
2084 else:
2085 result.append(
2086 (
2087 "addflag",
2088 "forbidden",
2089 gettext("Mark as forbidden translation"),
2090 )
2091 )
2092 if "terminology" in flags:
2093 result.append(
2094 (
2095 "removeflag",
2096 "terminology",
2097 gettext("Unmark as terminology"),
2098 )
2099 )
2100 else:
2101 result.append(
2102 (
2103 "addflag",
2104 "terminology",
2105 gettext("Mark as terminology"),
2106 )
2107 )
2108 return result
2110 def invalidate_related_cache(self) -> None:
2111 # Invalidate stats counts
2112 self.translation.invalidate_cache()
2113 # Invalidate unit cached properties
2114 for key in ["all_comments", "suggestions"]:
2115 if key in self.__dict__:
2116 del self.__dict__[key]
2118 def update_explanation(
2119 self, explanation: str, user: User, save: bool = True
2120 ) -> bool:
2121 """Update glossary explanation."""
2122 verify_in_transaction()
2123 old = self.old_unit["explanation"]
2124 if old == explanation:
2125 return False
2126 self.explanation = explanation
2127 file_format_support = (
2128 self.translation.component.file_format_cls.supports_explanation
2129 )
2130 units: Iterable[Unit] = []
2131 if self.is_source:
2132 units = self.unit_set.exclude(id=self.id).select_for_update()
2133 # Mark change as pending if file format supports this
2134 if file_format_support:
2135 for unit in units:
2136 PendingUnitChange.store_unit_change(
2137 unit=unit,
2138 author=user,
2139 )
2141 # translation file does not exist for source strings in bilingual formats
2142 if (
2143 not self.is_source
2144 or self.translation.component.file_format_cls.monolingual
2145 ):
2146 PendingUnitChange.store_unit_change(
2147 unit=self,
2148 author=user,
2149 )
2151 if save:
2152 self.save(update_fields=["explanation"], only_save=True)
2154 # Always generate change for self
2155 units = [*units, self]
2157 for unit in units:
2158 unit.generate_change(
2159 user=user,
2160 author=user,
2161 change_action=ActionEvents.EXPLANATION,
2162 check_new=False,
2163 save=True,
2164 target=explanation,
2165 old=old,
2166 )
2167 return True
2169 def update_extra_flags(
2170 self, extra_flags: str, user: User, save: bool = True
2171 ) -> None:
2172 """Update unit extra flags."""
2173 verify_in_transaction()
2174 old = self.old_unit["extra_flags"]
2175 if old == extra_flags:
2176 return
2177 self.extra_flags = extra_flags
2178 units: Iterable[Unit] = []
2179 if self.is_source:
2180 units = self.unit_set.select_for_update().exclude(id=self.id)
2181 # Always generate change for self
2182 units = [*units, self]
2183 if save:
2184 self.save(update_fields=["extra_flags"], same_content=True)
2186 for unit in units:
2187 unit.generate_change(
2188 user=user,
2189 author=user,
2190 change_action=ActionEvents.EXTRA_FLAGS,
2191 check_new=False,
2192 save=True,
2193 old=old,
2194 target=self.extra_flags,
2195 )
2197 @cached_property
2198 def glossary_sort_key(self):
2199 return (self.translation.component.priority, self.source.lower())
2201 def update_translation_memory(
2202 self, user: User | None = None, *, needs_user_check: bool = True
2203 ) -> None:
2204 if needs_user_check and ( 2204 ↛ 2213line 2204 didn't jump to line 2213 because the condition on line 2204 was never true
2205 not user
2206 or user.is_bot
2207 or not user.is_active
2208 or (
2209 self.target == self.old_unit["target"]
2210 and self.state == self.old_unit["state"]
2211 )
2212 ):
2213 return
2215 translation = self.translation
2216 component = translation.component
2217 if ( 2217 ↛ 2223line 2217 didn't jump to line 2223 because the condition on line 2217 was never true
2218 (not translation.is_source or component.intermediate)
2219 and (self.state >= STATE_TRANSLATED or self.state != self.old_unit["state"])
2220 and not component.is_glossary
2221 and is_valid_memory_entry(source=self.source, target=self.target)
2222 ):
2223 handle_unit_translation_change(self, user)
2225 @property
2226 def has_pending_changes(self) -> bool:
2227 return self.pending_changes.exists()