"""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 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["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] contract_col = column_index_from_string(allocation_cols["contract_hours_weekly"]) project_col = column_index_from_string(allocation_cols["project_key"]) for row_idx in range(header_row + 1, allocation_ws.max_row + 1): 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: issues.append( ValidationIssue( code="INVALID_CONTRACT_HOURS", message=f"Invalid 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