underwriting-workflow / utils /tabs /document_validation_tab.py
vamsidharmuthireddy's picture
Upload 90 files
52c1998 verified
import streamlit as st
from utils.prep_validators_payload import process_extracted_data
import pandas as pd
from utils.logger import setup_logger
from utils.json_utils import DocumentTypeByCategory, extract_document_types_from_transformed
logger = setup_logger(__name__)
def validate_documents(current):
st.header("Validation Results")
# if current is not None:
# if st.session_state['uploads'][current]['application_form'] is None:
# custom_app_form = {
# "application_summary_full_name": "Jodie Pippa",
# "application_summary_bank_name": "HSBC",
# "application_summary_employer_name": "ABC Ltd",
# "application_summary_complete_address": "123 Maple Street, London, UK, SW1A 1AA",
# "full_name_err_msgs": None,
# "bank_name_err_msgs": None,
# "employer_name_err_msgs": None,
# "complete_employee_address_err_msgs": None,
# "is_incomplete": False,
# }
# else:
# custom_app_form = st.session_state['uploads'][current]['application_form'].set_index(0)[1].to_dict()
# custom_app_form = CustomAppFormUpload.model_validate(
# uploaded_custom_form_dict).model_dump()
current_upload = st.session_state.get('current_upload')
if current_upload:
if st.session_state['uploads'][current]['application_form'] is None:
custom_app_form = {
"application_summary_full_name": "Jodie Pippa",
"application_summary_bank_name": "HSBC",
"application_summary_employer_name": "ABC Ltd",
"application_summary_complete_address": "123 Maple Street, London, UK, SW1A 1AA",
"full_name_err_msgs": None,
"bank_name_err_msgs": None,
"employer_name_err_msgs": None,
"complete_employee_address_err_msgs": None,
"is_incomplete": False,
}
else:
custom_app_form = st.session_state['uploads'][current]['application_form'].set_index(0)[1].to_dict()
current_data = st.session_state['uploads'][current_upload]
full_data = current_data['values_raw']
full_data_transformed = current_data['results_transformed']
logger.info(f"full_data : {full_data}")
if full_data_transformed is not None:
logger.info(f"full_data_transformed : {full_data_transformed}")
full_data_transformed_categories = extract_document_types_from_transformed(full_data_transformed)
global_validations = DocumentTypeByCategory.model_validate(full_data_transformed_categories).to_dataframe()
st.markdown("## Document Type Upload validations")
st.table(global_validations)
st.markdown("---")
if full_data is not None:
validations, cross_docs_name_eq_check = process_extracted_data(full_data, custom_app_form, full_data_transformed)
# st.markdown("## Document validations")
# document_types_not_uploaded = set()
# for key in validations:
# if not validations[key]:
# document_types_not_uploaded.add(key)
# if document_types_not_uploaded:
# document_types_not_uploaded_list = list(document_types_not_uploaded)
# document_types_not_uploaded_list.sort()
# doc_types_not_uploaded_pts = "\n".join(
# [f"- {item}" for item in document_types_not_uploaded_list]
# )
# st.markdown(doc_types_not_uploaded_pts)
# else:
# st.markdown("*All document types uploaded*")
st.markdown("## Document validations")
for k, v in validations.items():
st.markdown(f"Validations: {k}")
if len(v) > 0:
df = v[0]['validation_policy_status_df']
df.index += 1
logger.info(f"df: {df}")
df['Status'] = df['Status'].apply(lambda x: 'βœ…' if x else '❌')
# df = pd.DataFrame(v).T
# st.dataframe(df, use_container_width=True)
for col in df.select_dtypes(include='object'):
df[col] = df[col].astype(str)
st.table(df)
# st.markdown(f"**RedFlag**: {v[0]['is_red_flagged']}")
st.markdown("---")
st.markdown("---")
st.markdown("## General Guidance")
cross_docs_name_eq_check_df = pd.DataFrame([cross_docs_name_eq_check])
cross_docs_name_eq_check_df['Status'] = cross_docs_name_eq_check_df['Status'].apply(lambda x: 'βœ…' if x else '❌')
cross_docs_name_eq_check_df.index += 1
st.table(cross_docs_name_eq_check_df)