Erster Versuch

fehlerbericht/mail_import.py

@@ -0,0 +1,80 @@
+from typing import Optional
+from sqlalchemy import Column, create_engine, String
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
+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)
+
+
+
+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 mail_import(self):
+        with MailBox(self.mailbox[0]).login(self.mailbox[1], self.mailbox[2]) as mb:
+            mb.folder.set("Test")
+            messages = mb.fetch(mark_seen=True, limit=100, headers_only=False)
+
+            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)
+
+    def cleanup(self):
+        pass
+
+
+if __name__ == '__main__':
+    mi = MailImport()
+    mi.mail_import()
+    mi.cleanup()