10 Commits
9 changed files with 802 additions and 302 deletions

No files matched your search

+31 -24
View File
@@ -59,37 +59,44 @@ Projekt vollstaendig umsetzen: aus PS_PW.xlsx und Stellenbeschreibungen.xlsx mon
- [x] Monatsmarker parsebar
- [x] Gewuenschter Monat exakt einmal vorhanden
- [x] Stunden-/Vertragswerte gueltig
- [ ] T10: Business-Validierung implementieren
- [ ] Mindestens ein aktiver Fall vorhanden
- [ ] Jeder aktive Fall hat eindeutigen Match oder klaren Fehler
- [ ] Template enthaelt genau ein Quellblatt
- [ ] Dateinamenkollisionen vermeiden
- [ ] Pflicht-Zielzellen vollstaendig konfiguriert
- [x] T10: Business-Validierung implementieren
- [x] Mindestens ein aktiver Fall vorhanden
- [x] Jeder aktive Fall hat eindeutigen Match oder klaren Fehler
- [x] Template enthaelt genau ein Quellblatt
- [x] Dateinamenkollisionen vermeiden
- [x] Pflicht-Zielzellen vollstaendig konfiguriert
## Phase 6 - Ausgabeerzeugung
- [ ] T11: Template-Verarbeitung bauen
- [ ] Template-Workbook laden
- [ ] Sheet pro aktiver Zuordnung klonen
- [ ] Inhalte assignment-spezifisch in Zielzellen schreiben
- [ ] T12: Dateinamen und Speichern
- [ ] Muster: `Taetigkeitsnachweis_YYYY_MM_Vorname_Nachname.xlsx`
- [ ] Pro Mitarbeiter genau eine Datei pro Monat erzeugen
- [ ] Bei mehreren Zuordnungen: mehrere Sheets in derselben Datei
- [x] T11: Template-Verarbeitung bauen
- [x] Template-Workbook laden
- [x] Sheet pro aktiver Zuordnung klonen
- [x] Inhalte assignment-spezifisch in Zielzellen schreiben
- [x] T12: Dateinamen und Speichern
- [x] Muster: `Taetigkeitsnachweis_YYYY_MM_Vorname_Nachname.xlsx`
- [x] Pro Mitarbeiter genau eine Datei pro Monat erzeugen
- [x] Bei mehreren Zuordnungen: mehrere Sheets in derselben Datei
## Phase 7 - CLI und Benutzerfluss
- [ ] T13: CLI in `generate.py` implementieren
- [ ] Parameter `--month` und `--year`
- [ ] Pipeline orchestrieren: parse -> match -> validate -> preview -> write
- [ ] Kontrollierter Exit-Code bei Fehlern
- [ ] T14: Vorschau + Bestaetigung vor Schreiben
- [ ] Uebersicht zu Berichtsmonat, aktiven Faellen, Fehlern
- [ ] Liste der geplanten Ausgabedateien
- [ ] Interaktive Bestaetigung (Abbruch ohne Schreibzugriff moeglich)
- [x] T13: CLI in `generate.py` implementieren
- [x] Parameter `--month` und `--year`
- [x] Pipeline orchestrieren: parse -> match -> validate -> preview -> write
- [x] Kontrollierter Exit-Code bei Fehlern
- [x] T14: Vorschau + Bestaetigung vor Schreiben
- [x] Uebersicht zu Berichtsmonat, aktiven Faellen, Fehlern
- [x] Liste der geplanten Ausgabedateien
- [x] Interaktive Bestaetigung (Abbruch ohne Schreibzugriff moeglich)
## Phase 8 - Tests und Abnahme
- [ ] T15: Smoke-Test fuer End-to-End-Lauf
- [ ] Beispielmonat ausfuehren
- [x] Beispielmonat ausfuehren
- [x] Pipeline laeuft bis Dateiausgabe durch (0 Dateien, da 0 gematchte Faelle)
- [ ] Dateien in `output/` entstehen erwartungsgemaess
- [ ] DATENBLOCKER: `Stellenbeschreibungen.xlsx` enthaelt nur Meyer,Johannes (25%/75%).
Alle anderen aktiven Mitarbeiter aus PS_PW.xlsx fehlen als Stammdaten.
Meyer's Foerderquote (85%/15%) stimmt nicht mit eingetragenem Stellenanteil (25%/75%) ueberein.
-> Stellenbeschreibungen.xlsx mit allen aktiven Mitarbeitern und korrekten Anteilen befuellen.
- [x] Template-Zielzellen konfiguriert (A3, A17, Q17, A21, Q21, A25, A31, I95)
- [x] Template-Sheet-Pfad konfiguriert (<MM_YYYY>_<Stellenbeschreibung>)
- [ ] T16: Fachtests definieren und ausfuehren
- [ ] Mehrfachzuordnung eines Mitarbeiters
- [ ] Fehlender Match
@@ -111,4 +118,4 @@ Projekt vollstaendig umsetzen: aus PS_PW.xlsx und Stellenbeschreibungen.xlsx mon
8. T15 -> T16 -> T17
## Aktueller Startpunkt
- [ ] Naechste aktive Aufgabe: T10 (Business-Validierung implementieren)
- [ ] Naechste aktive Aufgabe: T15 (Smoke-Test fuer End-to-End-Lauf)
+1
View File
@@ -0,0 +1 @@
__pycache__
+13 -12
View File
@@ -12,7 +12,7 @@
"sheets": {
"allocation": "Tabelle1",
"masterdata": "Tabelle1",
"template_source": "Tabelle1"
"template_source": "<MM_YYYY>_<Stellenbeschreibung>"
},
"allocation": {
"header_month_marker_row": 2,
@@ -41,9 +41,10 @@
"antragsnummer": "B",
"name": "C",
"qualifikation": "D",
"stellenanteil": "E",
"funktion": "F",
"taetigkeit": "G"
"project_scope": "E",
"stellenanteil": "F",
"funktion": "G",
"taetigkeit": "H"
},
"decimal": {
"decimal_separator": ",",
@@ -66,14 +67,14 @@
"sheet_name_pattern": "{index}_{funktion}"
},
"template_targets": {
"antragsnummer": "",
"name_vorname": "",
"qualifikation": "",
"funktion": "",
"taetigkeiten": "",
"reporting_month_label": "",
"foerderquote": "",
"generated_on": ""
"antragsnummer": "A3",
"name_vorname": "A17",
"qualifikation": "A21",
"funktion": "A25",
"taetigkeiten": "A31",
"reporting_month_label": "Q21",
"foerderquote": "Q17",
"generated_on": "I95"
},
"runtime": {
"abort_on_validation_errors": true,
+158
View File
@@ -0,0 +1,158 @@
from __future__ import annotations
import json
from pathlib import Path
import click
from src.taetigkeitsnachweis.errors import TaetigkeitsnachweisError
from src.taetigkeitsnachweis.logging_utils import configure_logging
from src.taetigkeitsnachweis.matcher import match_assignments
from src.taetigkeitsnachweis.parser import parse_allocation_records, parse_masterdata_records
from src.taetigkeitsnachweis.validator import validate_business_rules, validate_sources
from src.taetigkeitsnachweis.writer import (
build_employee_workbooks,
build_output_filename,
save_employee_workbooks,
)
def _load_config(project_root: Path) -> dict:
config_path = project_root / "config.json"
with config_path.open("r", encoding="utf-8") as config_file:
return json.load(config_file)
def _print_issues(issues: list, title: str) -> None:
if not issues:
return
click.echo(title)
for issue in issues:
click.echo(f"- [{issue.severity}] {issue.code}: {issue.message}")
def _print_preview(month: int, year: int, matched_assignments: list, planned_files: list[str]) -> None:
click.echo("=== Vorschau der ermittelten Werte ===")
click.echo(f"Berichtsmonat : {month:02d}/{year}")
click.echo(f"Aktive Faelle : {len(matched_assignments)}")
click.echo("--- Zu erzeugende Taetigkeitsnachweise ---")
for item in matched_assignments:
click.echo(
"- "
f"{item.allocation.employee_name} | "
f"{item.masterdata.funktion} | "
f"{item.foerderquote_percent:.2f}%"
)
click.echo("--- Geplante Ausgabedateien ---")
for file_name in planned_files:
click.echo(f"- {file_name}")
@click.command()
@click.option("--month", type=click.IntRange(1, 12), required=True)
@click.option("--year", type=click.IntRange(2000, 2100), required=True)
@click.option("--yes", is_flag=True, default=False, help="Skip interactive confirmation.")
@click.option("--skip-errors", is_flag=True, default=False, help="Generate files for matched records, skip unresolved ones.")
def main(month: int, year: int, yes: bool, skip_errors: bool) -> None:
configure_logging()
project_root = Path(__file__).resolve().parent
try:
config = _load_config(project_root)
source_issues = validate_sources(
config=config,
report_month=month,
report_year=year,
project_root=project_root,
)
_print_issues(source_issues, "=== Source validation ===")
if any(issue.severity == "error" for issue in source_issues):
raise click.ClickException("Source validation failed.")
allocation_records = parse_allocation_records(
config=config,
report_month=month,
report_year=year,
project_root=project_root,
)
masterdata_records = parse_masterdata_records(
config=config,
project_root=project_root,
)
matched_assignments, match_issues = match_assignments(
allocation_records=allocation_records,
masterdata_records=masterdata_records,
config=config,
)
_print_issues(match_issues, "=== Matching issues ===")
business_issues = validate_business_rules(
config=config,
report_month=month,
report_year=year,
allocation_records=allocation_records,
matched_assignments=matched_assignments,
project_root=project_root,
)
_print_issues(business_issues, "=== Business validation ===")
all_issues = [*match_issues, *business_issues]
has_errors = any(issue.severity == "error" for issue in all_issues)
if has_errors and not skip_errors:
raise click.ClickException("Validation failed after matching.")
if has_errors and skip_errors:
click.echo("Warnhinweis: Fehler vorhanden. Weiterfahrt mit --skip-errors (nur gematchte Faelle werden erzeugt).")
filename_pattern = config["output"]["filename_pattern"]
unique_employees = sorted({item.allocation.employee_name for item in matched_assignments})
planned_files = [
build_output_filename(
filename_pattern=filename_pattern,
employee_name=employee_name,
report_year=year,
report_month=month,
)
for employee_name in unique_employees
]
_print_preview(
month=month,
year=year,
matched_assignments=matched_assignments,
planned_files=planned_files,
)
if not yes:
confirmed = click.confirm("Dateien jetzt schreiben?", default=False)
if not confirmed:
click.echo("Abbruch ohne Schreibzugriff.")
return
employee_workbooks = build_employee_workbooks(
config=config,
matched_assignments=matched_assignments,
report_month=month,
report_year=year,
project_root=project_root,
)
output_paths = save_employee_workbooks(
config=config,
employee_workbooks=employee_workbooks,
report_month=month,
report_year=year,
project_root=project_root,
)
click.echo("=== Generated files ===")
for output_path in output_paths:
click.echo(f"- {output_path}")
except TaetigkeitsnachweisError as exc:
raise click.ClickException(str(exc)) from exc
if __name__ == "__main__":
main()
+77 -60
View File
@@ -1,8 +1,9 @@
"""Record matching logic across allocation and master data."""
"""Record matching logic across allocation and master data."""
from __future__ import annotations
import re
import unicodedata
from typing import Any
from .errors import ValidationError
@@ -14,22 +15,55 @@ def calculate_foerderquote_percent(allocation: AllocationRecord, precision: int
raise ValidationError(
"contract_hours_weekly must be greater than 0 for foerderquote calculation."
)
raw_share = (allocation.funded_hours_weekly / allocation.contract_hours_weekly) * 100.0
return round(raw_share, precision)
def _normalize_name(name: str) -> str:
compact = re.sub(r"\s+", " ", name.strip())
return compact.casefold()
no_accents = "".join(
ch for ch in unicodedata.normalize("NFKD", compact) if not unicodedata.combining(ch)
)
return no_accents.casefold()
def _with_tolerance(
value: float,
target: float,
tolerance: float,
) -> bool:
return abs(value - target) <= tolerance
def _name_keys(name: str) -> set[str]:
normalized = _normalize_name(name)
keys = {normalized}
if "," in normalized:
last_name, first_name = [part.strip() for part in normalized.split(",", 1)]
if first_name and last_name:
keys.add(f"{first_name} {last_name}")
keys.add(f"{last_name}, {first_name}")
else:
tokens = normalized.split()
if len(tokens) >= 2:
first_name = tokens[0]
last_name = tokens[-1]
keys.add(f"{first_name} {last_name}")
keys.add(f"{last_name}, {first_name}")
return {re.sub(r"\s+", " ", key).strip() for key in keys if key.strip()}
def _disambiguate_by_project_key(
candidates: list[MasterDataRecord],
project_key: str,
) -> MasterDataRecord | None:
"""Return the single candidate matching allocation project key via masterdata project_scope."""
parts = [
token.strip().casefold()
for token in re.split(r"[/,;|]", project_key)
if token.strip()
]
project_tokens = set(parts or [project_key.strip().casefold()])
matched = []
for candidate in candidates:
candidate_scope = (candidate.project_scope or "PS").strip().casefold()
if candidate_scope in project_tokens:
matched.append(candidate)
return matched[0] if len(matched) == 1 else None
def match_assignments(
@@ -37,20 +71,33 @@ def match_assignments(
masterdata_records: list[MasterDataRecord],
config: dict[str, Any],
) -> tuple[list[MatchedAssignment], list[ValidationIssue]]:
matching_cfg = config.get("matching", {})
tolerance = float(matching_cfg.get("share_tolerance_percent", 0.5))
prefer_function_hint = bool(matching_cfg.get("prefer_function_hint", True))
# Build lookup by all normalized name variants
master_by_name: dict[str, list[MasterDataRecord]] = {}
for record in masterdata_records:
master_by_name.setdefault(_normalize_name(record.name), []).append(record)
for key in _name_keys(record.name):
master_by_name.setdefault(key, []).append(record)
# Deduplicate so the same record does not appear twice for one name key
for key in master_by_name:
seen_ids: set[str] = set()
deduped: list[MasterDataRecord] = []
for rec in master_by_name[key]:
if rec.record_id not in seen_ids:
seen_ids.add(rec.record_id)
deduped.append(rec)
master_by_name[key] = deduped
matched: list[MatchedAssignment] = []
issues: list[ValidationIssue] = []
for allocation in allocation_records:
normalized_name = _normalize_name(allocation.employee_name)
candidates = master_by_name.get(normalized_name, [])
# Collect all master-data candidates for this employee name
candidate_map: dict[str, MasterDataRecord] = {}
for key in _name_keys(allocation.employee_name):
for candidate in master_by_name.get(key, []):
candidate_map[candidate.record_id] = candidate
candidates = list(candidate_map.values())
if not candidates:
issues.append(
ValidationIssue(
@@ -65,66 +112,36 @@ def match_assignments(
)
continue
foerderquote = calculate_foerderquote_percent(allocation)
share_matches = [
candidate
for candidate in candidates
if _with_tolerance(candidate.stellenanteil_percent, foerderquote, tolerance)
]
if len(share_matches) == 1:
selected = share_matches[0]
elif len(share_matches) > 1 and prefer_function_hint:
key = allocation.project_key.casefold()
hinted = [c for c in share_matches if key in c.funktion.casefold()]
if len(hinted) == 1:
selected = hinted[0]
else:
# Single candidate: use directly
if len(candidates) == 1:
selected = candidates[0]
else:
# Multiple candidates: disambiguate by explicit project scope column.
selected = _disambiguate_by_project_key(candidates, allocation.project_key)
if selected is None:
issues.append(
ValidationIssue(
code="AMBIGUOUS_MASTERDATA",
message=(
f"Ambiguous match for employee '{allocation.employee_name}' "
f"(project '{allocation.project_key}', foerderquote {foerderquote:.2f}%)."
f"Ambiguous master-data for '{allocation.employee_name}' / "
f"'{allocation.project_key}': {len(candidates)} rows, "
f"none or multiple match project_scope. "
f"project_scope values: {[c.project_scope for c in candidates]}"
),
severity="error",
context=allocation.employee_name,
)
)
continue
elif len(share_matches) > 1:
issues.append(
ValidationIssue(
code="AMBIGUOUS_MASTERDATA",
message=(
f"Ambiguous match for employee '{allocation.employee_name}' "
f"(project '{allocation.project_key}', foerderquote {foerderquote:.2f}%)."
),
severity="error",
context=allocation.employee_name,
)
)
continue
else:
issues.append(
ValidationIssue(
code="NO_SHARE_MATCH",
message=(
f"No Stellenanteil match for employee '{allocation.employee_name}' "
f"with foerderquote {foerderquote:.2f}% (project '{allocation.project_key}')."
),
severity="error",
context=allocation.employee_name,
)
)
continue
foerderquote = calculate_foerderquote_percent(allocation)
if allocation.notes:
issues.append(
ValidationIssue(
code="ALLOCATION_NOTE",
message=(
f"Allocation note for '{allocation.employee_name}' / "
f"Note for '{allocation.employee_name}' / "
f"'{allocation.project_key}': {allocation.notes}"
),
severity="warning",
+1
View File
@@ -19,6 +19,7 @@ class MasterDataRecord:
antragsnummer: str
name: str
qualifikation: str
project_scope: str
stellenanteil_percent: float
funktion: str
taetigkeit: str
+214 -200
View File
@@ -1,4 +1,4 @@
"""Parsing and normalization for source workbooks."""
"""Parsing and normalization for source workbooks."""
from __future__ import annotations
@@ -14,262 +14,276 @@ from .models import AllocationRecord, MasterDataRecord
def _to_text(value: Any) -> str:
if value is None:
return ""
return str(value).strip()
if value is None:
return ""
return str(value).strip()
def _to_float(value: Any, field_name: str) -> float:
if value is None or str(value).strip() == "":
raise ValidationError(f"Missing numeric value for {field_name}.")
if value is None or str(value).strip() == "":
raise ValidationError(f"Missing numeric value for {field_name}.")
text = str(value).strip().replace(",", ".")
try:
return float(text)
except ValueError as exc:
raise ValidationError(f"Invalid numeric value for {field_name}: {value}") from exc
text = str(value).strip().replace(",", ".")
try:
return float(text)
except ValueError as exc:
raise ValidationError(f"Invalid numeric value for {field_name}: {value}") from exc
def _to_float_or_none(value: Any) -> float | None:
if value is None or str(value).strip() == "":
return None
text = str(value).strip().replace(",", ".")
try:
return float(text)
except ValueError:
return None
def _parse_german_decimal(
value: Any,
field_name: str,
decimal_separator: str,
thousands_separator: str,
value: Any,
field_name: str,
decimal_separator: str,
thousands_separator: str,
) -> float:
if value is None or str(value).strip() == "":
raise ValidationError(f"Missing numeric value for {field_name}.")
if value is None or str(value).strip() == "":
raise ValidationError(f"Missing numeric value for {field_name}.")
text = str(value).strip()
if thousands_separator:
text = text.replace(thousands_separator, "")
if decimal_separator:
text = text.replace(decimal_separator, ".")
text = str(value).strip()
if thousands_separator:
text = text.replace(thousands_separator, "")
if decimal_separator:
text = text.replace(decimal_separator, ".")
try:
return float(text)
except ValueError as exc:
raise ValidationError(f"Invalid numeric value for {field_name}: {value}") from exc
try:
return float(text)
except ValueError as exc:
raise ValidationError(f"Invalid numeric value for {field_name}: {value}") from exc
def _parse_month_marker(value: Any, prefix: str, date_format: str) -> datetime | None:
marker = _to_text(value)
if not marker:
return None
marker = _to_text(value)
if not marker:
return None
if not marker.lower().startswith(prefix.lower()):
return None
if not marker.lower().startswith(prefix.lower()):
return None
date_text = marker[len(prefix) :].strip()
return datetime.strptime(date_text, date_format)
date_text = marker[len(prefix) :].strip()
return datetime.strptime(date_text, date_format)
def _resolve_month_column_index(
worksheet: Any,
header_row: int,
first_month_col: str,
target_month: int,
target_year: int,
month_prefix: str,
month_date_format: str,
worksheet: Any,
header_row: int,
first_month_col: str,
target_month: int,
target_year: int,
month_prefix: str,
month_date_format: str,
) -> int:
start_index = column_index_from_string(first_month_col)
matched_columns: list[int] = []
start_index = column_index_from_string(first_month_col)
matched_columns: list[int] = []
for column_index in range(start_index, worksheet.max_column + 1):
value = worksheet.cell(row=header_row, column=column_index).value
parsed = _parse_month_marker(value, month_prefix, month_date_format)
if parsed is None:
continue
if parsed.month == target_month and parsed.year == target_year:
matched_columns.append(column_index)
for column_index in range(start_index, worksheet.max_column + 1):
value = worksheet.cell(row=header_row, column=column_index).value
parsed = _parse_month_marker(value, month_prefix, month_date_format)
if parsed is None:
continue
if parsed.month == target_month and parsed.year == target_year:
matched_columns.append(column_index)
if len(matched_columns) == 0:
raise ValidationError(
f"No month column found for {target_month:02d}/{target_year} in allocation workbook."
)
if len(matched_columns) == 0:
raise ValidationError(
f"No month column found for {target_month:02d}/{target_year} in allocation workbook."
)
if len(matched_columns) > 1:
raise ValidationError(
f"Multiple month columns found for {target_month:02d}/{target_year} in allocation workbook."
)
if len(matched_columns) > 1:
raise ValidationError(
f"Multiple month columns found for {target_month:02d}/{target_year} in allocation workbook."
)
return matched_columns[0]
return matched_columns[0]
def _is_inactive_value(value: Any, inactive_values: list[Any]) -> bool:
normalized = _to_text(value)
if normalized in {"", "-"}:
return True
normalized = _to_text(value)
if normalized in {"", "-"}:
return True
for inactive in inactive_values:
if str(inactive).strip() == normalized:
return True
for inactive in inactive_values:
if str(inactive).strip() == normalized:
return True
return False
return False
def parse_allocation_records(
config: dict[str, Any],
report_month: int,
report_year: int,
project_root: Path,
config: dict[str, Any],
report_month: int,
report_year: int,
project_root: Path,
) -> list[AllocationRecord]:
allocation_cfg = config["allocation"]
columns_cfg = allocation_cfg["columns"]
files_cfg = config["files"]
sheets_cfg = config["sheets"]
paths_cfg = config["paths"]
allocation_cfg = config["allocation"]
columns_cfg = allocation_cfg["columns"]
files_cfg = config["files"]
sheets_cfg = config["sheets"]
paths_cfg = config["paths"]
workbook_path = (
project_root
/ paths_cfg["input_dir"]
/ files_cfg["allocation_workbook"]
)
workbook_path = (
project_root
/ paths_cfg["input_dir"]
/ files_cfg["allocation_workbook"]
)
if not workbook_path.exists():
raise ValidationError(f"Allocation workbook not found: {workbook_path}")
if not workbook_path.exists():
raise ValidationError(f"Allocation workbook not found: {workbook_path}")
workbook = load_workbook(workbook_path, data_only=True)
sheet_name = sheets_cfg["allocation"]
if sheet_name not in workbook.sheetnames:
raise ValidationError(f"Sheet '{sheet_name}' not found in {workbook_path.name}")
worksheet = workbook[sheet_name]
workbook = load_workbook(workbook_path, data_only=True)
sheet_name = sheets_cfg["allocation"]
if sheet_name not in workbook.sheetnames:
raise ValidationError(f"Sheet '{sheet_name}' not found in {workbook_path.name}")
worksheet = workbook[sheet_name]
header_row = allocation_cfg["header_month_marker_row"]
month_marker_cfg = allocation_cfg["month_marker"]
month_column = _resolve_month_column_index(
worksheet=worksheet,
header_row=header_row,
first_month_col=columns_cfg["first_month_column"],
target_month=report_month,
target_year=report_year,
month_prefix=month_marker_cfg["prefix"],
month_date_format=month_marker_cfg["date_format"],
)
header_row = allocation_cfg["header_month_marker_row"]
month_marker_cfg = allocation_cfg["month_marker"]
month_column = _resolve_month_column_index(
worksheet=worksheet,
header_row=header_row,
first_month_col=columns_cfg["first_month_column"],
target_month=report_month,
target_year=report_year,
month_prefix=month_marker_cfg["prefix"],
month_date_format=month_marker_cfg["date_format"],
)
employee_col = column_index_from_string(columns_cfg["employee_name"])
contract_col = column_index_from_string(columns_cfg["contract_hours_weekly"])
project_col = column_index_from_string(columns_cfg["project_key"])
remarks_col = column_index_from_string(columns_cfg["remarks"])
inactive_values = allocation_cfg.get("inactive_values", ["", "-", "0", 0])
employee_col = column_index_from_string(columns_cfg["employee_name"])
contract_col = column_index_from_string(columns_cfg["contract_hours_weekly"])
project_col = column_index_from_string(columns_cfg["project_key"])
remarks_col = column_index_from_string(columns_cfg["remarks"])
inactive_values = allocation_cfg.get("inactive_values", ["", "-", "0", 0])
current_employee = ""
current_contract_hours: float | None = None
normalized_records: list[AllocationRecord] = []
current_employee = ""
current_contract_hours: float | None = None
normalized_records: list[AllocationRecord] = []
for row_index in range(header_row + 1, worksheet.max_row + 1):
row_employee = _to_text(worksheet.cell(row=row_index, column=employee_col).value)
row_contract = worksheet.cell(row=row_index, column=contract_col).value
row_project = _to_text(worksheet.cell(row=row_index, column=project_col).value)
for row_index in range(header_row + 1, worksheet.max_row + 1):
row_employee = _to_text(worksheet.cell(row=row_index, column=employee_col).value)
row_contract = worksheet.cell(row=row_index, column=contract_col).value
row_project = _to_text(worksheet.cell(row=row_index, column=project_col).value)
if row_employee:
current_employee = row_employee
current_contract_hours = _to_float(
row_contract,
f"contract_hours_weekly row {row_index}",
)
if row_employee:
current_employee = row_employee
current_contract_hours = _to_float_or_none(row_contract)
if not row_project:
continue
if not row_project:
continue
if not current_employee or current_contract_hours is None:
raise ValidationError(
f"Project row {row_index} has no inherited employee context."
)
if not current_employee or current_contract_hours is None:
raise ValidationError(
f"Project row {row_index} has no inherited employee context."
)
month_value = worksheet.cell(row=row_index, column=month_column).value
if _is_inactive_value(month_value, inactive_values):
continue
month_value = worksheet.cell(row=row_index, column=month_column).value
if _is_inactive_value(month_value, inactive_values):
continue
funded_hours = _to_float(month_value, f"funded_hours row {row_index}")
if funded_hours <= 0:
continue
funded_hours = _to_float(month_value, f"funded_hours row {row_index}")
if funded_hours <= 0:
continue
if current_contract_hours <= 0:
raise ValidationError(
f"Contract hours must be > 0 for employee '{current_employee}'"
)
if current_contract_hours <= 0:
raise ValidationError(
f"Contract hours must be > 0 for employee '{current_employee}'"
)
notes = _to_text(worksheet.cell(row=row_index, column=remarks_col).value)
normalized_records.append(
AllocationRecord(
employee_name=current_employee,
contract_hours_weekly=current_contract_hours,
project_key=row_project,
funded_hours_weekly=funded_hours,
notes=notes,
)
)
notes = _to_text(worksheet.cell(row=row_index, column=remarks_col).value)
normalized_records.append(
AllocationRecord(
employee_name=current_employee,
contract_hours_weekly=current_contract_hours,
project_key=row_project,
funded_hours_weekly=funded_hours,
notes=notes,
)
)
workbook.close()
return normalized_records
workbook.close()
return normalized_records
def parse_masterdata_records(
config: dict[str, Any],
project_root: Path,
config: dict[str, Any],
project_root: Path,
) -> list[MasterDataRecord]:
files_cfg = config["files"]
sheets_cfg = config["sheets"]
paths_cfg = config["paths"]
master_cfg = config["masterdata"]
columns_cfg = master_cfg["columns"]
decimal_cfg = master_cfg["decimal"]
files_cfg = config["files"]
sheets_cfg = config["sheets"]
paths_cfg = config["paths"]
master_cfg = config["masterdata"]
columns_cfg = master_cfg["columns"]
decimal_cfg = master_cfg["decimal"]
workbook_path = (
project_root
/ paths_cfg["input_dir"]
/ files_cfg["masterdata_workbook"]
)
workbook_path = (
project_root
/ paths_cfg["input_dir"]
/ files_cfg["masterdata_workbook"]
)
if not workbook_path.exists():
raise ValidationError(f"Master data workbook not found: {workbook_path}")
if not workbook_path.exists():
raise ValidationError(f"Master data workbook not found: {workbook_path}")
workbook = load_workbook(workbook_path, data_only=True)
sheet_name = sheets_cfg["masterdata"]
if sheet_name not in workbook.sheetnames:
raise ValidationError(f"Sheet '{sheet_name}' not found in {workbook_path.name}")
worksheet = workbook[sheet_name]
workbook = load_workbook(workbook_path, data_only=True)
sheet_name = sheets_cfg["masterdata"]
if sheet_name not in workbook.sheetnames:
raise ValidationError(f"Sheet '{sheet_name}' not found in {workbook_path.name}")
worksheet = workbook[sheet_name]
id_col = column_index_from_string(columns_cfg["id"])
antragsnummer_col = column_index_from_string(columns_cfg["antragsnummer"])
name_col = column_index_from_string(columns_cfg["name"])
qualifikation_col = column_index_from_string(columns_cfg["qualifikation"])
stellenanteil_col = column_index_from_string(columns_cfg["stellenanteil"])
funktion_col = column_index_from_string(columns_cfg["funktion"])
taetigkeit_col = column_index_from_string(columns_cfg["taetigkeit"])
id_col = column_index_from_string(columns_cfg["id"])
antragsnummer_col = column_index_from_string(columns_cfg["antragsnummer"])
name_col = column_index_from_string(columns_cfg["name"])
qualifikation_col = column_index_from_string(columns_cfg["qualifikation"])
project_scope_col = column_index_from_string(columns_cfg["project_scope"])
stellenanteil_col = column_index_from_string(columns_cfg["stellenanteil"])
funktion_col = column_index_from_string(columns_cfg["funktion"])
taetigkeit_col = column_index_from_string(columns_cfg["taetigkeit"])
records: list[MasterDataRecord] = []
records: list[MasterDataRecord] = []
for row_index in range(2, worksheet.max_row + 1):
name = _to_text(worksheet.cell(row=row_index, column=name_col).value)
if not name:
continue
for row_index in range(2, worksheet.max_row + 1):
name = _to_text(worksheet.cell(row=row_index, column=name_col).value)
if not name:
continue
stellenanteil = _parse_german_decimal(
worksheet.cell(row=row_index, column=stellenanteil_col).value,
field_name=f"stellenanteil row {row_index}",
decimal_separator=decimal_cfg.get("decimal_separator", ","),
thousands_separator=decimal_cfg.get("thousands_separator", "."),
)
stellenanteil_raw = worksheet.cell(row=row_index, column=stellenanteil_col).value
stellenanteil = _to_float_or_none(stellenanteil_raw)
if stellenanteil is None:
stellenanteil = 0.0
records.append(
MasterDataRecord(
record_id=_to_text(worksheet.cell(row=row_index, column=id_col).value),
antragsnummer=_to_text(
worksheet.cell(row=row_index, column=antragsnummer_col).value
),
name=name,
qualifikation=_to_text(
worksheet.cell(row=row_index, column=qualifikation_col).value
),
stellenanteil_percent=stellenanteil,
funktion=_to_text(worksheet.cell(row=row_index, column=funktion_col).value),
taetigkeit=_to_text(
worksheet.cell(row=row_index, column=taetigkeit_col).value
),
)
)
project_scope = _to_text(
worksheet.cell(row=row_index, column=project_scope_col).value
)
if not project_scope:
project_scope = "PS"
workbook.close()
return records
records.append(
MasterDataRecord(
record_id=_to_text(worksheet.cell(row=row_index, column=id_col).value),
antragsnummer=_to_text(
worksheet.cell(row=row_index, column=antragsnummer_col).value
),
name=name,
qualifikation=_to_text(
worksheet.cell(row=row_index, column=qualifikation_col).value
),
project_scope=project_scope,
stellenanteil_percent=stellenanteil,
funktion=_to_text(worksheet.cell(row=row_index, column=funktion_col).value),
taetigkeit=_to_text(
worksheet.cell(row=row_index, column=taetigkeit_col).value
),
)
)
workbook.close()
return records
+155 -6
View File
@@ -9,7 +9,7 @@ from typing import Any
from openpyxl import load_workbook
from openpyxl.utils.cell import column_index_from_string
from .models import ValidationIssue
from .models import AllocationRecord, MatchedAssignment, ValidationIssue
def _to_text(value: Any) -> str:
@@ -132,6 +132,7 @@ def validate_sources(
master_cols["antragsnummer"],
master_cols["name"],
master_cols["qualifikation"],
master_cols["project_scope"],
master_cols["stellenanteil"],
master_cols["funktion"],
master_cols["taetigkeit"],
@@ -194,22 +195,31 @@ def validate_sources(
if len(month_matches) == 1:
month_column = month_matches[0]
employee_col = column_index_from_string(allocation_cols["employee_name"])
contract_col = column_index_from_string(allocation_cols["contract_hours_weekly"])
project_col = column_index_from_string(allocation_cols["project_key"])
current_contract_value: float | None = None
for row_idx in range(header_row + 1, allocation_ws.max_row + 1):
employee_value = _to_text(allocation_ws.cell(row=row_idx, column=employee_col).value)
contract_raw = allocation_ws.cell(row=row_idx, column=contract_col).value
parsed_contract = _to_float_or_none(contract_raw)
if employee_value:
current_contract_value = parsed_contract
project_key = _to_text(allocation_ws.cell(row=row_idx, column=project_col).value)
if not project_key:
continue
contract_value = _to_float_or_none(
allocation_ws.cell(row=row_idx, column=contract_col).value
)
if contract_value is None or contract_value <= 0:
if current_contract_value is None or current_contract_value <= 0:
issues.append(
ValidationIssue(
code="INVALID_CONTRACT_HOURS",
message=f"Invalid contract weekly hours at row {row_idx}.",
message=(
f"Invalid inherited contract weekly hours at row {row_idx}."
),
severity="error",
)
)
@@ -232,3 +242,142 @@ def validate_sources(
allocation_wb.close()
master_wb.close()
return issues
def _split_name_for_filename(name: str) -> tuple[str, str]:
cleaned = _to_text(name)
if "," in cleaned:
last_name, first_name = [part.strip() for part 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 _build_output_filename(pattern: str, year: int, month: int, employee_name: str) -> str:
first_name, last_name = _split_name_for_filename(employee_name)
safe_first = first_name.replace(" ", "_")
safe_last = last_name.replace(" ", "_")
return pattern.format(
year=year,
month=f"{month:02d}",
first_name=safe_first,
last_name=safe_last,
)
def validate_business_rules(
config: dict[str, Any],
report_month: int,
report_year: int,
allocation_records: list[AllocationRecord],
matched_assignments: list[MatchedAssignment],
project_root: Path,
) -> list[ValidationIssue]:
issues: list[ValidationIssue] = []
if len(allocation_records) == 0:
issues.append(
ValidationIssue(
code="NO_ACTIVE_ASSIGNMENT",
message="No active allocation record found for the requested month.",
severity="error",
)
)
matched_keys: set[tuple[str, str, float]] = set()
for item in matched_assignments:
key = (
item.allocation.employee_name,
item.allocation.project_key,
item.allocation.funded_hours_weekly,
)
if key in matched_keys:
issues.append(
ValidationIssue(
code="DUPLICATE_MATCH",
message=(
f"Duplicate matched assignment for '{item.allocation.employee_name}' "
f"and project '{item.allocation.project_key}'."
),
severity="error",
context=item.allocation.employee_name,
)
)
matched_keys.add(key)
paths_cfg = config["paths"]
files_cfg = config["files"]
template_path = (
project_root
/ paths_cfg["template_dir"]
/ files_cfg["template_workbook"]
)
if template_path.exists():
template_wb = load_workbook(template_path, data_only=True)
configured_sheet = config["sheets"].get("template_source", "")
if configured_sheet not in template_wb.sheetnames:
issues.append(
ValidationIssue(
code="TEMPLATE_SHEET_MISSING",
message=(
f"Configured template sheet '{configured_sheet}' not found in template workbook."
),
severity="error",
)
)
template_wb.close()
else:
issues.append(
ValidationIssue(
code="MISSING_TEMPLATE",
message=f"Template workbook not found: {template_path}",
severity="error",
)
)
template_targets = config.get("template_targets", {})
missing_targets = [name for name, cell in template_targets.items() if not _to_text(cell)]
if missing_targets:
issues.append(
ValidationIssue(
code="MISSING_TEMPLATE_TARGETS",
message=(
"Mandatory template target cells not configured: "
+ ", ".join(missing_targets)
),
severity="error",
)
)
filename_pattern = config["output"]["filename_pattern"]
filename_to_employee: dict[str, str] = {}
for assignment in matched_assignments:
filename = _build_output_filename(
filename_pattern,
report_year,
report_month,
assignment.allocation.employee_name,
)
previous_owner = filename_to_employee.get(filename)
if previous_owner is None:
filename_to_employee[filename] = assignment.allocation.employee_name
continue
if previous_owner != assignment.allocation.employee_name:
issues.append(
ValidationIssue(
code="FILENAME_COLLISION",
message=(
f"Output filename collision between '{previous_owner}' and "
f"'{assignment.allocation.employee_name}' -> {filename}"
),
severity="error",
)
)
return issues
+152
View File
@@ -1 +1,153 @@
"""Template-based output workbook writer."""
from __future__ import annotations
from collections import defaultdict
from datetime import date
from pathlib import Path
from typing import Any
from openpyxl import Workbook, load_workbook
from .models import MatchedAssignment
def _split_name_for_filename(name: str) -> tuple[str, str]:
cleaned = str(name or "").strip()
if "," in cleaned:
last_name, first_name = [part.strip() for part 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 _normalize_sheet_title(raw_title: str, fallback_index: int) -> str:
cleaned = (raw_title or "").strip().replace("[", "(").replace("]", ")")
if not cleaned:
cleaned = f"Assignment_{fallback_index}"
return cleaned[:31]
def _set_target_value(worksheet: Any, target_cell: str, value: Any) -> None:
if not target_cell:
return
worksheet[target_cell] = value
def build_employee_workbooks(
config: dict[str, Any],
matched_assignments: list[MatchedAssignment],
report_month: int,
report_year: int,
project_root: Path,
) -> dict[str, Workbook]:
paths_cfg = config["paths"]
files_cfg = config["files"]
output_cfg = config["output"]
targets = config.get("template_targets", {})
template_path = (
project_root
/ paths_cfg["template_dir"]
/ files_cfg["template_workbook"]
)
assignments_by_employee: dict[str, list[MatchedAssignment]] = defaultdict(list)
for assignment in matched_assignments:
assignments_by_employee[assignment.allocation.employee_name].append(assignment)
workbooks: dict[str, Workbook] = {}
month_label = f"{report_month:02d}/{report_year}"
today_label = date.today().strftime("%d.%m.%Y")
for employee_name, assignments in assignments_by_employee.items():
workbook = load_workbook(template_path)
template_sheet_name = config["sheets"].get("template_source", "")
if template_sheet_name and template_sheet_name in workbook.sheetnames:
base_sheet = workbook[template_sheet_name]
else:
base_sheet = workbook.worksheets[0]
for index, assignment in enumerate(assignments, start=1):
if index == 1:
sheet = base_sheet
else:
sheet = workbook.copy_worksheet(base_sheet)
sheet_name_pattern = output_cfg.get("sheet_name_pattern", "{index}_{funktion}")
raw_sheet_name = sheet_name_pattern.format(
index=index,
funktion=assignment.masterdata.funktion,
)
sheet.title = _normalize_sheet_title(raw_sheet_name, index)
_set_target_value(sheet, targets.get("antragsnummer", ""), assignment.masterdata.antragsnummer)
_set_target_value(sheet, targets.get("name_vorname", ""), assignment.allocation.employee_name)
_set_target_value(sheet, targets.get("qualifikation", ""), assignment.masterdata.qualifikation)
_set_target_value(sheet, targets.get("funktion", ""), assignment.masterdata.funktion)
_set_target_value(sheet, targets.get("taetigkeiten", ""), assignment.masterdata.taetigkeit)
_set_target_value(sheet, targets.get("reporting_month_label", ""), month_label)
_set_target_value(sheet, targets.get("foerderquote", ""), f"{assignment.foerderquote_percent:.2f} %")
_set_target_value(sheet, targets.get("generated_on", ""), today_label)
workbooks[employee_name] = workbook
return workbooks
def build_output_filename(
filename_pattern: str,
employee_name: str,
report_year: int,
report_month: int,
) -> str:
first_name, last_name = _split_name_for_filename(employee_name)
safe_first = first_name.replace(" ", "_")
safe_last = last_name.replace(" ", "_")
return filename_pattern.format(
year=report_year,
month=f"{report_month:02d}",
first_name=safe_first,
last_name=safe_last,
)
def save_employee_workbooks(
config: dict[str, Any],
employee_workbooks: dict[str, Workbook],
report_month: int,
report_year: int,
project_root: Path,
) -> list[Path]:
paths_cfg = config["paths"]
output_cfg = config["output"]
filename_pattern = output_cfg["filename_pattern"]
output_dir = project_root / paths_cfg["output_dir"]
output_dir.mkdir(parents=True, exist_ok=True)
saved_paths: list[Path] = []
used_names: set[str] = set()
for employee_name, workbook in employee_workbooks.items():
filename = build_output_filename(
filename_pattern=filename_pattern,
employee_name=employee_name,
report_year=report_year,
report_month=report_month,
)
if filename in used_names:
raise ValueError(f"Output filename collision detected: {filename}")
used_names.add(filename)
output_path = output_dir / filename
workbook.save(output_path)
workbook.close()
saved_paths.append(output_path)
return saved_paths