import pandas as pd
from sqlalchemy import create_engine, MetaData
from datetime import datetime

# ---------- CONFIG ----------
CSV_PATH = 'Dinesafe.csv'
LOG_PATH = 'etl_log.txt'
DROPPED_PATH = 'dropped_rows.csv'
DB_URL = "mssql+pyodbc://DESKTOP-TDFM6CE\\SQLEXPRESS/DineSafe?driver=ODBC+Driver+17+for+SQL+Server"

# ---------- LOGGING ----------


def log(message):
    timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
    with open(LOG_PATH, 'a') as f:
        f.write(f"[{timestamp}] {message}\n")
    print(message)

# ---------- LOAD ----------


def load_raw_data(path):
    df = pd.read_csv(path)
    df.columns = df.columns.str.strip()
    return df

# ---------- TRANSFORM ----------


def split_address(address):
    if pd.isna(address):
        return pd.Series([None, None])
    parts = str(address).split(',')
    street = parts[0].strip()
    unit = parts[1].strip() if len(parts) > 1 else None
    return pd.Series([street, unit])


def split_severity(severity):
    if pd.isna(severity):
        return pd.Series([None, None])
    parts = str(severity).split('-', 1)
    return pd.Series([parts[0].strip(), parts[1].strip() if len(parts) > 1 else None])


def split_infraction(detail):
    if pd.isna(detail) or '-' not in detail:
        return pd.Series([None, detail])
    parts = str(detail).rsplit('-', 1)
    return pd.Series([parts[1].strip(), parts[0].strip()])


def transform(df):
    df['Inspection ID'] = pd.to_numeric(
        df['Inspection ID'], errors='coerce').astype('Int64')
    df_valid = df.dropna(subset=['Inspection ID']).copy()
    df_invalid = df[df['Inspection ID'].isna()].copy()

    for d in [df_valid, df_invalid]:
        d[['StreetAddress', 'UnitNumber']
          ] = d['Establishment Address'].apply(split_address)
        d['RawAddress'] = d['Establishment Address']
        d['City'] = None
        d['PostalCode'] = None

    df_valid[['SeverityCode', 'SeverityLabel']
             ] = df_valid['Severity'].apply(split_severity)
    df_valid[['InfractionCode', 'InfractionText']
             ] = df_valid['Infraction Details'].apply(split_infraction)

    return df_valid, df_invalid

# ---------- PREPARE TABLES ----------


def prepare_tables(df_valid, df_invalid):
    all_types = pd.concat([df_valid['Establishment Type'],
                          df_invalid['Establishment Type']]).dropna().unique()
    est_type_df = pd.DataFrame({'TypeName': all_types})

    establishment_df = pd.concat([df_valid, df_invalid])[[
        'Establishment ID', 'Establishment Name', 'Establishment Type',
        'StreetAddress', 'UnitNumber', 'RawAddress',
        'City', 'PostalCode', 'Establishment Status',
        'Latitude', 'Longitude'
    ]].copy()
    establishment_df.columns = ['EstablishmentId', 'Name', 'TypeName',
                                'StreetAddress', 'UnitNumber', 'RawAddress',
                                'City', 'PostalCode', 'Status',
                                'Latitude', 'Longitude']
    establishment_df = establishment_df.drop_duplicates(
        subset='EstablishmentId')

    inspection_df = df_valid[['Inspection ID', 'Establishment ID', 'Inspection Date',
                              'Outcome', 'Amount Fined']].copy()
    inspection_df.columns = ['InspectionId', 'EstablishmentId', 'InspectionDate',
                             'OutcomeType', 'AmountFined']
    inspection_df['OutcomeDetail'] = None
    inspection_df = inspection_df.drop_duplicates(subset='InspectionId')

    infraction_df = df_valid[['InfractionText',
                              'InfractionCode']].dropna().drop_duplicates()

    inspection_infraction_df = df_valid[['unique_id', 'Inspection ID',
                                         'InfractionText', 'InfractionCode',
                                         'SeverityCode', 'SeverityLabel', 'Action']].copy()
    inspection_infraction_df.columns = ['UniqueId', 'InspectionId',
                                        'InfractionText', 'InfractionCode',
                                        'SeverityCode', 'SeverityLabel', 'Action']
    inspection_infraction_df = inspection_infraction_df.dropna(
        subset=['InfractionText', 'InfractionCode'])
    inspection_infraction_df = inspection_infraction_df.drop_duplicates(
        subset='UniqueId')

    return est_type_df, establishment_df, inspection_df, infraction_df, inspection_infraction_df

