Coverage for app/venv/lib/python3.14/site-packages/weblate/addons/base.py: 31%
263 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 os
8import subprocess
9from contextlib import suppress
10from itertools import chain
11from typing import TYPE_CHECKING, Any, ClassVar, TypedDict, cast
13from django.conf import settings
14from django.core.exceptions import ValidationError
15from django.utils.functional import cached_property
16from django.utils.translation import gettext
18from weblate.addons.events import POST_CONFIGURE_EVENTS, AddonEvent
19from weblate.trans.exceptions import FileParseError
20from weblate.trans.models import Component
21from weblate.trans.templatetags.translations import format_json
22from weblate.trans.util import get_clean_env
23from weblate.utils import messages
24from weblate.utils.errors import report_error
25from weblate.utils.files import cleanup_error_message
26from weblate.utils.html import format_html_join_comma, list_to_tuples
27from weblate.utils.render import render_template
28from weblate.utils.validators import validate_filename
30if TYPE_CHECKING: 30 ↛ 31line 30 didn't jump to line 31 because the condition on line 30 was never true
31 from collections.abc import Generator
33 from django.forms.boundfield import BoundField
34 from django_stubs_ext import StrOrPromise
36 from weblate.addons.forms import BaseAddonForm
37 from weblate.addons.models import Addon, AddonActivityLog
38 from weblate.auth.models import AuthenticatedHttpRequest, User
39 from weblate.trans.models import Change, Project, Translation, Unit
42class CompatDict(TypedDict, total=False):
43 vcs: set[str]
44 file_format: set[str]
45 edit_template: set[bool]
48class BaseAddon:
49 """Base class for Weblate add-ons."""
51 events: ClassVar[set[AddonEvent]] = set()
52 settings_form: type[BaseAddonForm] | None = None
53 name = ""
54 compat: ClassVar[CompatDict] = {}
55 multiple = False
56 verbose: StrOrPromise = "Base add-on"
57 description: StrOrPromise = "Base add-on"
58 icon = "cog.svg"
59 project_scope = False
60 repo_scope = False
61 needs_component = False
62 has_summary = False
63 alert: str = ""
64 trigger_update = False
65 stay_on_create = False
66 user_name = ""
67 user_verbose = ""
69 def __init__(self, storage: Addon) -> None:
70 self.instance: Addon = storage
71 self.alerts: list[dict[str, str]] = []
72 self.extra_files: list[str] = []
74 def __repr__(self) -> str:
75 return f"<{self.__class__.__name__} instance={self.instance}>"
77 @cached_property
78 def doc_anchor(self) -> str:
79 return self.get_doc_anchor()
81 @classmethod
82 def get_doc_anchor(cls) -> str:
83 return "addon-{}".format(cls.name.replace(".", "-").replace("_", "-"))
85 @classmethod
86 def has_settings(cls) -> bool:
87 return cls.settings_form is not None
89 @classmethod
90 def get_identifier(cls) -> str:
91 return cls.name
93 @classmethod
94 def create_object(
95 cls,
96 *,
97 component: Component | None = None,
98 project: Project | None = None,
99 acting_user: User | None = None,
100 **kwargs,
101 ) -> Addon:
102 from weblate.addons.models import Addon
104 result = Addon(
105 project=project,
106 component=component,
107 name=cls.name,
108 acting_user=acting_user,
109 **kwargs,
110 )
112 result.addon_class = cls
113 return result
115 @classmethod
116 def create(
117 cls,
118 *,
119 component: Component | None = None,
120 project: Project | None = None,
121 run: bool = True,
122 acting_user: User | None = None,
123 **kwargs,
124 ) -> BaseAddon:
125 storage = cls.create_object(
126 component=component, project=project, acting_user=acting_user, **kwargs
127 )
128 storage.save(force_insert=True)
129 result = cls(storage)
130 result.post_configure(run=run)
131 return result
133 @classmethod
134 def get_add_form(
135 cls,
136 user: User | None,
137 *,
138 component: Component | None = None,
139 project: Project | None = None,
140 **kwargs,
141 ) -> BaseAddonForm | None:
142 """Return configuration form for adding new add-on."""
143 if cls.settings_form is None:
144 return None
145 storage = cls.create_object(
146 component=component, project=project, acting_user=user
147 )
148 instance = cls(storage)
149 return cls.settings_form(user, instance, **kwargs)
151 def get_settings_form(self, user: User | None, **kwargs) -> BaseAddonForm | None:
152 """Return configuration form for this add-on."""
153 if self.settings_form is None:
154 return None
155 if "data" not in kwargs:
156 kwargs["data"] = self.instance.configuration
157 return self.settings_form(user, self, **kwargs)
159 def show_setting_field(self, field: BoundField) -> bool:
160 return not field.is_hidden and field.value()
162 def get_setting_value(self, field: BoundField) -> StrOrPromise:
163 value = field.value()
164 if value is True:
165 return gettext("enabled")
166 if field_choices := getattr(field.field, "choices", None):
167 choices: dict[str, str] = {
168 str(choice): value for choice, value in field_choices
169 }
170 if isinstance(value, list):
171 return format_html_join_comma(
172 "{}",
173 list_to_tuples(choices.get(val, val) for val in value),
174 )
175 return choices.get(value, value)
176 return value
178 def get_settings_fields(self) -> list[tuple[StrOrPromise, StrOrPromise]]:
179 form = self.get_settings_form(None)
180 if form is None:
181 return []
183 return [
184 (
185 field.label,
186 self.get_setting_value(field),
187 )
188 for field in form
189 if self.show_setting_field(field)
190 ]
192 def configure(self, configuration: dict[str, Any]) -> None:
193 """Save configuration."""
194 self.instance.configuration = configuration
195 self.instance.save()
196 self.post_configure()
198 def post_configure(self, run: bool = True) -> None:
199 from weblate.addons.tasks import postconfigure_addon
201 self.instance.log_debug("configuring events for %s add-on", self.name)
203 # Configure events to current status
204 self.instance.configure_events(self.events)
206 if run:
207 if settings.CELERY_TASK_ALWAYS_EAGER:
208 postconfigure_addon(self.instance.pk, self.instance)
209 else:
210 postconfigure_addon.delay_on_commit(self.instance.pk)
212 def post_configure_run(self) -> None:
213 # Trigger post events to ensure direct processing
214 if component := self.instance.component:
215 if self.repo_scope and component.linked_component:
216 component = component.linked_component
217 self.post_configure_run_component(component)
219 if project := self.instance.project:
220 for component in project.component_set.iterator():
221 if self.can_install(component, None):
222 self.post_configure_run_component(component)
224 def post_configure_run_component(self, component: Component) -> None:
225 from weblate.addons.models import execute_addon_event
227 # Trigger post configure event for a VCS component
228 previous = component.repository.last_revision
229 if not (POST_CONFIGURE_EVENTS & self.events):
230 return
232 base_event_args = (self.instance, component, component)
233 if AddonEvent.EVENT_POST_COMMIT in self.events:
234 component.log_debug("running post_commit add-on: %s", self.name)
235 execute_addon_event(
236 *(base_event_args),
237 AddonEvent.EVENT_POST_COMMIT,
238 "post_commit",
239 (
240 component,
241 True,
242 ),
243 )
244 if AddonEvent.EVENT_POST_UPDATE in self.events:
245 component.log_debug("running post_update add-on: %s", self.name)
246 # The post_update typically operates on files, so make sure these are updated
247 component.commit_pending("add-on", None)
248 execute_addon_event(
249 *(base_event_args),
250 AddonEvent.EVENT_POST_UPDATE,
251 "post_update",
252 (component, "", False),
253 )
254 if AddonEvent.EVENT_COMPONENT_UPDATE in self.events:
255 component.log_debug("running component_update add-on: %s", self.name)
256 execute_addon_event(
257 *(base_event_args),
258 AddonEvent.EVENT_COMPONENT_UPDATE,
259 "component_update",
260 (component,),
261 )
262 if AddonEvent.EVENT_POST_PUSH in self.events:
263 component.log_debug("running post_push add-on: %s", self.name)
264 execute_addon_event(
265 *(base_event_args),
266 AddonEvent.EVENT_POST_PUSH,
267 "post_push",
268 (component,),
269 )
270 if AddonEvent.EVENT_DAILY in self.events:
271 component.log_debug("running daily add-on: %s", self.name)
272 execute_addon_event(
273 *(base_event_args), AddonEvent.EVENT_DAILY, "daily", (component,)
274 )
276 current = component.repository.last_revision
277 if previous != current:
278 component.log_debug(
279 "add-ons updated repository from %s to %s", previous, current
280 )
281 component.create_translations()
283 def post_uninstall(self) -> None:
284 pass
286 def save_state(self) -> None:
287 """Save add-on state information."""
288 self.instance.save(update_fields=["state"])
290 @classmethod
291 def can_install(cls, component: Component, user: User | None) -> bool: # noqa: ARG003
292 """Check whether add-on is compatible with given component."""
293 return all(
294 getattr(component, key) in cast("set", values)
295 for key, values in cls.compat.items()
296 )
298 def pre_push(
299 self, component: Component, activity_log_id: int | None = None
300 ) -> dict | None:
301 """Event handler before repository is pushed upstream."""
302 # To be implemented in a subclass
304 def post_push(
305 self, component: Component, activity_log_id: int | None = None
306 ) -> dict | None:
307 """Event handler after repository is pushed upstream."""
308 # To be implemented in a subclass
310 def pre_update(
311 self, component: Component, activity_log_id: int | None = None
312 ) -> dict | None:
313 """Event handler before repository is updated from upstream."""
314 # To be implemented in a subclass
316 def post_update(
317 self,
318 component: Component,
319 previous_head: str,
320 skip_push: bool,
321 activity_log_id: int | None = None,
322 ) -> dict | None:
323 """
324 Event handler after repository is updated from upstream.
326 :param str previous_head: HEAD of the repository prior to update, can
327 be blank on initial clone.
328 :param bool skip_push: Whether the add-on operation should skip pushing
329 changes upstream. Usually you can pass this to
330 underlying methods as ``commit_and_push`` or
331 ``commit_pending``.
332 """
333 # To be implemented in a subclass
335 def pre_commit(
336 self,
337 translation: Translation,
338 author: str,
339 store_hash: bool,
340 activity_log_id: int | None = None,
341 ) -> dict | None:
342 """Event handler before changes are committed to the repository."""
343 # To be implemented in a subclass
345 def post_commit(
346 self,
347 component: Component,
348 store_hash: bool,
349 activity_log_id: int | None = None,
350 ) -> dict | None:
351 """Event handler after changes are committed to the repository."""
352 # To be implemented in a subclass
354 def post_add(
355 self, translation: Translation, activity_log_id: int | None = None
356 ) -> dict | None:
357 """Event handler after new translation is added."""
358 # To be implemented in a subclass
360 def unit_pre_create(
361 self, unit: Unit, activity_log_id: int | None = None
362 ) -> dict | None:
363 """Event handler before new unit is created."""
364 # To be implemented in a subclass
366 def daily(
367 self, component: Component, activity_log_id: int | None = None
368 ) -> dict | None:
369 """Event handler daily."""
370 # To be implemented in a subclass
372 def component_update(
373 self, component: Component, activity_log_id: int | None = None
374 ) -> dict | None:
375 """Event handler for component update."""
376 # To be implemented in a subclass
378 def change_event(
379 self, change: Change, activity_log_id: int | None = None
380 ) -> dict | None:
381 """Event handler for change event."""
382 # To be implemented in a subclass
384 def execute_process(
385 self, component: Component, cmd: list[str], env: dict[str, str] | None = None
386 ) -> None:
387 component.log_debug("%s add-on exec: %s", self.name, " ".join(cmd))
388 try:
389 output = subprocess.check_output(
390 cmd,
391 env=get_clean_env(env),
392 cwd=component.full_path,
393 stderr=subprocess.STDOUT,
394 text=True,
395 )
396 component.log_debug("exec result: %s", output)
397 except (OSError, subprocess.CalledProcessError) as err:
398 output = getattr(err, "output", "")
399 component.log_error("failed to exec %s: %s", repr(cmd), err)
400 for line in output.splitlines():
401 component.log_error("program output: %s", line)
402 self.alerts.append(
403 {
404 "addon": self.name,
405 "command": " ".join(cmd),
406 "output": cleanup_error_message(output),
407 "error": str(err),
408 }
409 )
410 report_error("Add-on script error", project=component.project)
412 def trigger_alerts(self, component: Component) -> None:
413 if self.alerts:
414 component.add_alert(self.alert, occurrences=self.alerts)
415 self.alerts = []
416 else:
417 component.delete_alert(self.alert)
419 def commit_and_push(
420 self,
421 component: Component,
422 files: list[str] | None = None,
423 skip_push: bool = False,
424 ) -> bool:
425 if files is None:
426 files = list(
427 chain.from_iterable(
428 translation.filenames
429 for translation in component.translation_set.iterator()
430 )
431 )
432 files += self.extra_files
433 repository = component.repository
434 if not files or not repository.needs_commit(files):
435 return False
436 with repository.lock:
437 component.commit_files(
438 template=component.addon_message,
439 extra_context={"addon_name": self.verbose},
440 files=files,
441 skip_push=skip_push,
442 )
443 return True
445 def render_repo_filename(
446 self, template: str, translation: Translation
447 ) -> str | None:
448 component = translation.component
450 # Render the template
451 filename = render_template(template, translation=translation)
453 # Validate filename (not absolute or linking to parent dir)
454 try:
455 validate_filename(filename)
456 except ValidationError:
457 return None
459 # Absolute path
460 filename = os.path.join(component.full_path, filename)
462 # Check if parent directory exists
463 dirname = os.path.dirname(filename)
464 if not os.path.exists(dirname):
465 os.makedirs(dirname)
467 # Validate if there is not a symlink out of the tree
468 try:
469 component.repository.resolve_symlinks(dirname)
470 if os.path.exists(filename):
471 component.repository.resolve_symlinks(filename)
472 except ValueError:
473 component.log_error("refused to write out of repository: %s", filename)
474 return None
476 return filename
478 @classmethod
479 def pre_install(
480 cls, obj: Component | Project | None, request: AuthenticatedHttpRequest
481 ) -> None:
482 from weblate.trans.tasks import perform_update
484 if cls.trigger_update and isinstance(obj, Component):
485 perform_update.delay("Component", obj.pk, auto=True)
486 if obj.repo_needs_merge():
487 messages.warning(
488 request,
489 gettext(
490 "The repository is outdated, you might not get "
491 "expected results until you update it."
492 ),
493 )
495 @cached_property
496 def user(self) -> User:
497 """Weblate user used to track changes by this add-on."""
498 from weblate.auth.models import User
500 if not self.user_name or not self.user_verbose:
501 msg = f"{self.__class__.__name__} is missing user_name and user_verbose!"
502 raise ValueError(msg)
504 return User.objects.get_or_create_bot(
505 scope="addon",
506 name=self.user_name,
507 verbose=self.user_verbose,
508 )
510 def render_activity_log(self, activity: AddonActivityLog) -> str:
511 # The details might be empty for pending entries
512 result = activity.details.get("result")
513 if result is None:
514 return ""
515 if isinstance(result, str):
516 return result
517 if isinstance(result, dict):
518 return format_json(result)
519 return str(result)
522class UpdateBaseAddon(BaseAddon):
523 """
524 Base class for add-ons updating translation files.
526 It hooks to post update and commits all changed translations.
527 """
529 events: ClassVar[set[AddonEvent]] = {
530 AddonEvent.EVENT_POST_UPDATE,
531 }
533 @staticmethod
534 def iterate_translations(component: Component) -> Generator[Translation]:
535 for translation in component.translation_set.iterator():
536 if not translation.is_source or component.intermediate:
537 yield translation
539 def update_translations(self, component: Component, previous_head: str) -> None:
540 raise NotImplementedError
542 def post_update(
543 self,
544 component: Component,
545 previous_head: str,
546 skip_push: bool,
547 activity_log_id: int | None = None,
548 ) -> None:
549 # Ignore file parse error, it will be properly tracked as an alert
550 with component.repository.lock:
551 with suppress(FileParseError):
552 self.update_translations(component, previous_head)
553 self.commit_and_push(component, skip_push=skip_push)
556class ChangeBaseAddon(BaseAddon):
557 """Base class for add-ons that listen for Change notifications."""
559 events: ClassVar[set[AddonEvent]] = {
560 AddonEvent.EVENT_CHANGE,
561 }
563 multiple = False