python scripts/baseline_diff.py baseline/golden_2025-07 baseline/post_refactor_legacy_2025-07

[ERROR] Old directory does not exist: baseline\golden_2025-07

# api/bigquery/identifiers.py
"""
BigQuery identifier builder with security validation.

Prevents SQL injection via environment variables by validating identifiers
against BigQuery naming rules and rejecting dangerous characters.
"""

import os
import re
import logging
from typing import Literal

logger = logging.getLogger(__name__)

# BigQuery naming patterns (per official documentation)
# Project: 6-30 chars, starts with lowercase letter, contains [a-z0-9-]
PROJECT_PATTERN = re.compile(r'^[a-z][a-z0-9-]{5,29}$')

# Dataset/Table: starts with letter or underscore, contains [a-zA-Z0-9_], no hyphens
DATASET_PATTERN = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_]*$')
TABLE_PATTERN = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_]*$')

# Dangerous characters that must be rejected
DANGEROUS_CHARS = ['`', ' ', '.', ';', '\n', '\r', '\t', "'", '"']


def validate_identifier(value: str, identifier_type: str) -> None:
    """
    Validate BigQuery identifier against naming rules.
    
    Args:
        value: Identifier value to validate
        identifier_type: Type of identifier ('project', 'dataset', or 'table')
    
    Raises:
        RuntimeError: If identifier is invalid or contains dangerous characters
    """
    if not isinstance(value, str):
        raise RuntimeError(f"{identifier_type} must be a string, got {type(value).__name__}")
    
    if not value:
        raise RuntimeError(f"{identifier_type} cannot be empty")
    
    # Check for dangerous characters
    for char in DANGEROUS_CHARS:
        if char in value:
            raise RuntimeError(
                f"{identifier_type} contains dangerous character '{char}': {value}"
            )
    
    # Validate against pattern
    if identifier_type == "project":
        if not PROJECT_PATTERN.match(value):
            raise RuntimeError(
                f"Invalid project ID format: {value} "
                "(must be 6-30 chars, start with lowercase letter, contain [a-z0-9-])"
            )
    elif identifier_type in ("dataset", "table"):
        pattern = DATASET_PATTERN if identifier_type == "dataset" else TABLE_PATTERN
        if not pattern.match(value):
            raise RuntimeError(
                f"Invalid {identifier_type} format: {value} "
                "(must start with letter/underscore, contain [a-zA-Z0-9_], no hyphens)"
            )
    else:
        raise ValueError(f"Unknown identifier_type: {identifier_type}")


def get_project_id() -> str:
    """Get GCP project ID from environment with validation."""
    project_id = os.getenv("GCP_PROJECT_ID", "payroll-bi-gauntlet")
    validate_identifier(project_id, "project")
    return project_id


def get_dataset_analytics() -> str:
    """Get analytics dataset name from environment with validation."""
    dataset = os.getenv("BQ_DATASET_ANALYTICS", "payroll_analytics")
    validate_identifier(dataset, "dataset")
    return dataset


def get_dataset_raw() -> str:
    """Get raw dataset name from environment with validation."""
    dataset = os.getenv("BQ_DATASET_RAW", "payroll_raw")
    validate_identifier(dataset, "dataset")
    return dataset


def get_dataset_processed() -> str:
    """Get processed dataset name from environment with validation."""
    dataset = os.getenv("BQ_DATASET_PROCESSED", "payroll_processed")
    validate_identifier(dataset, "dataset")
    return dataset


def bq_table(
    dataset_kind: Literal["analytics", "raw", "processed"],
    table: str
) -> str:
    """
    Build fully-qualified BigQuery table identifier with validation.
    
    Args:
        dataset_kind: Type of dataset ('analytics', 'raw', or 'processed')
        table: Table/view name
    
    Returns:
        Fully-qualified identifier: `project.dataset.table`
    
    Raises:
        RuntimeError: If any identifier is invalid
    """
    # Validate table name
    validate_identifier(table, "table")
    
    # Get project and dataset
    project_id = get_project_id()
    
    if dataset_kind == "analytics":
        dataset = get_dataset_analytics()
    elif dataset_kind == "raw":
        dataset = get_dataset_raw()
    elif dataset_kind == "processed":
        dataset = get_dataset_processed()
    else:
        raise ValueError(f"Unknown dataset_kind: {dataset_kind}")
    
    # Validate dataset
    validate_identifier(dataset, "dataset")
    
    # Return backticked identifier (BigQuery standard format)
    return f"`{project_id}.{dataset}.{table}`"


