move the whole app in subfolder
This commit is contained in:
Rob committed
1 parent
fa3e99dec0
commit
94e245a3a0
27 files changed
No files matched your search
@@ -0,0 +1 @@
|
||||
"""Tätigkeitsnachweis-Generator package."""
|
||||
@@ -0,0 +1,85 @@
|
||||
"""Tätigkeitsnachweis-Generator – Konfigurations-Loader."""
|
||||
|
||||
import configparser
|
||||
import json
|
||||
import fnmatch
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
from .models import AppConfig, FieldSchema, OutputMapping
|
||||
|
||||
|
||||
def load_settings(config_dir: Path, project_root: Path) -> AppConfig:
|
||||
"""Lädt settings.ini und gibt ein AppConfig-Objekt zurück."""
|
||||
ini_path = config_dir / "settings.ini"
|
||||
if not ini_path.exists():
|
||||
raise FileNotFoundError(f"settings.ini nicht gefunden: {ini_path}")
|
||||
|
||||
parser = configparser.ConfigParser()
|
||||
parser.read(ini_path, encoding="utf-8")
|
||||
|
||||
paths = parser["paths"] if "paths" in parser else {}
|
||||
output = parser["output"] if "output" in parser else {}
|
||||
parsing = parser["parsing"] if "parsing" in parser else {}
|
||||
font = parser["font"] if "font" in parser else {}
|
||||
|
||||
return AppConfig(
|
||||
input_dir=str(project_root / paths.get("input_dir", "input")),
|
||||
template_dir=str(project_root / paths.get("template_dir", "template")),
|
||||
output_dir=str(project_root / paths.get("output_dir", "output")),
|
||||
filename_pattern=output.get("filename_pattern", "{last_name}_{first_name}_Tätigkeitsnachweis.xlsx"),
|
||||
month_sheet_pattern=parsing.get("month_sheet_pattern", "* 20??"),
|
||||
skip_empty_name=parsing.getboolean("skip_empty_name", True),
|
||||
skip_empty_antragsnummer=parsing.getboolean("skip_empty_antragsnummer", True),
|
||||
fontsize=font.getfloat("fontsize", 11.0),
|
||||
)
|
||||
|
||||
|
||||
def load_input_schema(config_dir: Path) -> list[FieldSchema]:
|
||||
"""Lädt input-fields.json."""
|
||||
schema_path = config_dir / "input-fields.json"
|
||||
if not schema_path.exists():
|
||||
raise FileNotFoundError(f"input-fields.json nicht gefunden: {schema_path}")
|
||||
|
||||
with open(schema_path, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
|
||||
return [FieldSchema(**item) for item in data]
|
||||
|
||||
|
||||
def load_output_mapping(config_dir: Path) -> list[OutputMapping]:
|
||||
"""Lädt output-fields.json."""
|
||||
mapping_path = config_dir / "output-fields.json"
|
||||
if not mapping_path.exists():
|
||||
raise FileNotFoundError(f"output-fields.json nicht gefunden: {mapping_path}")
|
||||
|
||||
with open(mapping_path, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
|
||||
return [OutputMapping(**item) for item in data]
|
||||
|
||||
|
||||
def get_available_files(directory: Path, extension: str = ".xlsx") -> list[str]:
|
||||
"""Listet alle Dateien mit der angegebenen Erweiterung im Verzeichnis auf."""
|
||||
if not directory.exists():
|
||||
return []
|
||||
files = sorted(
|
||||
f.name for f in directory.iterdir()
|
||||
if f.is_file() and f.suffix.lower() == extension and not f.name.startswith("~")
|
||||
)
|
||||
return files
|
||||
|
||||
|
||||
def get_available_input_files(config: AppConfig) -> list[str]:
|
||||
"""Listet alle .xlsx-Dateien im input-Ordner."""
|
||||
return get_available_files(Path(config.input_dir))
|
||||
|
||||
|
||||
def get_available_templates(config: AppConfig) -> list[str]:
|
||||
"""Listet alle .pdf-Dateien im template-Ordner."""
|
||||
return get_available_files(Path(config.template_dir), extension=".pdf")
|
||||
|
||||
|
||||
def is_month_sheet(sheet_name: str, pattern: str) -> bool:
|
||||
"""Prüft, ob ein Sheet-Name dem Monatsmuster entspricht."""
|
||||
return fnmatch.fnmatch(sheet_name.strip(), pattern)
|
||||
@@ -0,0 +1,75 @@
|
||||
"""Tätigkeitsnachweis-Generator – Datenmodelle."""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class EmployeeRecord:
|
||||
"""Eine Zeile aus einem Monats-Sheet der Input-Datei."""
|
||||
name: str
|
||||
antragsnummer: str
|
||||
qualifikation: str
|
||||
funktion: str
|
||||
stellenanteil: str
|
||||
monat: str # z. B. "01.2025"
|
||||
taetigkeit: str
|
||||
id: Optional[int] = None
|
||||
kommentar: Optional[str] = None
|
||||
|
||||
@property
|
||||
def month_index(self) -> int:
|
||||
"""Extrahiere den Monatsindex (1–12) aus dem Monatsfeld."""
|
||||
try:
|
||||
return int(self.monat.split(".")[0])
|
||||
except (ValueError, IndexError):
|
||||
return 0
|
||||
|
||||
@property
|
||||
def year(self) -> int:
|
||||
"""Extrahiere das Jahr aus dem Monatsfeld."""
|
||||
try:
|
||||
return int(self.monat.split(".")[1])
|
||||
except (ValueError, IndexError):
|
||||
return 0
|
||||
|
||||
|
||||
@dataclass
|
||||
class FieldSchema:
|
||||
"""Definition eines Feldes aus input-fields.json."""
|
||||
column: str
|
||||
field_name: str
|
||||
type: str
|
||||
required: bool
|
||||
|
||||
|
||||
@dataclass
|
||||
class OutputMapping:
|
||||
"""Definition eines Mappings aus output-fields.json."""
|
||||
source_field: str
|
||||
format: Optional[str] = None
|
||||
pdf_field: Optional[str] = None # Name des _Field-Platzhalters im PDF (z.B. "Auftragsnummer_Field")
|
||||
|
||||
|
||||
@dataclass
|
||||
class AppConfig:
|
||||
"""Gesamte App-Konfiguration."""
|
||||
input_dir: str
|
||||
template_dir: str
|
||||
output_dir: str
|
||||
filename_pattern: str
|
||||
input_schema: list = field(default_factory=list)
|
||||
output_mapping: list = field(default_factory=list)
|
||||
month_sheet_pattern: str = "* 20??"
|
||||
skip_empty_name: bool = True
|
||||
skip_empty_antragsnummer: bool = True
|
||||
fontsize: float = 11.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class ValidationResult:
|
||||
"""Ergebnis einer Validierung."""
|
||||
is_valid: bool
|
||||
errors: list = field(default_factory=list)
|
||||
warnings: list = field(default_factory=list)
|
||||
info: dict = field(default_factory=dict)
|
||||
@@ -0,0 +1,134 @@
|
||||
"""Input-Datei-Parser: liest monatliche Sheets und erzeugt EmployeeRecords."""
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional
|
||||
|
||||
from openpyxl import load_workbook
|
||||
from openpyxl.utils import column_index_from_string
|
||||
|
||||
from .config import is_month_sheet
|
||||
from .models import EmployeeRecord, FieldSchema
|
||||
|
||||
|
||||
def _to_text(value: Any) -> str:
|
||||
if value is None:
|
||||
return ""
|
||||
return str(value).strip()
|
||||
|
||||
|
||||
def _to_int_or_none(value: Any) -> Optional[int]:
|
||||
if value is None or str(value).strip() == "":
|
||||
return None
|
||||
try:
|
||||
return int(value)
|
||||
except (ValueError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
def _to_percent_string(value: Any) -> str:
|
||||
"""Wandelt einen Prozentwert in einen String ohne %-Zeichen um.
|
||||
|
||||
Excel speichert Prozentwerte als Dezimalbruch (50 % = 0.5).
|
||||
Falls der Rohwert eine Zahl ≤ 1 ist, wird er als Excel-Prozentwert
|
||||
interpretiert und mit 100 multipliziert.
|
||||
"""
|
||||
if value is None or str(value).strip() == "":
|
||||
return ""
|
||||
# Falls es bereits ein String mit %-Zeichen ist
|
||||
if isinstance(value, str):
|
||||
return value.strip().rstrip("% ")
|
||||
# Numerischer Wert: Excel-Prozentwert (0.5 = 50 %) oder Ganzzahl (50 = 50 %)
|
||||
if isinstance(value, (int, float)):
|
||||
if 0 < value <= 1:
|
||||
return str(int(value * 100))
|
||||
if value == 0:
|
||||
return "0"
|
||||
# Wert > 1: bereits als Ganzzahl gemeint (z. B. 50)
|
||||
return str(int(value))
|
||||
return str(value).strip()
|
||||
|
||||
|
||||
def _build_column_map(schema: list[FieldSchema]) -> dict[str, int]:
|
||||
"""Baut ein Mapping von field_name → 0-basierter Spaltenindex."""
|
||||
col_map: dict[str, int] = {}
|
||||
for field in schema:
|
||||
col_map[field.field_name] = column_index_from_string(field.column) - 1
|
||||
return col_map
|
||||
|
||||
|
||||
def parse_input_file(
|
||||
filepath: Path,
|
||||
schema: list[FieldSchema],
|
||||
month_sheet_pattern: str = "* 20??",
|
||||
skip_empty_name: bool = True,
|
||||
skip_empty_antragsnummer: bool = True,
|
||||
) -> list[EmployeeRecord]:
|
||||
"""
|
||||
Öffnet die Input-Datei, iteriert über alle Monats-Sheets
|
||||
und gibt eine Liste von EmployeeRecords zurück.
|
||||
"""
|
||||
wb = load_workbook(filepath, data_only=True)
|
||||
col_map = _build_column_map(schema)
|
||||
records: list[EmployeeRecord] = []
|
||||
|
||||
for sheet_name in wb.sheetnames:
|
||||
if not is_month_sheet(sheet_name, month_sheet_pattern):
|
||||
continue
|
||||
|
||||
ws = wb[sheet_name]
|
||||
# Header ist Zeile 1, Daten ab Zeile 2
|
||||
# min_col=1 explizit setzen für korrekte Spaltenzuordnung
|
||||
schema_col_count = len(schema)
|
||||
for row in ws.iter_rows(min_row=2, max_row=ws.max_row, min_col=1, max_col=schema_col_count, values_only=True):
|
||||
name = _to_text(row[col_map.get("Name", 0)]) if len(row) > col_map.get("Name", 0) else ""
|
||||
antragsnummer = _to_text(row[col_map.get("Antragsnummer", 0)]) if len(row) > col_map.get("Antragsnummer", 0) else ""
|
||||
|
||||
if skip_empty_name and not name:
|
||||
continue
|
||||
if skip_empty_antragsnummer and not antragsnummer:
|
||||
continue
|
||||
|
||||
qualifikation = _to_text(row[col_map.get("Qualifikation", 0)]) if len(row) > col_map.get("Qualifikation", 0) else ""
|
||||
funktion = _to_text(row[col_map.get("Funktion", 0)]) if len(row) > col_map.get("Funktion", 0) else ""
|
||||
stellenanteil_raw = row[col_map.get("Stellenanteil in %", 0)] if len(row) > col_map.get("Stellenanteil in %", 0) else None
|
||||
stellenanteil = _to_percent_string(stellenanteil_raw)
|
||||
monat = _to_text(row[col_map.get("Monat (MM.JJ)", 0)]) if len(row) > col_map.get("Monat (MM.JJ)", 0) else ""
|
||||
taetigkeit = _to_text(row[col_map.get("Tätigkeit", 0)]) if len(row) > col_map.get("Tätigkeit", 0) else ""
|
||||
record_id = _to_int_or_none(row[col_map.get("ID", 0)]) if len(row) > col_map.get("ID", 0) else None
|
||||
kommentar = _to_text(row[col_map.get("Kommentar", 1)]) if "Kommentar" in col_map and len(row) > col_map["Kommentar"] else None
|
||||
|
||||
records.append(EmployeeRecord(
|
||||
id=record_id,
|
||||
name=name,
|
||||
antragsnummer=antragsnummer,
|
||||
qualifikation=qualifikation,
|
||||
funktion=funktion,
|
||||
stellenanteil=stellenanteil,
|
||||
monat=monat,
|
||||
taetigkeit=taetigkeit,
|
||||
kommentar=kommentar,
|
||||
))
|
||||
|
||||
wb.close()
|
||||
return records
|
||||
|
||||
|
||||
def group_by_employee(records: list[EmployeeRecord]) -> dict[str, list[EmployeeRecord]]:
|
||||
"""Gruppiert Records nach Mitarbeitername."""
|
||||
groups: dict[str, list[EmployeeRecord]] = {}
|
||||
for record in records:
|
||||
if record.name not in groups:
|
||||
groups[record.name] = []
|
||||
groups[record.name].append(record)
|
||||
return groups
|
||||
|
||||
|
||||
def get_unique_antragsnummern(records: list[EmployeeRecord]) -> list[str]:
|
||||
"""Gibt eine sortierte Liste aller eindeutigen Antragsnummern zurück."""
|
||||
return sorted(set(r.antragsnummer for r in records if r.antragsnummer))
|
||||
|
||||
|
||||
def get_unique_employees(records: list[EmployeeRecord]) -> list[str]:
|
||||
"""Gibt eine sortierte Liste aller eindeutigen Mitarbeiternamen zurück."""
|
||||
return sorted(set(r.name for r in records if r.name))
|
||||
@@ -0,0 +1,206 @@
|
||||
"""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,
|
||||
)
|
||||
@@ -0,0 +1,289 @@
|
||||
"""PDF-basierte Activity-Report-Generierung.
|
||||
|
||||
Verwendet ein mehrseitiges PDF-Template mit _Field-Platzhaltern.
|
||||
Für jeden Mitarbeiter wird ein PDF erzeugt (alle Template-Seiten pro Monat).
|
||||
Die _Field-Marker werden per PyMuPDF (fitz) gesucht und durch echte Daten ersetzt.
|
||||
"""
|
||||
|
||||
import calendar
|
||||
from pathlib import Path
|
||||
|
||||
import fitz # PyMuPDF
|
||||
|
||||
from .models import EmployeeRecord, OutputMapping
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Namens-Helfer
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def _split_name_for_filename(name: str) -> tuple[str, str]:
|
||||
"""Teilt 'Nachname, Vorname' oder 'Vorname Nachname' in (Vorname, Nachname)."""
|
||||
cleaned = str(name or "").strip()
|
||||
if "," in cleaned:
|
||||
last_name, first_name = [p.strip() for p in cleaned.split(",", 1)]
|
||||
return first_name or "Unknown", last_name or "Unknown"
|
||||
parts = cleaned.split()
|
||||
if len(parts) >= 2:
|
||||
return parts[0], parts[-1]
|
||||
if len(parts) == 1:
|
||||
return parts[0], "Unknown"
|
||||
return "Unknown", "Unknown"
|
||||
|
||||
|
||||
def _format_value(value, fmt: str | None) -> str:
|
||||
"""Formatiert einen Wert gemäß Format-String."""
|
||||
if value is None:
|
||||
return ""
|
||||
if fmt is None:
|
||||
return str(value)
|
||||
if "%" in fmt:
|
||||
try:
|
||||
return fmt % float(value)
|
||||
except (ValueError, TypeError):
|
||||
return str(value)
|
||||
return str(value)
|
||||
|
||||
|
||||
def _format_month_for_template(monat_str: str) -> str:
|
||||
"""Konvertiert 'MM.JJ' zu 'MM.JJJJ'."""
|
||||
monat_str = monat_str.strip()
|
||||
parts = monat_str.split(".")
|
||||
if len(parts) == 2:
|
||||
month = parts[0]
|
||||
year = parts[1]
|
||||
if len(year) == 2:
|
||||
year = "20" + year
|
||||
return f"{month}.{year}"
|
||||
return monat_str
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Daten-Extraktion aus EmployeeRecord
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def _get_field_value(record: EmployeeRecord, employee_name: str, source_field: str) -> str:
|
||||
"""Liefert den formatierten Wert für ein Feld im PDF."""
|
||||
if source_field == "Monat (MM.JJ)":
|
||||
return _format_month_for_template(record.monat)
|
||||
elif source_field == "Stellenanteil in %":
|
||||
try:
|
||||
value = int(float(str(record.stellenanteil).strip()))
|
||||
return f"{value},00"
|
||||
except (ValueError, TypeError):
|
||||
return str(record.stellenanteil)
|
||||
elif source_field == "Name":
|
||||
return employee_name
|
||||
elif source_field == "Antragsnummer":
|
||||
return record.antragsnummer
|
||||
elif source_field == "Qualifikation":
|
||||
return record.qualifikation
|
||||
elif source_field == "Funktion":
|
||||
return record.funktion
|
||||
elif source_field == "Tätigkeit":
|
||||
return record.taetigkeit
|
||||
elif source_field in ("Ort", "Ort 2"):
|
||||
return "Dresden"
|
||||
elif source_field == "Datum":
|
||||
parts = record.monat.split(".")
|
||||
if len(parts) == 2:
|
||||
try:
|
||||
month = int(parts[0])
|
||||
year = int(parts[1])
|
||||
last_day = calendar.monthrange(year, month)[1]
|
||||
return f"{last_day:02d}.{month:02d}.{year}"
|
||||
except (ValueError, IndexError):
|
||||
return ""
|
||||
return ""
|
||||
return ""
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# PDF-Template Pre-Scan: _Field-Positionen auf ALLEN Seiten ermitteln
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def _scan_template_fields(
|
||||
template_path: Path,
|
||||
mapping: list[OutputMapping],
|
||||
fontsize: float,
|
||||
) -> dict[str, list[dict]]:
|
||||
"""
|
||||
Scannt ALLE Seiten des PDF-Templates nach _Field-Platzhaltern.
|
||||
Rückgabe: dict[pdf_field] → Liste von {source_field, page_num, rect, fontsize, fontname, ...}
|
||||
Felder mit pdf_field=None werden übersprungen.
|
||||
"""
|
||||
doc = fitz.open(template_path)
|
||||
|
||||
# Sammle alle pdf_field → source_field Zuordnungen
|
||||
pdf_to_source: dict[str, str] = {}
|
||||
for m in mapping:
|
||||
if m.pdf_field is not None:
|
||||
pdf_to_source[m.pdf_field] = m.source_field
|
||||
|
||||
field_info: dict[str, list[dict]] = {}
|
||||
|
||||
# Alle Vorkommen aller PDF-Feld-Marker finden (alle Seiten)
|
||||
for pdf_field, source_field in pdf_to_source.items():
|
||||
occurrences: list[dict] = []
|
||||
for page_num in range(doc.page_count):
|
||||
page = doc[page_num]
|
||||
found = page.search_for(pdf_field)
|
||||
for rect in found:
|
||||
occurrences.append({
|
||||
"source_field": source_field,
|
||||
"page_num": page_num,
|
||||
"rect": rect,
|
||||
"fontsize": fontsize,
|
||||
"fontname": "helv",
|
||||
})
|
||||
field_info[pdf_field] = occurrences
|
||||
|
||||
doc.close()
|
||||
return field_info
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Template-Seiten mit Daten befüllen
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def _fill_pdf_pages(
|
||||
pages: list[fitz.Page], # eine Page pro Template-Seite
|
||||
record: EmployeeRecord,
|
||||
employee_name: str,
|
||||
mapping: list[OutputMapping],
|
||||
field_info: dict[str, list[dict]],
|
||||
fontsize: float,
|
||||
) -> None:
|
||||
"""Befüllt alle Template-Seiten: redacted _Field-Platzhalter und fügt echte Daten ein."""
|
||||
# Mehrzeilige Felder, die als Textbox (mit Zeilenumbruch) behandelt werden
|
||||
MULTILINE_FIELDS = {"Tätigkeiten_Field", "Dienstreisen_Field"}
|
||||
TEXTBOX_WIDTH_PT = 400 # Breite der Textbox in Punkt
|
||||
|
||||
for m in mapping:
|
||||
if m.pdf_field is None:
|
||||
continue
|
||||
|
||||
occurrences = field_info.get(m.pdf_field, [])
|
||||
if not occurrences:
|
||||
continue # Feld nicht im Template gefunden
|
||||
|
||||
value = _get_field_value(record, employee_name, m.source_field)
|
||||
formatted = _format_value(value, m.format)
|
||||
|
||||
# Jedes Vorkommen des Feldes im PDF befüllen
|
||||
for info in occurrences:
|
||||
page_num = info["page_num"]
|
||||
if page_num >= len(pages):
|
||||
continue
|
||||
|
||||
page = pages[page_num]
|
||||
rect = info["rect"]
|
||||
|
||||
# Platzhalter redacten
|
||||
page.add_redact_annot(rect)
|
||||
page.apply_redactions()
|
||||
|
||||
if not formatted:
|
||||
continue
|
||||
|
||||
# Mehrzeilige Felder: Zeilenweise per insert_text schreiben
|
||||
if m.pdf_field in MULTILINE_FIELDS:
|
||||
# ... (unverändert)
|
||||
clean_text = "\n".join(
|
||||
line for line in formatted.split("\n") if line.strip()
|
||||
)
|
||||
if not clean_text:
|
||||
continue
|
||||
|
||||
lines = clean_text.split("\n")
|
||||
line_spacing = fontsize * 1.2
|
||||
|
||||
output_lines: list[str] = []
|
||||
for line in lines:
|
||||
max_width = TEXTBOX_WIDTH_PT
|
||||
words = line.split(" ")
|
||||
current_line = ""
|
||||
for word in words:
|
||||
test_line = current_line + (" " if current_line else "") + word
|
||||
text_width = fitz.get_text_length(test_line, fontname="helv", fontsize=fontsize)
|
||||
if text_width <= max_width or not current_line:
|
||||
current_line = test_line
|
||||
else:
|
||||
output_lines.append(current_line)
|
||||
current_line = word
|
||||
if current_line:
|
||||
output_lines.append(current_line)
|
||||
|
||||
y = rect.y0 + fontsize
|
||||
for current_line in output_lines:
|
||||
page.insert_text(
|
||||
(rect.x0, y), current_line,
|
||||
fontname="helv", fontsize=fontsize,
|
||||
)
|
||||
y += line_spacing
|
||||
else:
|
||||
# Einzeilige Felder: vertikal zentriert platzieren
|
||||
mid_y = (rect.y0 + rect.y1) / 2
|
||||
baseline_y = mid_y + fontsize * 0.35
|
||||
page.insert_text(
|
||||
(rect.x0, baseline_y),
|
||||
formatted,
|
||||
fontname="helv",
|
||||
fontsize=fontsize,
|
||||
)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────
|
||||
# Hauptfunktion: Reports generieren
|
||||
# ─────────────────────────────────────────────
|
||||
|
||||
def generate_activity_reports(
|
||||
employee_records: dict[str, list[EmployeeRecord]],
|
||||
template_path: Path,
|
||||
mapping: list[OutputMapping],
|
||||
output_dir: Path,
|
||||
filename_pattern: str,
|
||||
fontsize: float = 11.0,
|
||||
) -> list[Path]:
|
||||
"""
|
||||
Erzeugt für jeden Mitarbeiter ein PDF.
|
||||
Pro Monat werden ALLE Template-Seiten kopiert und mit Daten befüllt.
|
||||
"""
|
||||
output_dir.mkdir(parents=True, exist_ok=True)
|
||||
generated_files: list[Path] = []
|
||||
|
||||
# Pre-Scan: Feld-Positionen auf allen Template-Seiten ermitteln
|
||||
field_info = _scan_template_fields(template_path, mapping, fontsize)
|
||||
|
||||
for employee_name, records in employee_records.items():
|
||||
if not records:
|
||||
continue
|
||||
|
||||
first_name, last_name = _split_name_for_filename(employee_name)
|
||||
safe_first = first_name.replace(" ", "_").replace("/", "_")
|
||||
safe_last = last_name.replace(" ", "_").replace("/", "_")
|
||||
|
||||
filename = filename_pattern.format(first_name=safe_first, last_name=safe_last)
|
||||
output_path = output_dir / filename
|
||||
|
||||
sorted_records = sorted(records, key=lambda r: r.month_index)
|
||||
|
||||
# Ergebnis-PDF erstellen
|
||||
result_doc = fitz.open()
|
||||
|
||||
for record in sorted_records:
|
||||
# Template frisch öffnen für jeden Monat
|
||||
temp_doc = fitz.open(template_path)
|
||||
pages = [temp_doc[i] for i in range(temp_doc.page_count)]
|
||||
|
||||
_fill_pdf_pages(pages, record, employee_name, mapping, field_info, fontsize)
|
||||
|
||||
# Alle modifizierten Seiten ins Ergebnis-PDF einfügen
|
||||
result_doc.insert_pdf(temp_doc)
|
||||
temp_doc.close()
|
||||
|
||||
result_doc.save(output_path)
|
||||
result_doc.close()
|
||||
generated_files.append(output_path)
|
||||
|
||||
return generated_files
|
||||
Reference in new issue
Block a user