# ---------- INSERT ----------


def insert_table(df, table_name, engine, key_column=None):
    try:
        if key_column:
            existing_keys = pd.read_sql(
                f'SELECT {key_column} FROM {table_name}', con=engine)
            df = df[~df[key_column].isin(existing_keys[key_column])]
    except Exception:
        log(f"{table_name} table not found or empty.")
    df.to_sql(table_name, con=engine, if_exists='append',
              index=False, chunksize=500)
    log(f"Inserted {len(df)} rows into {table_name}.")


def insert_infraction_rows(df, engine):
    metadata = MetaData()
    metadata.reflect(bind=engine)
    infraction_table = metadata.tables['Infraction']
    existing = pd.read_sql(
        'SELECT InfractionText, InfractionCode FROM Infraction', con=engine)
    existing_set = set(
        zip(existing['InfractionText'], existing['InfractionCode']))
    inserted = 0
    with engine.begin() as conn:
        for _, row in df.iterrows():
            key = (row['InfractionText'], row['InfractionCode'])
            if key not in existing_set:
                conn.execute(infraction_table.insert().values(
                    InfractionText=row['InfractionText'],
                    InfractionCode=row['InfractionCode']
                ))
                inserted += 1
    log(f"Inserted {inserted} new infractions.")


def resolve_type_ids(establishment_df, engine):
    type_lookup = pd.read_sql(
        'SELECT TypeId, TypeName FROM EstablishmentType', con=engine)
    type_map = dict(zip(type_lookup['TypeName'], type_lookup['TypeId']))
    establishment_df['TypeId'] = establishment_df['TypeName'].map(type_map)
    return establishment_df.drop(columns=['TypeName'])


def resolve_infraction_ids(inspection_infraction_df, engine):
    infraction_lookup = pd.read_sql(
        'SELECT InfractionId, InfractionText, InfractionCode FROM Infraction', con=engine)
    infraction_map = {(row['InfractionText'], row['InfractionCode']): row['InfractionId']
                      for _, row in infraction_lookup.iterrows()}
    inspection_infraction_df['InfractionId'] = inspection_infraction_df.apply(
        lambda row: infraction_map.get((row['InfractionText'], row['InfractionCode'])), axis=1)
    return inspection_infraction_df.drop(columns=['InfractionText', 'InfractionCode'])

# ---------- MAIN ----------


def run_etl():
    engine = create_engine(DB_URL, fast_executemany=True)
    df = load_raw_data(CSV_PATH)
    df_valid, df_invalid = transform(df)

    dropped = df[df['Inspection ID'].isna() | df['Infraction Details'].isna()]
    dropped.to_csv(DROPPED_PATH, index=False)
    log(f"Logged {len(dropped)} dropped rows to {DROPPED_PATH}.")

    est_type_df, establishment_df, inspection_df, infraction_df, inspection_infraction_df = prepare_tables(
        df_valid, df_invalid)

    insert_table(est_type_df, 'EstablishmentType', engine)
    establishment_df = resolve_type_ids(establishment_df, engine)
    insert_table(establishment_df, 'Establishment', engine, 'EstablishmentId')
    insert_table(inspection_df, 'Inspection', engine, 'InspectionId')
    insert_infraction_rows(infraction_df, engine)
    inspection_infraction_df = resolve_infraction_ids(
        inspection_infraction_df, engine)
    insert_table(inspection_infraction_df,
                 'InspectionInfraction', engine, 'UniqueId')

    log("ETL process completed successfully.")


# ---------- EXECUTE ----------
if __name__ == "__main__":
    run_etl()
