Automating Excel to CSV Conversion for Budget Tracking: Tactical Implementation & Compliance Routing

import csv
import hashlib
import json
import logging
from decimal import Decimal, ROUND_HALF_UP, InvalidOperation
from pathlib import Path
from typing import Dict, List, Optional, Generator
from datetime import datetime, timezone

import openpyxl
from openpyxl.worksheet.worksheet import Worksheet

# ---------------------------------------------------------------------------
# Structured Audit Logger Configuration
# ---------------------------------------------------------------------------
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(message)s",
    handlers=[logging.FileHandler("grant_budget_audit.log", encoding="utf-8")]
)
logger = logging.getLogger("grant_budget_converter")

# ---------------------------------------------------------------------------
# Custom Exception Hierarchy for Explicit Error Routing
# ---------------------------------------------------------------------------
class BudgetConversionError(Exception):
    """Base exception for pipeline-stage failures."""

class SchemaDriftError(BudgetConversionError):
    """Raised when canonical headers cannot be resolved."""

class PrecisionValidationError(BudgetConversionError):
    """Raised when decimal coercion fails or exceeds audit thresholds."""

class MemoryBoundExceededError(BudgetConversionError):
    """Raised when row iteration exceeds configured memory limits."""

# ---------------------------------------------------------------------------
# Compliance & Configuration Constants
# ---------------------------------------------------------------------------
CANONICAL_SCHEMA = [
    "line_item_id", "category", "description",
    "fiscal_year", "budget_amount", "actuals", "variance_pct"
]

ALIAS_MAP: Dict[str, str] = {
    "Personnel": "category",
    "Line Item": "line_item_id",
    "FY Budget": "budget_amount",
    "Actual Spend": "actuals",
    "Variance %": "variance_pct",
    "Description/Notes": "description",
    "Fiscal Year": "fiscal_year"
}

QUARANTINE_DIR = Path("./quarantine")
QUARANTINE_DIR.mkdir(exist_ok=True)

# ---------------------------------------------------------------------------
# Core Conversion Pipeline
# ---------------------------------------------------------------------------
def _compute_sha256(file_path: Path) -> str:
    """Deterministic file hashing for 2 CFR §200.302 audit trails."""
    sha = hashlib.sha256()
    with open(file_path, "rb") as f:
        for chunk in iter(lambda: f.read(8192), b""):
            sha.update(chunk)
    return sha.hexdigest()

def _resolve_headers(raw_headers: List[str]) -> Dict[int, str]:
    """Map raw Excel headers to canonical schema. Fails fast on drift."""
    mapping: Dict[int, str] = {}
    normalized = [h.strip().lower().replace(" ", "_") for h in raw_headers]
    
    for idx, header in enumerate(normalized):
        if header in CANONICAL_SCHEMA:
            mapping[idx] = header
        elif header in ALIAS_MAP:
            mapping[idx] = ALIAS_MAP[header]
        else:
            logger.warning(f"Unmapped header detected: {raw_headers[idx]}")
            
    missing = set(CANONICAL_SCHEMA) - set(mapping.values())
    if missing:
        raise SchemaDriftError(f"Canonical columns missing after alias resolution: {missing}")
    return mapping

def _coerce_decimal(value: Optional[str], field: str) -> str:
    """Enforce 2 CFR §200.403 precision rules. Returns string for CSV serialization."""
    if value is None or str(value).strip() == "":
        return "0.00"
    try:
        dec = Decimal(str(value).replace(",", "").strip())
        rounded = dec.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
        return str(rounded)
    except (InvalidOperation, ValueError) as e:
        raise PrecisionValidationError(f"Invalid monetary value for {field}: {value}") from e

def _stream_rows(ws: Worksheet, header_map: Dict[int, str]) -> Generator[Dict[str, str], None, None]:
    """Memory-safe row iteration with explicit type preservation."""
    for row_idx, row in enumerate(ws.iter_rows(min_row=2, values_only=True), start=2):
        if row_idx > 500_000:  # Hard memory bound to prevent OOM
            raise MemoryBoundExceededError("Row limit exceeded. Enforce chunked processing upstream.")
            
        record: Dict[str, str] = {}
        for col_idx, canonical_name in header_map.items():
            raw_val = row[col_idx] if col_idx < len(row) else None
            if canonical_name in ("budget_amount", "actuals", "variance_pct"):
                record[canonical_name] = _coerce_decimal(str(raw_val), canonical_name)
            else:
                record[canonical_name] = str(raw_val).strip() if raw_val is not None else ""
        yield record

def convert_excel_to_csv(
    input_path: Path,
    output_path: Path,
    sheet_name: Optional[str] = None
) -> Dict[str, str]:
    """
    Deterministic Excel-to-CSV conversion with audit logging and compliance routing.
    Maps directly to 2 CFR §200.302/403 requirements.
    """
    input_hash = _compute_sha256(input_path)
    logger.info(f"Starting conversion | input={input_path.name} | hash={input_hash}")

    try:
        wb = openpyxl.load_workbook(str(input_path), read_only=True, data_only=True)
        ws = wb[sheet_name] if sheet_name else wb.active
        header_row = next(ws.iter_rows(min_row=1, max_row=1, values_only=True))
        
        header_map = _resolve_headers(list(header_row))
        
        with open(output_path, "w", newline="", encoding="utf-8") as csvfile:
            writer = csv.DictWriter(csvfile, fieldnames=CANONICAL_SCHEMA)
            writer.writeheader()
            
            row_count = 0
            for record in _stream_rows(ws, header_map):
                writer.writerow(record)
                row_count += 1
                
        wb.close()
        output_hash = _compute_sha256(output_path)
        
        audit_record = {
            "timestamp": datetime.now(timezone.utc).isoformat(),
            "input_file": input_path.name,
            "input_hash": input_hash,
            "output_file": output_path.name,
            "output_hash": output_hash,
            "rows_processed": row_count,
            "compliance_status": "PASS",
            "standard": "2 CFR §200.302/403"
        }
        logger.info(json.dumps(audit_record))
        return audit_record

    except (SchemaDriftError, PrecisionValidationError, MemoryBoundExceededError) as e:
        logger.error(f"Conversion failed | input={input_path.name} | error={e}")
        _route_to_quarantine(input_path, e)
        raise
    except Exception as e:
        logger.critical(f"Unhandled exception | input={input_path.name} | error={e}")
        raise BudgetConversionError(f"Critical pipeline failure: {e}") from e

def _route_to_quarantine(file_path: Path, error: Exception) -> None:
    """Explicit fallback routing for non-compliant or corrupted inputs."""
    quarantine_path = QUARANTINE_DIR / f"QUARANTINED_{file_path.name}"
    file_path.rename(quarantine_path)
    
    manifest = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "original_file": file_path.name,
        "error_type": error.__class__.__name__,
        "error_message": str(error),
        "routing_action": "QUARANTINE",
        "requires_manual_review": True
    }
    manifest_path = QUARANTINE_DIR / f"MANIFEST_{file_path.stem}.json"
    with open(manifest_path, "w", encoding="utf-8") as f:
        json.dump(manifest, f, indent=2)
    logger.warning(f"File quarantined | path={quarantine_path} | manifest={manifest_path}")