"""Record matching logic across allocation and master data.""" from __future__ import annotations import re from typing import Any from .errors import ValidationError from .models import AllocationRecord, MasterDataRecord, MatchedAssignment, ValidationIssue def calculate_foerderquote_percent(allocation: AllocationRecord, precision: int = 2) -> float: if allocation.contract_hours_weekly <= 0: 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() def _with_tolerance( value: float, target: float, tolerance: float, ) -> bool: return abs(value - target) <= tolerance def match_assignments( allocation_records: list[AllocationRecord], 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)) master_by_name: dict[str, list[MasterDataRecord]] = {} for record in masterdata_records: master_by_name.setdefault(_normalize_name(record.name), []).append(record) 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, []) if not candidates: issues.append( ValidationIssue( code="MISSING_MASTERDATA", message=( f"No master-data match for employee '{allocation.employee_name}' " f"(project '{allocation.project_key}')." ), severity="error", context=allocation.employee_name, ) ) 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: 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 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 if allocation.notes: issues.append( ValidationIssue( code="ALLOCATION_NOTE", message=( f"Allocation note for '{allocation.employee_name}' / " f"'{allocation.project_key}': {allocation.notes}" ), severity="warning", context=allocation.employee_name, ) ) matched.append( MatchedAssignment( allocation=allocation, masterdata=selected, foerderquote_percent=foerderquote, ) ) return matched, issues