I have a python script that migrates data from 2 tables in a schema called "migration" into a few other tables in a different schema. Migration involves encrypting a column and doing some simple transformations on others. If the primary key cif already exists, we only update a few columns instead of inserting a new record.
I intend to migrate several million records in batches of 100,000. Modify this script to be as efficient as possible to make it run as quickly as possible. (Maybe we we should parallelize encryption separately/do inserts, updates in batches/ maybe use pandas?/something else? I'll let you decide the best approach)

```
from utils.encryption import encrypt_aes 
import pandas as pd 
import traceback 
import psycopg2 
import time 
from datetime import datetime 

# Encryption key for password encryption
ENCRYPTION_KEY = 'CiIyciDeIgyIBhecJtdhbYTRHR3ojFdvrrJF1xkzLzw=' 

# Database connection parameters for source database
src_db_params = { 
    'host': 'localhost',  
    'database': 'SBI', 
    'user': 'postgres',  
    'password': '1234',  
    'port': '5432' 
} 

# Database connection parameters for target database
target_db_params = { 
    'host': 'localhost',  
    'database': 'venugopal-test', 
    'user': 'postgres',  
    'password': '1234',  
    'port': '5432' 
} 

# Column mapping for the source table
ref_inb_col_mapping = { 
    'user_id': 0,  
    'user_name': 1,  
    'password': 2,  
    'user_state': 3,  
    'lock_time': 4,  
    'lock_count': 5,  
    'perm_lock_count': 6,  
    'user_block': 7,  
    'last_login_date': 8,  
    'pwd_changed_date': 9,  
    'cif': 10,  
    'login_count': 11,  
    'user_cr_time': 12,  
    'creation_time': 13,  
    'last_mod_time': 14,  
    'mig_status_flag': 15,  
    'migrated_time': 16,  
    'gg_sync_time': 17,  
    'fsl_process_start_time': 18,  
    'gg_txn_time': 19 
} 

def normalize_cif(s): 
    """Normalize CIF by padding with zeros to ensure 17 digit length"""
    return s.zfill(17) 

def check_and_create_migration_status_table(cur, conn):
    try:
        # Check if the migration schema exists
        cur.execute("""
            SELECT EXISTS (
                SELECT FROM information_schema.schemata 
                WHERE schema_name = 'migration'
            );
        """)
        schema_exists = cur.fetchone()[0]

        if not schema_exists:
            print("Schema 'migration' does not exist. Creating it now.")
            cur.execute('CREATE SCHEMA migration;')
            conn.commit()
            print("Schema 'migration' created successfully.")

        # Check if the table exists
        cur.execute("""
            SELECT EXISTS (
                SELECT FROM information_schema.tables 
                WHERE  table_schema = 'migration'
                AND    table_name   = 'sbi_fsl_inb_mig_status_new'
            );
        """)
        table_exists = cur.fetchone()[0]

        if not table_exists:
            print("Table 'migration.sbi_fsl_inb_mig_status_new' does not exist. Creating it now.")
            cur.execute('''
                CREATE TABLE migration.sbi_fsl_inb_mig_status_new
                (
                    user_id numeric(38,0),
                    user_name character varying(255),
                    mig_status_flag character(1),
                    migrated_time timestamp without time zone
                )
            ''')
            conn.commit()
            print("Table 'migration.sbi_fsl_inb_mig_status_new' created successfully.")
        else:
            print("Table 'migration.sbi_fsl_inb_mig_status_new' already exists.")
    except Exception as e:
        print(f"Error checking/creating migration status table: {e}")
        conn.rollback()
        raise

def update_or_insert_migration_status(cur, conn, user_id, user_name, mig_status_flag, migrated_time):
    try:
        # Check if the user_id already exists
        cur.execute('''
            SELECT COUNT(*) FROM migration.sbi_fsl_inb_mig_status_new 
            WHERE user_id = %s
        ''', (user_id,))
        user_exists = cur.fetchone()[0] > 0

        if user_exists:
            # Update the existing record
            cur.execute('''
                UPDATE migration.sbi_fsl_inb_mig_status_new 
                SET mig_status_flag = %s, migrated_time = %s
                WHERE user_id = %s
            ''', (mig_status_flag, migrated_time, user_id))
            print(f"Updated migration status for user_id: {user_id}")
        else:
            # Insert a new record
            cur.execute('''
                INSERT INTO migration.sbi_fsl_inb_mig_status_new 
                (user_id, user_name, mig_status_flag, migrated_time) 
                VALUES (%s, %s, %s, %s)
            ''', (user_id, user_name, mig_status_flag, migrated_time))
            print(f"Inserted new migration status for user_id: {user_id}")

        conn.commit()
    except Exception as e:
        print(f"Error updating/inserting migration status for user_id {user_id}: {e}")
        conn.rollback()
        raise

def rinb_main_debug(): 
    try: 
        print("Migration process initiated.") 
        
        # Establish database connections
        source_conn = psycopg2.connect(**src_db_params) 
        source_cur = source_conn.cursor() 
        target_conn = psycopg2.connect(**target_db_params) 
        target_cur = target_conn.cursor() 
        
        # Check and create migration status table in the source database
        check_and_create_migration_status_table(source_cur, source_conn)
        
        # Fetch CIFs to process
        query = """SELECT cif from migration.sbi_yono1_fsl_mpin_staging where batch_id=2;""" 
        source_cur.execute(query) 
        INPUT_CIF_LIST = [x[0] for x in source_cur.fetchall()]
        
        print(f"Processing {len(INPUT_CIF_LIST)} CIFs")

        # Get column information from source table
        query = """SELECT column_name, data_type from information_schema.columns 
        where table_schema='migration' AND table_name='sbi_inb_fsl_gg';""" 
        source_cur.execute(query) 
        columns = source_cur.fetchall() 
        
        for i, col in enumerate(columns): 
            ref_inb_col_mapping[col[0]] = i 
        print("Column mapping completed successfully")     
        
        # Start migration process
        migration_start_time = time.time() 
        migration_timestamp = datetime.now() 
        
        # Filter out duplicate CIFs 
        query_duplicated = """ 
            SELECT cif 
            FROM migration.sbi_inb_fsl_gg 
            GROUP BY cif 
            HAVING COUNT(*) > 1; 
        """ 
        source_cur.execute(query_duplicated) 
        duplicated_cifs = [row[0] for row in source_cur.fetchall()] 
        duplicated_cifs_in_input = set(duplicated_cifs).intersection(set(INPUT_CIF_LIST)) 
        print("Duplicate CIFs found in INB table for the following CIFs", duplicated_cifs_in_input) 
        filtered_cif_tuple = tuple([c for c in INPUT_CIF_LIST if c not in duplicated_cifs]) 
        print("Filtered CIF list:", filtered_cif_tuple) 
        
        # Fetch rows to migrate
        query = """SELECT * FROM migration.sbi_inb_fsl_gg WHERE cif in %s;""" 
        source_cur.execute(query, (filtered_cif_tuple,)) 
        rows = source_cur.fetchall() 
        print(f"Migrating {len(rows)} rows")  
        
        mapped_cmaster_rows = [] 
        mapped_lock_mgmt_rows = [] 
        migration_report = [] 
        customer_id_i = 1 
        customer_lock_mgmt_id_i = 1 
        last_insert_customer_id = None 
        failed_records = [] 
        
        for i, row in enumerate(rows): 
            print(f"Processing row {i}")
            try: 
                # Extract data from source row
                inb_user_id_ = row[ref_inb_col_mapping['user_id']] 
                inb_cif_ = row[ref_inb_col_mapping['cif']] 
                print("CIF:", inb_cif_) 
                inb_login_ct_ = row[ref_inb_col_mapping['login_count']] 
                cif_no = normalize_cif(inb_cif_) if inb_cif_ is not None else inb_cif_ 
                inb_creation_time_ = row[ref_inb_col_mapping['creation_time']] 
                inb_modification_time_ = row[ref_inb_col_mapping['last_mod_time']] 
                created_by = 2 
                inb_user_state_ = row[ref_inb_col_mapping['user_state']] 
                
                # Determine customer status
                if inb_user_state_ == 15: 
                    customer_status = 0 
                else: 
                    customer_status = row[ref_inb_col_mapping['user_state']] 
                
                # Determine created date
                if inb_creation_time_: 
                    created_date = inb_creation_time_  
                elif customer_status != 15: 
                    created_date = migration_start_time 
                else: 
                    print("Invalid created date. Skipping record...") 
                    failed_records.append((inb_user_id_, inb_cif_)) 
                    continue 
                
                email_id = None 
                global_block_system = None 
                global_block_time = None 
                global_lock = True if inb_user_state_ in [1, 7, 9, 11] else False 
                mob_country_code = None 
                
                # Set mobile_number with a default value if not present
                mobile_number_index = ref_inb_col_mapping.get('mobile_number', None)
                if mobile_number_index is not None and mobile_number_index < len(row):
                    mobile_number = row[mobile_number_index] if row[mobile_number_index] else 'N/A'
                else:
                    mobile_number = None  # Default value if the column is not found
                
                modified_date = inb_modification_time_ 
                modified_by = 2 if inb_modification_time_ else None 
                mpin = None 
                
                # Encrypt password
                encryption_success, password = encrypt_aes(row[ref_inb_col_mapping['password']], key_base64=ENCRYPTION_KEY) 
                if not encryption_success: 
                    print("Encryption failed... Adding to failed records", (inb_user_id_, cif_no, row[ref_inb_col_mapping['password']])) 
                    continue 
                
                inb_perm_lock_ct_ = row[ref_inb_col_mapping['perm_lock_count']] 
                perm_lock_count = 0 if inb_perm_lock_ct_ is None else inb_perm_lock_ct_ 
                perm_lock_time = None 
                pseudo_id = inb_user_id_ 
                user_name = row[ref_inb_col_mapping['user_name']] 
                cmpin = None 
                
                # Check if customer already exists
                target_cur.execute(''' 
                    SELECT customer_id FROM public.emas_customer_master WHERE cif_no = %s; 
                ''', (cif_no,))
                existing_customer = target_cur.fetchone()
                print(f"Existing customer: {existing_customer}")
                
                if existing_customer:
                    # Update existing customer
                    customer_id = existing_customer[0]
                    last_insert_customer_id = customer_id
                    target_cur.execute(''' 
                        UPDATE public.emas_customer_master 
                        SET created_by = %s, created_date = %s, email_id = %s, global_block_system = %s, 
                            global_block_time = %s, global_lock = %s, mob_country_code = %s, 
                            mobile_number = %s, mpin = %s, password = %s, 
                            perm_lock_count = %s, perm_lock_time = %s, pseudo_id = %s, 
                            user_name = %s, cmpin = %s, customer_status = %s 
                        WHERE customer_id = %s; 
                    ''', (created_by, created_date, email_id, global_block_system, global_block_time, 
                          global_lock, mob_country_code, mobile_number, mpin, password, 
                          perm_lock_count, perm_lock_time, pseudo_id, user_name, cmpin, customer_status, 
                          customer_id))
                else:
                    # Insert new customer
                    target_cur.execute(''' 
                        INSERT INTO public.emas_customer_master (cif_no, created_by, created_date, email_id, 
                            global_block_system, global_block_time, global_lock, mob_country_code, 
                            mobile_number, mpin, password, perm_lock_count, perm_lock_time, 
                            pseudo_id, user_name, cmpin, customer_status)  
                        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) 
                        RETURNING customer_id; 
                    ''', (cif_no, created_by, created_date, email_id, global_block_system, global_block_time, 
                          global_lock, mob_country_code, mobile_number, mpin, password, 
                          perm_lock_count, perm_lock_time, pseudo_id, user_name, cmpin, customer_status))
                    
                    last_insert_customer_id = target_cur.fetchone()[0] 
                
                # Prepare data for emas_customer_lock_mgmt
                customer_lock_mgmt_id = customer_lock_mgmt_id_i 
                channel_blocked_by = None 
                channel_blocked_time = None 
                channel_status = 15 if inb_user_state_ == 15 else 0 
                created_by = 2 
                created_date = inb_creation_time_ 
                hard_lock_count = 0 
                hard_lock_time = None 
                last_failed_login_time = None 
                last_success_login_time = row[ref_inb_col_mapping['last_login_date']] 
                last_updated_time = row[ref_inb_col_mapping['pwd_changed_date']] 
                if not last_updated_time: 
                    last_updated_time = inb_creation_time_ if inb_creation_time_ else migration_timestamp 
                
                modified_date = inb_modification_time_ 
                modified_by = 2 if modified_date else None 
                inb_lock_ct_ = row[ref_inb_col_mapping['lock_count']] 
                temp_lock_count = inb_lock_ct_ if inb_lock_ct_ else 0 
                temp_lock_time = row[ref_inb_col_mapping['lock_time']] 
                channel_id = 1 # WEB channel 
                login_count = inb_login_ct_ if inb_login_ct_ else 0 
                
                if last_insert_customer_id is None: 
                    print("Error fetching customer ID from emas_customer_master") 
                    continue 
                customer_id = last_insert_customer_id
                
                # Update or insert emas_customer_lock_mgmt
                target_cur.execute('''
                    UPDATE public.emas_customer_lock_mgmt
                    SET channel_blocked_by = %s,
                        channel_blocked_time = %s,
                        channel_status = %s,
                        created_by = %s,
                        created_date = %s,
                        hard_lock_count = %s,
                        hard_lock_time = %s,
                        last_failed_login_time = %s,
                        last_success_login_time = %s,
                        last_updated_time = %s,
                        modified_date = %s,
                        modified_by = %s,
                        temp_lock_count = %s,
                        temp_lock_time = %s,
                        channel_id = %s,
                        login_count = %s
                    WHERE customer_id = %s
                ''', (channel_blocked_by, channel_blocked_time, channel_status, created_by, created_date, 
                      hard_lock_count, hard_lock_time, last_failed_login_time, last_success_login_time,
                      last_updated_time, modified_date, modified_by, temp_lock_count, temp_lock_time,
                      channel_id, login_count, customer_id))
            
                # If no rows were updated, insert a new record
                if target_cur.rowcount == 0:
                    target_cur.execute('''
                        INSERT INTO public.emas_customer_lock_mgmt 
                        (channel_blocked_by, channel_blocked_time, channel_status, created_by, created_date,
                         hard_lock_count, hard_lock_time, last_failed_login_time, last_success_login_time,
                         last_updated_time, modified_date, modified_by, temp_lock_count, temp_lock_time,
                         channel_id, customer_id, login_count)  
                        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) 
                    ''', (channel_blocked_by, channel_blocked_time, channel_status, created_by, created_date,
                          hard_lock_count, hard_lock_time, last_failed_login_time, last_success_login_time,
                          last_updated_time, modified_date, modified_by, temp_lock_count, temp_lock_time,
                          channel_id, customer_id, login_count))
                
                # Update or insert migration status
                mig_status_flag = 'P'  # Successful migration
                migrated_time = datetime.now()
                update_or_insert_migration_status(source_cur, source_conn, inb_user_id_, user_name, mig_status_flag, migrated_time)
                
                migration_report.append((cif_no, customer_id))
                target_conn.commit()
            
            except Exception as e:
                print(f"Failed processing row for user_id {inb_user_id_}", traceback.format_exc())
                target_conn.rollback()
                
                # Update or insert failed migration status
                try:
                    mig_status_flag = 'F'  # Failed migration
                    migrated_time = datetime.now()
                    update_or_insert_migration_status(source_cur, source_conn, inb_user_id_, user_name, mig_status_flag, migrated_time)
                except Exception as status_error:
                    print(f"Failed to update/insert failure status for user_id: {inb_user_id_}. Error: {status_error}")
        
        print("Migrated Data", migration_report) 
        pd.DataFrame(migration_report).to_csv(f'output/migration_report_{datetime.now().strftime("%Y_%m_%d_%H_%M_%S_%f")}.csv') 
        print("Time to migrate:", time.time() - migration_start_time) 
        
    except Exception as e: 
        print("An error occurred during the migration process:", traceback.format_exc()) 
    finally:
        # Close all database connections
        if 'source_cur' in locals():
            source_cur.close() 
        if 'source_conn' in locals():
            source_conn.close() 
        if 'target_cur' in locals():
            target_cur.close() 
        if 'target_conn' in locals():
            target_conn.close() 

# Execute the main function
if __name__ == "__main__":
    rinb_main_debug()
```