545 lines
22 KiB
Python
545 lines
22 KiB
Python
|
import os
|
||
|
|
import pandas as pd
|
||
|
|
import boto3
|
||
|
|
from botocore.exceptions import ClientError
|
||
|
|
import re
|
||
|
|
from datetime import datetime
|
||
|
|
from valid import STATE_MAP, AC_IND_LANG_COLS, B_IND_LANG_COLS
|
||
|
|
import utils
|
||
|
|
import preprocessing_funcs
|
||
|
|
import config
|
||
|
|
from openpyxl import Workbook
|
||
|
|
from openpyxl.utils.dataframe import dataframe_to_rows
|
||
|
|
|
||
|
|
def detect_date_anomalies(series, field_name, is_merged=False):
|
||
|
|
total_count = len(series)
|
||
|
|
unique_values = series.fillna("").nunique()
|
||
|
|
|
||
|
|
date_fields = [
|
||
|
|
"CONTRACT_BASE_EFFECTIVE_DT",
|
||
|
|
"CONTRACT_EFFECTIVE_DT",
|
||
|
|
"CONTRACT_EFFECTIVE_DT_SIGNATORY",
|
||
|
|
"CONTRACT_TERMINATION_DT",
|
||
|
|
"CONTRACT_SIGNATORY_DT",
|
||
|
|
]
|
||
|
|
date_pattern = r"(\d{1,2}[/-]\d{1,2}[/-]\d{2,4}|\d{2,4}[/-]\d{1,2}[/-]\d{1,2})"
|
||
|
|
|
||
|
|
date_like_count = 0
|
||
|
|
|
||
|
|
if pd.api.types.is_string_dtype(series) or pd.api.types.is_object_dtype(series):
|
||
|
|
series = series.astype(str)
|
||
|
|
if field_name in date_fields:
|
||
|
|
date_like_count = series.str.count(date_pattern).sum()
|
||
|
|
else:
|
||
|
|
date_like_count = series.str.count(r"\d{1,2}/\d{1,2}/\d{2,4}").sum()
|
||
|
|
|
||
|
|
data_type = "float64" if pd.api.types.is_float_dtype(series) else "string"
|
||
|
|
|
||
|
|
dataset = "Merged DataFrame" if is_merged else "Original DataFrame"
|
||
|
|
result = {
|
||
|
|
f"Total entries ({dataset})": total_count,
|
||
|
|
f"Unique values ({dataset})": unique_values,
|
||
|
|
f"Date-like entries ({dataset})": date_like_count,
|
||
|
|
}
|
||
|
|
|
||
|
|
if is_merged:
|
||
|
|
result[f"Data type ({dataset})"] = data_type
|
||
|
|
|
||
|
|
return result
|
||
|
|
|
||
|
|
def tin_check(tin: str) -> tuple[str, bool]:
|
||
|
|
pattern = r'\b\d{2}-\d{7}\b'
|
||
|
|
|
||
|
|
if not isinstance(tin, str):
|
||
|
|
try:
|
||
|
|
tin = str(tin)
|
||
|
|
except TypeError:
|
||
|
|
check = False
|
||
|
|
|
||
|
|
return tin, check
|
||
|
|
|
||
|
|
check=bool(re.match(pattern, tin))
|
||
|
|
|
||
|
|
return tin, check
|
||
|
|
|
||
|
|
|
||
|
|
def npi_check(npi: str) -> tuple[str, bool]:
|
||
|
|
pattern = r'^[1-2]\d{9}$'
|
||
|
|
if not isinstance(npi, str):
|
||
|
|
try:
|
||
|
|
npi = str(npi)
|
||
|
|
except TypeError:
|
||
|
|
check = False
|
||
|
|
|
||
|
|
return npi, check
|
||
|
|
|
||
|
|
check = bool(re.match(pattern, npi))
|
||
|
|
|
||
|
|
return npi, check
|
||
|
|
|
||
|
|
def date_format_check(date: str) -> tuple:
|
||
|
|
|
||
|
|
format = '%m/%d/%Y'
|
||
|
|
|
||
|
|
try:
|
||
|
|
dt = datetime.strptime(date, format)
|
||
|
|
date = dt.strftime('%m/%d/%Y')
|
||
|
|
check = True
|
||
|
|
except ValueError:
|
||
|
|
date = date
|
||
|
|
check = False
|
||
|
|
|
||
|
|
return date, check
|
||
|
|
|
||
|
|
def yn_check(yn: str) -> tuple[str, bool]:
|
||
|
|
if not isinstance(yn, str):
|
||
|
|
try:
|
||
|
|
yn = str(yn)
|
||
|
|
except TypeError:
|
||
|
|
check = False
|
||
|
|
|
||
|
|
return yn, check
|
||
|
|
|
||
|
|
return yn, (yn == 'Y' or yn == 'N')
|
||
|
|
|
||
|
|
|
||
|
|
def state_check(state: str) -> tuple[str, bool]:
|
||
|
|
|
||
|
|
if not isinstance(state, str):
|
||
|
|
try:
|
||
|
|
state = str(state)
|
||
|
|
except TypeError:
|
||
|
|
check = False
|
||
|
|
return state, check
|
||
|
|
|
||
|
|
if state in STATE_MAP.keys():
|
||
|
|
state = STATE_MAP[state]
|
||
|
|
check = True
|
||
|
|
elif state.strip().title() in STATE_MAP.values():
|
||
|
|
check = True
|
||
|
|
else:
|
||
|
|
check = False
|
||
|
|
|
||
|
|
return state, check
|
||
|
|
|
||
|
|
def pages_pagenum_check(pages: str, pagenum: str) -> bool:
|
||
|
|
|
||
|
|
if not isinstance(pages, str):
|
||
|
|
try:
|
||
|
|
pages = str(pages)
|
||
|
|
except TypeError:
|
||
|
|
check = False
|
||
|
|
return check
|
||
|
|
elif not isinstance(pagenum, str):
|
||
|
|
try:
|
||
|
|
pagenum = str(pagenum)
|
||
|
|
except TypeError:
|
||
|
|
check = False
|
||
|
|
return check
|
||
|
|
else:
|
||
|
|
check = (pagenum < pages)
|
||
|
|
|
||
|
|
return check
|
||
|
|
|
||
|
|
|
||
|
|
def perform_qc_qa_tests_both(ac_df, b_df, merged_df, input_dict, processed_files):
|
||
|
|
|
||
|
|
file_name_column = config.FILE_NAME_COLUMN
|
||
|
|
|
||
|
|
total_input_files = len(input_dict) if input_dict else 0
|
||
|
|
processed_ac_files = len(
|
||
|
|
[f for f in processed_files if f in ac_df[file_name_column].values]
|
||
|
|
)
|
||
|
|
processed_b_files = len(
|
||
|
|
[f for f in processed_files if f in b_df[file_name_column].values]
|
||
|
|
)
|
||
|
|
failed_ac_files = (
|
||
|
|
total_input_files - processed_ac_files if total_input_files > 0 else 0
|
||
|
|
)
|
||
|
|
failed_b_files = (
|
||
|
|
total_input_files - processed_b_files if total_input_files > 0 else 0
|
||
|
|
)
|
||
|
|
|
||
|
|
yn_cols_b = [col for col in b_df.columns if '(Y/N)' in col]
|
||
|
|
yn_cols_ac = [col for col in ac_df.columns if '(Y/N)' in col]
|
||
|
|
|
||
|
|
general_stats = {
|
||
|
|
"Total Number of Contracts processed": total_input_files,
|
||
|
|
"# of contracts successfully processed for AC fields": processed_ac_files,
|
||
|
|
"# of contracts successfully processed for B fields": processed_b_files,
|
||
|
|
"# of contracts failed for AC fields": failed_ac_files,
|
||
|
|
"# of contracts failed for B fields": failed_b_files,
|
||
|
|
"# of rows for AC run": len(ac_df),
|
||
|
|
"# of rows for B run": len(b_df),
|
||
|
|
"Total rows in merged DataFrame": len(merged_df),
|
||
|
|
"Unique Filenames in AC DataFrame": ac_df[file_name_column].nunique(),
|
||
|
|
"Unique Filenames in B DataFrame": b_df[file_name_column].nunique(),
|
||
|
|
"Unique Filenames in merged DataFrame": merged_df[file_name_column].nunique(),
|
||
|
|
"Contracts with all AC fields missing": merged_df[
|
||
|
|
[col for col in ac_df.columns if col != file_name_column]
|
||
|
|
]
|
||
|
|
.isnull()
|
||
|
|
.all(axis=1)
|
||
|
|
.sum(),
|
||
|
|
"Contracts with all B fields missing": merged_df[
|
||
|
|
[col for col in b_df.columns if col != file_name_column]
|
||
|
|
]
|
||
|
|
.isnull()
|
||
|
|
.all(axis=1)
|
||
|
|
.sum(),
|
||
|
|
"# of duplicate rows for AC run": ac_df.duplicated().sum(),
|
||
|
|
"# of duplicate rows for B run": b_df.duplicated().sum(),
|
||
|
|
"Number of duplicate rows in merged DataFrame": merged_df.duplicated().sum(),
|
||
|
|
"# of fields for AC run" : len(ac_df.columns),
|
||
|
|
"# of fields for B run" : len(b_df.columns),
|
||
|
|
"# of incorrectly formatted Provider Group TINs" : ac_df["IRS #"].apply(lambda x: not tin_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||
|
|
"# of incorrectly formatted Provider Group NPIs" : ac_df["NPI (10-digits)"].apply(lambda x: not npi_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||
|
|
"# of incorrectly formatted Provider Group Signatory TINs": ac_df["PROV_GROUP_TIN_SIGNATORY"].apply(lambda x: not tin_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||
|
|
"# of incorrectly formatted Provider TINs (other)" : ac_df["PROV_TIN_OTHER"].apply(lambda x: not tin_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||
|
|
"# of incorrectly formatted Provider NPIs (other)": ac_df["PROV_NPI_OTHER"].apply(lambda x: not npi_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||
|
|
"# of incorrect Health Plan States" : ac_df['Health Plan State'].apply(lambda x: not state_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||
|
|
"# of rows with Pages greater than Page Num" : merged_df.apply(lambda row: pages_pagenum_check(row['Pages'], row['Page_Num']) if (not utils.is_empty(row['Pages']) and utils.is_empty(row['Page_Num'])) else False, axis=1).sum(),
|
||
|
|
"# of Payer Names found in Notice to Provider Name" : ac_df.apply(lambda row: row['PAYER NAME'] in row['Notice to Provider Name'] if not utils.is_empty(row['Notice to Provider Name']) else False, axis = 1).sum(),
|
||
|
|
"# of invalid Y/N values for B run" : b_df[yn_cols_b].apply(lambda x: yn_check(x)[1], axis = 1).sum().sum(),
|
||
|
|
"# of invalid Y/N values for AC run" : ac_df[yn_cols_ac].apply(lambda x: yn_check(x)[1], axis = 1).sum().sum(),
|
||
|
|
"# of invalid Termination Dates" : ac_df['Termination Date'].apply(lambda x: not date_format_check(x) if not utils.is_empty(x) else x).sum(),
|
||
|
|
"# of invalid Conract Effective Dates" : ac_df['Contract Effective Date'].apply(lambda x: not date_format_check(x) if not utils.is_empty(x) else x).sum()
|
||
|
|
}
|
||
|
|
|
||
|
|
ac_stats = {}
|
||
|
|
for field in [col for col in ac_df.columns if col != file_name_column]:
|
||
|
|
missing_contracts = merged_df[merged_df[field].isnull()][
|
||
|
|
file_name_column
|
||
|
|
].nunique()
|
||
|
|
ac_stats[field] = {
|
||
|
|
"Contracts missing this field (Merged DataFrame)": missing_contracts,
|
||
|
|
**detect_date_anomalies(ac_df[field], field, is_merged=False),
|
||
|
|
**detect_date_anomalies(merged_df[field], field, is_merged=True),
|
||
|
|
}
|
||
|
|
|
||
|
|
b_stats = {}
|
||
|
|
for field in [col for col in b_df.columns if col != file_name_column]:
|
||
|
|
missing_contracts = (
|
||
|
|
merged_df.groupby(file_name_column)[field]
|
||
|
|
.apply(lambda x: x.isnull().all())
|
||
|
|
.sum()
|
||
|
|
)
|
||
|
|
contracts_with_missing = (
|
||
|
|
merged_df.groupby(file_name_column)[field]
|
||
|
|
.apply(lambda x: x.isnull().any())
|
||
|
|
.sum()
|
||
|
|
)
|
||
|
|
total_missing = merged_df[field].isnull().sum()
|
||
|
|
b_stats[field] = {
|
||
|
|
"Contracts missing all entries (Merged DataFrame)": missing_contracts,
|
||
|
|
"Contracts with at least one missing entry (Merged DataFrame)": contracts_with_missing,
|
||
|
|
"Total missing values (Merged DataFrame)": total_missing,
|
||
|
|
**detect_date_anomalies(b_df[field], field, is_merged=False),
|
||
|
|
**detect_date_anomalies(merged_df[field], field, is_merged=True),
|
||
|
|
}
|
||
|
|
|
||
|
|
return general_stats, ac_stats, b_stats
|
||
|
|
|
||
|
|
|
||
|
|
def perform_qc_qa_tests_ac(ac_df, input_dict, processed_files):
|
||
|
|
|
||
|
|
file_name_column = config.FILE_NAME_COLUMN
|
||
|
|
|
||
|
|
total_input_files = len(input_dict) if input_dict else 0
|
||
|
|
processed_ac_files = len(
|
||
|
|
[f for f in processed_files if f in ac_df[file_name_column].values]
|
||
|
|
)
|
||
|
|
failed_ac_files = (
|
||
|
|
total_input_files - processed_ac_files if total_input_files > 0 else 0
|
||
|
|
)
|
||
|
|
|
||
|
|
yn_cols_ac = [col for col in ac_df.columns if '(Y/N)' in col]
|
||
|
|
|
||
|
|
general_stats = {
|
||
|
|
"Total Number of Contracts processed": total_input_files,
|
||
|
|
"# of contracts successfully processed for AC fields": processed_ac_files,
|
||
|
|
"# of contracts failed for AC fields": failed_ac_files,
|
||
|
|
"# of rows for AC run": len(ac_df),
|
||
|
|
"Unique Filenames in AC DataFrame": ac_df[file_name_column].nunique(),
|
||
|
|
"Contracts with all AC fields missing": ac_df[
|
||
|
|
[col for col in ac_df.columns if col != file_name_column]
|
||
|
|
]
|
||
|
|
.isnull()
|
||
|
|
.all(axis=1)
|
||
|
|
.sum(),
|
||
|
|
"# of duplicate rows for AC run": ac_df.duplicated().sum(),
|
||
|
|
"# of fields for AC run" : len(ac_df.columns),
|
||
|
|
"# of incorrectly formatted Provider Group TINs" : ac_df["IRS #"].apply(lambda x: not tin_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||
|
|
"# of incorrectly formatted Provider Group NPIs" : ac_df["NPI (10-digits)"].apply(lambda x: not npi_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||
|
|
"# of incorrectly formatted Provider Group Signatory TINs": ac_df["PROV_GROUP_TIN_SIGNATORY"].apply(lambda x: not tin_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||
|
|
"# of incorrectly formatted Provider TINs (other)" : ac_df["PROV_TIN_OTHER"].apply(lambda x: not tin_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||
|
|
"# of incorrectly formatted Provider NPIs (other)": ac_df["PROV_NPI_OTHER"].apply(lambda x: not npi_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||
|
|
"# of incorrect Health Plan States" : ac_df['Health Plan State'].apply(lambda x: not state_check(x)[1] if not utils.is_empty(x) else False).sum(),
|
||
|
|
"# of Payer Names found in Notice to Provider Name" : ac_df.apply(lambda row: row['PAYER NAME'] in row['Notice to Provider Name'] if not utils.is_empty(row['Notice to Provider Name']) else False, axis = 1).sum(),
|
||
|
|
"# of invalid Y/N values for AC run" : ac_df[yn_cols_ac].apply(lambda x: yn_check(x)[1], axis = 1).sum().sum(),
|
||
|
|
"# of invalid Termination Dates" : ac_df['Termination Date'].apply(lambda x: not date_format_check(x) if not utils.is_empty(x) else x).sum(),
|
||
|
|
"# of invalid Conract Effective Dates" : ac_df['Contract Effective Date'].apply(lambda x: not date_format_check(x) if not utils.is_empty(x) else x).sum()
|
||
|
|
}
|
||
|
|
|
||
|
|
ac_stats = {}
|
||
|
|
for field in [col for col in ac_df.columns if col != file_name_column]:
|
||
|
|
missing_contracts = ac_df[ac_df[field].isnull()][file_name_column].nunique()
|
||
|
|
ac_stats[field] = {
|
||
|
|
"Contracts missing this field (Merged DataFrame)": missing_contracts,
|
||
|
|
**detect_date_anomalies(ac_df[field], field, is_merged=False),
|
||
|
|
}
|
||
|
|
|
||
|
|
return general_stats, ac_stats
|
||
|
|
|
||
|
|
|
||
|
|
def perform_qc_qa_tests_b(b_df, input_dict, processed_files):
|
||
|
|
|
||
|
|
file_name_column = config.FILE_NAME_COLUMN
|
||
|
|
|
||
|
|
total_input_files = len(input_dict) if input_dict else 0
|
||
|
|
processed_b_files = len(
|
||
|
|
[f for f in processed_files if f in b_df[file_name_column].values]
|
||
|
|
)
|
||
|
|
failed_b_files = (
|
||
|
|
total_input_files - processed_b_files if total_input_files > 0 else 0
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
yn_cols_b = [col for col in b_df.columns if '(Y/N)' in col]
|
||
|
|
|
||
|
|
general_stats = {
|
||
|
|
"Total Number of Contracts processed": total_input_files,
|
||
|
|
"# of contracts successfully processed for B fields": processed_b_files,
|
||
|
|
"# of contracts failed for B fields": failed_b_files,
|
||
|
|
"# of rows for B run": len(b_df),
|
||
|
|
"Unique Filenames in B DataFrame": b_df[file_name_column].nunique(),
|
||
|
|
"Contracts with all B fields missing": b_df[
|
||
|
|
[col for col in b_df.columns if col != file_name_column]
|
||
|
|
]
|
||
|
|
.isnull()
|
||
|
|
.all(axis=1)
|
||
|
|
.sum(),
|
||
|
|
"# of duplicate rows for B run": b_df.duplicated().sum(),
|
||
|
|
"# of fields for B run" : len(b_df.columns),
|
||
|
|
"# of rows with Pages greater than Page Num" : b_df.apply(lambda row: pages_pagenum_check(row['Pages'], row['Page_Num']), axis=1).sum(),
|
||
|
|
"# of invalid Y/N values for B run" : b_df[yn_cols_b].apply(lambda x: yn_check(x)[1], axis = 1).sum().sum()}
|
||
|
|
|
||
|
|
b_stats = {}
|
||
|
|
for field in [col for col in b_df.columns if col != file_name_column]:
|
||
|
|
missing_contracts = (
|
||
|
|
b_df.groupby(file_name_column)[field]
|
||
|
|
.apply(lambda x: x.isnull().all())
|
||
|
|
.sum()
|
||
|
|
)
|
||
|
|
contracts_with_missing = (
|
||
|
|
b_df.groupby(file_name_column)[field]
|
||
|
|
.apply(lambda x: x.isnull().any())
|
||
|
|
.sum()
|
||
|
|
)
|
||
|
|
total_missing = b_df[field].isnull().sum()
|
||
|
|
b_stats[field] = {
|
||
|
|
"Contracts missing all entries (B DataFrame)": missing_contracts,
|
||
|
|
"Contracts with at least one missing entry (B DataFrame)": contracts_with_missing,
|
||
|
|
"Total missing values (Merged DataFrame)": total_missing,
|
||
|
|
**detect_date_anomalies(b_df[field], field, is_merged=False),
|
||
|
|
}
|
||
|
|
|
||
|
|
return general_stats, b_stats
|
||
|
|
|
||
|
|
|
||
|
|
|
||
|
|
def get_table_stats(text_dict):
|
||
|
|
table_count, num_table_pages, rate_count = 0, 0, 0
|
||
|
|
table_pages, rate_pages, rates_more_than_10 = [], [], []
|
||
|
|
for page_num, page_text in text_dict.items():
|
||
|
|
if utils.contains_reimbursement(page_text, page_num):
|
||
|
|
rate_pages.append(page_num)
|
||
|
|
rates_on_page = sum(1 for char in page_text if char in {"$", "%"})
|
||
|
|
rate_count += rates_on_page
|
||
|
|
if rates_on_page >= 10:
|
||
|
|
rates_more_than_10.append(page_num)
|
||
|
|
|
||
|
|
table_texts = re.findall(
|
||
|
|
r"-------Table Start--------(.*?)-------Table End--------",
|
||
|
|
page_text,
|
||
|
|
re.DOTALL,
|
||
|
|
)
|
||
|
|
table_page_count = len(table_texts)
|
||
|
|
if table_page_count > 0:
|
||
|
|
table_count += table_page_count
|
||
|
|
num_table_pages += 1
|
||
|
|
table_pages.append(page_num)
|
||
|
|
|
||
|
|
return {
|
||
|
|
"Table Count": table_count,
|
||
|
|
"Table Page Count": num_table_pages,
|
||
|
|
"Table Pages": table_pages,
|
||
|
|
"Rate Count": rate_count,
|
||
|
|
"Rate Pages": rate_pages,
|
||
|
|
"Num Rate Pages": len(rate_pages),
|
||
|
|
">=10 Rate Pages": rates_more_than_10,
|
||
|
|
"Num >=10 Rates": len(rates_more_than_10),
|
||
|
|
}
|
||
|
|
|
||
|
|
def table_stats_qaqc(input_dict, ac_df, b_df, file_name_column) -> list[dict]:
|
||
|
|
|
||
|
|
table_stats = []
|
||
|
|
if input_dict:
|
||
|
|
for filename, content in input_dict.items():
|
||
|
|
filename_without_ext = utils.remove_txt_extension(filename).lower()
|
||
|
|
text_dict = preprocessing_funcs.split_text(content)
|
||
|
|
stats = get_table_stats(text_dict)
|
||
|
|
table_stats.append(
|
||
|
|
{
|
||
|
|
"Contract Name": filename,
|
||
|
|
"Has B output": (
|
||
|
|
filename_without_ext in b_df[file_name_column].values
|
||
|
|
if b_df is not None
|
||
|
|
else "No B Output Given"
|
||
|
|
),
|
||
|
|
"Has AC output": (
|
||
|
|
filename_without_ext in ac_df[file_name_column].values
|
||
|
|
if ac_df is not None
|
||
|
|
else "No AC Output Given"
|
||
|
|
),
|
||
|
|
"# Tables": stats["Table Count"],
|
||
|
|
"# Reimb rates": stats["Rate Count"],
|
||
|
|
"Table Pages": ", ".join(map(str, stats["Table Pages"])),
|
||
|
|
"Rate Pages": ", ".join(map(str, stats["Rate Pages"])),
|
||
|
|
">=10 Rate Pages": ", ".join(map(str, stats[">=10 Rate Pages"])),
|
||
|
|
}
|
||
|
|
)
|
||
|
|
|
||
|
|
return table_stats
|
||
|
|
|
||
|
|
def perform_qc_qa(ac_final_df, b_final_df, abc_final_df, input_dict):
|
||
|
|
|
||
|
|
# Perform QA/QC analysis
|
||
|
|
wb = Workbook()
|
||
|
|
|
||
|
|
ws_general = wb.active
|
||
|
|
ws_general.title = "General Stats"
|
||
|
|
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
|
|
|
||
|
|
|
||
|
|
# Get list of all processed files
|
||
|
|
processed_files = []
|
||
|
|
if ac_final_df is not None:
|
||
|
|
processed_files.extend(ac_final_df["Contract Name"].tolist())
|
||
|
|
if b_final_df is not None:
|
||
|
|
processed_files.extend(b_final_df["Contract Name"].tolist())
|
||
|
|
processed_files = list(set(processed_files))
|
||
|
|
|
||
|
|
# Generate QA/QC stats based on available dataframes
|
||
|
|
if ac_final_df is not None and b_final_df is not None:
|
||
|
|
general_stats, ac_stats, b_stats = perform_qc_qa_tests_both(
|
||
|
|
ac_final_df, b_final_df, abc_final_df, input_dict, processed_files
|
||
|
|
)
|
||
|
|
elif ac_final_df is not None:
|
||
|
|
general_stats, ac_stats = perform_qc_qa_tests_ac(
|
||
|
|
ac_final_df, input_dict, processed_files
|
||
|
|
)
|
||
|
|
else:
|
||
|
|
general_stats, b_stats = perform_qc_qa_tests_b(
|
||
|
|
b_final_df, input_dict, processed_files
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
# Add metadata to general stats
|
||
|
|
general_stats_with_metadata = {
|
||
|
|
"Report Generation Time": timestamp,
|
||
|
|
"Batch ID": config.BATCH_ID,
|
||
|
|
"AC Processing": "Yes" if ("a" in config.FIELDS and "c" in config.FIELDS) or (config.AC_DF) else "No",
|
||
|
|
"B Processing": "Yes" if ("b" in config.FIELDS) or (config.B_DF) else "No",
|
||
|
|
**general_stats, }
|
||
|
|
|
||
|
|
|
||
|
|
# Write general stats
|
||
|
|
general_df = pd.DataFrame.from_dict(
|
||
|
|
general_stats_with_metadata, orient="index", columns=["Value"]
|
||
|
|
)
|
||
|
|
for r in dataframe_to_rows(general_df, index=True, header=True):
|
||
|
|
ws_general.append(r)
|
||
|
|
|
||
|
|
|
||
|
|
# Write AC stats if available
|
||
|
|
if ac_final_df is not None:
|
||
|
|
ws_ac = wb.create_sheet("AC Fields Stats")
|
||
|
|
ac_stats_df = pd.DataFrame(ac_stats).T
|
||
|
|
for r in dataframe_to_rows(ac_stats_df, index=True, header=True):
|
||
|
|
ws_ac.append(r)
|
||
|
|
|
||
|
|
|
||
|
|
# Write B stats if available
|
||
|
|
if b_final_df is not None:
|
||
|
|
ws_b = wb.create_sheet("B Fields Stats")
|
||
|
|
b_stats_df = pd.DataFrame(b_stats).T
|
||
|
|
for r in dataframe_to_rows(b_stats_df, index=True, header=True):
|
||
|
|
ws_b.append(r)
|
||
|
|
|
||
|
|
|
||
|
|
# Generate table statistics
|
||
|
|
table_stats = table_stats_qaqc(input_dict, ac_final_df, b_final_df, config.FILE_NAME_COLUMN)
|
||
|
|
|
||
|
|
ws_table_stats = wb.create_sheet("Table and Reimbursement Stats")
|
||
|
|
table_stats_df = pd.DataFrame(table_stats)
|
||
|
|
for r in dataframe_to_rows(table_stats_df, index=False, header=True):
|
||
|
|
ws_table_stats.append(r)
|
||
|
|
|
||
|
|
return wb
|
||
|
|
|
||
|
|
def save_qcqa_wb(wb, run_timestamp):
|
||
|
|
|
||
|
|
# Save QC report locally
|
||
|
|
report_filename = f'QC_QA_Report_{config.BATCH_ID}_{datetime.now().strftime("%Y%m%d_%H%M%S")}.xlsx'
|
||
|
|
report_path = os.path.join(config.CONSOLIDATED_OUTPUT_DIRECTORY, report_filename)
|
||
|
|
wb.save(report_path)
|
||
|
|
print(f"QC/QA Report saved to {report_path}")
|
||
|
|
|
||
|
|
|
||
|
|
# Upload to S3
|
||
|
|
if config.WRITE_TO_S3:
|
||
|
|
BUCKET_NAME = config.S3_OUTPUT_BUCKET
|
||
|
|
print(f"Uploading final outputs to s3://{BUCKET_NAME}/{run_timestamp}")
|
||
|
|
|
||
|
|
|
||
|
|
# Upload consolidated outputs
|
||
|
|
consolidated_files = [
|
||
|
|
f"{config.BATCH_ID}-AC.csv",
|
||
|
|
f"{config.BATCH_ID}-B.csv",
|
||
|
|
f"{config.BATCH_ID}-ABC.csv",
|
||
|
|
report_filename #Add QC Report to upload list
|
||
|
|
]
|
||
|
|
|
||
|
|
|
||
|
|
for filename in consolidated_files:
|
||
|
|
local_path = os.path.join(config.CONSOLIDATED_OUTPUT_DIRECTORY, filename)
|
||
|
|
if os.path.exists(local_path):
|
||
|
|
s3_key = f"{run_timestamp}/consolidated/{filename}"
|
||
|
|
try:
|
||
|
|
print(f"uploading {local_path} to s3://{BUCKET_NAME}/{s3_key}")
|
||
|
|
config.S3_CLIENT.upload_file(local_path, BUCKET_NAME, s3_key)
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error uploading {local_path}: {str(e)}")
|
||
|
|
else:
|
||
|
|
print(f"File not found: {local_path}")
|
||
|
|
|
||
|
|
|
||
|
|
# Upload tracking files
|
||
|
|
tracking_files = {
|
||
|
|
'cost_log': config.COST_LOG_NAME,
|
||
|
|
'file_log': config.FILE_LOG_NAME
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
for file_type, file_name in tracking_files.items():
|
||
|
|
if os.path.exists(file_name):
|
||
|
|
try:
|
||
|
|
print(f"uploading final {file_name}")
|
||
|
|
s3_key = os.path.join(run_timestamp, file_name).replace("\\", "/")
|
||
|
|
config.S3_CLIENT.upload_file(file_name, BUCKET_NAME, s3_key)
|
||
|
|
|
||
|
|
except Exception as e:
|
||
|
|
print(f"Error uploading {file_name}: {str(e)}")
|
||
|
|
|
||
|
|
print(f"Upload completed to s3://{BUCKET_NAME}/{run_timestamp}")
|