path: root/alot/mail/reply.py

# This file is released under the GNU GPL, version 3 or a later revision.
# For further details see the COPYING file

import email.policy
from   email.utils import getaddresses
from   enum        import Enum, auto
import logging

from .                      import headers as HDR

from ..helper               import formataddr
from ..settings.const       import settings

ACT_REPLY  = 'reply'
ACT_BOUNCE = 'bounce'
ACT_FWD    = 'forward'

def determine_account(headers, action = ACT_REPLY):
    """
    Determine the local account to use for acting on a message with the given
    headers.

    :param headers: headers of the email to inspect
    :type headers: `db.message._MessageHeaders`
    :param action: intended use case: one of the ACT_* constants
    :type action: str
    :returns: 2-tuple of (from address, account)
    """
    # get accounts
    my_accounts = settings.get_accounts()
    assert my_accounts, 'no accounts set!'

    # extract list of addresses to check for my address
    # X-Envelope-To and Envelope-To are used to store the recipient address
    # if not included in other fields
    # Process the headers in order of importance: if a mail was sent with
    # account X, with account Y in e.g. CC or delivered-to, make sure that
    # account X is the one selected and not account Y.
    for candidate_header in settings.get("reply_account_header_priority"):
        candidate_addresses = getaddresses(headers.get_all(candidate_header))

        logging.debug('candidate addresses: %s', candidate_addresses)
        # pick the most important account that has an address in candidates
        # and use that account's realname and the address found here
        for account in my_accounts:
            for seen_name, seen_address in candidate_addresses:
                if account.matches_address(seen_address):
                    if settings.get(action + '_force_realname'):
                        realname = account.realname
                    else:
                        realname = seen_name
                    if settings.get(action + '_force_address'):
                        address = str(account.address)
                    else:
                        address = seen_address

                    logging.debug('using realname: "%s"', realname)
                    logging.debug('using address: %s', address)

                    from_value = formataddr((realname, address))
                    return from_value, account

    # revert to default account if nothing found
    account = my_accounts[0]
    realname = account.realname
    address = account.address
    logging.debug('using realname: "%s"', realname)
    logging.debug('using address: %s', address)

    from_value = formataddr((realname, str(address)))
    return from_value, account

class ReplyMode(Enum):
    AUTHOR = auto()
    GROUP  = auto()
    LIST   = auto()

class AddrList:
    """
    An ordered set of unique  addresses.
    """

    # list of Address objects
    _addrs      = None
    # mapping of addr-spec -> Address
    _addr_specs = None

    def __init__(self, addresses):
        self._addrs      = []
        self._addr_specs = {}

        for addr in addresses:
            self.add(addr)

    def __str__(self):
        return ', '.join((str(a) for a in self._addrs))

    def __len__(self):
        return len(self._addrs)

    def __iter__(self):
        return iter(self._addrs)

    def _addr_to_addr_spec(self, addr):
        if hasattr(addr, 'addr_spec'):
            # headerregistry.Address
            return addr.addr_spec

        # account.Address
        return str(addr)

    def __contains__(self, addr):
        return self._addr_to_addr_spec(addr) in self._addr_specs

    def __delitem__(self, addr):
        addr_spec = self._addr_to_addr_spec(addr)
        idx       = self._addrs.index(addr_spec)
        del self._addrs[idx]
        del self._addr_specs[addr_spec]

    def __add__(self, other):
        ret = self.__class__(self)
        for a in other:
            ret.add(a)
        return ret

    def add(self, addr):
        if not addr.addr_spec:
            raise ValueError('Missing addr-spec')

        if addr in self:
            idx = self._addrs.index(self._addr_specs[addr.addr_spec])

            # if the address we already have is missing display name,
            # replace it with the new one
            if not self._addrs[idx].display_name and addr.display_name:
                self._addrs[idx]                 = addr
                self._addr_specs[addr.addr_spec] = addr

            return

        self._addrs.append(addr)
        self._addr_specs[addr.addr_spec] = addr

