notify: Make skip logic generic

There is already enough copies of the code, so let's make it generic.

weblate/accounts/migrations/0026_alter_subscription_notification.py

@@ -0,0 +1,45 @@
+# Copyright © Michal Čihař <[email protected]>
+#
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+# Generated by Django 4.2.1 on 2023-06-29 12:58
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("accounts", "0025_profile_theme"),
+    ]
+
+    operations = [
+        migrations.AlterField(
+            model_name="subscription",
+            name="notification",
+            field=models.CharField(
+                choices=[
+                    ("RepositoryNotification", "Repository operation"),
+                    ("LockNotification", "Component locking"),
+                    ("LicenseNotification", "Changed license"),
+                    ("ParseErrorNotification", "Parse error"),
+                    ("NewStringNotificaton", "New string"),
+                    ("NewContributorNotificaton", "New contributor"),
+                    ("NewSuggestionNotificaton", "New suggestion"),
+                    ("NewCommentNotificaton", "New comment"),
+                    ("MentionCommentNotificaton", "Mentioned in comment"),
+                    ("LastAuthorCommentNotificaton", "Comment on own translation"),
+                    ("TranslatedStringNotificaton", "Translated string"),
+                    ("ChangedStringNotificaton", "Changed string"),
+                    ("ApprovedStringNotificaton", "Approved string"),
+                    ("NewTranslationNotificaton", "New language"),
+                    ("NewComponentNotificaton", "New translation component"),
+                    ("NewAnnouncementNotificaton", "New announcement"),
+                    ("NewAlertNotificaton", "New alert"),
+                    ("MergeFailureNotification", "Repository failure"),
+                    ("PendingSuggestionsNotification", "Pending suggestions"),
+                    ("ToDoStringsNotification", "Unfinished strings"),
+                ],
+                max_length=100,
+            ),
+        ),
+    ]

weblate/accounts/notifications.py

@@ -5,7 +5,7 @@
 from collections import defaultdict
 from copy import copy
 from email.utils import formataddr
-from typing import Iterable, Optional
+from typing import Any, Iterable, Optional
 
 from dateutil.relativedelta import relativedelta
 from django.conf import settings
@@ -84,6 +84,7 @@ class Notification:
     ignore_watched: bool = False
     any_watched: bool = False
     required_attr: Optional[str] = None
+    skip_when_notify: Optional[Any] = None
 
     def __init__(self, outgoing, perm_cache=None):
         self.outgoing = outgoing
@@ -340,8 +341,17 @@ def send_immediate(
                 self.get_headers(context),
             )
 
+    def _convert_change_skip(self, change):
+        return change
+
     def should_skip(self, user, change):
-        return False
+        if self.skip_when_notify is None:
+            return False
+        if self.child_notify is None:
+            self.child_notify = self.skip_when_notify(None, self.perm_cache)
+        return bool(
+            list(self.child_notify.get_users(FREQ_INSTANT, change, users=[user.pk]))
+        )
 
     def notify_immediate(self, change):
         for user in self.get_users(FREQ_INSTANT, change):
@@ -415,28 +425,6 @@ def notify_monthly(self):
         self.notify_digest(FREQ_MONTHLY, self.filter_changes(months=1))
 
 
