adelgado/imap_filter
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add account details in mailbox definition

Browse source
This commit is contained in:
Antonio J. Delgado 2025-02-07 12:08:11 +02:00
parent 4977c05c38
commit 8d8dbd39d1
1 changed files with 43 additions and 20 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

63
imap_filter/imap_filter.py
View file

@ -26,6 +26,14 @@ except ImportError:
import click
import click_config_file
REQUIRED_MAILBOX_FIELDS = [
'mailbox',
'imap_user',
'imap_password_file',
'imap_server',
'imap_port',
'imap_ssl'
]
class ImapFilter:
'''IMAP filter tool'''
@ -46,6 +54,7 @@ class ImapFilter:
)
self._init_log()
signal(SIGINT, self._signal_handler)
self.imap = None
with open(self.config['filters_file'], 'r', encoding='UTF-8') as filters_file:
self._log.debug(
"Reading filters file '%s'...",
@ -67,7 +76,6 @@ class ImapFilter:
"Filters file is empty. Use --help to see more details."
)
sys.exit(1)
self.connect_imap()
self._process_filters()
self._log.debug(
"Took %s seconds to process",
@ -77,12 +85,21 @@ class ImapFilter:
def _process_filters(self):
matches = 0
for mailbox in self.config['mailboxes']:
if 'mailbox' not in mailbox:
self._log.warning(
"No mailbox name in %s, skipping.",
mailbox
)
break
for field in REQUIRED_MAILBOX_FIELDS:
if field not in mailbox:
self._log.warning(
"Missing required field '%s' in mailbox '%s', skipping.",
field,
mailbox
)
break
self.connect_imap(
imap_server=mailbox['imap_server'],
imap_port=mailbox['imap_port'],
ssl=mailbox['imap_ssl'],
imap_user=mailbox['imap_user'],
imap_password_file=mailbox['imap_password_file'],
)
if mailbox['mailbox'] == 'INBOX' and self.config['sieve_scripts_path']:
self._create_sieve_script(mailbox['filters'])
else:
@ -405,38 +422,44 @@ class ImapFilter:
return False
return True
def connect_imap(self):
def connect_imap(self, imap_server, imap_port, ssl, imap_user, imap_password_file):
'''Create connection object to the IMAP server'''
self._log.debug(
'Connecting to server %s:%s...',
self.config['imap_server'],
self.config['imap_port']
imap_server,
imap_port,
)
if self.config['ssl']:
if ssl:
try:
self.imap = imaplib.IMAP4_SSL(self.config['imap_server'], self.config['imap_port'])
self.imap = imaplib.IMAP4_SSL(imap_server, imap_port)
except Exception as error:
self._log.error(
"Error connecting securely to IMAP server '%s'. %s",
self.config['imap_server'],
error
imap_server,
error,
)
sys.exit(1)
else:
try:
self.imap = imaplib.IMAP4(self.config['imap_server'], self.config['imap_port'])
self.imap = imaplib.IMAP4(imap_server, imap_port)
except Exception as error:
self._log.error(
"Error connecting to IMAP server '%s'. %s",
self.config['imap_server'],
error
imap_server,
error,
)
sys.exit(2)
with open(imap_password_file, 'r', encoding='utf-8') as pass_file:
imap_password = pass_file.read().trim()
try:
self._log.debug('Authenticating as user %s...', self.config['imap_user'])
self.imap.login(self.config['imap_user'], self.config['imap_password'])
self._log.debug('Authenticating as user %s...', imap_user)
self.imap.login(imap_user, imap_password)
except Exception as error:
self._log.error("Error while login as '%s'. %s'", self.config['imap_user'], error)
self._log.error(
"Error while login as '%s'. %s'",
imap_user,
error,
)
self.imap.close()
sys.exit(3)