core/domain/email_manager.py - develop - githubmirror/oppia

core/domain/email_manager.py
3 131 строка112 KB
ateeksh22
Fix #26623: Migrate datetime usage to utils functions (#26648)
05 авг 2026, 07:42
05 авг 2026, 07:421d13470
# coding: utf-8
#
# Copyright 2014 The Oppia Authors. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS-IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Config properties and functions for managing email notifications."""

from __future__ import annotations

import logging
import pathlib
import tempfile

from core import feconf, utils
from core.constants import constants
from core.domain import (
    change_domain,
    email_services,
    html_cleaner,
    platform_parameter_list,
    platform_parameter_registry,
    platform_parameter_services,
    rights_domain,
    story_domain,
    subscription_services,
    suggestion_registry,
    taskqueue_services,
    user_services,
)
from core.platform import models

from typing import (
    Callable,
    DefaultDict,
    Dict,
    Final,
    List,
    Mapping,
    Optional,
    Sequence,
    Set,
    Tuple,
    TypedDict,
    Union,
)

MYPY = False
if MYPY:  # pragma: no cover
    from mypy_imports import (
        app_identity_services,
        email_models,
        secrets_services,
        suggestion_models,
        transaction_services,
    )

email_models, suggestion_models = models.Registry.import_models(
    [models.Names.EMAIL, models.Names.SUGGESTION]
)
app_identity_services = models.Registry.import_app_identity_services()
transaction_services = models.Registry.import_transaction_services()
secrets_services = models.Registry.import_secrets_services()

PlatformParameterRegistry = platform_parameter_registry.Registry


NEW_CD_USER_EMAIL_DATA: Dict[str, Dict[str, str]] = {
    constants.CD_USER_RIGHTS_CATEGORY_REVIEW_TRANSLATION: {
        'task': 'review',
        'category': 'translations',
        'to_review': 'translation suggestions',
        'description_template': '%s language translations',
        'rights_message_template': (
            'review translation suggestions made by contributors in the %s '
            'language'
        ),
    },
    constants.CD_USER_RIGHTS_CATEGORY_REVIEW_QUESTION: {
        'task': 'review',
        'category': 'questions',
        'to_review': 'question suggestions',
        'description': 'questions',
        'rights_message': 'review question suggestions made by contributors',
    },
    constants.CD_USER_RIGHTS_CATEGORY_SUBMIT_QUESTION: {
        'task': 'submit',
        'category': 'questions',
        'to_submit': 'question suggestions',
        'description': 'questions',
        'rights_message': 'submit question suggestions',
    },
}

REMOVED_CD_USER_EMAIL_DATA: Dict[str, Dict[str, str]] = {
    constants.CD_USER_RIGHTS_CATEGORY_REVIEW_TRANSLATION: {
        'category': 'translation',
        'role_description_template': (
            'translation reviewer role in the %s language'
        ),
        'rights_message_template': (
            'review translation suggestions made by contributors in the %s '
            'language'
        ),
    },
    constants.CD_USER_RIGHTS_CATEGORY_REVIEW_QUESTION: {
        'category': 'question',
        'role_description': 'question reviewer role',
        'rights_message': 'review question suggestions made by contributors',
    },
    constants.CD_USER_RIGHTS_CATEGORY_SUBMIT_QUESTION: {
        'category': 'question',
        'role_description': 'question submitter role',
        'rights_message': 'submit question suggestions',
    },
}

EXPLORATION_ROLE_MANAGER: Final = 'manager rights'
EXPLORATION_ROLE_EDITOR: Final = 'editor rights'
EXPLORATION_ROLE_VOICE_ARTIST: Final = 'voice artist rights'
EXPLORATION_ROLE_PLAYTESTER: Final = 'playtest access'

EDITOR_ROLE_EMAIL_HTML_ROLES: Dict[str, str] = {
    rights_domain.ROLE_OWNER: EXPLORATION_ROLE_MANAGER,
    rights_domain.ROLE_EDITOR: EXPLORATION_ROLE_EDITOR,
    rights_domain.ROLE_VOICE_ARTIST: EXPLORATION_ROLE_VOICE_ARTIST,
    rights_domain.ROLE_VIEWER: EXPLORATION_ROLE_PLAYTESTER,
}

_EDITOR_ROLE_EMAIL_HTML_RIGHTS: Dict[str, str] = {
    'can_manage': '<li>Change the exploration permissions</li><br>',
    'can_edit': '<li>Edit the exploration</li><br>',
    'can_voiceover': '<li>Voiceover the exploration</li><br>',
    'can_play': '<li>View and playtest the exploration</li><br>',
}

# We don't include "can_voiceover" for managers and editors, since this is
# implied by the email description for "can_edit".
EDITOR_ROLE_EMAIL_RIGHTS_FOR_ROLE: Dict[str, str] = {
    EXPLORATION_ROLE_MANAGER: (
        '%s%s%s'
        % (
            _EDITOR_ROLE_EMAIL_HTML_RIGHTS['can_manage'],
            _EDITOR_ROLE_EMAIL_HTML_RIGHTS['can_edit'],
            _EDITOR_ROLE_EMAIL_HTML_RIGHTS['can_play'],
        )
    ),
    EXPLORATION_ROLE_EDITOR: (
        '%s%s'
        % (
            _EDITOR_ROLE_EMAIL_HTML_RIGHTS['can_edit'],
            _EDITOR_ROLE_EMAIL_HTML_RIGHTS['can_play'],
        )
    ),
    EXPLORATION_ROLE_VOICE_ARTIST: (
        '%s%s'
        % (
            _EDITOR_ROLE_EMAIL_HTML_RIGHTS['can_voiceover'],
            _EDITOR_ROLE_EMAIL_HTML_RIGHTS['can_play'],
        )
    ),
    EXPLORATION_ROLE_PLAYTESTER: _EDITOR_ROLE_EMAIL_HTML_RIGHTS['can_play'],
}

CONTRIBUTOR_DASHBOARD_REVIEWER_NOTIFICATION_EMAIL_DATA: Dict[str, str] = {
    'email_body_template': (
        'Hi %s,'
        '<br><br>'
        'There are new review opportunities that we think you might be '
        'interested in on the <a href="%s%s">Contributor Dashboard</a>. '
        'Here are some examples of contributions that have been waiting '
        'the longest for review:'
        '<br><br>'
        '<ul>%s</ul><br>'
        'Please take some time to review any of the above contributions (if '
        'they still need a review) or any other contributions on the '
        'dashboard. We appreciate your help!'
        '<br><br>'
        'Thanks again, and happy reviewing!<br>'
        '- The Oppia Contributor Dashboard Team'
        '<br><br>%s'
    ),
    'email_subject': 'Contributor Dashboard Reviewer Opportunities',
}

CURRICULUM_ADMIN_CHAPTER_NOTIFICATION_EMAIL_DATA: Dict[str, str] = {
    'overdue_chapters_template': (
        'The following stories have unpublished chapters which are behind '
        'schedule. Please publish them or adjust the planned publication date.'
        '<br><br>'
        '<ol>%s</ol>'
    ),
    'upcoming_chapters_template': (
        'The following stories have unpublished chapters which are due for '
        'publication in the next %s days. Please ensure they are published '
        'on or before the planned date or adjust the planned publication date.'
        '<br><br>'
        '<ol>%%s</ol>'
    )
    % constants.CHAPTER_PUBLICATION_NOTICE_PERIOD_IN_DAYS,
    'email_subject': 'Chapter Publication Notifications',
}

VOICEOVER_TECH_LEADS_REGENERATION_NOTIFICATION_EMAIL: Dict[str, str] = {
    'email_body_template': (
        'Hi Tech-Lead (cc Voiceover Admins),<br><br>'
        'There were some errors when synthesizing automatic voiceovers for '
        'the exploration <a href="%s">%s</a> in language %s, triggered on '
        '%s at %s (UTC).<br><br>'
        'The attached file contains detailed information about the errors. '
        'Please look into these errors and fix them, as follows:<br><br>'
        '<ul>'
        '<li>If you see connection or timeout issues (e.g., ServiceTimeout, '
        'ConnectionFailure, ServiceUnavailable, TooManyRequests, '
        'ServiceRedirectTemporary), try <a href="%s">regenerating the '
        'voiceovers</a> after 10-15 minutes.</li>'
        '<li>If you see content-related issues (e.g., BadRequest, '
        'EmbeddedModelError), analyze the error and collaborate with the '
        'creator team to fix the relevant content in order to help address '
        'the issue. Once resolved, proceed with regenerating the voiceovers.'
        '</li>'
        '<li>If you see Azure configuration issues (e.g., '
        'AuthenticationFailure, ServiceError, RuntimeError, Forbidden), '
        'follow the instructions in <a href="%s">this doc</a> for '
        'troubleshooting.</li>'
        '</ul><br>'
        'Once the issue is resolved, attempt to <a href="%s">regenerating '
        'the voiceovers</a> of this exploration for the failed content IDs '
        'using <a href="%s">this guide</a>. Consider regenerating voiceovers '
        'for the entire exploration if the majority of voiceover synthesis '
        'fails.<br><br>'
        'Thank you.'
    ),
    'email_subject': '[Attention needed] Automatic Voiceover Generation Failed',
}

VOICEOVER_ADMINS_REGENERATION_NOTIFICATION_EMAIL: Dict[str, str] = {
    'email_body_template': (
        'Hi Voiceover Admins,<br><br>'
        '%s has initiated the generation of automatic voiceovers for the '
        'exploration titled <a href="%s">%s</a> in language(s) %s.<br>'
        'This generation was completed at %s %s. '
        'Below is a summary of the voiceover synthesis status.<br><br>'
        '<ul>'
        '<li>Total contents for speech synthesis: %s.</li>'
        '<li>Successfully generated voiceovers: %s.</li>'
        '<li>Failed contents for voiceover generation: %s.</li>'
        '</ul>%s'
        'Thank you.'
    ),
    'email_subject': 'Report on Automatic Voiceovers Generated for %s',
}

VOICEOVER_ADMIN_GOOGLE_GROUP: Final = 'voiceovers-leads@oppia.org'
VOICEOVER_TECH_LEADS_GOOGLE_GROUP: Final = 'voiceover-tech-support@oppia.org'

HTML_FOR_SUGGESTION_DESCRIPTION: Dict[
    str, Dict[str, Union[str, Callable[[Dict[str, str]], Tuple[str, ...]]]]
] = {
    # The templates below are for listing the information for each suggestion
    # type offered on the Contributor Dashboard.
    'suggestion_template': {
        feconf.SUGGESTION_TYPE_TRANSLATE_CONTENT: (
            '<li>The following %s translation suggestion was submitted for '
            'review %s ago:<br>%s</li><br>'
        ),
        feconf.SUGGESTION_TYPE_ADD_QUESTION: (
            '<li>The following question suggestion was submitted for review '
            '%s ago:<br>%s</li><br>'
        ),
    },
    # Each suggestion type has a lambda function to retrieve the values needed
    # to populate the above suggestion template.
    'suggestion_template_values_getter_functions': {
        feconf.SUGGESTION_TYPE_TRANSLATE_CONTENT: (
            lambda values_dict: (
                values_dict['language'],
                values_dict['review_wait_time'],
                values_dict['suggestion_content'],
            )
        ),
        feconf.SUGGESTION_TYPE_ADD_QUESTION: (
            lambda values_dict: (
                values_dict['review_wait_time'],
                values_dict['suggestion_content'],
            )
        ),
    },
}

ADMIN_NOTIFICATION_FOR_REVIEWER_SHORTAGE_EMAIL_DATA: Dict[str, str] = {
    'email_body_template': (
        'Hi %s,'
        '<br><br>'
        'In the <a href="%s%s#/roles">admin roles page,</a> please add '
        'reviewers to the Contributor Dashboard community by entering their '
        'username(s) and allow reviewing for the suggestion types that need '
        'more reviewers bolded below.'
        '<br><br>%s'
        'Thanks so much - we appreciate your help!<br><br>'
        'Best Wishes!<br>'
        '- The Oppia Contributor Dashboard Team'
    ),
    'email_subject': 'Reviewers Needed for Contributor Dashboard',
    # The templates below are for listing the information for each suggestion
    # type that needs more reviewers. For translation languages there are two
    # templates to account for: whether one or multiple languages needs more
    # reviewers.
    'one_language_template': (
        'There have been <b>%s translation suggestions</b> created on the '
        '<a href="%s%s">Contributor Dashboard page</a> where there are not '
        'enough reviewers.<br><br>'
    ),
    'multi_language_template': (
        'There have been <b>translation suggestions</b> created on the '
        '<a href="%s%s">Contributor Dashboard page</a> in languages where '
        'there are not enough reviewers. The languages that need more '
        'reviewers are:'
        '<br><ul>%s</ul><br>'
    ),
    'question_template': (
        'There have been <b>question suggestions</b> created on the '
        '<a href="%s%s">Contributor Dashboard page</a> where there are not '
        'enough reviewers.<br><br>'
    ),
}

