HiveBrain v1.2.0
Get Started
← Back to all entries
patternpythonMinor

Email redirection script in python

Submitted by: @import:stackexchange-codereview··
0
Viewed 0 times
scriptpythonemailredirection

Problem

I am new to python and wrote a email redirect script. It extracts mails from a server and than resend them. I would like to get some feedback:

```
import sys
import smtplib
import poplib
import re
import argparse

def validate_email_address(address):
email_pattern = '^[^@]+@[^@]+\.[^@]+$'
if not re.match(email_pattern, address ):
raise argparse.ArgumentTypeError('{} is not a valid email address'.format(address))
return address

arg_parser = argparse.ArgumentParser()
arg_parser.add_argument('popServer', help='pop server host or address of extracted account')
arg_parser.add_argument('smtpServer', help='smtp server host or address of extracted account')
arg_parser.add_argument('address', help='extract messages from this account', type=validate_email_address)
arg_parser.add_argument('password', help='password of extracted account')
arg_parser.add_argument('target', help='redirect messages to this account', type=validate_email_address)
arg_parser.add_argument('-useTSL', action='store_true')
arg_parser.add_argument('-useSSL', action='store_true')
args = arg_parser.parse_args()
timeout = 10
try:
if args.useSSL: pop_conn = poplib.POP3_SSL(args.popServer, timeout=timeout)
else: pop_conn = poplib.POP3(args.popServer, timeout=timeout)
pop_conn.user(args.address)
pop_conn.pass_(args.password)
mail_count = len(pop_conn.list()[1])
print('Mail count:{}'.format(mail_count))
if mail_count > 0:
print('Requesting mails from {0} for {1}'.format(args.popServer, args.address))
for i in range(1, mail_count + 1):
try:
print("Receiving mail {0}/{1}".format(i,mail_count))
message = (pop_conn.retr(i)[1])
pop_conn.dele(i)
message = [line.decode('utf-8', errors='ignore') for line in message]
message = '\n'.join(message)
print("Redirecting mail {0}/{1}".format(i,mail_count))
if(args.useSSL): smtp_conn = smtpli

Solution

OK, here it goes:

Currently all of your code is bunched up in a code block. It would be better to follow the single responsibility principle and have functions for every job.

Therefore, I used a main function, a parse_args function, a pop_connect function to login with the POP3 server, a smtp_connect function to do the same with the SMTP server, a get_mail function to retrieve a mail from the server and catch any exception and a send_mail function to encode the message and send it.

For the SMTP connection I used the with..as functionality of python. This makes sure that the SMTP connection is closed, regardless of any errors in between. Also, it is sufficient to connect to the SMTP server once and then re-use that connection, unless your latency is really bad (and in that case, just increase the timeout, which I mad an optional argument). Have a look at this website to read a more thorough explanation on how to use a contextmanager on a function.

I also made some formats slightly simpler. If the arguments are in order, it is not necessary, to use "{0}{1}".format(x, y), "{}{}".format(x, y) is sufficient. On the other hand, I also used the fact that format understands a few rudimentary things, namely indices and attributes:

>>> l = [1,2,3]
>>> "{0[1]}".format(l)
2
>>> from collections import namedtuple
>>> Box = namedtuple("Box", "width height")
>>> b = Box(2, 3)
>>> "{0.width}, {0.height}".format(b)
2, 3


Finally, I used the if __name__ == "__main__": idiom, to allow importing these functions from another module without running all the code.

```
import sys
import smtplib
import poplib
import re
import argparse

from contextlib import contextmanager

def validate_email_address(address):
email_pattern = '^[^@]+@[^@]+\.[^@]+$'
if not re.match(email_pattern, address):
raise argparse.ArgumentTypeError(
'{} is not a valid email address'.format(address))
return address

def parse_args():
arg_parser = argparse.ArgumentParser()
arg_parser.add_argument(
'popServer', help='pop server host or address of extracted account')
arg_parser.add_argument(
'smtpServer', help='smtp server host or address of extracted account')
arg_parser.add_argument(
'address', help='extract messages from this account', type=validate_email_address)
arg_parser.add_argument('password', help='password of extracted account')
arg_parser.add_argument(
'target', help='redirect messages to this account', type=validate_email_address)
arg_parser.add_argument('--useTSL', action='store_true')
arg_parser.add_argument('--useSSL', action='store_true')
arg_parser.add_argument('--timeout', type=int,
help="Timeout for the servers", default=10)
return arg_parser.parse_args()

def pop_connect(server, user, password, timeout=10, use_SSL=False):
pop = poplib.POP3_SSL if use_SSL else poplib.POP3
pop_conn = pop(server, timeout=timeout)
pop_conn.user(user)
pop_conn.pass_(password)
return pop_conn

