doczyai-pipelines/archive/scripts/cnc_hotfixes/apply_cnc_hotfix.py

import concurrent.futures
import os

import pandas as pd
from rapidfuzz import fuzz, process

import cnc_hotfix
import config
import io_utils
import keywords
import preprocessing_funcs
import valid

clean_file_name = config.get_arg_value('clean_file', '')
ABC_PATH = f'{clean_file_name}'

def get_highest_similarity(target_str, str_list):
    if target_str in str_list:
        return target_str
    best_match = process.extractOne(target_str, str_list)
    return best_match[0] if best_match else None


def process_hotfix(abc, contract_text):
    """
    Processes a contract by applying a series of hotfixes to the original output data.

    Parameters:
    abc (DataFrame): The original output DataFrame for one contract. This DataFrame contains 114 columns.
    contract_text (str): The text of the contract.

    Returns:
    DataFrame: The original abc DataFrame augmented with additional columns for each hotfix applied.
    """

    filename = abc['Contract Name'].unique()[0]
    
    original_length = len(abc)

    if contract_text is not None:
        # print(filename, len(contract_text))

        # Preprocess
        contract_text = preprocessing_funcs.clean_newlines(contract_text)
        contract_text = preprocessing_funcs.clean_law_symbols(contract_text)

        text_dict = preprocessing_funcs.split_text(
            contract_text
        )  # return a dictionary with keys - page_num (str), values as the page_text
        text_dict, top_sheet_dict = preprocessing_funcs.filter_quick_review(text_dict)
        ac_chunks = preprocessing_funcs.smart_chunk_ac(text_dict, contract_text, keywords.GROUPED_KEYWORD_MAPPINGS)

        ######################################### Hot fixes ##########################################

        ######### Rule/Regex-based #########
        # Default
        abc = cnc_hotfix.check_default_term(abc)
        abc = cnc_hotfix.check_default_rate(abc)
        abc = cnc_hotfix.populate_default_term(abc, text_dict)
        abc.drop_duplicates(inplace=True)

        # Health Plan State
        abc = cnc_hotfix.clean_healthplan_state(abc, filename, text_dict)
        abc.drop_duplicates(inplace=True)

        # IRS
        abc = cnc_hotfix.clean_irs(abc, filename, text_dict)
        abc.drop_duplicates(inplace=True)
            
        # NPI
        abc = cnc_hotfix.clean_npi(abc, filename, text_dict, top_sheet_dict)
        abc.drop_duplicates(inplace=True)
            
        # Exhibit
        abc = cnc_hotfix.clean_exhibit(abc, text_dict)
        abc.drop_duplicates(inplace=True)

        # LOB
        abc = cnc_hotfix.clean_lob(abc, text_dict)
        abc.drop_duplicates(inplace=True)

        # Lesser
        abc = cnc_hotfix.clean_lesser(abc, text_dict)
        abc.drop_duplicates(inplace=True)
        
        # Provider Type Level 2
        abc = cnc_hotfix.clean_provider_type_2(abc)
        abc.drop_duplicates(inplace=True)

        ######## Prompt-based #########
        # Rate Standard
        abc = cnc_hotfix.clean_rate_standard(abc)
        abc.drop_duplicates(inplace=True)

        # Contract Effective Date
        abc = cnc_hotfix.clean_contract_effective_date(abc, filename, text_dict)
        abc.drop_duplicates(inplace=True)
        
        # Term Group
        abc = cnc_hotfix.clean_term_group(abc, filename, text_dict, ac_chunks)
        abc.drop_duplicates(inplace=True)

        # Agreement Name
        abc = cnc_hotfix.clean_agreement_name(abc, filename, text_dict)
        abc.drop_duplicates(inplace=True)

    else:
        abc["Contract Duplicate Issue"] = "No Contract Found"
    
    ################## CREATE OUTPUT DIRECTORIES ##################
    base_filename = os.path.splitext(filename)[0].strip()
    output_dir = os.path.join(config.OUTPUT_DIRECTORY, base_filename)
    os.makedirs(output_dir, exist_ok=True)

    ################## WRITE UNPROCESSED OUTPUT ##################
    # Reorder
    abc = abc[[col for col in cnc_hotfix.HOTFIX_ORDER if col in abc.columns] + [col for col in abc.columns if col not in cnc_hotfix.HOTFIX_ORDER]]
    # Write to Excel
    abc.to_csv(
        os.path.join(output_dir, config.B_RESULTS_NAME), index=False
    )

    return abc


