hiffi-artist-intelligence/audit_system_integrity.py

"""
audit_system_integrity.py
Complete pre-deployment audit tool for HIFFI Artist Intelligence.
Checks CSV data hygiene, CPI engine calculations, Excel exports, and FastAPI imports.
"""

from pathlib import Path
import pandas as pd
import numpy as np

DATA_PATH = Path("data/hiffi_multiplatform_500.csv")

def run_pre_deployment_audit():
    print("==================================================")
    print("🔍 RUNNING PRE-DEPLOYMENT SYSTEM INTEGRITY AUDIT")
    print("==================================================\n")
    
    all_passed = True

    # ----------------------------------------------------
    # CHECK 1: CSV File Existence & Row Count
    # ----------------------------------------------------
    print("[1/5] Checking master CSV dataset...")
    if not DATA_PATH.exists():
        print("  ❌ FAIL: data/hiffi_multiplatform_500.csv does not exist!")
        return False
    
    df = pd.read_csv(DATA_PATH)
    if len(df) == 500:
        print(f"  ✅ PASS: Exact cohort size verified: {len(df)} rows.")
    else:
        print(f"  ⚠️ WARNING: Expected 500 rows, but found {len(df)} rows.")
        all_passed = False

    # ----------------------------------------------------
    # CHECK 2: Data Provenance & Sales Fields Hygiene
    # ----------------------------------------------------
    print("\n[2/5] Checking sales telemetry & status completeness...")
    valid_statuses = {"VERIFIED_CSPC", "VERIFIED_RIAA", "ESTIMATED_SEA"}
    actual_statuses = set(df["sales_data_status"].dropna().unique())
    invalid_statuses = actual_statuses - valid_statuses

    if not invalid_statuses:
        print(f"  ✅ PASS: All rows use clean governance statuses: {actual_statuses}")
    else:
        print(f"  ❌ FAIL: Unhandled/dirty statuses detected: {invalid_statuses}")
        all_passed = False

    # Check for NaN in total_eas_cspc
    nan_sales = df["total_eas_cspc"].isna().sum()
    if nan_sales == 0:
        print("  ✅ PASS: Zero NaN values in 'total_eas_cspc'.")
    else:
        print(f"  ❌ FAIL: Found {nan_sales} NaN entries in 'total_eas_cspc'!")
        all_passed = False

    # Check that sales rank is complete
    nan_sales_rank = df["cohort_sales_rank"].isna().sum()
    if nan_sales_rank == 0:
        print("  ✅ PASS: 100% of cohort_sales_rank populated.")
    else:
        print(f"  ❌ FAIL: Found {nan_sales_rank} missing sales ranks.")
        all_passed = False

    # ----------------------------------------------------
    # CHECK 3: Commercial Power Engine (CPI) Verification
    # ----------------------------------------------------
    print("\n[3/5] Testing CommercialPowerEngine on master data...")
    try:
        from core.cpi_engine import CommercialPowerEngine
        engine = CommercialPowerEngine()
        
        # Test CPI calculation
        cpi_series = engine.calculate_cpi(df)
        if len(cpi_series) == len(df) and not cpi_series.isna().any():
            print(f"  ✅ PASS: CPI calculated for all rows (Min: {cpi_series.min():.1f}, Max: {cpi_series.max():.1f}).")
        else:
            print("  ❌ FAIL: CPI calculation produced NaN values.")
            all_passed = False

        # Test Simulation (matched_only)
        sim_matched = engine.simulate_ranking(df[df["sales_data_status"] == "VERIFIED_CSPC"], commercial_weight=0.15, mode="matched_only")
        if not sim_matched.empty and "simulated_rank" in sim_matched.columns:
            print("  ✅ PASS: Simulation engine (matched_only) ran cleanly.")
        else:
            print("  ❌ FAIL: Simulation engine failed on matched_only.")
            all_passed = False

        # Test Simulation (global_redistribution)
        sim_global = engine.simulate_ranking(df, commercial_weight=0.15, mode="global_redistribution")
        if not sim_global.empty and "simulated_rank" in sim_global.columns:
            print("  ✅ PASS: Simulation engine (global_redistribution) ran cleanly.")
        else:
            print("  ❌ FAIL: Simulation engine failed on global_redistribution.")
            all_passed = False

    except Exception as e:
        print(f"  ❌ FAIL: Error running CommercialPowerEngine: {e}")
        all_passed = False

    # ----------------------------------------------------
    # CHECK 4: Excel Audit Workbook Exporter
    # ----------------------------------------------------
    print("\n[4/5] Testing Excel multi-tab workbook exporter...")
    try:
        from exporters.excel_workbook import generate_executive_sales_workbook
        excel_bytes = generate_executive_sales_workbook(df, commercial_weight=0.15, mode="global_redistribution")
        if excel_bytes and len(excel_bytes) > 1000:
            print(f"  ✅ PASS: Excel workbook generated successfully ({len(excel_bytes):,} bytes).")
        else:
            print("  ❌ FAIL: Excel workbook payload is empty.")
            all_passed = False
    except Exception as e:
        print(f"  ❌ FAIL: Error in generate_executive_sales_workbook: {e}")
        all_passed = False

    # ----------------------------------------------------
    # CHECK 5: FastAPI App Loadability
    # ----------------------------------------------------
    print("\n[5/5] Testing FastAPI routes and schemas in api.py...")
    try:
        import api
        routes = [route.path for route in api.app.routes]
        expected_routes = ["/health", "/telemetry/coverage", "/artists", "/simulate"]
        missing = [r for r in expected_routes if r not in routes]
        if not missing:
            print(f"  ✅ PASS: All required API endpoints present: {expected_routes}")
        else:
            print(f"  ❌ FAIL: Missing API routes: {missing}")
            all_passed = False
    except Exception as e:
        print(f"  ❌ FAIL: Error importing api.py: {e}")
        all_passed = False

    # ----------------------------------------------------
    # FINAL VERDICT
    # ----------------------------------------------------
    print("\n==================================================")
    if all_passed:
        print("🎯 AUDIT RESULT: 100% CLEAN! ZERO DISCREPANCIES.")
        print("All modules, formulas, exports, and APIs are verified.")
    else:
        print("⚠️ AUDIT RESULT: ISSUES DETECTED. Review logs above.")
    print("==================================================")

if __name__ == "__main__":
    run_pre_deployment_audit()