processing.py

import requests
import pandas as pd
import io
import json
import re
import clevercsv
import csv
from config import supabase, GEMINI_API_KEY
from typing import Tuple, Dict, Any, Optional
import os
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.backends import default_backend
import google.generativeai as genai
from SAGAR_QC.qc_pipeline import run_qc_pipeline, generate_quality_report
from SAGAR_QC.compliance_checker import run_compliance_check_from_bytes
from SAGAR_QC.compliance_report import generate_compliance_report
from logging_system import add_log, LogLevel

# --- HELPER: Pure Pandas Cleaning ---

def clean_column_names(df: pd.DataFrame) -> pd.DataFrame:
    """
    Cleans column names by stripping whitespace and removing special chars,
    BUT preserves the original Casing (Title Case).
    """
    new_columns = []
    for col in df.columns:
        s = str(col).strip()
        # Remove quotes commonly found in headers
        s = s.replace('"', '').replace("'", "")
        new_columns.append(s)
    
    df.columns = new_columns
    return df

# --- HELPER: JSON Flattener ---

def flatten_json_to_df(text_content: str) -> pd.DataFrame:
    """Branch A Helper: Flattens nested JSON into a DataFrame."""
    try:
        data = json.loads(text_content)
        # If it's a single object, wrap in list
        if isinstance(data, dict):
            data = [data]
        # Pandas json_normalize handles nested structures (e.g., geometry.coordinates)
        return pd.json_normalize(data)
    except:
        return None

# --- TIER 1: FAST INTELLIGENT CLEANING (THE MASTER ROUTER) ---

