extract_csv_metadata_service.py

import logging
from typing import List, Dict, Any, Optional
import pandas as pd
import numpy as np

# Configure Logging
logging.basicConfig(
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    level=logging.INFO
)
logger = logging.getLogger("csv_metadata_service")

def extract_csv_metadata_logic(csv_url: str) -> List[Dict[str, Any]]:
    """
    Analyzes a CSV file from a URL to infer column metadata.
    
    Args:
        csv_url (str): The public URL of the CSV file.
        
    Returns:
        List[Dict[str, Any]]: A list of dictionaries containing field metadata.
    """
    try:
        logger.info(f"Extracting metadata from: {csv_url}")
        
        # Load a sample of the CSV to infer types (first 50 rows is usually enough)
        # Using on_bad_lines='skip' to be robust against malformed rows
        # storage_options={'User-Agent': ...} can be added if requests are blocked
        df = pd.read_csv(csv_url, nrows=5, on_bad_lines='skip') 
        
        fields = []
        
        for col in df.columns:
            dtype = df[col].dtype
            
            # 1. Default type
            generic_type = 'string'
            
            # 2. Map Pandas/Numpy types to generic keys
            if pd.api.types.is_integer_dtype(dtype):
                generic_type = 'integer'
            elif pd.api.types.is_float_dtype(dtype):
                generic_type = 'decimal'
            elif pd.api.types.is_bool_dtype(dtype):
                generic_type = 'boolean'
            elif pd.api.types.is_datetime64_any_dtype(dtype):
                generic_type = 'date'
            else:
                # 3. Attempt to detect dates in object/string columns
                # We check the first non-null value to see if it parses as a date
                if len(df) > 0:
                    try:
                        first_valid = df[col].dropna().iloc[0] if not df[col].dropna().empty else ""
                        val_str = str(first_valid).strip()
                        
                        # Simple check to avoid treating plain numbers as dates
                        if val_str and not val_str.isdigit():
                            pd.to_datetime(val_str)
                            generic_type = 'date'
                    except (ValueError, TypeError):
                        # Not a date, keep as string
                        pass

            # 4. Determine Nullability
            is_nullable = bool(df[col].isnull().any())
            
            # 5. Heuristic for Primary Key
            lower_name = col.lower()
            potential_names = ['id', 'uuid', '_id', 'pk']
            
            # It's likely a PK if it has a common ID name OR ends in _id
            is_pk_name = lower_name in potential_names or lower_name.endswith('_id')
            
            # It must also be unique in our sample
            is_unique = df[col].is_unique if not df.empty else False
            
            is_primary_key = is_pk_name and is_unique

            fields.append({
                "name": col,
                "type": generic_type,
                "nullable": is_nullable,
                "isPrimaryKey": is_primary_key
            })
            
        logger.info(f"Successfully extracted {len(fields)} fields.")
        return fields

    except Exception as e:
        logger.error(f"CSV Analysis Error for {csv_url}: {str(e)}")
        # Re-raise to be handled by the controller's exception handler
        raise e