Excel Budget Template Sync

import logging
import hashlib
import uuid
from datetime import datetime
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import List, Dict, Any, Optional

import pandas as pd
import openpyxl
from pydantic import BaseModel, Field, ValidationError, field_validator, model_validator
from pydantic import ConfigDict

# Structured audit logger configuration
AUDIT_LOGGER = logging.getLogger("grant_budget_audit")
AUDIT_LOGGER.setLevel(logging.INFO)
formatter = logging.JSONFormatter() if hasattr(logging, 'JSONFormatter') else logging.Formatter(
    '{"timestamp":"%(asctime)s","level":"%(levelname)s","trace_id":"%(trace_id)s","event":"%(message)s"}'
)
handler = logging.StreamHandler()
handler.setFormatter(formatter)
AUDIT_LOGGER.addHandler(handler)

class BudgetLineItem(BaseModel):
    model_config = ConfigDict(arbitrary_types_allowed=True)
    category: str
    description: str
    amount: Decimal
    period_start: datetime
    period_end: datetime
    indirect_rate: Optional[Decimal] = Field(None, ge=0, le=Decimal("0.15"))

    @field_validator("amount", mode="before")
    @classmethod
    def coerce_decimal(cls, v: Any) -> Decimal:
        try:
            return Decimal(str(v).replace(",", "").replace("$", "").strip())
        except InvalidOperation:
            raise ValueError(f"Non-numeric budget value detected: {v}")

    @field_validator("category", mode="before")
    @classmethod
    def validate_category(cls, v: str) -> str:
        allowed = {"PERSONNEL", "TRAVEL", "EQUIPMENT", "INDIRECT", "OTHER_DIRECT"}
        if v.upper() not in allowed:
            raise ValueError(f"Invalid budget category: {v}")
        return v.upper()

    @model_validator(mode="after")
    def validate_period_alignment(self) -> "BudgetLineItem":
        if self.period_end <= self.period_start:
            raise ValueError("Period end must strictly follow period start")
        return self

class BudgetValidator:
    def __init__(self, file_path: Path, schema_version: str = "v2.1"):
        self.file_path = file_path
        self.schema_version = schema_version
        self.trace_id = str(uuid.uuid4())
        self.checksum = self._compute_checksum()
        self.validation_log: List[Dict[str, Any]] = []

    def _compute_checksum(self) -> str:
        sha256 = hashlib.sha256()
        with open(self.file_path, "rb") as f:
            for chunk in iter(lambda: f.read(8192), b""):
                sha256.update(chunk)
        return sha256.hexdigest()

    def _log_audit(self, level: str, message: str, **kwargs):
        extra = {"trace_id": self.trace_id, **kwargs}
        AUDIT_LOGGER.log(getattr(logging, level.upper()), message, extra=extra)

    def validate(self) -> Dict[str, Any]:
        self._log_audit("INFO", "Initiating budget template validation", 
                        file=str(self.file_path), schema=self.schema_version)
        
        # 1. Structural Integrity via openpyxl
        try:
            wb = openpyxl.load_workbook(self.file_path, read_only=True, data_only=True)
            ws = wb.active
            if ws.sheet_properties.tabColor is not None:
                self._log_audit("WARN", "Non-standard sheet styling detected")
            if any(ws.merged_cells.ranges):
                raise ValueError("Merged cells violate structural integrity policy")
            if ws.protection.sheet:
                raise ValueError("Protected sheets block programmatic validation")
            wb.close()
        except Exception as e:
            self._log_audit("ERROR", f"Structural validation failed: {str(e)}")
            return {"status": "QUARANTINE", "reason": "STRUCTURAL_FAILURE", "trace_id": self.trace_id}

        # 2. Schema & Data Validation via pandas + pydantic
        try:
            df = pd.read_excel(self.file_path, engine="openpyxl")
            required_cols = {"category", "description", "amount", "period_start", "period_end"}
            missing = required_cols - set(df.columns.str.lower())
            if missing:
                raise ValueError(f"Missing required headers: {missing}")
            
            # Normalize headers
            df.columns = df.columns.str.lower().str.strip()
            
            # Parse dates
            df["period_start"] = pd.to_datetime(df["period_start"], errors="raise")
            df["period_end"] = pd.to_datetime(df["period_end"], errors="raise")
            
            # Vectorized validation
            validated_rows = []
            for idx, row in df.iterrows():
                try:
                    validated_rows.append(BudgetLineItem(**row.to_dict()))
                except ValidationError as ve:
                    self._log_audit("ERROR", f"Row {idx} validation failed", errors=ve.errors())
                    return {"status": "QUARANTINE", "reason": "SCHEMA_VIOLATION", "trace_id": self.trace_id}

            # Cross-column dependency check
            total_expenses = sum(item.amount for item in validated_rows)
            df_total = df["amount"].sum()
            if abs(total_expenses - df_total) > Decimal("0.01"):
                raise ValueError("Arithmetic mismatch between line items and declared total")

            self._log_audit("INFO", "Validation successful", 
                            row_count=len(validated_rows), total_expenses=str(total_expenses))
            
            return {
                "status": "VALIDATED",
                "trace_id": self.trace_id,
                "checksum": self.checksum,
                "schema_version": self.schema_version,
                "validated_count": len(validated_rows),
                "total_budget": str(total_expenses),
                "audit_timestamp": datetime.utcnow().isoformat()
            }

        except Exception as e:
            self._log_audit("ERROR", f"Data validation failed: {str(e)}")
            return {"status": "QUARANTINE", "reason": "DATA_FAILURE", "trace_id": self.trace_id}