def fast_intelligent_cleaning(file_bytes: bytes, filename: str) -> bytes:
    """
    Tier 1 Processing: A robust multi-branch router that detects file structure.
    
    BRANCH A: JSON Data -> Auto-flatten.
    BRANCH B: Sensor Logs -> Regex cleaning ($GPS, ADCP).
    BRANCH C: Structured CSV/TSV -> CleverCSV parsing.
    BRANCH D: "Data Miner" -> Finds data blocks in unknown formats.
    """
    print(f"⚡ Attempting Fast Intelligent Cleaning for {filename}...")
    
    # 1. Decode Bytes
    text_content = ""
    encodings = ['utf-8', 'latin-1', 'iso-8859-1', 'cp1252']
    
    for encoding in encodings:
        try:
            text_content = file_bytes.decode(encoding)
            break
        except UnicodeDecodeError:
            continue
        
    if not text_content:
        raise ValueError("Could not decode file with standard encodings")

    lines = text_content.split('\n')
    first_100 = lines[:100]
    clean_snippet = '\n'.join([l.strip() for l in first_100 if l.strip()])

    df = None
    
    # --- BRANCH A: JSON DETECTION ---
    # Check if file starts with valid JSON indicators
    if clean_snippet.startswith('{') or clean_snippet.startswith('['):
        print("📍 Branch A: Detected JSON format")
        df = flatten_json_to_df(text_content)

    # --- BRANCH B: SENSOR LOG DETECTION ---
    # Explicit check for the messy formats we know (ADCP, GPS)
    if df is None:
        is_sensor = False
        sensor_keywords = ["$GPS", "$MET", "$SST"]
        
        for line in first_100:
            if any(k in line for k in sensor_keywords) or line.strip().startswith("Ens"):
                is_sensor = True
                break
        
        if is_sensor:
            print("📍 Branch B: Detected Sensor Log format")
            # 1. Smart Skip: Find the header row
            header_idx = 0
            for i, line in enumerate(lines[:100]):
                if line.startswith("Ens") or line.startswith("$GPS"):
                    header_idx = i
                    break
            
            content_to_parse = '\n'.join(lines[header_idx:])
            
            # 2. Regex Cleaning
            content_to_parse = content_to_parse.replace("$GPS", "").replace("$MET", "").replace("$SST", "")
            # Fix detached coords (0958.257 N -> 0958.257N)
            content_to_parse = re.sub(r'(\d+\.?\d*)\s+([NSEW])\b', r'\1\2', content_to_parse)
            content_to_parse = re.sub(r'[ \t]+', ' ', content_to_parse) # Normalize to single space
            
            # 3. Parse
            df = pd.read_csv(io.StringIO(content_to_parse), sep=r"\s+", on_bad_lines='skip', engine='python')

    # --- BRANCH C: STRUCTURED DATA DETECTION ---
    # Check for Darwin Core / Standard headers
    if df is None:
        structured_keywords = ["institutionCode", "scientificName", "basisOfRecord", "decimalLatitude", "collectionCode", "id"]
        is_structured = False
        for line in first_100:
            if any(k in line for k in structured_keywords):
                is_structured = True
                break
    
        if is_structured:
            print("📍 Branch C: Detected Structured Data (Darwin Core/OBIS)")
            try:
                # CleverCSV: Use read_csv_strings for StringIO content
                # Note: clevercsv.read_dataframe expects a file path, not StringIO
                # So we'll use pandas with clevercsv's detect_dialect if needed
                import tempfile
                with tempfile.NamedTemporaryFile(mode='w', suffix='.csv', delete=False, encoding='utf-8') as tmp_file:
                    tmp_file.write(text_content)
                    tmp_path = tmp_file.name
                
                try:
                    df = clevercsv.read_dataframe(tmp_path)
                finally:
                    # Clean up temp file
                    if os.path.exists(tmp_path):
                        os.remove(tmp_path)
            except Exception as e:
                print(f"⚠️ CleverCSV failed, using pandas fallback: {e}")
                # Fallback to python engine sniffing
                df = pd.read_csv(io.StringIO(text_content), sep=None, engine='python', on_bad_lines='skip')

    # --- BRANCH D: THE "DATA MINER" (Universal Fallback) ---
    # If unknown, scan for the "densest" block of data
    if df is None:
        print("📍 Branch D: Unknown format. Initiating Data Miner...")
        
        # Heuristic: The data section usually has the most consistent number of separators
        max_cols = 0
        best_sep = ','
        start_row = 0
        
        # Check common separators
        for sep in [',', '\t', ';', '|']:
            consistent_rows = 0
            temp_cols = 0
            for i, line in enumerate(lines[:50]):
                cols = len(line.split(sep))
                if cols > 1:
                    if cols > temp_cols:
                        temp_cols = cols
                        start_row = i
            
            if temp_cols > max_cols:
                max_cols = temp_cols
                best_sep = sep
        
        print(f"⛏️  Miner found best separator '{best_sep}' with {max_cols} columns starting row {start_row}")
        
        # Try to parse from that row
        try:
            content_to_parse = '\n'.join(lines[start_row:])
            df = pd.read_csv(io.StringIO(content_to_parse), sep=best_sep, on_bad_lines='skip', engine='python')
        except Exception as e:
            print(f"Miner failed: {e}")

    if df is None:
        raise ValueError("All parsing branches failed.")

    # --- FINAL CLEANUP (Applied to all branches) ---
    try:
        df = clean_column_names(df)
        
        # Drop rows where all columns are empty
        df = df.dropna(how='all', axis=0).dropna(how='all', axis=1)
        
        # Drop "unit" rows (common in scientific logs)
        if len(df) > 0:
            first_row_vals = df.iloc[0].astype(str).values
            # If >50% of values are very short strings (< 4 chars), it's likely a unit row (m, cm, %, etc)
            short_vals = sum(1 for v in first_row_vals if len(v) < 4 and v.lower() not in ['yes', 'no', 'nan'])
            if short_vals > len(df.columns) / 2:
                print("🧹 Dropping detected unit row...")
                df = df.iloc[1:]

    except Exception as e:
        print(f"Pandas cleaning warning: {e}")

    # Export back to clean CSV bytes
    output_buffer = io.StringIO()
    df.to_csv(output_buffer, index=False)
    return output_buffer.getvalue().encode('utf-8')


# --- TIER 2: GENERATIVE AI CLEANING (Gemini) ---

