Paket aufgeräumt und auf lokalem Testserver

fehlerbericht/config.py

@@ -0,0 +1,8 @@
+# DSN = "mysql+pymysql://gaps:Gcbs12ma@192.168.2.41/tasks"
+DSN = "mysql+pymysql://gaps:Gcbs12ma@localhost/tasks"
+MAILBOX = {
+    "server": "imap.1und1.de",
+    "user": "status@global-cube.de",
+    "password": "(6QNLU=7m3R?f2?6]JO4WH(K",
+    "folder": "Test",  # "INBOX"
+}

fehlerbericht/mail_import.py

@@ -1,59 +1,41 @@
+from config import DSN, MAILBOX
+from model import StatusMeldung
+
 from datetime import date, time
 from email.message import EmailMessage
-from sqlalchemy import Integer, create_engine, String, Date, Time
-from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
-from sqlalchemy.dialects.mysql.base import LONGTEXT
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
 from sqlalchemy.exc import IntegrityError
 from imap_tools import MailBox
 import json
 
 
-class Base(DeclarativeBase):
-    pass
-
-
-class StatusMeldung(Base):
-    __tablename__ = "status_meldung"
-    datum: Mapped[date] = mapped_column(Date, primary_key=True)
-    kunde: Mapped[str] = mapped_column(String(50), primary_key=True)
-    aufgabe: Mapped[str] = mapped_column(String(30))
-    start: Mapped[time] = mapped_column(Time, primary_key=True)
-    ende: Mapped[time] = mapped_column(Time)
-    fehlerbericht: Mapped[LONGTEXT] = mapped_column(LONGTEXT)
-    anzahl: Mapped[int] = mapped_column(Integer, default=0)
-    bearbeitet: Mapped[int] = mapped_column(Integer, default=0)
-    kommentar_id: Mapped[int] = mapped_column(Integer, default=0)
-
-
 class MailImport:
-    database = "mysql+pymysql://gaps:Gcbs12ma@192.168.2.41/tasks"
-    mailbox = ("imap.1und1.de", "status@global-cube.de", "(6QNLU=7m3R?f2?6]JO4WH(K")
-
-    def __init__(self):
-        self.engine = create_engine(self.database)
-
     def open_mailbox(self) -> MailBox:
-        return MailBox(self.mailbox[0]).login(self.mailbox[1], self.mailbox[2])
+        return MailBox(MAILBOX["server"]).login(
+            MAILBOX["user"], MAILBOX["password"], MAILBOX["folder"]
+        )
 
     def mail_import(self):
         with (
-            Session(self.engine) as self.session,
+            Session(create_engine(DSN)) as self.session,
             self.open_mailbox() as mb,
         ):
-            mb.folder.set("Test")
-            messages = mb.fetch(mark_seen=True, limit=100, headers_only=False)
+            messages = mb.fetch(mark_seen=True, limit=100)
             imported = []
 
             for msg in messages:
                 status = self.get_status_message(msg)
+                if status is None:
+                    continue
 
                 self.session.add(status)
                 try:
                     self.session.commit()
-                    imported.append(msg.uid)
                 except IntegrityError as e:
                     self.session.rollback()
                     print(e.args[0])
+                imported.append(msg.uid)
 
             # mb.delete(imported)
 
@@ -65,7 +47,7 @@ class MailImport:
 
         attachments = dict([(att.filename, att.payload) for att in msg.attachments])
         fehlerbericht_json = json.loads(attachments.get("fehlerbericht.json", ""))
-        fehlerbericht = json.dumps(fehlerbericht_json, indent=2)
+        fehlerbericht_import = json.dumps(fehlerbericht_json)
 
         aufgabe = ""
         if len(subject) > 5:
@@ -87,7 +69,7 @@ class MailImport:
             aufgabe=aufgabe,
             start=time.fromisoformat(subject[1]),
             ende=time.fromisoformat(subject[2]),
-            fehlerbericht=fehlerbericht,
+            fehlerbericht_import=fehlerbericht_import,
             anzahl=int(subject[4]),
         )
 

fehlerbericht/model.py

@@ -0,0 +1,22 @@
+from sqlalchemy import Integer, String, Date, Time
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+from sqlalchemy.dialects.mysql.base import LONGTEXT
+from datetime import date, time
+
+
+class Base(DeclarativeBase):
+    pass
+
+
+class StatusMeldung(Base):
+    __tablename__ = "status_meldung"
+    datum: Mapped[date] = mapped_column(Date, primary_key=True)
+    kunde: Mapped[str] = mapped_column(String(50), primary_key=True)
+    aufgabe: Mapped[str] = mapped_column(String(30))
+    start: Mapped[time] = mapped_column(Time, primary_key=True)
+    ende: Mapped[time] = mapped_column(Time)
+    fehlerbericht_import: Mapped[LONGTEXT] = mapped_column(LONGTEXT, nullable=True)
+    fehlerbericht: Mapped[LONGTEXT] = mapped_column(LONGTEXT, nullable=True)
+    anzahl: Mapped[int] = mapped_column(Integer, default=0)
+    bearbeitet: Mapped[int] = mapped_column(Integer, default=0)
+    kommentar_id: Mapped[int] = mapped_column(Integer, default=0)