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