ADMIN_NOTIFICATION_FOR_SUGGESTIONS_NEEDING_REVIEW_EMAIL_DATA: Dict[str, str] = {
    'email_body_template': (
        'Hi %s,<br><br>'
        'There are suggestions on the <a href="%s%s">Contributor Dashboard</a> '
        'that have been waiting for more than %s days for review. Please take '
        'a look at the suggestions mentioned below and help them get reviewed '
        'by going to the <a href="%s%s">admin roles page</a> and either:'
        '<br><br><ul>'
        '<li>Add more reviewers to the suggestion types that have suggestions '
        'waiting too long for a review</li><br>'
        '<li>Find the existing reviewers and email reviewers directly about '
        'the suggestions waiting for a review</li><br>'
        '</ul><br>'
        'Here are the suggestions that have been waiting too long for a review:'
        '<br><br>'
        '<ul>%s</ul><br>'
        'Thanks so much - we appreciate your help!<br>'
        'Best Wishes!<br><br>'
        '- The Oppia Contributor Dashboard Team'
    ),
    'email_subject': (
        'Contributor Dashboard Suggestions Have Been Waiting Too Long for '
        'Review'
    ),
}

CONTRIBUTOR_RANK_ACHIEVEMENT_NOTIFICATION: Dict[
    str, Dict[str, Dict[str, str]]
] = {
    feconf.CONTRIBUTION_TYPE_TRANSLATION: {
        feconf.CONTRIBUTION_SUBTYPE_ACCEPTANCE: {
            'email_body_template': (
                'Hi %s,<br><br>'
                'This is to let you know that you have successfully achieved '
                'the %s rank for submitting translations in %s. Your efforts '
                'help Oppia grow better every day and support students around '
                'the world.<br><br>'
                'You can check all the achievements you earned in the '
                '<a href="%s%s">Contributor Dashboard</a>.<br><br>'
                'Best wishes and we hope you can continue to contribute!'
                '<br><br>'
                'The Oppia Contributor Dashboard Team'
            ),
            'email_subject': ('Oppia Translator Rank Achievement!'),
        },
        feconf.CONTRIBUTION_SUBTYPE_REVIEW: {
            'email_body_template': (
                'Hi %s,<br><br>'
                'This is to let you know that you have successfully achieved '
                'the %s rank for reviewing translations in %s. Your efforts '
                'help Oppia grow better every day and support students around '
                'the world.<br><br>'
                'You can check all the achievements you earned in the '
                '<a href="%s%s">Contributor Dashboard</a>.<br><br>'
                'Best wishes and we hope you can continue to contribute!'
                '<br><br>'
                'The Oppia Contributor Dashboard Team'
            ),
            'email_subject': ('Oppia Translation Reviewer Rank Achievement!'),
        },
        feconf.CONTRIBUTION_SUBTYPE_EDIT: {
            'email_body_template': (
                'Hi %s,<br><br>'
                'This is to let you know that you have successfully achieved '
                'the %s rank for correcting translations in %s. Your efforts '
                'help Oppia grow better every day and support students around '
                'the world.<br><br>'
                'You can check all the achievements you earned in the '
                '<a href="%s%s">Contributor Dashboard</a>.<br><br>'
                'Best wishes and we hope you can continue to contribute!'
                '<br><br>'
                'The Oppia Contributor Dashboard Team'
            ),
            'email_subject': ('Oppia Translation Reviewer Rank Achievement!'),
        },
    },
    feconf.CONTRIBUTION_TYPE_QUESTION: {
        feconf.CONTRIBUTION_SUBTYPE_ACCEPTANCE: {
            'email_body_template': (
                'Hi %s,<br><br>'
                'This is to let you know that you have successfully achieved '
                'the %s rank for submitting practice questions. Your efforts '
                'help Oppia grow better every day and support students around '
                'the world.<br><br>'
                'You can check all the achievements you earned in the '
                '<a href="%s%s">Contributor Dashboard</a>.<br><br>'
                'Best wishes and we hope you can continue to contribute!'
                '<br><br>'
                'The Oppia Contributor Dashboard Team'
            ),
            'email_subject': ('Oppia Question Submitter Rank Achievement!'),
        },
        feconf.CONTRIBUTION_SUBTYPE_REVIEW: {
            'email_body_template': (
                'Hi %s,<br><br>'
                'This is to let you know that you have successfully '
                'achieved the %s '
                'rank for reviewing  practice questions. Your efforts help '
                'Oppia grow better every day and support students around the '
                'world.<br><br>'
                'You can check all the achievements you earned in the '
                '<a href="%s%s">Contributor Dashboard</a>.<br><br>'
                'Best wishes and we hope you can continue to contribute!'
                '<br><br>'
                'The Oppia Contributor Dashboard Team'
            ),
            'email_subject': ('Oppia Question Reviewer Rank Achievement!'),
        },
        feconf.CONTRIBUTION_SUBTYPE_EDIT: {
            'email_body_template': (
                'Hi %s,<br><br>'
                'This is to let you know that you have successfully '
                'achieved the %s '
                'rank for correcting practice questions. Your efforts help '
                'Oppia grow better every day and support students around '
                'the world.<br><br>'
                'You can check all the achievements you earned in the '
                '<a href="%s%s">Contributor Dashboard</a>.<br><br>'
                'Best wishes and we hope you can continue to '
                'contribute!<br><br>'
                'The Oppia Contributor Dashboard Team'
            ),
            'email_subject': ('Oppia Question Reviewer Rank Achievement!'),
        },
    },
}

SENDER_VALIDATORS: Dict[str, Union[bool, Callable[[str], bool]]] = {
    feconf.EMAIL_INTENT_SIGNUP: (lambda x: x == feconf.SYSTEM_COMMITTER_ID),
    feconf.EMAIL_INTENT_UNPUBLISH_EXPLORATION: user_services.is_moderator,
    feconf.EMAIL_INTENT_DAILY_BATCH: (
        lambda x: x == feconf.SYSTEM_COMMITTER_ID
    ),
    feconf.EMAIL_INTENT_EDITOR_ROLE_NOTIFICATION: (
        lambda x: x == feconf.SYSTEM_COMMITTER_ID
    ),
    feconf.EMAIL_INTENT_FEEDBACK_MESSAGE_NOTIFICATION: (
        lambda x: x == feconf.SYSTEM_COMMITTER_ID
    ),
    feconf.EMAIL_INTENT_SUGGESTION_NOTIFICATION: (
        lambda x: x == feconf.SYSTEM_COMMITTER_ID
    ),
    feconf.EMAIL_INTENT_SUBSCRIPTION_NOTIFICATION: (
        lambda x: x == feconf.SYSTEM_COMMITTER_ID
    ),
    feconf.EMAIL_INTENT_MARKETING: (lambda x: x == feconf.SYSTEM_COMMITTER_ID),
    feconf.EMAIL_INTENT_DELETE_EXPLORATION: user_services.is_moderator,
    feconf.EMAIL_INTENT_REPORT_BAD_CONTENT: (
        lambda x: x == feconf.SYSTEM_COMMITTER_ID
    ),
    feconf.EMAIL_INTENT_ONBOARD_CD_USER: (
        lambda x: x == feconf.SYSTEM_COMMITTER_ID
    ),
    feconf.EMAIL_INTENT_REMOVE_CD_USER: (
        lambda x: x == feconf.SYSTEM_COMMITTER_ID
    ),
    feconf.EMAIL_INTENT_REVIEW_CREATOR_DASHBOARD_SUGGESTIONS: (
        lambda x: x == feconf.SYSTEM_COMMITTER_ID
    ),
    feconf.EMAIL_INTENT_NOTIFY_CURRICULUM_ADMINS_CHAPTERS: (
        lambda x: x == feconf.SYSTEM_COMMITTER_ID
    ),
    feconf.EMAIL_INTENT_ADDRESS_CONTRIBUTOR_DASHBOARD_SUGGESTIONS: (
        lambda x: x == feconf.SYSTEM_COMMITTER_ID
    ),
    feconf.EMAIL_INTENT_REVIEW_CONTRIBUTOR_DASHBOARD_SUGGESTIONS: (
        lambda x: x == feconf.SYSTEM_COMMITTER_ID
    ),
    feconf.EMAIL_INTENT_NOTIFY_CONTRIBUTOR_DASHBOARD_ACHIEVEMENTS: (
        lambda x: x == feconf.SYSTEM_COMMITTER_ID
    ),
    feconf.EMAIL_INTENT_ADD_CONTRIBUTOR_DASHBOARD_REVIEWERS: (
        lambda x: x == feconf.SYSTEM_COMMITTER_ID
    ),
    feconf.EMAIL_INTENT_ACCOUNT_DELETED: (
        lambda x: x == feconf.SYSTEM_COMMITTER_ID
    ),
    feconf.EMAIL_INTENT_VOICEOVER_REGENERATION: (
        lambda x: x == feconf.SYSTEM_COMMITTER_ID
    ),
}


class FeedbackMessagesDict(TypedDict):
    """Dictionary representing the feedback message for email."""

    title: str
    messages: List[str]


def require_sender_id_is_valid(intent: str, sender_id: str) -> None:
    """Ensure that the sender ID is valid, based on the email's intent.

    Many emails are only allowed to be sent by a certain user or type of user,
    e.g. 'admin' or an admin/moderator. This function will raise an exception
    if the given sender is not allowed to send this type of email.

    Args:
        intent: str. The intent string, i.e. the purpose of the email.
            Valid intent strings are defined in feconf.py.
        sender_id: str. The ID of the user sending the email.

    Raises:
        Exception. The email intent is invalid.
        Exception. The sender_id is not appropriate for the given intent.
    """

    if intent not in SENDER_VALIDATORS:
        raise Exception('Invalid email intent string: %s' % intent)

    sender_validator_fn = SENDER_VALIDATORS[intent]
    # Ruling out the possibility of bool for mypy type checking. Because here,
    # the return value of 'SENDER_VALIDATORS[intent]' is always called.
    assert callable(sender_validator_fn)

    if not sender_validator_fn(sender_id):
        logging.error(
            'Invalid sender_id %s for email with intent \'%s\''
            % (sender_id, intent)
        )
        raise Exception(
            'Invalid sender_id for email with intent \'%s\'' % intent
        )


def get_rendered_email_footer() -> str:
    """Returns the email footer with its preferences-page URL resolved.

    EMAIL_FOOTER may contain
    feconf.EMAIL_FOOTER_PREFERENCES_LINK_PLACEHOLDER as a placeholder
    for the preferences-page URL. A footer without the placeholder is
    returned unchanged.

    Returns:
        str. The rendered email footer.
    """
    email_footer = platform_parameter_services.get_platform_parameter_value(
        platform_parameter_list.ParamName.EMAIL_FOOTER.value
    )
    assert isinstance(email_footer, str)

    if feconf.EMAIL_FOOTER_PREFERENCES_LINK_PLACEHOLDER not in email_footer:
        return email_footer

    oppia_site_url_for_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS.value
        )
    )
    assert isinstance(oppia_site_url_for_emails, str)

    preferences_url = oppia_site_url_for_emails + feconf.PREFERENCES_URL
    return email_footer.replace(
        feconf.EMAIL_FOOTER_PREFERENCES_LINK_PLACEHOLDER,
        preferences_url,
    )


