email_fetcher.DecodedMail.get_key() - Code Metrics - tracim/tracim_v2 - Measure and Improve Code Quality continuously with Scrutinizer

email_fetcher.DecodedMail.get_key() A
last analyzed 2018-11-14 14:00 UTC

↳ Parent: email_fetcher

Complexity

Conditions

Size

Total Lines	19
Code Lines	11

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	11
dl	0
loc	19
rs	9.85
c	0
b	0
f	0
cc	4
nop	1

# -*- coding: utf-8 -*-

import json
import socket
import ssl
import time
import typing
from email import message_from_bytes
from email.header import decode_header
from email.header import make_header
from email.message import Message
from email.utils import parseaddr

import filelock
import imapclient
import markdown
import requests
from email_reply_parser import EmailReplyParser

from tracim_backend.exceptions import BadStatusCode
from tracim_backend.exceptions import EmptyEmailBody
from tracim_backend.exceptions import NoSpecialKeyFound
from tracim_backend.exceptions import UnsupportedRequestMethod
from tracim_backend.lib.mail_fetcher.email_processing.parser import ParsedHTMLMail  # nopep8
from tracim_backend.lib.mail_fetcher.email_processing.sanitizer import HtmlSanitizer  # nopep8
from tracim_backend.lib.utils.authentification import TRACIM_API_KEY_HEADER
from tracim_backend.lib.utils.authentification import TRACIM_API_USER_EMAIL_LOGIN_HEADER  # nopep8
from tracim_backend.lib.utils.logger import logger

TRACIM_SPECIAL_KEY_HEADER = 'X-Tracim-Key'
CONTENT_TYPE_TEXT_PLAIN = 'text/plain'
CONTENT_TYPE_TEXT_HTML = 'text/html'

IMAP_CHECKED_FLAG = imapclient.FLAGGED
IMAP_SEEN_FLAG = imapclient.SEEN

MAIL_FETCHER_FILELOCK_TIMEOUT = 10
MAIL_FETCHER_CONNECTION_TIMEOUT = 60*3
MAIL_FETCHER_IDLE_RESPONSE_TIMEOUT = 60*9   # this should be not more
# that 29 minutes according to rfc2177.(server wait 30min by default)


class MessageContainer(object):
    def __init__(self, message: Message, uid: int) -> None:
        self.message = message
        self.uid = uid


class DecodedMail(object):
    def __init__(self, message: Message, uid: int=None) -> None:
        self._message = message
        self.uid = uid

    def _decode_header(self, header_title: str) -> typing.Optional[str]:
        # FIXME : Handle exception
        if header_title in self._message:
            return str(make_header(decode_header(self._message[header_title])))
        else:
            return None

    def get_subject(self) -> typing.Optional[str]:
        return self._decode_header('subject')

    def get_from_address(self) -> str:
        return parseaddr(self._message['From'])[1]

    def get_to_address(self) -> str:
        return parseaddr(self._message['To'])[1]

    def get_first_ref(self) -> str:
        return parseaddr(self._message['References'])[1]

    def get_special_key(self) -> typing.Optional[str]:
        return self._decode_header(TRACIM_SPECIAL_KEY_HEADER)

    def get_body(
            self,
            use_html_parsing=True,
            use_txt_parsing=True,
    ) -> typing.Optional[str]:
        body_part = self._get_mime_body_message()
        body = None
        if body_part:
            charset = body_part.get_content_charset('iso-8859-1')
            content_type = body_part.get_content_type()
            if content_type == CONTENT_TYPE_TEXT_PLAIN:
                txt_body = body_part.get_payload(decode=True).decode(
                    charset)
                if use_txt_parsing:
                    txt_body = EmailReplyParser.parse_reply(txt_body)
                html_body = markdown.markdown(txt_body)
                body = HtmlSanitizer.sanitize(html_body)

            elif content_type == CONTENT_TYPE_TEXT_HTML:
                html_body = body_part.get_payload(decode=True).decode(
                    charset)
                if use_html_parsing:
                    html_body = str(ParsedHTMLMail(html_body))
                body = HtmlSanitizer.sanitize(html_body)
            if not body:
                raise EmptyEmailBody()
        return body

    def _get_mime_body_message(self) -> typing.Optional[Message]:
        # TODO - G.M - 2017-11-16 - Use stdlib msg.get_body feature for py3.6+
        part = None
        # Check for html
        for part in self._message.walk():
            content_type = part.get_content_type()
            content_dispo = str(part.get('Content-Disposition'))
            if content_type == CONTENT_TYPE_TEXT_HTML \
                    and 'attachment' not in content_dispo:
                return part
        # check for plain text
        for part in self._message.walk():
            content_type = part.get_content_type()
            content_dispo = str(part.get('Content-Disposition'))
            if content_type == CONTENT_TYPE_TEXT_PLAIN \
                    and 'attachment' not in content_dispo:
                return part
        return part

    def get_key(self) -> typing.Optional[str]:

        """
        key is the string contain in some mail header we need to retrieve.
        First try checking special header, them check 'to' header
        and finally check first(oldest) mail-id of 'references' header
        """
        first_ref = self.get_first_ref()
        to_address = self.get_to_address()
        special_key = self.get_special_key()

        if special_key:
            return special_key
        if to_address:
            return DecodedMail.find_key_from_mail_address(to_address)
        if first_ref:
            return DecodedMail.find_key_from_mail_address(first_ref)

        raise NoSpecialKeyFound()

    @classmethod
    def find_key_from_mail_address(
        cls,
        mail_address: str,
    ) -> typing.Optional[str]:
        """ Parse mail_adress-like string
        to retrieve key.

        :param mail_address: user+key@something like string
        :return: key
        """
        username = mail_address.split('@')[0]
        username_data = username.split('+')
        if len(username_data) == 2:
            return username_data[1]
        return None