def preprocess_with_gemini(file_bytes: bytes, filename: str) -> bytes:
    """Tier 2 Processing: AI-based cleaning fallback."""
    try:
        if not GEMINI_API_KEY:
            raise ValueError("GEMINI_API_KEY not configured.")
        
        try:
            file_text = file_bytes.decode('utf-8')
        except UnicodeDecodeError:
            file_text = file_bytes.decode('latin-1', errors='ignore')
        
        max_chars = 150000
        if len(file_text) > max_chars:
            header_lines = file_text.split('\n')[:50]
            last_lines = file_text.split('\n')[-50:]
            preview_text = '\n'.join(header_lines) + '\n...[trimmed]...\n' + '\n'.join(last_lines)
        else:
            preview_text = file_text
        
        prompt = f"""Act as a Data Engineering API. Convert this raw file content into a standard CSV.
        
File: {filename}
Content Snippet:
{preview_text}

STRICT RULES:
1. Detect the format (TSV, CSV, or Log).
2. Identify the real header row (ignore metadata above it).
3. If it is a sensor log ($GPS), remove tags and merge coordinates (10 N -> 10N).
4. If it is a standard dataset, respect quoted fields (e.g. "Smith, 2004").
5. OUTPUT ONLY THE RAW CSV DATA. No markdown.
"""

        model = genai.GenerativeModel('gemini-2.5-flash')
        print(f"🤖 Calling Gemini 2.5 Flash for {filename}...")
        response = model.generate_content(prompt)
        
        csv_content = response.text.strip()
        
        if csv_content.startswith("```"):
            csv_content = re.sub(r"^```(csv)?|```$", "", csv_content, flags=re.MULTILINE).strip()
            
        if len(csv_content) < 10:
            raise ValueError("Gemini produced invalid output")
            
        return csv_content.encode('utf-8')
        
    except Exception as e:
        raise ValueError(f"Gemini processing failed: {e}")


# --- TIER 3: RULE BASED FALLBACK ---

def clean_csv_content(file_bytes: bytes) -> bytes:
    """Tier 3 Processing: Basic rule-based string manipulation."""
    try:
        text = file_bytes.decode('utf-8', errors='replace')
        text = text.replace("$GPS", "").replace("$MET", "").replace("$SST", "")
        text = re.sub(r'(\d+\.?\d*)\s+([NSEW])\b', r'\1\2', text)
        text = re.sub(r'[ \t]+', ',', text) 
        return text.encode('utf-8')
    except:
        return file_bytes


# --- DATA PROCESSING HELPERS ---

def process_occurrence_data(file_bytes: bytes) -> pd.DataFrame:
    buffer = io.BytesIO(file_bytes)
    try:
        # Standard CSV read
        df = pd.read_csv(buffer, on_bad_lines='skip', engine='python')
    except:
        buffer.seek(0)
        df = pd.read_csv(buffer, sep=None, on_bad_lines='skip', engine='python')
    
    # Try multiple date columns
    date_cols = ['Date', 'date', 'eventDate', 'First Ensemble Date', 'Ens Date']
    for dc in date_cols:
        if dc in df.columns:
            # Coerce dates, handling various formats
            df[dc] = pd.to_datetime(df[dc], errors='coerce', utc=True)

    # Coerce numerics
    # Handle Lat/Lon carefully - if they were string with N/E, extract float
    for col in ['Latitude', 'Longitude', 'decimalLatitude', 'decimalLongitude']:
        if col in df.columns:
             if df[col].dtype == 'object':
                 df[col] = df[col].astype(str).str.extract(r'(\d+\.?\d*)').astype(float)
             else:
                 df[col] = pd.to_numeric(df[col], errors='coerce')

    cols_to_numeric = ['Speed', 'depth', 'minimumDepthInMeters', 'maximumDepthInMeters']
    for col in cols_to_numeric:
        if col in df.columns:
            df[col] = pd.to_numeric(df[col], errors='coerce')
    
    df = df.dropna(axis=1, how='all')
    return df