def _send_email(
    recipient_id: str,
    sender_id: str,
    intent: str,
    email_subject: str,
    email_html_body: str,
    sender_email: str,
    bcc_admin: bool = False,
    cc_emails: Optional[List[str]] = None,
    sender_name: Optional[str] = None,
    recipient_email: Optional[str] = None,
    attachments: Optional[List[Dict[str, str]]] = None,
) -> None:
    """Sends an email to the given recipient.

    This function should be used for sending all user-facing emails.

    Raises an Exception if the sender_id is not appropriate for the given
    intent. Currently we support only system-generated emails and emails
    initiated by moderator actions.

    Args:
        recipient_id: str. The user ID of the recipient.
        sender_id: str. The user ID of the sender.
        intent: str. The intent string for the email, i.e. the purpose/type.
        email_subject: str. The subject of the email.
        email_html_body: str. The body (message) of the email.
        sender_email: str. The sender's email address.
        bcc_admin: bool. Whether to send a copy of the email to the admin's
            email address.
        cc_emails: list(str)|None. Optional argument. A list of email addresses
            to be CCed in the email. If None, no emails will be CCed.
        sender_name: str or None. The name to be shown in the "sender" field of
            the email.
        recipient_email: str or None. Override for the recipient email.
            This should only be used when the user with user_id equal to
            recipient_id does not exist or is deleted and their email cannot be
            retrieved via get_email_from_user_id.
        attachments: list(dict)|None. Optional argument. A list of
            dictionaries, where each dictionary includes the keys `filename`
            and `path` with their corresponding values.
    """

    if sender_name is None:
        email_sender_name = (
            platform_parameter_services.get_platform_parameter_value(
                platform_parameter_list.ParamName.EMAIL_SENDER_NAME.value
            )
        )
        assert isinstance(email_sender_name, str)
        sender_name = email_sender_name

    require_sender_id_is_valid(intent, sender_id)

    if recipient_email is None:
        recipient_email_address = user_services.get_email_from_user_id(
            recipient_id
        )
    else:
        recipient_email_address = recipient_email

    cleaned_html_body = html_cleaner.clean(email_html_body)
    if cleaned_html_body != email_html_body:
        logging.error(
            'Original email HTML body does not match cleaned HTML body:\n'
            'Original:\n%s\n\nCleaned:\n%s\n'
            % (email_html_body, cleaned_html_body)
        )
        return

    raw_plaintext_body = (
        cleaned_html_body.replace('<br/>', '\n')
        .replace('<br>', '\n')
        .replace('<li>', '<li>- ')
        .replace('</p><p>', '</p>\n<p>')
    )
    cleaned_plaintext_body = html_cleaner.strip_html_tags(raw_plaintext_body)

    if email_models.SentEmailModel.check_duplicate_message(
        recipient_id, email_subject, cleaned_plaintext_body
    ):
        logging.error(
            'Duplicate email:\n'
            'Details:\n%s %s\n%s\n\n'
            % (recipient_id, email_subject, cleaned_plaintext_body)
        )
        return

    @transaction_services.run_in_transaction_wrapper
    def _send_email_transactional() -> None:
        """Sends the email to a single recipient."""
        sender_name_email = '%s <%s>' % (sender_name, sender_email)

        try:
            email_services.send_mail(
                sender_name_email,
                recipient_email_address,
                email_subject,
                cleaned_plaintext_body,
                cleaned_html_body,
                cc_emails=cc_emails,
                bcc_admin=bcc_admin,
                attachments=attachments,
            )
        except Exception as e:
            logging.error(
                'Email to %s failed to send: %s. Enqueuing for retry.',
                recipient_email_address,
                e,
            )

            payload = {
                'sender_email': sender_name_email,
                'recipient_id': recipient_email_address,
                'subject': email_subject,
                'html_body': cleaned_html_body,
                'text_body': cleaned_plaintext_body,
            }

            taskqueue_services.enqueue_task(
                feconf.TASK_URL_RETRY_FAILED_EMAIL, payload, 0
            )

        email_models.SentEmailModel.create(
            recipient_id,
            recipient_email_address,
            sender_id,
            sender_name_email,
            intent,
            email_subject,
            cleaned_html_body,
            utils.get_current_utc_datetime(),
        )

    _send_email_transactional()


def send_dummy_mail_to_admin(username: str) -> None:
    """Send an email from the specified email address to admin.

    Args:
        username: str. Username of the sender.
    """

    email_body = 'This is a test mail from %s.' % (username)
    email_subject = 'Test Mail'
    system_name_email = '%s <%s>' % (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SYSTEM_EMAIL_NAME.value
        ),
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SYSTEM_EMAIL_ADDRESS.value
        ),
    )
    admin_email_address = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.ADMIN_EMAIL_ADDRESS.value
        )
    )
    assert isinstance(admin_email_address, str)

    email_services.send_mail(
        system_name_email,
        admin_email_address,
        email_subject,
        email_body,
        email_body.replace('\n', '<br/>'),
        bcc_admin=False,
    )


def send_mail_to_admin(email_subject: str, email_body: str) -> None:
    """Send an email to the admin email address.

    The email is sent to the ADMIN_EMAIL_ADDRESS set in feconf.py.

    Args:
        email_subject: str. Subject of the email.
        email_body: str. Body (message) of the email.
    """

    app_id = app_identity_services.get_application_id()
    body = '(Sent from %s)\n\n%s' % (app_id, email_body)
    system_name_email = '%s <%s>' % (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SYSTEM_EMAIL_NAME.value
        ),
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SYSTEM_EMAIL_ADDRESS.value
        ),
    )
    admin_email_address = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.ADMIN_EMAIL_ADDRESS.value
        )
    )
    assert isinstance(admin_email_address, str)
    email_services.send_mail(
        system_name_email,
        admin_email_address,
        email_subject,
        body,
        body.replace('\n', '<br/>'),
        bcc_admin=False,
    )


def send_post_signup_email(
    user_id: str, test_for_duplicate_email: bool = False
) -> None:
    """Sends a post-signup email to the given user.

    Raises an exception if emails are not allowed to be sent to users (i.e.
    SERVER_CAN_SEND_EMAILS platform parameter is False).

    Args:
        user_id: str. User ID of the user that signed up.
        test_for_duplicate_email: bool. For testing duplicate emails.
    """

    email_subject_content_param_name = (
        platform_parameter_list.ParamName.SIGNUP_EMAIL_SUBJECT_CONTENT.value
    )
    email_subject_content = (
        platform_parameter_services.get_platform_parameter_value(
            email_subject_content_param_name
        )
    )
    # Here we use assert because the get_platform_parameter_value returns
    # value of type platform_parameter_domain.PlatformDataTypes and we
    # are sure that email_subject_content is of type str. This helps us
    # avoid the mypy error.
    assert isinstance(email_subject_content, str)
    email_body_content_param_name = (
        platform_parameter_list.ParamName.SIGNUP_EMAIL_BODY_CONTENT.value
    )
    email_body_content = (
        platform_parameter_services.get_platform_parameter_value(
            email_body_content_param_name
        )
    )
    assert isinstance(email_body_content, str)
    if not test_for_duplicate_email:
        email_subject_content_default_value = (
            platform_parameter_registry.Registry.get_platform_parameter(
                email_subject_content_param_name
            ).default_value
        )
        email_body_content_default_value = (
            platform_parameter_registry.Registry.get_platform_parameter(
                email_body_content_param_name
            ).default_value
        )
        if not email_subject_content or (
            email_subject_content_default_value == email_subject_content
        ):
            logging.error(
                'Please ensure that the value for the admin platform '
                'property SIGNUP_EMAIL_SUBJECT_CONTENT is set, before allowing '
                'post-signup emails to be sent.'
            )
            return
        if not email_body_content or (
            email_body_content_default_value == email_body_content
        ):
            logging.error(
                'Please ensure that the value for the admin platform '
                'property SIGNUP_EMAIL_BODY_CONTENT is set, before allowing '
                'post-signup emails to be sent.'
            )
            return

    recipient_username = user_services.get_username(user_id)
    email_footer = get_rendered_email_footer()
    email_body = 'Hi %s,<br><br>%s<br><br>%s' % (
        recipient_username,
        email_body_content,
        email_footer,
    )
    noreply_email_address = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
        )
    )
    assert isinstance(noreply_email_address, str)

    _send_email(
        user_id,
        feconf.SYSTEM_COMMITTER_ID,
        feconf.EMAIL_INTENT_SIGNUP,
        email_subject_content,
        email_body,
        noreply_email_address,
    )


def get_moderator_unpublish_exploration_email() -> str:
    """Returns a draft of the text of the body for an email sent immediately
    when a moderator unpublishes an exploration. An empty body is a signal to
    the frontend that no email will be sent.

    Returns:
        str. Draft of the email body for an email sent after the moderator
        unpublishes an exploration, or an empty string if no email should
        be sent.
    """

    try:
        require_moderator_email_prereqs_are_satisfied()
    except utils.ValidationError:
        return ''

    unpublish_exp_email_html_body = platform_parameter_services.get_platform_parameter_value(
        platform_parameter_list.ParamName.UNPUBLISH_EXPLORATION_EMAIL_HTML_BODY.value
    )
    # Ruling out the possibility of Any for mypy type checking.
    assert isinstance(unpublish_exp_email_html_body, str)
    return unpublish_exp_email_html_body


def require_moderator_email_prereqs_are_satisfied() -> None:
    """Raises an exception if, for any reason, moderator emails cannot be sent.

    Raises:
        ValidationError. The SERVER_CAN_SEND_EMAILS platform parameter is False.
    """

    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        raise utils.ValidationError(
            'For moderator emails to be sent, please ensure that '
            'SERVER_CAN_SEND_EMAILS is set to True.'
        )


def send_moderator_action_email(
    sender_id: str,
    recipient_id: str,
    intent: str,
    exploration_title: str,
    email_body: str,
) -> None:
    """Sends a email immediately following a moderator action (unpublish,
    delete) to the given user.

    Raises an exception if emails are not allowed to be sent to users (i.e.
    SERVER_CAN_SEND_EMAILS platform parameter is False).

    Args:
        sender_id: str. User ID of the sender.
        recipient_id: str. User ID of the recipient.
        intent: str. The intent string (cause/purpose) of the email.
        exploration_title: str. The title of the exploration on which the
            moderator action was taken.
        email_body: str. The email content/message.
    """

    require_moderator_email_prereqs_are_satisfied()
    email_config = feconf.VALID_MODERATOR_ACTIONS[intent]

    recipient_username = user_services.get_username(recipient_id)
    sender_username = user_services.get_username(sender_id)
    email_subject_fn = email_config['email_subject_fn']
    # Ruling out the possibility of str for mypy type checking. Because here,
    # the return value of 'email_config['email_subject_fn']' is always called.
    assert callable(email_subject_fn)
    email_subject = email_subject_fn(exploration_title)
    email_salutation_html_fn = email_config['email_salutation_html_fn']
    # Ruling out the possibility of str for mypy type checking. Because here,
    # the return value of 'email_config['email_salutation_html_fn']' is always
    # called.
    assert callable(email_salutation_html_fn)
    email_salutation_html = email_salutation_html_fn(recipient_username)
    email_signoff_html_fn = email_config['email_signoff_html_fn']
    # Ruling out the possibility of str for mypy type checking. Because here,
    # the return value of 'email_config['email_signoff_html_fn']' is always
    # called.
    assert callable(email_signoff_html_fn)
    email_signoff_html = email_signoff_html_fn(sender_username)
    email_footer = get_rendered_email_footer()
    full_email_content = '%s<br><br>%s<br><br>%s<br><br>%s' % (
        email_salutation_html,
        email_body,
        email_signoff_html,
        email_footer,
    )
    system_email_address = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SYSTEM_EMAIL_ADDRESS.value
        )
    )
    assert isinstance(system_email_address, str)
    _send_email(
        recipient_id,
        sender_id,
        intent,
        email_subject,
        full_email_content,
        system_email_address,
        bcc_admin=True,
    )