class BadIMAPFetchResponse(Exception):
    pass


class MailFetcher(object):
    def __init__(
        self,
        host: str,
        port: str,
        user: str,
        password: str,
        use_ssl: bool,
        folder: str,
        use_idle: bool,
        connection_max_lifetime: int,
        heartbeat: int,
        api_base_url: str,
        api_key: str,
        use_html_parsing: bool,
        use_txt_parsing: bool,
        lockfile_path: str,
        burst: bool,
    ) -> None:
        """
        Fetch mail from a mailbox folder through IMAP and add their content to
        Tracim through http according to mail Headers.
        Fetch is regular.
        :param host: imap server hostname
        :param port: imap connection port
        :param user: user login of mailbox
        :param password: user password of mailbox
        :param use_ssl: use imap over ssl connection
        :param folder: mail folder where new mail are fetched
        :param use_idle: use IMAP IDLE(server notification) when available
        :param heartbeat: seconds to wait before fetching new mail again
        :param connection_max_lifetime: maximum duration allowed for a
             connection . connection are automatically renew when their
             lifetime excess this duration.
        :param api_base_url: url to get access to tracim api
        :param api_key: tracim api key
        :param use_html_parsing: parse html mail
        :param use_txt_parsing: parse txt mail
        :param burst: if true, run only one time,
        if false run as continous daemon.
        """
        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.use_ssl = use_ssl
        self.folder = folder
        self.heartbeat = heartbeat
        self.use_idle = use_idle
        self.connection_max_lifetime = connection_max_lifetime
        self.api_base_url = api_base_url
        self.api_key = api_key
        self.use_html_parsing = use_html_parsing
        self.use_txt_parsing = use_txt_parsing
        self.lock = filelock.FileLock(lockfile_path)
        self._is_active = True
        self.burst = burst

    def run(self) -> None:
        logger.info(self, 'Starting MailFetcher')
        while self._is_active:
            imapc = None
            sleep_after_connection = True
            try:
                imapc = imapclient.IMAPClient(
                    self.host,
                    self.port,
                    ssl=self.use_ssl,
                    timeout=MAIL_FETCHER_CONNECTION_TIMEOUT
                )
                imapc.login(self.user, self.password)

                logger.debug(self, 'Select folder {}'.format(
                    self.folder,
                ))
                imapc.select_folder(self.folder)

                # force renew connection when deadline is reached
                deadline = time.time() + self.connection_max_lifetime
                while True:
                    if not self._is_active:
                        logger.warning(self, 'Mail Fetcher process aborted')
                        sleep_after_connection = False
                        break

                    if time.time() > deadline:
                        logger.debug(
                            self,
                            "MailFetcher Connection Lifetime limit excess"
                            ", Try Re-new connection")
                        sleep_after_connection = False
                        break

                    # check for new mails
                    self._check_mail(imapc)

                    if self.use_idle and imapc.has_capability('IDLE'):
                        # IDLE_mode wait until event from server
                        logger.debug(self, 'wail for event(IDLE)')
                        imapc.idle()
                        imapc.idle_check(
                            timeout=MAIL_FETCHER_IDLE_RESPONSE_TIMEOUT
                        )
                        imapc.idle_done()
                    else:
                        if self.use_idle and not imapc.has_capability('IDLE'):
                            log = 'IDLE mode activated but server do not' \
                                  'support it, use polling instead.'
                            logger.warning(self, log)

                        if self.burst:
                            self.stop()
                            break
                        # normal polling mode : sleep a define duration
                        logger.debug(self,
                                     'sleep for {}'.format(self.heartbeat))
                        time.sleep(self.heartbeat)

                    if self.burst:
                        self.stop()
                        break
            # Socket
            except (socket.error,
                    socket.gaierror,
                    socket.herror) as e:
                log = 'Socket fail with IMAP connection {}'
                logger.error(self, log.format(e.__str__()))

            except socket.timeout as e:
                log = 'Socket timeout on IMAP connection {}'
                logger.error(self, log.format(e.__str__()))

            # SSL
            except ssl.SSLError as e:
                log = 'SSL error on IMAP connection'
                logger.error(self, log.format(e.__str__()))

            except ssl.CertificateError as e:
                log = 'SSL Certificate verification failed on IMAP connection'
                logger.error(self, log.format(e.__str__()))

            # Filelock
            except filelock.Timeout as e:
                log = 'Mail Fetcher Lock Timeout {}'
                logger.warning(self, log.format(e.__str__()))

            # IMAP
            # TODO - G.M - 10-01-2017 - Support imapclient exceptions
            # when Imapclient stable will be 2.0+

            except BadIMAPFetchResponse as e:
                log = 'Imap Fetch command return bad response.' \
                      'Is someone else connected to the mailbox ?: ' \
                      '{}'
                logger.error(self, log.format(e.__str__()))
            # Others
            except Exception as e:
                log = 'Mail Fetcher error {}'
                logger.error(self, log.format(e.__str__()))

            finally:
                # INFO - G.M - 2018-01-09 - Connection closing
                # Properly close connection according to
                # https://github.com/mjs/imapclient/pull/279/commits/043e4bd0c5c775c5a08cb5f1baa93876a46732ee
                # TODO : Use __exit__ method instead when imapclient stable will
                # be 2.0+ .
                if imapc:
                    logger.debug(self, 'Try logout')
                    try:
                        imapc.logout()
                    except Exception:
                        try:
                            imapc.shutdown()
                        except Exception as e:
                            log = "Can't logout, connection broken ? {}"
                            logger.error(self, log.format(e.__str__()))

            if self.burst:
                self.stop()
                break

            if sleep_after_connection:
                logger.debug(self, 'sleep for {}'.format(self.heartbeat))
                time.sleep(self.heartbeat)

        log = 'Mail Fetcher stopped'
        logger.debug(self, log)

    def _check_mail(self, imapc: imapclient.IMAPClient) -> None:
        with self.lock.acquire(
                timeout=MAIL_FETCHER_FILELOCK_TIMEOUT
        ):
            messages = self._fetch(imapc)
            cleaned_mails = [DecodedMail(m.message, m.uid)
                             for m in messages]
            self._notify_tracim(cleaned_mails, imapc)

    def stop(self) -> None:
        self._is_active = False

    def _fetch(
        self, 
        imapc: imapclient.IMAPClient,
    ) -> typing.List[MessageContainer]:
        """
        Get news message from mailbox
        :return: list of new mails
        """
        messages = []

        logger.debug(self, 'Fetch unflagged messages')
        uids = imapc.search(['UNFLAGGED'])
        logger.debug(self, 'Found {} unflagged mails'.format(
            len(uids),
        ))
        for msgid, data in imapc.fetch(uids, ['BODY.PEEK[]']).items():
            # INFO - G.M - 2017-12-08 - Fetch BODY.PEEK[]
            # Retrieve all mail(body and header) but don't set mail
            # as seen because of PEEK
            # see rfc3501
            logger.debug(self, 'Fetch mail "{}"'.format(
                msgid,
            ))

            try:
                msg = message_from_bytes(data[b'BODY[]'])
            except KeyError as e:
                # INFO - G.M - 12-01-2018 - Fetch may return events response
                # In some specific case, fetch command may return events
                # response unrelated to fetch request.
                # This should happen only when someone-else use the mailbox
                # at the same time of the fetcher.
                # see https://github.com/mjs/imapclient/issues/334
                except_msg = 'fetch response : {}'.format(str(data))
                raise BadIMAPFetchResponse(except_msg) from e

            msg_container = MessageContainer(msg, msgid)
            messages.append(msg_container)

        return messages

    def _notify_tracim(
        self,
        mails: typing.List[DecodedMail],
        imapc: imapclient.IMAPClient
    ) -> None:
        """
        Send http request to tracim endpoint
        :param mails: list of mails to send
        :return: none
        """
        logger.debug(self, 'Notify tracim about {} new responses'.format(
            len(mails),
        ))
        # TODO BS 20171124: Look around mail.get_from_address(), mail.get_key()
        # , mail.get_body() etc ... for raise InvalidEmailError if missing
        #  required informations (actually get_from_address raise IndexError
        #  if no from address for example) and catch it here
        while mails:
            mail = mails.pop()
            try:
                method, endpoint, json_body_dict = self._create_comment_request(mail)  # nopep8
            except NoSpecialKeyFound as exc:
                log = 'Failed to create comment request due to missing specialkey in mail {}'  # nopep8
                logger.error(self, log.format(exc.__str__()))
                continue
            except EmptyEmailBody as exc:
                log = 'Empty body, skip mail'
                logger.error(self, log)
                continue
            except Exception as exc:
                log = 'Failed to create comment request in mail fetcher error {}'  # nopep8
                logger.error(self, log.format(exc.__str__()))
                continue

            try:
                self._send_request(
                    mail=mail,
                    imapc=imapc,
                    method=method,
                    endpoint=endpoint,
                    json_body_dict=json_body_dict,
                )
            except requests.exceptions.Timeout as e:
                log = 'Timeout error to transmit fetched mail to tracim : {}'
                logger.error(self, log.format(str(e)))
            except requests.exceptions.RequestException as e:
                log = 'Fail to transmit fetched mail to tracim : {}'
                logger.error(self, log.format(str(e)))

    def _get_auth_headers(self, user_email) -> dict:
        return {
            TRACIM_API_KEY_HEADER: self.api_key,
            TRACIM_API_USER_EMAIL_LOGIN_HEADER: user_email
        }

    def _get_content_info(self, content_id, user_email):
        endpoint = '{api_base_url}contents/{content_id}'.format(
            api_base_url=self.api_base_url,
            content_id=content_id,
        )
        result = requests.get(
            endpoint,
            headers=self._get_auth_headers(user_email)
        )
        if result.status_code not in [200, 204]:
            details = result.json().get('message')
            msg = 'bad status code {}(200 is valid) response when trying to get info about a content: {}'  # nopep8
            msg = msg.format(str(result.status_code), details)
            raise BadStatusCode(msg)
        return result.json()

    def _create_comment_request(self, mail: DecodedMail) -> typing.Tuple[str, str, dict]:  # nopep8
        content_id = mail.get_key()
        content_info = self._get_content_info(content_id, mail.get_from_address())  # nopep8
        mail_body = mail.get_body(
            use_html_parsing=self.use_html_parsing,
            use_txt_parsing=self.use_txt_parsing,
        )
        endpoint = '{api_base_url}workspaces/{workspace_id}/contents/{content_id}/comments'.format(  # nopep8
            api_base_url=self.api_base_url,
            content_id=content_id,
            workspace_id=content_info['workspace_id']
        )
        method = 'POST'
        body = {
            'raw_content': mail_body
        }
        return method, endpoint, body

    def _send_request(
            self,
            mail: DecodedMail,
            imapc: imapclient.IMAPClient,
            method: str,
            endpoint: str,
            json_body_dict: dict
    ):
        logger.debug(
            self,
            'Contact API on {endpoint} with method {method} with body {body}'.format(   # nopep8
                endpoint=endpoint,
                method=method,
                body=str(json_body_dict),
            ),
        )
        if method == 'POST':
            request_method = requests.post
        else:
            # TODO - G.M - 2018-08-24 - Better handling exception
            raise UnsupportedRequestMethod('Request method not supported')

        r = request_method(
            url=endpoint,
            json=json_body_dict,
            headers=self._get_auth_headers(mail.get_from_address()),
        )
        if r.status_code not in [200, 204]:
            details = r.json().get('message')
            msg = 'bad status code {} (200 and 204 are valid) response when sending mail to tracim: {}'  # nopep8
            msg = msg.format(str(r.status_code), details)
            raise BadStatusCode(msg)
        # Flag all correctly checked mail
        if r.status_code in [200, 204]:
            imapc.add_flags((mail.uid,), IMAP_CHECKED_FLAG)
            imapc.add_flags((mail.uid,), IMAP_SEEN_FLAG)


