import dask.dataframe as dd
from sqlalchemy import create_engine, text, types, inspect
import pandas as pd
import numpy as np
import re
import warnings
from datetime import date 


class ETLPipeline:
    def __init__(self, start_year=2000, end_year=2030):
        self.start_date = date(start_year, 1, 1)
        self.end_date   = date(end_year, 12, 31)
        print(f"Date dimension range set from {self.start_date} to {self.end_date}")

        self.source_engine = create_engine(
            "mysql+mysqlconnector://root:Root%40123@localhost/vms_test"
        )
        self.target_engine = create_engine(
            "mysql+mysqlconnector://root:Root%40123@localhost/olap_2"
        )

    # def extract_company_dimension(self):
    #     print("Starting company extraction...")
    #     source_engine = create_engine("mysql+mysqlconnector://root:Root%40123@localhost/vms_test")
    #     try:
    #         with source_engine.connect() as conn:
    #             has_org_id = conn.execute(text("""
    #                 SELECT COUNT(*)
    #                 FROM information_schema.COLUMNS
    #                 WHERE TABLE_SCHEMA = 'vms_test'
    #                 AND TABLE_NAME = 'tbl_companies'
    #                 AND COLUMN_NAME = 'organisation_id'
    #             """)).scalar() > 0

    #             if has_org_id:
    #                 # Filter to only get companies with organisation_id = 5
    #                 df = pd.read_sql_query(
    #                     "SELECT id, company_name, company_code, account_number, is_active, organisation_id FROM tbl_companies WHERE organisation_id = 5",
    #                     conn
    #                 )
    #                 # print(f"Extracted {len(df)} companies with organisation_id = 5")
    #             else:
    #                 # print("The tbl_companies table does not have an organisation_id column. Cannot filter by organisation_id = 5.")
    #                 return None

    #             if df.empty:
    #                 # print("No companies found with organisation_id = 5.")
    #                 return None

    #             df['id'] = df['id'].astype(np.int64)
    #             df['is_active'] = df['is_active'].astype(np.int8) 
    #             if 'organisation_id' in df.columns:
    #                 df['organisation_id'] = df['organisation_id'].astype('Int64')

    #             df['company_key'] = range(1, len(df) + 1)

    #             dim_company_df = df.rename(columns={
    #                 'id': 'company_source_id',
    #                 'is_active': 'is_company_active'
    #             })

    #             final_columns = [
    #                 'company_key', 'company_source_id', 'company_name',
    #                 'company_code', 'account_number', 'is_company_active',
    #                 'organisation_id'
    #             ]
    #             dim_company_df = dim_company_df[final_columns]

    #             npartitions = max(1, len(dim_company_df) // 50000)  
    #             dim_company = dd.from_pandas(dim_company_df, npartitions=npartitions)

    #             return dim_company
    #     finally:
    #         source_engine.dispose()

    # def load_company_dimension(self, dim_company):
    #     if dim_company is None or dim_company.npartitions == 0:
    #         #  print("No company data to load.")
    #          return False

    #     print("Starting company load...")
    #     dim_company_pd = dim_company.compute()

    #     if dim_company_pd.empty:
    #         # print("Computed company DataFrame is empty. Skipping load.")
    #         return False

    #     target_engine = create_engine("mysql+mysqlconnector://root:Root%40123@localhost/olap_2")
    #     try:
    #         with target_engine.begin() as connection:
    #             # print("Disabling foreign key checks.")
    #             connection.execute(text("SET FOREIGN_KEY_CHECKS=0"))
    #             # print("Dropping existing dim_company table.")
    #             connection.execute(text("DROP TABLE IF EXISTS dim_company"))

    #             # print("Creating new dim_company table.")
    #             connection.execute(text("""
    #                 CREATE TABLE dim_company (
    #                     company_key INT NOT NULL PRIMARY KEY,
    #                     company_source_id BIGINT,
    #                     company_name VARCHAR(255),
    #                     company_code VARCHAR(50),
    #                     account_number VARCHAR(50),
    #                     is_company_active TINYINT,
    #                     organisation_id INT DEFAULT NULL
    #                 )
    #             """))

    #             # print(f"Inserting {len(dim_company_pd)} records into dim_company...")
    #             insert_count = 0
    #             for _, row in dim_company_pd.iterrows():
    #                 try:
    #                     params = {
    #                         'company_key': int(row['company_key']),
    #                         'company_source_id': int(row['company_source_id']) if pd.notna(row['company_source_id']) else None,
    #                         'company_name': str(row['company_name']) if pd.notna(row['company_name']) else None,
    #                         'company_code': str(row['company_code']) if pd.notna(row['company_code']) else None,
    #                         'account_number': str(row['account_number']) if pd.notna(row['account_number']) else None,
    #                         'is_company_active': int(row['is_company_active']) if pd.notna(row['is_company_active']) else None,
    #                         'organisation_id': int(row['organisation_id']) if pd.notna(row['organisation_id']) else None
    #                     }
    #                     connection.execute(text("""
    #                         INSERT INTO dim_company
    #                         (company_key, company_source_id, company_name, company_code, account_number, is_company_active, organisation_id)
    #                         VALUES (:company_key, :company_source_id, :company_name, :company_code, :account_number, :is_company_active, :organisation_id)
    #                     """), params)
    #                     insert_count += 1
    #                 except Exception as e:
    #                     # print(f"Error inserting company row {row.get('company_key', 'N/A')}: {e}")
    #                     # print(f"Problematic row data: {row.to_dict()}")
    #                     continue

    #             # print(f"Successfully inserted {insert_count} company records.")
    #             # print("Re-enabling foreign key checks.")
    #             connection.execute(text("SET FOREIGN_KEY_CHECKS = 1"))
    #         return True
    #     except Exception as e:
    #         #  print(f"An error occurred during company load: {e}")
    #          raise
    #     finally:
    #         target_engine.dispose()

    # def extract_fitment_reason_dimension(self):
    #     print("Starting fitment_reason extraction...")
    #     source_engine = create_engine("mysql+mysqlconnector://root:Root%40123@localhost/vms_test")
    #     try:
    #         with source_engine.connect() as conn:
    #             table_exists = conn.execute(text("""
    #                 SELECT COUNT(*)
    #                 FROM information_schema.TABLES
    #                 WHERE TABLE_SCHEMA = 'vms_test'
    #                   AND TABLE_NAME = 'tbl_job_fitment_reasons'
    #             """)).scalar() > 0

    #             df = None

    #             if not table_exists:
    #                 # print("Source table tbl_job_fitment_reasons not found. Using default data.")
    #                 default_reasons = [
    #                     'Worn', 'Blowout', 'Flatspot', 'Sidewall Damage', 'Mismatch',
    #                     'Punctured', 'Tread Damage', 'Complaint', 'Uneven Wear',
    #                     'MOT Inspection', 'De-fleet', 'Run Flat', 'Tread Lift',
    #                     'Hankook Test Tyres', 'Unknown Reason', 'Conversion',
    #                     'Out Of Shape', 'Cut To Cords', 'Aged', 'Regroove',
    #                     'Vehicle Vibrating', 'Damaged', 'Twinning', 'Default'
    #                 ]
    #                 data = {
    #                     'fitment_reason_source_id': range(1, len(default_reasons) + 1),
    #                     'fitment_reason_name': default_reasons,
    #                     'organisation_id': [None] * len(default_reasons)
    #                 }
    #                 df = pd.DataFrame(data)
    #             else:
    #                 # print("Source table tbl_job_fitment_reasons found. Extracting data.")
    #                 has_org_id = conn.execute(text("""
    #                     SELECT COUNT(*)
    #                     FROM information_schema.COLUMNS
    #                     WHERE TABLE_SCHEMA = 'vms_test'
    #                       AND TABLE_NAME = 'tbl_job_fitment_reasons'
    #                       AND COLUMN_NAME = 'organisation_id'
    #                 """)).scalar() > 0

    #                 columns = conn.execute(text("""
    #                     SELECT COLUMN_NAME
    #                     FROM information_schema.COLUMNS
    #                     WHERE TABLE_SCHEMA = 'vms_test'
    #                       AND TABLE_NAME = 'tbl_job_fitment_reasons'
    #                       AND COLUMN_NAME IN ('id', 'fitment_reason_id', 'name', 'fitment_reason_name')
    #                 """)).fetchall()
    #                 column_names = {col[0] for col in columns}

    #                 id_col = 'id' if 'id' in column_names else 'fitment_reason_id'
    #                 name_col = 'name' if 'name' in column_names else 'fitment_reason_name'

    #                 if id_col not in column_names or name_col not in column_names:
    #                      raise ValueError(f"Could not find required columns ({id_col}, {name_col}) in tbl_job_fitment_reasons")


    #                 if has_org_id:
    #                     # print("Extracting fitment reasons with organisation_id.")
    #                     query = f"""SELECT {id_col} as fitment_reason_source_id,
    #                                        {name_col} as fitment_reason_name,
    #                                        organisation_id
    #                                 FROM tbl_job_fitment_reasons"""
    #                 else:
    #                     # print("Extracting fitment reasons without organisation_id (adding placeholder).")
    #                     query = f"""SELECT {id_col} as fitment_reason_source_id,
    #                                        {name_col} as fitment_reason_name
    #                                 FROM tbl_job_fitment_reasons"""

    #                 df = pd.read_sql_query(query, conn)
    #                 if 'organisation_id' not in df.columns:
    #                     df['organisation_id'] = None


    #             if df is None or df.empty:
    #                 #  print("No fitment reason data extracted.")
    #                  return None

    #             df['fitment_reason_source_id'] = df['fitment_reason_source_id'].astype(np.int64)
    #             df['organisation_id'] = df['organisation_id'].astype('Int64')
    #             df['fitment_reason_name'] = df['fitment_reason_name'].astype(str).fillna('Unknown Reason')


    #             allowed_values_enum = [
    #                 'Worn','Blowout','Flatspot','Sidewall Damage','Mismatch',
    #                 'Punctured','Tread Damage','Complaint','Uneven Wear',
    #                 'MOT Inspection','De-fleet','Run Flat','Tread Lift',
    #                 'Hankook Test Tyres','Unknown Reason','Conversion',
    #                 'Out Of Shape','Cut To Cords','Aged','Regroove',
    #                 'Vehicle Vibrating','Damaged','Twinning','Default'
    #             ]
    #             allowed_values_set = set(allowed_values_enum)

    #             def map_to_allowed_value(value):
    #                 value_str = str(value).strip()
    #                 if value_str in allowed_values_set:
    #                     return value_str
    #                 return "Unknown Reason"

    #             df['fitment_reason_name'] = df['fitment_reason_name'].apply(map_to_allowed_value)
    #             df = df.drop_duplicates(subset=['fitment_reason_name'], keep='first')

    #             df = df.reset_index(drop=True)
    #             df['fitment_reason_key'] = df.index + 1

    #             final_columns = [
    #                 'fitment_reason_key', 'fitment_reason_source_id',
    #                 'fitment_reason_name', 'organisation_id'
    #             ]
    #             df = df[final_columns]

    #             npartitions = max(1, len(df) // 10000)   #Every 10,000 row in dataframe a new partition is created.
    #             dim_fitment_reason = dd.from_pandas(df, npartitions=npartitions)

    #             # print(f"Extracted and transformed {len(df)} unique fitment reason records.")
    #             return dim_fitment_reason
    #     finally:
    #         source_engine.dispose()

    # def load_fitment_reason_dimension(self, dim_fitment_reason):
    #     if dim_fitment_reason is None or dim_fitment_reason.npartitions == 0:
    #         # print("No fitment reason data to load.")
    #         return False

    #     print("Starting fitment reason load...")
    #     dim_fitment_reason_pd = dim_fitment_reason.compute()

    #     if dim_fitment_reason_pd.empty:
    #         # print("Computed fitment reason DataFrame is empty. Skipping load.")
    #         return False

    #     target_engine = create_engine("mysql+mysqlconnector://root:Root%40123@localhost/olap_2")
    #     try:
    #         with target_engine.begin() as connection:
    #             # print("Disabling foreign key checks.")
    #             connection.execute(text("SET FOREIGN_KEY_CHECKS = 0"))
    #             connection.execute(text("SET SQL_SAFE_UPDATES = 0"))
    #             # print("Dropping existing dim_fitment_reason table.")
    #             connection.execute(text("DROP TABLE IF EXISTS dim_fitment_reason"))

    #             allowed_values_enum = [
    #                 'Worn','Blowout','Flatspot','Sidewall Damage','Mismatch',
    #                 'Punctured','Tread Damage','Complaint','Uneven Wear',
    #                 'MOT Inspection','De-fleet','Run Flat','Tread Lift',
    #                 'Hankook Test Tyres','Unknown Reason','Conversion',
    #                 'Out Of Shape','Cut To Cords','Aged','Regroove',
    #                 'Vehicle Vibrating','Damaged','Twinning','Default'
    #             ]
    #             enum_values = ",".join([f"'{val}'" for val in allowed_values_enum])

    #             # print("Creating new dim_fitment_reason table.")
    #             connection.execute(text(f"""
    #                 CREATE TABLE dim_fitment_reason (
    #                     fitment_reason_key INT NOT NULL PRIMARY KEY,
    #                     fitment_reason_source_id INT NOT NULL,
    #                     fitment_reason_name ENUM({enum_values}),
    #                     organisation_id INT DEFAULT NULL
    #                 )
    #             """))

    #             # print(f"Inserting {len(dim_fitment_reason_pd)} records into dim_fitment_reason...")
    #             insert_count = 0
    #             for _, row in dim_fitment_reason_pd.iterrows():
    #                 try:
    #                      params = {
    #                         'fitment_reason_key': int(row['fitment_reason_key']),
    #                         'fitment_reason_source_id': int(row['fitment_reason_source_id']) if pd.notna(row['fitment_reason_source_id']) else None,
    #                         'fitment_reason_name': row['fitment_reason_name'],
    #                         'organisation_id': int(row['organisation_id']) if pd.notna(row['organisation_id']) else None
    #                     }
    #                      connection.execute(text("""
    #                         INSERT INTO dim_fitment_reason
    #                         (fitment_reason_key, fitment_reason_source_id, fitment_reason_name, organisation_id)
    #                         VALUES (:fitment_reason_key, :fitment_reason_source_id, :fitment_reason_name, :organisation_id)
    #                     """), params)
    #                      insert_count += 1
    #                 except Exception as e:
    #                     # print(f"Error inserting fitment reason row {row.get('fitment_reason_key', 'N/A')}: {e}")
    #                     # print(f"Problematic row data: {row.to_dict()}")
    #                     continue

    #             # print(f"Successfully inserted {insert_count} fitment reason records.")
    #             # print("Re-enabling foreign key checks.")
    #             connection.execute(text("SET FOREIGN_KEY_CHECKS = 1"))
    #         return True
    #     except Exception as e:
    #         #  print(f"An error occurred during fitment reason load: {e}")
    #          raise
    #     finally:
    #         target_engine.dispose()

    # def extract_product_dimension(self):
    #     print("Starting product extraction...")
    #     src_engine = create_engine("mysql+mysqlconnector://root:Root%40123@localhost/vms_test")

    #     try:
    #          with src_engine.connect() as conn:
    #               has_org_id = conn.execute(text("""
    #                 SELECT COUNT(*)
    #                 FROM information_schema.COLUMNS
    #                 WHERE TABLE_SCHEMA = 'vms_test'
    #                   AND TABLE_NAME = 'tbl_products'
    #                   AND COLUMN_NAME = 'organisation_id'
    #               """)).scalar() > 0

    #               org_id_select = "p.organisation_id AS organisation_id" if has_org_id else "NULL AS organisation_id"
    #             #   print(f"Extracting products {'with' if has_org_id else 'without'} organisation_id.")

    #          sql = f"""
    #          SELECT
    #              p.id                   AS product_source_id,
    #              p.name                 AS description,
    #              p.name                 AS product_name,
    #              pg.group_name          AS group_name,
    #              pg.group_code          AS group_code,
    #              CASE WHEN t.id IS NULL THEN 0 ELSE 1 END AS is_tyre,
    #              CASE
    #                  WHEN t.id IS NULL THEN 0
    #                  WHEN t.remould = 0 AND t.regrooved = 0 THEN 1
    #                  ELSE 0
    #              END                    AS is_tyre_new,
    #              pat.name               AS pattern,
    #              t.width                AS tyre_width,
    #              t.rim                  AS tyre_rim,
    #              t.load_rating          AS tyre_load_rating,
    #              t.speed_rating         AS tyre_speed_rating,
    #              t.remould              AS is_tyre_remould,
    #              t.regrooved            AS is_tyre_regrooved, # Select directly
    #              t.profile              AS tyre_profile,
    #              b.name                 AS brand_name,
    #              {org_id_select}
    #          FROM      tbl_products           p
    #          LEFT JOIN tbl_product_groups   pg  ON pg.id = p.product_group_id
    #          LEFT JOIN tbl_product_patterns pat ON pat.id = p.product_pattern_id
    #          LEFT JOIN tbl_product_brands   b   ON b.id = p.product_brand_id
    #          LEFT JOIN tbl_product_tyres    t   ON t.product_id = p.id
    #          """
    #          df = pd.read_sql_query(sql, src_engine)

    #          if df.empty:
    #             #  print("No product data extracted.")
    #              return None

    #         #  print(f"Extracted {len(df)} raw product records. Performing transformations...")

    #          df['product_source_id'] = df['product_source_id'].astype(np.int64)
    #          df['organisation_id'] = df['organisation_id'].astype('Int64')


    #          for col in ['is_tyre', 'is_tyre_new', 'is_tyre_remould', 'is_tyre_regrooved']:
    #              df[col] = df[col].astype('Int8')

    #          string_cols = ['description', 'product_name', 'group_name', 'group_code',
    #                         'pattern', 'tyre_width', 'tyre_rim', 'tyre_load_rating',
    #                         'tyre_speed_rating', 'tyre_profile', 'brand_name']
    #          for col in string_cols:
    #              df[col] = df[col].astype(object).where(pd.notna(df[col]), None)


    #          df = df.reset_index(drop=True)
    #          df["product_key"] = range(1, len(df) + 1)

    #          final_columns = [
    #              'product_key', 'product_source_id', 'description', 'product_name',
    #              'group_name', 'group_code', 'is_tyre', 'is_tyre_new', 'pattern',
    #              'tyre_width', 'tyre_rim', 'tyre_load_rating', 'tyre_speed_rating',
    #              'is_tyre_remould', 'is_tyre_regrooved', 'tyre_profile', 'brand_name',
    #              'organisation_id'
    #          ]
    #          df = df[final_columns]

    #          npartitions = max(1, len(df) // 50000)
    #          dim_product = dd.from_pandas(df, npartitions=npartitions)
    #         #  print(f"Transformed {len(df)} product records.")
    #          return dim_product
    #     except Exception as e:
    #         # print(f"Error during product extraction: {e}")
    #         raise
    #     finally:
    #         src_engine.dispose()

    # def load_product_dimension(self, dim_product):
    #     if dim_product is None or dim_product.npartitions == 0:
    #         # print("No product data to load.")
    #         return False

    #     print("Starting product load...")
    #     pdf = dim_product.compute()

    #     if pdf.empty:
    #         # print("Computed product DataFrame is empty. Skipping load.")
    #         return False

    #     pdf = pdf.replace({np.nan: None, pd.NaT: None})

    #     tgt_engine = create_engine("mysql+mysqlconnector://root:Root%40123@localhost/olap_2")
    #     try:
    #         with tgt_engine.begin() as conn:
    #             # print("Disabling foreign key checks.")
    #             conn.execute(text("SET FOREIGN_KEY_CHECKS = 0"))
    #             # print("Dropping existing dim_product table.")
    #             conn.execute(text("DROP TABLE IF EXISTS dim_product"))

    #             # print("Creating new dim_product table.")
    #             conn.execute(text("""
    #                 CREATE TABLE dim_product (
    #                     product_key        INT PRIMARY KEY,
    #                     product_source_id  BIGINT,
    #                     description        VARCHAR(255),
    #                     group_name         VARCHAR(100),
    #                     group_code         VARCHAR(50),
    #                     is_tyre_new        TINYINT,
    #                     is_tyre            TINYINT,
    #                     pattern            VARCHAR(100),
    #                     tyre_width         VARCHAR(50),
    #                     tyre_rim           VARCHAR(50),
    #                     tyre_load_rating   VARCHAR(50),
    #                     tyre_speed_rating  VARCHAR(50),
    #                     is_tyre_remould    TINYINT, -- Allows NULL
    #                     is_tyre_regrooved  TINYINT, -- Allows NULL
    #                     tyre_profile       VARCHAR(50),
    #                     brand_name         VARCHAR(100),
    #                     product_name       VARCHAR(255),
    #                     organisation_id    INT DEFAULT NULL
    #                 )
    #             """)) # Ensure TINYINT columns in DB allow NULL if needed

    #             # print(f"Inserting {len(pdf)} records into dim_product...")
    #             insert_count = 0
    #             for _, row in pdf.iterrows():
    #                 try:
    #                      # Convert row to dictionary to handle potential missing keys gracefully if needed
    #                      # row_dict = row.to_dict() # Alternative way

    #                      params = {
    #                         'product_key': int(row['product_key']),
    #                         'product_source_id': int(row['product_source_id']) if pd.notna(row['product_source_id']) else None,
    #                         'description': row['description'], # Already None if null
    #                         'group_name': row['group_name'],
    #                         'group_code': row['group_code'],
    #                         # Use pd.notna to check for None before casting TINYINT flags
    #                         'is_tyre_new': int(row['is_tyre_new']) if pd.notna(row['is_tyre_new']) else None,
    #                         'is_tyre': int(row['is_tyre']) if pd.notna(row['is_tyre']) else None,
    #                         'pattern': row['pattern'],
    #                         'tyre_width': row['tyre_width'],
    #                         'tyre_rim': row['tyre_rim'],
    #                         'tyre_load_rating': row['tyre_load_rating'],
    #                         'tyre_speed_rating': row['tyre_speed_rating'],
    #                         # --- *** CORRECTED LOADING CHECKS *** ---
    #                         'is_tyre_remould': int(row['is_tyre_remould']) if pd.notna(row['is_tyre_remould']) else None,
    #                         'is_tyre_regrooved': int(row['is_tyre_regrooved']) if pd.notna(row['is_tyre_regrooved']) else None,
    #                         'tyre_profile': row['tyre_profile'],
    #                         'brand_name': row['brand_name'],
    #                         'product_name': row['product_name'],
    #                         'organisation_id': int(row['organisation_id']) if pd.notna(row['organisation_id']) else None
    #                     }

    #                      conn.execute(text("""
    #                         INSERT INTO dim_product (
    #                             product_key, product_source_id, description, group_name,
    #                             group_code, is_tyre_new, is_tyre, pattern, tyre_width,
    #                             tyre_rim, tyre_load_rating, tyre_speed_rating,
    #                             is_tyre_remould, is_tyre_regrooved, tyre_profile,
    #                             brand_name, product_name, organisation_id
    #                         ) VALUES (
    #                             :product_key, :product_source_id, :description, :group_name,
    #                             :group_code, :is_tyre_new, :is_tyre, :pattern, :tyre_width,
    #                             :tyre_rim, :tyre_load_rating, :tyre_speed_rating,
    #                             :is_tyre_remould, :is_tyre_regrooved, :tyre_profile,
    #                             :brand_name, :product_name, :organisation_id
    #                         )
    #                     """), params)
    #                      insert_count += 1
    #                 except Exception as e:
    #                     # print(f"Error inserting product row {row.get('product_key', 'N/A')}: {e}")
    #                     # print(f"Problematic row data: {row.to_dict()}")
    #                     continue

    #             # print(f"Successfully inserted {insert_count} product records.")
    #             # print("Re-enabling foreign key checks.")
    #             conn.execute(text("SET FOREIGN_KEY_CHECKS = 1"))
    #         return True
    #     except Exception as e:
    #         #  print(f"An error occurred during product load: {e}")
    #          raise
    #     finally:
    #         tgt_engine.dispose()

    # # --- Date Dimension Methods (assuming they exist from previous step) ---
    # def extract_date_dimension(self):
    #     """Generates the date dimension data within the specified range."""
    #     print("Starting date dimension generation...")
    #     try:
    #         date_range = pd.date_range(self.start_date, self.end_date, freq='D')
    #         dim_date_df = pd.DataFrame(date_range, columns=['full_date'])

    #         dim_date_df['date_key'] = dim_date_df['full_date'].dt.strftime('%Y%m%d').astype(int)
    #         dim_date_df['day'] = dim_date_df['full_date'].dt.day.astype(np.int8)
    #         dim_date_df['month'] = dim_date_df['full_date'].dt.month.astype(np.int8)
    #         dim_date_df['year'] = dim_date_df['full_date'].dt.year.astype(np.int16)
    #         dim_date_df['day_of_week'] = dim_date_df['full_date'].dt.dayofweek.astype(np.int8)
    #         dim_date_df['day_name'] = dim_date_df['full_date'].dt.day_name()
    #         dim_date_df['month_name'] = dim_date_df['full_date'].dt.month_name()
    #         dim_date_df['quarter'] = dim_date_df['full_date'].dt.quarter.astype(np.int8)
    #         dim_date_df['is_weekend'] = (dim_date_df['day_of_week'] >= 5).astype(np.int8)


    #         final_columns = [
    #             'date_key', 'full_date', 'day', 'month', 'year', 'day_of_week',
    #             'day_name', 'month_name', 'quarter', 'is_weekend'
    #         ]
    #         dim_date_df = dim_date_df[final_columns]

    #         npartitions = max(1, len(dim_date_df) // 10000)
    #         dim_date = dd.from_pandas(dim_date_df, npartitions=npartitions)

    #         # print(f"Generated {len(dim_date_df)} date dimension records.")
    #         return dim_date

    #     except Exception as e:
    #         # print(f"Error generating date dimension: {e}")
    #         raise

    # def load_date_dimension(self, dim_date):
    #     """Loads the generated date dimension data into the target table."""
    #     if dim_date is None or dim_date.npartitions == 0:
    #         #  print("No date data to load.")
    #          return False

    #     # print("Starting date dimension load...")
    #     dim_date_pd = dim_date.compute()

    #     if dim_date_pd.empty:
    #         # print("Computed date DataFrame is empty. Skipping load.")
    #         return False

    #     target_engine = create_engine("mysql+mysqlconnector://root:Root%40123@localhost/olap_2")
    #     try:
    #         with target_engine.begin() as connection:
    #             print("Disabling foreign key checks.")
    #             connection.execute(text("SET FOREIGN_KEY_CHECKS=0"))
    #             print("Dropping existing dim_date table.")
    #             connection.execute(text("DROP TABLE IF EXISTS dim_date"))

    #             print("Creating new dim_date table.")
    #             connection.execute(text("""
    #                 CREATE TABLE dim_date (
    #                     date_key INT PRIMARY KEY,
    #                     full_date DATE NOT NULL,
    #                     day TINYINT NOT NULL,
    #                     month TINYINT NOT NULL,
    #                     year SMALLINT NOT NULL,
    #                     day_of_week TINYINT NOT NULL,
    #                     day_name VARCHAR(10) NOT NULL,
    #                     month_name VARCHAR(10) NOT NULL,
    #                     quarter TINYINT NOT NULL,
    #                     is_weekend TINYINT NOT NULL
    #                 ) ENGINE=InnoDB;
    #             """))

    #             # print(f"Inserting {len(dim_date_pd)} records into dim_date...")
    #             insert_count = 0
    #             for _, row in dim_date_pd.iterrows():
    #                 try:
    #                     params = {
    #                         'date_key': int(row['date_key']),
    #                         'full_date': row['full_date'],
    #                         'day': int(row['day']),
    #                         'month': int(row['month']),
    #                         'year': int(row['year']),
    #                         'day_of_week': int(row['day_of_week']),
    #                         'day_name': str(row['day_name']),
    #                         'month_name': str(row['month_name']),
    #                         'quarter': int(row['quarter']),
    #                         'is_weekend': int(row['is_weekend'])
    #                     }
    #                     connection.execute(text("""
    #                         INSERT INTO dim_date
    #                         (date_key, full_date, day, month, year, day_of_week, day_name, month_name, quarter, is_weekend)
    #                         VALUES (:date_key, :full_date, :day, :month, :year, :day_of_week, :day_name, :month_name, :quarter, :is_weekend)
    #                     """), params)
    #                     insert_count += 1
    #                 except Exception as e:
    #                     # print(f"Error inserting date row {row.get('date_key', 'N/A')}: {e}")
    #                     # print(f"Problematic row data: {row.to_dict()}")
    #                     continue

    #             # print(f"Successfully inserted {insert_count} date dimension records.")
    #             # print("Re-enabling foreign key checks.")
    #             connection.execute(text("SET FOREIGN_KEY_CHECKS = 1"))
    #         return True
    #     except Exception as e:
    #         #  print(f"An error occurred during date dimension load: {e}")
    #          raise
    #     finally:
    #         target_engine.dispose()
                 
    # def extract_service_provider_dimension(self):
    #     print("Starting service provider extraction…")
    #     src = create_engine(
    #         "mysql+mysqlconnector://root:Root%40123@localhost/vms_test",
    #         pool_pre_ping=True,
    #     )

    #     try:
    #         with src.connect() as cn:

    #             if not cn.scalar(text("""
    #                 SELECT COUNT(*) FROM information_schema.TABLES
    #                 WHERE TABLE_SCHEMA = 'vms_test'
    #                 AND TABLE_NAME   = 'tbl_serviceproviders'
    #             """)):
    #                 print("tbl_serviceproviders not found – skipping.")
    #                 return None

    #             centres_exist = cn.scalar(text("""
    #                 SELECT COUNT(*) FROM information_schema.TABLES
    #                 WHERE TABLE_SCHEMA = 'vms_test'
    #                 AND TABLE_NAME   = 'tbl_serviceprovider_centres'
    #             """)) > 0
                
    #             sp_has_org = cn.scalar(text("""
    #                 SELECT COUNT(*) FROM information_schema.COLUMNS
    #                 WHERE TABLE_SCHEMA = 'vms_test'
    #                 AND TABLE_NAME   = 'tbl_serviceproviders'
    #                 AND COLUMN_NAME  = 'organisation_id'
    #             """)) > 0
    #             sp_org_sel = "sp.organisation_id" if sp_has_org else "NULL AS organisation_id"

    #             if centres_exist:
    #                 fk_candidates = [
    #                     'service_provider_id', 'ServiceProviderId', 'ServiceProviderID',
    #                     'serviceProviderId', 'serviceprovider_id'
    #                 ]
    #                 fk_col = cn.execute(text(f"""
    #                     SELECT COLUMN_NAME
    #                     FROM information_schema.COLUMNS
    #                     WHERE TABLE_SCHEMA = 'vms_test'
    #                     AND TABLE_NAME   = 'tbl_serviceprovider_centres'
    #                     AND COLUMN_NAME IN ({",".join(repr(c) for c in fk_candidates)})
    #                     LIMIT 1
    #                 """)).scalar()

    #                 if fk_col is None:
    #                     # Fall back to a harmless self‑join – we still want the base table
    #                     centres_exist = False

    #             if centres_exist:
    #                 sql = f"""
    #                 SELECT
    #                     sp.id                           AS service_provider_source_id,
    #                     sp.service_provider_name        AS name,
    #                     sp.ServiceProviderCode          AS code,
    #                     sp.`3rdParty`                   AS is_3rd_party,
    #                     sp.is_active                    AS is_active,
    #                     spc.centre_name                 AS centre_name,
    #                     spc.centre_code                 AS centre_code,
    #                     {sp_org_sel}
    #                 FROM tbl_serviceproviders            sp
    #                 LEFT JOIN tbl_serviceprovider_centres spc
    #                     ON sp.id = spc.`{fk_col}`
    #                 """
    #             else:
    #                 sql = f"""
    #                 SELECT
    #                     sp.id                    AS service_provider_source_id,
    #                     sp.service_provider_name AS name,
    #                     sp.ServiceProviderCode   AS code,
    #                     sp.`3rdParty`            AS is_3rd_party,
    #                     sp.is_active             AS is_active,
    #                     NULL                     AS centre_name,
    #                     NULL                     AS centre_code,
    #                     {sp_org_sel}
    #                 FROM tbl_serviceproviders sp
    #                 """

    #             df = pd.read_sql_query(sql, cn)

    #             if df.empty:
    #                 # print("No service‑provider rows returned.")
    #                 return None

    #             # numeric dtypes
    #             df['service_provider_source_id'] = df['service_provider_source_id'].astype('int64')
    #             for col in ('is_3rd_party', 'is_active'):
    #                 df[col] = df[col].astype('Int8')

    #             # organisation_id is optional
    #             if 'organisation_id' in df.columns:
    #                 df['organisation_id'] = df['organisation_id'].astype('Int64')
    #             else:
    #                 df['organisation_id'] = pd.NA

    #             # clean string columns (don’t convert NaN → 'nan')
    #             for col in ('name', 'code', 'centre_name', 'centre_code'):
    #                 df[col] = df[col].where(df[col].notna(), None)

    #             df = df.reset_index(drop=True)
    #             df['service_provider_key'] = df.index + 1

    #             cols = [
    #                 'service_provider_key', 'service_provider_source_id', 'name',
    #                 'code', 'is_3rd_party', 'is_active',
    #                 'centre_name', 'centre_code', 'organisation_id'
    #             ]
    #             nparts = max(1, len(df) // 10_000)
    #             return dd.from_pandas(df[cols], npartitions=nparts)

    #     finally:
    #         src.dispose()

    # def load_service_provider_dimension(self, dim_service_provider):
    #     if dim_service_provider is None or dim_service_provider.npartitions == 0:
    #         # print("No service provider data to load.")
    #         return False
            
    #     print("Starting service provider load...")
    #     dim_service_provider_pd = dim_service_provider.compute()
        
    #     if dim_service_provider_pd.empty:
    #         # print("Computed service provider DataFrame is empty. Skipping load.")
    #         return False
        
    #     # Replace NaN with None for SQL compatibility
    #     dim_service_provider_pd = dim_service_provider_pd.replace({np.nan: None, pd.NaT: None})
            
    #     target_engine = create_engine("mysql+mysqlconnector://root:Root%40123@localhost/olap_2")
        
    #     try:
    #         with target_engine.begin() as connection:
    #             # Disable foreign key checks
    #             connection.execute(text("SET FOREIGN_KEY_CHECKS = 0"))
                
    #             # Drop existing table
    #             connection.execute(text("DROP TABLE IF EXISTS dim_service_provider"))
                
    #             # Create new table
    #             connection.execute(text("""
    #                 CREATE TABLE dim_service_provider (
    #                     service_provider_key INT NOT NULL PRIMARY KEY,
    #                     service_provider_source_id INT NOT NULL,
    #                     name VARCHAR(255),
    #                     code VARCHAR(50),
    #                     is_3rd_party TINYINT,
    #                     is_active TINYINT,
    #                     centre_name VARCHAR(255),
    #                     centre_code VARCHAR(50),
    #                     organisation_id INT DEFAULT NULL
    #                 )
    #             """))
                
    #             # Insert data
    #             insert_count = 0
    #             for _, row in dim_service_provider_pd.iterrows():
    #                 try:
    #                     params = {
    #                         'service_provider_key': int(row['service_provider_key']),
    #                         'service_provider_source_id': int(row['service_provider_source_id']) if pd.notna(row['service_provider_source_id']) else None,
    #                         'name': str(row['name']) if pd.notna(row['name']) else None,
    #                         'code': str(row['code']) if pd.notna(row['code']) else None,
    #                         'is_3rd_party': int(row['is_3rd_party']) if pd.notna(row['is_3rd_party']) else 0,
    #                         'is_active': int(row['is_active']) if pd.notna(row['is_active']) else 0,
    #                         'centre_name': str(row['centre_name']) if pd.notna(row['centre_name']) else None,
    #                         'centre_code': str(row['centre_code']) if pd.notna(row['centre_code']) else None,
    #                         'organisation_id': int(row['organisation_id']) if pd.notna(row['organisation_id']) else None
    #                     }
                        
    #                     connection.execute(text("""
    #                         INSERT INTO dim_service_provider
    #                         (service_provider_key, service_provider_source_id, name, code, 
    #                         is_3rd_party, is_active, centre_name, centre_code, organisation_id)
    #                         VALUES
    #                         (:service_provider_key, :service_provider_source_id, :name, :code,
    #                         :is_3rd_party, :is_active, :centre_name, :centre_code, :organisation_id)
    #                     """), params)
                        
    #                     insert_count += 1
                        
    #                     # Print progress
    #                     # if insert_count % 100 == 0:
    #                     #     print(f"Inserted {insert_count} records so far...")
                            
    #                 except Exception as e:
    #                     # print(f"Error inserting service provider row {row.get('service_provider_key', 'N/A')}: {e}")
    #                     # print(f"Problematic row data: {row.to_dict()}")
    #                     continue
                        
    #             # print(f"Successfully inserted {insert_count} service provider records.")
                
    #             # Re-enable foreign key checks
    #             connection.execute(text("SET FOREIGN_KEY_CHECKS = 1"))
                
    #         return True
    #     except Exception as e:
    #         print(f"An error occurred during service provider load: {e}")
    #         raise
    #     finally:
    #         target_engine.dispose()
            
            
    # def extract_contract_dimension(self):
    #         print("Starting contract extraction…")
            
    #         # Create engines for source (vms_test) and target (olap_2)
    #         src_engine = create_engine(
    #             "mysql+mysqlconnector://root:Root%40123@localhost/vms_test",
    #             pool_pre_ping=True,
    #         )
    #         tgt_engine = create_engine(
    #             "mysql+mysqlconnector://root:Root%40123@localhost/olap_2",
    #             pool_pre_ping=True,
    #         )
            
    #         try:
    #             with src_engine.connect() as cn:
    #                 # Query from the source; note we select c.company_id as company_id.
    #                 sql = """
    #                     SELECT
    #                         c.id                 AS contract_source_id,
    #                         c.contract_no        AS contract_number,
    #                         c.contract_name      AS contract_name,
    #                         ct.name              AS contract_type,
    #                         c.status             AS contract_status,
    #                         ct.is_active         AS is_contract_active,
    #                         c.company_id         AS company_id,
    #                         t.start_date         AS start_date,
    #                         t.end_date           AS end_date
    #                     FROM tbl_contracts c
    #                     LEFT JOIN tbl_contract_types ct ON ct.id = c.contract_type_id
    #                     LEFT JOIN tbl_contract_terms t  ON t.contract_id = c.id
    #                 """
    #                 contract_df = pd.read_sql_query(sql, cn)
                    
    #                 if contract_df.empty:
    #                     print("No contract rows returned.")
    #                     return None
                
    #             with tgt_engine.connect() as cn_tgt:
    #                 company_df = pd.read_sql_query(
    #                     "SELECT company_key, company_source_id FROM dim_company", 
    #                     cn_tgt
    #                 )
                
    #             if company_df.empty:
    #                 raise ValueError("dim_company is empty; cannot resolve company_key for contracts.")
                

    #             merged_df = contract_df.merge(
    #                 company_df,
    #                 how="left",
    #                 left_on="company_id",
    #                 right_on="company_source_id"
    #             )
                
    #             missing = merged_df["company_key"].isna().sum()
    #             if missing:
    #                 warnings.warn(
    #                     f"{missing} contract row(s) have no matching company in dim_company and will be skipped."
    #                 )
    #                 merged_df = merged_df[merged_df["company_key"].notna()]
                
    #             merged_df["contract_source_id"] = merged_df["contract_source_id"].astype("int64")
    #             merged_df["company_key"]        = merged_df["company_key"].astype("int64")
    #             merged_df["is_contract_active"] = merged_df["is_contract_active"].astype("Int8")
                
    #             merged_df = merged_df.reset_index(drop=True)
    #             merged_df["contract_key"] = merged_df.index + 1
                
    #             final_cols = [
    #                 "contract_key", "contract_source_id", "contract_number",
    #                 "contract_name", "contract_type", "contract_status",
    #                 "is_contract_active", "company_key", "start_date", "end_date",
    #             ]
    #             nparts = max(1, len(merged_df) // 10000)
    #             return dd.from_pandas(merged_df[final_cols], npartitions=nparts)
            
    #         finally:
    #             src_engine.dispose()
    #             tgt_engine.dispose()
                
    # def load_contract_dimension(self, dim_contract):
    #         if dim_contract is None or dim_contract.npartitions == 0:
    #             # print("No contract data to load.")
    #             return False

    #         print("Starting contract load…")
    #         pdf = dim_contract.compute().replace({np.nan: None, pd.NaT: None})

    #         tgt_engine = create_engine(
    #             "mysql+mysqlconnector://root:Root%40123@localhost/olap_2",
    #             pool_pre_ping=True,
    #         )
    #         try:
    #             with tgt_engine.begin() as connection:
    #                 connection.execute(text("SET FOREIGN_KEY_CHECKS = 0"))
    #                 connection.execute(text("DROP TABLE IF EXISTS dim_contract"))
    #                 connection.execute(text("""
    #                     CREATE TABLE dim_contract (
    #                         contract_key       INT       PRIMARY KEY,
    #                         contract_source_id INT,
    #                         contract_number    VARCHAR(50),
    #                         contract_name      VARCHAR(255),
    #                         contract_type      VARCHAR(50),
    #                         contract_status    VARCHAR(50),
    #                         is_contract_active TINYINT,
    #                         company_key        INT,
    #                         start_date         DATE,
    #                         end_date           DATE
    #                     )
    #                 """))

    #                 insert_sql = text("""
    #                     INSERT INTO dim_contract (
    #                         contract_key, contract_source_id, contract_number,
    #                         contract_name, contract_type, contract_status,
    #                         is_contract_active, company_key, start_date, end_date
    #                     )
    #                     VALUES (
    #                         :contract_key, :contract_source_id, :contract_number,
    #                         :contract_name, :contract_type, :contract_status,
    #                         :is_contract_active, :company_key, :start_date, :end_date
    #                     )
    #                 """)

    #                 inserted = 0
    #                 for _, r in pdf.iterrows():
    #                     try:
    #                         connection.execute(insert_sql, {
    #                             "contract_key":       int(r["contract_key"]),
    #                             "contract_source_id": int(r["contract_source_id"]),
    #                             "contract_number":    r["contract_number"],
    #                             "contract_name":      r["contract_name"],
    #                             "contract_type":      r["contract_type"],
    #                             "contract_status":    r["contract_status"],
    #                             "is_contract_active": int(r["is_contract_active"]) if pd.notna(r["is_contract_active"]) else None,
    #                             "company_key":        int(r["company_key"]),
    #                             "start_date":         r["start_date"],
    #                             "end_date":           r["end_date"],
    #                         })
    #                         inserted += 1
    #                     except Exception as e:
    #                         print(f"Error inserting contract_key {r['contract_key']}: {e}")

    #                 connection.execute(text("SET FOREIGN_KEY_CHECKS = 1"))
    #                 # print(f"Inserted {inserted} contract row(s).")
    #             return True
    #         finally:
    #             tgt_engine.dispose()

    # def extract_vehicle_dimension(self):
    #     print("Starting vehicle extraction...")
    #     src_engine = create_engine("mysql+mysqlconnector://root:Root%40123@localhost/vms_test")
    #     try:
    #         with src_engine.connect() as conn:
    #             sql = """
    #                 SELECT
    #                     vd.id AS vehicle_source_id,
    #                     vd.registration AS registration_number,
    #                     vd.alternative_registration,
    #                     vd.fleet_number,
    #                     vd.is_active,
    #                     vd.is_suspended,
    #                     vd.estimated_annual_mileage,
    #                     vd.actual_annual_mileage,
    #                     vd.total_contract_mileage,
    #                     vd.current_mileage,
    #                     vd.estimated_mileage,
    #                     vd.contract_start_date,
    #                     vd.fee AS vehicle_fee,
    #                     vm.name AS make,
    #                     vmo.name AS model,
    #                     vd.last_inspection AS last_inspection_date
    #                 FROM tbl_vehicle_details vd
    #                 LEFT JOIN tbl_vehicle_makes vm ON vd.make_id = vm.id
    #                 LEFT JOIN tbl_vehicle_models vmo ON vd.model_id = vmo.id
    #             """
    #             df = pd.read_sql_query(sql, conn)
    #             if df.empty:
    #                 print("No vehicle data extracted.")
    #                 return None
    #     finally:
    #         src_engine.dispose()

    #     # Revised safe conversion function: convert elementwise to integer (or pd.NA)
    #     def safe_convert_to_int(series):
    #         series = series.replace({"N/A": np.nan, "": np.nan})
    #         numeric_series = pd.to_numeric(series, errors='coerce')
    #         if numeric_series.isna().sum() > 0:
    #             problematic = series[numeric_series.isna()].unique()
    #             print(f"Warning: The following values in column '{series.name}' could not be converted and will be set to NA: {problematic}")
    #         converted = numeric_series.apply(lambda x: pd.NA if pd.isna(x) else int(round(x)))
    #         return converted.astype("Int64")

    #     int_cols = [
    #         "vehicle_source_id",
    #         "estimated_annual_mileage",
    #         "actual_annual_mileage",
    #         "total_contract_mileage",
    #         "current_mileage",
    #         "estimated_mileage"
    #     ]
    #     for col in int_cols:
    #         df[col] = safe_convert_to_int(df[col])
    #     for col in ["is_active", "is_suspended"]:
    #         df[col] = pd.to_numeric(df[col], errors='coerce').fillna(0).astype('Int8')
    #     df["vehicle_fee"] = pd.to_numeric(
    #         df["vehicle_fee"].astype(str).str.replace('[^0-9.-]', '', regex=True),
    #         errors='coerce'
    #     )
    #     date_cols = ["contract_start_date", "last_inspection_date"]
    #     for col in date_cols:
    #         df[col] = pd.to_datetime(df[col], errors="coerce").dt.date
    #     df["company_key"] = None
    #     df["contract_key"] = None
    #     df = df.reset_index(drop=True)
    #     df["vehicle_key"] = df.index + 1
    #     final_columns = [
    #         "vehicle_key",
    #         "vehicle_source_id",
    #         "registration_number",
    #         "alternative_registration",
    #         "fleet_number",
    #         "is_active",
    #         "is_suspended",
    #         "estimated_annual_mileage",
    #         "actual_annual_mileage",
    #         "total_contract_mileage",
    #         "current_mileage",
    #         "estimated_mileage",
    #         "contract_start_date",
    #         "vehicle_fee",
    #         "make",
    #         "model",
    #         "last_inspection_date",
    #         "company_key",
    #         "contract_key"
    #     ]
    #     df = df[final_columns]
    #     npartitions = max(1, len(df) // 50000)
    #     dim_vehicle = dd.from_pandas(df, npartitions=npartitions)
    #     # print(f"Extracted {len(df)} vehicle records.")
    #     return dim_vehicle

    # def load_vehicle_dimension(self, dim_vehicle):
    #     if dim_vehicle is None or dim_vehicle.npartitions == 0:
    #         # print("No vehicle data to load.")
    #         return False
    #     print("Starting vehicle load...")
    #     vehicle_df = dim_vehicle.compute().replace({np.nan: None, pd.NaT: None})
    #     tgt_engine = create_engine("mysql+mysqlconnector://root:Root%40123@localhost/olap_2")
    #     try:
    #         with tgt_engine.begin() as connection:
    #             # print("Disabling foreign key checks.")
    #             connection.execute(text("SET FOREIGN_KEY_CHECKS = 0"))
    #             # print("Dropping existing dim_vehicle table.")
    #             connection.execute(text("DROP TABLE IF EXISTS dim_vehicle"))
    #             # print("Creating new dim_vehicle table.")
    #             connection.execute(text("""
    #                 CREATE TABLE dim_vehicle (
    #                     vehicle_key               INT PRIMARY KEY,
    #                     vehicle_source_id         INT,
    #                     registration_number       VARCHAR(50),
    #                     alternative_registration  VARCHAR(50),
    #                     fleet_number              VARCHAR(50),
    #                     is_active                 TINYINT,
    #                     is_suspended              TINYINT,
    #                     estimated_annual_mileage  INT,
    #                     actual_annual_mileage     INT,
    #                     total_contract_mileage    INT,
    #                     current_mileage           INT,
    #                     estimated_mileage         INT,
    #                     contract_start_date       DATE,
    #                     vehicle_fee               DECIMAL(10,2),
    #                     make                      VARCHAR(100),
    #                     model                     VARCHAR(100),
    #                     last_inspection_date      DATE,
    #                     company_key               INT,
    #                     contract_key              INT
    #                 )
    #             """))
    #             # print(f"Inserting {len(vehicle_df)} vehicle records...")
    #             insert_count = 0
    #             for _, row in vehicle_df.iterrows():
    #                 try:
    #                     params = {
    #                         'vehicle_key': int(row['vehicle_key']),
    #                         'vehicle_source_id': int(row['vehicle_source_id']) if row['vehicle_source_id'] is not None else None,
    #                         'registration_number': row['registration_number'],
    #                         'alternative_registration': row['alternative_registration'],
    #                         'fleet_number': row['fleet_number'],
    #                         'is_active': int(row['is_active']) if row['is_active'] is not None else None,
    #                         'is_suspended': int(row['is_suspended']) if row['is_suspended'] is not None else None,
    #                         'estimated_annual_mileage': int(row['estimated_annual_mileage']) if row['estimated_annual_mileage'] is not None else None,
    #                         'actual_annual_mileage': int(row['actual_annual_mileage']) if row['actual_annual_mileage'] is not None else None,
    #                         'total_contract_mileage': int(row['total_contract_mileage']) if row['total_contract_mileage'] is not None else None,
    #                         'current_mileage': int(row['current_mileage']) if row['current_mileage'] is not None else None,
    #                         'estimated_mileage': int(row['estimated_mileage']) if row['estimated_mileage'] is not None else None,
    #                         'contract_start_date': row['contract_start_date'],
    #                         'vehicle_fee': float(row['vehicle_fee']) if row['vehicle_fee'] is not None else None,
    #                         'make': str(row['make']) if row['make'] is not None else None,
    #                         'model': str(row['model']) if row['model'] is not None else None,
    #                         'last_inspection_date': row['last_inspection_date'],
    #                         'company_key': int(row['company_key']) if row['company_key'] is not None else None,
    #                         'contract_key': int(row['contract_key']) if row['contract_key'] is not None else None
    #                     }
    #                     connection.execute(text("""
    #                         INSERT INTO dim_vehicle (
    #                             vehicle_key, vehicle_source_id, registration_number,
    #                             alternative_registration, fleet_number, is_active, is_suspended,
    #                             estimated_annual_mileage, actual_annual_mileage, total_contract_mileage,
    #                             current_mileage, estimated_mileage, contract_start_date, vehicle_fee,
    #                             make, model, last_inspection_date, company_key, contract_key
    #                         ) VALUES (
    #                             :vehicle_key, :vehicle_source_id, :registration_number,
    #                             :alternative_registration, :fleet_number, :is_active, :is_suspended,
    #                             :estimated_annual_mileage, :actual_annual_mileage, :total_contract_mileage,
    #                             :current_mileage, :estimated_mileage, :contract_start_date, :vehicle_fee,
    #                             :make, :model, :last_inspection_date, :company_key, :contract_key
    #                         )
    #                     """), params)
    #                     insert_count += 1
    #                 except Exception as e:
    #                     # print(f"Error inserting vehicle row {row.get('vehicle_key', 'N/A')}: {e}")
    #                     continue
    #             # print(f"Successfully inserted {insert_count} vehicle records.")
    #             # print("Re-enabling foreign key checks.")
    #             connection.execute(text("SET FOREIGN_KEY_CHECKS = 1"))
    #         return True
    #     except Exception as e:
    #         print(f"An error occurred during vehicle load: {e}")
    #         raise
    #     finally:
    #         tgt_engine.dispose()

    # def extract_fact_invoice_dimension(self):
    #     # print("Extracting fact_invoices via full-table reads + pandas…")

    #     cid = pd.read_sql_table('tbl_customer_invoice_detail',self.source_engine,schema='vms_test').rename(columns={'id':  'invoice_detail_source_id','fee': 'CustomerInvoiceValue'})
    #     ci = pd.read_sql_table('tbl_customer_invoices',self.source_engine,schema='vms_test').rename(columns={'id':'invoice_source_id','customer_invoice_group_id':'invoice_group_source_id','contract_id':'contract_source_id'})

    #     df = cid.merge(ci,left_on='customer_invoice_id',right_on='invoice_source_id',how='left')
    #     df['invoice_number'] = df['invoice_group_source_id'].astype(str)

    #     if 'interval_id' in df.columns:
    #         ii = pd.read_sql_table('tbl_invoice_intervals',self.source_engine,schema='vms_test').rename(columns={'name':'invoice_interval'})
    #         df = df.merge(ii[['id','invoice_interval']],left_on='interval_id',right_on='id',how='left')
    #     else:
    #         df['invoice_interval'] = None

    #     if 'vehicle_use_id' in df.columns:
    #         vu = pd.read_sql_table('tbl_vehicle_uses',self.source_engine,schema='vms_test').rename(columns={'name':'business_sector'})
    #         df = df.merge(vu[['id','business_sector']],left_on='vehicle_use_id',right_on='id',how='left')
    #     else:
    #         df['business_sector'] = None

    #     contract_dim = pd.read_sql_table('dim_contract',self.target_engine,schema='olap_2',columns=['contract_key','contract_source_id','company_key'])
    #     df = df.merge(contract_dim,on='contract_source_id',how='left')

    #     df['start_period'] = pd.to_datetime(df['start_period'])
    #     df['end_period']   = pd.to_datetime(df['end_period'])
    #     df['date_key']     = df['start_period'].dt.strftime('%Y%m%d').astype(int)
    #     df['invoice_date'] = df['start_period'].dt.date
    #     df['start_period'] = df['invoice_date']
    #     df['end_period']   = df['end_period'].dt.date
    #     df['period_label'] = df['start_period'].astype(str) + ' - ' + df['end_period'].astype(str)

    #     # 6) Surrogate key
    #     df = df.reset_index(drop=True)
    #     df['fact_invoice_key'] = df.index + 1

    #     final_cols = [
    #         'fact_invoice_key',
    #         'invoice_detail_source_id','invoice_source_id',
    #         'invoice_group_source_id','invoice_number',
    #         'contract_key','company_key',
    #         'date_key','invoice_date','CustomerInvoiceValue',
    #         'start_period','end_period','period_label',
    #         'invoice_interval','business_sector'
    #     ]
    #     df = df[final_cols]

    #     return dd.from_pandas(df, npartitions=max(1, len(df)//50000))

    # def load_fact_invoice_dimension(self, dim_fact_invoice):
    #     if dim_fact_invoice is None or dim_fact_invoice.npartitions == 0:
    #         print("No fact_invoices to load, skipping.")
    #         return False

    #     print("Loading fact_invoices via autocommit connection…")
    #     df = dim_fact_invoice.compute().replace({pd.NaT: None, np.nan: None})

    #     with self.target_engine.connect().execution_options(autocommit=True) as conn:
    #         df.to_sql(
    #             'fact_invoices',
    #             con=conn,
    #             schema='olap_2',
    #             if_exists='replace',
    #             index=False,
    #             chunksize=5000,
    #             dtype={
    #                 'fact_invoice_key':         types.BigInteger,
    #                 'invoice_detail_source_id': types.Integer,
    #                 'invoice_source_id':        types.Integer,
    #                 'invoice_group_source_id':  types.Integer,
    #                 'invoice_number':           types.String(50),
    #                 'contract_key':             types.Integer,
    #                 'company_key':              types.Integer,
    #                 'date_key':                 types.Integer,
    #                 'invoice_date':             types.Date,
    #                 'CustomerInvoiceValue':     types.Numeric(10,2),
    #                 'start_period':             types.Date,
    #                 'end_period':               types.Date,
    #                 # increase length here to fit e.g. "2010-05-31 - 2010-06-30" (23 chars)
    #                 'period_label':             types.String(30),
    #                 'invoice_interval':         types.String(50),
    #                 'business_sector':          types.String(100),
    #             }
    #         )

    #     print("fact_invoices loaded.")
    #     return True

    # def extract_fact_jobs_dimension(self):
    #     detail_cols = ['id','JobId','ProductId','JobFitmentReasonId','Quantity','UnitCost','UnitPrice','IsTyre','IsRecharge','CustomerInvoiceDetailId','Wheel']
       
    #     jd = pd.read_sql_table('tbl_job_details', self.source_engine,schema='vms_test', columns=detail_cols).rename(columns={
    #         'id':'job_detail_source_id',
    #         'JobId':'job_source_id',
    #         'ProductId':'product_source_id',
    #         'JobFitmentReasonId':'fitment_reason_source_id',
    #         'Quantity':'quantity',
    #         'UnitCost':'cost_to_us',
    #         'UnitPrice':'selling_price_for_PAYG_contract',
    #         'IsTyre':'is_tyre',
    #         'IsRecharge':'is_recharge',
    #         'CustomerInvoiceDetailId':'customer_invoice_detail_id',
    #         'Wheel':'wheel_function_id',
    #     })

    #     hdr = pd.read_sql_table('tbl_jobs', self.source_engine,schema='vms_test',columns=['id','AuthorisationDate','vehicle_reg_id','contract_id','company_id','service_provider_id']
    #     ).rename(columns={
    #         'id':'job_source_id',
    #         'AuthorisationDate':'authorisation_datetime',
    #         'vehicle_reg_id':'vehicle_source_id',
    #         'contract_id':'contract_source_id',
    #         'company_id':'company_source_id',
    #         'service_provider_id':'service_provider_source_id'
    #     })
    #     hdr['authorisation_datetime'] = pd.to_datetime(hdr['authorisation_datetime'], errors='coerce')
       
    #     hdr['authorisation_date_key'] = pd.to_numeric(hdr['authorisation_datetime'].dt.strftime('%Y%m%d'),errors='coerce').astype('Int64')

    #     df = jd.merge(hdr, on='job_source_id', how='left')

    #     dims = [
    #         ('dim_product','product_source_id','product_key'),
    #         ('dim_fitment_reason','fitment_reason_source_id','fitment_reason_key'),
    #         ('dim_vehicle','vehicle_source_id','vehicle_key'),
    #         ('dim_contract','contract_source_id','contract_key'),
    #         ('dim_company','company_source_id','company_key'),
    #         ('dim_service_provider','service_provider_source_id','service_provider_key'),
    #     ]
    #     for table, src, tgt in dims:
    #         small = pd.read_sql_table(table, self.target_engine,schema='olap_2', columns=[src, tgt])
    #         lookup = dict(zip(small[src], small[tgt]))
    #         df[tgt] = df[src].map(lookup).astype('Int64')

    #     # Get the invoice detail information
    #     inv_det = pd.read_sql_query(
    #         "SELECT id AS customer_invoice_detail_id, customer_invoice_id AS invoice_source_id "
    #         "FROM tbl_customer_invoice_detail",
    #         self.source_engine
    #     )
    #     df = df.merge(inv_det, on='customer_invoice_detail_id', how='left')

    #     # Get the invoice information
    #     inv = pd.read_sql_query(
    #         "SELECT id AS invoice_source_id, total_invoice_exc_vat, "
    #         "total_invoice_vat AS customer_invoice_value "
    #         "FROM tbl_customer_invoices",
    #         self.source_engine
    #     )
    #     df = df.merge(inv, on='invoice_source_id', how='left')
        
    #     # Use the tbl_customer_invoice_groups.id as the invoice_number
    #     # Create a mapping between invoice_source_id and invoice_group_id
    #     invoice_mapping = pd.read_sql_query(
    #         "SELECT id AS invoice_source_id, customer_invoice_group_id AS invoice_group_id "
    #         "FROM tbl_customer_invoices",
    #         self.source_engine
    #     )
    #     # Merge to get the invoice_group_id
    #     df = df.merge(invoice_mapping, on='invoice_source_id', how='left')
        
    #     # Use the invoice_group_id as the invoice_number as per the requirement
    #     df['invoice_number'] = df['invoice_group_id'].astype(str)

    #     df['profit_for_PAYG_contract'] = (df['selling_price_for_PAYG_contract'] - df['cost_to_us'])

    #     jt = pd.read_sql_query(
    #         "SELECT id AS product_source_id, report_code AS job_type FROM tbl_products",
    #         self.source_engine
    #     )
    #     df = df.merge(jt, on='product_source_id', how='left')

    #     td = pd.read_sql_query("SELECT * FROM tbl_job_tread_depths", self.source_engine)
    #     jd_key = next((c for c in td.columns if 'job' in c.lower() and 'detail' in c.lower()), None)
    #     depth_c = next((c for c in td.columns if 'tread' in c.lower()), None)
    #     if jd_key and depth_c:
    #         td = td[[jd_key, depth_c]].rename(columns={
    #             jd_key: 'job_detail_source_id',
    #             depth_c: 'tread_depth'
    #         })
    #         df = df.merge(td, on='job_detail_source_id', how='left')
    #     else:
    #         df['tread_depth'] = None

    #     wf = pd.read_sql_table('tbl_wheel_functions', self.source_engine,schema='vms_test', columns=['id','name']).rename(columns={'id':'wheel_function_id','name':'wheel_position'})
    #     df = df.merge(wf, on='wheel_function_id', how='left')

    #     df['call_out_type'] = None

    #     df = df.reset_index(drop=True)
    #     df['fact_jobs_key'] = df.index + 1

    #     final_cols = [
    #         'fact_jobs_key',
    #         'job_detail_source_id','job_source_id',
    #         'product_key','vehicle_key','contract_key','company_key',
    #         'fitment_reason_key','service_provider_key',
    #         'authorisation_date_key','authorisation_datetime',
    #         'job_type','quantity','cost_to_us',
    #         'selling_price_for_PAYG_contract','profit_for_PAYG_contract',
    #         'is_tyre','is_recharge',
    #         'total_invoice_exc_vat','customer_invoice_value',
    #         'invoice_number','tread_depth','wheel_position',
    #         'call_out_type'
    #     ]
    #     return dd.from_pandas(df[final_cols], npartitions=max(1, len(df)//50000))
    
    # def load_fact_jobs_dimension(self, dim_fact_jobs):
    #     if dim_fact_jobs is None or dim_fact_jobs.npartitions == 0:
    #         print("No fact_jobs to load, skipping.")
    #         return False

    #     print("Loading fact_jobs via autocommit connection…")
    #     df = dim_fact_jobs.compute().replace({pd.NaT: None, np.nan: None})

    #     df['job_type'] = df['job_type'].replace({'': None})

    #     with self.target_engine.connect().execution_options(autocommit=True) as conn:
    #         df.to_sql(
    #             'fact_jobs',
    #             con=conn,
    #             schema='olap_2',
    #             if_exists='replace',
    #             index=False,
    #             chunksize=5000,
    #             dtype={
    #                 'fact_jobs_key':                   types.BigInteger,
    #                 'job_detail_source_id':            types.Integer,
    #                 'job_source_id':                   types.Integer,
    #                 'product_key':                     types.Integer,
    #                 'vehicle_key':                     types.Integer,
    #                 'contract_key':                    types.Integer,
    #                 'company_key':                     types.Integer,
    #                 'fitment_reason_key':              types.Integer,
    #                 'service_provider_key':            types.Integer,
    #                 'authorisation_date_key':          types.Integer,
    #                 'authorisation_datetime':          types.DateTime,
    #                 # Use VARCHAR(50) instead of ENUM
    #                 'job_type':                        types.String(50),
    #                 'quantity':                        types.Numeric(10,2),
    #                 'cost_to_us':                      types.Numeric(10,2),
    #                 'selling_price_for_PAYG_contract': types.Numeric(10,2),
    #                 'profit_for_PAYG_contract':        types.Numeric(10,2),
    #                 'is_tyre':                         types.Integer,
    #                 'is_recharge':                     types.Integer,
    #                 'total_invoice_exc_vat':           types.Numeric(10,2),
    #                 'customer_invoice_value':          types.Numeric(10,2),
    #                 'invoice_number':                  types.String(50),
    #                 'tread_depth':                     types.Numeric(5,2),
    #                 'wheel_position':                  types.String(50),
    #                 'call_out_type':                   types.Enum('DAY','NIGHT','UNKNOWN')
    #             }
    #         )

    #     print("fact_jobs loaded.")
    #     return True
            
    # # --- Runner Methods ---
    # def run_company_etl(self):
    #     print("\n--- Starting Company ETL ---")
    #     dim_company = self.extract_company_dimension()
    #     if dim_company is not None:
    #         self.load_company_dimension(dim_company)
    #     else:
    #         print("Company extraction returned no data. Skipping load.")

             
        
    # def run_fitment_reason_etl(self):
    #     print("\n--- Starting Fitment Reason ETL ---")
    #     dim_fitment_reason = self.extract_fitment_reason_dimension()
    #     if dim_fitment_reason is not None:
    #         self.load_fitment_reason_dimension(dim_fitment_reason)
    #     else:
    #         print("Fitment Reason extraction returned no data. Skipping load.")

        
    # def run_service_provider_etl(self):
    #     print("\n--- Starting Service Provider ETL ---")
    #     dim_service_provider = self.extract_service_provider_dimension()
    #     if dim_service_provider is not None:
    #         self.load_service_provider_dimension(dim_service_provider)
    #     else:
    #         print("Service Provider extraction returned no data. Skipping load.")
    

    # def run_product_etl(self):
    #     print("\n--- Starting Product ETL ---")
    #     dim_product = self.extract_product_dimension()
    #     if dim_product is not None:
    #         self.load_product_dimension(dim_product)
    #     else:
    #         print("Product extraction returned no data. Skipping load.")
   

    # def run_date_etl(self):
    #     print("\n--- Starting Date Dimension ETL ---")
    #     dim_date = self.extract_date_dimension()
    #     if dim_date is not None:
    #         self.load_date_dimension(dim_date)
    #     else:
    #         print("Date dimension generation returned no data. Skipping load.")
        
    # def run_contract_etl(self):
    #     print("\n--- Starting Contract ETL ---")
    #     dim_contract = self.extract_contract_dimension()
    #     if dim_contract is not None:
    #         self.load_contract_dimension(dim_contract)
    #     else:
    #         print("Contract extraction returned no data. Skipping load.")

        
        
    # def run_vehicle_etl(self):
    #     print("\n--- Starting Vehicle ETL ---")
    #     dim_vehicle = self.extract_vehicle_dimension()
    #     if dim_vehicle is not None:
    #         self.load_vehicle_dimension(dim_vehicle)
    #     else:
    #         print("Vehicle extraction returned no data. Skipping load.")
    
    
    # def run_fact_invoice_etl(self):
    #     print("\n--- Starting Fact Invoice ETL ---")
    #     ddf = self.extract_fact_invoice_dimension()
    #     if ddf is not None:
    #         self.load_fact_invoice_dimension(ddf)
    #     else:
    #         print("Extraction returned no data; skipping.")
    
    

    # def run_fact_jobs_etl(self):
    #     print("\n--- Starting Fact Jobs ETL ---")
    #     ddf = self.extract_fact_jobs_dimension()
    #     if ddf is not None:
    #         self.load_fact_jobs_dimension(ddf)
    #     else:
    #         print("Extraction returned no data; skipping.")

    def close(self):
        print("\n--- ETL pipeline execution finished ---")

# --- Main Execution Block ---
if __name__ == "__main__":
    etl = ETLPipeline(start_year=2000, end_year=2030)
    etl.run_date_etl()
    # etl.run_company_etl()
    # etl.run_fitment_reason_etl()
    # etl.run_product_etl()
    # etl.run_service_provider_etl()
    # etl.run_contract_etl()
    # etl.run_vehicle_etl()
    # etl.run_fact_invoice_etl()
    # etl.run_fact_jobs_etl()
    # etl.run_fact_inspections_etl()
    etl.close()
    
Create a complete ETL pipeline with different Python files organized in a modular structure, with a final file called main.py to orchestrate the process. Please include logic for incremental ETL in the code to efficiently process only new or changed data since the last extraction.