def send_role_notification_email(
    inviter_id: str,
    recipient_id: str,
    recipient_role: str,
    exploration_id: str,
    exploration_title: str,
) -> None:
    """Sends a email when a new user is given activity rights (Manager, Editor,
    Viewer) to an exploration by creator of exploration.

    Email will only be sent if recipient wants to receive these emails (i.e.
    'can_receive_editor_role_email' is set True in recipent's preferences).

    Args:
        inviter_id: str. ID of the user who invited the recipient to the new
            role.
        recipient_id: str. User ID of the recipient.
        recipient_role: str. Role given to the recipient. Must be defined in
            EDITOR_ROLE_EMAIL_HTML_ROLES.
        exploration_id: str. ID of the exploration for which the recipient has
            been given the new role.
        exploration_title: str. Title of the exploration for which the recipient
            has been given the new role.

    Raises:
        Exception. The role is invalid (i.e. not defined in
            EDITOR_ROLE_EMAIL_HTML_ROLES).
    """

    # Editor role email body and email subject templates.
    email_subject_template = '%s - invitation to collaborate'

    email_body_template = (
        'Hi %s,<br>'
        '<br>'
        '<b>%s</b> has granted you %s to their exploration, '
        '"<a href="https://www.oppia.org/create/%s">%s</a>", on Oppia.org.<br>'
        '<br>'
        'This allows you to:<br>'
        '<ul>%s</ul>'
        'You can find the exploration '
        '<a href="https://www.oppia.org/create/%s">here</a>.<br>'
        '<br>'
        'Thanks, and happy collaborating!<br>'
        '<br>'
        'Best wishes,<br>'
        'The Oppia Team<br>'
        '<br>%s'
    )

    # Return from here if sending email is turned off.
    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return

    # Return from here is sending editor role email is disabled.
    if not feconf.CAN_SEND_TRANSACTIONAL_EMAILS:
        logging.error('This app cannot send editor role emails to users.')
        return

    recipient_username = user_services.get_username(recipient_id)
    inviter_username = user_services.get_username(inviter_id)
    recipient_preferences = user_services.get_email_preferences(recipient_id)

    if not recipient_preferences.can_receive_editor_role_email:
        # Do not send email if recipient has declined.
        return

    if recipient_role not in EDITOR_ROLE_EMAIL_HTML_ROLES:
        raise Exception('Invalid role: %s' % recipient_role)

    role_description = EDITOR_ROLE_EMAIL_HTML_ROLES[recipient_role]
    rights_html = EDITOR_ROLE_EMAIL_RIGHTS_FOR_ROLE[role_description]

    email_subject = email_subject_template % exploration_title
    email_footer = get_rendered_email_footer()
    email_body = email_body_template % (
        recipient_username,
        inviter_username,
        role_description,
        exploration_id,
        exploration_title,
        rights_html,
        exploration_id,
        email_footer,
    )
    noreply_email_address = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
        )
    )
    assert isinstance(noreply_email_address, str)

    _send_email(
        recipient_id,
        feconf.SYSTEM_COMMITTER_ID,
        feconf.EMAIL_INTENT_EDITOR_ROLE_NOTIFICATION,
        email_subject,
        email_body,
        noreply_email_address,
        sender_name=inviter_username,
    )


def send_emails_to_subscribers(
    creator_id: str, exploration_id: str, exploration_title: str
) -> None:
    """Sends an email to all the subscribers of the creators when the creator
    publishes an exploration.

    Args:
        creator_id: str. The id of the creator who has published an exploration
            and to whose subscribers we are sending emails.
        exploration_id: str. The id of the exploration which the creator has
            published.
        exploration_title: str. The title of the exploration which the creator
            has published.
    """

    creator_name = user_services.get_username(creator_id)
    email_subject = '%s has published a new exploration!' % creator_name
    email_body_template = (
        'Hi %s,<br>'
        '<br>'
        '%s has published a new exploration! You can play it here: '
        '<a href="https://www.oppia.org/explore/%s">%s</a><br>'
        '<br>'
        'Thanks, and happy learning!<br>'
        '<br>'
        'Best wishes,<br>'
        '- The Oppia Team<br>'
        '<br>%s'
    )

    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return

    if not feconf.CAN_SEND_TRANSACTIONAL_EMAILS:
        logging.error('This app cannot send subscription emails to users.')
        return

    recipient_list = subscription_services.get_all_subscribers_of_creator(
        creator_id
    )
    recipients_usernames = user_services.get_usernames(
        recipient_list, strict=True
    )
    recipients_preferences = user_services.get_users_email_preferences(
        recipient_list
    )
    noreply_email_address = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
        )
    )
    assert isinstance(noreply_email_address, str)
    for index, username in enumerate(recipients_usernames):
        if recipients_preferences[index].can_receive_subscription_email:
            email_footer = get_rendered_email_footer()
            email_body = email_body_template % (
                username,
                creator_name,
                exploration_id,
                exploration_title,
                email_footer,
            )
            _send_email(
                recipient_list[index],
                feconf.SYSTEM_COMMITTER_ID,
                feconf.EMAIL_INTENT_SUBSCRIPTION_NOTIFICATION,
                email_subject,
                email_body,
                noreply_email_address,
            )


def send_feedback_message_email(
    recipient_id: str, feedback_messages: Dict[str, FeedbackMessagesDict]
) -> None:
    """Sends an email when creator receives feedback message to an exploration.

    Args:
        recipient_id: str. User ID of recipient.
        feedback_messages: dict. Contains feedback messages. Example:

            {
                'exploration_id': {
                    'title': 'Exploration 1234',
                    'messages': ['Feedback message 1', 'Feedback message 2']
                }
            }
    """
    email_subject_template = (
        'You\'ve received %s new message%s on your explorations'
    )

    email_body_template = (
        'Hi %s,<br>'
        '<br>'
        'You\'ve received %s new message%s on your Oppia explorations:<br>'
        '<ul>%s</ul>'
        'You can view and reply to your messages from your '
        '<a href="https://www.oppia.org/creator-dashboard">dashboard</a>.'
        '<br>'
        '<br>Thanks, and happy teaching!<br>'
        '<br>'
        'Best wishes,<br>'
        'The Oppia Team<br>'
        '<br>%s'
    )

    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return

    if not feconf.CAN_SEND_TRANSACTIONAL_EMAILS:
        logging.error('This app cannot send feedback message emails to users.')
        return

    if not feedback_messages:
        return

    recipient_username = user_services.get_username(recipient_id)

    messages_html = ''
    count_messages = 0
    for exp_id, reference in feedback_messages.items():
        messages_html += (
            '<li><a href="https://www.oppia.org/create/%s#/feedback">'
            '%s</a>:<br><ul>' % (exp_id, reference['title'])
        )
        for message in reference['messages']:
            messages_html += '<li>%s<br></li>' % message
            count_messages += 1
        messages_html += '</ul></li>'

    email_subject = email_subject_template % (
        (count_messages, 's') if count_messages > 1 else ('a', '')
    )

    email_footer = get_rendered_email_footer()

    email_body = email_body_template % (
        recipient_username,
        count_messages if count_messages > 1 else 'a',
        's' if count_messages > 1 else '',
        messages_html,
        email_footer,
    )

    noreply_email_address = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
        )
    )
    assert isinstance(noreply_email_address, str)

    _send_email(
        recipient_id,
        feconf.SYSTEM_COMMITTER_ID,
        feconf.EMAIL_INTENT_FEEDBACK_MESSAGE_NOTIFICATION,
        email_subject,
        email_body,
        noreply_email_address,
    )


def can_users_receive_thread_email(
    recipient_ids: List[str], exploration_id: str, has_suggestion: bool
) -> List[bool]:
    """Returns if users can receive email.

    Args:
        recipient_ids: list(str). IDs of persons that should receive the email.
        exploration_id: str. ID of exploration that received new message.
        has_suggestion: bool. True if thread contains suggestion.

    Returns:
        list(bool). True if user can receive the email, False otherwise.
    """
    users_global_prefs = user_services.get_users_email_preferences(
        recipient_ids
    )
    users_exploration_prefs = (
        user_services.get_users_email_preferences_for_exploration(
            recipient_ids, exploration_id
        )
    )
    zipped_preferences = list(zip(users_global_prefs, users_exploration_prefs))

    result = []
    if has_suggestion:
        for user_global_prefs, user_exploration_prefs in zipped_preferences:
            result.append(
                user_global_prefs.can_receive_feedback_message_email
                and not user_exploration_prefs.mute_suggestion_notifications
            )
    else:
        for user_global_prefs, user_exploration_prefs in zipped_preferences:
            result.append(
                user_global_prefs.can_receive_feedback_message_email
                and not user_exploration_prefs.mute_feedback_notifications
            )

    return result


def send_suggestion_email(
    exploration_title: str,
    exploration_id: str,
    author_id: str,
    recipient_list: List[str],
) -> None:
    """Send emails to notify the given recipients about new suggestion.

    Each recipient will only be emailed if their email preferences allow for
    incoming feedback message emails.

    Args:
        exploration_title: str. Title of the exploration with the new
            suggestion.
        exploration_id: str. The ID of the exploration with the new suggestion.
        author_id: str. The user ID of the author of the suggestion.
        recipient_list: list(str). The user IDs of the email recipients.
    """

    email_subject = 'New suggestion for "%s"' % exploration_title

    email_body_template = (
        'Hi %s,<br>'
        '%s has submitted a new suggestion for your Oppia exploration, '
        '<a href="https://www.oppia.org/create/%s">"%s"</a>.<br>'
        'You can accept or reject this suggestion by visiting the '
        '<a href="https://www.oppia.org/create/%s#/feedback">feedback page</a> '
        'for your exploration.<br>'
        '<br>'
        'Thanks!<br>'
        '- The Oppia Team<br>'
        '<br>%s'
    )

    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return

    if not feconf.CAN_SEND_TRANSACTIONAL_EMAILS:
        logging.error('This app cannot send feedback message emails to users.')
        return

    author_username = user_services.get_username(author_id)
    can_users_receive_email = can_users_receive_thread_email(
        recipient_list, exploration_id, True
    )
    email_footer = get_rendered_email_footer()
    noreply_email_address = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
        )
    )
    assert isinstance(noreply_email_address, str)
    for index, recipient_id in enumerate(recipient_list):
        recipient_username = user_services.get_username(recipient_id)
        # Send email only if recipient wants to receive.
        if can_users_receive_email[index]:
            email_body = email_body_template % (
                recipient_username,
                author_username,
                exploration_id,
                exploration_title,
                exploration_id,
                email_footer,
            )
            _send_email(
                recipient_id,
                feconf.SYSTEM_COMMITTER_ID,
                feconf.EMAIL_INTENT_SUGGESTION_NOTIFICATION,
                email_subject,
                email_body,
                noreply_email_address,
            )


def send_instant_feedback_message_email(
    recipient_id: str,
    sender_id: str,
    message: str,
    email_subject: str,
    exploration_title: str,
    exploration_id: str,
    thread_title: str,
) -> None:
    """Send an email when a new message is posted to a feedback thread, or when
    the thread's status is changed.

    Args:
        recipient_id: str. The user ID of the recipient.
        sender_id: str. The user ID of the sender.
        message: str. The message text or status change text from the sender.
        email_subject: str. The subject line to be sent in the email.
        exploration_title: str. The title of the exploration.
        exploration_id: str. ID of the exploration the feedback thread is about.
        thread_title: str. The title of the feedback thread.
    """

    email_body_template = (
        'Hi %s,<br><br>'
        'New update to thread "%s" on '
        '<a href="https://www.oppia.org/create/%s#/feedback">%s</a>:<br>'
        '<ul><li>%s: %s<br></li></ul>'
        '(You received this message because you are a '
        'participant in this thread.)<br><br>'
        'Best wishes,<br>'
        'The Oppia team<br>'
        '<br>%s'
    )

    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return

    if not feconf.CAN_SEND_TRANSACTIONAL_EMAILS:
        logging.error('This app cannot send feedback message emails to users.')
        return

    sender_username = user_services.get_username(sender_id)
    recipient_username = user_services.get_username(recipient_id)
    recipient_preferences = user_services.get_email_preferences(recipient_id)

    if recipient_preferences.can_receive_feedback_message_email:
        email_footer = get_rendered_email_footer()
        email_body = email_body_template % (
            recipient_username,
            thread_title,
            exploration_id,
            exploration_title,
            sender_username,
            message,
            email_footer,
        )
        noreply_email_address = (
            platform_parameter_services.get_platform_parameter_value(
                platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
            )
        )
        assert isinstance(noreply_email_address, str)
        _send_email(
            recipient_id,
            feconf.SYSTEM_COMMITTER_ID,
            feconf.EMAIL_INTENT_FEEDBACK_MESSAGE_NOTIFICATION,
            email_subject,
            email_body,
            noreply_email_address,
        )