def process_tabular_data(file_bytes: bytes) -> Tuple[pd.DataFrame, str, Dict[str, Any]]:
    df = process_occurrence_data(file_bytes)
    metadata_payload = {
        "columns": list(df.columns),
        "inferred_types": {col: str(dtype) for col, dtype in df.dtypes.items()},
        "total_rows": len(df)
    }
    return df, "tabular", metadata_payload

def process_netcdf_data(file_bytes: bytes) -> Tuple[pd.DataFrame, str, Dict[str, Any]]:
    import xarray as xr
    buffer = io.BytesIO(file_bytes)
    ds = xr.open_dataset(buffer)
    try:
        df = ds.to_dataframe().reset_index()
        metadata_payload = {
            "dimensions": dict(ds.dims),
            "variables": list(ds.data_vars.keys()),
            "attributes": ds.attrs
        }
    finally:
        ds.close()
    df = df.dropna(axis=1, how='all')
    return df, "netcdf", metadata_payload

def _sanitize_dataframe_for_parquet(df: pd.DataFrame) -> pd.DataFrame:
    sanitized = df.copy()
    sanitized.columns = [str(c).strip() for c in sanitized.columns]
    return sanitized

# --- AES ENCRYPTION HELPERS ---

def _get_aes_key() -> bytes:
    """
    Get AES key from env as hex string.
    Example: export AES_ENCRYPTION_KEY='00112233...ffeedd'
    Length must be 16/24/32 bytes (128/192/256-bit AES).
    """
    key_hex = os.environ.get("AES_ENCRYPTION_KEY")
    if not key_hex:
        raise ValueError("AES_ENCRYPTION_KEY environment variable not set")

    key = bytes.fromhex(key_hex)
    if len(key) not in (16, 24, 32):
        raise ValueError("AES_ENCRYPTION_KEY must be 16, 24 or 32 bytes (in hex)")

    return key


def encrypt_bytes_aes_gcm(plaintext: bytes) -> bytes:
    """
    Encrypt bytes with AES-GCM.
    Stored format:  [12-byte nonce][16-byte tag][ciphertext...]
    """
    key = _get_aes_key()
    nonce = os.urandom(12)  # GCM standard nonce size

    encryptor = Cipher(
        algorithms.AES(key),
        modes.GCM(nonce),
        backend=default_backend()
    ).encryptor()

    ciphertext = encryptor.update(plaintext) + encryptor.finalize()
    tag = encryptor.tag  # 16 bytes

    return nonce + tag + ciphertext


def decrypt_bytes_aes_gcm(blob: bytes) -> bytes:
    """
    Decrypt bytes encrypted with encrypt_bytes_aes_gcm.
    Expects: [nonce(12)][tag(16)][ciphertext]
    """
    key = _get_aes_key()

    nonce = blob[:12]
    tag = blob[12:28]
    ciphertext = blob[28:]

    decryptor = Cipher(
        algorithms.AES(key),
        modes.GCM(nonce, tag),
        backend=default_backend()
    ).decryptor()

    plaintext = decryptor.update(ciphertext) + decryptor.finalize()
    return plaintext


def upload_parquet_to_storage(bucket_name: str, remote_file_path: str, data: pd.DataFrame, encrypt: bool = True):
    """
    Upload parquet file to Supabase storage.
    
    Args:
        bucket_name: Name of the Supabase storage bucket
        remote_file_path: Path where the file should be stored
        data: DataFrame to convert to parquet
        encrypt: If True, encrypt the file and add .enc extension. If False, upload unencrypted.
    
    Returns:
        Upload result from Supabase
    """
    # 1. Convert DF → Parquet bytes
    parquet_bytes = data.to_parquet(engine='pyarrow')

    if encrypt:
        # 2. Encrypt with AES-GCM
        encrypted_bytes = encrypt_bytes_aes_gcm(parquet_bytes)

        # Change extension so you know this is encrypted
        # e.g. myfile.parquet.enc
        if not remote_file_path.endswith(".enc"):
            remote_file_path = remote_file_path + ".enc"

        # 3. Upload encrypted blob
        result = supabase.storage.from_(bucket_name).upload(
            path=remote_file_path,
            file=encrypted_bytes,
            file_options={"content-type": "application/octet-stream", "upsert": "true"}
        )
    else:
        # Upload unencrypted parquet file
        result = supabase.storage.from_(bucket_name).upload(
            path=remote_file_path,
            file=parquet_bytes,
            file_options={"content-type": "application/x-parquet", "upsert": "true"}
        )
    
    return result


