zev/imbox
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #129 from zevaverbach/encapsulate_generator

Browse Source 
Created Message class to encapsulate the "messages" generator
This commit is contained in:
Martin Rusev
2018-07-25 19:46:49 +02:00
committed by GitHub
parent a1801af56e 73fafcb368
commit 34149efed5
6 changed files with 87 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
.gitignore vendored
View File

@@ -31,3 +31,6 @@ nosetests.xml
example.*
example.py
# PyCharm
.idea/

89
imbox/__init__.py
View File

@@ -1,8 +1,9 @@
from imbox.imap import ImapTransport
from imbox.parser import parse_email
from imbox.parser import parse_email, fetch_email_by_uid
from imbox.query import build_search_query
import logging
logger = logging.getLogger(__name__)
__version_info__ = (0, 9, 5)
@@ -36,29 +37,6 @@ class Imbox:
logger.info("Disconnected from IMAP Server {username}@{hostname}".format(
hostname=self.hostname, username=self.username))
def query_uids(self, **kwargs):
query = build_search_query(**kwargs)
message, data = self.connection.uid('search', None, query)
if data[0] is None:
return []
return data[0].split()
def fetch_by_uid(self, uid):
message, data = self.connection.uid('fetch', uid, '(BODY.PEEK[])')
logger.debug("Fetched message for UID {}".format(int(uid)))
raw_email = data[0][1]
email_object = parse_email(raw_email, policy=self.parser_policy)
return email_object
def fetch_list(self, **kwargs):
uid_list = self.query_uids(**kwargs)
logger.debug("Fetch all messages for UID in {}".format(uid_list))
for uid in uid_list:
yield (uid, self.fetch_by_uid(uid))
def mark_seen(self, uid):
logger.info("Mark UID {} with \\Seen FLAG".format(int(uid)))
self.connection.uid('STORE', uid, '+FLAGS', '(\\Seen)')
@@ -69,7 +47,6 @@ class Imbox:
def delete(self, uid):
logger.info("Mark UID {} with \\Deleted FLAG and expunge.".format(int(uid)))
mov, data = self.connection.uid('STORE', uid, '+FLAGS', '(\\Deleted)')
self.connection.expunge()
def copy(self, uid, destination_folder):
@@ -81,7 +58,7 @@ class Imbox:
if self.copy(uid, destination_folder):
self.delete(uid)
def messages(self, *args, **kwargs):
def messages(self, **kwargs):
folder = kwargs.get('folder', False)
msg = ""
@@ -90,7 +67,65 @@ class Imbox:
msg = " from folder '{}'".format(folder)
logger.info("Fetch list of messages{}".format(msg))
return self.fetch_list(**kwargs)
return Messages(connection=self.connection,
parser_policy=self.parser_policy,
**kwargs)
def folders(self):
return self.connection.list()
class Messages:
def __init__(self,
connection,
parser_policy,
**kwargs):
self.connection = connection
self.parser_policy = parser_policy
self.kwargs = kwargs
self._uid_list = self._query_uids(**kwargs)
logger.debug("Fetch all messages for UID in {}".format(self._uid_list))
def _fetch_email(self, uid):
return fetch_email_by_uid(uid=uid,
connection=self.connection,
parser_policy=self.parser_policy)
def _query_uids(self, **kwargs):
query_ = build_search_query(**kwargs)
message, data = self.connection.uid('search', None, query_)
if data[0] is None:
return []
return data[0].split()
def _fetch_email_list(self):
for uid in self._uid_list:
yield uid, self._fetch_email(uid)
def __repr__(self):
if len(self.kwargs) > 0:
return 'Messages({})'.format('\n'.join('{}={}'.format(key, value)
for key, value in self.kwargs.items()))
return 'Messages(ALL)'
def __iter__(self):
return self._fetch_email_list()
def __next__(self):
return self
def __len__(self):
return len(self._uid_list)
def __getitem__(self, index):
uids = self._uid_list[index]
if not isinstance(uids, list):
uid = uids
return uid, self._fetch_email(uid)
return [(uid, self._fetch_email(uid))
for uid in uids]

