File size: 3,510 Bytes
c0fce26 03993cf c0fce26 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 | 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 |