diff --git a/lambda/textract_logs_lambda.py b/lambda/textract_logs_lambda.py new file mode 100644 index 0000000..262edd1 --- /dev/null +++ b/lambda/textract_logs_lambda.py @@ -0,0 +1,266 @@ +import json +import snowflake.connector + + +""" +Sample input event for Insert operation: +{ + "operation": "insert", + "data": { + "BATCH_ID": 12345, + "JOB_ID": "job_67890", + "STAGE": "Preprocessing", + "TEXTRACT_STATUS": "Pending", + "BUCKET_NAME": "my-bucket", + "FILE_NAME": "document.pdf", + "FILE_PATH": "/path/to/document.pdf", + "DOCUMENT_TYPE": "Type A", + "PAYER_SIGNED": False, + "PROVIDER_SIGNED": True, + "GROUP_ID": "group_123", + "CREATED_TIME": "2023-01-01 12:00:00", + "MODIFIED_TIME": "2023-01-01 12:00:00", + "CREATED_BY": "user_1", + "MODIFIED_BY": "user_1", + "ORIGINAL_FILE_EXTENSION": "pdf", + "NO_OF_PAGES": 10, + "FILE_SIZE": 204800 + } +} + + + +Sample input event for Update operation: + { + "operation": "update", + "data": { + "DOCUMENT_ID": 1001, + "TEXTRACT_STATUS": "Completed", + "MODIFIED_TIME": "2023-01-02 13:00:00", + "MODIFIED_BY": "user_2", + "NO_OF_PAGES": 12, + "FILE_SIZE": 304800 + } + } + +Operation can be: 'insert' or 'update' +Data is a dictionary with the columns and values to be inserted or updated + +""" + + + +def construct_doc_insert_sql(data): + """ + Constructs the SQL for an insert operation + + Sample return value: + INSERT INTO STG.DOCUMENT_LOGS (BATCH_ID, JOB_ID, STAGE, TEXTRACT_STATUS, BUCKET_NAME, FILE_NAME, FILE_PATH, DOCUMENT_TYPE, PAYER_SIGNED, PROVIDER_SIGNED, GROUP_ID, CREATED_TIME, MODIFIED_TIME, CREATED_BY, MODIFIED_BY, ORIGINAL_FILE_EXTENSION, NO_OF_PAGES, FILE_SIZE) + VALUES (101, 'J123456', 'Processing', 'Success', 'doc-bucket', 'file1.pdf', '/documents/2023/', 'Report', True, False, 'G100', '2024-03-21 10:00:00', '2024-03-21 10:00:00', 'admin', 'admin', 'pdf', 10, 1048576); + """ + columns = ', '.join(data.keys()) + values = ', '.join(["'" + str(value).replace("'", "''") + "'" if isinstance(value, str) else str(value) for value in data.values()]) + sql = f"INSERT INTO STG.DOCUMENT_LOGS ({columns}) VALUES ({values});" + return sql + + +def construct_doc_update_sql(data, document_id): + """ + Constructs the SQL for an update operation + + Sample return value: + UPDATE STG.DOCUMENT_LOGS SET TEXTRACT_STATUS = 'Failed', MODIFIED_TIME = '2024-03-22 15:00:00', MODIFIED_BY = 'admin' WHERE DOCUMENT_ID = 1001; + """ + set_clauses = ', '.join([f"{key} = '" + str(value).replace("'", "''") + "'" if isinstance(value, str) else f"{key} = {value}" for key, value in data.items()]) + sql = f"UPDATE STG.DOCUMENT_LOGS SET {set_clauses} WHERE DOCUMENT_ID = {document_id};" + return sql + + +def construct_batch_insert_sql(data): + """ + Constructs the SQL for an insert operation + + Sample return value: + INSERT INTO STG.BATCH_LOGS (CLIENT_ID, EXECUTION_START_TIME, NO_OF_DOCUMENTS, USER_NAME) VALUES ('C200', '2024-03-21 09:00:00', 150, 'batch_processor'); + """ + columns = ', '.join(data.keys()) + values = ', '.join(["'" + str(value).replace("'", "''") + "'" if isinstance(value, str) else str(value) for value in data.values()]) + sql = f"INSERT INTO STG.BATCH_LOGS ({columns}) VALUES ({values});" + return sql + +def construct_client_insert_sql(data): + """ + Constructs the SQL for an insert operation + + Sample return value: + INSERT INTO STG.CLIENT_LOGS (CLIENT_ID, CLIENT_NAME, BUCKET_NAME) VALUES ('CL300', 'Acme Corporation', 'acme-docs'); + """ + columns = ', '.join(data.keys()) + values = ', '.join(["'" + str(value).replace("'", "''") + "'" if isinstance(value, str) else str(value) for value in data.values()]) + sql = f"INSERT INTO STG.CLIENT_LOGS ({columns}) VALUES ({values});" + return sql + + +def construct_client_update_sql(data, client_id): + """ + Constructs the SQL for an update operation + + Sample return value: + UPDATE STG.CLIENT_LOGS SET BUCKET_NAME = 'new-acme-docs' WHERE CLIENT_ID = 'CL300'; + """ + set_clauses = ', '.join([f"{key} = '" + str(value).replace("'", "''") + "'" if isinstance(value, str) else f"{key} = {value}" for key, value in data.items()]) + sql = f"UPDATE STG.CLIENT_LOGS SET {set_clauses} WHERE CLIENT_ID = '{client_id}';" + return sql + +def construct_batch_update_sql(data, batch_id): + """ + Constructs the SQL for an update operation + + Sample return value: + UPDATE STG.BATCH_LOGS SET NO_OF_DOCUMENTS = 155, USER_NAME = 'updated_processor' WHERE BATCH_ID = 1; + """ + set_clauses = ', '.join([f"{key} = '" + str(value).replace("'", "''") + "'" if isinstance(value, str) else f"{key} = {value}" for key, value in data.items()]) + sql = f"UPDATE STG.BATCH_LOGS SET {set_clauses} WHERE BATCH_ID = {batch_id};" + return sql + + +# Main Lambda handler +def lambda_handler(event, context): + # Extract operation type and payload from event + + + operation = event['operation'] # 'insert' or 'update' + data = event['data'] + table = event['table'] + + # Get conn from Secrets Manager + + try: + # with conn.cursor() as cursor: + if table == 'DOCUMENT_LOGS': + if operation == 'insert': + sql = construct_doc_insert_sql(data) + elif operation == 'update': + document_id = data.pop('DOCUMENT_ID', None) + sql = construct_doc_update_sql(data, document_id) + + elif table == 'BATCH_LOGS': + if operation == 'insert': + sql = construct_batch_insert_sql(data) + elif operation == 'update': + batch_id = data.pop('BATCH_ID', None) + sql = construct_batch_update_sql(data, batch_id) + + elif table == 'CLIENT_LOGS': + if operation == 'insert': + sql = construct_client_insert_sql(data) + elif operation == 'update': + client_id = data.pop('CLIENT_ID', None) + sql = construct_client_update_sql(data, client_id) + else: + raise ValueError("Unsupported table.") + print(sql) + # cursor.execute(sql) + return {'statusCode': 200, 'body': json.dumps('Operation successful')} + + except Exception as e: + return {'statusCode': 400, 'body': json.dumps(str(e))} + finally: + # conn.close() + pass + + +# Sample input events +doc_input_event = { + "operation": "insert", + "table": "DOCUMENT_LOGS", + "data": { + "BATCH_ID": 101, + "JOB_ID": "J123456", + "STAGE": "Processing", + "TEXTRACT_STATUS": "Success", + "BUCKET_NAME": "doc-bucket", + "FILE_NAME": "file1.pdf", + "FILE_PATH": "/documents/2023/", + "DOCUMENT_TYPE": "Report", + "PAYER_SIGNED": True, + "PROVIDER_SIGNED": False, + "GROUP_ID": "G100", + "CREATED_TIME": "2024-03-21 10:00:00", + "MODIFIED_TIME": "2024-03-21 10:00:00", + "CREATED_BY": "admin", + "MODIFIED_BY": "admin", + "ORIGINAL_FILE_EXTENSION": "pdf", + "NO_OF_PAGES": 10, + "FILE_SIZE": 1048576 + } +} + + + +doc_update_event = { + "operation": "update", + "table": "DOCUMENT_LOGS", + "data": { + "DOCUMENT_ID": 1001, + "TEXTRACT_STATUS": "Failed", + "MODIFIED_TIME": "2024-03-22 15:00:00", + "MODIFIED_BY": "admin" + } +} + + + +batch_insert_event = { + "operation": "insert", + "table": "BATCH_LOGS", + "data": { + "CLIENT_ID": "C200", + "EXECUTION_START_TIME": "2024-03-21 09:00:00", + "NO_OF_DOCUMENTS": 150, + "USER_NAME": "batch_processor" + } +} + +batch_update_event = { + "operation": "update", + "table": "BATCH_LOGS", + "data": { + "BATCH_ID": 102, + "NO_OF_DOCUMENTS": 155, + "USER_NAME": "updated_processor" + } +} + +client_insert_event = { + "operation": "insert", + "table": "CLIENT_LOGS", + "data": { + "CLIENT_ID": "CL300", + "CLIENT_NAME": "Acme Corporation", + "BUCKET_NAME": "acme-docs" + } +} + +client_update_event = { + "operation": "update", + "table": "CLIENT_LOGS", + "data": { + "CLIENT_ID": "CL300", + "BUCKET_NAME": "new-acme-docs" + } +} + + +lambda_handler(doc_input_event, None) + +lambda_handler(doc_update_event, None) + +lambda_handler(batch_insert_event, None) + +lambda_handler(batch_update_event, None) + +lambda_handler(client_insert_event, None) + +lambda_handler(client_update_event, None) + diff --git a/on_demand_scripts/training_data_pre_processing.py b/on_demand_scripts/training_data_pre_processing.py new file mode 100644 index 0000000..957f4fb --- /dev/null +++ b/on_demand_scripts/training_data_pre_processing.py @@ -0,0 +1,111 @@ +import pandas as pd +from datetime import datetime + + +def export_column_config(column_names: list): + """ + This function exports the column names and datatypes to a csv file + This will then be ingested to the training data column config + """ + # Create a data frame from the 2 lists and export as csv with the current date and time as filename + + # Create a datatypes list that is all VARCHAR strings equal to the length of the column_names list + try: + column_datatypes = ["VARCHAR" for i in range(len(column_names))] + df = pd.DataFrame(list(zip(column_names, column_datatypes)), columns=["Column_Name", "Data_Type"]) + date = datetime.now() + timestamp = str(date.strftime("%m%d%Y_%H%M%S")) + df.to_csv(f"column_config_{timestamp}.csv", index=False) + return "Column config created successfully" + except Exception as e: + return str(e) + + +def process_xls(file_name: str): + """ + This function processes the master_doczy_db.xlsx file and creates a csv file with the processed data + This will then be ingested to the training data raw table + """ + try: + xl_df = pd.read_excel(file_name, sheet_name="Data Base", header=4) # Passing header as 4 to use sf_col as header + datatypes = xl_df.iloc[0].values.tolist() # grab the datatypes + xl_df2 = xl_df[26:] # Trim the df to remove the first 26 rows where the data is not useful + xl_df2 = xl_df2.reset_index(drop=True) + xl_df2.columns.values[7] = "DOCUMENT_NAME" # works + xl_df2 = xl_df2.iloc[:, 7:] # Drop columns before DOCUMENT_NAME + + start_idx = xl_df2.columns.get_loc('DOCUMENT_NAME') + 1 # +1 because we don't want to drop 'DOCUMENT_NAME' + + # Get index of 'CONTRACT_TITLE' column + end_idx = xl_df2.columns.get_loc('CONTRACT_TITLE') + + # Create a list of column names to drop, which are between 'DOCUMENT_NAME' and 'CONTRACT_TITLE' + cols_to_drop = xl_df2.columns[start_idx:end_idx] + + # Drop the columns + xl_df2.drop(columns=cols_to_drop, inplace=True) + + xl_df2.dropna(axis=1, how='all') + date = datetime.now() + timestamp = str(date.strftime("%m%d%Y_%H%M%S")) + + xl_df2 = xl_df2.loc[:, ~xl_df2.columns.str.startswith('Unnamed')] # Dropping any unnamed columns (Question cols without SF_COL_NAME) + + date = datetime.now() + timestamp = str(date.strftime("%m%d%Y_%H%M%S")) + + xl_df2.columns = xl_df2.columns.str.replace('.', '_', regex=False) # Replace '.' with '_' in column names so that snowflake can ingest + + xl_df2.to_csv(f"processed_training_data-{timestamp}.csv", index=False) + print("Processed training data created successfully") + return xl_df2 + + except Exception as e: + return str(e) + + +def create_business_config_table(file_name: str): + """ + This function creates a business config table from the Business excel file + Where we extact the sf_columns, interrogation question, priority, group_no and theme + """ + try: + xl_df = pd.read_excel(file_name, sheet_name="Data Base", header=2) # Passing header as 4 to use sf_col as header + xl_df = xl_df.iloc[:,12:] # Drop columns before DOCUMENT_NAME + + questions = xl_df.columns.tolist() # Grab the questions that are in the header row + sf_cols = xl_df.iloc[1].tolist() # Grab the sf_cols + priority = xl_df.iloc[3].tolist() # Grab the priority + group_no = xl_df.iloc[4].tolist() # Grab the group_no + theme = xl_df.iloc[5].tolist() # Grab the theme + + # Create a dataframe from the lists + df_internal = pd.DataFrame({'Column_name': sf_cols, 'Question': questions, 'priority': priority, 'group_no': group_no, 'theme': theme}) + + # Drop rows where the question is 'Unnamed' and the column_name is NaN (Pandas automatically fills NaN with 'Unnamed' when reading excel files depending on the formatting) + df_cleaned = df_internal[~df_internal['Question'].str.contains('Unnamed', na=False) & ~df_internal['Column_name'].isna()] + + date = datetime.now() + timestamp = str(date.strftime("%m%d%Y_%H%M%S")) + + df_cleaned.to_csv(f'biz_config-{timestamp}.csv', index=False) + except Exception as e: + return str(e) + + + +def main(): + """ + For this script to work, ensure all rows in the excel file are expanded between the header and actual values. + We need the sf_column name, group no, priority etc to be accessible for ingestion + """ + file_name = "master_doczy_db.xlsx" + xl_df = process_xls(file_name) + status = export_column_config(xl_df.columns.tolist()) + print(status) + status = create_business_config_table(file_name) + print(status) + + +if __name__ == "__main__": + main() diff --git a/snowflake/scripts/training_interface/R__001_INIT_OBJS.sql b/snowflake/scripts/training_interface/R__001_INIT_OBJS.sql index f2d151f..992ece8 100644 --- a/snowflake/scripts/training_interface/R__001_INIT_OBJS.sql +++ b/snowflake/scripts/training_interface/R__001_INIT_OBJS.sql @@ -49,3 +49,18 @@ CREATE TABLE IF NOT EXISTS STG.TRAINING_DATA_COLUMN_CONFIG( COLUMN_NAME VARCHAR, COLUMN_DATATYPE VARCHAR ); + +-- Creating a separate file format for the training data +CREATE FILE FORMAT IF NOT EXISTS STG.TRAINING_DATA_FILE_FORMAT + TYPE = 'CSV' + FIELD_DELIMITER = ',' + FILE_EXTENSION = '.csv' + RECORD_DELIMITER = '\\n' + DATE_FORMAT = AUTO + TRIM_SPACE = TRUE + NULL_IF = ('NULL', '', 'N/A','?','~','\\N') + SKIP_HEADER = 1 + EMPTY_FIELD_AS_NULL = TRUE + FIELD_OPTIONALLY_ENCLOSED_BY = '"' + error_on_column_count_mismatch=false + SKIP_BLANK_LINES = TRUE; \ No newline at end of file diff --git a/snowflake/scripts/training_interface/R__004_CREATE_TRAINING_DATA_TABLE_SP.sql b/snowflake/scripts/training_interface/R__004_CREATE_TRAINING_DATA_TABLE_SP.sql index 6a49bc9..eeb3dbc 100644 --- a/snowflake/scripts/training_interface/R__004_CREATE_TRAINING_DATA_TABLE_SP.sql +++ b/snowflake/scripts/training_interface/R__004_CREATE_TRAINING_DATA_TABLE_SP.sql @@ -10,10 +10,10 @@ DECLARE cur_config cursor FOR SELECT column_name, column_datatype FROM STG.TRAINING_DATA_COLUMN_CONFIG; BEGIN - + -- Creating the raw training data table with all columns as VARCHAR due to the dynamic nature of the columns and fields OPEN cur_config; FOR rec IN cur_config DO - dynamic_ddl := dynamic_ddl || rec.column_name || '' '' || rec.column_datatype || '',''; + dynamic_ddl := dynamic_ddl || rec.column_name || '' '' || ''VARCHAR'' || '',''; END FOR; dynamic_ddl := LEFT(dynamic_ddl, LENGTH(dynamic_ddl) - 1); diff --git a/snowflake/scripts/training_interface/R__006_LOAD_TRAINING_DATA_RAW_SP.sql b/snowflake/scripts/training_interface/R__006_LOAD_TRAINING_DATA_RAW_SP.sql index 9f5bd06..d4ea76a 100644 --- a/snowflake/scripts/training_interface/R__006_LOAD_TRAINING_DATA_RAW_SP.sql +++ b/snowflake/scripts/training_interface/R__006_LOAD_TRAINING_DATA_RAW_SP.sql @@ -22,8 +22,8 @@ BEGIN call stg.log_audit(:procedure_name, 'Section 2', 99, 'START'); - -- Truncate table as we are using the KILL & FILL approach - TRUNCATE TABLE STG.TRAINING_DATA_RAW; + -- Recreating the training data table by calling the SP. This will replace the existing table with new column definitions + call STG.CREATE_TRAINING_DATA_TABLE() call stg.log_audit(:procedure_name, 'Section 2', 99, 'END'); @@ -31,7 +31,7 @@ BEGIN -- Copy command to load data COPY INTO STG.TRAINING_DATA_RAW FROM @STG.RAW_TRAINING_DATA_STAGE - FILE_FORMAT = (FORMAT_NAME = 'STG.CSV_HEADER') + FILE_FORMAT = (FORMAT_NAME = 'STG.TRAINING_DATA_FILE_FORMAT') ON_ERROR = ABORT_STATEMENT; call stg.log_audit(:procedure_name, 'Section 3', 99, 'END');