Coverage for app/venv/lib/python3.14/site-packages/weblate/utils/stats.py: 71%
843 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 time
8from datetime import datetime, timedelta
9from itertools import chain
10from operator import itemgetter
11from types import GeneratorType
12from typing import TYPE_CHECKING, cast
14import sentry_sdk
15from django.conf import settings
16from django.core.cache import cache
17from django.core.exceptions import ObjectDoesNotExist
18from django.db import transaction
19from django.db.models import Count, F, Q
20from django.db.models.functions import Length
21from django.urls import reverse
22from django.utils import timezone
23from django.utils.functional import cached_property
25from weblate.checks.models import CHECKS
26from weblate.lang.models import Language
27from weblate.trans.checklists import TranslationChecklistMixin
28from weblate.trans.mixins import BaseURLMixin
29from weblate.trans.util import translation_percent
30from weblate.utils.random import get_random_identifier
31from weblate.utils.site import get_site_url
32from weblate.utils.state import (
33 STATE_APPROVED,
34 STATE_EMPTY,
35 STATE_FUZZY,
36 STATE_READONLY,
37 STATE_TRANSLATED,
38)
40if TYPE_CHECKING: 40 ↛ 41line 40 didn't jump to line 41 because the condition on line 40 was never true
41 from collections.abc import Generator, Iterable
43 from django.db.models import Model
45 from weblate.trans.models import Category, Component, Project
47StatItem = int | float | str | datetime | None
48StatDict = dict[str, StatItem]
50BASICS = {
51 "all",
52 "fuzzy",
53 "todo",
54 "readonly",
55 "nottranslated",
56 "translated",
57 "approved",
58 "allchecks",
59 "translated_checks",
60 "dismissed_checks",
61 "suggestions",
62 "nosuggestions",
63 "comments",
64 "approved_suggestions",
65 "unlabeled",
66 "unapproved",
67}
68BASIC_KEYS = frozenset(
69 (
70 *(f"{x}_words" for x in BASICS),
71 *(f"{x}_chars" for x in BASICS),
72 *BASICS,
73 "languages",
74 "last_changed",
75 "last_author",
76 "recent_changes",
77 "monthly_changes",
78 "total_changes",
79 "stats_timestamp",
80 )
81)
82SOURCE_KEYS = frozenset(
83 (
84 *BASIC_KEYS,
85 "source_strings",
86 "source_words",
87 "source_chars",
88 )
89)
91# TODO: Drop in Weblate 6
92LEGACY_KEYS = {
93 "unapproved",
94 "unapproved_chars",
95 "unapproved_words",
96 "recent_changes",
97 "monthly_changes",
98 "total_changes",
99 "stats_timestamp",
100}
102SOURCE_MAP = {
103 "source_chars": "all_chars",
104 "source_words": "all_words",
105 "source_strings": "all",
106}
109def zero_stats(keys: Iterable[str]) -> StatDict:
110 stats: StatDict = dict.fromkeys(keys, 0)
111 stats["last_changed"] = None
112 stats["last_author"] = None
113 stats["stats_timestamp"] = 0
114 return stats
117def yield_stats(queryset) -> Iterable[BaseStats]:
118 """
119 Yield stats attribute from the iterable.
121 This is an effective wrapper to use iterator over queryset
122 to generate all stats items.
123 """
124 yield from (item.stats for item in queryset.iterator())
127def prefetch_stats(queryset):
128 """Fetch stats from cache for a queryset."""
129 # Force evaluating queryset/iterator, we need all objects
130 objects = list(queryset)
132 # This function can either accept queryset, in which case it is
133 # returned with prefetched stats, or iterator, in which case new list
134 # is returned.
135 # This is needed to allow using such querysets further and to support
136 # processing iterator when it is more effective.
137 result = objects if isinstance(queryset, (chain, GeneratorType)) else queryset
139 # Bail out in case the query is empty
140 if not objects:
141 return result
143 # Use stats prefetch
144 objects[0].stats.prefetch_many([i.stats for i in objects])
146 return result
149def get_non_glossary_stats(
150 stats_obj: ProjectLanguageStats | ProjectStats | GlobalStats,
151) -> dict[str, int]:
152 """Return a dictionary with all, source_strings, and translated strings count excluding glossary content."""
153 result = {
154 "all": stats_obj.all,
155 "translated": stats_obj.translated,
156 "source_strings": getattr(stats_obj, "source_strings", stats_obj.all),
157 }
159 if isinstance(stats_obj, ProjectLanguageStats):
160 from weblate.trans.models import Translation
162 glossaries = Translation.objects.filter(
163 language=stats_obj.language, component__in=stats_obj.project.glossaries
164 ).prefetch()
165 elif isinstance(stats_obj, ProjectStats):
166 glossaries = stats_obj._object.glossaries # noqa: SLF001
167 elif isinstance(stats_obj, GlobalStats):
168 from weblate.trans.models import Component
170 glossaries = Component.objects.filter(is_glossary=True)
171 else:
172 # other stat types do not concern glossaries
173 return result
175 for glossary in prefetch_stats(glossaries):
176 result["all"] -= glossary.stats.all
177 result["translated"] -= glossary.stats.translated
178 result["source_strings"] -= getattr(
179 glossary.stats, "source_strings", glossary.stats.all
180 )
182 return result
185class BaseStats:
186 """Caching statistics calculator."""
188 basic_keys = BASIC_KEYS
189 is_ghost = False
191 def __init__(self, obj) -> None:
192 self._object = obj
193 self._data: StatDict = {}
194 self._loaded: bool = False
195 self._pending_save: bool = False
196 self.last_change_cache = None
197 self._collected_update_objects: list[BaseStats] | None = None
199 def __repr__(self) -> str:
200 return f"<{self.__class__.__name__}:{self.cache_key}>"
202 @property
203 def pk(self) -> int:
204 return self._object.pk
206 def get_absolute_url(self) -> str:
207 return self._object.get_absolute_url()
209 def get_translate_url(self) -> str:
210 return self._object.get_translate_url()
212 @property
213 def obj(self):
214 return self._object
216 @property
217 def stats(self):
218 return self
220 @property
221 def is_loaded(self) -> bool:
222 return self._loaded
224 def set_data(self, data: StatDict) -> None:
225 self._loaded = True
226 self._data = data
228 def get_data(self) -> StatDict:
229 """
230 Return a copy of data including percents.
232 Used in stats endpoints.
233 """
234 self.ensure_loaded()
235 percents = [
236 "translated_percent",
237 "approved_percent",
238 "fuzzy_percent",
239 "readonly_percent",
240 "allchecks_percent",
241 "translated_checks_percent",
242 "translated_words_percent",
243 "approved_words_percent",
244 "fuzzy_words_percent",
245 "readonly_words_percent",
246 "allchecks_words_percent",
247 "translated_checks_words_percent",
248 ]
249 data: StatDict = {
250 percent: self.calculate_percent(percent) for percent in percents
251 }
252 data.update(self._data)
253 return data
255 @staticmethod
256 def prefetch_many(stats) -> None:
257 lookup = {i.cache_key: i for i in stats if not i.is_loaded}
258 if not lookup: 258 ↛ 259line 258 didn't jump to line 259 because the condition on line 258 was never true
259 return
260 data = cache.get_many(lookup.keys())
261 for item, value in data.items():
262 lookup[item].set_data(value)
263 for item in set(lookup.keys()) - set(data.keys()):
264 lookup[item].set_data({})
266 @cached_property
267 def has_review(self) -> bool:
268 return True
270 @cached_property
271 def cache_key(self) -> str:
272 return f"stats-{self._object.cache_key}"
274 def remove_stats(self, *names: str) -> None:
275 self.ensure_loaded()
276 if not self._data:
277 return
278 changed = False
279 for name in names:
280 if name in self._data:
281 del self._data[name]
282 changed = True
283 if changed:
284 self.save()
286 def ensure_loaded(self) -> None:
287 """Load from cache if not already done."""
288 if not self._loaded:
289 self.force_load()
291 def force_load(self) -> None:
292 """Enforced loading of stats."""
293 self._data = self.load()
294 self._loaded = True
296 def aggregate_get(self, name: str) -> StatItem:
297 """
298 Fast getter for aggregation.
300 - expects data is present
301 - expects valid basic keys only
302 - maps source keys
303 """
304 try:
305 return self._data[name]
306 except KeyError:
307 # Handle source_* keys as virtual on translation level for easier aggregation
308 if name.startswith("source_"): 308 ↛ 314line 308 didn't jump to line 314 because the condition on line 308 was always true
309 return self._data[SOURCE_MAP[name]]
310 # Legacy keys were calculated on demand before and are precalculated
311 # since Weblate 5.2, so they are missing on stats calculated before.
312 # Using zero here is most likely a wrong value, but safe and cheap.
313 # TODO: Drop in Weblate 6
314 if name in LEGACY_KEYS:
315 return 0
316 raise
318 def __getattr__(self, name: str):
319 if name.startswith("_"): 319 ↛ 320line 319 didn't jump to line 320 because the condition on line 319 was never true
320 msg = f"Invalid stats for {self}: {name}"
321 raise AttributeError(msg)
323 self.ensure_loaded()
325 # Calculate virtual percents
326 if name.endswith("_percent"):
327 return self.calculate_percent(name)
329 if name == "stats_timestamp":
330 # TODO: Drop in Weblate 6
331 # Migration path for legacy stat data
332 return self._data.get(name, 0)
334 # Virtual fields
335 if name == "translated_without_checks": 335 ↛ 336line 335 didn't jump to line 336 because the condition on line 335 was never true
336 return self.translated - self.translated_checks
337 if name == "translated_without_checks_words": 337 ↛ 338line 337 didn't jump to line 338 because the condition on line 337 was never true
338 return self.translated_words - self.translated_checks_words
339 if name == "translated_without_checks_chars": 339 ↛ 340line 339 didn't jump to line 340 because the condition on line 339 was never true
340 return self.translated_chars - self.translated_checks_chars
342 # Calculate missing data
343 if name not in self._data:
344 was_pending = self._pending_save
345 self._pending_save = True
346 self.calculate_by_name(name)
347 if name not in self._data: 347 ↛ 348line 347 didn't jump to line 348 because the condition on line 347 was never true
348 msg = f"Unsupported stats for {self}: {name}"
349 self._pending_save = was_pending
350 raise AttributeError(msg)
351 if not was_pending: 351 ↛ 355line 351 didn't jump to line 355 because the condition on line 351 was always true
352 self.save()
353 self._pending_save = False
355 return self._data[name]
357 def calculate_by_name(self, name: str) -> None:
358 if name in self.basic_keys: 358 ↛ exitline 358 didn't return from function 'calculate_by_name' because the condition on line 358 was always true
359 self.calculate_basic()
360 self.save()
362 def load(self) -> StatDict:
363 return cache.get(self.cache_key, {})
365 def delete(self):
366 return cache.delete(self.cache_key)
368 def save(self, update_parents: bool = True) -> None:
369 """Save stats to cache."""
370 cache.set(self.cache_key, self._data, 30 * 86400)
372 def get_update_objects(self, *, full: bool = True) -> Generator[BaseStats]:
373 yield GlobalStats()
375 def collect_update_objects(self) -> None:
376 """
377 Collect update objects.
379 This is used on the pre_delete signal as the objects
380 cannot be collected once the object is deleted.
381 """
382 # Use list to force materializing the generator
383 self._collected_update_objects = list(self.get_update_objects())
385 def _iterate_update_objects(
386 self, *, extra_objects: Iterable[BaseStats] | None = None
387 ) -> Generator[BaseStats]:
388 """Get list of stats to update."""
389 stat_objects: Iterable[BaseStats]
390 if self._collected_update_objects is not None: 390 ↛ 391line 390 didn't jump to line 391 because the condition on line 390 was never true
391 stat_objects = self._collected_update_objects
392 self._collected_update_objects = None
393 else:
394 stat_objects = self.get_update_objects()
395 if extra_objects:
396 stat_objects = chain(stat_objects, extra_objects)
398 # Prefetch stats data from cache
399 fetched_stat_objects: list[BaseStats] = prefetch_stats(stat_objects)
401 # Discard references to no longer needed objects
402 while fetched_stat_objects:
403 yield fetched_stat_objects.pop(0)
405 def update_parents(
406 self, *, extra_objects: Iterable[BaseStats] | None = None
407 ) -> None:
408 """Update parent statistics."""
409 for stat in self._iterate_update_objects(extra_objects=extra_objects):
410 if self.stats_timestamp and self.stats_timestamp <= stat.stats_timestamp:
411 self._object.log_debug("skipping update of %s", stat)
412 else:
413 self._object.log_debug("updating stats %s", stat)
414 stat.update_stats()
416 def clear(self) -> None:
417 """Clear local cache."""
418 self._loaded = True
419 self._data = {}
421 def store(self, key: str, value: StatItem) -> None:
422 if value is None and not key.startswith("last_"): 422 ↛ 423line 422 didn't jump to line 423 because the condition on line 422 was never true
423 self._data[key] = 0
424 else:
425 self._data[key] = value
427 def update_stats(self, update_parents: bool = True) -> None:
428 self.clear()
429 if settings.STATS_LAZY: 429 ↛ 430line 429 didn't jump to line 430 because the condition on line 429 was never true
430 self.save(update_parents=update_parents)
431 else:
432 self.calculate_basic()
433 self.save(update_parents=update_parents)
435 def calculate_basic(self) -> None:
436 with sentry_sdk.start_span(op="stats", name=f"CALCULATE {self.cache_key}"):
437 self.ensure_loaded()
438 self._calculate_basic()
440 def _calculate_basic(self) -> None:
441 raise NotImplementedError
443 def calculate_percent(self, item: str) -> float:
444 """Calculate percent value for given item."""
445 base = item[:-8]
447 if base.endswith("_words"):
448 total = self.all_words
449 elif base.endswith("_chars"):
450 total = self.all_chars
451 else:
452 total = self.all
454 if self.has_review:
455 completed = {"approved", "approved_words", "approved_chars"}
456 else:
457 completed = {"translated", "translated_words", "translated_chars"}
459 return translation_percent(
460 getattr(self, base), total, zero_complete=(base in completed)
461 )
463 @property
464 def waiting_review_percent(self):
465 return self.translated_percent - self.approved_percent - self.readonly_percent
467 @property
468 def waiting_review(self):
469 return self.translated - self.approved - self.readonly
471 @property
472 def waiting_review_words_percent(self):
473 return (
474 self.translated_words_percent
475 - self.approved_words_percent
476 - self.readonly_words_percent
477 )
479 @property
480 def waiting_review_words(self):
481 return self.translated_words - self.approved_words - self.readonly_words
483 @property
484 def waiting_review_chars_percent(self):
485 return (
486 self.translated_chars_percent
487 - self.approved_chars_percent
488 - self.readonly_chars_percent
489 )
491 @property
492 def waiting_review_chars(self):
493 return self.translated_chars - self.approved_chars - self.readonly_chars
496class DummyTranslationStats(BaseStats):
497 """
498 Dummy stats to report 0 in all cases.
500 Used when given language does not exist in a component.
501 """
503 def __init__(self, obj) -> None:
504 super().__init__(obj)
505 self.language = obj
507 @property
508 def pk(self) -> int:
509 return 100000 + self.language.pk
511 @cached_property
512 def cache_key(self) -> str:
513 return ""
515 def save(self, update_parents: bool = True) -> None:
516 return
518 def load(self) -> StatDict:
519 return {}
521 def _calculate_basic(self) -> None:
522 self._data = zero_stats(self.basic_keys)
525class TranslationStats(BaseStats):
526 """Per translation stats."""
528 @cached_property
529 def is_source(self):
530 return self._object.is_source
532 def save(self, update_parents: bool = True) -> None:
533 from weblate.utils.tasks import update_translation_stats_parents
535 super().save()
537 if update_parents:
538 if settings.CELERY_TASK_ALWAYS_EAGER: 538 ↛ 539line 538 didn't jump to line 539 because the condition on line 538 was never true
539 transaction.on_commit(self.update_parents)
540 else:
541 pk = self._object.pk
542 update_translation_stats_parents.delay_on_commit(pk)
544 def get_update_objects(self, *, full: bool = True) -> Generator[BaseStats]:
545 translation = self._object
546 component = translation.component
548 # Language
549 yield translation.language.stats
551 # Project / language
552 yield component.project.stats.get_single_language_stats(translation.language)
554 # Linked project / language
555 for link in component.cached_links: 555 ↛ 556line 555 didn't jump to line 556 because the loop on line 555 never started
556 yield link.stats.get_single_language_stats(translation.language)
558 # Category / language
559 category = component.category
560 while category: 560 ↛ 561line 560 didn't jump to line 561 because the condition on line 560 was never true
561 yield category.stats.get_single_language_stats(translation.language)
562 category = category.category
564 if full:
565 # Component
566 yield component.stats
567 # Component list, category, project and global
568 yield from component.stats.get_update_objects()
570 @property
571 def language(self):
572 return self._object.language
574 @cached_property
575 def has_review(self):
576 return self._object.enable_review
578 def _calculate_basic(self) -> None: # noqa: PLR0914
579 values = (
580 "state",
581 "num_words",
582 "active_checks_count",
583 "dismissed_checks_count",
584 "suggestion_count",
585 "source_label_count",
586 "label_count",
587 "comment_count",
588 "num_chars",
589 )
591 # Calculate summary for each unit in the database.
592 # We only need presence check, not actual count, but using Exists(OuterRef())
593 # creates a subquery for each field, while Count() creates a single join
594 # and calculates based on that, what performs better.
595 units = self._object.unit_set.annotate(
596 active_checks_count=Count("check", filter=Q(check__dismissed=False)),
597 dismissed_checks_count=Count("check", filter=Q(check__dismissed=True)),
598 suggestion_count=Count("suggestion"),
599 source_label_count=Count("source_unit__labels"),
600 label_count=Count("source_unit__labels"),
601 comment_count=Count("comment", filter=Q(comment__resolved=False)),
602 num_chars=Length("source"),
603 ).values_list(*values)
605 (
606 get_state,
607 get_num_words,
608 get_active_checks_count,
609 get_dismissed_checks_count,
610 get_suggestion_count,
611 get_source_label_count,
612 get_label_count,
613 get_comment_count,
614 get_num_chars,
615 ) = (itemgetter(i) for i in range(len(values)))
617 # Sum stats in Python, this is way faster than conditional sums in the database
618 units_all = units
619 units_fuzzy = [unit for unit in units if get_state(unit) == STATE_FUZZY]
620 units_readonly = [unit for unit in units if get_state(unit) == STATE_READONLY]
621 units_nottranslated = [unit for unit in units if get_state(unit) == STATE_EMPTY]
622 units_unapproved = [
623 unit for unit in units if get_state(unit) == STATE_TRANSLATED
624 ]
625 units_approved = [unit for unit in units if get_state(unit) == STATE_APPROVED]
626 units_translated = [
627 unit for unit in units if get_state(unit) >= STATE_TRANSLATED
628 ]
629 units_todo = [unit for unit in units if get_state(unit) < STATE_TRANSLATED]
630 units_unlabeled = [
631 unit
632 for unit in units
633 if not get_source_label_count(unit) and not get_label_count(unit)
634 ]
635 units_allchecks = [unit for unit in units if get_active_checks_count(unit)]
636 units_translated_checks = [
637 unit
638 for unit in units
639 if get_active_checks_count(unit)
640 and get_state(unit) in {STATE_TRANSLATED, STATE_APPROVED}
641 ]
642 units_dismissed_checks = [
643 unit for unit in units if get_dismissed_checks_count(unit)
644 ]
645 units_suggestions = [unit for unit in units if get_suggestion_count(unit)]
646 units_nosuggestions = [
647 unit
648 for unit in units
649 if not get_suggestion_count(unit) and get_state(unit) < STATE_TRANSLATED
650 ]
651 units_approved_suggestions = [
652 unit
653 for unit in units
654 if get_suggestion_count(unit) and get_state(unit) == STATE_APPROVED
655 ]
656 units_comments = [unit for unit in units if get_comment_count(unit)]
658 # Store in a cache
659 self.store("all", len(units_all))
660 self.store("all_words", sum(get_num_words(unit) for unit in units_all))
661 self.store("all_chars", sum(get_num_chars(unit) for unit in units_all))
662 self.store("fuzzy", len(units_fuzzy))
663 self.store("fuzzy_words", sum(get_num_words(unit) for unit in units_fuzzy))
664 self.store("fuzzy_chars", sum(get_num_chars(unit) for unit in units_fuzzy))
665 self.store("readonly", len(units_readonly))
666 self.store(
667 "readonly_words", sum(get_num_words(unit) for unit in units_readonly)
668 )
669 self.store(
670 "readonly_chars", sum(get_num_chars(unit) for unit in units_readonly)
671 )
672 self.store("translated", len(units_translated))
673 self.store(
674 "translated_words", sum(get_num_words(unit) for unit in units_translated)
675 )
676 self.store(
677 "translated_chars", sum(get_num_chars(unit) for unit in units_translated)
678 )
679 self.store("todo", len(units_todo))
680 self.store("todo_words", sum(get_num_words(unit) for unit in units_todo))
681 self.store("todo_chars", sum(get_num_chars(unit) for unit in units_todo))
682 self.store("nottranslated", len(units_nottranslated))
683 self.store(
684 "nottranslated_words",
685 sum(get_num_words(unit) for unit in units_nottranslated),
686 )
687 self.store(
688 "nottranslated_chars",
689 sum(get_num_chars(unit) for unit in units_nottranslated),
690 )
691 # Review workflow
692 self.store("approved", len(units_approved))
693 self.store(
694 "approved_words", sum(get_num_words(unit) for unit in units_approved)
695 )
696 self.store(
697 "approved_chars", sum(get_num_chars(unit) for unit in units_approved)
698 )
699 self.store("unapproved", len(units_unapproved))
700 self.store(
701 "unapproved_words", sum(get_num_words(unit) for unit in units_unapproved)
702 )
703 self.store(
704 "unapproved_chars", sum(get_num_chars(unit) for unit in units_unapproved)
705 )
706 # Labels
707 self.store("unlabeled", len(units_unlabeled))
708 self.store(
709 "unlabeled_words", sum(get_num_words(unit) for unit in units_unlabeled)
710 )
711 self.store(
712 "unlabeled_chars", sum(get_num_chars(unit) for unit in units_unlabeled)
713 )
714 # Checks
715 self.store("allchecks", len(units_allchecks))
716 self.store(
717 "allchecks_words", sum(get_num_words(unit) for unit in units_allchecks)
718 )
719 self.store(
720 "allchecks_chars", sum(get_num_chars(unit) for unit in units_allchecks)
721 )
722 self.store("translated_checks", len(units_translated_checks))
723 self.store(
724 "translated_checks_words",
725 sum(get_num_words(unit) for unit in units_translated_checks),
726 )
727 self.store(
728 "translated_checks_chars",
729 sum(get_num_chars(unit) for unit in units_translated_checks),
730 )
731 self.store("dismissed_checks", len(units_dismissed_checks))
732 self.store(
733 "dismissed_checks_words",
734 sum(get_num_words(unit) for unit in units_dismissed_checks),
735 )
736 self.store(
737 "dismissed_checks_chars",
738 sum(get_num_chars(unit) for unit in units_dismissed_checks),
739 )
740 # Suggestions
741 self.store("suggestions", len(units_suggestions))
742 self.store(
743 "suggestions_words", sum(get_num_words(unit) for unit in units_suggestions)
744 )
745 self.store(
746 "suggestions_chars", sum(get_num_chars(unit) for unit in units_suggestions)
747 )
748 self.store("nosuggestions", len(units_nosuggestions))
749 self.store(
750 "nosuggestions_words",
751 sum(get_num_words(unit) for unit in units_nosuggestions),
752 )
753 self.store(
754 "nosuggestions_chars",
755 sum(get_num_chars(unit) for unit in units_nosuggestions),
756 )
757 self.store("approved_suggestions", len(units_approved_suggestions))
758 self.store(
759 "approved_suggestions_words",
760 sum(get_num_words(unit) for unit in units_approved_suggestions),
761 )
762 self.store(
763 "approved_suggestions_chars",
764 sum(get_num_chars(unit) for unit in units_approved_suggestions),
765 )
766 # Comments
767 self.store("comments", len(units_comments))
768 self.store(
769 "comments_words", sum(get_num_words(unit) for unit in units_comments)
770 )
771 self.store(
772 "comments_chars", sum(get_num_chars(unit) for unit in units_comments)
773 )
775 # There is single language here, but it is aggregated at higher levels
776 self.store("languages", 1)
778 # Last change timestamp
779 self.fetch_last_change()
781 self.count_changes()
783 # Store timestamp
784 self.store("stats_timestamp", time.time())
786 def get_last_change_obj(self):
787 from weblate.trans.models import Change
789 # This is set in Change.save
790 if self.last_change_cache is not None:
791 return self.last_change_cache
793 cache_key = Change.get_last_change_cache_key(self._object.pk)
794 change_pk = cache.get(cache_key)
795 if change_pk == 0:
796 # No change
797 return None
798 if change_pk is not None:
799 try:
800 return Change.objects.get(pk=change_pk)
801 except Change.DoesNotExist:
802 pass
803 try:
804 last_change = self._object.change_set.order()[0]
805 except IndexError:
806 Change.store_last_change(self._object, None)
807 return None
808 last_change.update_cache_last_change()
809 return last_change
811 def fetch_last_change(self) -> None:
812 last_change = self.get_last_change_obj()
814 if last_change is None:
815 self.store("last_changed", None)
816 self.store("last_author", None)
817 else:
818 self.store("last_changed", last_change.timestamp)
819 self.store("last_author", last_change.author_id)
821 def count_changes(self) -> None:
822 if self.last_changed:
823 monthly = timezone.now() - timedelta(days=30)
824 recently = self.last_changed - timedelta(hours=6)
825 changes = self._object.change_set.content().aggregate(
826 total=Count("id"),
827 recent=Count("id", filter=Q(timestamp__gt=recently)),
828 monthly=Count("id", filter=Q(timestamp__gt=monthly)),
829 )
830 self.store("recent_changes", changes["recent"])
831 self.store("monthly_changes", changes["monthly"])
832 self.store("total_changes", changes["total"])
833 else:
834 self.store("recent_changes", 0)
835 self.store("monthly_changes", 0)
836 self.store("total_changes", 0)
838 def calculate_by_name(self, name: str) -> None:
839 super().calculate_by_name(name)
840 if name.startswith("check:"): 840 ↛ 841line 840 didn't jump to line 841 because the condition on line 840 was never true
841 self.calculate_checks()
842 elif name.startswith("label:"): 842 ↛ 843line 842 didn't jump to line 843 because the condition on line 842 was never true
843 self.calculate_labels()
845 def calculate_checks(self) -> None:
846 """Prefetch check stats."""
847 self.ensure_loaded()
848 allchecks = {check.url_id for check in CHECKS.values()}
849 stats = (
850 self._object.unit_set.filter(check__dismissed=False)
851 .values("check__name")
852 .annotate_stats()
853 )
854 for check, strings, words, chars in stats.values_list(
855 "check__name", "strings", "words", "chars"
856 ):
857 # Filtering here is way more effective than in SQL
858 if check is None:
859 continue
860 check = f"check:{check}"
861 self.store(check, strings)
862 self.store(f"{check}_words", words)
863 self.store(f"{check}_chars", chars)
864 allchecks.discard(check)
865 for check in allchecks:
866 self.store(check, 0)
867 self.store(f"{check}_words", 0)
868 self.store(f"{check}_chars", 0)
869 self.save()
871 def calculate_labels(self) -> None:
872 """Prefetch check stats."""
873 from weblate.trans.models.label import TRANSLATION_LABELS
875 self.ensure_loaded()
876 alllabels = set(
877 self._object.component.project.label_set.values_list("name", flat=True)
878 )
879 stats = (
880 self._object.unit_set.values("source_unit__labels__name")
881 .annotate_stats()
882 .values_list("source_unit__labels__name", "strings", "words", "chars")
883 )
884 translation_stats = (
885 self._object.unit_set.filter(labels__name__in=TRANSLATION_LABELS)
886 .values("labels__name")
887 .annotate_stats()
888 .values_list("labels__name", "strings", "words", "chars")
889 )
891 for label_name, strings, words, chars in chain(stats, translation_stats):
892 # Filtering here is way more effective than in SQL
893 if label_name is None:
894 continue
895 label = f"label:{label_name}"
896 self.store(label, strings)
897 self.store(f"{label}_words", words)
898 self.store(f"{label}_chars", chars)
899 alllabels.discard(label_name)
900 for label_name in alllabels:
901 label = f"label:{label_name}"
902 self.store(label, 0)
903 self.store(f"{label}_words", 0)
904 self.store(f"{label}_chars", 0)
905 self.save()
908class AggregatingStats(BaseStats):
909 basic_keys = SOURCE_KEYS
910 sum_source_keys = True
912 def get_child_objects(self) -> Iterable[Model]:
913 raise NotImplementedError
915 def get_category_objects(self):
916 return []
918 def calculate_source(self, stats: dict, all_stats: list) -> None:
919 return
921 @cached_property
922 def aggregated_stats(self) -> list[BaseStats]:
923 return [
924 obj.stats
925 for obj in prefetch_stats(
926 chain(self.get_child_objects(), self.get_category_objects())
927 )
928 ]
930 def _calculate_basic(self) -> None:
931 stats = zero_stats(self.basic_keys)
932 all_stats: list[BaseStats] = self.aggregated_stats
934 # Ensure all objects have data available so that we can use _dict directly
935 for stats_obj in all_stats:
936 if "all" not in stats_obj._data: # noqa: SLF001
937 stats_obj.calculate_basic()
938 stats_obj.save()
940 for item in self.basic_keys:
941 if not self.sum_source_keys and item.startswith("source_"):
942 # Handle in calculate_source when logic for source strings differs
943 continue
945 # Extract all values by dedicated getter
946 values = (stats_obj.aggregate_get(item) for stats_obj in all_stats)
948 if item == "stats_timestamp":
949 stats[item] = max(cast("Generator[float]", values), default=time.time())
950 elif item == "last_changed":
951 # We need to access values twice here
952 values_list = list(values)
953 stats[item] = max_last_changed = max(
954 (value for value in values_list if value is not None), default=None
955 )
956 if max_last_changed is not None:
957 offset = values_list.index(max_last_changed)
958 stats["last_author"] = all_stats[offset].last_author
959 elif item == "last_author":
960 # The last_author is calculated together with last_changed
961 continue
962 else:
963 stats[item] = sum(cast("Generator[float]", values))
965 if not self.sum_source_keys:
966 self.calculate_source(stats, all_stats)
968 for key, value in stats.items():
969 self.store(key, value)
972class SingleLanguageStats(AggregatingStats):
973 def _calculate_basic(self) -> None:
974 super()._calculate_basic()
975 self.store("languages", 1)
977 def get_single_language_stats(self, language):
978 return self
980 @cached_property
981 def is_source(self):
982 return self.obj.is_source
985class ParentAggregatingStats(AggregatingStats):
986 sum_source_keys = True
989class LanguageStats(AggregatingStats):
990 def get_child_objects(self):
991 return self._object.translation_set.only("id", "language")
993 @property
994 def language(self):
995 return self._object
998class ComponentStats(AggregatingStats):
999 sum_source_keys = False
1001 def get_child_objects(self):
1002 return self._object.translation_set.only("id", "component", "language")
1004 @cached_property
1005 def has_review(self):
1006 return self._object.enable_review
1008 def calculate_source(self, stats: dict, all_stats: list) -> None:
1009 """Fetch source info from source translation."""
1010 for obj in all_stats: 1010 ↛ exitline 1010 didn't return from function 'calculate_source' because the loop on line 1010 didn't complete
1011 if obj.is_source:
1012 stats["source_chars"] = obj.all_chars
1013 stats["source_words"] = obj.all_words
1014 stats["source_strings"] = obj.all
1015 break
1017 def get_update_objects(self, *, full: bool = True) -> Generator[BaseStats]:
1018 # Component lists
1019 yield from yield_stats(self._object.componentlist_set.only("id", "slug"))
1021 # Projects this component is shared to
1022 yield from yield_stats(self._object.cached_links)
1024 if self._object.category: 1024 ↛ 1026line 1024 didn't jump to line 1026 because the condition on line 1024 was never true
1025 # Category
1026 yield self._object.category.stats
1027 # Category parents, project and global
1028 yield from self._object.category.stats.get_update_objects()
1029 else:
1030 # Project
1031 yield self._object.project.stats
1032 # Global
1033 yield from self._object.project.stats.get_update_objects()
1035 def update_language_stats_parents(self) -> None:
1036 # Fetch language stats to update
1037 extras: Iterable[Iterable[BaseStats]] = (
1038 translation.stats.get_update_objects(full=False)
1039 for translation in prefetch_stats(
1040 self.get_child_objects().select_related("language")
1041 )
1042 )
1044 # Update all parents
1045 self.update_parents(extra_objects=chain.from_iterable(extras))
1047 def update_language_stats(self) -> None:
1048 from weblate.utils.tasks import update_language_stats_parents
1050 # Update languages
1051 for translation in prefetch_stats(self.get_child_objects()):
1052 translation.stats.update_stats(update_parents=False)
1054 # Update our stats
1055 self.update_stats()
1057 # Update all parents
1058 if settings.CELERY_TASK_ALWAYS_EAGER: 1058 ↛ 1059line 1058 didn't jump to line 1059 because the condition on line 1058 was never true
1059 transaction.on_commit(self.update_language_stats_parents)
1060 else:
1061 pk = self._object.pk
1062 update_language_stats_parents.delay_on_commit(pk)
1064 def get_language_stats(self):
1065 return (
1066 translation.stats
1067 for translation in prefetch_stats(
1068 self._object.translation_set.select_related("language")
1069 )
1070 )
1072 def get_single_language_stats(self, language):
1073 try:
1074 return TranslationStats(self._object.translation_set.get(language=language))
1075 except ObjectDoesNotExist:
1076 return DummyTranslationStats(language)
1079class ProjectLanguageComponent:
1080 is_glossary = False
1082 def __init__(self, parent) -> None:
1083 self.slug = "-"
1084 self.parent = parent
1086 @property
1087 def translation_set(self):
1088 return self.parent.translation_set
1090 @property
1091 def context_label(self):
1092 return self.translation_set[0].component.context_label
1094 @property
1095 def source_language(self):
1096 return self.translation_set[0].component.source_language
1099class ProjectLanguage(BaseURLMixin, TranslationChecklistMixin):
1100 """Wrapper class used in project-language listings and stats."""
1102 remove_permission = "translation.delete"
1103 settings_permission = "project.edit"
1105 def __init__(self, project, language: Language) -> None:
1106 self.project = project
1107 self.language = language
1108 self.component = ProjectLanguageComponent(self)
1110 def __str__(self) -> str:
1111 return f"{self.project} - {self.language}"
1113 @property
1114 def enable_review(self) -> bool:
1115 return self.project.enable_review
1117 @property
1118 def is_readonly(self) -> bool:
1119 return False
1121 @property
1122 def code(self):
1123 return self.language.code
1125 @cached_property
1126 def stats(self):
1127 return ProjectLanguageStats(self)
1129 def get_share_url(self):
1130 """Return absolute URL usable for sharing."""
1131 return get_site_url(
1132 reverse(
1133 "engage",
1134 kwargs={"path": self.get_url_path()},
1135 )
1136 )
1138 def get_widgets_url(self) -> str:
1139 """Return absolute URL for widgets."""
1140 return f"{self.project.get_widgets_url()}?lang={self.language.code}"
1142 @cached_property
1143 def pk(self) -> str:
1144 return f"{self.project.pk}-{self.language.pk}"
1146 @cached_property
1147 def cache_key(self) -> str:
1148 return f"{self.project.cache_key}-{self.language.pk}"
1150 def get_url_path(self):
1151 return [*self.project.get_url_path(), "-", self.language.code]
1153 def get_absolute_url(self) -> str:
1154 return reverse("show", kwargs={"path": self.get_url_path()})
1156 def get_translate_url(self):
1157 return reverse("translate", kwargs={"path": self.get_url_path()})
1159 @cached_property
1160 def translation_set(self):
1161 all_langs = self.language.translation_set.prefetch()
1162 result = all_langs.filter(component__project=self.project)
1163 if self.project.has_shared_components:
1164 result |= all_langs.filter(component__links=self.project)
1165 result = result.distinct()
1166 for item in result:
1167 item.is_shared = (
1168 None
1169 if item.component.project == self.project
1170 else item.component.project
1171 )
1172 return sorted(
1173 result,
1174 key=lambda trans: (trans.component.priority, trans.component.name.lower()),
1175 )
1177 @cached_property
1178 def is_source(self):
1179 return self.language.id in self.project.source_language_ids
1181 @cached_property
1182 def change_set(self):
1183 return self.project.change_set.filter(language=self.language)
1185 @cached_property
1186 def workflow_settings(self):
1187 from weblate.trans.models.workflow import WorkflowSetting
1189 workflow_settings = WorkflowSetting.objects.filter(
1190 Q(project=None) | Q(project=self.project),
1191 language=self.language,
1192 ).order_by(F("project").desc(nulls_last=True))
1193 if len(workflow_settings) == 0: 1193 ↛ 1196line 1193 didn't jump to line 1196 because the condition on line 1193 was always true
1194 return None
1195 # Project specific is first, project NULL is last
1196 return workflow_settings[0]
1199class ChecklistStats(SingleLanguageStats):
1200 def calculate_by_name(self, name: str) -> None:
1201 super().calculate_by_name(name)
1202 if name.startswith("check:"):
1203 self.calculate_checks()
1204 elif name.startswith("label:"):
1205 self.calculate_labels()
1207 def aggregate_stats(self, keys: Iterable[str]) -> None:
1208 self.ensure_loaded()
1209 all_stats: list[BaseStats] = self.aggregated_stats
1210 suffixes: tuple[str, ...] = ("", "_words", "_chars")
1211 for key in keys:
1212 for suffix in suffixes:
1213 name = f"{key}{suffix}"
1214 # The attribute might be missing in some corner cases such as
1215 # when component is shared in different project and calculating
1216 # label stats.
1217 values = (getattr(stats_obj, name, 0) for stats_obj in all_stats)
1218 self.store(name, sum(values))
1219 self.save()
1221 def calculate_checks(self) -> None:
1222 """Prefetch check stats."""
1223 self.aggregate_stats(check.url_id for check in CHECKS.values())
1225 def calculate_labels(self) -> None:
1226 """Prefetch check stats."""
1227 self.aggregate_stats(
1228 f"label:{name}"
1229 for name in self._object.project.label_set.values_list("name", flat=True)
1230 )
1233class ProjectLanguageStats(ChecklistStats):
1234 def __init__(self, obj: ProjectLanguage) -> None:
1235 self.language = obj.language
1236 self.project = obj.project
1237 super().__init__(obj)
1238 obj.stats = self
1240 @cached_property
1241 def has_review(self):
1242 return self.project.source_review or self.project.translation_review
1244 def get_child_objects(self):
1245 return (
1246 self.language.translation_set.filter(
1247 Q(component__project=self.project) | Q(component__links=self.project)
1248 )
1249 .distinct()
1250 .only("id", "language")
1251 )
1254class CategoryLanguage(BaseURLMixin, TranslationChecklistMixin):
1255 """Wrapper class used in category-language listings and stats."""
1257 remove_permission = "translation.delete"
1259 def __init__(self, category: Category, language: Language) -> None:
1260 self.category = category
1261 self.language = language
1262 self.component = ProjectLanguageComponent(self)
1264 def __str__(self) -> str:
1265 return f"{self.category} - {self.language}"
1267 @property
1268 def project(self) -> Project:
1269 return self.category.project
1271 @property
1272 def enable_review(self) -> bool:
1273 return self.project.enable_review
1275 @property
1276 def is_readonly(self) -> bool:
1277 return False
1279 @property
1280 def code(self):
1281 return self.language.code
1283 @property
1284 def name(self):
1285 return self.category.name
1287 @cached_property
1288 def stats(self):
1289 return CategoryLanguageStats(self)
1291 @cached_property
1292 def pk(self) -> str:
1293 return f"{self.category.pk}-{self.language.pk}"
1295 @cached_property
1296 def cache_key(self) -> str:
1297 return f"{self.category.cache_key}-{self.language.pk}"
1299 def get_url_path(self):
1300 return [*self.category.get_url_path(), "-", self.language.code]
1302 def get_absolute_url(self) -> str:
1303 return reverse("show", kwargs={"path": self.get_url_path()})
1305 def get_translate_url(self):
1306 return reverse("translate", kwargs={"path": self.get_url_path()})
1308 @cached_property
1309 def translation_set(self):
1310 result = self.language.translation_set.filter(
1311 component__category=self.category
1312 ).prefetch()
1313 for item in result:
1314 item.is_shared = (
1315 None
1316 if item.component.project == self.category.project
1317 else item.component.project
1318 )
1319 return sorted(
1320 result,
1321 key=lambda trans: (trans.component.priority, trans.component.name.lower()),
1322 )
1324 @cached_property
1325 def is_source(self):
1326 return self.language.id in self.category.source_language_ids
1328 @cached_property
1329 def change_set(self):
1330 return self.language.change_set.for_category(self.category)
1333class CategoryLanguageStats(ChecklistStats):
1334 def __init__(self, obj: CategoryLanguage) -> None:
1335 self.language = obj.language
1336 self.category = obj.category
1337 super().__init__(obj)
1338 obj.stats = self
1340 @cached_property
1341 def has_review(self):
1342 return (
1343 self.category.project.source_review
1344 or self.category.project.translation_review
1345 )
1347 def get_category_objects(self):
1348 return [
1349 CategoryLanguage(category, self.language)
1350 for category in self.category.stats.get_category_objects()
1351 ]
1353 def get_child_objects(self):
1354 return self.language.translation_set.filter(
1355 component__category=self.category
1356 ).only("id", "language")
1359class CategoryStats(ParentAggregatingStats):
1360 def get_update_objects(self, *, full: bool = True) -> Generator[BaseStats]:
1361 if self._object.category:
1362 yield self._object.category.stats
1363 yield from self._object.category.stats.get_update_objects()
1364 else:
1365 # Global
1366 yield from self._object.project.stats.get_update_objects()
1368 def get_child_objects(self):
1369 return self._object.component_set.only("id", "slug", "category", "check_flags")
1371 def get_category_objects(self):
1372 return self._object.category_set.only("id", "slug", "category")
1374 def get_single_language_stats(self, language):
1375 return CategoryLanguageStats(CategoryLanguage(self._object, language))
1377 def get_language_stats(self):
1378 return prefetch_stats(
1379 self.get_single_language_stats(language)
1380 for language in self._object.languages
1381 )
1384class ProjectStats(ParentAggregatingStats):
1385 @cached_property
1386 def has_review(self):
1387 return self._object.enable_review
1389 def get_child_objects(self):
1390 own = self._object.component_set.only("id", "project", "check_flags")
1391 shared = self._object.shared_components.only("id", "project", "check_flags")
1392 if shared: 1392 ↛ 1393line 1392 didn't jump to line 1393 because the condition on line 1392 was never true
1393 return (own | shared).distinct()
1394 return own
1396 def get_single_language_stats(self, language: Language) -> ProjectLanguageStats:
1397 return ProjectLanguageStats(ProjectLanguage(self._object, language))
1399 def get_language_stats(self) -> list[ProjectLanguageStats]:
1400 return prefetch_stats(
1401 self.get_single_language_stats(language)
1402 for language in self._object.languages
1403 )
1405 def _calculate_basic(self) -> None:
1406 super()._calculate_basic()
1407 self.store("languages", self._object.languages.count())
1410class ComponentListStats(ParentAggregatingStats):
1411 def get_child_objects(self):
1412 return self._object.components.only(
1413 "id", "slug", "componentlist", "check_flags"
1414 )
1417class GlobalStats(ParentAggregatingStats):
1418 def __init__(self) -> None:
1419 super().__init__(None)
1421 def get_child_objects(self):
1422 from weblate.trans.models import Project
1424 return Project.objects.only("id", "slug")
1426 def _calculate_basic(self) -> None:
1427 super()._calculate_basic()
1428 self.store("languages", Language.objects.have_translation().count())
1430 @cached_property
1431 def cache_key(self) -> str:
1432 return "stats-global"
1434 def get_single_language_stats(self, language):
1435 return LanguageStats(language)
1437 def get_language_stats(self):
1438 return prefetch_stats(
1439 self.get_single_language_stats(language)
1440 for language in Language.objects.have_translation()
1441 )
1443 # The following fields are used in MetricsSerializer in API
1444 def get_languages(self):
1445 return Language.objects.count()
1447 def get_users(self):
1448 from weblate.auth.models import User
1450 return User.objects.count()
1452 def get_projects(self):
1453 from weblate.trans.models import Project
1455 return Project.objects.count()
1457 def get_components(self):
1458 from weblate.trans.models import Component
1460 return Component.objects.count()
1462 def get_translations(self):
1463 from weblate.trans.models import Translation
1465 return Translation.objects.count()
1467 def get_checks(self):
1468 from weblate.checks.models import Check
1470 return Check.objects.count()
1472 def get_configuration_errors(self):
1473 from weblate.wladmin.models import ConfigurationError
1475 return ConfigurationError.objects.filter(ignored=False).count()
1477 def get_suggestions(self):
1478 from weblate.trans.models import Suggestion
1480 return Suggestion.objects.count()
1482 def get_celery_queues(self):
1483 from weblate.utils.celery import get_queue_stats
1485 return get_queue_stats()
1487 def get_name(self):
1488 return settings.SITE_TITLE
1491class GhostStats(BaseStats):
1492 basic_keys = SOURCE_KEYS
1493 is_ghost = True
1495 def __init__(self, base=None) -> None:
1496 super().__init__(None)
1497 self.base = base
1499 @cached_property
1500 def pk(self):
1501 return get_random_identifier()
1503 def _calculate_basic(self) -> None:
1504 stats = zero_stats(self.basic_keys)
1505 if self.base is not None:
1506 for skey, dkey, tkey in (
1507 ("source_strings", "all", "todo"),
1508 ("source_words", "all_words", "todo_words"),
1509 ("source_chars", "all_chars", "todo_chars"),
1510 ):
1511 stats[tkey] = stats[dkey] = getattr(self.base, skey)
1512 for key, value in stats.items():
1513 self.store(key, value)
1515 @cached_property
1516 def cache_key(self) -> str:
1517 return "stats-zero"
1519 def save(self, update_parents: bool = True) -> None:
1520 pass
1522 def load(self) -> StatDict:
1523 return {}
1525 def get_absolute_url(self) -> str:
1526 return ""
1528 def base_obj(self) -> Project | Component:
1529 raise NotImplementedError
1532class GhostProjectLanguageStats(GhostStats):
1533 language: Language
1534 project: Project
1535 is_source: bool = False
1537 def __init__(self, project: Project, language: Language) -> None:
1538 super().__init__(project.stats)
1539 self.project = project
1540 self.language = language
1542 def base_obj(self):
1543 return self.project
1546class GhostCategoryLanguageStats(GhostStats):
1547 category: Category
1548 language: Language
1549 is_source: bool = False
1551 def __init__(self, category: Category, language: Language) -> None:
1552 super().__init__(category.stats)
1553 self.project = category.project
1554 self.category = category
1555 self.language = language
1557 def base_obj(self):
1558 return self.category