207 lines
7.1 KiB
Python
207 lines
7.1 KiB
Python
"""Validierung von Input-Dateien und Templates."""
|
||
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
import fitz # PyMuPDF
|
||
from openpyxl import load_workbook
|
||
from openpyxl.utils import column_index_from_string
|
||
|
||
from .config import is_month_sheet
|
||
from .models import FieldSchema, OutputMapping, ValidationResult
|
||
|
||
|
||
def validate_input_file(
|
||
filepath: Path,
|
||
schema: list[FieldSchema],
|
||
month_sheet_pattern: str = "* 20??",
|
||
) -> ValidationResult:
|
||
"""
|
||
Validiert eine Input-Datei:
|
||
- Datei existiert
|
||
- Erwartete Spalten sind vorhanden (Header in Zeile 1)
|
||
- Mindestens ein Monats-Daten-Sheet enthält Daten
|
||
- Nicht-Monats-Sheets (z. B. "Alle Rollen") werden übersprungen
|
||
"""
|
||
errors: list[str] = []
|
||
warnings: list[str] = []
|
||
info: dict = {}
|
||
|
||
if not filepath.exists():
|
||
errors.append(f"Input-Datei nicht gefunden: {filepath}")
|
||
return ValidationResult(is_valid=False, errors=errors)
|
||
|
||
try:
|
||
wb = load_workbook(filepath, data_only=True)
|
||
except Exception as e:
|
||
errors.append(f"Datei konnte nicht geöffnet werden: {e}")
|
||
return ValidationResult(is_valid=False, errors=errors)
|
||
|
||
sheet_count = len(wb.sheetnames)
|
||
info["sheet_count"] = sheet_count
|
||
info["sheet_names"] = wb.sheetnames
|
||
|
||
# Prüfe nur Monats-Sheets (überspringe z. B. "Alle Rollen")
|
||
schema_col_count = len(schema)
|
||
sheets_with_data = 0
|
||
for sheet_name in wb.sheetnames:
|
||
if not is_month_sheet(sheet_name, month_sheet_pattern):
|
||
continue
|
||
|
||
ws = wb[sheet_name]
|
||
if ws.max_row < 2:
|
||
continue # Kein Header + Daten
|
||
|
||
# min_col=1 explizit setzen, um Verschiebungen bei
|
||
# ws.min_column > 1 zu vermeiden
|
||
header_row = list(ws.iter_rows(
|
||
min_row=1, max_row=1, min_col=1, max_col=schema_col_count,
|
||
values_only=True,
|
||
))[0]
|
||
missing_cols = []
|
||
for field in schema:
|
||
col_idx = column_index_from_string(field.column) - 1
|
||
if col_idx >= len(header_row):
|
||
missing_cols.append(f"{field.column} ({field.field_name})")
|
||
continue
|
||
header_val = str(header_row[col_idx]).strip() if header_row[col_idx] is not None else ""
|
||
if header_val != field.field_name:
|
||
warnings.append(
|
||
f"Sheet '{sheet_name}', Spalte {field.column}: "
|
||
f"Erwartet '{field.field_name}', gefunden '{header_val}'"
|
||
)
|
||
|
||
if not missing_cols:
|
||
sheets_with_data += 1
|
||
|
||
# Zähle Datenzeilen
|
||
data_rows = ws.max_row - 1
|
||
info[f"rows_{sheet_name}"] = data_rows
|
||
|
||
info["sheets_with_data"] = sheets_with_data
|
||
wb.close()
|
||
|
||
# ── Dateninhalts-Prüfung: leere Pflichtfelder ───────────────────
|
||
required_fields = [f for f in schema if f.required]
|
||
if required_fields and sheets_with_data > 0:
|
||
# Mapping field_name → column index
|
||
field_col_map: dict[str, int] = {}
|
||
for f in required_fields:
|
||
field_col_map[f.field_name] = column_index_from_string(f.column) - 1
|
||
max_col = max(field_col_map.values()) + 1
|
||
|
||
# {field_name: {name1, name2, ...}} für leere Pflichtwerte
|
||
empty_by_field: dict[str, set[str]] = {f.field_name: set() for f in required_fields}
|
||
|
||
wb2 = load_workbook(filepath, data_only=True)
|
||
for sheet_name in wb2.sheetnames:
|
||
if not is_month_sheet(sheet_name, month_sheet_pattern):
|
||
continue
|
||
ws = wb2[sheet_name]
|
||
if ws.max_row < 2:
|
||
continue
|
||
for row in ws.iter_rows(min_row=2, max_row=ws.max_row, min_col=1, max_col=max_col, values_only=True):
|
||
name_val = str(row[field_col_map.get("Name", 0)] or "").strip() if len(row) > field_col_map.get("Name", 0) else ""
|
||
if not name_val:
|
||
continue
|
||
for f in required_fields:
|
||
col_idx = field_col_map[f.field_name]
|
||
if col_idx < len(row):
|
||
cell_val = str(row[col_idx] or "").strip() if row[col_idx] is not None else ""
|
||
if not cell_val:
|
||
empty_by_field[f.field_name].add(name_val)
|
||
wb2.close()
|
||
|
||
for field_name, names in empty_by_field.items():
|
||
if names:
|
||
sorted_names = sorted(names)
|
||
total = len(sorted_names)
|
||
preview = sorted_names[:5]
|
||
suffix = " …" if total > 5 else ""
|
||
warnings.append(
|
||
f"Feld '{field_name}' ist leer bei {total} Person(en): "
|
||
f"{', '.join(preview)}{suffix}"
|
||
)
|
||
# ──────────────────────────────────────────────────────────────────
|
||
|
||
if sheets_with_data == 0:
|
||
errors.append("Kein Sheet mit den erwarteten Spalten gefunden.")
|
||
|
||
return ValidationResult(
|
||
is_valid=len(errors) == 0,
|
||
errors=errors,
|
||
warnings=warnings,
|
||
info=info,
|
||
)
|
||
|
||
|
||
def validate_template(
|
||
template_path: Path,
|
||
mapping: list[OutputMapping],
|
||
) -> ValidationResult:
|
||
"""
|
||
Validiert ein PDF-Template:
|
||
- Datei existiert
|
||
- Alle _Field-Platzhalter (pdf_field) sind im PDF auf irgendeiner Seite vorhanden
|
||
"""
|
||
errors: list[str] = []
|
||
warnings: list[str] = []
|
||
info: dict = {}
|
||
|
||
if not template_path.exists():
|
||
errors.append(f"Template-Datei nicht gefunden: {template_path}")
|
||
return ValidationResult(is_valid=False, errors=errors)
|
||
|
||
try:
|
||
doc = fitz.open(template_path)
|
||
except Exception as e:
|
||
errors.append(f"PDF-Template konnte nicht geöffnet werden: {e}")
|
||
return ValidationResult(is_valid=False, errors=errors)
|
||
|
||
if doc.page_count == 0:
|
||
errors.append("PDF-Template enthält keine Seiten.")
|
||
doc.close()
|
||
return ValidationResult(is_valid=False, errors=errors)
|
||
|
||
found_count = 0
|
||
missing_fields = []
|
||
|
||
for m in mapping:
|
||
if m.pdf_field is None:
|
||
continue # Kein Marker erwartet
|
||
|
||
# Suche auf allen Seiten
|
||
found_any = False
|
||
for page_num in range(doc.page_count):
|
||
found = doc[page_num].search_for(m.pdf_field)
|
||
if found:
|
||
found_any = True
|
||
break
|
||
|
||
if found_any:
|
||
found_count += 1
|
||
else:
|
||
missing_fields.append(m.source_field)
|
||
|
||
# Zähle unique pdf_fields (nicht die source_fields, da Tätigkeit doppelt vorkommt)
|
||
unique_pdf_fields = set(m.pdf_field for m in mapping if m.pdf_field is not None)
|
||
total_pages = doc.page_count
|
||
|
||
doc.close()
|
||
|
||
info["found_fields"] = found_count
|
||
info["total_fields"] = len(unique_pdf_fields)
|
||
info["page_count"] = total_pages
|
||
|
||
if missing_fields:
|
||
errors.append(
|
||
f"Platzhalter nicht im PDF-Template gefunden: {', '.join(missing_fields)}"
|
||
)
|
||
|
||
return ValidationResult(
|
||
is_valid=len(errors) == 0,
|
||
errors=errors,
|
||
warnings=warnings,
|
||
info=info,
|
||
)
|