def send_flag_exploration_email(
    exploration_title: str,
    exploration_id: str,
    reporter_id: str,
    report_text: str,
) -> None:
    """Send an email to all moderators when an exploration is flagged.

    Args:
        exploration_title: str. The title of the flagged exporation.
        exploration_id: str. The ID of the flagged exploration.
        reporter_id: str. The user ID of the reporter.
        report_text: str. The message entered by the reporter.
    """
    email_subject = 'Exploration flagged by user: "%s"' % exploration_title

    email_body_template = (
        'Hello Moderator,<br>'
        '%s has flagged exploration "%s" on the following '
        'grounds: <br>'
        '%s .<br>'
        'You can modify the exploration by clicking '
        '<a href="https://www.oppia.org/create/%s">here</a>.<br>'
        '<br>'
        'Thanks!<br>'
        '- The Oppia Team<br>'
        '<br>%s'
    )

    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return

    reporter_username = user_services.get_username(reporter_id)

    email_footer = get_rendered_email_footer()

    email_body = email_body_template % (
        reporter_username,
        exploration_title,
        report_text,
        exploration_id,
        email_footer,
    )

    noreply_email_address = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
        )
    )
    assert isinstance(noreply_email_address, str)

    recipient_list = user_services.get_user_ids_by_role(
        feconf.ROLE_ID_MODERATOR
    )
    for recipient_id in recipient_list:
        _send_email(
            recipient_id,
            feconf.SYSTEM_COMMITTER_ID,
            feconf.EMAIL_INTENT_REPORT_BAD_CONTENT,
            email_subject,
            email_body,
            noreply_email_address,
        )


def send_mail_to_onboard_new_reviewers(
    recipient_id: str, category: str
) -> None:
    """Sends an email to users who have crossed the threshold score.

    Args:
        recipient_id: str. The ID of the user who is being offered to become a
            reviewer.
        category: str. The category that the user is being offered to review.
    """

    email_subject = 'Invitation to review suggestions'

    email_body_template = (
        'Hi %s,<br><br>'
        'Thank you for actively contributing high-quality suggestions for '
        'Oppia\'s lessons in %s, and for helping to make these lessons better '
        'for students around the world!<br><br>'
        'In recognition of your contributions, we would like to invite you to '
        'become one of Oppia\'s reviewers. As a reviewer, you will be able to '
        'review suggestions in %s, and contribute to helping ensure that any '
        'edits made to lessons preserve the lessons\' quality and are '
        'beneficial for students.<br><br>'
        'If you\'d like to help out as a reviewer, please visit your '
        '<a href="https://www.oppia.org/creator-dashboard/">dashboard</a>. '
        'and set your review preferences accordingly. Note that, if you accept,'
        'you will receive occasional emails inviting you to review incoming '
        'suggestions by others.<br><br>'
        'Again, thank you for your contributions to the Oppia community!<br>'
        '- The Oppia Team<br>'
        '<br>%s'
    )

    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return

    recipient_username = user_services.get_username(recipient_id)
    can_user_receive_email = user_services.get_email_preferences(
        recipient_id
    ).can_receive_email_updates

    # Send email only if recipient wants to receive.
    if can_user_receive_email:
        email_footer = get_rendered_email_footer()
        email_body = email_body_template % (
            recipient_username,
            category,
            category,
            email_footer,
        )
        noreply_email_address = (
            platform_parameter_services.get_platform_parameter_value(
                platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
            )
        )
        assert isinstance(noreply_email_address, str)
        _send_email(
            recipient_id,
            feconf.SYSTEM_COMMITTER_ID,
            feconf.EMAIL_INTENT_ONBOARD_CD_USER,
            email_subject,
            email_body,
            noreply_email_address,
        )


def send_mail_to_notify_users_to_review(
    recipient_id: str, category: str
) -> None:
    """Sends an email to users to review suggestions in categories they have
    agreed to review for.

    Args:
        recipient_id: str. The id of the user who is being pinged to review
            suggestions.
        category: str. The category of the suggestions to review.
    """

    email_subject = 'Notification to review suggestions'

    email_body_template = (
        'Hi %s,<br><br>'
        'Just a heads-up that there are new suggestions to '
        'review in %s, which you are registered as a reviewer for.'
        '<br><br>Please take a look at and accept/reject these suggestions at'
        ' your earliest convenience. You can visit your '
        '<a href="https://www.oppia.org/creator-dashboard/">dashboard</a> '
        'to view the list of suggestions that need a review.<br><br>'
        'Thank you for helping improve Oppia\'s lessons!'
        '- The Oppia Team<br>'
        '<br>%s'
    )

    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return

    recipient_username = user_services.get_username(recipient_id)
    can_user_receive_email = user_services.get_email_preferences(
        recipient_id
    ).can_receive_email_updates

    # Send email only if recipient wants to receive.
    if can_user_receive_email:
        email_footer = get_rendered_email_footer()
        email_body = email_body_template % (
            recipient_username,
            category,
            email_footer,
        )
        noreply_email_address = (
            platform_parameter_services.get_platform_parameter_value(
                platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
            )
        )
        assert isinstance(noreply_email_address, str)
        _send_email(
            recipient_id,
            feconf.SYSTEM_COMMITTER_ID,
            feconf.EMAIL_INTENT_REVIEW_CREATOR_DASHBOARD_SUGGESTIONS,
            email_subject,
            email_body,
            noreply_email_address,
        )


def _create_html_for_reviewable_suggestion_email_info(
    reviewable_suggestion_email_info: (
        suggestion_registry.ReviewableSuggestionEmailInfo
    ),
) -> str:
    """Creates the html for the given reviewable_suggestion_email_info. This
    html content is used to provide information about a suggestion in an
    email.

    Args:
        reviewable_suggestion_email_info: ReviewableSuggestionEmailInfo. The
            information about the suggestion that will be used to form the
            html for the email. This includes the suggestion type, language,
            content and review submission date.

    Returns:
        str. A string containing the html that represents the suggestion
        information.
    """
    # Get the language of the suggestion.
    language = utils.get_supported_audio_language_description(
        reviewable_suggestion_email_info.language_code
    )
    # Calculate how long the suggestion has been waiting for review.
    suggestion_review_wait_time = utils.get_current_utc_datetime() - (
        reviewable_suggestion_email_info.submission_datetime
    )
    # Get a string composed of the largest time unit that has a
    # value, followed by that time unit. For example, if the
    # suggestion had been waiting for review for 5 days and 2 hours,
    # '5 days' would be returned. This is more user friendly since a
    # high level of precision is not needed.
    human_readable_review_wait_time = (
        utils.create_string_from_largest_unit_in_timedelta(
            suggestion_review_wait_time
        )
    )
    values_to_populate_suggestion_template_dict = {
        'language': language,
        'review_wait_time': human_readable_review_wait_time,
        'suggestion_content': (
            reviewable_suggestion_email_info.suggestion_content
        ),
    }
    get_values_to_populate_suggestion_template = (
        HTML_FOR_SUGGESTION_DESCRIPTION[
            'suggestion_template_values_getter_functions'
        ][reviewable_suggestion_email_info.suggestion_type]
    )
    # Ruling out the possibility of str for mypy type checking. Because from
    # implementation it is clear that, this is used as a function not string.
    assert callable(get_values_to_populate_suggestion_template)
    suggestion_template = HTML_FOR_SUGGESTION_DESCRIPTION[
        'suggestion_template'
    ][reviewable_suggestion_email_info.suggestion_type]
    # Ruling out the possibility of callable for mypy type checking.
    assert isinstance(suggestion_template, str)
    return suggestion_template % (
        get_values_to_populate_suggestion_template(
            values_to_populate_suggestion_template_dict
        )
    )


def send_mail_to_notify_admins_suggestions_waiting_long(
    admin_ids: List[str],
    translation_admin_ids: List[str],
    question_admin_ids: List[str],
    reviewable_suggestion_email_infos: List[
        suggestion_registry.ReviewableSuggestionEmailInfo
    ],
) -> None:
    """Sends an email to admins to inform them about the suggestions that have
    been waiting longer than
    suggestion_models.SUGGESTION_REVIEW_WAIT_TIME_THRESHOLD_IN_DAYS days for a
    review on the Contributor Dashboard. Admins can be informed about at most
    suggestion_models.MAX_NUMBER_OF_SUGGESTIONS_TO_EMAIL_ADMIN suggestions.
    The information about the suggestions is organized in descending order
    by the suggestion's review wait time.

    Args:
        admin_ids: list(str). The user ids of the admins to notify.
        translation_admin_ids: list(str). The user ids of the translation
            admins to notify.
        question_admin_ids: list(str). The user ids of the question admins
            to notify.
        reviewable_suggestion_email_infos: list(ReviewableSuggestionEmailInfo).
            list(ReviewableSuggestionEmailContentInfo). A list of suggestion
            email content info objects that represent suggestions
            that have been waiting too long for review to notify the admins
            about. Each object contains includes the suggestion type, language,
            content and review submission date. The objects are sorted in
            descending order based on review wait time.
    """
    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return

    if not platform_parameter_services.get_platform_parameter_value(
        platform_parameter_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_SUGGESTIONS_NEEDING_REVIEW.value
    ):
        logging.error(
            'The "notify_admins_suggestions_waiting_too_long" property '
            'must be enabled on the admin config page in order to send '
            'admins the emails.'
        )
        return

    if not reviewable_suggestion_email_infos:
        logging.info(
            'There were no Contributor Dashboard suggestions that were waiting '
            'too long for a review.'
        )
        return

    if not admin_ids:
        logging.error('There were no admins to notify.')
        return

    translation_suggestion_descriptions = []
    question_suggestion_descriptions = []
    # Get the html for the list of suggestions that have been waiting too long
    # for a review.
    for reviewable_suggestion_email_info in reviewable_suggestion_email_infos:
        if (
            feconf.SUGGESTION_TYPE_TRANSLATE_CONTENT
            == reviewable_suggestion_email_info.suggestion_type
        ):
            translation_suggestion_descriptions.append(
                _create_html_for_reviewable_suggestion_email_info(
                    reviewable_suggestion_email_info
                )
            )
        if (
            feconf.SUGGESTION_TYPE_ADD_QUESTION
            == reviewable_suggestion_email_info.suggestion_type
        ):
            question_suggestion_descriptions.append(
                _create_html_for_reviewable_suggestion_email_info(
                    reviewable_suggestion_email_info
                )
            )

    list_of_translation_suggestion_descriptions = ''.join(
        translation_suggestion_descriptions
    )
    if list_of_translation_suggestion_descriptions:
        user_ids = []
        user_ids.extend(admin_ids)
        user_ids.extend(translation_admin_ids)
        _send_suggestions_waiting_too_long_email(
            user_ids, list_of_translation_suggestion_descriptions
        )
    list_of_question_suggestion_descriptions = ''.join(
        question_suggestion_descriptions
    )
    if list_of_question_suggestion_descriptions:
        user_ids = []
        user_ids.extend(admin_ids)
        user_ids.extend(question_admin_ids)
        _send_suggestions_waiting_too_long_email(
            user_ids, list_of_question_suggestion_descriptions
        )


def _send_suggestions_waiting_too_long_email(
    admin_ids: List[str], list_of_suggestion_descriptions: str
) -> None:
    """Helper method for send_mail_to_notify_admins_suggestions_waiting_long
    that allows sending of emails to the list of admin ids provided.

    Args:
        admin_ids: list(str). The user ids of the admins to notify.
        list_of_suggestion_descriptions: str. Suggestion descriptions HTML to
            send in the email.
    """
    email_subject = (
        ADMIN_NOTIFICATION_FOR_SUGGESTIONS_NEEDING_REVIEW_EMAIL_DATA[
            'email_subject'
        ]
    )
    email_body_template = (
        ADMIN_NOTIFICATION_FOR_SUGGESTIONS_NEEDING_REVIEW_EMAIL_DATA[
            'email_body_template'
        ]
    )
    # Get the emails and usernames of the admins.
    admin_user_settings = user_services.get_users_settings(admin_ids)
    curriculum_admin_usernames, admin_emails = list(
        zip(
            *[
                (
                    (admin_user_setting.username, admin_user_setting.email)
                    if admin_user_setting is not None
                    else (None, None)
                )
                for admin_user_setting in admin_user_settings
            ]
        )
    )
    noreply_email_address = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
        )
    )
    assert isinstance(noreply_email_address, str)

    oppia_site_url = platform_parameter_services.get_platform_parameter_value(
        platform_parameter_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS.value
    )
    assert isinstance(oppia_site_url, str)

    for index, admin_id in enumerate(admin_ids):
        if not admin_emails[index]:
            logging.error(
                'There was no email for the given admin id: %s.' % admin_id
            )
            continue
        email_body = email_body_template % (
            curriculum_admin_usernames[index],
            oppia_site_url,
            feconf.CONTRIBUTOR_DASHBOARD_URL,
            suggestion_models.SUGGESTION_REVIEW_WAIT_TIME_THRESHOLD_IN_DAYS,
            oppia_site_url,
            feconf.CONTRIBUTOR_DASHBOARD_ADMIN_URL,
            list_of_suggestion_descriptions,
        )

        _send_email(
            admin_id,
            feconf.SYSTEM_COMMITTER_ID,
            feconf.EMAIL_INTENT_ADDRESS_CONTRIBUTOR_DASHBOARD_SUGGESTIONS,
            email_subject,
            email_body,
            noreply_email_address,
            recipient_email=admin_emails[index],
        )


