Spaces:
Build error
Build error
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) | |