Coverage for app/venv/lib/python3.14/site-packages/weblate/trans/bulk.py: 9%

1# Copyright © Michal Čihař <michal@weblate.org> 

2# 

3# SPDX-License-Identifier: GPL-3.0-or-later 

4from __future__ import annotations 

5 

6from typing import TYPE_CHECKING 

7 

8from django.db import transaction 

9 

10from weblate.checks.flags import Flags 

11from weblate.trans.actions import ActionEvents 

12from weblate.trans.models import Component, Unit 

13from weblate.trans.models.label import TRANSLATION_LABELS 

14from weblate.trans.models.pending import PendingUnitChange 

15from weblate.utils.state import ( 

16 STATE_APPROVED, 

17 STATE_FUZZY, 

18 STATE_TRANSLATED, 

19) 

20 

21if TYPE_CHECKING: 21 ↛ 22line 21 didn't jump to line 22 because the condition on line 21 was never true

22 from django.db.models import QuerySet 

23 

24 from weblate.auth.models import User 

25 from weblate.trans.models import Label, Project 

26 from weblate.trans.models.unit import UnitQuerySet 

27 

28EDITABLE_STATES = {STATE_FUZZY, STATE_TRANSLATED, STATE_APPROVED} 

29 

30 

31def bulk_perform( # noqa: C901 

32 user: User | None, 

33 unit_set: UnitQuerySet, 

34 *, 

35 query: str, 

36 target_state: int | str, 

37 add_flags: str | Flags, 

38 remove_flags: str | Flags, 

39 add_labels: QuerySet[Label], 

40 remove_labels: QuerySet[Label], 

41 project: Project, 

42 components: QuerySet[Component] | list[Component] | None = None, 

43) -> int: 

44 matching = unit_set.search(query, project=project) 

45 if components is None: 

46 components = Component.objects.filter( 

47 id__in=matching.values_list("translation__component_id", flat=True) 

48 ) 

49 

50 if isinstance(target_state, str): 

51 target_state = int(target_state) 

52 if isinstance(add_flags, str): 

53 add_flags = Flags(add_flags) 

54 if isinstance(remove_flags, str): 

55 remove_flags = Flags(remove_flags) 

56 add_labels_pks = {label.pk for label in add_labels} 

57 remove_labels_pks = {label.pk for label in remove_labels} 

58 

59 update_source = add_flags or remove_flags or add_labels or remove_labels 

60 

61 updated = 0 

62 for component in components: 

63 prev_updated = updated 

64 component.batch_checks = True 

65 with transaction.atomic(): 

66 component_units = matching.filter(translation__component=component) 

67 

68 source_unit_ids = set() 

69 

70 if target_state == -1: 

71 # Only fetch source unit ids here 

72 source_unit_ids = set( 

73 component_units.values_list("source_unit_id", flat=True) 

74 ) 

75 else: 

76 to_update = [] 

77 source_units = [] 

78 unit_ids = list(component_units.values_list("id", flat=True)) 

79 # Generate changes for state change 

80 for unit in ( 

81 Unit.objects.filter(id__in=unit_ids, state__in=EDITABLE_STATES) 

82 .exclude(translation__filename="", state=target_state) 

83 .prefetch() 

84 .select_for_update() 

85 ): 

86 source_unit_ids.add(unit.source_unit_id) 

87 

88 if user is None or user.has_perm("unit.edit", unit): 

89 # Create change object for edit, update is done outside the loop 

90 unit.state = target_state 

91 unit.generate_change( 

92 user, user, ActionEvents.BULK_EDIT, check_new=False 

93 ) 

94 updated += 1 

95 to_update.append(unit) 

96 if unit.is_source: 

97 source_units.append(unit) 

98 

99 # Bulk update state 

100 Unit.objects.filter(pk__in=(unit.pk for unit in to_update)).update( 

101 state=target_state 

102 ) 

103 for unit in to_update: 

104 PendingUnitChange.store_unit_change(unit=unit, author=user) 

105 # Fire source_change event in bulk for source units 

106 for unit in source_units: 

107 # The change is already done in the database, we 

108 # need it here to recalculate state of translation 

109 # units 

110 unit.is_batch_update = True 

111 unit.source_unit_save() 

112 

113 if update_source and ( 

114 user is None or user.has_perm("source.edit", component) 

115 ): 

116 # Perform changes on the source units 

117 source_units = ( 

118 Unit.objects.filter(pk__in=source_unit_ids) 

119 .prefetch() 

120 .prefetch_bulk() 

121 ) 

122 if add_labels or remove_labels: 

123 source_units = source_units.prefetch_related("labels") 

124 for source_unit in source_units.select_for_update(): 

125 changed = False 

126 if add_flags or remove_flags: 

127 flags = Flags(source_unit.extra_flags) 

128 flags.merge(add_flags) 

129 flags.remove(remove_flags) 

130 new_flags = flags.format() 

131 if source_unit.extra_flags != new_flags: 

132 source_unit.is_batch_update = True 

133 source_unit.update_extra_flags(new_flags, user) 

134 changed = True 

135 

136 if add_labels or remove_labels: 

137 unit_label_pks = { 

138 label.pk for label in source_unit.labels.all() 

139 } 

140 

141 if add_labels_pks - unit_label_pks: 

142 source_unit.is_batch_update = True 

143 source_unit.labels.add(*add_labels) 

144 changed = True 

145 

146 if unit_label_pks & remove_labels_pks: 

147 source_unit.is_batch_update = True 

148 source_unit.labels.remove(*remove_labels) 

149 changed = True 

150 

151 if changed: 

152 updated += 1 

153 

154 # Handle translation labels 

155 translation_labels = [ 

156 label for label in remove_labels if label.name in TRANSLATION_LABELS 

157 ] 

158 if translation_labels: 

159 for unit in component_units.filter(labels__in=translation_labels): 

160 unit.labels.remove(*translation_labels) 

161 updated += 1 

162 

163 if prev_updated != updated: 

164 component.invalidate_cache() 

165 component.update_source_checks() 

166 component.run_batched_checks() 

167 

168 return updated