Added training data new script, 1st draft of logging lambda and some updates to SF objs
This commit is contained in:
Umang Mistry
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
@@ -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;
|
||||
@@ -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);
|
||||
|
||||
@@ -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');
|
||||
|
||||
Reference in New Issue
Block a user