Coverage for app/venv/lib/python3.14/site-packages/weblate/memory/tasks.py: 27%

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.machinery.base import get_machinery_language 

11from weblate.memory.models import Memory 

12from weblate.memory.utils import is_valid_memory_entry 

13from weblate.utils.celery import app 

14from weblate.utils.state import STATE_APPROVED, STATE_TRANSLATED 

15 

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

17 from weblate.auth.models import User 

18 from weblate.lang.models import Language 

19 from weblate.trans.models import Component, Project, Unit 

20 

21 

22@app.task(trail=False) 

23def import_memory(project_id: int, component_id: int | None = None) -> None: 

24 from weblate.trans.models import Project, Unit 

25 

26 project = Project.objects.get(pk=project_id) 

27 

28 components = project.component_set.prefetch() 

29 if component_id: 

30 components = components.filter(id=component_id) 

31 

32 for component in components: 

33 component.log_info("updating translation memory") 

34 with transaction.atomic(): 

35 units = Unit.objects.filter( 

36 translation__component=component, 

37 state__gte=STATE_TRANSLATED, 

38 translation__component__is_glossary=False, 

39 ).exclude(target="") 

40 if not component.intermediate: 40 ↛ 44line 40 didn't jump to line 44 because the condition on line 40 was always true

41 units = units.exclude( 

42 translation__language_id=component.source_language_id 

43 ) 

44 for unit in units.prefetch_related("translation", "translation__language"): 44 ↛ 45line 44 didn't jump to line 45 because the loop on line 44 never started

45 handle_unit_translation_change(unit, None, component, project) 

46 

47 

48def handle_unit_translation_change( 

49 unit: Unit, 

50 user: User | None = None, 

51 component: Component | None = None, 

52 project: Project | None = None, 

53) -> None: 

54 if component is None: 

55 component = unit.translation.component 

56 if project is None: 

57 project = component.project 

58 

59 # Do not keep per-user memory for bots 

60 if user is None or (user and user.is_bot): 

61 user_id = None 

62 add_user = False 

63 else: 

64 user_id = user.id 

65 add_user = user.profile.contribute_personal_tm 

66 

67 source_language: Language = get_machinery_language(component.source_language) 

68 target_language: Language = get_machinery_language(unit.translation.language) 

69 source = unit.source 

70 target = unit.target 

71 origin = component.full_slug 

72 

73 if not is_valid_memory_entry(source=source, target=target): 

74 return 

75 

76 update_memory.delay_on_commit( 

77 source_language_id=source_language.id, 

78 target_language_id=target_language.id, 

79 source=source, 

80 target=target, 

81 origin=origin, 

82 add_shared=project.contribute_shared_tm, 

83 user_id=user_id, 

84 project_id=project.id, 

85 add_project=component.contribute_project_tm, 

86 add_user=add_user, 

87 unit_state=unit.state, 

88 context=unit.context or "", 

89 ) 

90 

91 

92@app.task(trail=False) 

93def update_memory( 

94 *, 

95 source_language_id: int, 

96 target_language_id: int, 

97 source: str, 

98 context: str, 

99 target: str, 

100 origin: str, 

101 add_shared: bool, 

102 add_project: bool, 

103 add_user: bool, 

104 user_id: int | None, 

105 project_id: int, 

106 unit_state: int, 

107) -> None: 

108 from weblate.trans.models import Project 

109 

110 project = Project.objects.get(pk=project_id) 

111 check_matching = True 

112 if (project.translation_review and unit_state == STATE_APPROVED) or ( 

113 not project.translation_review and unit_state >= STATE_TRANSLATED 

114 ): 

115 memory_status = Memory.STATUS_ACTIVE 

116 if project.autoclean_tm: 

117 # delete old entries, including those with different targets 

118 Memory.objects.filter( 

119 from_file=False, 

120 source=source, 

121 origin=origin, 

122 context=context, 

123 source_language_id=source_language_id, 

124 target_language_id=target_language_id, 

125 ).delete() 

126 check_matching = False 

127 else: 

128 memory_status = Memory.STATUS_PENDING 

129 to_create = [] 

130 to_update = [] 

131 

132 if check_matching: 

133 # Check matching entries in memory 

134 for matching in Memory.objects.filter( 

135 from_file=False, 

136 source=source, 

137 target=target, 

138 origin=origin, 

139 source_language_id=source_language_id, 

140 target_language_id=target_language_id, 

141 ): 

142 if matching.target == target and matching.status != memory_status: 

143 matching.status = memory_status 

144 to_update.append(matching) 

145 

146 if ( 

147 matching.user_id is None 

148 and matching.project_id == project_id 

149 and not matching.shared 

150 ): 

151 add_project = False 

152 elif ( 

153 add_shared 

154 and matching.user_id is None 

155 and matching.project_id is None 

156 and matching.shared 

157 ): 

158 add_shared = False 

159 elif ( 

160 add_user 

161 and matching.user_id == user_id 

162 and matching.project_id is None 

163 and not matching.shared 

164 ): 

165 add_user = False 

166 

167 if add_project: 

168 to_create.append( 

169 Memory( 

170 user=None, 

171 project_id=project_id, 

172 from_file=False, 

173 shared=False, 

174 source=source, 

175 context=context, 

176 target=target, 

177 origin=origin, 

178 source_language_id=source_language_id, 

179 target_language_id=target_language_id, 

180 status=memory_status, 

181 ) 

182 ) 

183 if add_shared: 

184 to_create.append( 

185 Memory( 

186 user=None, 

187 project=None, 

188 from_file=False, 

189 shared=True, 

190 source=source, 

191 context=context, 

192 target=target, 

193 origin=origin, 

194 source_language_id=source_language_id, 

195 target_language_id=target_language_id, 

196 status=memory_status, 

197 ) 

198 ) 

199 if add_user: 

200 to_create.append( 

201 Memory( 

202 user_id=user_id, 

203 project=None, 

204 from_file=False, 

205 shared=False, 

206 source=source, 

207 context=context, 

208 target=target, 

209 origin=origin, 

210 source_language_id=source_language_id, 

211 target_language_id=target_language_id, 

212 status=memory_status, 

213 ) 

214 ) 

215 

216 if to_create: 

217 Memory.objects.bulk_create(to_create) 

218 

219 if to_update: 

220 Memory.objects.bulk_update(to_update, fields=["status"])