def create_bounding_box(df: pd.DataFrame) -> str | None:
    lat_col = next((c for c in df.columns if c.lower() in ['latitude', 'decimallatitude']), None)
    lon_col = next((c for c in df.columns if c.lower() in ['longitude', 'decimallongitude']), None)

    if lat_col and lon_col:
        lat_series = pd.to_numeric(df[lat_col], errors='coerce')
        lon_series = pd.to_numeric(df[lon_col], errors='coerce')
        valid_idx = lat_series.notna() & lon_series.notna()
        if valid_idx.any():
            min_lat, max_lat = lat_series[valid_idx].min(), lat_series[valid_idx].max()
            min_lon, max_lon = lon_series[valid_idx].min(), lon_series[valid_idx].max()
            return f"POLYGON(({min_lon} {min_lat}, {max_lon} {min_lat}, {max_lon} {max_lat}, {min_lon} {max_lat}, {min_lon} {min_lat}))"
    return None

def update_metadata_in_db(
    original_filename: str, 
    processed_file_location: str, 
    status: str, 
    file_type: str, 
    metadata_payload: dict, 
    geom: str | None,
    superset_view: str | None = None,
    superset_dash_id: str | None = None
):
    if superset_view and superset_dash_id:
        metadata_payload["superset_view"] = superset_view
        metadata_payload["superset_dashboard_id"] = superset_dash_id

    new_row = {
        'original_filename': original_filename,
        'processed_file_location': processed_file_location,
        'status': status,
        'file_type': file_type,
        'metadata_payload': metadata_payload,
        'geom': geom,
        'superset_view_name': superset_view,
        'superset_dashboard_id': superset_dash_id
    }
    response = supabase.table('metadata_sagar').insert(new_row).execute()
    return response


def upload_netcdf_to_storage(bucket_name: str, remote_file_path: str, file_bytes: bytes):
    """
    Upload netCDF file directly to Supabase storage (unencrypted).
    
    Note: netCDF files are stored unencrypted, similar to unencrypted Parquet files,
    to allow direct access by tools that need to read netCDF format.
    
    Args:
        bucket_name: Name of the Supabase storage bucket
        remote_file_path: Path where the file should be stored
        file_bytes: Raw bytes of the netCDF file
    
    Returns:
        Upload result from Supabase
    """
    result = supabase.storage.from_(bucket_name).upload(
        path=remote_file_path,
        file=file_bytes,
        file_options={"content-type": "application/x-netcdf", "upsert": "true"}
    )
    return result


def update_netcdf_in_db(
    original_filename: str,
    file_location: str,
    status: str,
    compliance_report: dict,
    metadata_payload: dict
):
    """
    Store netCDF file metadata in netcdf_files table.
    
    Args:
        original_filename: Original filename
        file_location: Location in Supabase storage
        status: Processing status
        compliance_report: Full compliance report
        metadata_payload: Additional metadata (dimensions, variables, attributes)
    
    Returns:
        Database response
    """
    new_row = {
        'original_filename': original_filename,
        'file_location': file_location,
        'status': status,
        'file_type': 'netcdf',
        'compliance_report': compliance_report,
        'metadata_payload': metadata_payload
    }
    response = supabase.table('netcdf_files').insert(new_row).execute()
    return response

def call_brain_api(filename, bucket_id):
    brain_url = "https://sagar-superset-brain.vercel.app/register-dataset" 
    payload = {"filename": filename, "bucket_id": bucket_id}
    try:
        print(f"🧠 Calling Brain API for {filename}...")
        response = requests.post(brain_url, json=payload, timeout=10)
        response.raise_for_status()
        data = response.json()
        print(f"✅ Brain Success: Dashboard {data.get('dashboard_id')}")
        return data
    except Exception as e:
        print(f"⚠️ Brain API Failed: {e}")
        return None