@contextmanager
def smtp_connect(server, user, password, timeout=10, use_SSL=False, use_TSL=False):
try:
smtp = smtplib.SMTP_SSL if use_SSL else smtplib.SMTP
smtp_conn = smtp(server, timeout=timeout)
if use_TSL:
smtp_conn.starttls()
smtp_conn.login(user, password)
yield smtp_conn
except smtplib.SMTPException as SMTP_exception:
print('Request of mail failed:{}'.format(SMTP_exception))
raise
finally:
try:
smtp_conn.quit()
except UnboundLocalError:
pass

def get_email(pop_conn, i):
try:
return pop_conn.retr(i)[1]
except poplib.error_proto as pop_error:
print('Receiving of mail failed:{0}'.format(pop_error))

def send_mail(smtp_conn, to_email, from_email, message):
message = '\n'.join(line.decode(
'utf-8', errors='ignore') for line in message)
smtp_conn.sendmail(from_email, to_email, message.encode(
encoding='ascii', errors='ignore'))

def main():
args = parse_args()
try:
pop_conn = pop_connect(args.popServer, args.address,
args.password, args.timeout, args.useSSL)
mail_count = len(pop_conn.list()[1])
print('Mail count:{}'.format(mail_count))
if mail_count:
print('Requesting mails from {0.popServer} for {0.address}'.format(args))
with smtp_connect(args.smtpServer, args.address, args.password, args.timeout, args.useSSL, args.useTSL) as smtp_conn:
for i in range(1, mail_count + 1):
print("Receiving mail {}/{}".format(i, mail_count))
message = get_email(pop_conn, i)
pop_conn.dele(i)
print("Redirecting mail {}/{}".format(i, mail_count))
send_mail(smtp_conn, args.address, args.target, message)
except poplib.error_proto as

Code Snippets

>>> l = [1,2,3]
>>> "{0[1]}".format(l)
2
>>> from collections import namedtuple
>>> Box = namedtuple("Box", "width height")
>>> b = Box(2, 3)
>>> "{0.width}, {0.height}".format(b)
2, 3
import sys
import smtplib
import poplib
import re
import argparse

from contextlib import contextmanager


def validate_email_address(address):
    email_pattern = '^[^@]+@[^@]+\.[^@]+$'
    if not re.match(email_pattern, address):
        raise argparse.ArgumentTypeError(
            '{} is not a valid email address'.format(address))
    return address


def parse_args():
    arg_parser = argparse.ArgumentParser()
    arg_parser.add_argument(
        'popServer', help='pop server host or address of extracted account')
    arg_parser.add_argument(
        'smtpServer', help='smtp server host or address of extracted account')
    arg_parser.add_argument(
        'address', help='extract messages from this account', type=validate_email_address)
    arg_parser.add_argument('password', help='password of extracted account')
    arg_parser.add_argument(
        'target', help='redirect messages to this account', type=validate_email_address)
    arg_parser.add_argument('--useTSL', action='store_true')
    arg_parser.add_argument('--useSSL', action='store_true')
    arg_parser.add_argument('--timeout', type=int,
                            help="Timeout for the servers", default=10)
    return arg_parser.parse_args()


def pop_connect(server, user, password, timeout=10, use_SSL=False):
    pop = poplib.POP3_SSL if use_SSL else poplib.POP3
    pop_conn = pop(server, timeout=timeout)
    pop_conn.user(user)
    pop_conn.pass_(password)
    return pop_conn


@contextmanager
def smtp_connect(server, user, password, timeout=10, use_SSL=False, use_TSL=False):
    try:
        smtp = smtplib.SMTP_SSL if use_SSL else smtplib.SMTP
        smtp_conn = smtp(server, timeout=timeout)
        if use_TSL:
            smtp_conn.starttls()
        smtp_conn.login(user, password)
        yield smtp_conn
    except smtplib.SMTPException as SMTP_exception:
        print('Request of mail failed:{}'.format(SMTP_exception))
        raise
    finally:
        try:
            smtp_conn.quit()
        except UnboundLocalError:
            pass


def get_email(pop_conn, i):
    try:
        return pop_conn.retr(i)[1]
    except poplib.error_proto as pop_error:
        print('Receiving of mail failed:{0}'.format(pop_error))


def send_mail(smtp_conn, to_email, from_email, message):
    message = '\n'.join(line.decode(
        'utf-8', errors='ignore') for line in message)
    smtp_conn.sendmail(from_email, to_email, message.encode(
        encoding='ascii', errors='ignore'))


def main():
    args = parse_args()
    try:
        pop_conn = pop_connect(args.popServer, args.address,
                               args.password, args.timeout, args.useSSL)
        mail_count = len(pop_conn.list()[1])
        print('Mail count:{}'.format(mail_count))
        if mail_count:
            print('Requesting mails from {0.popServer} for {0.address}'.format(args))
            with smtp_connect(args.smtpServer, args.address, args.password, args.timeout, args.useSSL, args.useTSL) as smtp_conn:

Context

StackExchange Code Review Q#148162, answer score: 3

Revisions (0)

No revisions yet.