import snowflake.connector import os from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.hazmat.primitives.asymmetric import dsa from cryptography.hazmat.primitives import serialization import csv import datetime from pathlib import Path import json import codecs import re import pandas as pd import ast # Increase CSV field size limit to handle large fields import sys csv.field_size_limit(sys.maxsize) FILE_ORDER = { "account": "01", "account_payment_term": "02", "account_payee": "03", "account_tax_info": "04", "contract": "05", "run_controller_contract": "06", "legacy_contract": "07", "account_contract": "08", "combined_contract_terms": "09" } def load_column_config(config_file_path): """ Load column configuration from a JSON or text file. Args: config_file_path (str): Path to the configuration file Returns: dict: Dictionary with table names as keys and list of columns as values Expected JSON format: { "account": ["account_id", "account_name", "created_at", "created_by", "last_modified", "last_modified_by", "sap_created_at"], "contract": ["contract_id", "contract_name", "status", "created_at"], ... } Expected text format (alternative): [TABLE_NAME] column1 column2 column3 [ANOTHER_TABLE] column1 column2 """ if not os.path.exists(config_file_path): print(f"Warning: Config file not found: {config_file_path}") return {} try: # Try to load as JSON first with open(config_file_path, 'r', encoding='utf-8') as f: if config_file_path.lower().endswith('.json'): config = json.load(f) # Convert all keys to lowercase for consistency return {k.lower(): v for k, v in config.items()} else: # Parse as text format content = f.read() config = {} current_table = None for line in content.strip().split('\n'): line = line.strip() if not line or line.startswith('#'): # Skip empty lines and comments continue if line.startswith('[') and line.endswith(']'): # Table name current_table = line[1:-1].lower() config[current_table] = [] elif current_table: # Column name config[current_table].append(line) return config except json.JSONDecodeError: print(f"Error: Invalid JSON format in config file: {config_file_path}") return {} except Exception as e: print(f"Error loading config file {config_file_path}: {str(e)}") return {} def get_table_columns(table_name, column_config): """ Get the list of columns for a specific table from the configuration. Args: table_name (str): Name of the table column_config (dict): Column configuration dictionary Returns: list: List of column names, or ["*"] if not configured """ table_key = table_name.lower() if table_key in column_config: return column_config[table_key] else: print(f"Warning: No column configuration found for table '{table_name}', using SELECT *") return ["*"] def get_filter_query(table_name): """Generate the appropriate filter query based on table name""" base_filter = """ select distinct account_id from orchard_app_reporting_v2.uat_royalty_accounting_royalty_accounting.contract c join orchard_app_reporting_v2.uat_royalty_accounting_royalty_accounting.run_controller_contract rcc on c.contract_id = rcc.contract_id join orchard_app_reporting_v2.uat_royalty_accounting_royalty_accounting.account_contract ac on ac.contract_id = c.contract_id where run_controller_id = 1 and account_id not in (select distinct account_id from orchard_app_reporting_v2.prod_royalty_accounting_royalty_accounting.account) and account_id not in (select distinct account_id from orchard_app_reporting_v2.qa_royalty_accounting_royalty_accounting.account) """ contract_filter = """ select distinct c.contract_id from orchard_app_reporting_v2.uat_royalty_accounting_royalty_accounting.contract c join orchard_app_reporting_v2.uat_royalty_accounting_royalty_accounting.run_controller_contract rcc on c.contract_id = rcc.contract_id where run_controller_id = 1 and c.contract_id not in (select distinct contract_id from orchard_app_reporting_v2.prod_royalty_accounting_royalty_accounting.contract) and c.contract_id not in (select distinct contract_id from orchard_app_reporting_v2.qa_royalty_accounting_royalty_accounting.contract) """ contract_term_condition_filter = """ select distinct c.contract_id from orchard_app_reporting_v2.uat_royalty_accounting_royalty_accounting.contract c join orchard_app_reporting_v2.uat_royalty_accounting_royalty_accounting.run_controller_contract rcc on c.contract_id = rcc.contract_id where run_controller_id = 1 and c.contract_id not in (select distinct contract_id from orchard_app_reporting_v2.prod_royalty_accounting_royalty_accounting.contract) and c.contract_id not in (select distinct contract_id from orchard_app_reporting_v2.qa_royalty_accounting_royalty_accounting.contract) """ # Tables that use account_id filter account_tables = ["account", "account_payment_term", "account_payee", "account_tax_info", "account_contract"] # Tables that use contract_id filter contract_tables = ["contract", "run_controller_contract", "legacy_contract"] if table_name.lower() in account_tables: return base_filter elif table_name.lower() in contract_tables: return contract_filter else: # Default to account filter for unknown tables return base_filter def snowflake_to_sql_insert(cs, table_name, output_dir, column_config): """ Read data from Snowflake table and convert to SQL INSERT statements. Args: cs: Snowflake cursor object table_name (str): Name of the table to read from output_dir (str): Directory to save the SQL file column_config (dict): Dictionary containing column configurations for each table """ table_name_lower = table_name.lower() # Generate output table name based on file order order_num = FILE_ORDER.get(table_name_lower, "00") output_table = f"20-{order_num}-migration-contracts_{table_name_lower}_qa" sql_file = os.path.join(output_dir, f"{output_table}.sql") try: # Get filter subquery filter_query = get_filter_query(table_name_lower) # Get configured columns for this table table_columns = get_table_columns(table_name, column_config) # Determine filter column based on table type account_tables = ["account", "account_payment_term", "account_payee", "account_tax_info", "account_contract"] if table_name_lower in account_tables: filter_column = "account_id" else: filter_column = "contract_id" # Build column list for SELECT if table_columns == ["*"]: column_list = "*" else: column_list = ", ".join(table_columns) # Build the main query main_query = f""" SELECT DISTINCT {column_list} FROM orchard_app_reporting_v2.uat_royalty_accounting_royalty_accounting.{table_name.upper()} WHERE {filter_column} IN ({filter_query}) """ print(f"Executing query for {table_name}: {main_query}") # Execute the query cs.execute(main_query) # Get column names from cursor description headers = [desc[0].lower() for desc in cs.description] # Fetch all results rows = cs.fetchall() if not rows: print(f"No data found for table {table_name}") return with open(sql_file, 'w', encoding='utf-8') as sql_output: # Write SQL file header sql_output.write(f"-- SQL Insert script for {output_table}\n") sql_output.write(f"-- Generated on {datetime.datetime.now()}\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_lower in id_columns_to_nullify: for id_col in id_columns_to_nullify[table_name_lower]: 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 rows: # 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 NULL values elif value is None: values.append("NULL") # Handle numeric values elif isinstance(value, (int, float)): values.append(str(value)) # Handle datetime values elif isinstance(value, datetime.datetime): values.append(f"'{value.strftime('%Y-%m-%d %H:%M:%S')}'") # Handle date values elif isinstance(value, datetime.date): values.append(f"'{value.strftime('%Y-%m-%d')}'") # Handle string values with proper escaping else: str_value = str(value).strip() escaped_value = str_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_lower, headers, batch_rows) batch_rows = [] # Write any remaining rows if batch_rows: write_batch_insert(sql_output, table_name_lower, headers, batch_rows) print(f"Created {sql_file} with {row_count} rows from Snowflake table {table_name}") except Exception as e: print(f"Error processing Snowflake table {table_name}: {str(e)}") import traceback traceback.print_exc() 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 value is None: return "NULL" elif isinstance(value, (int, float)): return str(value) elif isinstance(value, datetime.datetime): return f"'{value.strftime('%Y-%m-%d %H:%M:%S')}'" elif isinstance(value, datetime.date): return f"'{value.strftime('%Y-%m-%d')}'" else: if isinstance(value, str): # Handle JSON strings (conditions column) if value.strip().startswith('{') and value.strip().endswith('}'): try: # First, try to clean up double-escaped quotes clean_value = value.strip() # If the string has double quotes escaped as "", unescape them if '""' in clean_value: clean_value = clean_value.replace('""', '"') # Parse and re-serialize to ensure proper JSON formatting parsed = json.loads(clean_value) # Keep proper JSON format with double quotes and wrap in single quotes json_str = json.dumps(parsed, ensure_ascii=False) return f"'{json_str}'" except json.JSONDecodeError: pass # Not valid JSON, continue with normal string handling # Handle regular strings if value.startswith('"') and value.endswith('"'): value = value[1:-1] elif value.startswith('\'') and value.endswith('\''): value = value[1:-1] value = str(value).strip() escaped_value = value.replace("'", "''") return f"'{escaped_value}'" return f"'{str(value).strip()}'" def generate_combined_contract_terms_sql(contract_term_csv, contract_term_condition_csv, contract_term_condition_csv_delimited, contract_term_csv_delimited, 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. """ order_num = FILE_ORDER.get("combined_contract_terms", "09") combined_sql_file = os.path.join(output_dir, f"20-{order_num}-migration-contracts_combined_contract_terms_qa.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') as f: content = f.read() processed_content = '' i = 0 while i < len(content): if content[i] in '{[': # Find the matching closing bracket start_char = content[i] end_char = '}' if start_char == '{' else ']' bracket_count = 1 json_start = i i += 1 while i < len(content) and bracket_count > 0: if content[i] == start_char: bracket_count += 1 elif content[i] == end_char: bracket_count -= 1 i += 1 # Keep the JSON part intact processed_content += content[json_start:i] if i < len(content) and content[i] == ',': processed_content += '|' # or ',' if that's your preferred delimiter # i += 1 elif content[i] == ',': processed_content += '|' else: processed_content += content[i] i += 1 processed_content = processed_content.replace('\\,', ',') # Write processed content with open(contract_term_condition_csv_delimited, 'w') as f2: f2.write(processed_content) # Read with DictReader using the processed file with open(contract_term_condition_csv_delimited, 'r') as condition_file: condition_reader = csv.DictReader( condition_file, delimiter='|', quoting=csv.QUOTE_NONE, escapechar=None, doublequote=False ) condition_headers = condition_reader.fieldnames # Find the term ID column - handle case sensitivity print(condition_headers) 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 with open(contract_term_csv, 'r') as f: print("CSV Opened") with open(contract_term_csv_delimited, 'w') as f2: print("CSV Delimited Opened") for cnt, line in enumerate(f): line = line.replace('"\\', '"') if cnt == 0: line = line.replace(',', '|') f2.write(line) else: processed_line = '' i = 0 in_quotes = False in_special = False # for JSON/arrays while i < len(line): char = line[i] # Handle quote state if char == '"' and (i == 0 or line[i-1] != '\\'): in_quotes = not in_quotes # Handle JSON/array state elif char in '{[' and not in_quotes: in_special = True elif char in ']}' and not in_quotes: in_special = False # Handle comma replacement if char == ',' and not in_quotes and not in_special: processed_line += '|' else: processed_line += char i += 1 f2.write(processed_line) # Read the processed file with open(contract_term_csv_delimited, 'r', encoding='utf-8') as term_file: term_reader = csv.DictReader( term_file, delimiter='|', quoting=csv.QUOTE_MINIMAL, escapechar=None, doublequote=False, skipinitialspace=False ) 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'") elif header.upper() == 'CONDITIONS': # Handle conditions column directly try: # Clean the input string first - remove outer quotes if present clean_value = condition[header].strip() if clean_value.startswith('"') and clean_value.endswith('"'): clean_value = clean_value[1:-1] # Handle double-escaped quotes if '""' in clean_value: clean_value = clean_value.replace('""', '"') # Parse and re-serialize JSON to ensure proper formatting json_value = json.loads(clean_value) # Keep proper JSON format with double quotes json_str = json.dumps(json_value, ensure_ascii=False) # Just wrap in single quotes condition_values.append(f"'{json_str}'") except json.JSONDecodeError as e: # If not valid JSON, handle as regular string condition_values.append(format_sql_value(condition[header])) 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 clean_json_string(val): if isinstance(val, str) and val.startswith('['): # Remove newlines, extra spaces, and double double-quotes val = re.sub(r'\s+', '', val) # remove all whitespace val = val.replace('""', '"') # fix double quotes if isinstance(val, str) and val.startswith('["') and val.endswith('"]'): try: # Convert string to list items = ast.literal_eval(val) if isinstance(items, list): return f"[{', '.join(items)}]" except Exception: pass # In case of malformed strings, return as is return val def clean_json_output(val): """Clean and validate JSON string""" if not isinstance(val, str): return val # Remove extra whitespace val = val.strip() # Try to parse and reformat JSON to ensure it's valid try: if val.startswith('{') or val.startswith('['): parsed = json.loads(val) return json.dumps(parsed, ensure_ascii=False, separators=(',', ':')) except json.JSONDecodeError: pass return val def normalize_json(val): if isinstance(val, str) and val.strip().startswith('{'): try: # Parse JSON string -> dict -> one-line JSON string parsed = json.loads(val) return json.dumps(parsed, ensure_ascii=False, indent=None) # no escaping, no extra indent except Exception as e: return val # leave as-is if not valid JSON return val def generate_combined_contract_terms_sql_from_snowflake(cs, output_dir, column_config): """ Generate a combined SQL script for CONTRACT_TERM and CONTRACT_TERM_CONDITION tables reading data from Snowflake with proper relationship handling. Args: cs: Snowflake cursor object output_dir (str): Directory to save the SQL file column_config (dict): Dictionary containing column configurations for each table """ order_num = FILE_ORDER.get("combined_contract_terms", "09") combined_sql_file = os.path.join(output_dir, f"20-{order_num}-migration-contracts_combined_contract_terms_qa.sql") try: # Get filter for contract_id contract_filter = get_filter_query("contract") contract_term_condition_filter = get_filter_query("contract") # Get configured columns for contract_term and contract_term_condition contract_term_columns = get_table_columns("contract_term", column_config) contract_condition_columns = get_table_columns("contract_term_condition", column_config) # Build column lists if contract_term_columns == ["*"]: term_column_list = "*" else: term_column_list = ", ".join(contract_term_columns) if contract_condition_columns == ["*"]: condition_column_list = "ctc.*" else: condition_column_list = ", ".join([f"ctc.{col}" for col in contract_condition_columns]) # Query for contract terms contract_terms_query = f""" SELECT {term_column_list} FROM orchard_app_reporting_v2.uat_royalty_accounting_royalty_accounting.CONTRACT_TERM WHERE contract_id IN ({contract_filter}) """ # Query for contract term conditions contract_conditions_query = f""" SELECT {condition_column_list} FROM orchard_app_reporting_v2.uat_royalty_accounting_royalty_accounting.CONTRACT_TERM_CONDITION ctc JOIN orchard_app_reporting_v2.uat_royalty_accounting_royalty_accounting.CONTRACT_TERM ct ON ctc.contract_term_id = ct.contract_term_id WHERE ct.contract_id IN ({contract_term_condition_filter}) """ print("Fetching contract terms from Snowflake...") cs.execute(contract_terms_query) contract_terms = cs.fetchall() term_headers = [desc[0].lower() for desc in cs.description] # columns = [col[0] for col in cs.description] # Convert to DataFrame df_terms = pd.DataFrame(contract_terms, columns=term_headers) print("Fetching contract term conditions from Snowflake...") cs.execute(contract_conditions_query) contract_conditions = cs.fetchall() condition_headers = [desc[0].lower() for desc in cs.description] df_cond = pd.DataFrame(contract_conditions, columns=condition_headers) # --- Get directories --- 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 # --- Build file paths --- contract_term_csv = os.path.join(csv_dir, "contract_term.csv") contract_term_condition_csv = os.path.join(csv_dir, "contract_term_condition.csv") contract_term_condition_csv_delimited = os.path.join(csv_dir, "contract_term_condition_csv_delimited.csv") contract_term_csv_delimited = os.path.join(csv_dir, "contract_term_csv_delimited.csv") # --- Save CSVs --- # Replace newlines in all string columns # for col in df_terms.select_dtypes(include='object').columns: # df_terms[col] = df_terms[col].astype(str).str.replace(r'[\r\n]+', ' ', regex=True) # Clean multiline strings first for col in df_terms.select_dtypes(include='object').columns: df_terms[col] = df_terms[col].astype(str).str.replace(r'[\r\n]+', ' ', regex=True) # Write CSV in two steps: with open(contract_term_csv, 'w', newline='', encoding='utf-8') as f: # Step 1: write header manually (unquoted) writer = csv.writer(f) writer.writerow(df_terms.columns) # for col in df_terms.columns: # df_terms[col] = df_terms[col].apply(clean_json_string) for col in df_terms.select_dtypes(include='object').columns: df_terms[col] = df_terms[col].astype(str).str.replace(r'\s+', ' ', regex=True) # flatten multiline strings df_terms[col] = df_terms[col].str.replace('""', '"') # Step 2: write data using pandas (with quoting) df_terms.to_csv( f, index=False, header=False, # <- Do not write header again quoting=csv.QUOTE_NONE, lineterminator='\n', na_rep='NULL', quotechar='"', escapechar='\\', doublequote=True ) # df_terms.to_csv(contract_term_csv, index=False, na_rep='NULL', quoting=1, quotechar='"', lineterminator='\n') with open(contract_term_condition_csv, 'w', newline='', encoding='utf-8') as f: # Step 1: write header manually (unquoted) writer = csv.writer(f) writer.writerow(df_cond.columns) for col in df_cond.select_dtypes(include='object').columns: df_cond[col] = df_cond[col].astype(str).str.replace(r'\s+', ' ', regex=True) # flatten multiline strings # Export to CSV with proper JSON handling df_cond.to_csv( f, index=False, header=False, # <- Do not write header again quoting=csv.QUOTE_NONE, lineterminator='\n', na_rep='NULL', quotechar='"', escapechar='\\', doublequote=True ) # --- Ensure delimited CSVs exist (even if empty for now) --- for file_path in [contract_term_condition_csv_delimited, contract_term_csv_delimited]: if not os.path.exists(file_path): with open(file_path, 'w', encoding='utf-8') as f: pass # --- Call the SQL generation function --- 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, contract_term_condition_csv_delimited, contract_term_csv_delimited, 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}") except Exception as e: print(f"Error generating combined contract terms SQL from Snowflake: {str(e)}") import traceback traceback.print_exc() def main(): """ Main function to process Snowflake tables and generate SQL insert scripts. Requires a Snowflake cursor object 'cs' to be passed or available in scope. """ # Note: Assuming 'cs' (Snowflake cursor) is already created and available # cs = ctx.cursor() # This should be done before calling this script # Check if cursor is available with open("/Users/[user_name in laptop]/.ssh/snowflake/rsa_key.p8", "rb") as key: p_key= serialization.load_pem_private_key( key.read(), password=os.environ['PRIVATE_KEY_PASSPHRASE'].encode(), backend=default_backend() ) pkb = p_key.private_bytes( encoding=serialization.Encoding.DER, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption()) ctx = snowflake.connector.connect( account = "SME-ORCHARD", user = "{user_name}@sonymusic-pde.com", role = "DEV_ENGINEERING", warehouse= "DEV_PERFORMANCE_WAREHOUSE", private_key=pkb ) cs = ctx.cursor() try: cs except NameError: print("Error: Snowflake cursor 'cs' is not available. Please create it first with: cs = ctx.cursor()") return # Get configuration file path script_dir = os.path.dirname(os.path.abspath(__file__)) if __file__ else os.getcwd() default_config_path = os.path.join(script_dir, "table_columns_config.json") config_file_path = input(f"Enter path to column configuration file [default: {default_config_path}]: ").strip() or default_config_path # Load column configuration column_config = load_column_config(config_file_path) if not column_config: print("Warning: No column configuration loaded. Will use SELECT * for all tables.") column_config = {} # Define table 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 path for output 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: try: snowflake_to_sql_insert(cs, table_name, output_dir, column_config) except Exception as e: print(f"Error processing table {table_name}: {str(e)}") # Special handling for CONTRACT_TERM and CONTRACT_TERM_CONDITION try: print("Processing CONTRACT_TERM and CONTRACT_TERM_CONDITION with special handling...") generate_combined_contract_terms_sql_from_snowflake(cs, output_dir, column_config) except Exception as e: print(f"Error processing combined contract terms: {str(e)}") print(f"\nSQL generation complete. Files saved to: {output_dir}") def main_with_cursor(cs, config_file_path=None): """ Alternative main function that accepts the cursor as a parameter. Args: cs: Snowflake cursor object config_file_path (str, optional): Path to column configuration file """ # Get configuration file path if not config_file_path: script_dir = os.path.dirname(os.path.abspath(__file__)) if __file__ else os.getcwd() default_config_path = os.path.join(script_dir, "table_columns_config.json") config_file_path = input(f"Enter path to column configuration file [default: {default_config_path}]: ").strip() or default_config_path # Load column configuration column_config = load_column_config(config_file_path) if not column_config: print("Warning: No column configuration loaded. Will use SELECT * for all tables.") column_config = {} # Define table 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 path for output script_dir = os.path.dirname(os.path.abspath(__file__)) if __file__ else os.getcwd() 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: try: snowflake_to_sql_insert(cs, table_name, output_dir, column_config) except Exception as e: print(f"Error processing table {table_name}: {str(e)}") # Special handling for CONTRACT_TERM and CONTRACT_TERM_CONDITION try: print("Processing CONTRACT_TERM and CONTRACT_TERM_CONDITION with special handling...") generate_combined_contract_terms_sql_from_snowflake(cs, output_dir, column_config) except Exception as e: print(f"Error processing combined contract terms: {str(e)}") print(f"\nSQL generation complete. Files saved to: {output_dir}") if __name__ == "__main__": main() # To use this script with your cursor: # main_with_cursor(cs)