From aee36484f008408cb3af82972863f1192699b520 Mon Sep 17 00:00:00 2001 From: Anton Khirnov Date: Mon, 22 Nov 2021 18:37:52 +0100 Subject: commands/thread: split some reply handling code to a separate module --- alot/commands/thread.py | 134 ++++-------------------------------------------- alot/mail/reply.py | 129 ++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 140 insertions(+), 123 deletions(-) create mode 100644 alot/mail/reply.py diff --git a/alot/commands/thread.py b/alot/commands/thread.py index fe84695b..9b3c56b9 100644 --- a/alot/commands/thread.py +++ b/alot/commands/thread.py @@ -35,6 +35,7 @@ from ..helper import split_commandstring from ..mail.attachment import Attachment from ..mail.envelope import Envelope from ..mail import headers as HDR +from ..mail import reply from ..settings.const import settings from ..utils import argparse as cargparse from ..utils.mailcap import MailcapHandler @@ -70,64 +71,6 @@ def clear_my_address(my_account, value): new_value.append(formataddr((name, address))) return new_value -def determine_sender(headers, action='reply'): - """ - Inspect a given mail to reply/forward/bounce and find the most appropriate - account to act from and construct a suitable From-Header to use. - - :param headers: headers of the email to inspect - :type headers: `db.message._MessageHeaders` - :param action: intended use case: one of "reply", "forward" or "bounce" - :type action: str - """ - assert action in ['reply', 'forward', 'bounce'] - - # 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. - candidate_headers = settings.get("reply_account_header_priority") - for candidate_header in candidate_headers: - 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 - - @registerCommand(MODE, 'reply', arguments=[ (['--all'], {'action': 'store_true', 'help': 'reply to all'}), (['--list'], {'action': cargparse.BooleanAction, 'default': None, @@ -155,18 +98,6 @@ class ReplyCommand(Command): self.force_spawn = spawn super().__init__(**kwargs) - def _reply_subject(self, 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 _list_reply(self, message): # Auto-detect ML if HDR.LIST_ID in message.headers and self._force_list_reply is None: @@ -174,52 +105,6 @@ class ReplyCommand(Command): return bool(self._force_list_reply) - def _reply_references(self, 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(self, 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(addr in lists for n, addr in getaddresses(recipients)): - followupto = ', '.join(recipients) - logging.debug('mail followup to: %s', followupto) - return followupto - - def _build_body_text(self, 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 _determine_recipients(self, message, account): # set To sender = (message.headers.get(HDR.REPLY_TO) or @@ -286,11 +171,12 @@ class ReplyCommand(Command): # construct reply headers headers = {} - headers[HDR.SUBJECT] = self._reply_subject(message) + headers[HDR.SUBJECT] = reply.subject(message) # set From-header and sending account try: - from_header, account = determine_sender(message.headers, 'reply') + from_header, account = reply.determine_account(message.headers, + reply.ACT_REPLY) except AssertionError as e: ui.notify(str(e), priority='error') return @@ -304,16 +190,16 @@ class ReplyCommand(Command): # set In-Reply-To + References headers headers[HDR.IN_REPLY_TO] = '<%s>' % message.id - headers[HDR.REFERENCES] = self._reply_references(message) + headers[HDR.REFERENCES] = reply.references(message) # if any of the recipients is a mailinglist that we are subscribed to, # set Mail-Followup-To header so that duplicates are avoided # to and cc are already cleared of our own address - mft = self._mail_followup_to([to, cc]) + mft = reply.mail_followup_to([to, cc]) if mft: headers[HDR.MAIL_FOLLOWUP_TO] = mft - body_text = self._build_body_text(message, ui) + body_text = reply.body_text(message, ui) # construct the envelope envelope = Envelope(headers = headers, bodytext = body_text, @@ -396,7 +282,8 @@ class ForwardCommand(Command): # set From-header and sending account try: - from_header, account = determine_sender(message.headers, 'forward') + from_header, account = reply.determine_account(message.headers, + reply.ACT_FORWARD) except AssertionError as e: ui.notify(str(e), priority='error') return @@ -434,7 +321,8 @@ class BounceMailCommand(Command): # set Resent-From-header and sending account try: - resent_from_header, account = determine_sender(message.headers, 'bounce') + resent_from_header, account = reply.determine_account(message.headers, + reply.ACT_BOUNCE) except AssertionError as e: ui.notify(str(e), priority='error') return diff --git a/alot/mail/reply.py b/alot/mail/reply.py new file mode 100644 index 00000000..0e7d897e --- /dev/null +++ b/alot/mail/reply.py @@ -0,0 +1,129 @@ +# 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 +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 + + +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(addr in lists for n, addr in getaddresses(recipients)): + followupto = ', '.join(recipients) + logging.debug('mail followup to: %s', followupto) + return followupto -- cgit v1.2.3