fix: PDF extraction und Speicherung korrigiert
- db-Parameter in extract_and_process_attachments hinzugefügt - INVOICE_STORAGE_PATH in docker-compose.yml gesetzt - Ablagepfad: /tmp/invoice-storage im Container - Alte DB-Einträge aktualisiert Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
parent
937267b507
commit
97b5a7ec45
2 changed files with 175 additions and 12 deletions
|
|
@ -6,6 +6,7 @@ IMAP Email-Verarbeitung und Filterung
|
|||
import imaplib
|
||||
import email
|
||||
import os
|
||||
import tempfile
|
||||
from datetime import datetime
|
||||
from typing import List, Optional, Dict, Any
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
|
|
@ -32,6 +33,9 @@ GMAIL_IMAP_PASSWORD = os.getenv("GMAIL_IMAP_PASSWORD", "")
|
|||
# Filter-Kriterien
|
||||
INVOICE_KEYWORDS = ["rechnung", "invoice", "receipt", "bill", "factuur"]
|
||||
|
||||
# Ablagepfad (Evo-X2) - optional
|
||||
INVOICE_STORAGE_PATH = os.getenv("INVOICE_STORAGE_PATH", "/tmp/invoice-storage")
|
||||
|
||||
|
||||
def connect_imap(host: str, port: int, user: str, password: str) -> imaplib.IMAP4_SSL:
|
||||
"""Verbindet mit IMAP-Server"""
|
||||
|
|
@ -100,6 +104,36 @@ def extract_attachments(msg: email.message.Message) -> List[Dict[str, Any]]:
|
|||
return attachments
|
||||
|
||||
|
||||
def parse_email_date(date_str: str) -> Optional[datetime]:
|
||||
"""Parses various email date formats"""
|
||||
if not date_str:
|
||||
return None
|
||||
|
||||
from email.utils import parsedate_to_datetime
|
||||
|
||||
try:
|
||||
# Versuche email.utils parsedate_to_datetime
|
||||
return parsedate_to_datetime(date_str)
|
||||
except:
|
||||
pass
|
||||
|
||||
# Fallback: Versuche verschiedene Formate
|
||||
formats = [
|
||||
"%a, %d %b %Y %H:%M:%S %z",
|
||||
"%a, %d %b %Y %H:%M:%S %z (%Z)",
|
||||
"%d %b %Y %H:%M:%S %z",
|
||||
"%Y-%m-%d %H:%M:%S",
|
||||
]
|
||||
|
||||
for fmt in formats:
|
||||
try:
|
||||
return datetime.strptime(date_str.strip(), fmt)
|
||||
except:
|
||||
continue
|
||||
|
||||
return None
|
||||
|
||||
|
||||
@router.get("/search")
|
||||
async def search_emails(db: Session = Depends(get_db)):
|
||||
"""Sucht nach Rechnungs-Emails bei Strato"""
|
||||
|
|
@ -141,13 +175,110 @@ async def process_email(email_id: str, db: Session = Depends(get_db)):
|
|||
pass
|
||||
|
||||
|
||||
def extract_and_process_attachments(msg: email.message.Message, email_id: str, db: Session) -> List[Dict[str, Any]]:
|
||||
"""Extrahiert und verarbeitet PDF-Anhänge aus Email"""
|
||||
processed = []
|
||||
|
||||
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 not filename:
|
||||
continue
|
||||
|
||||
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()
|
||||
|
||||
# Nur PDFs verarbeiten
|
||||
if content_type != 'application/pdf' and not filename.lower().endswith('.pdf'):
|
||||
continue
|
||||
|
||||
payload = part.get_payload(decode=True)
|
||||
if not payload:
|
||||
continue
|
||||
|
||||
# Temporäre Datei erstellen
|
||||
with tempfile.NamedTemporaryFile(suffix='.pdf', delete=False) as tmp:
|
||||
tmp.write(payload)
|
||||
tmp_path = tmp.name
|
||||
|
||||
try:
|
||||
# PDF verarbeiten
|
||||
from backend.routers.invoices import extract_pdf_data, get_storage_path
|
||||
|
||||
pdf_data = extract_pdf_data(tmp_path)
|
||||
|
||||
if "error" not in pdf_data and pdf_data.get("amount"):
|
||||
# Rechnung speichern
|
||||
tax_type = pdf_data.get("tax_info", {}).get("tax_type", "unbekannt")
|
||||
tax_rate = pdf_data.get("tax_info", {}).get("tax_rate", 0)
|
||||
|
||||
# Speicherpfad erstellen
|
||||
storage_year = 2026 # Default, könnte aus Rechnungsdatum extrahiert werden
|
||||
storage_path = get_storage_path(storage_year, tax_type)
|
||||
|
||||
# PDF in Zielverzeichnis kopieren (optional)
|
||||
if INVOICE_STORAGE_PATH:
|
||||
os.makedirs(storage_path, exist_ok=True)
|
||||
|
||||
# Dateiname generieren
|
||||
invoice_number = pdf_data.get("invoice_number") or "unbekannt"
|
||||
file_name = f"Rechnung_{invoice_number}_{email_id}.pdf"
|
||||
dest_path = os.path.join(storage_path, file_name)
|
||||
|
||||
import shutil
|
||||
shutil.copy(tmp_path, dest_path)
|
||||
|
||||
# In DB speichern
|
||||
invoice = Invoice(
|
||||
email_id=email_id,
|
||||
file_path=dest_path,
|
||||
file_name=file_name,
|
||||
storage_year=storage_year,
|
||||
storage_path=storage_path.replace(INVOICE_STORAGE_PATH, ""),
|
||||
invoice_number=invoice_number,
|
||||
invoice_amount=pdf_data.get("amount"),
|
||||
issuer_name=pdf_data.get("issuer"),
|
||||
invoice_tax_type=tax_type,
|
||||
invoice_tax_rate=tax_rate,
|
||||
processed=True,
|
||||
confidence_score=pdf_data.get("tax_info", {}).get("confidence", 0)
|
||||
)
|
||||
db.add(invoice)
|
||||
db.commit()
|
||||
|
||||
processed.append({
|
||||
"filename": filename,
|
||||
"invoice_number": invoice_number,
|
||||
"amount": pdf_data.get("amount"),
|
||||
"tax_type": tax_type,
|
||||
"saved_to": dest_path
|
||||
})
|
||||
|
||||
finally:
|
||||
# Temporäre Datei löschen
|
||||
try:
|
||||
os.unlink(tmp_path)
|
||||
except:
|
||||
pass
|
||||
|
||||
return processed
|
||||
|
||||
|
||||
@router.get("/scan-all")
|
||||
async def scan_all_emails(db: Session = Depends(get_db)):
|
||||
"""Scant alle Emails auf Rechnungen"""
|
||||
"""Scant alle Emails, extrahiert PDFs und verarbeitet Rechnungen"""
|
||||
results = {
|
||||
"start_time": datetime.utcnow().isoformat(),
|
||||
"processed": 0,
|
||||
"invoices_found": 0,
|
||||
"pdfs_processed": 0,
|
||||
"errors": []
|
||||
}
|
||||
|
||||
|
|
@ -162,10 +293,11 @@ async def scan_all_emails(db: Session = Depends(get_db)):
|
|||
|
||||
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
|
||||
# Prüfen ob bereits verarbeitet
|
||||
log_entry = db.query(EmailLog).filter(EmailLog.email_id == email_id_str).first()
|
||||
if log_entry and log_entry.processed:
|
||||
continue
|
||||
|
||||
result, msg_data = mail.fetch(email_id, '(RFC822)')
|
||||
if result != "OK":
|
||||
|
|
@ -174,7 +306,10 @@ async def scan_all_emails(db: Session = Depends(get_db)):
|
|||
msg = email.message_from_bytes(msg_data[0][1])
|
||||
subject = msg.get("Subject", "")
|
||||
from_addr = msg.get("From", "")
|
||||
date = msg.get("Date", "")
|
||||
date_str = msg.get("Date", "")
|
||||
|
||||
# Datum parsen
|
||||
email_date = parse_email_date(date_str)
|
||||
|
||||
# Subject decodieren
|
||||
decoded_subject, encoding = decode_header(subject)[0]
|
||||
|
|
@ -186,20 +321,30 @@ async def scan_all_emails(db: Session = Depends(get_db)):
|
|||
matched_keywords = [kw for kw in INVOICE_KEYWORDS if kw in subject_lower]
|
||||
|
||||
if matched_keywords:
|
||||
# Email als Rechnung markieren
|
||||
# Email als Rechnung markieren (auch wenn keine PDF dabei ist)
|
||||
log_entry = EmailLog(
|
||||
email_id=email_id_str,
|
||||
email_subject=subject,
|
||||
email_from=from_addr,
|
||||
email_date=date,
|
||||
email_date=email_date,
|
||||
email_folder="INBOX",
|
||||
matched_keywords=matched_keywords,
|
||||
has_attachment=True,
|
||||
processed=False
|
||||
processed=True
|
||||
)
|
||||
db.add(log_entry)
|
||||
results["invoices_found"] += 1
|
||||
|
||||
# PDFs extrahieren und verarbeiten
|
||||
pdfs = extract_and_process_attachments(msg, email_id_str, db)
|
||||
results["pdfs_processed"] += len(pdfs)
|
||||
|
||||
# Log aktualisieren mit Ergebnis
|
||||
if pdfs and log_entry:
|
||||
log_entry.processing_result = f"PDFs processed: {len(pdfs)}"
|
||||
log_entry.processed = True
|
||||
db.commit()
|
||||
|
||||
results["processed"] += 1
|
||||
|
||||
db.commit()
|
||||
|
|
@ -208,6 +353,8 @@ async def scan_all_emails(db: Session = Depends(get_db)):
|
|||
|
||||
except Exception as e:
|
||||
results["errors"].append(f"Strato scan failed: {str(e)}")
|
||||
import traceback
|
||||
results["errors"].append(traceback.format_exc())
|
||||
|
||||
# Gmail Mail (optional)
|
||||
if GMAIL_IMAP_USER and GMAIL_IMAP_PASSWORD:
|
||||
|
|
@ -221,9 +368,9 @@ async def scan_all_emails(db: Session = Depends(get_db)):
|
|||
|
||||
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:
|
||||
log_entry = db.query(EmailLog).filter(EmailLog.email_id == email_id_str).first()
|
||||
if log_entry and log_entry.processed:
|
||||
continue
|
||||
|
||||
result, msg_data = mail.fetch(email_id, '(RFC822)')
|
||||
|
|
@ -233,6 +380,9 @@ async def scan_all_emails(db: Session = Depends(get_db)):
|
|||
msg = email.message_from_bytes(msg_data[0][1])
|
||||
subject = msg.get("Subject", "")
|
||||
from_addr = msg.get("From", "")
|
||||
date_str = msg.get("Date", "")
|
||||
|
||||
email_date = parse_email_date(date_str)
|
||||
|
||||
decoded_subject, encoding = decode_header(subject)[0]
|
||||
if isinstance(decoded_subject, bytes):
|
||||
|
|
@ -246,15 +396,23 @@ async def scan_all_emails(db: Session = Depends(get_db)):
|
|||
email_id=email_id_str,
|
||||
email_subject=subject,
|
||||
email_from=from_addr,
|
||||
email_date=datetime.utcnow().isoformat(),
|
||||
email_date=email_date,
|
||||
email_folder="INBOX",
|
||||
matched_keywords=matched_keywords,
|
||||
has_attachment=True,
|
||||
processed=False
|
||||
processed=True
|
||||
)
|
||||
db.add(log_entry)
|
||||
results["invoices_found"] += 1
|
||||
|
||||
pdfs = extract_and_process_attachments(msg, email_id_str, db)
|
||||
results["pdfs_processed"] += len(pdfs)
|
||||
|
||||
if pdfs:
|
||||
log_entry.processing_result = f"PDFs processed: {len(pdfs)}"
|
||||
log_entry.processed = True
|
||||
db.commit()
|
||||
|
||||
results["processed"] += 1
|
||||
|
||||
db.commit()
|
||||
|
|
@ -263,6 +421,8 @@ async def scan_all_emails(db: Session = Depends(get_db)):
|
|||
|
||||
except Exception as e:
|
||||
results["errors"].append(f"Gmail scan failed: {str(e)}")
|
||||
import traceback
|
||||
results["errors"].append(traceback.format_exc())
|
||||
|
||||
results["end_time"] = datetime.utcnow().isoformat()
|
||||
return results
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@ services:
|
|||
- SMTP_PORT=587
|
||||
- SMTP_USER=${SMTP_USER}
|
||||
- SMTP_PASSWORD=${SMTP_PASSWORD}
|
||||
- INVOICE_STORAGE_PATH=/tmp/invoice-storage
|
||||
depends_on:
|
||||
- db
|
||||
networks:
|
||||
|
|
@ -25,6 +26,7 @@ services:
|
|||
volumes:
|
||||
- ./backend/tests:/app/backend/tests
|
||||
- /tmp/test-invoices:/tmp/test-invoices
|
||||
- invoice-storage:/tmp/invoice-storage
|
||||
|
||||
db:
|
||||
image: postgres:16-alpine
|
||||
|
|
@ -45,3 +47,4 @@ networks:
|
|||
|
||||
volumes:
|
||||
postgres-data:
|
||||
invoice-storage:
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue