import pandas as pd
from sqlalchemy import create_engine, event
import urllib
import time

# --- CONFIGURACIÓN ---
FILE_PATH = 'Padron_ARCA.txt' # Ruta a tu archivo de 11.5GB
SERVER = 'LOCALHOST'          # Tu servidor (ej: LOCALHOST\SQLEXPRESS o IP)
DATABASE = 'MiBaseDeDatos'    # Nombre de tu base
USERNAME = 'sa'               # Usuario (o usa autenticación de windows)
PASSWORD = 'TuPassword'       # Password
DRIVER = 'ODBC Driver 17 for SQL Server' # O versión 18
CHUNK_SIZE = 100000           # 100k filas por lote es un buen balance RAM/CPU

# Definición de anchos fijos según documentación
# CUIT(11), DENOM(160), ACT(6), BAJA(1), REEMP(11), FECHA(10), SEX(1), CP(10), PROV(2), FALL(8)
WIDTHS = [11, 160, 6, 1, 11, 10, 1, 10, 2, 8]
COL_NAMES = [
    'cuit', 'denominacion', 'actividad', 'marca_baja',
    'cuit_reemplazo', 'fecha_nac_contrato', 'sexo',
    'codigo_postal', 'provincia', 'fecha_fallecimiento'
]

def get_engine():
    """Crea el motor de conexión optimizado para MS SQL"""
    connection_string = f'DRIVER={{{DRIVER}}};SERVER={SERVER};DATABASE={DATABASE};UID={USERNAME};PWD={PASSWORD};Trusted_Connection=no;'
    # Si usas autenticación de Windows, usa: Trusted_Connection=yes; y quita UID/PWD

    params = urllib.parse.quote_plus(connection_string)

    # fast_executemany=True es CRÍTICO para performance en MS SQL
    engine = create_engine(f"mssql+pyodbc:///?odbc_connect={params}", fast_executemany=True)
    return engine

def clean_chunk(df):
    """Lógica de limpieza y parseo eficiente"""

    # 1. Trimming masivo de espacios en blanco (vectorizado es muy rápido)
    df = df.apply(lambda x: x.str.strip() if x.dtype == "object" else x)

    # 2. Manejo de Fechas Nulas según documentación
    # Reemplazamos los valores 'dummy' por None para que SQL los tome como NULL

    # Fecha Nacimiento: 1901-01-01 -> NULL
    mask_nac = (df['fecha_nac_contrato'] == '1901-01-01')
    df.loc[mask_nac, 'fecha_nac_contrato'] = None

    # Fecha Fallecimiento: 19010101 -> NULL
    # Nota: El TXT trae formato AAAAMMDD en este campo, SQL lo suele entender,
    # pero a veces es mejor enviarlo separado. Si falla, parsear a datetime.
    mask_fall = (df['fecha_fallecimiento'] == '19010101')
    df.loc[mask_fall, 'fecha_fallecimiento'] = None

    # Opcional: Convertir a objetos datetime de Pandas si SQL se queja del formato string
    # df['fecha_nac_contrato'] = pd.to_datetime(df['fecha_nac_contrato'], errors='coerce')
    # df['fecha_fallecimiento'] = pd.to_datetime(df['fecha_fallecimiento'], format='%Y%m%d', errors='coerce')

    return df

def main():
    print(f"Iniciando carga masiva a MS SQL Server 2022...")
    engine = get_engine()
    start_time = time.time()
    total_rows = 0

    # Iterador de lectura (Streaming)
    # dtype=str evita que pandas intente adivinar tipos y pierda ceros a la izquierda en CUITs
    with pd.read_fwf(FILE_PATH, widths=WIDTHS, names=COL_NAMES,
                     chunksize=CHUNK_SIZE, encoding='latin-1', dtype=str) as reader:

        for i, chunk in enumerate(reader):
            batch_start = time.time()

            # Limpieza
            clean_df = clean_chunk(chunk)

            # Inserción
            # method=None usa la optimización del engine (fast_executemany)
            clean_df.to_sql('padron_arca', engine, if_exists='append', index=False, schema='dbo')

            rows_in_chunk = len(clean_df)
            total_rows += rows_in_chunk
            batch_time = time.time() - batch_start

            print(f"Lote {i+1} procesado: {rows_in_chunk} filas en {batch_time:.2f}s. Total: {total_rows:,}")

    print(f"--- FINALIZADO ---")
    print(f"Tiempo total: {(time.time() - start_time)/60:.2f} minutos.")
    print(f"Total registros: {total_rows:,}")

if __name__ == "__main__":
    main()