268 lines
8.6 KiB
Python
268 lines
8.6 KiB
Python
"""
|
|
C-Stone Invoice Check — Email Router
|
|
IMAP Email-Verarbeitung und Filterung
|
|
"""
|
|
|
|
import imaplib
|
|
import email
|
|
import os
|
|
from datetime import datetime
|
|
from typing import List, Optional, Dict, Any
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from sqlalchemy.orm import Session
|
|
from email.header import decode_header
|
|
|
|
from backend.database import get_db
|
|
from backend.models import EmailLog, Invoice
|
|
|
|
router = APIRouter()
|
|
|
|
# IMAP Konfiguration
|
|
IMAP_HOST = os.getenv("IMAP_HOST", "imap.strato.de")
|
|
IMAP_PORT = int(os.getenv("IMAP_PORT", 993))
|
|
IMAP_USER = os.getenv("IMAP_USER", "")
|
|
IMAP_PASSWORD = os.getenv("IMAP_PASSWORD", "")
|
|
|
|
# Gmail Konfiguration
|
|
GMAIL_IMAP_HOST = os.getenv("GMAIL_IMAP_HOST", "imap.gmail.com")
|
|
GMAIL_IMAP_PORT = int(os.getenv("GMAIL_IMAP_PORT", 993))
|
|
GMAIL_IMAP_USER = os.getenv("GMAIL_IMAP_USER", "")
|
|
GMAIL_IMAP_PASSWORD = os.getenv("GMAIL_IMAP_PASSWORD", "")
|
|
|
|
# Filter-Kriterien
|
|
INVOICE_KEYWORDS = ["rechnung", "invoice", "receipt", "bill", "factuur"]
|
|
|
|
|
|
def connect_imap(host: str, port: int, user: str, password: str) -> imaplib.IMAP4_SSL:
|
|
"""Verbindet mit IMAP-Server"""
|
|
mail = imaplib.IMAP4_SSL(host, port)
|
|
mail.login(user, password)
|
|
return mail
|
|
|
|
|
|
def search_invoices(mail: imaplib.IMAP4_SSL, folder: str = "INBOX") -> List[str]:
|
|
"""Sucht nach Emails mit Rechnungs-Stichworten"""
|
|
mail.select(folder)
|
|
result, data = mail.search(None, 'ALL')
|
|
if result != "OK":
|
|
return []
|
|
|
|
email_ids = data[0].split()
|
|
matching_ids = []
|
|
|
|
for email_id in email_ids:
|
|
result, msg_data = mail.fetch(email_id, '(RFC822)')
|
|
if result != "OK":
|
|
continue
|
|
|
|
msg = email.message_from_bytes(msg_data[0][1])
|
|
subject = msg.get("Subject", "")
|
|
from_addr = msg.get("From", "")
|
|
|
|
# Subject decodieren
|
|
decoded_subject, encoding = decode_header(subject)[0]
|
|
if isinstance(decoded_subject, bytes):
|
|
subject = decoded_subject.decode(encoding or "utf-8")
|
|
|
|
# Prüfen auf Keywords
|
|
subject_lower = subject.lower()
|
|
if any(kw in subject_lower for kw in INVOICE_KEYWORDS):
|
|
matching_ids.append(email_id.decode())
|
|
|
|
return matching_ids
|
|
|
|
|
|
def extract_attachments(msg: email.message.Message) -> List[Dict[str, Any]]:
|
|
"""Extrahiert Anhänge aus Email"""
|
|
attachments = []
|
|
for part in msg.walk():
|
|
if part.get_content_maintype() == 'multipart':
|
|
continue
|
|
if part.get('Content-Disposition') is None:
|
|
continue
|
|
|
|
filename = part.get_filename()
|
|
if filename:
|
|
decoded_filename, encoding = decode_header(filename)[0]
|
|
if isinstance(decoded_filename, bytes):
|
|
filename = decoded_filename.decode(encoding or "utf-8")
|
|
|
|
content_type = part.get_content_type()
|
|
payload = part.get_payload(decode=True)
|
|
|
|
attachments.append({
|
|
"filename": filename,
|
|
"content_type": content_type,
|
|
"data": payload,
|
|
"size": len(payload) if payload else 0
|
|
})
|
|
|
|
return attachments
|
|
|
|
|
|
@router.get("/search")
|
|
async def search_emails(db: Session = Depends(get_db)):
|
|
"""Sucht nach Rechnungs-Emails bei Strato"""
|
|
results = {
|
|
"processed": 0,
|
|
"invoices_found": 0,
|
|
"errors": []
|
|
}
|
|
|
|
try:
|
|
mail = connect_imap(IMAP_HOST, IMAP_PORT, IMAP_USER, IMAP_PASSWORD)
|
|
|
|
# Alle Ordner durchsuchen
|
|
for folder in ["INBOX", "Sent", "Gesendet"]:
|
|
try:
|
|
email_ids = search_invoices(mail, folder)
|
|
results[f"{folder}_count"] = len(email_ids)
|
|
|
|
for email_id in email_ids:
|
|
# Verarbeitung auslagern
|
|
pass
|
|
|
|
except Exception as e:
|
|
results["errors"].append(f"Folder {folder}: {str(e)}")
|
|
|
|
mail.logout()
|
|
results["processed"] = len(email_ids) if 'email_ids' in dir() else 0
|
|
|
|
except Exception as e:
|
|
results["errors"].append(f"IMAP Connection failed: {str(e)}")
|
|
|
|
return results
|
|
|
|
|
|
@router.get("/process/{email_id}")
|
|
async def process_email(email_id: str, db: Session = Depends(get_db)):
|
|
"""Verarbeitet einzelne Email"""
|
|
# TODO: Implementierung
|
|
pass
|
|
|
|
|
|
@router.get("/scan-all")
|
|
async def scan_all_emails(db: Session = Depends(get_db)):
|
|
"""Scant alle Emails auf Rechnungen"""
|
|
results = {
|
|
"start_time": datetime.utcnow().isoformat(),
|
|
"processed": 0,
|
|
"invoices_found": 0,
|
|
"errors": []
|
|
}
|
|
|
|
# Strato Mail
|
|
try:
|
|
mail = connect_imap(IMAP_HOST, IMAP_PORT, IMAP_USER, IMAP_PASSWORD)
|
|
mail.select("INBOX")
|
|
|
|
result, data = mail.search(None, 'ALL')
|
|
if result == "OK":
|
|
email_ids = data[0].split()
|
|
|
|
for email_id in email_ids:
|
|
email_id_str = email_id.decode()
|
|
log_entry = db.query(EmailLog).filter(EmailLog.email_id == email_id_str).first()
|
|
|
|
if log_entry:
|
|
continue # Bereits verarbeitet
|
|
|
|
result, msg_data = mail.fetch(email_id, '(RFC822)')
|
|
if result != "OK":
|
|
continue
|
|
|
|
msg = email.message_from_bytes(msg_data[0][1])
|
|
subject = msg.get("Subject", "")
|
|
from_addr = msg.get("From", "")
|
|
date = msg.get("Date", "")
|
|
|
|
# Subject decodieren
|
|
decoded_subject, encoding = decode_header(subject)[0]
|
|
if isinstance(decoded_subject, bytes):
|
|
subject = decoded_subject.decode(encoding or "utf-8")
|
|
|
|
# Prüfen auf Keywords
|
|
subject_lower = subject.lower()
|
|
matched_keywords = [kw for kw in INVOICE_KEYWORDS if kw in subject_lower]
|
|
|
|
if matched_keywords:
|
|
# Email als Rechnung markieren
|
|
log_entry = EmailLog(
|
|
email_id=email_id_str,
|
|
email_subject=subject,
|
|
email_from=from_addr,
|
|
email_date=date,
|
|
email_folder="INBOX",
|
|
matched_keywords=matched_keywords,
|
|
has_attachment=True,
|
|
processed=False
|
|
)
|
|
db.add(log_entry)
|
|
results["invoices_found"] += 1
|
|
|
|
results["processed"] += 1
|
|
|
|
db.commit()
|
|
|
|
mail.logout()
|
|
|
|
except Exception as e:
|
|
results["errors"].append(f"Strato scan failed: {str(e)}")
|
|
|
|
# Gmail Mail (optional)
|
|
if GMAIL_IMAP_USER and GMAIL_IMAP_PASSWORD:
|
|
try:
|
|
mail = connect_imap(GMAIL_IMAP_HOST, GMAIL_IMAP_PORT, GMAIL_IMAP_USER, GMAIL_IMAP_PASSWORD)
|
|
mail.select("INBOX")
|
|
|
|
result, data = mail.search(None, 'ALL')
|
|
if result == "OK":
|
|
email_ids = data[0].split()
|
|
|
|
for email_id in email_ids:
|
|
email_id_str = email_id.decode()
|
|
log_entry = db.query(EmailLog).filter(EmailLog.email_id == email_id_str).first()
|
|
|
|
if log_entry:
|
|
continue
|
|
|
|
result, msg_data = mail.fetch(email_id, '(RFC822)')
|
|
if result != "OK":
|
|
continue
|
|
|
|
msg = email.message_from_bytes(msg_data[0][1])
|
|
subject = msg.get("Subject", "")
|
|
from_addr = msg.get("From", "")
|
|
|
|
decoded_subject, encoding = decode_header(subject)[0]
|
|
if isinstance(decoded_subject, bytes):
|
|
subject = decoded_subject.decode(encoding or "utf-8")
|
|
|
|
subject_lower = subject.lower()
|
|
matched_keywords = [kw for kw in INVOICE_KEYWORDS if kw in subject_lower]
|
|
|
|
if matched_keywords:
|
|
log_entry = EmailLog(
|
|
email_id=email_id_str,
|
|
email_subject=subject,
|
|
email_from=from_addr,
|
|
email_date=datetime.utcnow().isoformat(),
|
|
email_folder="INBOX",
|
|
matched_keywords=matched_keywords,
|
|
has_attachment=True,
|
|
processed=False
|
|
)
|
|
db.add(log_entry)
|
|
results["invoices_found"] += 1
|
|
|
|
results["processed"] += 1
|
|
|
|
db.commit()
|
|
|
|
mail.logout()
|
|
|
|
except Exception as e:
|
|
results["errors"].append(f"Gmail scan failed: {str(e)}")
|
|
|
|
results["end_time"] = datetime.utcnow().isoformat()
|
|
return results
|