Coverage for app/venv/lib/python3.14/site-packages/weblate/trans/models/translation.py: 42%
1038 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 codecs
8import contextlib
9import os
10import tempfile
11from datetime import UTC
12from itertools import chain
13from pathlib import Path
14from typing import TYPE_CHECKING, BinaryIO, Literal, NotRequired, TypedDict, overload
16import sentry_sdk
17from django.core.cache import cache
18from django.core.exceptions import ObjectDoesNotExist, ValidationError
19from django.db import IntegrityError, models, transaction
20from django.db.models import F, Q
21from django.urls import reverse
22from django.utils import timezone
23from django.utils.functional import cached_property
24from django.utils.html import format_html
25from django.utils.translation import gettext, ngettext
27from weblate.checks.flags import Flags
28from weblate.formats.auto import try_load
29from weblate.formats.base import UnitNotFoundError
30from weblate.formats.helpers import CONTROLCHARS, NamedBytesIO
31from weblate.lang.models import Language, Plural
32from weblate.trans.actions import ActionEvents
33from weblate.trans.checklists import TranslationChecklistMixin
34from weblate.trans.defines import FILENAME_LENGTH
35from weblate.trans.exceptions import (
36 FailedCommitError,
37 FileParseError,
38 PluralFormsMismatchError,
39)
40from weblate.trans.mixins import CacheKeyMixin, LockMixin, LoggerMixin, URLMixin
41from weblate.trans.models.change import Change
42from weblate.trans.models.pending import PendingUnitChange
43from weblate.trans.models.suggestion import Suggestion
44from weblate.trans.models.unit import Unit
45from weblate.trans.models.variant import Variant
46from weblate.trans.signals import component_post_update, vcs_pre_commit
47from weblate.trans.util import is_plural, join_plural, split_plural
48from weblate.trans.validators import validate_check_flags
49from weblate.utils import messages
50from weblate.utils.errors import report_error
51from weblate.utils.html import format_html_join_comma
52from weblate.utils.render import render_template
53from weblate.utils.site import get_site_url
54from weblate.utils.state import (
55 STATE_APPROVED,
56 STATE_EMPTY,
57 STATE_FUZZY,
58 STATE_READONLY,
59 STATE_TRANSLATED,
60)
61from weblate.utils.stats import GhostStats, TranslationStats
62from weblate.utils.version import GIT_VERSION
64if TYPE_CHECKING: 64 ↛ 65line 64 didn't jump to line 65 because the condition on line 64 was never true
65 from datetime import datetime
67 from weblate.auth.models import AuthenticatedHttpRequest, User
68 from weblate.formats.base import TranslationFormat, TranslationUnit
69 from weblate.utils.state import (
70 StringState,
71 )
73 from .project import Project
75UploadResult = tuple[int, int, int, int]
78class NewUnitParams(TypedDict):
79 context: NotRequired[str]
80 source: str | list[str]
81 target: NotRequired[str | list[str] | None]
82 auto_context: NotRequired[bool]
83 extra_flags: NotRequired[str | None]
84 explanation: NotRequired[str]
85 state: NotRequired[int | None]
86 skip_existing: NotRequired[bool]
89class TranslationManager(models.Manager):
90 def check_sync(
91 self,
92 component,
93 lang,
94 code,
95 path,
96 force=False,
97 request: AuthenticatedHttpRequest | None = None,
98 change=None,
99 ):
100 """Parse translation meta info and updates translation object."""
101 translation, _created = component.translation_set.get_or_create(
102 language=lang,
103 defaults={"filename": path, "language_code": code, "plural": lang.plural},
104 )
105 if translation.filename != path or translation.language_code != code:
106 force = True
107 translation.filename = path
108 translation.language_code = code
109 translation.save(update_fields=["filename", "language_code"])
110 flags = ""
111 if (not component.edit_template and translation.is_template) or (
112 not component.has_template() and translation.is_source
113 ):
114 flags = "read-only"
115 if translation.check_flags != flags: 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true
116 force = True
117 translation.check_flags = flags
118 translation.save(update_fields=["check_flags"])
119 translation.check_sync(force, request=request, change=change)
120 return translation
123class TranslationQuerySet(models.QuerySet):
124 def prefetch(self, *, defer_huge: bool = True):
125 from weblate.trans.models import Component
127 component_prefetch: str | models.Prefetch
128 if defer_huge:
129 component_prefetch = models.Prefetch(
130 "component", queryset=Component.objects.defer_huge()
131 )
132 else:
133 component_prefetch = "component"
135 return self.prefetch_related(
136 component_prefetch,
137 "component__project",
138 "component__category",
139 "component__category__project",
140 "component__category__category",
141 "component__category__category__project",
142 "component__category__category__category",
143 "component__category__category__category__project",
144 ).prefetch_meta()
146 def prefetch_meta(self):
147 from weblate.trans.models import Alert, Component
149 return self.prefetch_related(
150 "language",
151 models.Prefetch(
152 "component__linked_component", queryset=Component.objects.defer_huge()
153 ),
154 "component__linked_component__project",
155 models.Prefetch(
156 "component__alert_set",
157 queryset=Alert.objects.filter(dismissed=False),
158 to_attr="all_active_alerts",
159 ),
160 )
162 def prefetch_plurals(self):
163 return self.prefetch_related("language__plural_set")
165 def filter_access(self, user: User):
166 result = self
167 if user.needs_project_filter: 167 ↛ 168line 167 didn't jump to line 168 because the condition on line 167 was never true
168 result = result.filter(component__project__in=user.allowed_projects)
169 if user.needs_component_restrictions_filter: 169 ↛ 170line 169 didn't jump to line 170 because the condition on line 169 was never true
170 result = result.filter(
171 Q(component__restricted=False)
172 | Q(component_id__in=user.component_permissions)
173 )
174 return result
176 def order(self):
177 return self.order_by(
178 "component__priority", "component__project__name", "component__name"
179 )
182class Translation(
183 models.Model,
184 URLMixin,
185 LoggerMixin,
186 CacheKeyMixin,
187 LockMixin,
188 TranslationChecklistMixin,
189):
190 component = models.ForeignKey(
191 "trans.Component", on_delete=models.deletion.CASCADE, db_index=False
192 )
193 language = models.ForeignKey(Language, on_delete=models.deletion.CASCADE)
194 plural = models.ForeignKey(Plural, on_delete=models.deletion.CASCADE)
195 revision = models.CharField(max_length=200, default="", blank=True)
196 filename = models.CharField(max_length=FILENAME_LENGTH)
198 language_code = models.CharField(max_length=50, default="", blank=True)
200 check_flags = models.TextField(
201 verbose_name="Translation flags",
202 default="",
203 validators=[validate_check_flags],
204 blank=True,
205 )
207 objects = TranslationManager.from_queryset(TranslationQuerySet)()
209 remove_permission = "translation.delete"
210 settings_permission = "component.edit"
212 class Meta:
213 app_label = "trans"
214 unique_together = [("component", "language")] # noqa: RUF012
215 verbose_name = "translation"
216 verbose_name_plural = "translations"
218 def __str__(self) -> str:
219 return f"{self.component} — {self.language}"
221 def __init__(self, *args, **kwargs) -> None:
222 super().__init__(*args, **kwargs)
223 self.stats = TranslationStats(self)
224 self.addon_commit_files: list[str] = []
225 self.reason = ""
226 self._invalidate_scheduled = False
227 self.update_changes: list[Change] = []
228 # Project backup integration
229 self.original_id = -1
231 self.create_unit_change_action = ActionEvents.NEW_UNIT_REPO
232 self.update_unit_change_action = ActionEvents.STRING_REPO_UPDATE
234 @property
235 def code(self):
236 return self.language.code
238 def log_hook(self, level, msg, *args) -> None:
239 self.component.store_log(self.full_slug, msg, *args)
241 @property
242 def project(self) -> Project:
243 return self.component.project
245 @cached_property
246 def is_template(self) -> bool:
247 """
248 Check whether this is template translation.
250 This means that translations should be propagated as sources to others.
251 """
252 return (
253 self.component.has_template() and self.filename == self.component.template
254 )
256 @cached_property
257 def is_source(self) -> bool:
258 """
259 Check whether this is source strings.
261 This means that translations should be propagated as sources to others.
262 """
263 return self.language_id == self.component.source_language_id
265 @cached_property
266 def all_flags(self):
267 """Return parsed list of flags."""
268 return Flags(self.component.all_flags, self.check_flags)
270 @cached_property
271 def is_readonly(self):
272 return "read-only" in self.all_flags
274 def clean(self) -> None:
275 """Validate that filename exists and can be opened using translate-toolkit."""
276 filename = self.get_filename()
277 if filename is None:
278 # Should not actually happen
279 msg = "Translation without a filename!"
280 raise ValidationError(msg)
281 if not os.path.exists(filename):
282 raise ValidationError(
283 gettext(
284 "Filename %s not found in repository! To add new "
285 "translation, add language file into repository."
286 )
287 % self.filename
288 )
289 try:
290 self.load_store()
291 except Exception as error:
292 raise ValidationError(
293 gettext("Could not parse file %(file)s: %(error)s")
294 % {"file": self.filename, "error": str(error)}
295 ) from error
297 def get_url_path(self):
298 return (*self.component.get_url_path(), self.language.code)
300 def get_widgets_url(self) -> str:
301 """Return absolute URL for widgets."""
302 return f"{self.component.project.get_widgets_url()}?lang={self.language.code}&component={self.component.pk}"
304 def get_share_url(self):
305 """Return absolute URL usable for sharing."""
306 return get_site_url(
307 reverse(
308 "engage",
309 kwargs={"path": [self.component.project.slug, "-", self.language.code]},
310 )
311 )
313 def get_translate_url(self):
314 return reverse("translate", kwargs={"path": self.get_url_path()})
316 def get_filename(self) -> str | None:
317 """Return absolute filename."""
318 if not self.filename:
319 return None
320 return os.path.join(self.component.full_path, self.filename)
322 def load_store(self, fileobj=None, force_intermediate=False):
323 """Load translate-toolkit storage from disk."""
324 # Use intermediate store as template for source translation
325 with sentry_sdk.start_span(
326 op="translation.load_store", name=self.get_filename()
327 ):
328 if force_intermediate or (self.is_template and self.component.intermediate): 328 ↛ 329line 328 didn't jump to line 329 because the condition on line 328 was never true
329 template = self.component.intermediate_store
330 else:
331 template = self.component.template_store
332 if fileobj is None: 332 ↛ 337line 332 didn't jump to line 337 because the condition on line 332 was always true
333 fileobj = self.get_filename()
334 if fileobj is None: 334 ↛ 335line 334 didn't jump to line 335 because the condition on line 334 was never true
335 msg = "Attempt to parse store without a filename."
336 raise ValueError(msg)
337 elif self.is_template:
338 template = self.component.load_template_store(
339 NamedBytesIO(fileobj.name, fileobj.read())
340 )
341 fileobj.seek(0)
342 return self.component.file_format_cls(
343 fileobj,
344 template,
345 language_code=self.language_code,
346 source_language=self.language_code
347 if self.component.has_template()
348 else self.component.source_language.code,
349 is_template=self.is_template,
350 existing_units=self.unit_set.all(),
351 file_format_params=self.component.file_format_params,
352 )
354 @cached_property
355 def store(self):
356 """Return translate-toolkit storage object for a translation."""
357 try:
358 return self.load_store()
359 except FileParseError:
360 raise
361 except Exception as exc:
362 report_error(
363 "Translation parse error", project=self.component.project, print_tb=True
364 )
365 self.component.handle_parse_error(exc, self)
367 def pre_process_unit(
368 self,
369 *,
370 dbunits: dict[int, Unit],
371 id_hash: int,
372 unit: TranslationUnit,
373 pos: int,
374 ) -> Unit:
375 try:
376 newunit = dbunits[id_hash]
377 is_new = False
378 except KeyError:
379 newunit = Unit(translation=self, id_hash=id_hash, state=-1)
380 # Avoid fetching empty list of checks from the database
381 newunit.all_checks = []
382 # Avoid fetching empty list of variants
383 newunit._prefetched_objects_cache = { # noqa: SLF001
384 "defined_variants": Variant.objects.none()
385 }
386 is_new = True
388 newunit.store_unit_attributes(
389 unit=unit, pos=pos, created=is_new, id_hash=id_hash
390 )
391 return newunit
393 def check_sync( # noqa: C901
394 self,
395 force: bool = False,
396 request: AuthenticatedHttpRequest | None = None,
397 change: int | None = None,
398 author: User | None = None,
399 ) -> None:
400 """Check whether database is in sync with git and possibly updates."""
401 with sentry_sdk.start_span(op="translation.check_sync", name=self.full_slug):
402 if change is None: 402 ↛ 404line 402 didn't jump to line 404 because the condition on line 402 was always true
403 change = ActionEvents.UPDATE
404 user = None if request is None else request.user
406 details = {
407 "filename": self.filename,
408 }
409 self.update_changes = []
411 # Check if we're not already up to date
412 try:
413 new_revision = self.get_git_blob_hash()
414 except FileNotFoundError:
415 self.reason = ""
416 return
417 except Exception as exc:
418 report_error("Translation parse error", project=self.component.project)
419 self.component.handle_parse_error(exc, self)
420 if not self.revision:
421 self.reason = "new file"
422 elif self.revision != new_revision: 422 ↛ 423line 422 didn't jump to line 423 because the condition on line 422 was never true
423 self.reason = "content changed"
425 # Include changed filename in the details
426 old_parts = self.revision.split(",")
427 new_parts = new_revision.split(",")
428 if len(old_parts) == len(new_parts):
429 filenames = self.get_hash_filenames()
430 for i, old_part in enumerate(old_parts):
431 if old_part != new_parts[i]:
432 details["filename"] = filenames[i][
433 len(self.component.full_path) :
434 ].lstrip("/")
435 break
437 elif force: 437 ↛ 440line 437 didn't jump to line 440 because the condition on line 437 was always true
438 self.reason = "check forced"
439 else:
440 self.reason = ""
441 return
442 details["reason"] = self.reason
444 self.component.check_template_valid()
446 # List of updated units (used for cleanup and duplicates detection)
447 updated: dict[int, Unit] = {}
449 try:
450 store = self.store
451 translation_store = None
453 try:
454 store_units = store.content_units
455 except ValueError as error:
456 raise FileParseError(str(error)) from error
458 self.log_info(
459 "processing %s, %s, %d strings",
460 self.filename,
461 self.reason,
462 len(store_units),
463 )
465 # Store plural
466 plural = store.get_plural(self.language, store)
467 if plural != self.plural: 467 ↛ 468line 467 didn't jump to line 468 because the condition on line 467 was never true
468 self.plural = plural
469 self.save(update_fields=["plural"])
471 # Select all current units for update
472 dbunits = {
473 unit.id_hash: unit
474 for unit in self.unit_set.prefetch_bulk().select_for_update()
475 }
476 duplicates: list[Unit] = []
478 # Process based on intermediate store if available
479 if self.component.intermediate: 479 ↛ 480line 479 didn't jump to line 480 because the condition on line 479 was never true
480 translation_store = store
481 store = self.load_store(force_intermediate=True)
482 try:
483 store_units = store.content_units
484 except ValueError as error:
485 raise FileParseError(str(error)) from error
487 for pos, unit in enumerate(store_units):
488 # Use translation store if exists and if it contains the string
489 if translation_store is not None: 489 ↛ 490line 489 didn't jump to line 490 because the condition on line 489 was never true
490 try:
491 translated_unit, created = translation_store.find_unit(
492 unit.context, unit.source
493 )
494 if translated_unit and not created:
495 unit = translated_unit
496 else:
497 # Patch unit to have matching source
498 unit.source = translated_unit.source
499 except UnitNotFoundError:
500 pass
501 if ( 501 ↛ 507line 501 didn't jump to line 507 because the condition on line 501 was never true
502 self.component.file_format_cls.monolingual
503 and self.component.key_filter_re
504 and self.component.key_filter_re.match(unit.context) is None
505 ):
506 # This is where the key filtering take place
507 self.log_info(
508 "Doesn't match with key_filter, skipping: %s (%s)",
509 unit.context,
510 repr(unit.source),
511 )
512 continue
514 try:
515 id_hash = unit.id_hash
516 except Exception as error:
517 self.component.handle_parse_error(error, self)
519 # Check for possible duplicate units
520 if id_hash in updated:
521 newunit = updated[id_hash]
522 self.log_warning(
523 "duplicate string to translate: %s (%s)",
524 newunit,
525 repr(newunit.source),
526 )
527 duplicates.append(newunit)
528 continue
530 # Collect source strings
531 newunit = self.pre_process_unit(
532 dbunits=dbunits,
533 id_hash=id_hash,
534 unit=unit,
535 pos=pos + 1,
536 )
538 # Store current unit ID
539 updated[id_hash] = newunit
541 # Create source strings
542 if not self.is_source and not self.component.template:
543 with sentry_sdk.start_span(
544 op="component.bulk_create_source", name=self.full_slug
545 ):
546 self.component.bulk_create_sources(
547 [
548 newunit.unit_attributes
549 for newunit in updated.values()
550 if newunit.unit_attributes is not None
551 ]
552 )
554 # Create/update translations
555 for newunit in updated.values():
556 with sentry_sdk.start_span(
557 op="unit.update_from_unit", name=f"{self.full_slug}:{pos}"
558 ):
559 newunit.update_from_unit(user=user, author=author)
561 # Trigger duplicate alerts
562 for newunit in duplicates:
563 self.component.trigger_alert(
564 "DuplicateString",
565 language_code=self.language.code,
566 source=newunit.source,
567 unit_pk=newunit.pk,
568 )
570 except FileParseError as error:
571 report_error(
572 "Could not parse file on update", project=self.component.project
573 )
574 self.log_warning("skipping update due to parse error: %s", error)
575 self.store_update_changes()
576 return
578 # Delete stale units
579 stale = set(dbunits) - set(updated)
580 if stale:
581 self.log_info("deleting %d stale strings", len(stale))
582 self.unit_set.filter(id_hash__in=stale).delete()
583 self.component.needs_cleanup = True
585 # We should also do cleanup on source strings tracking objects
587 # Update revision and stats
588 self.store_hash()
590 # Store change entry
591 self.update_changes.append(
592 Change(
593 translation=self,
594 action=change,
595 user=user,
596 author=author,
597 details=details,
598 )
599 )
601 self.store_update_changes()
603 # Invalidate keys cache
604 transaction.on_commit(self.invalidate_keys)
605 self.log_info("updating completed")
607 # Use up to date list as prefetch for source
608 if self.is_source:
609 self.component.preload_sources(updated)
611 # Unload the store, this is intentionally in a way that it would break
612 # further consumers as no further consumer is expected after this and
613 # we do not want to parse the file again.
614 self.__dict__["store"] = None
616 def store_update_changes(self) -> None:
617 # Save change
618 Change.objects.bulk_create(self.update_changes, batch_size=500)
619 self.update_changes.clear()
621 def do_update(
622 self, request: AuthenticatedHttpRequest | None = None, method: str | None = None
623 ):
624 return self.component.do_update(request, method=method)
626 def do_push(self, request: AuthenticatedHttpRequest | None = None):
627 return self.component.do_push(request)
629 def do_reset(
630 self,
631 request: AuthenticatedHttpRequest | None = None,
632 *,
633 keep_changes: bool = False,
634 ) -> bool:
635 return self.component.do_reset(request, keep_changes=keep_changes)
637 def do_cleanup(self, request: AuthenticatedHttpRequest | None = None):
638 return self.component.do_cleanup(request)
640 def do_file_sync(self, request: AuthenticatedHttpRequest | None = None):
641 return self.component.do_file_sync(request)
643 def do_file_scan(self, request: AuthenticatedHttpRequest | None = None):
644 return self.component.do_file_scan(request)
646 def can_push(self) -> bool:
647 return self.component.can_push()
649 def has_push_configuration(self) -> bool:
650 return self.component.has_push_configuration()
652 def get_hash_filenames(self) -> list[str]:
653 """Return filenames to include in the hash."""
654 component = self.component
655 filenames = [self.get_filename()]
657 if component.has_template(): 657 ↛ 659line 657 didn't jump to line 659 because the condition on line 657 was never true
658 # Include template
659 filenames.append(component.get_template_filename())
661 filename = component.get_intermediate_filename()
662 if component.intermediate and os.path.exists(filename):
663 # Include intermediate language as it might add new strings
664 filenames.append(filename)
666 return filenames
668 def get_git_blob_hash(self) -> str:
669 """Return current VCS blob hash for file."""
670 get_object_hash = self.component.repository.get_object_hash
671 filenames = self.get_hash_filenames()
672 return ",".join(get_object_hash(filename) for filename in filenames)
674 def store_hash(self) -> None:
675 """Store current hash in database."""
676 self.revision = self.get_git_blob_hash()
677 self.save(update_fields=["revision"])
679 def get_last_author(self):
680 """Return last author of change done in Weblate."""
681 if not self.stats.last_author:
682 return None
683 from weblate.auth.models import User
685 return User.objects.get(pk=self.stats.last_author).get_visible_name()
687 @transaction.atomic
688 def commit_pending(self, reason: str, user: User | None, skip_push: bool = False):
689 """Commit any pending changes."""
690 if not self.needs_commit(): 690 ↛ 692line 690 didn't jump to line 692 because the condition on line 690 was always true
691 return False
692 return self.component.commit_pending(reason, user, skip_push=skip_push)
694 @transaction.atomic
695 def _commit_pending(self, reason: str, user: User | None) -> bool:
696 """
697 Commit pending translation.
699 Assumptions:
701 - repository lock is held
702 - the source translation needs to be committed first
703 - signals and alerts are updated by the caller
704 - repository push is handled by the caller
705 """
706 try:
707 store = self.store
708 except FileParseError as error:
709 report_error(
710 "Could not parse file on commit", project=self.component.project
711 )
712 self.log_error("skipping commit due to error: %s", error)
713 return False
715 try:
716 store.ensure_index()
717 except ValueError as error:
718 report_error(
719 "Could not parse file on commit", project=self.component.project
720 )
721 self.log_error("skipping commit due to error: %s", error)
722 return False
724 pending_changes = list(
725 PendingUnitChange.objects.for_translation(self)
726 .prefetch_related("unit", "author")
727 .order_by("timestamp")
728 .select_for_update()
729 )
731 if not pending_changes: 731 ↛ 732line 731 didn't jump to line 732 because the condition on line 731 was never true
732 return False
734 self.log_info(
735 "committing %d pending changes (%s)", len(pending_changes), reason
736 )
738 all_changes_status = {}
739 units_to_clear_disk_state = set() # Track units for disk_state clearing
741 commit_groups = self._group_changes_by_author(pending_changes)
742 for author, changes in commit_groups:
743 author_name = author.get_author_name()
744 timestamp = max(change.timestamp for change in changes)
746 # Flush the grouped pending changes for this author
747 changes_status = self.update_units(changes, store, author_name)
748 all_changes_status.update(changes_status)
750 # Track successful units
751 for change in changes:
752 if changes_status.get(change.pk): 752 ↛ 751line 752 didn't jump to line 751 because the condition on line 752 was always true
753 units_to_clear_disk_state.add(change.unit_id)
755 # Commit changes if there was anything written out
756 if any(changes_status.values()): 756 ↛ 742line 756 didn't jump to line 742 because the condition on line 756 was always true
757 self.git_commit(
758 user, author_name, timestamp, skip_push=True, signals=False
759 )
761 # Short-circuit when no changes were processed
762 if not any(all_changes_status.values()): 762 ↛ 763line 762 didn't jump to line 763 because the condition on line 762 was never true
763 return False
765 # A pending change can be deleted from the database:
766 # 1. Always, if it has been applied successfully.
767 # 2. If it failed to apply, it can be removed only if a newer pending
768 # change targeting the same unit has since been successfully applied.
769 # In some cases, a previously failed change may be resolved by a subsequent
770 # change to the unit. Failed pending changes are not retried on every commit
771 # attempt and hence may be absent from the list of pending changes being
772 # currently processed. Hence, we track the latest timestamp up to which
773 # changes to a unit have been applied successfully and delete all pending
774 # changes for the unit until that timestamp.
775 success_times = {}
776 for change in pending_changes:
777 if all_changes_status.get(change.pk): 777 ↛ 776line 777 didn't jump to line 776 because the condition on line 777 was always true
778 prev = success_times.get(change.unit_id)
779 if prev is None or change.timestamp > prev: 779 ↛ 776line 779 didn't jump to line 776 because the condition on line 779 was always true
780 success_times[change.unit_id] = change.timestamp
782 for unit_id, ts in success_times.items():
783 PendingUnitChange.objects.filter(
784 unit_id=unit_id, timestamp__lte=ts
785 ).delete()
787 # Clear disk states for units that have no more pending changes
788 units_with_remaining_changes = set(
789 PendingUnitChange.objects.filter(
790 unit_id__in=units_to_clear_disk_state
791 ).values_list("unit_id", flat=True)
792 )
794 units_to_actually_clear = (
795 units_to_clear_disk_state - units_with_remaining_changes
796 )
798 if units_to_actually_clear:
799 units_to_update = list(
800 Unit.objects.filter(
801 id__in=units_to_actually_clear, details__has_key="disk_state"
802 ).select_for_update()
803 )
805 for unit in units_to_update:
806 del unit.details["disk_state"]
808 if units_to_update: 808 ↛ 812line 808 didn't jump to line 812 because the condition on line 808 was always true
809 Unit.objects.bulk_update(units_to_update, ["details"], batch_size=500)
811 # Update stats (the translated flag might have changed)
812 self.invalidate_cache()
814 # Make sure template cache is purged upon commit
815 self.drop_store_cache()
817 return True
819 @staticmethod
820 def _group_changes_by_author(
821 pending_changes: list[PendingUnitChange],
822 ) -> list[tuple[User, list[PendingUnitChange]]]:
823 """
824 Organize pending changes into groups where each group represents the set of changes that can be safely committed together.
826 Assuming pending_changes is sorted by timestamp in ascending order, the changes are
827 grouped according to the following logic:
829 1. Maintain active groups per author in a dict ({author_id: {"changes": [...], "units": {...}}})
830 2. For each pending change, check if the unit conflicts with ANY active group EXCEPT the change author's own group.
831 3. If conflict found: finalize conflicting group(s)
832 4. Add change to the author's existing group or create a new one if needed
834 Example:
835 Consider the following changes were made in chronological order:
837 - A1: Author 1 edits Unit A -> Group 1: Author1
838 - A2: Author 2 edits Unit A -> Unit A conflicts -> Finalize Group 1, start Group 2: Author2
839 - B3: Author 1 edits Unit B -> No conflicts -> Start Group 3: Author1
840 - C4: Author 2 edits Unit C -> No conflicts -> Add to Group 2: Author2
841 - B5: Author 1 edits Unit B -> conflicts with own change -> Do not break group, Add to Group 3: Author1
842 - D6: Author 1 edits Unit D -> No conflicts -> Add to Group 3: Author1
844 Result:
845 - Commit 1: Author1 [A1]
846 - Commit 2: Author2 [A2, C4]
847 - Commit 3: Author1 [B3, B5, D6]
849 """
850 author_map = {
851 change.author.id: change.author
852 for change in pending_changes
853 if change.author
854 }
856 active_groups = {}
857 finalized_groups = []
859 for change in pending_changes:
860 unit_id = change.unit.id
861 author_id = change.author.id
863 conflicting_authors = []
864 for other_author_id, group_data in active_groups.items():
865 if unit_id in group_data["units"] and author_id != other_author_id: 865 ↛ 866line 865 didn't jump to line 866 because the condition on line 865 was never true
866 conflicting_authors.append(other_author_id)
868 for conflicting_author_id in conflicting_authors: 868 ↛ 869line 868 didn't jump to line 869 because the loop on line 868 never started
869 group_data = active_groups.pop(conflicting_author_id)
870 author = author_map[conflicting_author_id]
871 finalized_groups.append((author, group_data["changes"]))
873 if author_id in active_groups:
874 active_groups[author_id]["changes"].append(change)
875 active_groups[author_id]["units"].add(unit_id)
876 else:
877 active_groups[author_id] = {"changes": [change], "units": {unit_id}}
879 for author_id, group_data in active_groups.items():
880 author = author_map[author_id]
881 finalized_groups.append((author, group_data["changes"]))
883 return finalized_groups
885 def get_commit_message(self, author: str, template: str, **kwargs):
886 """Format commit message based on project configuration."""
887 return render_template(template, translation=self, author=author, **kwargs)
889 @property
890 def count_pending_units(self):
891 """Return count of units with pending changes."""
892 return PendingUnitChange.objects.for_translation(self).count()
894 def needs_commit(self):
895 """Check whether there are some not committed changes."""
896 return self.count_pending_units > 0
898 def repo_needs_merge(self):
899 return self.component.repo_needs_merge()
901 def repo_needs_push(self):
902 return self.component.repo_needs_push()
904 @cached_property
905 def filenames(self) -> list[str]:
906 if not self.filename: 906 ↛ 907line 906 didn't jump to line 907 because the condition on line 906 was never true
907 return []
908 if self.component.file_format_cls.simple_filename: 908 ↛ 910line 908 didn't jump to line 910 because the condition on line 908 was always true
909 return [self.get_filename()]
910 return self.store.get_filenames()
912 def git_commit(
913 self,
914 user: User | None,
915 author: str,
916 timestamp: datetime | None = None,
917 skip_push=False,
918 signals=True,
919 template: str | None = None,
920 store_hash: bool = True,
921 ) -> bool:
922 """Commit translation to git."""
923 repository = self.component.repository
924 if template is None:
925 template = self.component.commit_message
926 with repository.lock:
927 # Pre commit hook
928 vcs_pre_commit.send(
929 sender=self.__class__,
930 translation=self,
931 author=author,
932 store_hash=store_hash,
933 )
935 # Do actual commit with git lock
936 if self.component.commit_files(
937 template=template,
938 author=author,
939 timestamp=timestamp,
940 skip_push=skip_push,
941 signals=signals,
942 files=self.filenames + self.addon_commit_files,
943 extra_context={"translation": self},
944 store_hash=store_hash,
945 ):
946 self.log_info("committed %s as %s", self.filenames, author)
947 self.change_set.create(
948 action=ActionEvents.COMMIT, user=user, author=user
949 )
951 # Store updated hash
952 if store_hash:
953 self.store_hash()
954 self.addon_commit_files = []
956 return True
958 def update_units(
959 self,
960 pending_changes: list[PendingUnitChange],
961 store: TranslationFormat,
962 author_name: str,
963 ) -> dict[int, bool]:
964 """Update backend file and unit."""
965 changes_status = {}
966 updated = False
967 for pending_change in pending_changes:
968 unit = pending_change.unit
970 # update the unit in memory so that get_target_plurals returns
971 # the correct result for this change
972 unit.target = pending_change.target
974 if pending_change.add_unit: 974 ↛ 975line 974 didn't jump to line 975 because the condition on line 974 was never true
975 pounit = store.new_unit(
976 unit.context, unit.get_source_plurals(), unit.get_target_plurals()
977 )
978 pounit.set_explanation(pending_change.explanation)
979 pounit.set_source_explanation(pending_change.source_unit_explanation)
980 # Check if context has changed while adding to storage
981 if pounit.context != unit.context:
982 if self.is_source:
983 # Update all matching translations
984 Unit.objects.filter(
985 context=unit.context, translation__component=self.component
986 ).update(context=pounit.context)
987 else:
988 # Update this unit only
989 Unit.objects.filter(pk=unit.pk).update(context=pounit.context)
990 unit.context = pounit.context
992 changes_status[pending_change.pk] = True
993 updated = True
994 else:
995 try:
996 pounit, add = store.find_unit(unit.context, unit.source)
997 except UnitNotFoundError:
998 # Bail out if we have not found anything
999 report_error(
1000 "String disappeared",
1001 project=self.component.project,
1002 skip_sentry=True,
1003 )
1004 unit.state = STATE_FUZZY
1005 # Use update instead of hitting expensive save()
1006 Unit.objects.filter(pk=unit.pk).update(state=STATE_FUZZY)
1007 unit.change_set.create(
1008 action=ActionEvents.SAVE_FAILED,
1009 target="Could not find string in the translation file",
1010 )
1011 pending_change.metadata.update(
1012 {
1013 "last_failed": timezone.now().isoformat(),
1014 "failed_revision": self.revision,
1015 "weblate_version": GIT_VERSION,
1016 "blocking_unit": True,
1017 }
1018 )
1019 pending_change.save()
1020 # this should be kept as pending, so that the changes are not lost
1021 changes_status[pending_change.pk] = False
1022 continue
1024 # Optionally add unit to translation file.
1025 # This has be done prior setting target as some formats
1026 # generate content based on target language.
1027 if add: 1027 ↛ 1028line 1027 didn't jump to line 1028 because the condition on line 1027 was never true
1028 store.add_unit(pounit)
1030 # Store translations
1031 try:
1032 if unit.is_plural: 1032 ↛ 1033line 1032 didn't jump to line 1033 because the condition on line 1032 was never true
1033 pounit.set_target(unit.get_target_plurals())
1034 else:
1035 pounit.set_target(pending_change.target)
1036 pounit.set_explanation(pending_change.explanation)
1037 pounit.set_source_explanation(
1038 pending_change.source_unit_explanation
1039 )
1040 except Exception as error:
1041 report_error(
1042 "Could not update unit", project=self.component.project
1043 )
1044 unit.state = STATE_FUZZY
1045 # Use update instead of hitting expensive save()
1046 Unit.objects.filter(pk=unit.pk).update(state=STATE_FUZZY)
1047 unit.change_set.create(
1048 action=ActionEvents.SAVE_FAILED,
1049 target=self.component.get_parse_error_message(error),
1050 )
1052 pending_change.metadata.update(
1053 {
1054 "last_failed": timezone.now().isoformat(),
1055 "failed_revision": self.revision,
1056 "weblate_version": GIT_VERSION,
1057 "blocking_unit": False,
1058 }
1059 )
1060 pending_change.save()
1061 # this should be kept as pending, so that the changes are not lost
1062 changes_status[pending_change.pk] = False
1063 continue
1065 changes_status[pending_change.pk] = True
1066 updated = True
1068 # Update fuzzy/approved flag
1069 pounit.set_state(pending_change.state)
1071 # Update disk state from the file to the unit
1072 unit.details["disk_state"] = {
1073 "target": pounit.target,
1074 "state": pending_change.state,
1075 "explanation": pounit.explanation,
1076 }
1077 unit.save(update_fields=["details"])
1079 # Did we do any updates?
1080 if not updated: 1080 ↛ 1081line 1080 didn't jump to line 1081 because the condition on line 1080 was never true
1081 return changes_status
1083 # Update po file header
1084 now = timezone.now()
1085 if not timezone.is_aware(now): 1085 ↛ 1086line 1085 didn't jump to line 1086 because the condition on line 1085 was never true
1086 now = timezone.make_aware(now, UTC)
1088 # Prepare headers to update
1089 headers = {
1090 "last_translator": author_name,
1091 "plural_forms": self.plural.plural_form,
1092 "language": self.language_code,
1093 "PO_Revision_Date": now.strftime("%Y-%m-%d %H:%M%z"),
1094 }
1096 # Optionally store language team with link to website
1097 if self.component.project.set_language_team:
1098 headers["language_team"] = (
1099 f"{self.language.name} <{get_site_url(self.get_absolute_url())}>"
1100 )
1102 # Optionally store email for reporting bugs in source
1103 report_source_bugs = self.component.report_source_bugs
1104 if report_source_bugs:
1105 headers["report_msgid_bugs_to"] = report_source_bugs
1107 # Update generic headers
1108 store.update_header(**headers)
1110 # save translation changes
1111 store.save()
1113 return changes_status
1115 @cached_property
1116 def workflow_settings(self):
1117 return self.component.project.project_languages[self.language].workflow_settings
1119 @cached_property
1120 def enable_review(self):
1121 project = self.component.project
1122 project_review = (
1123 project.source_review if self.is_source else project.translation_review
1124 )
1125 if not project_review: 1125 ↛ 1127line 1125 didn't jump to line 1127 because the condition on line 1125 was always true
1126 return False
1127 if self.workflow_settings is not None:
1128 return self.workflow_settings.translation_review
1129 return project_review
1131 @property
1132 def enable_suggestions(self):
1133 if self.workflow_settings is not None: 1133 ↛ 1134line 1133 didn't jump to line 1134 because the condition on line 1133 was never true
1134 return self.workflow_settings.enable_suggestions
1135 return self.component.enable_suggestions
1137 @property
1138 def suggestion_voting(self):
1139 if self.workflow_settings is not None: 1139 ↛ 1140line 1139 didn't jump to line 1140 because the condition on line 1139 was never true
1140 return self.workflow_settings.suggestion_voting
1141 return self.component.suggestion_voting
1143 @property
1144 def suggestion_autoaccept(self):
1145 if self.workflow_settings is not None:
1146 return self.workflow_settings.suggestion_autoaccept
1147 return self.component.suggestion_autoaccept
1149 def log_upload_not_found(
1150 self, not_found_log: list[str], unit: TranslationUnit
1151 ) -> None:
1152 not_found_log.append(unit.source)
1154 def show_upload_not_found(
1155 self,
1156 request: AuthenticatedHttpRequest,
1157 not_found_log: list[str],
1158 string_limit: int = 8,
1159 ) -> None:
1160 count = len(not_found_log)
1162 strings = format_html_join_comma(
1163 gettext("“{}”"), ((string,) for string in not_found_log[:string_limit])
1164 )
1165 if count > string_limit:
1166 strings = format_html_join_comma("{}", [(strings,), (gettext("…"),)])
1168 messages.warning(
1169 request,
1170 format_html(
1171 "{} {}",
1172 ngettext(
1173 "Could not find %d string:", "Could not find %d strings:", count
1174 )
1175 % count,
1176 strings,
1177 ),
1178 )
1180 def merge_translations(
1181 self,
1182 request: AuthenticatedHttpRequest,
1183 author: User,
1184 store2: TranslationFormat,
1185 conflicts: Literal["", "replace-approved", "replace-translated"],
1186 method: Literal["fuzzy", "approve", "translate"],
1187 fuzzy: Literal["", "process", "approve"],
1188 ):
1189 """
1190 Merge translation unit wise.
1192 Needed for template based translations to add new strings.
1193 """
1194 skipped = 0
1195 accepted = 0
1196 add_fuzzy = method == "fuzzy"
1197 add_approve = method == "approve"
1198 not_found_log: list[str] = []
1200 # Are there any translations to propagate?
1201 # This is just an optimalization to avoid doing that for every unit.
1202 propagate = (
1203 Translation.objects.filter(
1204 language=self.language,
1205 component__source_language_id=self.component.source_language_id,
1206 component__project=self.component.project,
1207 )
1208 .filter(component__allow_translation_propagation=True)
1209 .exclude(pk=self.pk)
1210 .exists()
1211 )
1213 unit_set = self.unit_set.select_for_update()
1215 for set_fuzzy, unit2 in store2.iterate_merge(fuzzy):
1216 try:
1217 unit = unit_set.get_unit(unit2)
1218 except Unit.DoesNotExist:
1219 self.log_upload_not_found(not_found_log, unit2)
1220 continue
1222 state = STATE_TRANSLATED
1223 if add_fuzzy or set_fuzzy:
1224 state = STATE_FUZZY
1225 elif add_approve:
1226 state = STATE_APPROVED
1228 if (
1229 (unit.translated and not conflicts)
1230 or (unit.approved and conflicts != "replace-approved")
1231 or unit.readonly
1232 or (not request.user.has_perm("unit.edit", unit))
1233 or (unit.target == unit2.target and unit.state == state)
1234 ):
1235 skipped += 1
1236 continue
1238 accepted += 1
1240 unit.translate(
1241 request.user,
1242 split_plural(unit2.target),
1243 state,
1244 change_action=ActionEvents.UPLOAD,
1245 propagate=propagate,
1246 author=author,
1247 request=request,
1248 )
1250 if accepted > 0:
1251 self.invalidate_cache()
1252 request.user.profile.increase_count("translated", accepted)
1254 if not_found_log:
1255 self.show_upload_not_found(request, not_found_log)
1257 return (len(not_found_log), skipped, accepted, len(store2.content_units))
1259 def merge_suggestions(
1260 self, request: AuthenticatedHttpRequest, author: User, store, fuzzy
1261 ):
1262 """Merge content of translate-toolkit store as a suggestions."""
1263 skipped = 0
1264 accepted = 0
1265 not_found_log: list[str] = []
1267 unit_set = self.unit_set.all()
1269 for _unused, unit in store.iterate_merge(fuzzy):
1270 # Grab database unit
1271 try:
1272 dbunit = unit_set.get_unit(unit)
1273 except Unit.DoesNotExist:
1274 self.log_upload_not_found(not_found_log, unit)
1275 continue
1277 # Add suggestion
1278 current_target = dbunit.get_target_plurals()
1279 new_target = unit.target
1280 if isinstance(new_target, str):
1281 new_target = [new_target]
1282 if current_target != new_target and not dbunit.readonly:
1283 if Suggestion.objects.add(
1284 dbunit,
1285 new_target,
1286 request,
1287 raise_exception=False,
1288 user=author,
1289 ):
1290 accepted += 1
1291 else:
1292 skipped += 1
1293 else:
1294 skipped += 1
1296 # Update suggestion count
1297 if accepted > 0:
1298 self.invalidate_cache()
1300 if not_found_log:
1301 self.show_upload_not_found(request, not_found_log)
1303 return (len(not_found_log), skipped, accepted, len(store.content_units))
1305 def drop_store_cache(self) -> None:
1306 if "store" in self.__dict__: 1306 ↛ 1308line 1306 didn't jump to line 1308 because the condition on line 1306 was always true
1307 del self.__dict__["store"]
1308 if self.is_source:
1309 self.component.drop_template_store_cache()
1311 def handle_upload_store_change(
1312 self,
1313 request: AuthenticatedHttpRequest,
1314 author: User,
1315 change_action: ActionEvents,
1316 ) -> None:
1317 component = self.component
1318 if not component.repository.needs_commit(self.filenames):
1319 return
1321 self.create_unit_change_action = ActionEvents.NEW_UNIT_UPLOAD
1322 self.update_unit_change_action = ActionEvents.STRING_UPLOAD_UPDATE
1323 self.component.acting_user = author
1325 previous_revision = component.repository.last_revision
1326 self.drop_store_cache()
1328 self.git_commit(
1329 author,
1330 author=author.get_author_name(),
1331 store_hash=False,
1332 signals=False,
1333 )
1335 self.handle_store_change(
1336 request, author, previous_revision, change=change_action
1337 )
1338 # Emit signals later to avoid cleanup add-on to store translation
1339 # revision before parsing
1340 component.send_post_commit_signal()
1342 self.create_unit_change_action = ActionEvents.NEW_UNIT_REPO
1343 self.update_unit_change_action = ActionEvents.STRING_REPO_UPDATE
1345 def handle_source(
1346 self, request: AuthenticatedHttpRequest, author: User, fileobj: BinaryIO
1347 ) -> UploadResult:
1348 """Replace source translations with uploaded one."""
1349 component = self.component
1350 filenames = []
1351 with component.repository.lock:
1352 # Commit pending changes
1353 try:
1354 component.commit_pending("source update", author)
1355 except Exception as error:
1356 raise FailedCommitError(
1357 gettext("Could not commit pending changes: %s")
1358 % str(error).replace(self.component.full_path, "")
1359 ) from error
1361 # Create actual file with the uploaded content
1362 with tempfile.NamedTemporaryFile(
1363 prefix="weblate-upload", dir=self.component.full_path, delete=False
1364 ) as temp:
1365 temp.write(fileobj.read())
1367 try:
1368 # Prepare msgmerge args based on add-ons (if configured)
1369 args = self.component.file_format_cls.get_msgmerge_args(component)
1370 # Update translation files
1371 for translation in component.translation_set.exclude(
1372 language=component.source_language
1373 ):
1374 filename = translation.get_filename()
1375 component.file_format_cls.update_bilingual(
1376 filename, temp.name, args=args
1377 )
1378 filenames.append(filename)
1379 finally:
1380 if os.path.exists(temp.name):
1381 if component.new_base:
1382 filename = component.get_new_base_filename()
1383 os.replace(temp.name, filename)
1384 filenames.append(filename)
1385 else:
1386 os.unlink(temp.name)
1388 # Commit changes
1389 self.handle_upload_store_change(
1390 request, author, change_action=ActionEvents.SOURCE_UPLOAD
1391 )
1392 return (0, 0, self.unit_set.count(), self.unit_set.count())
1394 def handle_replace(
1395 self, request: AuthenticatedHttpRequest, author: User, fileobj: BinaryIO
1396 ) -> UploadResult:
1397 """Replace file content with uploaded one."""
1398 filecopy = fileobj.read()
1399 fileobj.close()
1400 fileobj = NamedBytesIO(fileobj.name, filecopy)
1401 self.unit_set.select_for_update()
1402 with self.component.repository.lock:
1403 try:
1404 if self.is_source:
1405 self.component.commit_pending("replace file", author)
1406 else:
1407 self.commit_pending("replace file", author)
1408 except Exception as error:
1409 raise FailedCommitError(
1410 gettext("Could not commit pending changes: %s")
1411 % str(error).replace(self.component.full_path, "")
1412 ) from error
1413 # This will throw an exception in case of error
1414 store2 = self.load_store(fileobj)
1415 store2.check_valid()
1417 # Actually replace file content
1418 self.component.file_format_cls.save_atomic(
1419 self.get_filename(), lambda handle: handle.write(filecopy)
1420 )
1422 # Commit to VCS
1423 self.handle_upload_store_change(
1424 request, author, change_action=ActionEvents.REPLACE_UPLOAD
1425 )
1427 return (0, 0, self.unit_set.count(), len(store2.content_units))
1429 def handle_add_upload(
1430 self,
1431 request: AuthenticatedHttpRequest,
1432 author: User,
1433 store: TranslationFormat,
1434 fuzzy: Literal["", "process", "approve"] = "",
1435 ) -> UploadResult:
1436 component = self.component
1437 has_template = component.has_template()
1438 skipped = 0
1439 accepted = 0
1440 component.start_batched_checks()
1441 existing: set[str] | set[tuple[str, str]]
1442 if has_template:
1443 existing = set(self.unit_set.values_list("context", flat=True))
1444 else:
1445 existing = set(self.unit_set.values_list("context", "source"))
1446 for _set_fuzzy, unit in store.iterate_merge(fuzzy, only_translated=False):
1447 idkey = unit.context if has_template else (unit.context, unit.source)
1448 if idkey in existing:
1449 skipped += 1
1450 continue
1451 self.add_unit(
1452 request,
1453 unit.context,
1454 split_plural(unit.source),
1455 split_plural(unit.target) if not self.is_source else [],
1456 is_batch_update=True,
1457 state=STATE_READONLY if unit.is_readonly() else None,
1458 )
1459 existing.add(idkey)
1460 accepted += 1
1461 self.store_update_changes()
1462 component.invalidate_cache()
1463 if component.needs_variants_update:
1464 component.update_variants()
1465 component.schedule_sync_terminology()
1466 component.update_source_checks()
1467 component.run_batched_checks()
1468 component_post_update.send(sender=self.__class__, component=component)
1469 return (0, skipped, accepted, len(store.content_units))
1471 def load_uploaded_file(
1472 self,
1473 request: AuthenticatedHttpRequest,
1474 fileobj: BinaryIO,
1475 method: Literal["fuzzy", "approve", "translate"],
1476 ) -> TranslationFormat:
1477 component = self.component
1479 filecopy = fileobj.read()
1480 fileobj.close()
1482 # Strip possible UTF-8 BOM
1483 if filecopy[:3] == codecs.BOM_UTF8:
1484 filecopy = filecopy[3:]
1486 # Commit pending changes in template
1487 if component.has_template() and component.source_translation.needs_commit():
1488 try:
1489 component.commit_pending("upload", request.user)
1490 except Exception as error:
1491 raise FailedCommitError(
1492 gettext("Could not commit pending changes: %s")
1493 % str(error).replace(self.component.full_path, "")
1494 ) from error
1496 # Load backend file
1497 if method == "add" and self.is_template:
1498 template_store = try_load(
1499 fileobj.name,
1500 filecopy,
1501 component.file_format_cls,
1502 None,
1503 is_template=True,
1504 language_code=self.language_code,
1505 source_language=self.component.source_language.code,
1506 file_format_params=self.component.file_format_params,
1507 )
1509 else:
1510 template_store = component.template_store
1511 store = try_load(
1512 fileobj.name,
1513 filecopy,
1514 component.file_format_cls,
1515 template_store,
1516 language_code=self.language_code,
1517 source_language=self.component.source_language.code,
1518 file_format_params=self.component.file_format_params,
1519 )
1521 # Check valid plural forms
1522 if hasattr(store.store, "parseheader"):
1523 header = store.store.parseheader()
1524 try:
1525 number, formula = Plural.parse_plural_forms(header["Plural-Forms"])
1526 except (ValueError, KeyError):
1527 # Formula wrong or missing
1528 pass
1529 else:
1530 if not self.plural.same_plural(number, formula):
1531 raise PluralFormsMismatchError
1532 return store
1534 @transaction.atomic
1535 def handle_upload(
1536 self,
1537 request: AuthenticatedHttpRequest,
1538 fileobj: BinaryIO,
1539 conflicts: Literal["", "replace-approved", "replace-translated"],
1540 author_name: str | None = None,
1541 author_email: str | None = None,
1542 method: Literal["fuzzy", "approve", "translate"] = "translate",
1543 fuzzy: Literal["", "process", "approve"] = "",
1544 ) -> UploadResult:
1545 """Top level handler for file uploads."""
1546 from weblate.auth.models import User
1548 component = self.component
1550 # Get User object for author
1551 author = User.objects.get_author_by_email(
1552 author_name, author_email, request.user, request
1553 )
1554 result: UploadResult
1556 if method == "replace":
1557 result = self.handle_replace(request, author, fileobj)
1559 elif method == "source":
1560 result = self.handle_source(request, author, fileobj)
1561 else:
1562 store = self.load_uploaded_file(request, fileobj, method)
1564 if method in {"translate", "fuzzy", "approve"}:
1565 # Merge on units level
1566 result = self.merge_translations(
1567 request, author, store, conflicts, method, fuzzy
1568 )
1569 elif method == "add":
1570 with component.lock:
1571 result = self.handle_add_upload(request, author, store, fuzzy=fuzzy)
1572 else:
1573 # Add as suggestions
1574 result = self.merge_suggestions(request, author, store, fuzzy)
1576 self.change_set.create(
1577 action=ActionEvents.FILE_UPLOAD,
1578 user=request.user,
1579 author=author,
1580 details={
1581 "method": method,
1582 "not_found": result[0],
1583 "skipped": result[1],
1584 "accepted": result[2],
1585 "total": result[3],
1586 },
1587 )
1589 return result
1591 def _invalidate_trigger(self) -> None:
1592 self._invalidate_scheduled = False
1593 self.stats.update_stats()
1594 self.component.invalidate_glossary_cache()
1596 def invalidate_cache(self) -> None:
1597 """Invalidate any cached stats."""
1598 # Invalidate summary stats
1599 if self._invalidate_scheduled:
1600 return
1601 self._invalidate_scheduled = True
1602 transaction.on_commit(self._invalidate_trigger)
1604 def detect_completed_translation(self, change: Change, old_translated: int) -> None:
1605 translated = self.stats.translated
1606 if old_translated < translated and translated == self.stats.all:
1607 self.change_set.create(
1608 action=ActionEvents.COMPLETE,
1609 user=change.user,
1610 author=change.author,
1611 )
1613 # check if component is fully translated
1614 component = self.component
1615 if component.stats.translated == component.stats.all:
1616 self.component.change_set.create(
1617 action=ActionEvents.COMPLETED_COMPONENT,
1618 user=change.user,
1619 author=change.author,
1620 )
1622 @property
1623 def keys_cache_key(self) -> str:
1624 return f"translation-keys-{self.pk}"
1626 def invalidate_keys(self) -> None:
1627 cache.delete(self.keys_cache_key)
1629 def get_export_url(self):
1630 """Return URL of exported git repository."""
1631 return self.component.get_export_url()
1633 def remove(self, user: User) -> None:
1634 """Remove translation from the Database and VCS."""
1635 from weblate.glossary.tasks import cleanup_stale_glossaries
1637 author = user.get_author_name()
1638 # Log
1639 self.log_info("removing %s as %s", self.filenames, author)
1641 # Remove file from VCS
1642 if any(os.path.exists(name) for name in self.filenames):
1643 with self.component.repository.lock:
1644 self.component.repository.remove(
1645 self.filenames,
1646 self.get_commit_message(
1647 author, template=self.component.delete_message
1648 ),
1649 author,
1650 )
1651 self.component.push_if_needed()
1653 # Remove blank directory if still present (appstore)
1654 filename = Path(self.get_filename())
1655 if filename.is_dir():
1656 with contextlib.suppress(OSError):
1657 filename.rmdir()
1659 # Delete from the database
1660 self.delete()
1661 transaction.on_commit(self.stats.update_parents)
1662 transaction.on_commit(self.component.schedule_update_checks)
1664 # Record change
1665 self.component.change_set.create(
1666 action=ActionEvents.REMOVE_TRANSLATION,
1667 target=self.filename,
1668 user=user,
1669 author=user,
1670 )
1671 if not self.component.is_glossary:
1672 cleanup_stale_glossaries.delay_on_commit(self.component.project.id)
1674 def handle_store_change(
1675 self,
1676 request: AuthenticatedHttpRequest | None,
1677 user: User,
1678 previous_revision: str,
1679 change=None,
1680 ) -> None:
1681 self.drop_store_cache()
1682 # Explicit stats invalidation is needed here as the unit removal in
1683 # delete_unit might do changes in the database only and not touch the files
1684 # for pending new units
1685 if self.is_source:
1686 self.component.create_translations(request=request, change=change)
1687 self.component.invalidate_cache()
1688 else:
1689 self.check_sync(request=request, change=change, author=user)
1690 self.invalidate_cache()
1691 # Trigger post-update signal
1692 self.component.trigger_post_update(
1693 previous_head=previous_revision,
1694 skip_push=False,
1695 )
1697 def get_store_change_translations(self) -> list[Translation]:
1698 component = self.component
1699 result: list[Translation] = []
1700 if self.is_source:
1701 result.extend(component.translation_set.exclude(id=self.id))
1702 # Source is always at the end
1703 result.append(self)
1704 return result
1706 @overload
1707 def add_unit( 1707 ↛ anywhereline 1707 didn't jump anywhere: it always raised an exception.
1708 self,
1709 request: AuthenticatedHttpRequest,
1710 context: str,
1711 source: str | list[str],
1712 target: str | list[str] | None = None,
1713 *,
1714 extra_flags: str = "",
1715 explanation: str = "",
1716 auto_context: bool = False,
1717 is_batch_update: bool = False,
1718 skip_existing: bool = False,
1719 state: StringState | None = None,
1720 author: User | None = None,
1721 ) -> Unit | None: ...
1722 @overload
1723 def add_unit( 1723 ↛ anywhereline 1723 didn't jump anywhere: it always raised an exception.
1724 self,
1725 request: None,
1726 context: str,
1727 source: str | list[str],
1728 target: str | list[str] | None = None,
1729 *,
1730 extra_flags: str = "",
1731 explanation: str = "",
1732 auto_context: bool = False,
1733 is_batch_update: bool = False,
1734 skip_existing: bool = False,
1735 state: StringState | None = None,
1736 author: User,
1737 ) -> Unit | None: ...
1738 @transaction.atomic
1739 def add_unit( # noqa: C901, PLR0914, PLR0915, PLR0912
1740 self,
1741 request,
1742 context,
1743 source,
1744 target=None,
1745 *,
1746 extra_flags="",
1747 explanation="",
1748 auto_context=False,
1749 is_batch_update=False,
1750 skip_existing=False,
1751 state=None,
1752 author=None,
1753 ):
1754 if isinstance(source, list):
1755 source = join_plural(source)
1757 parsed_flags = Flags(extra_flags)
1759 user = request.user if request else author
1760 if user is None:
1761 msg = "Can not add unit without an author!"
1762 raise ValueError(msg)
1763 component = self.component
1764 add_terminology = False
1765 if is_plural(source) and not component.file_format_cls.supports_plural:
1766 msg = "Plurals not supported by format!"
1767 raise ValueError(msg)
1769 component_wide = True
1770 if self.is_source:
1771 translations = (
1772 self,
1773 *component.translation_set.exclude(id=self.id).select_related(
1774 "language"
1775 ),
1776 )
1777 elif component.is_glossary and "terminology" in parsed_flags:
1778 add_terminology = True
1779 translations = (
1780 component.source_translation,
1781 *component.translation_set.exclude(
1782 id=component.source_translation.id
1783 ).select_related("language"),
1784 )
1785 else:
1786 component_wide = False
1787 translations = (component.source_translation, self)
1788 has_template = component.has_template()
1789 source_unit = None
1790 result = None
1792 # Automatic context
1793 if auto_context:
1794 suffix = 0
1795 base = context
1796 filter_args = {"source": source} if not has_template else {}
1797 while self.unit_set.filter(context=context, **filter_args).exists():
1798 suffix += 1
1799 context = f"{base}{suffix}"
1801 unit_ids = []
1802 changes = []
1803 for translation in translations:
1804 is_source = translation.is_source
1805 kwargs = {}
1806 pending = is_source if has_template else not is_source
1808 if (self.is_source and is_source) or (not self.is_source and not is_source):
1809 kwargs["explanation"] = explanation
1810 if is_source:
1811 current_target = source
1812 kwargs["extra_flags"] = extra_flags
1813 elif add_terminology and translation != self:
1814 current_target = ""
1815 else:
1816 current_target = target
1817 if current_target is None:
1818 current_target = ""
1819 # Wipe target for untranslatable strings
1820 if component.is_glossary and "read-only" in parsed_flags:
1821 current_target = ""
1822 if isinstance(current_target, list):
1823 has_translation = any(current_target)
1824 current_target = join_plural(current_target)
1825 else:
1826 has_translation = bool(current_target)
1827 id_hash = component.file_format_cls.unit_class.calculate_id_hash(
1828 has_template, source, context
1829 )
1830 # When adding to a target the source string can already exist
1831 unit = None
1832 if (skip_existing or not self.is_source) and is_source:
1833 try:
1834 unit = component.get_source(id_hash)
1835 flags = Flags(unit.extra_flags)
1836 flags.merge(extra_flags)
1837 new_flags = flags.format()
1838 if not skip_existing and (
1839 unit.extra_flags != new_flags or unit.explanation != explanation
1840 ):
1841 unit.extra_flags = new_flags
1842 unit.explanation = explanation
1843 unit.save(
1844 update_fields=["extra_flags", "explanation"],
1845 same_content=True,
1846 sync_terminology=False,
1847 )
1848 except Unit.DoesNotExist:
1849 pass
1850 if unit is None:
1851 if "read-only" in translation.all_flags or (
1852 component.is_glossary and "read-only" in parsed_flags
1853 ):
1854 unit_state = STATE_READONLY
1855 elif state is None:
1856 unit_state = STATE_TRANSLATED if has_translation else STATE_EMPTY
1857 elif has_translation and state == STATE_EMPTY:
1858 unit_state = STATE_TRANSLATED
1859 elif not has_translation and state != STATE_EMPTY:
1860 unit_state = STATE_EMPTY
1861 else:
1862 unit_state = state
1863 unit = Unit(
1864 translation=translation,
1865 context=context,
1866 source=source,
1867 target=current_target,
1868 state=unit_state,
1869 source_unit=source_unit,
1870 id_hash=id_hash,
1871 position=translation.stats.all + 1,
1872 **kwargs,
1873 )
1874 unit.is_batch_update = is_batch_update
1875 unit.trigger_update_variants = False
1876 try:
1877 with transaction.atomic():
1878 unit.save(
1879 force_insert=True,
1880 sync_terminology=False,
1881 )
1882 if pending:
1883 PendingUnitChange.store_unit_change(
1884 unit=unit, author=user, add_unit=True
1885 )
1886 changes.append(
1887 unit.generate_change(
1888 user=user,
1889 author=author or user,
1890 change_action=ActionEvents.NEW_UNIT,
1891 check_new=False,
1892 save=False,
1893 )
1894 )
1895 except IntegrityError:
1896 if not skip_existing:
1897 raise
1898 unit = translation.unit_set.get(id_hash=id_hash)
1899 # The source language is always first in the translations array
1900 if source_unit is None:
1901 source_unit = unit
1902 component._sources[id_hash] = unit # noqa: SLF001
1903 if translation == self:
1904 result = unit
1905 unit_ids.append(unit.pk)
1907 if changes:
1908 if is_batch_update:
1909 self.update_changes.extend(changes)
1910 else:
1911 Change.objects.bulk_create(changes)
1913 if not is_batch_update:
1914 if self.component.needs_variants_update:
1915 component.update_variants(
1916 updated_units=Unit.objects.filter(pk__in=unit_ids)
1917 )
1918 if component_wide:
1919 component.invalidate_cache()
1920 else:
1921 for translation in translations:
1922 translation.invalidate_cache()
1923 component_post_update.send(sender=self.__class__, component=component)
1924 return result
1926 def notify_deletion(self, unit, user: User) -> None:
1927 self.change_set.create(
1928 action=ActionEvents.STRING_REMOVE,
1929 user=user,
1930 target=unit.target,
1931 details={
1932 "source": unit.source,
1933 "target": unit.target,
1934 "context": unit.context,
1935 },
1936 )
1938 @transaction.atomic
1939 def delete_unit(self, request: AuthenticatedHttpRequest | None, unit: Unit) -> None:
1940 from weblate.auth.models import get_anonymous
1942 component = self.component
1943 user = request.user if request else get_anonymous()
1944 with component.repository.lock:
1945 component.commit_pending("delete unit", user)
1946 previous_revision = self.component.repository.last_revision
1947 cleanup_variants = False
1948 for translation in self.get_store_change_translations():
1949 # Does unit exist here?
1950 try:
1951 translation_unit = translation.unit_set.select_for_update().get(
1952 id_hash=unit.id_hash
1953 )
1954 except ObjectDoesNotExist:
1955 continue
1956 # Delete the removed unit from the database
1957 cleanup_variants |= translation_unit.variant_id is not None
1958 translation_unit.delete()
1959 translation.notify_deletion(translation_unit, user)
1960 # Skip file processing on source language without a storage
1961 if not translation.filename:
1962 continue
1963 # Does unit exist in the file?
1964 try:
1965 pounit, needs_add = translation.store.find_unit(
1966 unit.context, unit.source
1967 )
1968 except UnitNotFoundError:
1969 needs_add = True
1970 if not needs_add:
1971 # Commit changed file
1972 extra_files = translation.store.remove_unit(pounit.unit)
1973 translation.addon_commit_files.extend(extra_files)
1974 translation.drop_store_cache()
1975 translation.git_commit(
1976 user, user.get_author_name(), store_hash=False
1977 )
1978 # Adjust position as it will happen in most formats
1979 if translation_unit.position:
1980 translation.unit_set.filter(
1981 position__gt=translation_unit.position
1982 ).update(position=F("position") - 1)
1983 # Delete stale source units
1984 if not self.is_source and translation == self:
1985 source_unit = translation_unit.source_unit
1986 if source_unit.source_unit.unit_set.count() == 1:
1987 source_unit.delete()
1988 source_unit.translation.notify_deletion(source_unit, user)
1990 if self.is_source and unit.position and not component.has_template():
1991 # Adjust position is source language
1992 self.unit_set.filter(position__gt=unit.position).update(
1993 position=F("position") - 1
1994 )
1996 if cleanup_variants:
1997 self.component.update_variants()
1999 try:
2000 alert = self.component.alert_set.get(name="DuplicateString")
2001 except ObjectDoesNotExist:
2002 pass
2003 else:
2004 occurrences = [
2005 item
2006 for item in alert.details["occurrences"]
2007 if unit.pk != item["unit_pk"]
2008 ]
2009 if not occurrences:
2010 alert.delete()
2011 elif occurrences != alert.details["occurrences"]:
2012 alert.details["occurrences"] = occurrences
2013 alert.save(update_fields=["details"])
2015 self.handle_store_change(request, user, previous_revision)
2017 @transaction.atomic
2018 def sync_terminology(self) -> None:
2019 from weblate.auth.models import User
2021 if not self.is_source or not self.component.manage_units: 2021 ↛ 2022line 2021 didn't jump to line 2022 because the condition on line 2021 was never true
2022 return
2023 expected_count = self.component.translation_set.count()
2024 author: User | None = None
2025 added: bool = False
2026 for source in self.component.get_all_sources():
2027 # Is the string a terminology
2028 if "terminology" not in source.all_flags: 2028 ↛ 2030line 2028 didn't jump to line 2030 because the condition on line 2028 was always true
2029 continue
2030 if source.unit_set.count() == expected_count:
2031 continue
2032 added = True
2033 if author is None:
2034 author = User.objects.get_or_create_bot(
2035 scope="glossary",
2036 name="sync",
2037 verbose="Glossary sync",
2038 )
2039 # Add unit
2040 self.add_unit(
2041 None,
2042 source.context,
2043 source.get_source_plurals(),
2044 "",
2045 is_batch_update=True,
2046 skip_existing=True,
2047 author=author,
2048 )
2049 if added: 2049 ↛ 2050line 2049 didn't jump to line 2050 because the condition on line 2049 was never true
2050 self.store_update_changes()
2051 self.component.invalidate_cache()
2053 def validate_new_unit_data(
2054 self,
2055 context: str,
2056 source: str | list[str],
2057 target: str | list[str] | None = None,
2058 auto_context: bool = False,
2059 extra_flags: str | None = None,
2060 explanation: str = "",
2061 state: int | None = None,
2062 skip_existing: bool = False,
2063 ) -> None:
2064 component = self.component
2065 extra = {}
2066 if isinstance(source, str):
2067 source = [source]
2068 if len(source) > 1 and not component.file_format_cls.supports_plural:
2069 raise ValidationError(
2070 gettext("Plurals are not supported by the file format.")
2071 )
2072 for text in chain(source, [context]):
2073 if any(char in text for char in CONTROLCHARS):
2074 raise ValidationError(
2075 gettext("String contains control character: %s") % repr(text)
2076 )
2077 if state is not None:
2078 if state == STATE_EMPTY and any(source):
2079 raise ValidationError(
2080 gettext("Empty state is supported for blank strings only.")
2081 )
2082 if not any(source) and state != STATE_EMPTY:
2083 raise ValidationError(gettext("Blank strings require an empty state."))
2084 if state == STATE_APPROVED and not self.enable_review:
2085 raise ValidationError(
2086 gettext(
2087 "Approved state is not available as reviews are not enabled."
2088 )
2089 )
2090 if context:
2091 component.file_format_cls.validate_context(context)
2092 if not component.has_template():
2093 extra["source"] = join_plural(source)
2094 if not auto_context and self.unit_set.filter(context=context, **extra).exists():
2095 raise ValidationError(gettext("This string seems to already exist."))
2096 # Avoid using source translations without a filename
2097 if not self.filename:
2098 try:
2099 translation = component.translation_set.exclude(pk=self.pk)[0]
2100 except IndexError as error:
2101 raise ValidationError(
2102 gettext("Failed adding string, no translation found.")
2103 ) from error
2104 translation.validate_new_unit_data(
2105 context,
2106 source,
2107 target,
2108 auto_context=auto_context,
2109 extra_flags=extra_flags,
2110 explanation=explanation,
2111 )
2112 return
2114 @property
2115 def all_repo_components(self):
2116 return self.component.all_repo_components
2118 def get_source_plurals(self):
2119 """Return blank source fields for pluralized new string."""
2120 return [""] * self.plural.number
2122 def can_be_deleted(self) -> bool:
2123 """
2124 Check if a glossary can be deleted.
2126 It is possible to delete a glossary if:
2127 - it has no translations
2128 - it is managed by Weblate (i.e. repo == 'local:')
2129 """
2130 return self.stats.translated == 0 and self.component.repo == "local:"
2132 def get_glossaries(self) -> TranslationQuerySet:
2133 return (
2134 Translation.objects.filter(
2135 component__in=self.component.project.glossaries, language=self.language
2136 )
2137 .prefetch()
2138 .order()
2139 )
2142class GhostTranslation:
2143 """Ghost translation object used to show missing translations."""
2145 is_ghost = True
2147 def __init__(self, project, language, component) -> None:
2148 self.project = project
2149 self.language = language
2150 self.component = component
2151 self.stats = GhostStats(component.stats)
2152 self.pk = self.stats.pk
2153 self.is_source = False
2155 def __str__(self) -> str:
2156 return f"{self.component} — {self.language}"
2158 def get_absolute_url(self) -> str:
2159 return ""
2161 def base_obj(self):
2162 return self.component