-@register_notification
-class MergeFailureNotification(Notification):
-    actions = (
-        Change.ACTION_FAILED_MERGE,
-        Change.ACTION_FAILED_REBASE,
-        Change.ACTION_FAILED_PUSH,
-    )
-    # Translators: Notification name
-    verbose = gettext_lazy("Repository failure")
-    template_name = "repository_error"
-
-    def should_skip(self, user, change):
-        fake = copy(change)
-        fake.action = Change.ACTION_ALERT
-        fake.alert = Alert(name="MergeFailure", details={"error": ""})
-        if self.child_notify is None:
-            self.child_notify = NewAlertNotificaton(None, self.perm_cache)
-        return bool(
-            list(self.child_notify.get_users(FREQ_INSTANT, fake, users=[user.pk]))
-        )
-
-
 @register_notification
 class RepositoryNotification(Notification):
     actions = (
@@ -517,35 +505,26 @@ class NewSuggestionNotificaton(Notification):
 
 
 @register_notification
-class LastAuthorCommentNotificaton(Notification):
+class NewCommentNotificaton(Notification):
     actions = (Change.ACTION_COMMENT,)
     # Translators: Notification name
-    verbose = gettext_lazy("Comment on own translation")
+    verbose = gettext_lazy("New comment")
     template_name = "new_comment"
-    ignore_watched = True
+    filter_languages = True
     required_attr = "comment"
 
-    def should_skip(self, user, change):
-        if self.child_notify is None:
-            self.child_notify = MentionCommentNotificaton(None, self.perm_cache)
-        return bool(
-            list(self.child_notify.get_users(FREQ_INSTANT, change, users=[user.pk]))
-        )
+    def get_language_filter(self, change, translation):
+        if not change.comment.unit.is_source:
+            return translation.language
+        return None
 
-    def get_users(
-        self,
-        frequency,
-        change=None,
-        project=None,
-        component=None,
-        translation=None,
-        users=None,
-    ):
-        last_author = change.unit.get_last_content_change()[0]
-        users = [] if last_author.is_anonymous else [last_author.pk]
-        return super().get_users(
-            frequency, change, project, component, translation, users
-        )
+    def notify_immediate(self, change):
+        super().notify_immediate(change)
+
+        # Notify upstream
+        report_source_bugs = change.component.report_source_bugs
+        if change.comment and change.comment.unit.is_source and report_source_bugs:
+            self.send_immediate("en", report_source_bugs, change)
 
 
 @register_notification
@@ -556,13 +535,7 @@ class MentionCommentNotificaton(Notification):
     template_name = "new_comment"
     ignore_watched = True
     required_attr = "comment"
-
-    def should_skip(self, user, change):
-        if self.child_notify is None:
-            self.child_notify = NewCommentNotificaton(None, self.perm_cache)
-        return bool(
-            list(self.child_notify.get_users(FREQ_INSTANT, change, users=[user.pk]))
-        )
+    skip_when_notify = NewCommentNotificaton
 
     def get_users(
         self,
@@ -588,26 +561,38 @@ def get_users(
 
 
 @register_notification
-class NewCommentNotificaton(Notification):
+class LastAuthorCommentNotificaton(Notification):
     actions = (Change.ACTION_COMMENT,)
     # Translators: Notification name
-    verbose = gettext_lazy("New comment")
+    verbose = gettext_lazy("Comment on own translation")
     template_name = "new_comment"
-    filter_languages = True
+    ignore_watched = True
     required_attr = "comment"
+    skip_when_notify = MentionCommentNotificaton
 
-    def get_language_filter(self, change, translation):
-        if not change.comment.unit.is_source:
-            return translation.language
-        return None
+    def get_users(
+        self,
+        frequency,
+        change=None,
+        project=None,
+        component=None,
+        translation=None,
+        users=None,
+    ):
+        last_author = change.unit.get_last_content_change()[0]
+        users = [] if last_author.is_anonymous else [last_author.pk]
+        return super().get_users(
+            frequency, change, project, component, translation, users
+        )
 
-    def notify_immediate(self, change):
-        super().notify_immediate(change)
 
-        # Notify upstream
-        report_source_bugs = change.component.report_source_bugs
-        if change.comment and change.comment.unit.is_source and report_source_bugs:
-            self.send_immediate("en", report_source_bugs, change)
+@register_notification
+class TranslatedStringNotificaton(Notification):
+    actions = (Change.ACTION_CHANGE, Change.ACTION_NEW)
+    # Translators: Notification name
+    verbose = gettext_lazy("Translated string")
+    template_name = "translated_string"
+    filter_languages = True
 
 
 @register_notification
@@ -617,22 +602,7 @@ class ChangedStringNotificaton(Notification):
     verbose = gettext_lazy("Changed string")
     template_name = "changed_translation"
     filter_languages = True
-
-    def should_skip(self, user, change):
-        if self.child_notify is None:
-            self.child_notify = TranslatedStringNotificaton(None, self.perm_cache)
-        return bool(
-            list(self.child_notify.get_users(FREQ_INSTANT, change, users=[user.pk]))
-        )
-
-
-@register_notification
-class TranslatedStringNotificaton(Notification):
-    actions = (Change.ACTION_CHANGE, Change.ACTION_NEW)
-    # Translators: Notification name
-    verbose = gettext_lazy("Translated string")
-    template_name = "translated_string"
-    filter_languages = True
+    skip_when_notify = TranslatedStringNotificaton
 
 
 @register_notification
@@ -720,6 +690,25 @@ def should_skip(self, user, change):
         return False
 
 
+@register_notification
+class MergeFailureNotification(Notification):
+    actions = (
+        Change.ACTION_FAILED_MERGE,
+        Change.ACTION_FAILED_REBASE,
+        Change.ACTION_FAILED_PUSH,
+    )
+    # Translators: Notification name
+    verbose = gettext_lazy("Repository failure")
+    template_name = "repository_error"
+    skip_when_notify = NewAlertNotificaton
+
+    def _convert_change_skip(self, change):
+        fake = copy(change)
+        fake.action = Change.ACTION_ALERT
+        fake.alert = Alert(name="MergeFailure", details={"error": ""})
+        return fake
+
+
 class SummaryNotification(Notification):
     filter_languages = True