def send_reviewer_notifications(
    reviewer_ids_by_language: DefaultDict[str, List[str]],
    suggestions_by_language: DefaultDict[
        str, List[suggestion_registry.ReviewableSuggestionEmailInfo]
    ],
) -> None:
    """Sends email notifications to reviewers about new suggestions.

    Args:
        suggestions_by_language: dict. A dictionary that organizes
            new suggestions by language code.
        reviewer_ids_by_language: dict. A dictionary that organizes reviewer
            IDs by language code.
    """
    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return

    oppia_site_url = platform_parameter_services.get_platform_parameter_value(
        platform_parameter_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS.value
    )
    assert isinstance(oppia_site_url, str)

    for language_code, suggestions in suggestions_by_language.items():
        reviewer_ids = reviewer_ids_by_language[language_code]

        if not reviewer_ids:
            message = 'No reviewers found for language %s to notify'
            logging.error(message % language_code)
            continue

        email_subject = 'Contributor Dashboard New Reviewer Opportunities'
        email_body_template = (
            'Hi %s,<br><br>'
            'There are new <a href="%s%s">opportunities</a>'
            ' to review translations that we think you might be interested'
            ' in on the Contributor Dashboard page. Here are some examples'
            ' of contributions that are waiting for review:<br><br>'
            'The following suggestions are available for review: '
            '<br><br><ul>%s</ul><br>Please take some time to review any '
            'of the above contributions (if they still need a review)'
            ' or any other contributions on the dashboard.'
            ' We appreciate your help!<br><br>Thanks again, '
            'and happy reviewing!<br><br>The Oppia Contributor Dashboard Team'
        )
        suggestion_descriptions = []
        for suggestion in suggestions:
            suggestion_descriptions.append(
                _create_html_for_reviewable_suggestion_email_info(suggestion)
            )

        noreply_email_address = (
            platform_parameter_services.get_platform_parameter_value(
                platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
            )
        )
        assert isinstance(noreply_email_address, str)

        for reviewer_id in reviewer_ids:
            reviewer_username = user_services.get_username(reviewer_id)
            email_body = email_body_template % (
                reviewer_username,
                oppia_site_url,
                feconf.CONTRIBUTOR_DASHBOARD_URL,
                ''.join(suggestion_descriptions),
            )

            # Send the email to each reviewer.
            reviewer_email = user_services.get_email_from_user_id(reviewer_id)

            _send_email(
                reviewer_id,
                feconf.SYSTEM_COMMITTER_ID,
                feconf.EMAIL_INTENT_ADDRESS_CONTRIBUTOR_DASHBOARD_SUGGESTIONS,
                email_subject,
                email_body,
                noreply_email_address,
                recipient_email=reviewer_email,
            )


def send_mail_to_notify_admins_that_reviewers_are_needed(
    admin_ids: List[str],
    translation_admin_ids: List[str],
    question_admin_ids: List[str],
    suggestion_types_needing_reviewers: Dict[str, Set[str]],
) -> None:
    """Sends an email to admins to notify them that there are specific
    suggestion types on the Contributor Dashboard that need more reviewers.

    Note: it is assumed that all admins are super admins because only super
    admins have access to the admin page where reviewers can be added to the
    Contributor Dashboard. Also note that these emails are sent out regardless
    of the admins' email preferences.

    Args:
        admin_ids: list(str). The user ids of the admins to notify.
        translation_admin_ids: list(str). The user ids of the translation
            admins to notify.
        question_admin_ids: list(str). The user ids of the question admins
            to notify.
        suggestion_types_needing_reviewers: dict. A dictionary where the keys
            are suggestion types and each value corresponds to a set that
            contains the language codes within the suggestion type that need
            more reviewers. For example, for translation suggestions, the value
            would be a set of language codes that translations are offered in
            that need more reviewers.
    """
    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return

    if not platform_parameter_services.get_platform_parameter_value(
        platform_parameter_list.ParamName.ENABLE_ADMIN_NOTIFICATIONS_FOR_REVIEWER_SHORTAGE.value
    ):
        logging.error(
            'The "enable_admin_notifications_for_reviewer_shortage" '
            'property must be enabled on the admin config page in order to '
            'send admins the emails.'
        )
        return

    if not suggestion_types_needing_reviewers:
        logging.info(
            'There were no suggestion types that needed more reviewers on the '
            'Contributor Dashboard.'
        )
        return

    if not admin_ids:
        logging.error('There were no admins to notify.')
        return

    oppia_site_url = platform_parameter_services.get_platform_parameter_value(
        platform_parameter_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS.value
    )
    assert isinstance(oppia_site_url, str)

    if feconf.SUGGESTION_TYPE_TRANSLATE_CONTENT in (
        suggestion_types_needing_reviewers
    ):
        translation_suggestions_needing_reviewers_paragraphs = []
        language_codes_that_need_reviewers = suggestion_types_needing_reviewers[
            feconf.SUGGESTION_TYPE_TRANSLATE_CONTENT
        ]
        # There are different templates to handle whether multiple languages
        # need more reviewers or just one language.
        if len(language_codes_that_need_reviewers) == 1:
            translation_suggestions_needing_reviewers_paragraphs.append(
                ADMIN_NOTIFICATION_FOR_REVIEWER_SHORTAGE_EMAIL_DATA[
                    'one_language_template'
                ]
                % (
                    utils.get_supported_audio_language_description(
                        language_codes_that_need_reviewers.pop()
                    ),
                    oppia_site_url,
                    feconf.CONTRIBUTOR_DASHBOARD_URL,
                )
            )

        else:
            html_for_languages_that_need_more_reviewers = ''.join(
                [
                    '<li><b>%s</b></li><br>'
                    % (
                        utils.get_supported_audio_language_description(
                            language_code
                        )
                    )
                    for language_code in sorted(
                        language_codes_that_need_reviewers
                    )
                ]
            )
            translation_suggestions_needing_reviewers_paragraphs.append(
                ADMIN_NOTIFICATION_FOR_REVIEWER_SHORTAGE_EMAIL_DATA[
                    'multi_language_template'
                ]
                % (
                    oppia_site_url,
                    feconf.CONTRIBUTOR_DASHBOARD_URL,
                    html_for_languages_that_need_more_reviewers,
                )
            )
        translation_suggestions_needing_reviewers_html = ''.join(
            translation_suggestions_needing_reviewers_paragraphs
        )
        user_ids = []
        user_ids.extend(admin_ids)
        user_ids.extend(translation_admin_ids)
        _send_reviews_needed_email_to_admins(
            user_ids, translation_suggestions_needing_reviewers_html
        )

    if feconf.SUGGESTION_TYPE_ADD_QUESTION in (
        suggestion_types_needing_reviewers
    ):
        question_suggestions_needing_reviewers_paragraphs = []
        question_suggestions_needing_reviewers_paragraphs.append(
            ADMIN_NOTIFICATION_FOR_REVIEWER_SHORTAGE_EMAIL_DATA[
                'question_template'
            ]
            % (oppia_site_url, feconf.CONTRIBUTOR_DASHBOARD_URL)
        )
        question_suggestions_needing_reviewers_html = ''.join(
            question_suggestions_needing_reviewers_paragraphs
        )
        user_ids = []
        user_ids.extend(admin_ids)
        user_ids.extend(question_admin_ids)
        _send_reviews_needed_email_to_admins(
            user_ids, question_suggestions_needing_reviewers_html
        )


def _send_reviews_needed_email_to_admins(
    admin_ids: List[str], suggestions_needing_reviewers_html: str
) -> None:
    """Helper function for send_mail_to_notify_admins_that_reviewers_are_needed
    that allows sending email to the provided admin ids.

    Args:
        admin_ids: list(str). The user ids of the admins to notify.
        suggestions_needing_reviewers_html: str. The HTML representing
            the suggestion needing reviewers.
    """
    email_subject = ADMIN_NOTIFICATION_FOR_REVIEWER_SHORTAGE_EMAIL_DATA[
        'email_subject'
    ]
    email_body_template = ADMIN_NOTIFICATION_FOR_REVIEWER_SHORTAGE_EMAIL_DATA[
        'email_body_template'
    ]
    # Get the emails and usernames of the users.
    admin_user_settings = user_services.get_users_settings(admin_ids)
    curriculum_admin_usernames, admin_emails = list(
        zip(
            *[
                (
                    (admin_user_setting.username, admin_user_setting.email)
                    if admin_user_setting is not None
                    else (None, None)
                )
                for admin_user_setting in admin_user_settings
            ]
        )
    )
    noreply_email_address = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
        )
    )
    assert isinstance(noreply_email_address, str)

    oppia_site_url = platform_parameter_services.get_platform_parameter_value(
        platform_parameter_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS.value
    )
    assert isinstance(oppia_site_url, str)
    for index, admin_id in enumerate(admin_ids):
        if not admin_emails[index]:
            logging.error(
                'There was no email for the given admin id: %s.' % admin_id
            )
            continue
        email_body = email_body_template % (
            curriculum_admin_usernames[index],
            oppia_site_url,
            feconf.ADMIN_URL,
            suggestions_needing_reviewers_html,
        )

        _send_email(
            admin_id,
            feconf.SYSTEM_COMMITTER_ID,
            feconf.EMAIL_INTENT_ADD_CONTRIBUTOR_DASHBOARD_REVIEWERS,
            email_subject,
            email_body,
            noreply_email_address,
            recipient_email=admin_emails[index],
        )


def send_mail_to_notify_contributor_dashboard_reviewers(
    reviewer_ids: List[str],
    reviewers_suggestion_email_infos: List[
        List[suggestion_registry.ReviewableSuggestionEmailInfo]
    ],
) -> None:
    """Sends an email to each reviewer notifying them of the suggestions on the
    Contributor Dashboard that have been waiting the longest for review, and
    that the reviewer has permission to review.

    Args:
        reviewer_ids: list(str). A list of the Contributor Dashboard reviewer
            user ids to notify.
        reviewers_suggestion_email_infos:
            list(list(ReviewableSuggestionEmailInfo)). A list of suggestion
            email content info objects for each reviewer. These suggestion
            email content info objects contain the key information about the
            suggestions we're notifying reviewers about and will be used to
            compose the email body for each reviewer.
    """
    email_subject = CONTRIBUTOR_DASHBOARD_REVIEWER_NOTIFICATION_EMAIL_DATA[
        'email_subject'
    ]
    email_body_template = (
        CONTRIBUTOR_DASHBOARD_REVIEWER_NOTIFICATION_EMAIL_DATA[
            'email_body_template'
        ]
    )

    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return

    if not platform_parameter_services.get_platform_parameter_value(
        platform_parameter_list.ParamName.CONTRIBUTOR_DASHBOARD_REVIEWER_EMAILS_IS_ENABLED.value
    ):
        logging.error(
            'The "contributor_dashboard_reviewer_emails_is_enabled" property '
            'must be enabled on the platform parameters tab on the admin page '
            'in order to send reviewers the emails.'
        )
        return

    if not reviewer_ids:
        logging.error('No Contributor Dashboard reviewers to notify.')
        return

    reviewer_user_settings = user_services.get_users_settings(reviewer_ids)
    reviewer_usernames, reviewer_emails = list(
        zip(
            *[
                (
                    (
                        reviewer_user_setting.username,
                        reviewer_user_setting.email,
                    )
                    if reviewer_user_setting is not None
                    else (None, None)
                )
                for reviewer_user_setting in reviewer_user_settings
            ]
        )
    )

    email_footer = get_rendered_email_footer()
    noreply_email_address = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
        )
    )
    assert isinstance(noreply_email_address, str)

    oppia_site_url = platform_parameter_services.get_platform_parameter_value(
        platform_parameter_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS.value
    )
    assert isinstance(oppia_site_url, str)

    for index, reviewer_id in enumerate(reviewer_ids):
        if not reviewers_suggestion_email_infos[index]:
            logging.info(
                'There were no suggestions to recommend to the reviewer with '
                'user id: %s.' % reviewer_id
            )
            continue

        if not reviewer_emails[index]:
            logging.error(
                'There was no email for the given reviewer id: %s.'
                % (reviewer_id)
            )
            continue

        suggestion_descriptions = []
        for reviewer_suggestion_email_info in reviewers_suggestion_email_infos[
            index
        ]:
            suggestion_descriptions.append(
                _create_html_for_reviewable_suggestion_email_info(
                    reviewer_suggestion_email_info
                )
            )

        email_body = email_body_template % (
            reviewer_usernames[index],
            oppia_site_url,
            feconf.CONTRIBUTOR_DASHBOARD_URL,
            ''.join(suggestion_descriptions),
            email_footer,
        )

        _send_email(
            reviewer_id,
            feconf.SYSTEM_COMMITTER_ID,
            feconf.EMAIL_INTENT_REVIEW_CONTRIBUTOR_DASHBOARD_SUGGESTIONS,
            email_subject,
            email_body,
            noreply_email_address,
            recipient_email=reviewer_emails[index],
        )


