""" CSV to SQL INSERT Statement Generator Script ==================================== This script converts CSV files to SQL INSERT statements. USAGE: ------ 1. Place your CSV files in a directory (named exactly like the tables, e.g., "account.csv") 2. Run the script: python generate_contract_snapshot_sql_files.py 3. When prompted, enter the paths for your CSV files and where to save the SQL files (or press Enter to use the current directory as default) FEATURES: --------- - Creates 10 SQL files with INSERT statements from corresponding CSV files - Uses batch inserts (1000 rows per statement) for efficiency with large datasets - Handles special cases like NOW() function calls, NULL values, and proper string escaping - Formats SQL according to standard practices with proper value handling REQUIREMENTS: ------------ - Python 3.6 or higher (for f-string support) - Standard library only (no external dependencies) CSV FILE FORMAT: --------------- - First row should contain column names (headers) - Second row onward contains data values - Special value "NOW()" will be preserved as SQL function call - Empty values will be converted to NULL Example CSV (account.csv): ------------------------- id,name,currency_code,created_by,created_at 1,"Sample Account","USD","admin",NOW() 2,"Another Account","EUR","system",NOW() OUTPUT: ------- The script will generate SQL files with INSERT statements in the following format: INSERT INTO royalty_accounting.[table_name] (col1,col2,...) VALUES (val1,val2,...), (val1,val2,...), ... ; """ import os import csv import datetime from pathlib import Path # Increase CSV field size limit to handle large fields import sys csv.field_size_limit(sys.maxsize) def csv_to_sql_insert(csv_file, table_name, output_dir): """ Convert a CSV file to SQL INSERT statements. Args: csv_file (str): Path to the CSV file table_name (str): Name of the table for the INSERT statement output_dir (str): Directory to save the SQL file """ sql_file = os.path.join(output_dir, f"{table_name}.sql") try: with open(csv_file, 'r', encoding='utf-8') as csv_input, open(sql_file, 'w', encoding='utf-8') as sql_output: csv_reader = csv.reader(csv_input) headers = next(csv_reader) # Get column names from first row # Write a comment header in the SQL file sql_output.write(f"-- SQL Insert script for {table_name}\n") sql_output.write(f"-- Generated on {datetime.datetime.now()}\n") sql_output.write(f"-- Source file: {csv_file}\n\n") # Process data rows row_count = 0 batch_size = 1000 # Number of rows per batch insert batch_rows = [] # Define ID columns that should be set to NULL id_columns_to_nullify = { "ACCOUNT_PAYMENT_TERM": ["account_payment_term_id"], "ACCOUNT_PAYEE": ["account_payee_id"], "ACCOUNT_TAX_INFO": ["account_tax_info_id"], "RUN_CONTROLLER_CONTRACT": ["run_controller_contract_id"], "LEGACY_CONTRACT": ["legacy_contract_id"], "ACCOUNT_CONTRACT": ["account_contract_id"] } # Get index positions of headers to nullify for this table null_indices = [] if table_name in id_columns_to_nullify: for id_col in id_columns_to_nullify[table_name]: for i, header in enumerate(headers): if header.lower() == id_col.lower(): null_indices.append(i) # Get index positions for last_modified and last_modified_by last_modified_index = None last_modified_by_index = None for i, header in enumerate(headers): if header.lower() == 'last_modified': last_modified_index = i elif header.lower() == 'last_modified_by': last_modified_by_index = i for row in csv_reader: # Process each row row_count += 1 values = [] for i, value in enumerate(row): # Replace ID fields with NULL if i in null_indices: values.append("NULL") # Replace last_modified with NOW() elif i == last_modified_index: values.append("NOW()") # Replace last_modified_by elif i == last_modified_by_index: values.append("'orchard-migraton-test'") # Handle other values normally elif not value: # Handle empty values values.append("NULL") elif value.isdigit(): # Handle numeric values values.append(value) else: # Handle string values with proper escaping escaped_value = value.replace("'", "''") values.append(f"'{escaped_value}'") value_str = ','.join(values) batch_rows.append(f" ({value_str})") # Write batch if reaching batch size if len(batch_rows) >= batch_size: write_batch_insert(sql_output, table_name, headers, batch_rows) batch_rows = [] # Write any remaining rows if batch_rows: write_batch_insert(sql_output, table_name, headers, batch_rows) print(f"Created {sql_file} with {row_count} rows from {csv_file}") except Exception as e: print(f"Error processing {csv_file}: {str(e)}") def write_batch_insert(file_handle, table_name, headers, value_rows): """ Write a batch INSERT statement to the SQL file. Args: file_handle: Open file handle for the SQL file table_name (str): Name of the table headers (list): Column names value_rows (list): List of formatted value strings """ columns = ','.join(headers) file_handle.write(f"INSERT INTO royalty_accounting.{table_name} ({columns})\nVALUES\n") file_handle.write(',\n'.join(value_rows)) file_handle.write("\n;\n\n") def format_sql_value(value, is_last_modified=False, is_last_modified_by=False): """Format a value for SQL INSERT statement""" if is_last_modified: return "NOW()" elif is_last_modified_by: return "'orchard-migraton-test'" elif value == "NOW()": return value elif not value: return "NULL" elif value.isdigit(): return value else: escaped_value = value.replace("'", "''") return f"'{escaped_value}'" def generate_combined_contract_terms_sql(contract_term_csv, contract_term_condition_csv, output_dir): """ Generate a combined SQL script for CONTRACT_TERM and CONTRACT_TERM_CONDITION tables with proper relationship handling, ensuring each contract term is followed by its conditions. """ combined_sql_file = os.path.join(output_dir, "COMBINED_CONTRACT_TERMS.sql") try: # Load all contract term conditions into memory, organized by the contract_term_id conditions_by_term = {} with open(contract_term_condition_csv, 'r', encoding='utf-8') as condition_file: condition_reader = csv.DictReader(condition_file) condition_headers = condition_reader.fieldnames # Find the term ID column - handle case sensitivity term_id_column = None for header in condition_headers: if header.upper() == 'CONTRACT_TERM_ID': term_id_column = header break if not term_id_column: raise ValueError("The CONTRACT_TERM_CONDITION CSV must have a 'CONTRACT_TERM_ID' column (case insensitive)") for condition in condition_reader: term_id = condition[term_id_column] if term_id not in conditions_by_term: conditions_by_term[term_id] = [] conditions_by_term[term_id].append(condition) # Generate the SQL file with open(combined_sql_file, 'w', encoding='utf-8') as sql_file: sql_file.write(f"-- Combined SQL Insert script for CONTRACT_TERM and CONTRACT_TERM_CONDITION\n") sql_file.write(f"-- Generated on {datetime.datetime.now()}\n") sql_file.write(f"-- Handles parent-child relationship with LAST_INSERT_ID()\n\n") term_count = 0 condition_count = 0 # Process contract terms one by one and immediately add their conditions with open(contract_term_csv, 'r', encoding='utf-8') as term_file: term_reader = csv.DictReader(term_file) term_headers = term_reader.fieldnames # Find the ID column for contract terms - handle case sensitivity id_column = None for header in term_headers: if header.upper() == 'CONTRACT_TERM_ID': id_column = header break if not id_column: raise ValueError("The CONTRACT_TERM CSV must have an 'ID' column (case insensitive)") # Get indexes for last_modified and last_modified_by in term headers last_modified_term_index = -1 last_modified_by_term_index = -1 for i, header in enumerate(term_headers): if header.lower() == 'last_modified': last_modified_term_index = i elif header.lower() == 'last_modified_by': last_modified_by_term_index = i # Process each contract term for term in term_reader: term_count += 1 term_id = term.get(id_column, '') # Debug output print(f"Processing contract term ID: {term_id}") # Create values list for contract term, replacing ID with NULL term_values = [] for i, header in enumerate(term_headers): if header.upper() == 'CONTRACT_TERM_ID': term_values.append('NULL') # Let database auto-generate ID elif i == last_modified_term_index: term_values.append('NOW()') elif i == last_modified_by_term_index: term_values.append("'orchard-migraton-test'") else: term_values.append(format_sql_value(term[header])) # Write the contract term INSERT sql_file.write(f"INSERT INTO royalty_accounting.CONTRACT_TERM ({','.join(term_headers)})\nVALUES ({','.join(term_values)});\n\n") # Set variable to capture last inserted ID sql_file.write("SET @contract_term_id = LAST_INSERT_ID();\n\n") # Get indexes for last_modified and last_modified_by in condition headers last_modified_cond_index = -1 last_modified_by_cond_index = -1 for i, header in enumerate(condition_headers): if header.lower() == 'last_modified': last_modified_cond_index = i elif header.lower() == 'last_modified_by': last_modified_by_cond_index = i # Add associated conditions for this term if term_id in conditions_by_term: for condition in conditions_by_term[term_id]: condition_count += 1 # Create values list for condition, replacing ID with NULL and term_id with variable condition_values = [] for i, header in enumerate(condition_headers): if header.upper() == 'CONTRACT_TERM_CONDITION_ID': condition_values.append('NULL') # Let database auto-generate ID elif header.upper() == 'CONTRACT_TERM_ID': condition_values.append('@contract_term_id') # Use the variable elif i == last_modified_cond_index: condition_values.append('NOW()') elif i == last_modified_by_cond_index: condition_values.append("'orchard-migraton-test'") else: condition_values.append(format_sql_value(condition[header])) # Write the condition INSERT sql_file.write(f"INSERT INTO royalty_accounting.CONTRACT_TERM_CONDITION ({','.join(condition_headers)})\nVALUES ({','.join(condition_values)});\n\n") else: # Add a comment if no conditions exist for this term sql_file.write(f"-- No conditions found for contract term ID: {term_id}\n\n") print(f"Created {combined_sql_file} with {term_count} contract terms and {condition_count} conditions") except Exception as e: print(f"Error generating combined contract terms SQL: {str(e)}") import traceback traceback.print_exc() def main(): # Define file names to process (excluding the ones handled specially) table_names = [ "ACCOUNT", "ACCOUNT_PAYMENT_TERM", "ACCOUNT_PAYEE", "ACCOUNT_TAX_INFO", "CONTRACT", "RUN_CONTROLLER_CONTRACT", "LEGACY_CONTRACT", "ACCOUNT_CONTRACT" ] # Get directory paths script_dir = os.path.dirname(os.path.abspath(__file__)) csv_dir = input(f"Enter path to CSV files directory [default: {script_dir}]: ").strip() or script_dir output_dir = input(f"Enter path to save SQL files [default: {script_dir}]: ").strip() or script_dir # Create output directory if it doesn't exist Path(output_dir).mkdir(parents=True, exist_ok=True) # Process regular tables for table_name in table_names: csv_file = os.path.join(csv_dir, f"{table_name}.csv") if os.path.exists(csv_file): csv_to_sql_insert(csv_file, table_name, output_dir) else: print(f"Warning: CSV file not found: {csv_file}") # Special handling for CONTRACT_TERM and CONTRACT_TERM_CONDITION contract_term_csv = os.path.join(csv_dir, "CONTRACT_TERM.csv") contract_term_condition_csv = os.path.join(csv_dir, "CONTRACT_TERM_CONDITION.csv") if os.path.exists(contract_term_csv) and os.path.exists(contract_term_condition_csv): print("Processing CONTRACT_TERM and CONTRACT_TERM_CONDITION with special handling...") generate_combined_contract_terms_sql(contract_term_csv, contract_term_condition_csv, output_dir) else: if not os.path.exists(contract_term_csv): print(f"Warning: CONTRACT_TERM CSV file not found: {contract_term_csv}") if not os.path.exists(contract_term_condition_csv): print(f"Warning: CONTRACT_TERM_CONDITION CSV file not found: {contract_term_condition_csv}") print(f"\nSQL generation complete. Files saved to: {output_dir}") if __name__ == "__main__": main()