1			# -- coding: utf-8 --
2
3			import json
4			import socket
5			import ssl
6			import time
7			import typing
8			from email import message_from_bytes
9			from email.header import decode_header
10			from email.header import make_header
11			from email.message import Message
12			from email.utils import parseaddr
13
14			import filelock
15			import imapclient
16			import markdown
17			import requests
18			from email_reply_parser import EmailReplyParser
19
20			from tracim_backend.exceptions import BadStatusCode
21			from tracim_backend.exceptions import EmptyEmailBody
22			from tracim_backend.exceptions import NoSpecialKeyFound
23			from tracim_backend.exceptions import UnsupportedRequestMethod
24			from tracim_backend.lib.mail_fetcher.email_processing.parser import ParsedHTMLMail # nopep8
25			from tracim_backend.lib.mail_fetcher.email_processing.sanitizer import HtmlSanitizer # nopep8
26			from tracim_backend.lib.utils.authentification import TRACIM_API_KEY_HEADER
27			from tracim_backend.lib.utils.authentification import TRACIM_API_USER_EMAIL_LOGIN_HEADER # nopep8
28			from tracim_backend.lib.utils.logger import logger
29
30			TRACIM_SPECIAL_KEY_HEADER = 'X-Tracim-Key'
31			CONTENT_TYPE_TEXT_PLAIN = 'text/plain'
32			CONTENT_TYPE_TEXT_HTML = 'text/html'
33
34			IMAP_CHECKED_FLAG = imapclient.FLAGGED
35			IMAP_SEEN_FLAG = imapclient.SEEN
36
37			MAIL_FETCHER_FILELOCK_TIMEOUT = 10
38			MAIL_FETCHER_CONNECTION_TIMEOUT = 60*3
39			MAIL_FETCHER_IDLE_RESPONSE_TIMEOUT = 60*9 # this should be not more
40			# that 29 minutes according to rfc2177.(server wait 30min by default)
41
42
43			class MessageContainer(object):
44			def __init__(self, message: Message, uid: int) -> None:
45			self.message = message
46			self.uid = uid
47
48
49			class DecodedMail(object):
50			def __init__(self, message: Message, uid: int=None) -> None:
51			self._message = message
52			self.uid = uid
53
54			def _decode_header(self, header_title: str) -> typing.Optional[str]:
55			# FIXME : Handle exception
56			if header_title in self._message:
57			return str(make_header(decode_header(self._message[header_title])))
58			else:
59			return None
60
61			def get_subject(self) -> typing.Optional[str]:
62			return self._decode_header('subject')
63
64			def get_from_address(self) -> str:
65			return parseaddr(self._message['From'])[1]
66
67			def get_to_address(self) -> str:
68			return parseaddr(self._message['To'])[1]
69
70			def get_first_ref(self) -> str:
71			return parseaddr(self._message['References'])[1]
72
73			def get_special_key(self) -> typing.Optional[str]:
74			return self._decode_header(TRACIM_SPECIAL_KEY_HEADER)
75
76			def get_body(
77			self,
78			use_html_parsing=True,
79			use_txt_parsing=True,
80			) -> typing.Optional[str]:
81			body_part = self._get_mime_body_message()
82			body = None
83			if body_part:
84			charset = body_part.get_content_charset('iso-8859-1')
85			content_type = body_part.get_content_type()
86			if content_type == CONTENT_TYPE_TEXT_PLAIN:
87			txt_body = body_part.get_payload(decode=True).decode(
88			charset)
89			if use_txt_parsing:
90			txt_body = EmailReplyParser.parse_reply(txt_body)
91			html_body = markdown.markdown(txt_body)
92			body = HtmlSanitizer.sanitize(html_body)
93
94			elif content_type == CONTENT_TYPE_TEXT_HTML:
95			html_body = body_part.get_payload(decode=True).decode(
96			charset)
97			if use_html_parsing:
98			html_body = str(ParsedHTMLMail(html_body))
99			body = HtmlSanitizer.sanitize(html_body)
100			if not body:
101			raise EmptyEmailBody()
102			return body
103
104			def _get_mime_body_message(self) -> typing.Optional[Message]:
105			# TODO - G.M - 2017-11-16 - Use stdlib msg.get_body feature for py3.6+
106			part = None
107			# Check for html
108			for part in self._message.walk():
109			content_type = part.get_content_type()
110			content_dispo = str(part.get('Content-Disposition'))
111			if content_type == CONTENT_TYPE_TEXT_HTML \
112			and 'attachment' not in content_dispo:
113			return part
114			# check for plain text
115			for part in self._message.walk():
116			content_type = part.get_content_type()
117			content_dispo = str(part.get('Content-Disposition'))
118			if content_type == CONTENT_TYPE_TEXT_PLAIN \
119			and 'attachment' not in content_dispo:
120			return part
121			return part
122
123			def get_key(self) -> typing.Optional[str]:
124
125			"""
126			key is the string contain in some mail header we need to retrieve.
127			First try checking special header, them check 'to' header
128			and finally check first(oldest) mail-id of 'references' header
129			"""
130			first_ref = self.get_first_ref()
131			to_address = self.get_to_address()
132			special_key = self.get_special_key()
133
134			if special_key:
135			return special_key
136			if to_address:
137			return DecodedMail.find_key_from_mail_address(to_address)
138			if first_ref:
139			return DecodedMail.find_key_from_mail_address(first_ref)
140
141			raise NoSpecialKeyFound()
142
143			@classmethod
144			def find_key_from_mail_address(
145			cls,
146			mail_address: str,
147			) -> typing.Optional[str]:
148			""" Parse mail_adress-like string
149			to retrieve key.
150
151			:param mail_address: user+key@something like string
152			:return: key
153			"""
154			username = mail_address.split('@')[0]
155			username_data = username.split('+')
156			if len(username_data) == 2:
157			return username_data[1]
158			return None
159
160
161			class BadIMAPFetchResponse(Exception):
162			pass
163
164
165			class MailFetcher(object):
166			def __init__(
167			self,
168			host: str,
169			port: str,
170			user: str,
171			password: str,
172			use_ssl: bool,
173			folder: str,
174			use_idle: bool,
175			connection_max_lifetime: int,
176			heartbeat: int,
177			api_base_url: str,
178			api_key: str,
179			use_html_parsing: bool,
180			use_txt_parsing: bool,
181			lockfile_path: str,
182			burst: bool,
183			) -> None:
184			"""
185			Fetch mail from a mailbox folder through IMAP and add their content to
186			Tracim through http according to mail Headers.
187			Fetch is regular.
188			:param host: imap server hostname
189			:param port: imap connection port
190			:param user: user login of mailbox
191			:param password: user password of mailbox
192			:param use_ssl: use imap over ssl connection
193			:param folder: mail folder where new mail are fetched
194			:param use_idle: use IMAP IDLE(server notification) when available
195			:param heartbeat: seconds to wait before fetching new mail again
196			:param connection_max_lifetime: maximum duration allowed for a
197			connection . connection are automatically renew when their
198			lifetime excess this duration.
199			:param api_base_url: url to get access to tracim api
200			:param api_key: tracim api key
201			:param use_html_parsing: parse html mail
202			:param use_txt_parsing: parse txt mail
203			:param burst: if true, run only one time,
204			if false run as continous daemon.
205			"""
206			self.host = host
207			self.port = port
208			self.user = user
209			self.password = password
210			self.use_ssl = use_ssl
211			self.folder = folder
212			self.heartbeat = heartbeat
213			self.use_idle = use_idle
214			self.connection_max_lifetime = connection_max_lifetime
215			self.api_base_url = api_base_url
216			self.api_key = api_key
217			self.use_html_parsing = use_html_parsing
218			self.use_txt_parsing = use_txt_parsing
219			self.lock = filelock.FileLock(lockfile_path)
220			self._is_active = True
221			self.burst = burst
222
223			def run(self) -> None:
224			logger.info(self, 'Starting MailFetcher')
225			while self._is_active:
226			imapc = None
227			sleep_after_connection = True
228			try:
229			imapc = imapclient.IMAPClient(
230			self.host,
231			self.port,
232			ssl=self.use_ssl,
233			timeout=MAIL_FETCHER_CONNECTION_TIMEOUT
234			)
235			imapc.login(self.user, self.password)
236
237			logger.debug(self, 'Select folder {}'.format(
238			self.folder,
239			))
240			imapc.select_folder(self.folder)
241
242			# force renew connection when deadline is reached
243			deadline = time.time() + self.connection_max_lifetime
244			while True:
245			if not self._is_active:
246			logger.warning(self, 'Mail Fetcher process aborted')
247			sleep_after_connection = False
248			break
249
250			if time.time() > deadline:
251			logger.debug(
252			self,
253			"MailFetcher Connection Lifetime limit excess"
254			", Try Re-new connection")
255			sleep_after_connection = False
256			break
257
258			# check for new mails
259			self._check_mail(imapc)
260
261			if self.use_idle and imapc.has_capability('IDLE'):
262			# IDLE_mode wait until event from server
263			logger.debug(self, 'wail for event(IDLE)')
264			imapc.idle()
265			imapc.idle_check(
266			timeout=MAIL_FETCHER_IDLE_RESPONSE_TIMEOUT
267			)
268			imapc.idle_done()
269			else:
270			if self.use_idle and not imapc.has_capability('IDLE'):
271			log = 'IDLE mode activated but server do not' \
272			'support it, use polling instead.'
273			logger.warning(self, log)
274
275			if self.burst:
276			self.stop()
277			break
278			# normal polling mode : sleep a define duration
279			logger.debug(self,
280			'sleep for {}'.format(self.heartbeat))
281			time.sleep(self.heartbeat)
282
283			if self.burst:
284			self.stop()
285			break
286			# Socket
287			except (socket.error,
288			socket.gaierror,
289			socket.herror) as e:
290			log = 'Socket fail with IMAP connection {}'
291			logger.error(self, log.format(e.__str__()))
292
293			except socket.timeout as e:
294			log = 'Socket timeout on IMAP connection {}'
295			logger.error(self, log.format(e.__str__()))
296
297			# SSL
298			except ssl.SSLError as e:
299			log = 'SSL error on IMAP connection'
300			logger.error(self, log.format(e.__str__()))
301
302			except ssl.CertificateError as e:
303			log = 'SSL Certificate verification failed on IMAP connection'
304			logger.error(self, log.format(e.__str__()))
305
306			# Filelock
307			except filelock.Timeout as e:
308			log = 'Mail Fetcher Lock Timeout {}'
309			logger.warning(self, log.format(e.__str__()))
310
311			# IMAP
312			# TODO - G.M - 10-01-2017 - Support imapclient exceptions
313			# when Imapclient stable will be 2.0+
314
315			except BadIMAPFetchResponse as e:
316			log = 'Imap Fetch command return bad response.' \
317			'Is someone else connected to the mailbox ?: ' \
318			'{}'
319			logger.error(self, log.format(e.__str__()))
320			# Others
321			except Exception as e:
322			log = 'Mail Fetcher error {}'
323			logger.error(self, log.format(e.__str__()))
324
325			finally:
326			# INFO - G.M - 2018-01-09 - Connection closing
327			# Properly close connection according to
328			# https://github.com/mjs/imapclient/pull/279/commits/043e4bd0c5c775c5a08cb5f1baa93876a46732ee
329			# TODO : Use __exit__ method instead when imapclient stable will
330			# be 2.0+ .
331			if imapc:
332			logger.debug(self, 'Try logout')
333			try:
334			imapc.logout()
335			except Exception:
336			try:
337			imapc.shutdown()
338			except Exception as e:
339			log = "Can't logout, connection broken ? {}"
340			logger.error(self, log.format(e.__str__()))
341
342			if self.burst:
343			self.stop()
344			break
345
346			if sleep_after_connection:
347			logger.debug(self, 'sleep for {}'.format(self.heartbeat))
348			time.sleep(self.heartbeat)
349
350			log = 'Mail Fetcher stopped'
351			logger.debug(self, log)
352
353			def _check_mail(self, imapc: imapclient.IMAPClient) -> None:
354			with self.lock.acquire(
355			timeout=MAIL_FETCHER_FILELOCK_TIMEOUT
356			):
357			messages = self._fetch(imapc)
358			cleaned_mails = [DecodedMail(m.message, m.uid)
359			for m in messages]
360			self._notify_tracim(cleaned_mails, imapc)
361
362			def stop(self) -> None:
363			self._is_active = False
364
365			def _fetch(
366			self,
367			imapc: imapclient.IMAPClient,
368			) -> typing.List[MessageContainer]:
369			"""
370			Get news message from mailbox
371			:return: list of new mails
372			"""
373			messages = []
374
375			logger.debug(self, 'Fetch unflagged messages')
376			uids = imapc.search(['UNFLAGGED'])
377			logger.debug(self, 'Found {} unflagged mails'.format(
378			len(uids),
379			))
380			for msgid, data in imapc.fetch(uids, ['BODY.PEEK[]']).items():
381			# INFO - G.M - 2017-12-08 - Fetch BODY.PEEK[]
382			# Retrieve all mail(body and header) but don't set mail
383			# as seen because of PEEK
384			# see rfc3501
385			logger.debug(self, 'Fetch mail "{}"'.format(
386			msgid,
387			))
388
389			try:
390			msg = message_from_bytes(data[b'BODY[]'])
391			except KeyError as e:
392			# INFO - G.M - 12-01-2018 - Fetch may return events response
393			# In some specific case, fetch command may return events
394			# response unrelated to fetch request.
395			# This should happen only when someone-else use the mailbox
396			# at the same time of the fetcher.
397			# see https://github.com/mjs/imapclient/issues/334
398			except_msg = 'fetch response : {}'.format(str(data))
399			raise BadIMAPFetchResponse(except_msg) from e
400
401			msg_container = MessageContainer(msg, msgid)
402			messages.append(msg_container)
403
404			return messages
405
406			def _notify_tracim(
407			self,
408			mails: typing.List[DecodedMail],
409			imapc: imapclient.IMAPClient
410			) -> None:
411			"""
412			Send http request to tracim endpoint
413			:param mails: list of mails to send
414			:return: none
415			"""
416			logger.debug(self, 'Notify tracim about {} new responses'.format(
417			len(mails),
418			))
419			# TODO BS 20171124: Look around mail.get_from_address(), mail.get_key()
420			# , mail.get_body() etc ... for raise InvalidEmailError if missing
421			# required informations (actually get_from_address raise IndexError
422			# if no from address for example) and catch it here
423			while mails:
424			mail = mails.pop()
425			try:
426			method, endpoint, json_body_dict = self._create_comment_request(mail) # nopep8
427			except NoSpecialKeyFound as exc:
428			log = 'Failed to create comment request due to missing specialkey in mail {}' # nopep8
429			logger.error(self, log.format(exc.__str__()))
430			continue
431			except EmptyEmailBody as exc:
432			log = 'Empty body, skip mail'
433			logger.error(self, log)
434			continue
435			except Exception as exc:
436			log = 'Failed to create comment request in mail fetcher error {}' # nopep8
437			logger.error(self, log.format(exc.__str__()))
438			continue
439
440			try:
441			self._send_request(
442			mail=mail,
443			imapc=imapc,
444			method=method,
445			endpoint=endpoint,
446			json_body_dict=json_body_dict,
447			)
448			except requests.exceptions.Timeout as e:
449			log = 'Timeout error to transmit fetched mail to tracim : {}'
450			logger.error(self, log.format(str(e)))
451			except requests.exceptions.RequestException as e:
452			log = 'Fail to transmit fetched mail to tracim : {}'
453			logger.error(self, log.format(str(e)))
454
455			def _get_auth_headers(self, user_email) -> dict:
456			return {
457			TRACIM_API_KEY_HEADER: self.api_key,
458			TRACIM_API_USER_EMAIL_LOGIN_HEADER: user_email
459			}
460
461			def _get_content_info(self, content_id, user_email):
462			endpoint = '{api_base_url}contents/{content_id}'.format(
463			api_base_url=self.api_base_url,
464			content_id=content_id,
465			)
466			result = requests.get(
467			endpoint,
468			headers=self._get_auth_headers(user_email)
469			)
470			if result.status_code not in [200, 204]:
471			details = result.json().get('message')
472			msg = 'bad status code {}(200 is valid) response when trying to get info about a content: {}' # nopep8
473			msg = msg.format(str(result.status_code), details)
474			raise BadStatusCode(msg)
475			return result.json()
476
477			def _create_comment_request(self, mail: DecodedMail) -> typing.Tuple[str, str, dict]: # nopep8
478			content_id = mail.get_key()
479			content_info = self._get_content_info(content_id, mail.get_from_address()) # nopep8
480			mail_body = mail.get_body(
481			use_html_parsing=self.use_html_parsing,
482			use_txt_parsing=self.use_txt_parsing,
483			)
484			endpoint = '{api_base_url}workspaces/{workspace_id}/contents/{content_id}/comments'.format( # nopep8
485			api_base_url=self.api_base_url,
486			content_id=content_id,
487			workspace_id=content_info['workspace_id']
488			)
489			method = 'POST'
490			body = {
491			'raw_content': mail_body
492			}
493			return method, endpoint, body
494
495			def _send_request(
496			self,
497			mail: DecodedMail,
498			imapc: imapclient.IMAPClient,
499			method: str,
500			endpoint: str,
501			json_body_dict: dict
502			):
503			logger.debug(
504			self,
505			'Contact API on {endpoint} with method {method} with body {body}'.format( # nopep8
506			endpoint=endpoint,
507			method=method,
508			body=str(json_body_dict),
509			),
510			)
511			if method == 'POST':
512			request_method = requests.post
513			else:
514			# TODO - G.M - 2018-08-24 - Better handling exception
515			raise UnsupportedRequestMethod('Request method not supported')
516
517			r = request_method(
518			url=endpoint,
519			json=json_body_dict,
520			headers=self._get_auth_headers(mail.get_from_address()),
521			)
522			if r.status_code not in [200, 204]:
523			details = r.json().get('message')
524			msg = 'bad status code {} (200 and 204 are valid) response when sending mail to tracim: {}' # nopep8
525			msg = msg.format(str(r.status_code), details)
526			raise BadStatusCode(msg)
527			# Flag all correctly checked mail
528			if r.status_code in [200, 204]:
529			imapc.add_flags((mail.uid,), IMAP_CHECKED_FLAG)
530			imapc.add_flags((mail.uid,), IMAP_SEEN_FLAG)
531

tracim / tracim_v2

email_fetcher.DecodedMail.get_key() A last analyzed 2018-11-14 14:00 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

email_fetcher.DecodedMail.get_key() A
last analyzed 2018-11-14 14:00 UTC