# Initialize and log effective configuration at module import
def _log_configuration():
    """Log effective dataset configuration once at startup."""
    try:
        project_id = get_project_id()
        dataset_analytics = get_dataset_analytics()
        dataset_raw = get_dataset_raw()
        dataset_processed = get_dataset_processed()
        
        logger.info("="*60)
        logger.info("BigQuery Dataset Configuration:")
        logger.info(f"  Project ID: {project_id}")
        logger.info(f"  Analytics Dataset: {dataset_analytics}")
        logger.info(f"  Raw Dataset: {dataset_raw}")
        logger.info(f"  Processed Dataset: {dataset_processed}")
        logger.info("="*60)
    except Exception as e:
        logger.error(f"Failed to initialize dataset configuration: {e}")
        raise


# Call on module import (fail fast if invalid)
_log_configuration()

def get_ceo_metrics(tenant_id: str, period_label: str) -> Dict:
    """
    Get CEO-level metrics for a specific period
    
    Args:
        tenant_id: Tenant ID for isolation
        period_label: Period label (YYYY-MM-01)
    
    Returns:
        Dict with CEO metrics
    
    Raises:
        BigQueryConnectionError: If BigQuery client unavailable or query fails
        BigQueryNoDataError: If no data found for the period
    """
    try:
        if client is None:
            raise BigQueryConnectionError("BigQuery client not available. Check GCP_PROJECT_ID environment variable and authentication.")
        
        # Query 1: Get base metrics from ceo_snapshot
        ceo_query = f"""
        SELECT
            total_businesses,
            total_employees,
            gross_payout,
            chargebacks,
            net_payout
        FROM {bq_table("analytics", "ceo_snapshot")}
        WHERE period_label = CAST(@period_label AS DATE)
          AND tenant_id = @tenant_id
        """
        
        # Query 2: Get agent commissions from stage3
        agent_query = f"""
        SELECT
            SUM(agent_total) as agent_commissions
        FROM {bq_table("processed", "stage3_snapshots")}
        WHERE period_label = CAST(@period_label AS DATE)
          AND tenant_id = @tenant_id
          AND tpa_applied = TRUE
        """
        
        job_config = bigquery.QueryJobConfig(
            query_parameters=[
                bigquery.ScalarQueryParameter("tenant_id", "STRING", tenant_id),
                bigquery.ScalarQueryParameter("period_label", "DATE", period_label),
            ]
        )
        
        logger.info(f"Getting real CEO metrics for tenant {tenant_id}, period {period_label}")
        
        # Execute both queries
        ceo_job = client.query(ceo_query, job_config=job_config)
        ceo_results = list(ceo_job.result())
        
        agent_job = client.query(agent_query, job_config=job_config)
        agent_results = list(agent_job.result())
        
        if not ceo_results:
            raise BigQueryNoDataError(f"No CEO metrics data found for period {period_label} and tenant {tenant_id}")
        
        ceo_result = ceo_results[0]
        agent_commissions = float(agent_results[0].agent_commissions or 0) if agent_results and agent_results[0].agent_commissions is not None else 0
        
        # Calculate commissions using canonical formula
        gross_payout = float(ceo_result.gross_payout or 0)
        chargebacks = float(ceo_result.chargebacks or 0)
        
        # Owner Commission = Gross - Chargebacks - Agent Commissions
        business_owner_commission = gross_payout - chargebacks - agent_commissions
        # ... rest of function

