"""SMTP email backend class."""

import email.policy
import smtplib
import ssl
import threading
from email.headerregistry import Address, AddressHeader

from django.conf import settings
from django.core.mail import InvalidMailer
from django.core.mail.backends.base import BaseEmailBackend
from django.core.mail.utils import DNS_NAME
from django.utils.deprecation import RemovedInDjango70Warning, warn_about_external_use
from django.utils.encoding import force_str, punycode
from django.utils.functional import cached_property


class EmailBackend(BaseEmailBackend):
    """
    A wrapper that manages the SMTP network connection.
    """

    def __init__(
        self,
        host=None,
        port=None,
        username=None,
        password=None,
        use_tls=None,
        fail_silently=False,
        use_ssl=None,
        timeout=None,
        ssl_keyfile=None,
        ssl_certfile=None,
        **kwargs,
    ):
        # RemovedInDjango70Warning.
        if "alias" not in kwargs:
            msg = (
                "Directly creating EmailBackend instances is deprecated. "
                "Use mail.mailers instead."
            )
            warn_about_external_use(msg, RemovedInDjango70Warning)

        super().__init__(**kwargs)
        self.fail_silently = fail_silently
        self.connection = None
        self._partial_connection = None
        self._lock = threading.RLock()

        # RemovedInDjango70Warning.
        if self.alias is None:
            # Use deprecated settings when MAILERS not enabled.
            self.host = host or settings.EMAIL_HOST
            self.port = port or settings.EMAIL_PORT
            self.username = settings.EMAIL_HOST_USER if username is None else username
            self.password = (
                settings.EMAIL_HOST_PASSWORD if password is None else password
            )
            self.use_tls = settings.EMAIL_USE_TLS if use_tls is None else use_tls
            self.use_ssl = settings.EMAIL_USE_SSL if use_ssl is None else use_ssl
            self.timeout = settings.EMAIL_TIMEOUT if timeout is None else timeout
            self.ssl_keyfile = (
                settings.EMAIL_SSL_KEYFILE if ssl_keyfile is None else ssl_keyfile
            )
            self.ssl_certfile = (
                settings.EMAIL_SSL_CERTFILE if ssl_certfile is None else ssl_certfile
            )
            if self.use_ssl and self.use_tls:
                raise ValueError(
                    "EMAIL_USE_TLS/EMAIL_USE_SSL are mutually exclusive, so "
                    "only set one of those settings to True."
                )
            return

        self.host = host
        self.port = port
        self.username = username
        self.password = password
        self.use_tls = use_tls if use_tls is not None else False
        self.use_ssl = use_ssl if use_ssl is not None else False
        self.timeout = timeout
        self.ssl_keyfile = ssl_keyfile
        self.ssl_certfile = ssl_certfile
        if self.host is None:
            raise InvalidMailer("OPTIONS must define 'host'.", alias=self.alias)
        if self.use_ssl and self.use_tls:
            raise InvalidMailer(
                "The 'use_ssl' and 'use_tls' OPTIONS are incompatible. "
                "Set at most one of them to True.",
                alias=self.alias,
            )
        if self.port is None:
            if self.use_ssl:
                self.port = 465
            elif self.use_tls:
                self.port = 587
            else:
                self.port = 25

    @property
    def connection_class(self):
        return smtplib.SMTP_SSL if self.use_ssl else smtplib.SMTP

    @cached_property
    def ssl_context(self):
        if self.ssl_certfile or self.ssl_keyfile:
            ssl_context = ssl.SSLContext(protocol=ssl.PROTOCOL_TLS_CLIENT)
            ssl_context.load_cert_chain(self.ssl_certfile, self.ssl_keyfile)
            return ssl_context
        else:
            return ssl.create_default_context()

    def open(self):
        """
        Ensure an open connection to the email server. Return whether or not a
        new connection was required (True or False) or None if an exception
        passed silently.
        """
        if self.connection:
            # Nothing to do if the connection is already open.
            return False

        # If a connection was partially opened before, close it.
        if self._partial_connection is not None:
            self._close_connection(self._partial_connection)
            self._partial_connection = None

        # If local_hostname is not specified, socket.getfqdn() gets used.
        # For performance, we use the cached FQDN for local_hostname.
        connection_params = {"local_hostname": DNS_NAME.get_fqdn()}
        if self.timeout is not None:
            connection_params["timeout"] = self.timeout
        if self.use_ssl:
            connection_params["context"] = self.ssl_context
        try:
            self._partial_connection = self.connection_class(
                self.host, self.port, **connection_params
            )

            # TLS/SSL are mutually exclusive, so only attempt TLS over
            # non-secure connections.
            if not self.use_ssl and self.use_tls:
                self._partial_connection.starttls(context=self.ssl_context)
            if self.username and self.password:
                self._partial_connection.login(self.username, self.password)

            # Don't set connection until it's fully configured.
            self.connection = self._partial_connection
            self._partial_connection = None

            return True
        except OSError:
            if not self.fail_silently:
                raise

    def _close_connection(self, connection):
        try:
            connection.quit()
        except (ssl.SSLError, smtplib.SMTPServerDisconnected):
            # This happens when calling quit() on a TLS connection
            # sometimes, or when the connection was already disconnected
            # by the server.
            connection.close()
        except smtplib.SMTPException:
            if self.fail_silently:
                return
            raise

    def close(self):
        """Close the connection to the email server."""
        if self._partial_connection is not None:
            try:
                self._close_connection(self._partial_connection)
            finally:
                self._partial_connection = None
        if self.connection is not None:
            try:
                self._close_connection(self.connection)
            finally:
                self.connection = None

    def send_messages(self, email_messages):
        """
        Send one or more EmailMessage objects and return the number of email
        messages sent.
        """
        if not email_messages:
            return 0
        with self._lock:
            new_conn_created = self.open()
            if not self.connection or new_conn_created is None:
                # We failed silently on open().
                # Trying to send would be pointless.
                return 0
            num_sent = 0
            try:
                for message in email_messages:
                    sent = self._send(message)
                    if sent:
                        num_sent += 1
            finally:
                if new_conn_created:
                    self.close()
        return num_sent

    def _send(self, email_message):
        """A helper method that does the actual sending."""
        if not email_message.recipients():
            return False
        from_email = self.prep_address(email_message.from_email)
        recipients = [self.prep_address(addr) for addr in email_message.recipients()]
        message = email_message.message(policy=email.policy.SMTP)
        try:
            self.connection.sendmail(from_email, recipients, message.as_bytes())
        except smtplib.SMTPException:
            if not self.fail_silently:
                raise
            return False
        return True

    def prep_address(self, address, force_ascii=True):
        """
        Return the addr-spec portion of an email address. Raises ValueError for
        invalid addresses, including CR/NL injection.

        If force_ascii is True, apply IDNA encoding to non-ASCII domains, and
        raise ValueError for non-ASCII local-parts (which can't be encoded).
        Otherwise, leave Unicode characters unencoded (e.g., for sending with
        SMTPUTF8).
        """
        address = force_str(address)
        parsed = AddressHeader.value_parser(address)
        defects = set(str(defect) for defect in parsed.all_defects)
        # Django allows local mailboxes like "From: webmaster" (#15042).
        defects.discard("addr-spec local part with no domain")
        if not force_ascii:
            # PY315: Non-ASCII local-part is valid with SMTPUTF8. This check
            # can be removed once the minimum supported Python version is 3.15
            # (so the fix for https://github.com/python/cpython/issues/81074
            # is in all supported versions).
            defects.discard("local-part contains non-ASCII characters)")
        if defects:
            raise ValueError(f"Invalid address {address!r}: {'; '.join(defects)}")

        mailboxes = parsed.all_mailboxes
        if len(mailboxes) != 1:
            raise ValueError(f"Invalid address {address!r}: must be a single address")

        mailbox = mailboxes[0]
        if force_ascii and mailbox.domain and not mailbox.domain.isascii():
            # Re-compose an addr-spec with the IDNA encoded domain.
            domain = punycode(mailbox.domain)
            return str(Address(username=mailbox.local_part, domain=domain))
        else:
            return mailbox.addr_spec
