diff --git a/admin/notifications/views.py b/admin/notifications/views.py index e1c55c05f47..e307b60e8b6 100644 --- a/admin/notifications/views.py +++ b/admin/notifications/views.py @@ -11,6 +11,7 @@ from mako.lexer import Lexer from mako.parsetree import ControlLine import re +from string import Formatter def delete_selected_notifications(selected_ids): NotificationSubscription.objects.filter(id__in=selected_ids).delete() @@ -82,7 +83,7 @@ def generate_mock_json(structure, list_name=None): return result -def build_safe_context(template: str) -> dict: +def build_safe_context(template: str, subject: str) -> dict: templatenode = Lexer(text=template).parse() identifiers_location = [] for node in templatenode.get_children(): @@ -103,6 +104,9 @@ def build_safe_context(template: str) -> dict: mock_json = generate_mock_json(identifier_structure) context = {identifier: f'mock_{identifier}' for identifier in flatten_identifiers if identifier not in TEMPLATE_IDENTIFIER_BLACKLIST} context.update(mock_json) + + # subject + context.update({key: key for _, key, _, _ in Formatter().parse(subject) if key}) return context class NotificationsList(PermissionRequiredMixin, ListView): @@ -282,12 +286,12 @@ def get_context_data(self, *args, **kwargs): return kwargs else: if notification_type.is_digest_type: - inner_context = build_safe_context(notification_type.template) + inner_context = build_safe_context(notification_type.template, notification_type.subject) inner_template = _render_email_html(notification_type, ctx=inner_context, return_original_error=True) safe_context = {'notifications': [inner_template]} return_context = inner_context else: - safe_context = build_safe_context(notification_type.template) + safe_context = build_safe_context(notification_type.template, notification_type.subject) return_context = safe_context if notification_type.is_digest_type: @@ -300,6 +304,7 @@ def get_context_data(self, *args, **kwargs): except Exception as e: kwargs['rendered_template'] = f"Error rendering template: {str(e)}" + kwargs['rendered_subject'] = notification_type.subject.format(**safe_context) kwargs['context'] = json.dumps(return_context, indent=4) return kwargs diff --git a/admin/templates/notifications/notification_type_preview.html b/admin/templates/notifications/notification_type_preview.html index e9aefbe3284..14c9d507c3a 100644 --- a/admin/templates/notifications/notification_type_preview.html +++ b/admin/templates/notifications/notification_type_preview.html @@ -1,5 +1,6 @@

Notification Template Preview

Rendered Template

+

Subject: {{ rendered_subject|safe }}

