edx-platform/openedx/core/djangoapps/notifications/utils.py

"""
Utils function for notifications app
"""
from typing import Dict, List

from common.djangoapps.student.models import CourseAccessRole
from openedx.core.djangoapps.django_comment_common.models import Role
from openedx.core.djangoapps.notifications.config.waffle import ENABLE_NOTIFICATIONS
from openedx.core.djangoapps.notifications.models import create_notification_preference, NotificationPreference
from openedx.core.lib.cache_utils import request_cached


def get_show_notifications_tray():
    """
    Returns whether notifications tray is enabled via waffle flag
    """
    return ENABLE_NOTIFICATIONS.is_enabled()


def get_list_in_batches(input_list, batch_size):
    """
    Divides the list of objects into list of list of objects each of length batch_size.
    """
    list_length = len(input_list)
    for index in range(0, list_length, batch_size):
        yield input_list[index: index + batch_size]


@request_cached()
def get_notification_types_with_visibility_settings() -> Dict[str, List[str]]:
    """
    Get notification types with their visibility settings.

    :return: List of dictionaries with notification type names and corresponding visibility settings
    """
    from .base_notification import COURSE_NOTIFICATION_TYPES

    notification_types_with_visibility_settings = {}
    for notification_type in COURSE_NOTIFICATION_TYPES.values():
        if notification_type.get('visible_to'):
            notification_types_with_visibility_settings[notification_type['name']] = notification_type['visible_to']

    return notification_types_with_visibility_settings


def filter_out_visible_notifications(
    user_preferences: dict,
    notifications_with_visibility: Dict[str, List[str]],
    user_forum_roles: List[str],
    user_course_roles: List[str]
) -> dict:
    """
    Filter out notifications visible to forum roles from user preferences.

    :param user_preferences: User preferences dictionary
    :param notifications_with_visibility: List of dictionaries with notification type names and
    corresponding visibility settings
    :param user_forum_roles: List of forum roles for the user
    :param user_course_roles: List of course roles for the user
    :return: Updated user preferences dictionary
    """
    for user_preferences_app, app_config in user_preferences.items():
        if 'notification_types' in app_config:
            # Iterate over the types to remove and pop them from the dictionary
            for notification_type, is_visible_to in notifications_with_visibility.items():
                is_visible = False
                for role in is_visible_to:
                    if role in user_forum_roles or role in user_course_roles:
                        is_visible = True
                        break
                if is_visible:
                    continue
                if notification_type in user_preferences[user_preferences_app]['notification_types']:
                    user_preferences[user_preferences_app]['notification_types'].pop(notification_type)
    return user_preferences


def clean_arguments(kwargs):
    """
    Returns query arguments from command line arguments
    """
    clean_kwargs = {}
    for key in ['app_name', 'notification_type', 'course_id']:
        if kwargs.get(key):
            clean_kwargs[key] = kwargs[key]
    if kwargs.get('created', {}):
        clean_kwargs.update(kwargs.get('created'))
    return clean_kwargs


def get_user_forum_access_roles(user_id: int) -> List[str]:
    """
    Get forum roles for the given user in all course.

    :param user_id: User ID
    :return: List of forum roles
    """
    return list(Role.objects.filter(users__id=user_id).values_list('name', flat=True))


def exclude_inaccessible_preferences(user_preferences: dict, user):
    """
    Exclude notifications from user preferences that the user has no access to,
    based on forum and course roles.

    :param user_preferences: Dictionary of user notification preferences
    :param user: Django User object
    :return: Updated user_preferences dictionary (modified in-place)
    """
    forum_roles = get_user_forum_access_roles(user.id)
    visible_notifications = get_notification_types_with_visibility_settings()
    course_roles = CourseAccessRole.objects.filter(
        user=user
    ).values_list('role', flat=True)

    filter_out_visible_notifications(
        user_preferences,
        visible_notifications,
        forum_roles,
        course_roles
    )
    return user_preferences


def _get_missing_preference_objects(user_ids, existing_prefs, target_types):
    """
    Compares existing data against target needs and returns
    a list of unsaved model instances.
    """
    already_exists = {f"{p.user_id}-{p.type}" for p in existing_prefs}

    to_create = []
    for user_id in user_ids:
        for n_type in target_types:
            key = f"{int(user_id)}-{n_type}"

            if key not in already_exists:
                new_obj = create_notification_preference(
                    user_id=int(user_id),
                    notification_type=n_type
                )
                to_create.append(new_obj)
                already_exists.add(key)

    return to_create


def create_account_notification_pref_if_not_exists(user_ids, existing_preferences, notification_types):
    """
    Ensures that NotificationPreference objects exist for the given user IDs
    and notification types. Creates any missing preferences in bulk.
    Args:
        user_ids: Iterable of user IDs to check/create preferences for.
        existing_preferences: QuerySet of existing NotificationPreference objects.
        notification_types: Iterable of notification type strings to ensure exist.
    Returns:
        List of NotificationPreference objects including both existing and newly created ones.
    """
    new_prefs_to_save = _get_missing_preference_objects(
        user_ids,
        existing_preferences,
        notification_types
    )

    if not new_prefs_to_save:
        return list(existing_preferences)

    NotificationPreference.objects.bulk_create(
        new_prefs_to_save,
        ignore_conflicts=True
    )

    return list(existing_preferences) + new_prefs_to_save