def call_rag_embedding_api(filename, bucket_id):
    """
    Call the RAG service to generate embeddings for a newly uploaded file.
    Similar to call_brain_api(), this triggers the embedding generation process.
    """
    import os
    # Get RAG API URL from environment or use default
    rag_url = os.environ.get("RAG_API_URL", "https://rag.nikare.in/generate-embeddings")
    payload = {"filename": filename, "bucket_name": bucket_id}
    try:
        print(f"🔄 Calling RAG Embedding API for {filename}...")
        response = requests.post(rag_url, json=payload, timeout=60)  # Longer timeout for embedding generation
        response.raise_for_status()
        data = response.json()
        records_processed = data.get('records_processed', 0)
        print(f"✅ RAG Embedding Success: Processed {records_processed} records")
        return data
    except Exception as e:
        print(f"⚠️ RAG Embedding API Failed: {e}")
        return None

# --- MAIN ORCHESTRATOR ---

def process_csv_to_parquet(csv_bytes: bytes, original_filename: str, job_id: Optional[str] = None):
    import re
    processed_bucket = 'processed-data'
    if '.' in original_filename:
        processed_filename = re.sub(r'\.[^/.]+$', '.parquet', original_filename)
    else:
        processed_filename = f"{original_filename}.parquet"
    
    try:
        add_log(LogLevel.INFO, f"🚀 Starting Pipeline for {original_filename}", original_filename, job_id)
        
        lower_name = original_filename.lower()
        is_netcdf = lower_name.endswith(('.nc', '.nc4'))

        # --- NETCDF FILES: Special handling (compliance checker only, no QC/Superset/RAG) ---
        # Process netCDF files FIRST, before any cleaning attempts
        if is_netcdf:
            add_log(LogLevel.INFO, "📋 Processing netCDF file - Running Compliance Checker only...", original_filename, job_id)
            
            # Extract metadata from netCDF file
            try:
                import xarray as xr
                buffer = io.BytesIO(csv_bytes)
                ds = xr.open_dataset(buffer)
                metadata_payload = {
                    "dimensions": dict(ds.sizes),  # Use sizes instead of dims to avoid FutureWarning
                    "variables": list(ds.data_vars.keys()),
                    "coordinates": list(ds.coords.keys()),
                    "attributes": ds.attrs
                }
                ds.close()
            except Exception as meta_error:
                error_msg = f"⚠️ Could not extract netCDF metadata: {meta_error}"
                print(error_msg)
                add_log(LogLevel.WARNING, error_msg, original_filename, job_id)
                metadata_payload = {}
            
            # Run IOOS Compliance Checker
            compliance_results = None
            add_log(LogLevel.INFO, "📋 Running IOOS Compliance Checker (CF/ACDD standards)...", original_filename, job_id)
            print("📋 Running IOOS Compliance Checker (CF/ACDD standards)...")
            try:
                compliance_results = run_compliance_check_from_bytes(
                    csv_bytes,  # Use original bytes for netCDF
                    original_filename,
                    checker_names=['acdd', 'cf'],  # Run both ACDD and CF checks
                    criteria='normal'
                )
                
                if compliance_results.get('overall_passed', False):
                    success_msg = f"✅ Compliance Checker: PASSED ({compliance_results['summary']['passed_checkers']}/{compliance_results['summary']['total_checkers']} checkers passed)"
                    print(success_msg)
                    add_log(LogLevel.SUCCESS, success_msg, original_filename, job_id)
                else:
                    warning_msg = f"⚠️ Compliance Checker: FAILED ({compliance_results['summary']['failed_checkers']}/{compliance_results['summary']['total_checkers']} checkers failed)"
                    print(warning_msg)
                    add_log(LogLevel.WARNING, warning_msg, original_filename, job_id)
                
            except Exception as cc_error:
                error_msg = f"⚠️ Compliance Checker Error: {cc_error}"
                print(error_msg)
                add_log(LogLevel.ERROR, error_msg, original_filename, job_id)
                import traceback
                traceback.print_exc()
                compliance_results = {
                    'error': str(cc_error),
                    'overall_passed': False
                }
            
            # Generate compliance report
            compliance_report = generate_compliance_report(
                compliance_results,
                original_filename,
                additional_info={
                    'file_type': 'netcdf',
                    'metadata': metadata_payload
                }
            )
            
            # Upload netCDF file directly to storage (unencrypted, no conversion to parquet)
            netcdf_bucket = 'processed-data'
            netcdf_file_path = original_filename  # Keep original filename with .nc extension
            upload_msg = f"📤 Uploading netCDF file (unencrypted) to {netcdf_bucket} as {netcdf_file_path}..."
            print(upload_msg)
            add_log(LogLevel.INFO, upload_msg, original_filename, job_id)
            upload_result = upload_netcdf_to_storage(netcdf_bucket, netcdf_file_path, csv_bytes)
            
            # Store in netcdf_files table
            update_netcdf_in_db(
                original_filename=original_filename,
                file_location=netcdf_file_path,
                status='ready',
                compliance_report=compliance_report,
                metadata_payload=metadata_payload
            )
            
            success_msg = "🎉 netCDF Pipeline completed successfully!"
            print(success_msg)
            add_log(LogLevel.SUCCESS, success_msg, original_filename, job_id)
            return {
                "status": "success",
                "processed_file": netcdf_file_path,
                "file_type": "netcdf",
                "metadata": metadata_payload,
                "compliance_report": compliance_report,
                "upload_result": upload_result
            }
        
        # --- NON-NETCDF FILES: Standard processing (QC, Superset, RAG) ---
        # Skip cleaning for netCDF files - only process tabular data
        cleaned_csv_bytes = csv_bytes
        
        # Apply cleaning only for non-netCDF files
        try:
            # Tier 1 with 4-Branch Detection
            add_log(LogLevel.INFO, "Starting Tier 1: Fast Intelligent Cleaning...", original_filename, job_id)
            cleaned_csv_bytes = fast_intelligent_cleaning(csv_bytes, original_filename)
            success_msg = "✅ Tier 1: Fast Intelligent Cleaning Successful"
            print(success_msg)
            add_log(LogLevel.SUCCESS, success_msg, original_filename, job_id)
        except Exception as e1:
            error_msg = f"❌ Tier 1 Failed: {e1}"
            print(error_msg)
            add_log(LogLevel.WARNING, error_msg, original_filename, job_id)
            import traceback
            traceback.print_exc()
            try:
                fallback_msg = "⚠️ Falling back to Tier 2 (Gemini)..."
                print(fallback_msg)
                add_log(LogLevel.INFO, fallback_msg, original_filename, job_id)
                cleaned_csv_bytes = preprocess_with_gemini(csv_bytes, original_filename)
            except Exception as e2:
                error_msg2 = f"❌ Tier 2 Failed: {e2}"
                print(error_msg2)
                add_log(LogLevel.WARNING, error_msg2, original_filename, job_id)
                cleaned_csv_bytes = clean_csv_content(csv_bytes)
        
        # Process tabular data
        cleaned_df, file_type, metadata_payload = process_tabular_data(cleaned_csv_bytes)
        
        # --- SAGAR-QC: Run Quality Control Pipeline ---
        qc_msg = "🔍 Running SAGAR-QC Quality Control..."
        print(qc_msg)
        add_log(LogLevel.INFO, qc_msg, original_filename, job_id)
        try:
            # Run QC pipeline (adds 'qc_flag' column to DataFrame)
            cleaned_df, qc_stats = run_qc_pipeline(cleaned_df, config=None)
            
            # Generate quality report
            quality_report = generate_quality_report(
                qc_stats,
                original_filename,
                additional_info={
                    'file_type': file_type,
                    'total_columns': len(cleaned_df.columns)
                }
            )
            
            # Add QC summary to metadata
            metadata_payload['qc_summary'] = {
                'overall_quality_score': qc_stats['quality_metrics']['overall_quality_score'],
                'quality_status': quality_report['summary']['quality_status'],
                'flag_distribution': qc_stats['flag_distribution']
            }
            
            qc_success_msg = f"✅ SAGAR-QC Complete: Quality Score = {qc_stats['quality_metrics']['overall_quality_score']:.1f}%"
            print(qc_success_msg)
            add_log(LogLevel.SUCCESS, qc_success_msg, original_filename, job_id)
        except Exception as qc_error:
            qc_error_msg = f"⚠️ SAGAR-QC Error (continuing without QC): {qc_error}"
            print(qc_error_msg)
            add_log(LogLevel.WARNING, qc_error_msg, original_filename, job_id)
            import traceback
            traceback.print_exc()
            # Continue without QC flags if QC fails
            quality_report = None
            cleaned_df['qc_flag'] = 2  # UNKNOWN flag if QC fails
        
        geom_wkt = create_bounding_box(cleaned_df)
        cleaned_df = _sanitize_dataframe_for_parquet(cleaned_df)
        
        # Upload encrypted version to Supabase (for secure storage)
        encrypted_filename = processed_filename + ".enc"
        upload_enc_msg = f"🔐 Uploading encrypted Parquet to {processed_bucket} as {encrypted_filename}..."
        print(upload_enc_msg)
        add_log(LogLevel.INFO, upload_enc_msg, original_filename, job_id)
        upload_result = upload_parquet_to_storage(processed_bucket, encrypted_filename, cleaned_df, encrypt=True)
        
        # Upload unencrypted version to Supabase (for RAG and Superset Brain API)
        upload_unenc_msg = f"📤 Uploading unencrypted Parquet to {processed_bucket} as {processed_filename}..."
        print(upload_unenc_msg)
        add_log(LogLevel.INFO, upload_unenc_msg, original_filename, job_id)
        upload_result_unencrypted = upload_parquet_to_storage(processed_bucket, processed_filename, cleaned_df, encrypt=False)
        
        # Call Brain API with unencrypted filename (Superset needs to read the file)
        add_log(LogLevel.INFO, "🧠 Calling Brain API for Superset dashboard...", original_filename, job_id)
        brain_result = call_brain_api(processed_filename, processed_bucket)
        superset_view = brain_result.get("view") if brain_result else None
        superset_dash_id = brain_result.get("dashboard_id") if brain_result else None
        
        # Trigger RAG embedding generation with unencrypted filename (RAG needs to read the file)
        add_log(LogLevel.INFO, "🔄 Calling RAG Embedding API...", original_filename, job_id)
        rag_result = call_rag_embedding_api(processed_filename, processed_bucket)
        if rag_result:
            records_processed = rag_result.get('records_processed', 0)
            rag_success_msg = f"✅ RAG embeddings generated for {records_processed} records"
            print(rag_success_msg)
            add_log(LogLevel.SUCCESS, rag_success_msg, original_filename, job_id)

        # Store metadata with encrypted file location (this is the primary stored file)
        add_log(LogLevel.INFO, "💾 Storing metadata in database...", original_filename, job_id)
        update_metadata_in_db(
            original_filename=original_filename,
            processed_file_location=encrypted_filename,
            status='ready',
            file_type=file_type,
            metadata_payload=metadata_payload,
            geom=geom_wkt,
            superset_view=superset_view,
            superset_dash_id=superset_dash_id
        )
        
        final_success_msg = "🎉 Pipeline completed successfully!"
        print(final_success_msg)
        add_log(LogLevel.SUCCESS, final_success_msg, original_filename, job_id)
        result = {
            "status": "success",
            "processed_file": encrypted_filename,  # Return encrypted filename as primary file
            "processed_file_unencrypted": processed_filename,  # Also include unencrypted filename
            "metadata": metadata_payload,
            "upload_result": upload_result
        }
        
        # Add quality report if available
        if quality_report:
            result["quality_report"] = quality_report
        
        return result
        
    except Exception as e:
        error_msg = f"❌ Pipeline CRITICAL FAILURE: {e}"
        print(error_msg)
        add_log(LogLevel.ERROR, error_msg, original_filename, job_id)
        import traceback
        traceback.print_exc()
        return {"status": "failed", "error": str(e)}