Fix: Feinsortierung im Scheduler + DB-Reconnect
- execute_sortierregeln verarbeitet jetzt OrdnerRegel-Zuweisungen (wie der Button, nicht nur freie_ordner) - DB-Verbindung mit Retry bei Verbindungsfehlern (5 Versuche) - SQLAlchemy 2.0 kompatibel (text() für Raw-SQL) - Bessere Pool-Einstellungen für MariaDB Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
parent
f7a84c50e2
commit
bb84f200d0
4 changed files with 293 additions and 39 deletions
BIN
Docker - Image/V 2.6.tar
Executable file
BIN
Docker - Image/V 2.6.tar
Executable file
Binary file not shown.
|
|
@ -26,15 +26,53 @@ if is_sqlite:
|
|||
cursor.execute("PRAGMA busy_timeout=30000")
|
||||
cursor.close()
|
||||
else:
|
||||
# MariaDB/MySQL
|
||||
# MariaDB/MySQL mit Retry bei Verbindungsfehlern
|
||||
engine = create_engine(
|
||||
DATABASE_URL,
|
||||
echo=False,
|
||||
pool_pre_ping=True,
|
||||
pool_recycle=3600
|
||||
pool_pre_ping=True, # Prüft Verbindung vor Nutzung
|
||||
pool_recycle=1800, # Recycled Verbindungen nach 30 Min
|
||||
pool_size=5, # Max 5 Verbindungen im Pool
|
||||
max_overflow=10, # Bis zu 10 zusätzliche bei Bedarf
|
||||
pool_timeout=30, # Timeout beim Warten auf Verbindung
|
||||
connect_args={
|
||||
"connect_timeout": 10 # Timeout beim Verbindungsaufbau
|
||||
}
|
||||
)
|
||||
|
||||
SessionLocal = sessionmaker(bind=engine)
|
||||
|
||||
|
||||
def get_db_with_retry(max_retries: int = 5, retry_delay: int = 5):
|
||||
"""
|
||||
Gibt eine DB-Session zurück, mit Retry bei Verbindungsfehlern.
|
||||
Für Scheduler und Background-Tasks.
|
||||
"""
|
||||
import time
|
||||
from sqlalchemy.exc import OperationalError, DisconnectionError
|
||||
from sqlalchemy import text
|
||||
|
||||
last_error = None
|
||||
|
||||
for attempt in range(max_retries):
|
||||
try:
|
||||
db = SessionLocal()
|
||||
# Test-Query um Verbindung zu prüfen
|
||||
db.execute(text("SELECT 1"))
|
||||
return db
|
||||
except (OperationalError, DisconnectionError) as e:
|
||||
last_error = e
|
||||
if attempt < max_retries - 1:
|
||||
print(f"[DB] ⚠️ Verbindungsfehler (Versuch {attempt + 1}/{max_retries}): {e}", flush=True)
|
||||
print(f"[DB] Warte {retry_delay} Sekunden vor erneutem Versuch...", flush=True)
|
||||
time.sleep(retry_delay)
|
||||
else:
|
||||
print(f"[DB] ❌ Verbindung fehlgeschlagen nach {max_retries} Versuchen: {e}", flush=True)
|
||||
except Exception as e:
|
||||
print(f"[DB] ❌ Unerwarteter Fehler: {e}", flush=True)
|
||||
raise
|
||||
|
||||
raise last_error
|
||||
Base = declarative_base()
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -11,7 +11,7 @@ from pathlib import Path
|
|||
from typing import Optional, Dict, List
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
from ..models.database import SessionLocal, Zeitplan, Postfach, QuellOrdner
|
||||
from ..models.database import SessionLocal, Zeitplan, Postfach, QuellOrdner, OrdnerRegel, SortierRegel
|
||||
from ..modules.mail_fetcher import MailFetcher
|
||||
from ..modules.sorter import Sorter
|
||||
from ..config import INBOX_DIR
|
||||
|
|
@ -342,8 +342,46 @@ def create_trigger(zp: Zeitplan) -> Optional[CronTrigger]:
|
|||
|
||||
|
||||
def execute_zeitplan(zeitplan_id: int):
|
||||
"""Führt einen Zeitplan aus"""
|
||||
db = SessionLocal()
|
||||
"""Führt einen Zeitplan aus mit Retry bei DB-Verbindungsfehlern"""
|
||||
from sqlalchemy.exc import OperationalError, DisconnectionError
|
||||
from sqlalchemy import text
|
||||
import time
|
||||
|
||||
max_db_retries = 5
|
||||
retry_delay = 10
|
||||
|
||||
# Versuche DB-Verbindung mit Retry
|
||||
db = None
|
||||
for attempt in range(max_db_retries):
|
||||
try:
|
||||
db = SessionLocal()
|
||||
# Test-Query um Verbindung zu prüfen
|
||||
db.execute(text("SELECT 1"))
|
||||
break
|
||||
except (OperationalError, DisconnectionError) as e:
|
||||
print(f"[SCHEDULER] ⚠️ DB-Verbindungsfehler bei Zeitplan {zeitplan_id} (Versuch {attempt + 1}/{max_db_retries}): {e}", flush=True)
|
||||
if db:
|
||||
try:
|
||||
db.close()
|
||||
except:
|
||||
pass
|
||||
if attempt < max_db_retries - 1:
|
||||
print(f"[SCHEDULER] Warte {retry_delay} Sekunden vor erneutem Versuch...", flush=True)
|
||||
time.sleep(retry_delay)
|
||||
else:
|
||||
print(f"[SCHEDULER] ❌ DB-Verbindung fehlgeschlagen nach {max_db_retries} Versuchen - Zeitplan übersprungen", flush=True)
|
||||
logger.error(f"DB-Verbindung fehlgeschlagen für Zeitplan {zeitplan_id}")
|
||||
return
|
||||
except Exception as e:
|
||||
print(f"[SCHEDULER] ❌ Unerwarteter DB-Fehler: {e}", flush=True)
|
||||
logger.error(f"Unerwarteter DB-Fehler für Zeitplan {zeitplan_id}: {e}")
|
||||
if db:
|
||||
try:
|
||||
db.close()
|
||||
except:
|
||||
pass
|
||||
return
|
||||
|
||||
try:
|
||||
zeitplan = db.query(Zeitplan).filter(Zeitplan.id == zeitplan_id).first()
|
||||
if not zeitplan:
|
||||
|
|
@ -386,15 +424,36 @@ def execute_zeitplan(zeitplan_id: int):
|
|||
db.commit()
|
||||
logger.info(f"Zeitplan abgeschlossen: {zeitplan.name} - {zeitplan.letzter_status}")
|
||||
|
||||
except (OperationalError, DisconnectionError) as e:
|
||||
# DB-Verbindung während Ausführung verloren
|
||||
print(f"[SCHEDULER] ❌ DB-Verbindung verloren während Zeitplan '{zeitplan.name}': {e}", flush=True)
|
||||
logger.error(f"DB-Verbindung verloren bei Zeitplan {zeitplan.name}: {e}")
|
||||
# Versuche Status trotzdem zu speichern
|
||||
try:
|
||||
db.rollback()
|
||||
zeitplan.letzte_ausfuehrung = datetime.utcnow()
|
||||
zeitplan.letzter_status = "fehler"
|
||||
zeitplan.letzte_meldung = f"DB-Verbindungsfehler: {str(e)[:450]}"
|
||||
db.commit()
|
||||
except:
|
||||
pass
|
||||
|
||||
except Exception as e:
|
||||
zeitplan.letzte_ausfuehrung = datetime.utcnow()
|
||||
zeitplan.letzter_status = "fehler"
|
||||
zeitplan.letzte_meldung = str(e)[:500]
|
||||
db.commit()
|
||||
try:
|
||||
db.commit()
|
||||
except:
|
||||
pass
|
||||
logger.error(f"Fehler bei Zeitplan {zeitplan.name}: {e}")
|
||||
|
||||
finally:
|
||||
db.close()
|
||||
if db:
|
||||
try:
|
||||
db.close()
|
||||
except:
|
||||
pass
|
||||
|
||||
|
||||
def execute_mail_abruf(db, zeitplan: Zeitplan) -> Dict:
|
||||
|
|
@ -743,33 +802,193 @@ def execute_grobsortierung(db, zeitplan: Zeitplan) -> Dict:
|
|||
|
||||
|
||||
def execute_sortierregeln(db, zeitplan: Zeitplan) -> Dict:
|
||||
"""Führt nur Sortierregeln aus (freie Ordner von Regeln)"""
|
||||
from ..models.database import SortierRegel, VerarbeiteteDatei
|
||||
"""Führt Feinsortierung aus (Ordner-Zuweisungen + freie Ordner von Regeln)"""
|
||||
from ..models.database import SortierRegel, VerarbeiteteDatei, OrdnerRegel
|
||||
from ..modules.pdf_processor import PDFProcessor
|
||||
from pathlib import Path
|
||||
|
||||
# Regeln laden (optional spezifische Regel)
|
||||
if zeitplan.regel_id:
|
||||
regeln = db.query(SortierRegel).filter(
|
||||
SortierRegel.id == zeitplan.regel_id,
|
||||
SortierRegel.aktiv == True
|
||||
).all()
|
||||
else:
|
||||
regeln = db.query(SortierRegel).filter(SortierRegel.aktiv == True).all()
|
||||
|
||||
if not regeln:
|
||||
return {"erfolg": True, "meldung": "Keine aktiven Regeln gefunden"}
|
||||
print("", flush=True)
|
||||
print("[FEINSORTIERUNG] === START ===", flush=True)
|
||||
print(f"[FEINSORTIERUNG] Zeitplan: {zeitplan.name} (ID: {zeitplan.id})", flush=True)
|
||||
|
||||
pdf_processor = PDFProcessor()
|
||||
gesamt_sortiert = 0
|
||||
gesamt_fehler = 0
|
||||
gesamt_ohne_regel = 0
|
||||
fehler_meldungen = []
|
||||
|
||||
for regel in regeln:
|
||||
# Fallback-Regeln laden (gelten für alle Ordner wenn keine andere Regel passt)
|
||||
fallback_regeln = db.query(SortierRegel).filter(
|
||||
SortierRegel.aktiv == True,
|
||||
SortierRegel.ist_fallback == True
|
||||
).order_by(SortierRegel.prioritaet).all()
|
||||
|
||||
fallback_dicts = [{
|
||||
"id": r.id, "name": r.name, "prioritaet": r.prioritaet,
|
||||
"muster": r.muster, "extraktion": r.extraktion,
|
||||
"schema": r.schema, "unterordner": r.unterordner,
|
||||
"ziel_ordner": getattr(r, 'ziel_ordner', None),
|
||||
"nur_umbenennen": getattr(r, 'nur_umbenennen', False)
|
||||
} for r in fallback_regeln]
|
||||
|
||||
fallback_sorter = Sorter(fallback_dicts) if fallback_dicts else None
|
||||
print(f"[FEINSORTIERUNG] Fallback-Regeln: {len(fallback_dicts)}", flush=True)
|
||||
|
||||
# ============ TEIL 1: Ordner-Zuweisungen verarbeiten ============
|
||||
# (wie der Button /api/sortierung/starten)
|
||||
quell_ordner_liste = db.query(QuellOrdner).filter(QuellOrdner.aktiv == True).all()
|
||||
print(f"[FEINSORTIERUNG] Aktive Quellordner: {len(quell_ordner_liste)}", flush=True)
|
||||
|
||||
for quell_ordner in quell_ordner_liste:
|
||||
# Feinsortierung arbeitet im ZIEL-Ordner (wo Dateien nach Grobsortierung liegen)
|
||||
pfad = Path(quell_ordner.ziel_ordner)
|
||||
|
||||
print(f"[FEINSORTIERUNG] --- Ordner: {quell_ordner.name} ---", flush=True)
|
||||
print(f"[FEINSORTIERUNG] Ziel-Pfad: {pfad}", flush=True)
|
||||
|
||||
if not pfad.exists():
|
||||
print(f"[FEINSORTIERUNG] ⚠️ Pfad existiert nicht - überspringe", flush=True)
|
||||
continue
|
||||
|
||||
# Dateien sammeln
|
||||
dateien = [f for f in pfad.glob("**/*") if f.is_file() and f.suffix.lower() == ".pdf"]
|
||||
print(f"[FEINSORTIERUNG] Gefunden: {len(dateien)} PDF-Dateien", flush=True)
|
||||
|
||||
# Lade nur Regeln die diesem Ordner zugewiesen sind
|
||||
zuweisungen = db.query(OrdnerRegel).filter(OrdnerRegel.ordner_id == quell_ordner.id).all()
|
||||
zugewiesene_regel_ids = [z.regel_id for z in zuweisungen]
|
||||
|
||||
if zugewiesene_regel_ids:
|
||||
regeln = db.query(SortierRegel).filter(
|
||||
SortierRegel.id.in_(zugewiesene_regel_ids),
|
||||
SortierRegel.aktiv == True,
|
||||
SortierRegel.ist_fallback == False
|
||||
).order_by(SortierRegel.prioritaet).all()
|
||||
else:
|
||||
regeln = []
|
||||
|
||||
# Regeln in Dict-Format
|
||||
regeln_dicts = [{
|
||||
"id": r.id, "name": r.name, "prioritaet": r.prioritaet,
|
||||
"muster": r.muster, "extraktion": r.extraktion,
|
||||
"schema": r.schema, "unterordner": r.unterordner,
|
||||
"ziel_ordner": getattr(r, 'ziel_ordner', None),
|
||||
"nur_umbenennen": getattr(r, 'nur_umbenennen', False)
|
||||
} for r in regeln]
|
||||
|
||||
sorter = Sorter(regeln_dicts) if regeln_dicts else None
|
||||
print(f"[FEINSORTIERUNG] Zugewiesene Regeln: {len(regeln_dicts)}", flush=True)
|
||||
|
||||
for datei in dateien:
|
||||
try:
|
||||
text = ""
|
||||
ist_zugferd = False
|
||||
ocr_gemacht = False
|
||||
|
||||
# PDF verarbeiten
|
||||
ocr_erlaubt = getattr(quell_ordner, 'ocr_aktivieren', True)
|
||||
original_backup = getattr(quell_ordner, 'original_sichern', None)
|
||||
|
||||
pdf_result = pdf_processor.verarbeite(
|
||||
str(datei),
|
||||
ocr_erlaubt=ocr_erlaubt,
|
||||
original_backup_pfad=original_backup
|
||||
)
|
||||
|
||||
if pdf_result.get("fehler"):
|
||||
raise Exception(pdf_result["fehler"])
|
||||
|
||||
text = pdf_result.get("text", "")
|
||||
ist_zugferd = pdf_result.get("ist_zugferd", False)
|
||||
ocr_gemacht = pdf_result.get("ocr_durchgefuehrt", False)
|
||||
|
||||
# Dokument-Info für Regel-Matching
|
||||
doc_info = {
|
||||
"text": text,
|
||||
"original_name": datei.name,
|
||||
"absender": "",
|
||||
"dateityp": datei.suffix.lower()
|
||||
}
|
||||
|
||||
# Passende Regel finden (erst zugewiesene, dann Fallback)
|
||||
regel = None
|
||||
if sorter:
|
||||
regel = sorter.finde_passende_regel(doc_info)
|
||||
|
||||
# Fallback-Regel wenn keine zugewiesene passt
|
||||
ist_fallback = False
|
||||
if not regel and fallback_sorter:
|
||||
regel = fallback_sorter.finde_passende_regel(doc_info)
|
||||
if regel:
|
||||
ist_fallback = True
|
||||
|
||||
if not regel:
|
||||
print(f"[FEINSORTIERUNG] ⚠️ Keine Regel für: {datei.name}", flush=True)
|
||||
gesamt_ohne_regel += 1
|
||||
continue
|
||||
|
||||
print(f"[FEINSORTIERUNG] ✓ Regel '{regel.get('name')}' für: {datei.name}", flush=True)
|
||||
|
||||
# Felder extrahieren
|
||||
aktiver_sorter = sorter if not ist_fallback else fallback_sorter
|
||||
extrahiert = aktiver_sorter.extrahiere_felder(regel, doc_info)
|
||||
|
||||
# Dateiendung beibehalten
|
||||
schema = regel.get("schema", "{datum} - Dokument.pdf")
|
||||
if schema.endswith(".pdf"):
|
||||
schema = schema[:-4] + datei.suffix
|
||||
neuer_name = aktiver_sorter.generiere_dateinamen(
|
||||
{"schema": schema, **regel}, extrahiert
|
||||
)
|
||||
|
||||
# Zielordner bestimmen
|
||||
if regel.get("nur_umbenennen"):
|
||||
ziel = datei.parent
|
||||
elif regel.get("ziel_ordner"):
|
||||
ziel = Path(regel["ziel_ordner"])
|
||||
if regel.get("unterordner"):
|
||||
ziel = ziel / regel["unterordner"]
|
||||
else:
|
||||
ziel = pfad
|
||||
if regel.get("unterordner"):
|
||||
ziel = ziel / regel["unterordner"]
|
||||
ziel.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Verschieben
|
||||
print(f"[FEINSORTIERUNG] → Verschiebe: {datei.name} -> {ziel}/{neuer_name}", flush=True)
|
||||
neuer_pfad = aktiver_sorter.verschiebe_datei(str(datei), str(ziel), neuer_name)
|
||||
|
||||
gesamt_sortiert += 1
|
||||
|
||||
db.add(VerarbeiteteDatei(
|
||||
original_pfad=str(datei),
|
||||
original_name=datei.name,
|
||||
neuer_pfad=neuer_pfad,
|
||||
neuer_name=neuer_name,
|
||||
ist_zugferd=ist_zugferd,
|
||||
ocr_durchgefuehrt=ocr_gemacht,
|
||||
status="sortiert",
|
||||
extrahierte_daten=extrahiert
|
||||
))
|
||||
|
||||
except Exception as e:
|
||||
gesamt_fehler += 1
|
||||
print(f"[FEINSORTIERUNG] ❌ Fehler bei {datei.name}: {e}", flush=True)
|
||||
logger.error(f"Fehler bei Datei {datei}: {e}")
|
||||
|
||||
# ============ TEIL 2: Freie Ordner von Regeln verarbeiten ============
|
||||
print("", flush=True)
|
||||
print("[FEINSORTIERUNG] --- Freie Ordner ---", flush=True)
|
||||
|
||||
alle_regeln = db.query(SortierRegel).filter(SortierRegel.aktiv == True).all()
|
||||
|
||||
for regel in alle_regeln:
|
||||
freie_ordner = regel.freie_ordner if regel.freie_ordner else []
|
||||
if not freie_ordner:
|
||||
continue
|
||||
|
||||
print(f"[FEINSORTIERUNG] Regel '{regel.name}' hat {len(freie_ordner)} freie Ordner", flush=True)
|
||||
|
||||
regel_dict = {
|
||||
"id": regel.id,
|
||||
"name": regel.name,
|
||||
|
|
@ -786,23 +1005,22 @@ def execute_sortierregeln(db, zeitplan: Zeitplan) -> Dict:
|
|||
for freier_ordner_pfad in freie_ordner:
|
||||
freier_pfad = Path(freier_ordner_pfad)
|
||||
if not freier_pfad.exists() or not freier_pfad.is_dir():
|
||||
print(f"[FEINSORTIERUNG] ⚠️ Freier Ordner existiert nicht: {freier_ordner_pfad}", flush=True)
|
||||
continue
|
||||
|
||||
# Dateien sammeln
|
||||
dateien = [f for f in freier_pfad.glob("**/*") if f.is_file() and f.suffix.lower() == ".pdf"]
|
||||
print(f"[FEINSORTIERUNG] Freier Ordner {freier_ordner_pfad}: {len(dateien)} Dateien", flush=True)
|
||||
|
||||
for datei in dateien:
|
||||
try:
|
||||
ist_pdf = datei.suffix.lower() == ".pdf"
|
||||
text = ""
|
||||
ist_zugferd = False
|
||||
|
||||
if ist_pdf:
|
||||
pdf_result = pdf_processor.verarbeite(str(datei))
|
||||
if pdf_result.get("fehler"):
|
||||
raise Exception(pdf_result["fehler"])
|
||||
text = pdf_result.get("text", "")
|
||||
ist_zugferd = pdf_result.get("ist_zugferd", False)
|
||||
pdf_result = pdf_processor.verarbeite(str(datei))
|
||||
if pdf_result.get("fehler"):
|
||||
raise Exception(pdf_result["fehler"])
|
||||
text = pdf_result.get("text", "")
|
||||
ist_zugferd = pdf_result.get("ist_zugferd", False)
|
||||
|
||||
doc_info = {
|
||||
"text": text,
|
||||
|
|
@ -811,15 +1029,12 @@ def execute_sortierregeln(db, zeitplan: Zeitplan) -> Dict:
|
|||
"dateityp": datei.suffix.lower()
|
||||
}
|
||||
|
||||
# Prüfe ob Regel passt
|
||||
passend = regel_sorter.finde_passende_regel(doc_info)
|
||||
if not passend:
|
||||
continue
|
||||
|
||||
# Felder extrahieren
|
||||
extrahiert = regel_sorter.extrahiere_felder(passend, doc_info)
|
||||
|
||||
# Dateiname generieren
|
||||
schema = passend.get("schema", "{datum} - Dokument.pdf")
|
||||
if schema.endswith(".pdf"):
|
||||
schema = schema[:-4] + datei.suffix
|
||||
|
|
@ -827,23 +1042,18 @@ def execute_sortierregeln(db, zeitplan: Zeitplan) -> Dict:
|
|||
{"schema": schema, **passend}, extrahiert
|
||||
)
|
||||
|
||||
# Zielordner bestimmen
|
||||
if passend.get("nur_umbenennen"):
|
||||
# Nur umbenennen - Datei bleibt im aktuellen Ordner
|
||||
ziel = datei.parent
|
||||
elif passend.get("ziel_ordner"):
|
||||
# Regel hat eigenen Zielordner
|
||||
ziel = Path(passend["ziel_ordner"])
|
||||
if passend.get("unterordner"):
|
||||
ziel = ziel / passend["unterordner"]
|
||||
else:
|
||||
# Kein Zielordner - bleibt im freien Ordner
|
||||
ziel = freier_pfad
|
||||
if passend.get("unterordner"):
|
||||
ziel = ziel / passend["unterordner"]
|
||||
ziel.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
# Verschieben/Umbenennen
|
||||
regel_sorter.verschiebe_datei(str(datei), str(ziel), neuer_name)
|
||||
gesamt_sortiert += 1
|
||||
|
||||
|
|
@ -863,12 +1073,18 @@ def execute_sortierregeln(db, zeitplan: Zeitplan) -> Dict:
|
|||
|
||||
db.commit()
|
||||
|
||||
meldung = f"{gesamt_sortiert} Dateien mit Regeln sortiert"
|
||||
meldung = f"{gesamt_sortiert} Dateien sortiert"
|
||||
if gesamt_ohne_regel > 0:
|
||||
meldung += f", {gesamt_ohne_regel} ohne passende Regel"
|
||||
if gesamt_fehler > 0:
|
||||
meldung += f", {gesamt_fehler} Fehler"
|
||||
if fehler_meldungen:
|
||||
meldung += f" ({'; '.join(fehler_meldungen)})"
|
||||
|
||||
print("", flush=True)
|
||||
print(f"[FEINSORTIERUNG] === ENDE === {meldung}", flush=True)
|
||||
print("", flush=True)
|
||||
|
||||
return {"erfolg": gesamt_fehler == 0, "meldung": meldung}
|
||||
|
||||
|
||||
|
|
|
|||
BIN
dateiverwaltung-latest.tar.gz
Executable file
BIN
dateiverwaltung-latest.tar.gz
Executable file
Binary file not shown.
Loading…
Reference in a new issue