def send_mail_to_notify_contributor_ranking_achievement(
    contributor_ranking_email_info: (
        suggestion_registry.ContributorMilestoneEmailInfo
    ),
) -> None:
    """Sends an email to translation/question submitters and reviewers when
    they achieve a new rank.

    Args:
        contributor_ranking_email_info:
            ContributorMilestoneEmailInfo. An object with contributor ranking
            email information.
    """
    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return

    recipient_username = user_services.get_username(
        contributor_ranking_email_info.contributor_user_id
    )
    can_user_receive_email = user_services.get_email_preferences(
        contributor_ranking_email_info.contributor_user_id
    ).can_receive_email_updates

    oppia_site_url = platform_parameter_services.get_platform_parameter_value(
        platform_parameter_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS.value
    )
    assert isinstance(oppia_site_url, str)

    if can_user_receive_email:
        email_template = CONTRIBUTOR_RANK_ACHIEVEMENT_NOTIFICATION[
            contributor_ranking_email_info.contribution_type
        ][contributor_ranking_email_info.contribution_subtype]
        email_body = ''
        if contributor_ranking_email_info.contribution_type == (
            feconf.CONTRIBUTION_TYPE_TRANSLATION
        ):
            # Ruling out the possibility of None for mypy type checking. It is
            # obvious that for the contribution_type
            # CONTRIBUTION_TYPE_TRANSLATION the language_code will not be None.
            assert contributor_ranking_email_info.language_code is not None
            language = utils.get_supported_audio_language_description(
                contributor_ranking_email_info.language_code
            )
            email_body = email_template['email_body_template'] % (
                recipient_username,
                contributor_ranking_email_info.rank_name,
                language,
                oppia_site_url,
                feconf.CONTRIBUTOR_DASHBOARD_URL,
            )
        else:
            email_body = email_template['email_body_template'] % (
                recipient_username,
                contributor_ranking_email_info.rank_name,
                oppia_site_url,
                feconf.CONTRIBUTOR_DASHBOARD_URL,
            )

        noreply_email_address = (
            platform_parameter_services.get_platform_parameter_value(
                platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
            )
        )
        assert isinstance(noreply_email_address, str)

        _send_email(
            contributor_ranking_email_info.contributor_user_id,
            feconf.SYSTEM_COMMITTER_ID,
            feconf.EMAIL_INTENT_NOTIFY_CONTRIBUTOR_DASHBOARD_ACHIEVEMENTS,
            email_template['email_subject'],
            email_body,
            noreply_email_address,
        )


def send_reminder_mail_to_notify_curriculum_admins(
    curriculum_admin_ids: List[str],
    chapter_notifications_list: List[story_domain.StoryPublicationTimeliness],
) -> None:
    """Sends an email to curriculum admins to notify them about the
    behind-schedule and upcoming chapters in all the topics.

    Args:
        curriculum_admin_ids: list(str). The user ids of the admins to notify.
        chapter_notifications_list: list(StoryPublicationTimeliness). The list
            of stories having behind-schedule or upcoming chapters to be
            notified.
    """
    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return
    if len(curriculum_admin_ids) == 0:
        logging.error('There were no curriculum admins to notify.')
        return

    email_body_template = CURRICULUM_ADMIN_CHAPTER_NOTIFICATION_EMAIL_DATA
    email_subject = CURRICULUM_ADMIN_CHAPTER_NOTIFICATION_EMAIL_DATA[
        'email_subject'
    ]

    email_body = 'Dear Curriculum Admin, <br><br>'

    chapters_are_overdue = False
    chapters_are_upcoming = False

    overdue_stories_html = ''

    oppia_site_url = platform_parameter_services.get_platform_parameter_value(
        platform_parameter_list.ParamName.OPPIA_SITE_URL_FOR_EMAILS.value
    )
    assert isinstance(oppia_site_url, str)

    for overdue_story in chapter_notifications_list:
        if len(overdue_story.overdue_chapters) == 0:
            continue
        chapters_are_overdue = True
        story_link = '%s%s/%s' % (
            oppia_site_url,
            str(feconf.STORY_EDITOR_URL_PREFIX),
            overdue_story.id,
        )
        story_html = '<li>%s (%s) - <a href="%s">Link</a><ul>' % (
            overdue_story.story_name,
            overdue_story.topic_name,
            story_link,
        )
        for chapter in overdue_story.overdue_chapters:
            chapter_html = '<li>%s</li>' % chapter
            story_html += chapter_html
        story_html += '</ul></li>'
        overdue_stories_html += story_html
    if chapters_are_overdue:
        email_body += email_body_template['overdue_chapters_template'] % (
            overdue_stories_html
        )

    upcoming_stories_html = ''
    for upcoming_story in chapter_notifications_list:
        if len(upcoming_story.upcoming_chapters) == 0:
            continue
        chapters_are_upcoming = True
        story_link = '%s%s/%s' % (
            oppia_site_url,
            str(feconf.STORY_EDITOR_URL_PREFIX),
            upcoming_story.id,
        )
        story_html = '<li>%s (%s) - <a href="%s">Link</a><ul>' % (
            upcoming_story.story_name,
            upcoming_story.topic_name,
            story_link,
        )
        for chapter in upcoming_story.upcoming_chapters:
            chapter_html = '<li>%s</li>' % chapter
            story_html += chapter_html
        story_html += '</ul></li>'
        upcoming_stories_html += story_html
    if chapters_are_upcoming:
        email_body += email_body_template['upcoming_chapters_template'] % (
            upcoming_stories_html
        )

    email_body += 'Regards,<br> Oppia Foundation'

    noreply_email_address = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
        )
    )
    assert isinstance(noreply_email_address, str)
    system_email_name = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SYSTEM_EMAIL_NAME.value
        )
    )
    assert isinstance(system_email_name, str)

    if chapters_are_overdue or chapters_are_upcoming:
        for curriculum_admin_id in curriculum_admin_ids:
            _send_email(
                curriculum_admin_id,
                feconf.SYSTEM_COMMITTER_ID,
                feconf.EMAIL_INTENT_NOTIFY_CURRICULUM_ADMINS_CHAPTERS,
                email_subject,
                email_body,
                noreply_email_address,
                sender_name=system_email_name,
            )


def send_account_deleted_email(user_id: str, user_email: str) -> None:
    """Sends an email to user whose account was deleted.

    Args:
        user_id: str. The id of the user whose account got deleted.
        user_email: str. The email of the user whose account got deleted.
    """
    email_subject = 'Account deleted'

    email_body_template = (
        'Hi %s,<br><br>'
        'Your account was successfully deleted.<br><br>'
        '- The Oppia Team'
    )

    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return

    email_body = email_body_template % user_email
    noreply_email_address = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
        )
    )
    assert isinstance(noreply_email_address, str)
    _send_email(
        user_id,
        feconf.SYSTEM_COMMITTER_ID,
        feconf.EMAIL_INTENT_ACCOUNT_DELETED,
        email_subject,
        email_body,
        noreply_email_address,
        bcc_admin=True,
        recipient_email=user_email,
    )


def send_account_deletion_failed_email(user_id: str, user_email: str) -> None:
    """Sends an email to admin about the failure of the job that is supposed to
    delete the user.

    Args:
        user_id: str. The id of the user whose account failed to get deleted.
        user_email: str. The email of the user whose account failed to
            get deleted.
    """
    email_subject = 'WIPEOUT: Account deletion failed'
    email_body_template = (
        'The Wipeout process failed for the user '
        'with ID \'%s\' and email \'%s\'.' % (user_id, user_email)
    )
    send_mail_to_admin(email_subject, email_body_template)


def send_email_to_new_cd_user(
    recipient_id: str, category: str, language_code: Optional[str] = None
) -> None:
    """Sends an email to user who is assigned rights to either
       review or submit contributions.

    Args:
        recipient_id: str. The ID of the user.
        category: str. The category in which user can review or submit
            contributions.
        language_code: None|str. The language code for a language if
            the category is 'translation' else None.

    Raises:
        Exception. The contribution category is not valid.
        Exception. The language_code cannot be None if the
            category is 'translation'.
    """
    if category not in NEW_CD_USER_EMAIL_DATA:
        raise Exception('Invalid category: %s' % category)

    category_data = NEW_CD_USER_EMAIL_DATA[category]
    email_subject = 'You have been invited to %s Oppia %s' % (
        category_data['task'],
        category_data['category'],
    )

    if category in [
        constants.CD_USER_RIGHTS_CATEGORY_REVIEW_TRANSLATION,
    ]:
        if language_code is None:
            raise Exception(
                'The language_code cannot be None if the review category is'
                ' \'translation\''
            )
        language_description = utils.get_supported_audio_language_description(
            language_code
        ).capitalize()
        category_description = (
            category_data['description_template'] % language_description
        )
        rights_message = category_data['rights_message_template'] % (
            language_description
        )
    else:
        category_description = category_data['description']
        rights_message = category_data['rights_message']

    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return

    email_body_template = '%s %s %s %s'
    recipient_username = user_services.get_username(recipient_id)
    if category in [
        constants.CD_USER_RIGHTS_CATEGORY_REVIEW_TRANSLATION,
        constants.CD_USER_RIGHTS_CATEGORY_REVIEW_QUESTION,
    ]:
        to_review = category_data['to_review']
        email_body_template = (
            'Hi %s,<br><br>'
            'This is to let you know that the Oppia team has added you as a '
            'reviewer for %s. This allows you to %s.<br><br>'
            'You can check the %s waiting for review in the '
            '<a href="https://www.oppia.org/contributor-dashboard">'
            'Contributor Dashboard</a>.<br><br>'
            'Thanks, and happy contributing!<br><br>'
            'Best wishes,<br>'
            'The Oppia Community'
        )
        email_body = email_body_template % (
            recipient_username,
            category_description,
            rights_message,
            to_review,
        )

    elif category in [constants.CD_USER_RIGHTS_CATEGORY_SUBMIT_QUESTION]:
        email_body_template = (
            'Hi %s,<br><br>'
            'This is to let you know that the Oppia team has added you as a '
            'contributor to %s for use in lessons.<br><br>'
            'You can now start to submit %s in the '
            '<a href="https://www.oppia.org/contributor-dashboard">'
            'Contributor Dashboard</a>.<br><br>'
            'Thanks, and happy contributing!<br><br>'
            'Best wishes,<br>'
            'The Oppia Community'
        )
        email_body = email_body_template % (
            recipient_username,
            rights_message,
            category_description,
        )

    can_user_receive_email = user_services.get_email_preferences(
        recipient_id
    ).can_receive_email_updates

    # Send email only if recipient wants to receive.
    if can_user_receive_email:
        noreply_email_address = (
            platform_parameter_services.get_platform_parameter_value(
                platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
            )
        )
        assert isinstance(noreply_email_address, str)
        _send_email(
            recipient_id,
            feconf.SYSTEM_COMMITTER_ID,
            feconf.EMAIL_INTENT_ONBOARD_CD_USER,
            email_subject,
            email_body,
            noreply_email_address,
        )


