"""Parsing and normalization for source workbooks.""" 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 .errors import ValidationError from .models import AllocationRecord def _to_text(value: Any) -> str: 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}.") 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 _parse_month_marker(value: Any, prefix: str, date_format: str) -> datetime | None: marker = _to_text(value) if not marker: return None if not marker.lower().startswith(prefix.lower()): return None 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, ) -> 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) 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." ) return matched_columns[0] def _is_inactive_value(value: Any, inactive_values: list[Any]) -> bool: normalized = _to_text(value) if normalized in {"", "-"}: return True for inactive in inactive_values: if str(inactive).strip() == normalized: return True return False def parse_allocation_records( 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"] 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}") 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"], ) 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] = [] 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 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." ) 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 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, ) ) workbook.close() return normalized_records