import pandas as pd
import os

# 1. Fuente Oficial: NYC Taxi & Limousine Commission (Enero 2023)
url = "https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2023-01.parquet"

print(f"Descargando datos desde {url}...")
# Pandas lee Parquet nativamente (mucho más eficiente que CSV)
df = pd.read_parquet(url)

# 2. Muestreo: Tomamos 1 millón de filas para el ejercicio
df = df.sample(n=1000000, random_state=42)

print("Corrompiendo formato para simular Legacy CSV...")
# Guardamos SIN index y SIN header. Esto obliga a definir la estructura manualmente en SQL.
output_file = "C:\\Data\\taxi_dirty.csv" # Asegúrate que esta carpeta exista
df.to_csv(output_file, index=False, header=False)

print(f"¡Hecho! Archivo generado en: {output_file}")
print(f"Filas: {len(df)}")

CREATE DATABASE NYC_Taxi_Lab;
GO
USE NYC_Taxi_Lab;

-- 1. Tabla STAGING (Heap Table, sin Clustered Index)
-- Todo es VARCHAR para máxima permisividad.
CREATE TABLE Staging_Trips (
    VendorID VARCHAR(50),
    tpep_pickup_datetime VARCHAR(50),
    tpep_dropoff_datetime VARCHAR(50),
    passenger_count VARCHAR(50),
    trip_distance VARCHAR(50),
    RatecodeID VARCHAR(50),
    store_and_fwd_flag VARCHAR(50),
    PULocationID VARCHAR(50),
    DOLocationID VARCHAR(50),
    payment_type VARCHAR(50),
    fare_amount VARCHAR(50),
    extra VARCHAR(50),
    mta_tax VARCHAR(50),
    tip_amount VARCHAR(50),
    tolls_amount VARCHAR(50),
    improvement_surcharge VARCHAR(50),
    total_amount VARCHAR(50),
    congestion_surcharge VARCHAR(50),
    airport_fee VARCHAR(50)
);

-- 2. Carga de Alta Velocidad (Minimal Logging)
BULK INSERT Staging_Trips
FROM 'C:\Data\taxi_dirty.csv'
WITH (
    FIELDTERMINATOR = ',',  -- Separador de columnas CSV
    ROWTERMINATOR = '\n',   -- Salto de línea
    FIRSTROW = 1,           -- No hay headers
    TABLOCK                 -- IMPORTANTE: Permite carga paralela y minimiza log
);

-- Verificación rápida
SELECT TOP 5 * FROM Staging_Trips;

-- 1. Tabla Producción (Optimizada)
CREATE TABLE Trips (
    TripID INT IDENTITY(1,1) NOT NULL, -- Surrogate Key (Secuencial)
    VendorID TINYINT,
    Pickup DATETIME2(0),      -- Precisión de segundo
    Dropoff DATETIME2(0),
    PassengerCount TINYINT,   -- 1 byte (0-255)
    TripDistance DECIMAL(10,2),
    TotalAmount DECIMAL(10,2),
    
    -- El Clustered Index define el orden físico en disco.
    -- Usar IDENTITY evita la fragmentación (Page Splits).
    CONSTRAINT PK_Trips PRIMARY KEY CLUSTERED (TripID)
);

-- 2. ETL Intra-Base (Transformación)
INSERT INTO Trips (VendorID, Pickup, Dropoff, PassengerCount, TripDistance, TotalAmount)
SELECT 
    TRY_CAST(VendorID AS TINYINT),
    TRY_CAST(tpep_pickup_datetime AS DATETIME2(0)),
    TRY_CAST(tpep_dropoff_datetime AS DATETIME2(0)),
    -- Lógica de negocio: Si pasajeros es nulo/basura, asumir 1
    COALESCE(TRY_CAST(passenger_count AS TINYINT), 1),
    TRY_CAST(trip_distance AS DECIMAL(10,2)),
    TRY_CAST(total_amount AS DECIMAL(10,2))
FROM Staging_Trips
WHERE TRY_CAST(trip_distance AS DECIMAL(10,2)) > 0; -- Filtramos viajes inválidos

-- 3. La Prueba Final (Space Used)
EXEC sp_spaceused 'Staging_Trips';
EXEC sp_spaceused 'Trips'; 
-- ¡Deberías ver una reducción drástica en el tamaño 'Data'!