def send_email_to_removed_cd_user(
    user_id: str, category: str, language_code: Optional[str] = None
) -> None:
    """Sends an email to user who is removed from a specific
        contributor position.

    Args:
        user_id: str. The ID of the user.
        category: str. The category in which user can no longer review or
            submit contributions.
        language_code: None|str. The language code for a language if the review
            item is translation else None.

    Raises:
        Exception. The contribution category is not valid.
        Exception. The language_code cannot be None if the review category is
            'translation'.
    """
    if category not in REMOVED_CD_USER_EMAIL_DATA:
        raise Exception('Invalid category: %s' % category)

    category_data = REMOVED_CD_USER_EMAIL_DATA[category]
    if category == constants.CD_USER_RIGHTS_CATEGORY_SUBMIT_QUESTION:
        email_subject = 'You have been unassigned as a %s submitter' % (
            category_data['category']
        )
    else:
        email_subject = 'You have been unassigned as a %s reviewer' % (
            category_data['category']
        )

    if category == constants.CD_USER_RIGHTS_CATEGORY_REVIEW_TRANSLATION:
        if language_code is None:
            raise Exception(
                'The language_code cannot be None if the review category is'
                ' \'translation\''
            )
        language_description = utils.get_supported_audio_language_description(
            language_code
        ).capitalize()
        role_description = category_data['role_description_template'] % (
            language_description
        )
        rights_message = category_data['rights_message_template'] % (
            language_description
        )
    else:
        role_description = category_data['role_description']
        rights_message = category_data['rights_message']

    email_body_template = (
        'Hi %s,<br><br>'
        'The Oppia team has removed you from the %s. You won\'t be able to %s '
        'any more, but you can still contribute %ss through the '
        '<a href="https://www.oppia.org/contributor-dashboard">'
        'Contributor Dashboard</a>.<br><br>'
        'Thanks, and happy contributing!<br><br>'
        'Best wishes,<br>'
        'The Oppia Community'
    )

    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if not server_can_send_emails:
        logging.error('This app cannot send emails to users.')
        return

    recipient_username = user_services.get_username(user_id)
    can_user_receive_email = user_services.get_email_preferences(
        user_id
    ).can_receive_email_updates

    # Send email only if recipient wants to receive.
    if can_user_receive_email:
        email_body = email_body_template % (
            recipient_username,
            role_description,
            rights_message,
            category_data['category'],
        )
        noreply_email_address = (
            platform_parameter_services.get_platform_parameter_value(
                platform_parameter_list.ParamName.NOREPLY_EMAIL_ADDRESS.value
            )
        )
        assert isinstance(noreply_email_address, str)
        _send_email(
            user_id,
            feconf.SYSTEM_COMMITTER_ID,
            feconf.EMAIL_INTENT_REMOVE_CD_USER,
            email_subject,
            email_body,
            noreply_email_address,
        )


def send_not_mergeable_change_list_to_admin_for_review(
    exp_id: str,
    frontend_version: int,
    backend_version: int,
    change_list_dict: Sequence[
        Mapping[str, change_domain.AcceptableChangeDictTypes]
    ],
) -> None:
    """Sends an email to the admin to review the not mergeable change list
    to improve the functionality in future if possible.

    Args:
        exp_id: str. The ID of an exploration on which the change list was
            to be applied.
        frontend_version: int. Version of an exploration from frontend on
            which a user is working.
        backend_version: int. Latest version of an exploration on which the
            change list can not be applied.
        change_list_dict: dict. Dict of the changes made by the
            user on the frontend, which are not mergeable.
    """
    email_subject = 'Some changes were rejected due to a conflict'
    email_body_template = (
        'Hi Admin,<br><br>'
        'Some draft changes were rejected in exploration %s because the '
        'changes were conflicting and could not be saved. Please see the '
        'rejected change list below:<br>'
        'Discarded change list: %s <br><br>'
        'Frontend Version: %s<br>'
        'Backend Version: %s<br><br>'
        'Thanks!'
    )

    server_can_send_emails = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SERVER_CAN_SEND_EMAILS.value
        )
    )
    if server_can_send_emails:
        email_body = email_body_template % (
            exp_id,
            change_list_dict,
            frontend_version,
            backend_version,
        )
        send_mail_to_admin(email_subject, email_body)


def verify_mailchimp_secret(secret: str) -> bool:
    """Verifies the secret key for the mailchimp webhook.

    Args:
        secret: str. The secret key provided by the mailchimp webhook.

    Returns:
        bool. Whether the secret key is valid.
    """
    mailchimp_webhook_secret = secrets_services.get_secret(
        'MAILCHIMP_WEBHOOK_SECRET'
    )
    if mailchimp_webhook_secret is None:
        logging.error('Mailchimp Webhook secret is not available.')
        return False

    return secret == mailchimp_webhook_secret


def _generate_attachments_for_failed_voiceovers(
    date: str,
    time: str,
    voiceover_regeneration_error_messages: List[
        Dict[str, List[Tuple[str, str]] | str]
    ],
) -> List[Dict[str, str]]:
    """Generates a text file containing details of failed voiceover
    regeneration attempts.

    Args:
        date: str. The date when the voiceover regeneration was attempted.
        time: str. The time when the voiceover regeneration was attempted.
        voiceover_regeneration_error_messages: list(dict). A list of
            dictionaries containing details of failed voiceover regeneration
            attempts. Each dictionary should have keys 'exploration_id',
            'language_accent', and 'error_message'.

    Returns:
        list(dict). A list of dictionaries with 'filename' and 'path' keys,
        where 'filename' is the name of the generated file and 'path' is the
        absolute path to the file.
    """

    filename = 'voiceover_regeneration_errors.txt'

    temp_dir = tempfile.gettempdir()
    file_path = pathlib.Path(temp_dir) / filename

    lines = [
        'Synthesis details for each piece of content are presented below.\n',
        'Date: %s\n' % date,
        'Time: %s\n\n' % time,
    ]

    for error_dict in voiceover_regeneration_error_messages:
        exploration_id = error_dict.get('exploration_id', '')
        language_accent = error_dict.get('language_accent', '')
        error_messages = error_dict.get('error_messages', [])

        lines.append('Exploration ID: %s\n' % exploration_id)
        lines.append('Language Accent: %s\n' % language_accent)

        for error_message in error_messages:
            lines.append('\n----------------------------------------\n')
            lines.append(
                'Content ID: %s, Error Message: %s\n'
                % (error_message[0], error_message[1])
            )
            lines.append('\n----------------------------------------\n')

    with open(file_path, 'w', encoding='utf-8') as file:
        file.writelines(lines)

    return [{'filename': filename, 'path': str(file_path)}]


def _delete_voiceover_error_attachments(
    filename_to_path: List[Dict[str, str]],
) -> None:
    """Deletes the files created for voiceover regeneration notification emails.

    Args:
        filename_to_path: list(dict). A list of dictionaries containing
            'filename' and 'path' keys for each file to be deleted.
    """
    for file_info in filename_to_path:
        file_path = pathlib.Path(file_info['path'])
        if file_path.exists():
            file_path.unlink()


def send_emails_to_voiceover_admins(
    date: str,
    time: str,
    exploration_id: str,
    exploration_title: str,
    number_of_requested_voiceovers: int,
    number_of_successful_voiceovers: int,
    number_of_failed_voiceovers: int,
    language_descriptions: List[str],
    author_username: str,
) -> None:
    """Sends an email to the voiceover admins with details of the
    voiceover regeneration attempt.

    Args:
        date: str. The date when the voiceover regeneration was attempted.
        time: str. The time when the voiceover regeneration was attempted.
        exploration_id: str. The ID of the exploration for which the
            voiceover regeneration was attempted.
        exploration_title: str. The title of the exploration.
        number_of_requested_voiceovers: int. The total number of voiceovers
            requested for regeneration.
        number_of_successful_voiceovers: int. The number of voiceovers that
            were successfully regenerated.
        number_of_failed_voiceovers: int. The number of voiceovers that failed
            to regenerate.
        language_descriptions: list(str). A list of language descriptions for
            which the voiceover regeneration was attempted.
        author_username: str. The username of the author who triggered voiceover
            regeneration.
    """
    exploration_link = 'https://www.oppia.org/create/%s' % exploration_id

    optional_message = (
        (
            '<br>You have also been cc’d on a separate email, sent to the '
            'voiceover tech lead, to address the failed voiceover synthesis. '
            'Please follow up on that email as needed.<br><br>'
        )
        if number_of_failed_voiceovers > 0
        else '<br>'
    )

    email_body = VOICEOVER_ADMINS_REGENERATION_NOTIFICATION_EMAIL[
        'email_body_template'
    ] % (
        author_username,
        exploration_link,
        exploration_title,
        ', '.join(language_descriptions),
        date,
        time,
        number_of_requested_voiceovers,
        number_of_successful_voiceovers,
        number_of_failed_voiceovers,
        optional_message,
    )

    email_subject = (
        VOICEOVER_ADMINS_REGENERATION_NOTIFICATION_EMAIL['email_subject']
        % exploration_title
    )

    system_email_address = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SYSTEM_EMAIL_ADDRESS.value
        )
    )

    receipient_id = 'voiceovers-leads'
    _send_email(
        receipient_id,
        feconf.SYSTEM_COMMITTER_ID,
        feconf.EMAIL_INTENT_VOICEOVER_REGENERATION,
        email_subject,
        email_body,
        str(system_email_address),
        recipient_email=VOICEOVER_ADMIN_GOOGLE_GROUP,
    )


def send_emails_to_voiceover_tech_leads(
    exploration_id: str,
    exploration_title: str,
    date: str,
    time: str,
    language_descriptions: List[str],
    voiceover_regeneration_error_messages: List[
        Dict[str, List[Tuple[str, str]] | str]
    ],
) -> None:
    """Sends an email to the voiceover tech leads with details of failed
    voiceover regeneration attempts.

    Args:
        exploration_id: str. The ID of the exploration for which the
            voiceover regeneration failed.
        exploration_title: str. The title of the exploration.
        date: str. The date when the voiceover regeneration was attempted.
        time: str. The time when the voiceover regeneration was attempted.
        language_descriptions: list(str). A list of language descriptions for
            which the voiceover regeneration was attempted.
        voiceover_regeneration_error_messages: list(dict). A list of
            dictionaries containing details of failed voiceover regeneration
            attempts. Each dictionary should have keys 'exploration_id',
            'language_accent', and 'error_message'.
    """
    exploration_link = 'https://www.oppia.org/create/%s' % exploration_id
    document_link_1 = (
        'https://docs.google.com/document/d/1Wwd0Eg2jA3rnsiER6cf-ixOm3oh0T_c0'
        'lXnaRwbSWH4/edit?tab=t.0#heading=h.uc9ozkinrt05'
    )
    document_link_2 = (
        'https://docs.google.com/document/d/1Wwd0Eg2jA3rnsiER6cf-ixOm3oh0T_c0'
        'lXnaRwbSWH4/edit?tab=t.0#heading=h.y85o1y4ceo9y'
    )

    email_body = VOICEOVER_TECH_LEADS_REGENERATION_NOTIFICATION_EMAIL[
        'email_body_template'
    ] % (
        exploration_link,
        exploration_title,
        ', '.join(language_descriptions),
        date,
        time,
        document_link_1,
        document_link_2,
        document_link_1,
        document_link_2,
    )
    email_subject = VOICEOVER_TECH_LEADS_REGENERATION_NOTIFICATION_EMAIL[
        'email_subject'
    ]

    system_email_address = (
        platform_parameter_services.get_platform_parameter_value(
            platform_parameter_list.ParamName.SYSTEM_EMAIL_ADDRESS.value
        )
    )

    filename_to_path = _generate_attachments_for_failed_voiceovers(
        date, time, voiceover_regeneration_error_messages
    )

    receipient_id = 'voiceover-tech-support'
    _send_email(
        receipient_id,
        feconf.SYSTEM_COMMITTER_ID,
        feconf.EMAIL_INTENT_VOICEOVER_REGENERATION,
        email_subject,
        email_body,
        str(system_email_address),
        recipient_email=VOICEOVER_TECH_LEADS_GOOGLE_GROUP,
        cc_emails=[VOICEOVER_ADMIN_GOOGLE_GROUP],
        attachments=filename_to_path,
    )

    _delete_voiceover_error_attachments(filename_to_path)