wnd_ils_alarmfax_parser/app/imap_connect.py

from datetime import timedelta, datetime
import os
import ssl
import email
import logging
from imapclient import IMAPClient
from threading import Thread
from securecad_parser import parse_securecad_message
from hooks import webhook, alarminator_api, cups_print
import re

def run():
    threads = {}

    format = "%(asctime)s|%(threadName)s: %(message)s"
    logging.basicConfig(format=format, level=logging.INFO,
                        datefmt="%Y-%m-%d %H:%M:%S")

    def eventHandler(ELEMENT_NAME, uid, message_data, _server: IMAPClient = None):
        email_message = email.message_from_bytes(message_data[b'RFC822'])
        email_from: list[str] = re.findall(r'([\w\.-]+@[\w\.-]+)', email_message.get('From'))
        flags = _server.get_flags(uid)
        logging.info(ELEMENT_NAME + " - get Mail")
        if 'Processed_{}'.format(parser_id).encode() in flags[uid]:
            logging.info("Mail {} bereits verarbeitet.. ignoriere".format(uid))
            if not IS_DEV: 
                return
        else:
           _server.set_flags(uid, ['\\SEEN','Processed_{}'.format(parser_id)])
        logging.info("got Mail {} von {}".format(email_message.get('Subject'), email_from))
        if any(mail in filter_from for mail in email_from):
            # Get HTML body
            html_body = ""
            if email_message.is_multipart():
                for part in email_message.walk():
                    if part.get_content_type() == 'text/html':
                        html_body = part.get_payload(decode=True).decode(part.get_content_charset() or 'utf-8', errors='replace')
                        break
            else:
                if email_message.get_content_type() == 'text/html':
                    html_body = email_message.get_payload(decode=True).decode(email_message.get_content_charset() or 'utf-8', errors='replace')
    
            
            parsed_body = parse_securecad_message(html_body)
            logging.debug(parsed_body)
            if parsed_body != None:
                if 'ALARMDEPESCHE' in parsed_body:
                    logging.info("Alarm für: {}".format(parsed_body['ALARMDEPESCHE']))
                webhook(parsed_body)
                alarminator_api(parsed_body)
                cups_print(parsed_body,html_body)
            pass

    def folder_event_subscriber(folder: str):
        logging.info('folder_event_subscriber startet for Folder: {}'.format(folder))
        with IMAPClient(server, ssl_context=ssl_context) as _server:
            _server.login(username, password)
            while True:
                # filtern des ordners nach mails der letzten 24h, die nicht verarbeitet wurden
                now = datetime.now()
                _server.select_folder(folder, readonly=False)
                q = ['SENTSINCE', now - timedelta(days=1),'NOT','KEYWORD', 'Processed_{}'.format(parser_id)]
                if IS_DEV:
                    q = ['SENTSINCE', now - timedelta(days=1),'UNSEEN']
                    q = ['UNSEEN']
                messages = _server.search(q)
                cnt = messages.__len__()
                if cnt > 0:
                    logging.info("{} Mails nicht verarbeitet in den letzten 2 Tagen in ordner: {}".format(cnt, folder))
                    for uid, message_data in _server.fetch(messages, 'RFC822').items():
                        # IMAPClient ist nicht thread-safe, daher wird hier der _server übergeben und kein Thread verwendet.
                        eventHandler('SearchFolderEvent', uid, message_data, _server) 
                        # t = Thread(target=eventHandler, args=('SearchFolderEvent',uid,message_data,),name="eventHandler: SearchFolderEvent ({})".format(uid))
                        # t.start()
                # aktives warten auf streaming_events. maximal eine minute lang, dann wird nochmal der ordner durchsucht, falls mails angekommen sind während eines timeout/cooldown.
                _server.idle()
                try:
                    logging.debug("Idle check for folder: {}".format(folder))
                    messages = _server.idle_check(timeout=60)  # Timeout after 60 seconds
                    # if not messages:
                    #     logging.info("No new messages in folder: {}".format(folder))
                    #     continue
                    for item in messages:
                        if item[1] in (b'EXISTS'):
                            logging.info("New messages in folder: {}".format(folder))
                            for uid, message_data in _server.fetch([item[0]], 'RFC822').items():
                                if uid:
                                    eventHandler('NewMailEvent', uid, message_data, _server)                    
                except Exception as e:
                    logging.error("Error during idle check: {}".format(e))
                _server.idle_done()

    username = os.environ.get('username')
    password = os.environ.get('password')
    server = os.environ.get('server')
    folders = os.environ.get('folders',"")
    parser_id = os.environ.get('alarmfax_parser_id',"")
    primary_smtp_address = os.environ.get('primary_smtp_address')
    filter_from = os.environ.get('filter_from').split(";") if os.environ.get('filter_from') else []
    IS_DEV = True if os.environ.get('IS_DEV') and os.environ.get('IS_DEV') == "True" else False
    if IS_DEV:
        logging.getLogger().setLevel(logging.INFO)

    ssl_context = ssl.create_default_context()


    with IMAPClient(server, ssl_context=ssl_context) as _server:
        _server.login(username, password)
        _server.logout()
        folders_to_subscribe = []
        for f in folders.split(";"):
            if f == "":
                folders_to_subscribe.append('INBOX')
            else:
                folders_to_subscribe.append(f)

        while True:
            for f in folders_to_subscribe:
                if not f in threads or not threads[f].is_alive():
                    logging.info("folder_event_subscriber for folder \"{}\" not alive, starting".format(f))
                    t = Thread(target=folder_event_subscriber, args=(f,), daemon=True, name="folder_event_subscriber {}".format(f))
                    threads[f] = t
                    t.start()