Python-Projekte/fehlerbericht/mail_import.py

import sys
from config import DSN, MAILBOX
from model import StatusMeldung

from datetime import date, time, timedelta
from email.message import EmailMessage
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from imap_tools import MailBox, AND
import json


class MailImport:
    def open_mailbox(self) -> MailBox:
        return MailBox(MAILBOX["server"]).login(
            MAILBOX["user"], MAILBOX["password"], MAILBOX["folder"]
        )

    def mail_import(self) -> int:
        with (
            Session(create_engine(DSN)) as db_session,
            self.open_mailbox() as mb,
        ):
            messages = mb.fetch(mark_seen=True, limit=500)
            imported = []

            for msg in messages:
                status = self.get_status_message(msg)
                if status is None:
                    continue

                db_session.add(status)
                try:
                    db_session.commit()
                    imported.append(msg.uid)
                except IntegrityError as e:
                    db_session.rollback()
                    print(e.args[0])

            mb.delete(imported)
            return len(imported)

    def get_status_message(self, msg: EmailMessage) -> StatusMeldung:
        subject = msg.subject.split(";")
        if len(subject) < 5:
            print(msg.subject)
            return None

        attachments = dict(
            [(att.filename, att.payload) for att in msg.attachments if att.size > 0]
        )
        fehlerbericht_json = json.loads(attachments.get("fehlerbericht.json", "[]"))
        fehlerbericht_import = json.dumps(fehlerbericht_json)

        aufgabe = ""
        if len(subject) > 5:
            if subject[5][-4:] == ".bat":
                aufgabe = subject[5]
            elif ":" in subject[5]:
                aufgabe = "manuell"
        if ":" not in subject[1]:
            aufgabe = subject[1]
            subject[1] = "00:00"
        if aufgabe == "":
            files = [f["Name"] for f in fehlerbericht_json if ".bat" in f["Name"]]
            if len(files) > 0:
                aufgabe = files[0]

        return StatusMeldung(
            datum=date.fromisoformat(subject[3]),
            kunde=subject[0],
            aufgabe=aufgabe,
            start=time.fromisoformat(subject[1]),
            ende=time.fromisoformat(subject[2]),
            fehlerbericht_import=fehlerbericht_import,
            anzahl=int(subject[4]),
        )

    def cleanup(self):
        date_criteria = date.today() - timedelta(days=7)

        with self.open_mailbox() as mb:
            messages = mb.fetch(
                criteria=AND(date_lt=date_criteria, seen=True),
                limit=500,
            )
            selected = [msg.uid for msg in messages]
            mb.delete(selected)


if __name__ == "__main__":
    mi = MailImport()
    res = mi.mail_import()
    mi.cleanup()
    if res == 0:
        sys.exit(1)