14
imbox/imap.py
View File

@@ -10,22 +10,16 @@ class ImapTransport:
def __init__(self, hostname, port=None, ssl=True, ssl_context=None, starttls=False):
self.hostname = hostname
self.port = port
kwargs = {}
if ssl:
self.transport = IMAP4_SSL
if not self.port:
self.port = 993
self.port = port or 993
if ssl_context is None:
ssl_context = pythonssllib.create_default_context()
kwargs["ssl_context"] = ssl_context
self.server = IMAP4_SSL(self.hostname, self.port, ssl_context=ssl_context)
else:
self.transport = IMAP4
if not self.port:
self.port = 143
self.port = port or 143
self.server = IMAP4(self.hostname, self.port)
self.server = self.transport(self.hostname, self.port, **kwargs)
if starttls:
self.server.starttls()
logger.debug("Created IMAP4 transport for {host}:{port}"

19
imbox/parser.py
View File

@@ -9,6 +9,7 @@ from email.header import decode_header
from imbox.utils import str_encode, str_decode
import logging
logger = logging.getLogger(__name__)
@@ -66,7 +67,7 @@ def decode_param(param):
values = v.split('\n')
value_results = []
for value in values:
match = re.search(r'=\?((?:\w|-)+)\?(Q|B)\?(.+)\?=', value)
match = re.search(r'=\?((?:\w|-)+)\?([QB])\?(.+)\?=', value)
if match:
encoding, type_, code = match.groups()
if type_ == 'Q':
@@ -127,6 +128,16 @@ def decode_content(message):
return content
def fetch_email_by_uid(uid, connection, parser_policy):
message, data = connection.uid('fetch', uid, '(BODY.PEEK[])')
logger.debug("Fetched message for UID {}".format(int(uid)))
raw_email = data[0][1]
email_object = parse_email(raw_email, policy=parser_policy)
return email_object
def parse_email(raw_email, policy=None):
if isinstance(raw_email, bytes):
raw_email = str_encode(raw_email, 'utf-8', errors='ignore')
@@ -140,9 +151,7 @@ def parse_email(raw_email, policy=None):
except UnicodeEncodeError:
email_message = email.message_from_string(raw_email.encode('utf-8'), **email_parse_kwargs)
maintype = email_message.get_content_maintype()
parsed_email = {}
parsed_email['raw_email'] = raw_email
parsed_email = {'raw_email': raw_email}
body = {
"plain": [],
@@ -162,7 +171,7 @@ def parse_email(raw_email, policy=None):
content = decode_content(part)
is_inline = content_disposition is None \
or content_disposition.startswith("inline")
or content_disposition.startswith("inline")
if content_type == "text/plain" and is_inline:
body['plain'].append(content)
elif content_type == "text/html" and is_inline:

3
imbox/query.py
View File

@@ -8,8 +8,7 @@ logger = logging.getLogger(__name__)
def format_date(date):
if isinstance(date, datetime.date):
return date.strftime('%d-%b-%Y')
else:
return date
return date
def build_search_query(**kwargs):

4
imbox/utils.py
View File

@@ -1,14 +1,16 @@
import logging
logger = logging.getLogger(__name__)
def str_encode(value='', encoding=None, errors='strict'):
logger.debug("Encode str {} with and errors {}".format(value, encoding, errors))
return str(value, encoding, errors)
def str_decode(value='', encoding=None, errors='strict'):
if isinstance(value, str):
return bytes(value, encoding, errors).decode('utf-8')
elif isinstance(value, bytes):
return value.decode(encoding or 'utf-8', errors=errors)
else:
raise TypeError( "Cannot decode '{}' object".format(value.__class__) )
raise TypeError("Cannot decode '{}' object".format(value.__class__))