def _reply_list(message):
    # To choose the target of the reply --list
    # Reply-To is standart reply target RFC 2822:, RFC 1036: 2.2.1
    # X-BeenThere is needed by sourceforge ML also winehq
    # X-Mailing-List is also standart and is used by git-send-mail
    # Some mail server (gmail) will not resend you own mail, so we
    # fall back to 'To' as last resort
    to = message.headers.pick_addrlist_header(
            HDR.X_BEEN_THERE, HDR.X_MAILING_LIST, HDR.REPLY_TO, HDR.TO)

    # copy original Cc
    cc = message.headers.pick_addrlist_header(HDR.CC)

    return to, cc

def _reply_group(message, senders):
    # make sure that our own address is not included
    # if the message was self-sent, then our address is not included
    MFT = message.headers.pick_addrlist_header(HDR.MAIL_FOLLOWUP_TO)
    if MFT and settings.get('honor_followup_to'):
        logging.debug('honor followup to: %s', MFT)
        to = MFT
        cc = ()
    else:
        to = message.headers.pick_addrlist_header(HDR.TO) + senders
        cc = message.headers.pick_addrlist_header(HDR.CC)

    return to, cc

def determine_recipients(message, account, mode):
    # TODO: detect reply-to munging by ML here?
    senders = message.headers.pick_addrlist_header(HDR.MAIL_REPLY_TO,
                HDR.REPLY_TO, HDR.FROM)

    self_reply = len(senders) == 1 and account.matches_address(senders[0].addr_spec)
    if self_reply:
        senders = message.headers.get(HDR.TO)
        senders = senders.addresses if senders else ()
        logging.debug('Replying to own message, invert senders to: %s',
                      senders)

    if mode == ReplyMode.LIST:
        to, cc = _reply_list(message)
    elif mode == ReplyMode.GROUP:
        to, cc = _reply_group(message, senders)
    else: # mode == ReplyMode.AUTHOR
        to = senders
        cc = ()

    to = AddrList(to)
    cc = AddrList(cc)

    # remove duplicates across to/cc
    for a in to:
        if a in cc:
            del cc[a]

    # remove sending account from the address lists
    for a in [account.address] + account.aliases:
        # keep own address if it's the only recipient
        # i.e. the user deliberately wants to self-send the reply
        if a in to and len(to) > 1:
            del to[a]
        if a in cc:
            del cc[a]

    logging.debug('%s reply to: %s %s', mode, str(to), str(cc))

    return to, cc

def subject(message):
    subject = message.headers.get(HDR.SUBJECT) or ''
    reply_subject_hook = settings.get_hook('reply_subject')
    if reply_subject_hook:
        subject = reply_subject_hook(subject)
    else:
        rsp = settings.get('reply_subject_prefix')
        if not subject.lower().startswith(('re:', rsp.lower())):
            subject = rsp + subject

    return subject

def body_text(message, ui):
    name, address = message.get_author()
    timestamp     = message.date
    qf = settings.get_hook('reply_prefix')
    if qf:
        mail = email.message_from_bytes(message.as_bytes(),
                                        policy = email.policy.SMTP)
        quotestring = qf(name, address, timestamp,
                         message=mail, ui=ui, dbm=ui.dbman)
    else:
        quotestring = 'Quoting %s (%s)\n' % (name or address, timestamp)
    mailcontent = quotestring
    quotehook = settings.get_hook('text_quote')
    if quotehook:
        mailcontent += quotehook(message.get_body_text())
    else:
        quote_prefix = settings.get('quote_prefix')
        for line in message.get_body_text().splitlines():
            mailcontent += quote_prefix + line + '\n'

    return mailcontent

def references(message):
    old_references = message.headers.get(HDR.REFERENCES)
    references     = []

    if old_references:
        references = old_references.split()
        # limit to 16 references, including the one we add
        del references[1:-14]

    references.append('<%s>' % message.id)
    return ' '.join(references)

def mail_followup_to(recipients):
    if not settings.get('followup_to'):
        return None

    lists = settings.get('mailinglists')

    # check if any recipient address matches a known mailing list
    if any(a.addr_spec in lists for a in recipients):
        logging.debug('mail followup to: %s', recipients)
        return recipients