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 _server.idle_done() # In den events stehen nur vorhandenen nachrichten. Exists ist nicht die neue Nachricht, sondern eine bereits vorhandene. # for item in messages: # if item[1] == 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)) 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()