def main():

    REVERSE_MAPPING = {
        'Agreement Name (Contract Title)': 'Agreement_Name (Contract Title)',
        'Payer Name': 'PAYER NAME',
        'NPI (10-digits': 'NPI (10-digits)',
        'Prov Group TIN Signatory': 'PROV_GROUP_TIN_SIGNATORY',
        'Prov TIN Other': 'PROV_TIN_OTHER',
        'Prov NPI Other': 'PROV_NPI_OTHER',
        'Page Num': 'Page_Num',
        'Lesser of Logic Language, Included (Y/N)': 'Lesser of Logic language, included (Y/N)',
        'Reimb Methodology Short': 'Reimb. Methodology_Short',
        'If Rate is % of Payor or MCR [Standard]': 'If rate is % of Payor or MCR [STANDARD]',
        'If Rate is % of Payor or MCR [Standard] Short': 'If rate is % of Payor or MCR [STANDARD]_Short',
        'Flat Fee': 'FLAT FEE',
        'CDM Neutralization Language, Included (Y/N)': 'CDM Neutralization Language, included (Y/N)',
        'Contract Chargemaster Protection Language': 'CONTRACT_CHARGEMASTER_PROTECTION_LANGUAGE',
        'IP - DSH/IME/UC, Included (Y/N)': 'IP - DSH/IME/UC, included (Y/N)',
        'State': 'Health Plan State'
    }

    # Read clean data
    print("Reading clean data...")
    if '.xlsx' in ABC_PATH:
        abc_clean = pd.read_excel(ABC_PATH)
    elif '.csv' in ABC_PATH:
        abc_clean = pd.read_csv(ABC_PATH)
    elif '.xlsb' in ABC_PATH:
        abc_clean = io_utils.read_xlsb(ABC_PATH)
    

    # Print current columns and identify invalid ones
    print("Current columns:")
    for col in abc_clean.columns:
        print(col)

    abc_clean.rename(columns=REVERSE_MAPPING, inplace=True)

    print(f"Invalid columns: {[col for col in abc_clean.columns if col not in cnc_hotfix.HOTFIX_ORDER]}")
    
    

    abc_clean = abc_clean.dropna(how='all')
    abc_clean = abc_clean.dropna(axis=1, how='all')

    unique_output_files = abc_clean['Contract Name'].unique()

    print("ABC Clean: ", abc_clean.shape)
    print(len(unique_output_files), 'files in clean output...')
    # print(list(abc_clean.columns))
    abc_clean.rename(columns=valid.HOTFIX_MAPPING, inplace=True)
    abc_clean['Contract Name'] = abc_clean['Contract Name'].apply(lambda x: ('Filename: ' + str(x) if not str(x).startswith('Filename: ') else str(x)) + ('.txt' if not str(x).endswith('.txt') else ''))
    unique_output_files = abc_clean['Contract Name'].unique()
    
    already_ran = os.listdir(config.OUTPUT_DIRECTORY)

    # Read input dict
    print("Reading input .txt files...")
    input_dict = io_utils.read_input()
    input_dict = {'Filename: ' + filename.replace('.txt', '').strip() : contract_text for filename, contract_text in input_dict.items()}
    if config.FILTER_ALREADY_PROCESSED:
        input_dict = {filename : contract_text for filename, contract_text in input_dict.items() if filename not in already_ran}
    print("Input Dict: ", len(input_dict))
    
    # input_keys = list(input_dict.keys())
    # abc_clean['Contract Name'] = abc_clean['Contract Name'].apply(
    #     lambda x: get_highest_similarity(x, input_keys)
    # )

    unique_output_files = abc_clean['Contract Name'].unique()
    print(f"{len([file for file in input_dict.keys() if file not in unique_output_files])} files in full s3, not in output")
    print(f"{len([file for file in unique_output_files if file not in input_dict.keys()])} files in output, not in full s3")
    
    # Create list of tuples
    abc_clean_list = [(group, input_dict.get(name.replace('.txt', ''))) for name, group in abc_clean.groupby('Contract Name')]
    print(f"{len(abc_clean_list)} - files in input")
    print(f"{len([f for f in abc_clean_list if f[1]])} - valid files in input")
    del abc_clean
    del input_dict

    abc_clean_list = [f for f in abc_clean_list if f[1]]
    print(len(abc_clean_list))

    df_list = []
    with concurrent.futures.ThreadPoolExecutor(max_workers=config.MAX_WORKERS) as executor:
        # Create a list of future objects by submitting process_pair function to the executor
        futures = [executor.submit(process_hotfix, abc_df, contract_text) for abc_df, contract_text in abc_clean_list]
        # As each future completes, retrieve its result and add to df_list
        for future in concurrent.futures.as_completed(futures):
            try:
                new_df = future.result()
                if new_df is not None:
                    df_list.append(new_df)
            except Exception as e:
                # Log the exception or handle it otherwise
                print(f"An error occurred: {e}")
        
    abc_final = pd.concat(df_list, ignore_index=True)
    print(abc_final.shape)
    print(len(abc_final['Contract Name'].unique()), ' unique contracts')
    
    print("Writing to excel...")
    abc_final.to_excel(f'output_consolidated/{config.BATCH_ID}-Hotfix-Final.xlsx')

    print("Completed writing to Excel.")


if __name__ == "__main__":
    main()