Parsing Complex University Excel Grant Templates with pandas

import pandas as pd
import openpyxl
from pathlib import Path
import hashlib
import logging
import shutil
from datetime import datetime
from typing import Dict, Any

# Audit-safe logging with ISO timestamps and structured severity levels
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s | %(levelname)s | %(message)s',
    handlers=[
        logging.FileHandler('grant_parsing_audit.log'),
        logging.StreamHandler()
    ]
)

# Canonical compliance schema enforced across all university grant templates
# Aligns with NIH/NSF indirect cost tracking and OSHA/EPA lab inventory requirements
CANONICAL_SCHEMA: Dict[str, Any] = {
    'Award Number': str,
    'PI Last Name': str,
    'Department Code': str,
    'Direct Costs': float,
    'Indirect Cost Rate': float,
    'Start Date': 'datetime64[ns]',
    'End Date': 'datetime64[ns]'
}

def compute_sha256(filepath: Path) -> str:
    """Generate SHA-256 checksum for immutable file tracking."""
    sha256 = hashlib.sha256()
    with open(filepath, 'rb') as f:
        for chunk in iter(lambda: f.read(8192), b''):
            sha256.update(chunk)
    return sha256.hexdigest()

def resolve_merged_headers(workbook_path: Path, max_header_rows: int = 4) -> list[str]:
    """
    Resolve openpyxl merged cells into a flat, forward-filled header list.
    Handles multi-row headers common in NIH/NSF budget justification templates.
    """
    wb = openpyxl.load_workbook(workbook_path, read_only=True, data_only=True)
    ws = wb.active
    if ws is None:
        raise ValueError("No active worksheet found in grant template.")

    # Extract raw header rows
    raw_rows = []
    for row_idx in range(1, max_header_rows + 1):
        row_data = [cell.value for cell in next(ws.iter_rows(min_row=row_idx, max_row=row_idx))]
        raw_rows.append(row_data)

    col_count = max(len(r) for r in raw_rows)
    flat_header = [''] * col_count

    # Forward-fill across columns
    for row in raw_rows:
        for i, val in enumerate(row):
            if val is not None:
                flat_header[i] = val

    # Resolve merged cell ranges by propagating the anchor value
    for merged_range in ws.merged_cells.ranges:
        min_c, max_c = merged_range.min_col - 1, merged_range.max_col - 1
        anchor_val = None
        for r_idx, row in enumerate(raw_rows):
            for c_idx in range(min_c, max_c + 1):
                if c_idx < len(row) and row[c_idx] is not None:
                    anchor_val = row[c_idx]
                    break
            if anchor_val is not None:
                break
        if anchor_val is not None:
            for c in range(min_c, max_c + 1):
                if c < len(flat_header):
                    flat_header[c] = anchor_val

    wb.close()
    return [str(h).strip() if h else f"Column_{i}" for i, h in enumerate(flat_header)]

def validate_and_cast(df: pd.DataFrame, schema: Dict[str, Any]) -> pd.DataFrame:
    """Enforce canonical schema with strict type coercion and compliance validation."""
    missing = set(schema.keys()) - set(df.columns)
    if missing:
        raise KeyError(f"Missing mandatory compliance fields: {missing}")

    for col, dtype in schema.items():
        if dtype == 'datetime64[ns]':
            df[col] = pd.to_datetime(df[col], errors='coerce')
        else:
            df[col] = df[col].astype(dtype)

    # Drop rows with invalid dates (critical for NSF/NIH period-of-performance tracking)
    df.dropna(subset=['Start Date', 'End Date'], inplace=True)
    return df

def process_grant_template(
    source_dir: Path, 
    output_dir: Path, 
    quarantine_dir: Path, 
    schema: Dict[str, Any] = CANONICAL_SCHEMA
) -> None:
    """
    Idempotent pipeline for batch parsing university grant templates.
    Safe for repeated execution; skips already-processed files based on checksum.
    """
    for dir_path in (source_dir, output_dir, quarantine_dir):
        dir_path.mkdir(parents=True, exist_ok=True)

    for excel_file in source_dir.glob('*.xlsx'):
        file_hash = compute_sha256(excel_file)
        output_file = output_dir / f"{excel_file.stem}_{file_hash[:8]}.csv"

        # Idempotency guard
        if output_file.exists():
            logging.info(f"Skipping {excel_file.name}: Output already exists (hash: {file_hash[:8]})")
            continue

        try:
            logging.info(f"Processing {excel_file.name}...")
            headers = resolve_merged_headers(excel_file)
            
            # Read without assuming header row index
            df = pd.read_excel(excel_file, header=None, engine='openpyxl')
            df.columns = headers
            df = df.iloc[1:].reset_index(drop=True)

            # Strip whitespace from string columns
            df = df.apply(lambda x: x.str.strip() if x.dtype == 'object' else x)

            # Schema validation & type casting
            df_validated = validate_and_cast(df, schema)

            # Export with deterministic formatting
            df_validated.to_csv(output_file, index=False, date_format='%Y-%m-%d')
            logging.info(f"Successfully parsed and exported {output_file.name}")

        except Exception as e:
            logging.error(f"Failed to parse {excel_file.name}: {e}")
            quarantine_path = quarantine_dir / f"QUARANTINE_{excel_file.name}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
            shutil.copy2(excel_file, quarantine_path)
            logging.warning(f"File routed to quarantine: {quarantine_path}")