384 lines
13 KiB
Python
384 lines
13 KiB
Python
"""Source and business rule validation."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from openpyxl import load_workbook
|
|
from openpyxl.utils.cell import column_index_from_string
|
|
|
|
from .models import AllocationRecord, MatchedAssignment, ValidationIssue
|
|
|
|
|
|
def _to_text(value: Any) -> str:
|
|
if value is None:
|
|
return ""
|
|
return str(value).strip()
|
|
|
|
|
|
def _to_float_or_none(value: Any) -> float | None:
|
|
text = _to_text(value)
|
|
if not text:
|
|
return None
|
|
text = text.replace(",", ".")
|
|
try:
|
|
return float(text)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _parse_month_marker(value: Any, prefix: str, date_format: str) -> datetime | None:
|
|
marker = _to_text(value)
|
|
if not marker or not marker.lower().startswith(prefix.lower()):
|
|
return None
|
|
|
|
date_text = marker[len(prefix) :].strip()
|
|
try:
|
|
return datetime.strptime(date_text, date_format)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def validate_sources(
|
|
config: dict[str, Any],
|
|
report_month: int,
|
|
report_year: int,
|
|
project_root: Path,
|
|
) -> list[ValidationIssue]:
|
|
issues: list[ValidationIssue] = []
|
|
|
|
paths_cfg = config["paths"]
|
|
files_cfg = config["files"]
|
|
sheets_cfg = config["sheets"]
|
|
allocation_cfg = config["allocation"]
|
|
allocation_cols = allocation_cfg["columns"]
|
|
month_cfg = allocation_cfg["month_marker"]
|
|
|
|
input_dir = project_root / paths_cfg["input_dir"]
|
|
template_dir = project_root / paths_cfg["template_dir"]
|
|
|
|
allocation_path = input_dir / files_cfg["allocation_workbook"]
|
|
masterdata_path = input_dir / files_cfg["masterdata_workbook"]
|
|
template_path = template_dir / files_cfg["template_workbook"]
|
|
|
|
for required_path in [allocation_path, masterdata_path, template_path]:
|
|
if not required_path.exists():
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="MISSING_FILE",
|
|
message=f"Required file not found: {required_path}",
|
|
severity="error",
|
|
)
|
|
)
|
|
|
|
if any(issue.code == "MISSING_FILE" for issue in issues):
|
|
return issues
|
|
|
|
allocation_wb = load_workbook(allocation_path, data_only=True)
|
|
master_wb = load_workbook(masterdata_path, data_only=True)
|
|
|
|
allocation_sheet_name = sheets_cfg["allocation"]
|
|
master_sheet_name = sheets_cfg["masterdata"]
|
|
|
|
if allocation_sheet_name not in allocation_wb.sheetnames:
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="MISSING_SHEET",
|
|
message=f"Sheet '{allocation_sheet_name}' missing in {allocation_path.name}",
|
|
severity="error",
|
|
)
|
|
)
|
|
|
|
if master_sheet_name not in master_wb.sheetnames:
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="MISSING_SHEET",
|
|
message=f"Sheet '{master_sheet_name}' missing in {masterdata_path.name}",
|
|
severity="error",
|
|
)
|
|
)
|
|
|
|
if any(issue.code == "MISSING_SHEET" for issue in issues):
|
|
allocation_wb.close()
|
|
master_wb.close()
|
|
return issues
|
|
|
|
allocation_ws = allocation_wb[allocation_sheet_name]
|
|
master_ws = master_wb[master_sheet_name]
|
|
|
|
required_allocation_columns = [
|
|
allocation_cols["employee_name"],
|
|
allocation_cols["contract_hours_weekly"],
|
|
allocation_cols["project_key"],
|
|
allocation_cols["first_month_column"],
|
|
allocation_cols["remarks"],
|
|
]
|
|
max_allocation_col = allocation_ws.max_column
|
|
for col_name in required_allocation_columns:
|
|
if column_index_from_string(col_name) > max_allocation_col:
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="MISSING_COLUMN",
|
|
message=f"Allocation column '{col_name}' is not present.",
|
|
severity="error",
|
|
)
|
|
)
|
|
|
|
master_cols = config["masterdata"]["columns"]
|
|
required_master_columns = [
|
|
master_cols["id"],
|
|
master_cols["antragsnummer"],
|
|
master_cols["name"],
|
|
master_cols["qualifikation"],
|
|
master_cols["project_scope"],
|
|
master_cols["stellenanteil"],
|
|
master_cols["funktion"],
|
|
master_cols["taetigkeit"],
|
|
]
|
|
max_master_col = master_ws.max_column
|
|
for col_name in required_master_columns:
|
|
if column_index_from_string(col_name) > max_master_col:
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="MISSING_COLUMN",
|
|
message=f"Master-data column '{col_name}' is not present.",
|
|
severity="error",
|
|
)
|
|
)
|
|
|
|
header_row = allocation_cfg["header_month_marker_row"]
|
|
first_month_col = column_index_from_string(allocation_cols["first_month_column"])
|
|
month_matches: list[int] = []
|
|
parseable_markers = 0
|
|
|
|
for col_idx in range(first_month_col, allocation_ws.max_column + 1):
|
|
marker_value = allocation_ws.cell(row=header_row, column=col_idx).value
|
|
parsed = _parse_month_marker(
|
|
marker_value,
|
|
month_cfg["prefix"],
|
|
month_cfg["date_format"],
|
|
)
|
|
if parsed is None:
|
|
continue
|
|
|
|
parseable_markers += 1
|
|
if parsed.month == report_month and parsed.year == report_year:
|
|
month_matches.append(col_idx)
|
|
|
|
if parseable_markers == 0:
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="INVALID_MONTH_MARKERS",
|
|
message="No parseable month marker found in allocation header row.",
|
|
severity="error",
|
|
)
|
|
)
|
|
|
|
if len(month_matches) == 0:
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="MONTH_NOT_FOUND",
|
|
message=f"Requested month {report_month:02d}/{report_year} not found in allocation sheet.",
|
|
severity="error",
|
|
)
|
|
)
|
|
elif len(month_matches) > 1:
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="MONTH_AMBIGUOUS",
|
|
message=f"Requested month {report_month:02d}/{report_year} appears in multiple columns.",
|
|
severity="error",
|
|
)
|
|
)
|
|
|
|
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
|
|
|
|
if current_contract_value is None or current_contract_value <= 0:
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="INVALID_CONTRACT_HOURS",
|
|
message=(
|
|
f"Invalid inherited contract weekly hours at row {row_idx}."
|
|
),
|
|
severity="error",
|
|
)
|
|
)
|
|
|
|
monthly_value = allocation_ws.cell(row=row_idx, column=month_column).value
|
|
monthly_text = _to_text(monthly_value)
|
|
if monthly_text in {"", "-", "0"}:
|
|
continue
|
|
|
|
monthly_hours = _to_float_or_none(monthly_value)
|
|
if monthly_hours is None:
|
|
issues.append(
|
|
ValidationIssue(
|
|
code="INVALID_HOUR_VALUE",
|
|
message=f"Invalid funded hour value at row {row_idx}.",
|
|
severity="error",
|
|
)
|
|
)
|
|
|
|
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
|