def get_growth_loss_summary_snapshot(tenant_id: str, period_label: str, period_type: str) -> Dict:
    """Get summary using snapshot comparison for QTD/YTD
    
    Args:
        tenant_id: Tenant identifier
        period_label: End period (YYYY-MM-01 format)
        period_type: 'qtd' or 'ytd'
    
    Returns:
        Dict with nested structure:
        {
            "summary": {
                "gained_employees": int,
                "lost_employees": int,
                "net_employee_change": int,
                "pct_employee_change": float
            },
            "businesses": [{
                "business_name": str,
                "net_employee_change": int,
                "employee_change_pct": float,
                ...
            }],
            "error": bool,
            "message": str | None
        }
    """
    if not client:
        error_msg = "BigQuery client not available"
        logger.error(error_msg)
        return {
            'summary': {
                'gained_employees': 0,
                'lost_employees': 0,
                'net_employee_change': 0,
                'pct_employee_change': 0
            },
            'businesses': [],
            'error': True,
            'message': error_msg
        }
    
    # Initialize baseline_str before try block to avoid NameError in except clause
    baseline_str = None
    
    try:
        # Defensive check: verify client is still available during execution
        if not client:
            raise RuntimeError("BigQuery client became unavailable during execution")
        
        from datetime import datetime, date
        
        # Calculate baseline date
        current_date = datetime.strptime(period_label, '%Y-%m-%d').date()
        
        if period_type == 'qtd':
            # First day of current quarter
            quarter_month = ((current_date.month - 1) // 3) * 3 + 1
            baseline_date = date(current_date.year, quarter_month, 1)
        elif period_type == 'ytd':
            # January 1st of current year
            baseline_date = date(current_date.year, 1, 1)
        else:
            raise ValueError(f"Invalid period_type: {period_type}")
        
        baseline_str = baseline_date.strftime('%Y-%m-%d')
        
        # Comprehensive logging before query execution
        logger.info("[SNAPSHOT DEBUG] Starting snapshot comparison query")
        logger.info(f"[SNAPSHOT DEBUG] period_type={period_type}, period_label={period_label}")
        logger.info(f"[SNAPSHOT DEBUG] baseline_date={baseline_date}, baseline_str={baseline_str}, tenant_id={tenant_id}")
        
        # Parameterized query (SECURE - no SQL injection)
        query = f"""
        WITH current_snap AS (
            SELECT 
                UPPER(TRIM(business_label)) AS business_label,
                COUNT(DISTINCT member_id) AS employees
            FROM {bq_table("raw", "stage1_snapshots")}
            WHERE tenant_id = @tenant_id 
              AND period_label = CAST(@current_period AS DATE)
              AND COALESCE(TRIM(business_label), '') != ''
            GROUP BY UPPER(TRIM(business_label))
        ),
        baseline_snap AS (
            SELECT 
                UPPER(TRIM(business_label)) AS business_label,
                COUNT(DISTINCT member_id) AS employees
            FROM {bq_table("raw", "stage1_snapshots")}
            WHERE tenant_id = @tenant_id 
              AND period_label = CAST(@baseline_period AS DATE)
              AND COALESCE(TRIM(business_label), '') != ''
            GROUP BY UPPER(TRIM(business_label))
        )
        SELECT 
            COALESCE(c.business_label, b.business_label) AS business_name,
            COALESCE(c.employees, 0) AS current_employees,
            COALESCE(b.employees, 0) AS previous_employees,
            COALESCE(c.employees, 0) - COALESCE(b.employees, 0) AS net_employee_change,
            SAFE_DIVIDE(
                COALESCE(c.employees, 0) - COALESCE(b.employees, 0),
                NULLIF(COALESCE(b.employees, 0), 0)
            ) * 100 AS employee_change_pct,
            CASE 
                WHEN b.employees IS NULL THEN 'New Business'
                WHEN c.employees IS NULL THEN 'Lost Business'
                WHEN c.employees > b.employees THEN 'Growth'
                WHEN c.employees < b.employees THEN 'Decline'
                ELSE 'Stable'
            END AS change_category
        FROM current_snap c
        FULL OUTER JOIN baseline_snap b ON c.business_label = b.business_label
        ORDER BY ABS(COALESCE(c.employees, 0) - COALESCE(b.employees, 0)) DESC
        """
        
        job_config = bigquery.QueryJobConfig(
            query_parameters=[
                bigquery.ScalarQueryParameter("tenant_id", "STRING", tenant_id),
                bigquery.ScalarQueryParameter("current_period", "DATE", period_label),
                bigquery.ScalarQueryParameter("baseline_period", "DATE", baseline_str),
            ]
        )
        
        query_job = client.query(query, job_config=job_config)
        results = [dict(row) for row in query_job.result()]
        
        # ... rest of function (calculates summary statistics)

def get_business_health_distribution(tenant_id: str, period_label: str) -> List[Dict]:
    """Get business health distribution (Growth/Decline/Stable/New)"""
    try:
        if client is None:
            logger.warning("BigQuery client not available, returning empty list")
            return []
        
        query = f"""
        WITH categorized AS (
          SELECT
            CASE
              WHEN previous_employees IS NULL THEN 'New Business'
              WHEN net_employee_change > 0 THEN 'Growth'
              WHEN net_employee_change < 0 THEN 'Decline'
              ELSE 'Stable'
            END AS category
          FROM {bq_table("analytics", "business_growth_loss")}
          WHERE period_label = CAST(@period_label AS DATE)
            AND tenant_id = @tenant_id
        )
        SELECT 
          category,
          COUNT(*) as count,
          ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER (), 1) as percentage
        FROM categorized
        GROUP BY category
        ORDER BY count DESC
        """
        
        job_config = bigquery.QueryJobConfig(
            query_parameters=[
                bigquery.ScalarQueryParameter("tenant_id", "STRING", tenant_id),
                bigquery.ScalarQueryParameter("period_label", "DATE", period_label),
            ]
        )
        
        query_job = client.query(query, job_config=job_config)
        results = [dict(row) for row in query_job.result()]
        
        logger.info(f"Fetched business health distribution: {len(results)} categories")
        return results
        
    except Exception as e:
        logger.error(f"Error fetching business health distribution: {str(e)}")
        return []

python scripts/generate_wiring_map.py

================================================================================
GENERATING WIRING MAP
================================================================================
Found 29 query functions with table references
[OK] Updated C:\Projects\payroll-pipeline-cbs\docs\DASHBOARD_WIRING_VALIDATION.md

================================================================================
SUMMARY
================================================================================
Widgets mapped: 12
Endpoints mapped: 7
Query functions found: 29
Output: C:\Projects\payroll-pipeline-cbs\docs\DASHBOARD_WIRING_VALIDATION.md
================================================================================

python scripts/check_shadow_datasets.py

================================================================================
SHADOW DATASET EXISTENCE CHECK (GATE 1)
================================================================================
Project ID: payroll-bi-gauntlet

[OK] ANALYTICS: payroll-bi-gauntlet.payroll_analytics_shadow EXISTS
[OK] RAW: payroll-bi-gauntlet.payroll_raw_shadow EXISTS
[OK] PROCESSED: payroll-bi-gauntlet.payroll_processed_shadow EXISTS

================================================================================
DECISION LOGIC:
================================================================================
[OK] Analytics shadow exists (REQUIRED)
[OK] Raw shadow exists - can use payroll_raw_shadow
   Set BQ_DATASET_RAW=payroll_raw_shadow
[OK] Processed shadow exists - can use payroll_processed_shadow
   Set BQ_DATASET_PROCESSED=payroll_processed_shadow

================================================================================
RECOMMENDED ENV VARS FOR CUTOVER:
================================================================================
BQ_DATASET_ANALYTICS=payroll_analytics_shadow  # REQUIRED
BQ_DATASET_RAW=payroll_raw_shadow  # Optional (shadow exists)
BQ_DATASET_PROCESSED=payroll_processed_shadow  # Optional (shadow exists)
================================================================================

BQ_DATASET_ANALYTICS=payroll_analytics_shadow  # REQUIRED
BQ_DATASET_RAW=payroll_raw_shadow  # Optional (shadow exists)
BQ_DATASET_PROCESSED=payroll_processed_shadow  # Optional (shadow exists)