diff --git a/osf/email/notification_campaign.py b/osf/email/notification_campaign.py new file mode 100644 index 00000000000..322bf82e13b --- /dev/null +++ b/osf/email/notification_campaign.py @@ -0,0 +1,221 @@ +import logging +from osf.models import NotificationType, NotificationTypeEnum, OSFUser, UserActivityCounter +from django.db.models import Q +from django.db.models import OuterRef, Subquery, Exists, F +from django.db.models.functions import Coalesce +from framework.celery_tasks import app as celery_app +from celery import chord +from django.utils import timezone +from osf.models.notification_campaign import NotificationCampaign, NotificationCampaignRecipient, NotificationCampaignStatus, NotificationCampaignRecipientStatus + +logger = logging.getLogger(__name__) + + +counter_subquery = ( + UserActivityCounter.objects + .filter(_id=OuterRef('guids___id')) + .values('total')[:1] +) + + +def get_filtered_batches(filters, batch_size=1000, campaign_id=None): + already_sent_subquery = NotificationCampaignRecipient.objects.filter( + campaign_id=campaign_id, + user_id=OuterRef('pk'), + ) + + qs = ( + OSFUser.objects + .annotate(already_sent=Exists(already_sent_subquery)) + .filter(already_sent=False) + .filter(**filters) + .annotate(activity_total=Coalesce(Subquery(counter_subquery), 0)) + .order_by('-activity_total', '-date_registered', '-id') + ) + + last_total = None + last_date = None + last_id = None + + while True: + batch_qs = qs + + if last_total is not None: + batch_qs = batch_qs.filter( + Q(activity_total__lt=last_total) | + Q(activity_total=last_total, date_registered__lt=last_date) | + Q(activity_total=last_total, date_registered=last_date, id__lt=last_id) + ) + + batch = batch_qs[:batch_size] + + if not batch: + break + + rows = list(batch.values_list('id', 'activity_total', 'date_registered')) + + if not rows: + break + + batch_ids = [r[0] for r in rows] + + last_id, last_total, last_date = ( + rows[-1][0], + rows[-1][1], + rows[-1][2], + ) + + yield batch_ids + + +FILTER_PRESETS = { + 'all': {}, + 'active': {'is_active': True}, + 'internal': {'is_active': True, 'is_staff': True, 'username__endswith': '@cos.io'}, +} + +@celery_app.task(name='email.process_campaign_retry') +def process_campaign_retry(*args, **kwargs): + + campaign_id = kwargs.get('campaign_id') + campaign = NotificationCampaign.objects.get(id=campaign_id) + failed_recipients = NotificationCampaignRecipient.objects.filter(campaign=campaign, status=NotificationCampaignRecipientStatus.FAILED) + max_retries = campaign.metadata.get('execution', {}).get('max_retries', 3) + batch_size = campaign.metadata.get('execution', {}).get('batch_size', 1000) + failed_recipients_count = failed_recipients.count() + if not failed_recipients_count: + campaign.status = NotificationCampaignStatus.COMPLETED + campaign.completed_at = timezone.now() + campaign.failed_count = 0 + campaign.save(update_fields=['status', 'completed_at', 'failed_count']) + return + + if campaign.retries < max_retries: + logger.info(f'Retrying {failed_recipients_count} failed recipients for campaign {campaign_id}') + filters = {'id__in': failed_recipients.values_list('user_id', flat=True)} + tasks = [] + for batch in get_filtered_batches(filters=filters, batch_size=batch_size, campaign_id=campaign_id): + tasks.append( + send_campaign_batch.s( + notification_type_name=campaign.notification_type.name, + recipients_ids=batch, + context=campaign.metadata.get('context', {}), + campaign_id=campaign_id, + ) + ) + chord(tasks)( + process_campaign_retry.s(campaign_id=campaign_id) + ) + campaign.retries += 1 + campaign.save(update_fields=['retries']) + else: + campaign.failed_count = failed_recipients_count + campaign.status = NotificationCampaignStatus.PARTIALLY_COMPLETED + campaign.completed_at = timezone.now() + campaign.save(update_fields=['status', 'completed_at', 'failed_count']) + + +@celery_app.task(name='email.start_notification_campaign') +def start_notification_campaign(campaign_id): + campaign = NotificationCampaign.objects.get(id=campaign_id) + filters = campaign.metadata.get('filters', {}) + context = campaign.metadata.get('context', {}) + notification_type_name = campaign.notification_type.name + + if hasattr(NotificationTypeEnum, notification_type_name): + del getattr(NotificationTypeEnum, notification_type_name).instance + + if predefined_filter_name := filters.get('predefined'): + filters = FILTER_PRESETS.get(predefined_filter_name, {}) + + tasks = [] + total_recipients = 0 + batch_size = campaign.metadata.get('execution', {}).get('batch_size', 1000) + for batch in get_filtered_batches(filters=filters, batch_size=batch_size): + tasks.append( + send_campaign_batch.s( + notification_type_name=notification_type_name, + recipients_ids=batch, + context=context, + campaign_id=campaign_id, + ) + ) + total_recipients += len(batch) + + campaign.recipient_count = total_recipients + campaign.save(update_fields=['recipient_count']) + + chord(tasks)( + process_campaign_retry.s(campaign_id=campaign_id) + ) + + +@celery_app.task(name='email.send_campaign_batch', ignore_result=False) +def send_campaign_batch(context, recipients_ids, notification_type_name='blank', campaign_id=None): + campaign = NotificationCampaign.objects.get(id=campaign_id) + if campaign.status == NotificationCampaignStatus.CANCELLED: + logger.warning(f"Campaign {campaign_id} was cancelled") + return + if hasattr(NotificationTypeEnum, notification_type_name): + notification_type = getattr(NotificationTypeEnum, notification_type_name).instance + else: + notification_type = NotificationType.objects.filter( + name=notification_type_name + ).first() # TODO cache + + if notification_type is None: + return + + recipients_qs = OSFUser.objects.filter(id__in=recipients_ids) + recipient_records = { + 'to_create': [], + 'to_update': [], + } + existing = { + r.user_id: r + for r in NotificationCampaignRecipient.objects.filter( + campaign_id=campaign_id, + user_id__in=recipients_ids, + ) + } + success_count = 0 + failure_count = 0 + + for recipient in recipients_qs: + recipient_record = existing.get(recipient.id) + + if recipient_record is None: + recipient_record = NotificationCampaignRecipient( + campaign_id=campaign_id, + user=recipient, + ) + operation = 'to_create' + else: + operation = 'to_update' + + try: + notification_type.emit( + user=recipient, + event_context=context, + save=False, # Too many write operations + ) + + recipient_record.status = NotificationCampaignRecipientStatus.SENT + recipient_record.error_message = None + recipient_records[operation].append(recipient_record) + success_count += 1 + + except Exception as exc: + logger.error(exc) # TODO update error + recipient_record.status = NotificationCampaignRecipientStatus.FAILED + recipient_record.error_message = str(exc) + recipient_records[operation].append(recipient_record) + + failure_count += 1 + pass + + NotificationCampaignRecipient.objects.bulk_create(recipient_records['to_create']) + NotificationCampaignRecipient.objects.bulk_update(recipient_records['to_update'], ['status', 'error_message']) + + NotificationCampaign.objects.filter(pk=campaign_id).update(sent_count=F('sent_count') + success_count, failed_count=F('failed_count') + failure_count) + logger.info('Batch finished') # TODO: add/update logs diff --git a/osf/migrations/0045_notificationcampaign_notificationcampaignrecipient_and_more.py b/osf/migrations/0045_notificationcampaign_notificationcampaignrecipient_and_more.py new file mode 100644 index 00000000000..886ff3b29e9 --- /dev/null +++ b/osf/migrations/0045_notificationcampaign_notificationcampaignrecipient_and_more.py @@ -0,0 +1,53 @@ +# Generated by Django 4.2.26 on 2026-07-09 11:38 + +from django.conf import settings +from django.db import migrations, models +import django.db.models.deletion + + +class Migration(migrations.Migration): + + dependencies = [ + ('osf', '0044_notification_scheduled'), + ] + + operations = [ + migrations.CreateModel( + name='NotificationCampaign', + fields=[ + ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('created_at', models.DateTimeField(auto_now_add=True)), + ('updated_at', models.DateTimeField(auto_now=True)), + ('name', models.CharField(max_length=255)), + ('status', models.CharField(choices=[('created', 'Created'), ('running', 'Running'), ('completed', 'Completed'), ('partially_completed', 'Partially Completed'), ('failed', 'Failed'), ('cancelled', 'Cancelled'), ('ended', 'Ended')], default='created', max_length=20)), + ('started_at', models.DateTimeField(blank=True, null=True)), + ('completed_at', models.DateTimeField(blank=True, null=True)), + ('metadata', models.JSONField(blank=True, default=dict)), + ('recipient_count', models.PositiveIntegerField(default=0)), + ('sent_count', models.PositiveIntegerField(default=0)), + ('failed_count', models.PositiveIntegerField(default=0)), + ('retries', models.PositiveIntegerField(default=0)), + ('created_by', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, to=settings.AUTH_USER_MODEL)), + ('notification_type', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to='osf.notificationtype')), + ], + ), + migrations.CreateModel( + name='NotificationCampaignRecipient', + fields=[ + ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')), + ('updated_at', models.DateTimeField(auto_now=True)), + ('status', models.CharField(choices=[('pending', 'Pending'), ('sent', 'Sent'), ('failed', 'Failed')], db_index=True, default='pending', max_length=20)), + ('error_message', models.TextField(blank=True, null=True)), + ('campaign', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='recipients', to='osf.notificationcampaign')), + ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)), + ], + options={ + 'unique_together': {('campaign', 'user')}, + }, + ), + migrations.AddField( + model_name='osfuser', + name='received_notification_campaigns', + field=models.ManyToManyField(through='osf.NotificationCampaignRecipient', to='osf.notificationcampaign'), + ), + ] diff --git a/osf/models/notification_campaign.py b/osf/models/notification_campaign.py new file mode 100644 index 00000000000..8c5aa22a3ab --- /dev/null +++ b/osf/models/notification_campaign.py @@ -0,0 +1,101 @@ +from django.db import models +from django.utils import timezone + + +class NotificationCampaignStatus(models.TextChoices): + CREATED = 'created', 'Created' + RUNNING = 'running', 'Running' + COMPLETED = 'completed', 'Completed' + PARTIALLY_COMPLETED = 'partially_completed', 'Partially Completed' + FAILED = 'failed', 'Failed' + CANCELLED = 'cancelled', 'Cancelled' + ENDED = 'ended', 'Ended' + +class NotificationCampaignRecipientStatus(models.TextChoices): + PENDING = 'pending', 'Pending' + SENT = 'sent', 'Sent' + FAILED = 'failed', 'Failed' + + +class NotificationCampaign(models.Model): + created_at = models.DateTimeField(auto_now_add=True) + updated_at = models.DateTimeField(auto_now=True) + + name = models.CharField(max_length=255) + + notification_type = models.ForeignKey( + 'NotificationType', + on_delete=models.PROTECT, + ) + created_by = models.ForeignKey( + 'OSFUser', + null=True, + on_delete=models.SET_NULL, + ) + + status = models.CharField( + max_length=20, + choices=NotificationCampaignStatus.choices, + default=NotificationCampaignStatus.CREATED, + ) + + started_at = models.DateTimeField(null=True, blank=True) + completed_at = models.DateTimeField(null=True, blank=True) + + metadata = models.JSONField(default=dict, blank=True) + + # metadata structure: + # { + # "filters": { + # ... + # }, + # "context": { + # ... + # }, + # "execution": { + # "batch_size": , + # "max_retries": , + # }, + # "template": , + # } + + recipient_count = models.PositiveIntegerField(default=0) + sent_count = models.PositiveIntegerField(default=0) + failed_count = models.PositiveIntegerField(default=0) + retries = models.PositiveIntegerField(default=0) + + def start(self): + from osf.email.notification_campaign import start_notification_campaign + self.status = NotificationCampaignStatus.RUNNING + self.started_at = timezone.now() + + self.sent_count = 0 + self.failed_count = 0 + self.recipient_count = 0 + self.retries = 0 + self.metadata.update({'template': self.notification_type.template}) + self.save() + start_notification_campaign.delay(campaign_id=self.id) + + +class NotificationCampaignRecipient(models.Model): + campaign = models.ForeignKey( + NotificationCampaign, + on_delete=models.CASCADE, + related_name='recipients', + ) + user = models.ForeignKey( + 'OSFUser', + on_delete=models.CASCADE, + ) + updated_at = models.DateTimeField(auto_now=True) + status = models.CharField( + max_length=20, + choices=NotificationCampaignRecipientStatus.choices, + default=NotificationCampaignRecipientStatus.PENDING, + db_index=True, + ) + error_message = models.TextField(null=True, blank=True) + + class Meta: + unique_together = ('campaign', 'user') diff --git a/osf/models/user.py b/osf/models/user.py index 5739813cd48..b1e95010278 100644 --- a/osf/models/user.py +++ b/osf/models/user.py @@ -58,6 +58,7 @@ from .session import UserSessionMap from .tag import Tag from .validators import validate_email, validate_social, validate_history_item, has_domain_in_user_fields_for_names +from .notification_campaign import NotificationCampaign, NotificationCampaignRecipient from osf.utils.datetime_aware_jsonfield import DateTimeAwareJSONField from osf.utils.fields import NonNaiveDateTimeField, LowercaseEmailField, ensure_str from osf.utils.names import impute_names @@ -398,6 +399,11 @@ class OSFUser(DirtyFieldsMixin, GuidMixin, BaseModel, AbstractBaseUser, Permissi notifications_configured = DateTimeAwareJSONField(default=dict, blank=True) + received_notification_campaigns = models.ManyToManyField( + NotificationCampaign, + through=NotificationCampaignRecipient, + ) + # The time at which the user agreed to our updated ToS and Privacy Policy (GDPR, 25 May 2018) accepted_terms_of_service = NonNaiveDateTimeField(null=True, blank=True) diff --git a/website/settings/defaults.py b/website/settings/defaults.py index dfc78bb07ab..f87c293f245 100644 --- a/website/settings/defaults.py +++ b/website/settings/defaults.py @@ -483,6 +483,7 @@ class CeleryConfig: 'api.share.utils', 'scripts.check_manual_restart_approval', 'scripts.enhanced_stuck_registration_audit', + 'osf.email.notification_campaign', } background_migration_modules = { @@ -604,6 +605,7 @@ class CeleryConfig: 'scripts.disable_removed_beat_tasks', 'osf.management.commands.delete_withdrawn_or_failed_registration_files', 'osf.management.commands.migrate_osfmetrics_fix_6to8', + 'osf.email.notification_campaign', ) # Modules that need metrics and release requirements diff --git a/website/templates/blank.html.mako b/website/templates/blank.html.mako index f41301c1d1c..ced2e33c787 100644 --- a/website/templates/blank.html.mako +++ b/website/templates/blank.html.mako @@ -1,5 +1,62 @@ -<%inherit file="notify_base.mako" /> + + + + + + + -<%def name="content()"> +<%page args=" + domain='', + osf_contact_email='support@osf.io' +"/> - + + + + + + + + + + + + + + + +
+ OSF +
+ + + + +
+
+ + + + +
+

+ Copyright © 2026 + Center For Open Science, All rights reserved. | + + Privacy Policy + +

+

+ Questions? + ${osf_contact_email} +

+
+
+ + diff --git a/website/templates/notify_base.mako b/website/templates/notify_base.mako index 816ae3fbcf0..fde50148660 100644 --- a/website/templates/notify_base.mako +++ b/website/templates/notify_base.mako @@ -23,7 +23,7 @@ node_absolute_url=None, notification_settings_url=None, osf_contact_email='support@osf.io', - year=2025 + year=2026 "/>

- Copyright © 2025 + Copyright © 2026 Center For Open Science, All rights reserved. | Privacy Policy