erster erfolgreicher Testlauf

fehlerbericht/mail_import.py

@@ -1,80 +1,101 @@
-from typing import Optional
-from sqlalchemy import Column, create_engine, String
-from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+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.exc import IntegrityError
 from imap_tools import MailBox
-import re
-from dataclasses import dataclass
 import json
 
 
-@dataclass
-class StatusMail:
-    datum: str
-    kunde: str
-    start: str
-    ende: str
-    aufgabe: str = ""
-    betreff: str = ""
-    fehlerbericht: object = None
-    anzahl: int = 0
-    bearbeitet: int = 0
-    kommentar_id: int = 0
-
-
 class Base(DeclarativeBase):
     pass
 
 
 class StatusMeldung(Base):
-    __tablename__ = 'status_meldung'
-    id: Mapped[int] = mapped_column(primary_key=True)
-    name: Mapped[str] = mapped_column(String(30))
-    fullname: Mapped[Optional[str]]    
-    id=Column(Integer, primary_key=True)
-    title=Column('title', String(32))
-    in_stock=Column('in_stock', Boolean)
-    quantity=Column('quantity', Integer)
-    price=Column('price', Numeric)
-
+    __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")
+    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])
+
     def mail_import(self):
-        with MailBox(self.mailbox[0]).login(self.mailbox[1], self.mailbox[2]) as mb:
+        with (
+            Session(self.engine) as self.session,
+            self.open_mailbox() as mb,
+        ):
             mb.folder.set("Test")
             messages = mb.fetch(mark_seen=True, limit=100, headers_only=False)
+            imported = []
 
             for msg in messages:
-                subject = msg.subject.split(";")
-                if len(subject) < 5:
-                    print(msg.subject)
-                    continue
-                status = StatusMail(datum=subject[3], kunde=subject[0], start=subject[1], ende=subject[2], anzahl=subject[4], betreff=msg.subject)
-                if len(subject) > 5:
-                    if subject[5][-4:] == ".bat":
-                        status.aufgabe = subject[5]
-                    elif ":" in subject[5]:
-                        status.aufgabe = "manuell"
-                attachments = dict([(att.filename, att.payload) for att in msg.attachments])
-                status.fehlerbericht = json.dumps(json.loads(attachments.get("fehlerbericht.json", "")), indent=2)
-                self.insert_status_raw(status)
-
-    def insert_status_raw(self, content: StatusMail):
-        with self.engine.connect() as con:
-            query = f"INSERT INTO ping (kunde, pingtime, uptime) VALUES ('{kunde}', '{int(current_date)}', '{uptime}')"
-            con.execute(query)
+                status = self.get_status_message(msg)
+
+                self.session.add(status)
+                try:
+                    self.session.commit()
+                    imported.append(msg.uid)
+                except IntegrityError as e:
+                    self.session.rollback()
+                    print(e.args[0])
+
+            # mb.delete(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])
+        fehlerbericht_json = json.loads(attachments.get("fehlerbericht.json", ""))
+        fehlerbericht = json.dumps(fehlerbericht_json, indent=2)
+
+        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=fehlerbericht,
+            anzahl=int(subject[4]),
+        )
 
     def cleanup(self):
         pass
 
 
-if __name__ == '__main__':
+if __name__ == "__main__":
     mi = MailImport()
     mi.mail_import()
     mi.cleanup()