doczyai-pipelines/fieldExtraction/src/postprocessing_funcs.py

import pandas as pd
import numpy as np
import re
import difflib

import config
import prompts
import valid
from valid import DERIVED_INDICATOR_FIELDS
import claude_funcs
from utils import is_empty

import logging

logging.basicConfig(
    level=logging.ERROR, format="%(asctime)s - %(levelname)s - %(message)s"
)


def sanitize_value(value):
    try:
        if isinstance(value, list):
            return ", ".join(str(v) for v in value)
        elif isinstance(value, str):
            value = value.strip("[]")
            return ", ".join([item.strip(" '") for item in value.split(",")])
        elif pd.isna(value):
            return "N/A"
    except:
        return value


def exact_match(val, valid_values):
    val = val.strip().upper()
    for valid_val in valid_values:
        if val == valid_val.upper():
            return valid_val
    return None


def get_closest_match(val, valid_values, similarity_threshold=0.7):
    if pd.isna(val):
        return None
    val = val.strip().upper()
    matches = difflib.get_close_matches(
        val, [v.upper() for v in valid_values], n=1, cutoff=similarity_threshold
    )
    return matches[0] if matches else None


def correct_misplaced_values(df, columns, valid_values_dict):
    for index, row in df.iterrows():
        for col in columns:
            if pd.notna(row[col]):
                terms = row[col].split(",")
                for term in terms:
                    term = term.strip()
                    for target_col, valid_values in valid_values_dict.items():
                        if target_col != col:
                            match = exact_match(term, valid_values)
                            if match:
                                if (
                                    pd.isna(row[target_col])
                                    or not row[target_col].strip()
                                ):
                                    df.at[index, target_col] = match
                                    df.at[index, col] = None
                                else:
                                    current_value = row[target_col].strip()
                                    if (
                                        get_closest_match(
                                            match,
                                            [current_value],
                                            config.FUZZY_MATCH_THRESHOLD,
                                        )
                                        is None
                                    ):
                                        df.at[index, "Corrected_" + target_col] = (
                                            f"Found {term} in {col} cell"
                                        )
                                        df.at[index, col] = None
    return df


def filter_service_column(answer_dicts):
    filtered_list = []
    for d in answer_dicts:
        clean_dict = True
        for keyword in valid.SERVICE_FILTER:
            if (
                keyword.upper() in d["FULL_SERVICE"].upper()
                or d["FULL_SERVICE"].upper() in keyword.upper()
            ):
                clean_dict = False
        if clean_dict:
            filtered_list.append(d)
    return filtered_list


def filter_methodology_column(answer_dicts):
    filtered_list = []
    for d in answer_dicts:
        clean_dict = True
        for keyword in valid.METHODOLOGY_FILTER:
            if (
                keyword.upper() in d["FULL_METHODOLOGY"].upper()
                or d["FULL_METHODOLOGY"].upper() in keyword.upper()
            ):
                clean_dict = False
        if clean_dict:
            filtered_list.append(d)
    return filtered_list


def clean_td(td):
    td_clean = []
    for d in td:
        new_d = {}
        for k, v in d.items():
            if "DATE" in k:
                new_d[k] = v if isinstance(v, list) else [v]
            elif k not in ["page_num", "Filename"]:
                if isinstance(v, str) and "," in v:
                    new_d[k] = [item.strip() for item in v.split(",")]
                elif v == "N/A":
                    new_d[k] = []
                else:
                    new_d[k] = [v] if isinstance(v, str) else v
            else:
                new_d[k] = v
        td_clean.append(new_d)
    return td_clean


def get_parent_agreement_code(filename):
    try:
        filename = filename.split(".txt")[0]
        match = re.search(r"([^\sa-zA-Z]+)(?=\.\w+$|$)", filename)
        end = [i for i in match.group(1).split("_") if i]
        return end[0]
    except:
        return "N/A"


def consolidate_subheader(dict_list):
    modified_list = []
    for d in dict_list:
        if "FULL_SERVICE" in d and "SUBHEADER" in d:
            if d["SUBHEADER"] != "N/A":
                d["FULL_SERVICE"] = d["SUBHEADER"] + " - " + d["FULL_SERVICE"]
            # Remove the 'SUBHEADER' key
            del d["SUBHEADER"]
        modified_list.append(d)
    return modified_list


def clean_msr_lesser(df, ls=valid.VALID_MSR):
    pattern = "|".join(re.escape(item) for item in ls)
    mask = df["FULL_SERVICE"].str.contains(pattern, case=False, na=False)
    target_rows = df[mask]

    # Iterate over these rows
    for index, row in target_rows.iterrows():
        # Get all rows with the same 'EXHIBIT' value
        exhibit_rows = df[df["EXHIBIT"] == row["EXHIBIT"]]

        # Check the 'LESSER' values of these rows
        if (exhibit_rows["LESSER"] == "Y").any():
            # If any row has 'LESSER' == 'Y', set the 'LESSER' value of the original row to 'Y'
            df.at[index, "LESSER"] = "Y"

            # Move first LESSER_RATE where LESSER==Y to MSR row
            lesser_rate = exhibit_rows[exhibit_rows["LESSER"] == "Y"][
                "LESSER_RATE"
            ].iloc[0]
            df.at[index, "LESSER_RATE"] = lesser_rate
    return df


def clean_default_term(df, ls=valid.INVALID_DEFAULT):
    # Create a case-insensitive regex pattern that matches any of the values in ls
    pattern = "|".join(re.escape(term) for term in ls)

    # Find rows where the DEFAULT_TERM column contains any of the terms from ls
    df["DEFAULT_TERM"] = df["DEFAULT_TERM"].astype(str)
    mask = df["DEFAULT_TERM"].str.contains(pattern, case=False, na=False)

    # Replace these values with 'N/A'
    df.loc[mask, "DEFAULT_TERM"] = "N/A"
    df.loc[mask, "DEFAULT_RATE"] = "N/A"

    return df


def clean_lesser_rate(df):
    for index, row in df.iterrows():
        lesser = str(row["LESSER"]) if not pd.isna(row["LESSER"]) else ""
        lesser_rate = row["LESSER_RATE"]
        rate_standard = row["RATE_STANDARD"]

        if "Y" in lesser and is_empty(lesser_rate) and not is_empty(rate_standard):
            df.at[index, "LESSER_RATE"] = row["RATE_STANDARD"]
            df.at[index, "RATE_STANDARD"] = "N/A"
    return df


def clean_prov_2(df):
    valid_types = valid.select_valid_prov_2(df["PROV_TYPE"])
    target_rows = df[df["PROV_TYPE_LEVEL_2"].apply(is_empty)]

    def find_exact_match(text):
        if pd.isna(text) or text == "":
            return None

        words = re.findall(r"\b[\w/]+(?:[-\s][\w/]+)*\b", text)
        for i in range(len(words)):
            for j in range(i + 1, len(words) + 1):
                phrase = " ".join(words[i:j])
                if phrase in valid_types:  # Case-sensitive matching
                    return phrase
        return None

    for index, row in target_rows.iterrows():
        match = None

        if not is_empty(row["FULL_SERVICE"]):
            match = find_exact_match(str(row["FULL_SERVICE"]))
            if match:
                logging.info(f"Row {index}: Matched in FULL_SERVICE: {match}")
                df.at[index, "PROV_TYPE_LEVEL_2"] = match
                continue

        if not is_empty(row["EXHIBIT"]):
            match = find_exact_match(str(row["EXHIBIT"]))
            if match:
                logging.info(f"Row {index}: Matched in EXHIBIT: {match}")
                df.at[index, "PROV_TYPE_LEVEL_2"] = match
                continue

        exhibit_rows = df[df["EXHIBIT"] == row["EXHIBIT"]]
        if not exhibit_rows.empty:
            for _, exhibit_row in exhibit_rows.iterrows():
                if not is_empty(exhibit_row["PROV_TYPE_LEVEL_2"]):
                    match = find_exact_match(str(exhibit_row["PROV_TYPE_LEVEL_2"]))
                    if match:
                        logging.info(
                            f"Row {index}: Matched in other row's PROV_TYPE_LEVEL_2: {match}"
                        )
                        df.at[index, "PROV_TYPE_LEVEL_2"] = match
                        break
                elif not is_empty(exhibit_row["FULL_SERVICE"]):
                    match = find_exact_match(str(exhibit_row["FULL_SERVICE"]))
                    if match:
                        logging.info(
                            f"Row {index}: Matched in other row's FULL_SERVICE: {match}"
                        )
                        df.at[index, "PROV_TYPE_LEVEL_2"] = match
                        break

        if match is None:
            logging.warning(f"Row {index}: No valid match found")

    # Final check to ensure no invalid values were assigned
    invalid_assignments = df[
        (~df["PROV_TYPE_LEVEL_2"].isin(valid_types))
        & (~df["PROV_TYPE_LEVEL_2"].apply(is_empty))
    ]
    if not invalid_assignments.empty:
        logging.warning("Invalid assignments found:")
        for idx, row in invalid_assignments.iterrows():
            logging.warning(
                f"Row {idx}: Invalid PROV_TYPE_LEVEL_2: {row['PROV_TYPE_LEVEL_2']}"
            )
            logging.warning(f"    FULL_SERVICE: {row['FULL_SERVICE']}")
            logging.warning(f"    EXHIBIT: {row['EXHIBIT']}")

        df.loc[invalid_assignments.index, "PROV_TYPE_LEVEL_2"] = ""

    return df


def clean_ac_fields(final_df):
    # additional post-processing
    final_df = final_df.fillna("")
    # final_df.loc[final_df['CREDENTIALING_APP_IND'] != 'Yes', 'CREDENTIALING_APP_IND'] = "No"
    # final_df.loc[final_df['AFFILIATION_CLAUSE_IND'] != 'Yes', 'AFFILIATION_CLAUSE_IND'] = "No"
    # final_df.loc[final_df['ASSIGNMENTS_CLAUSE_IND'] != 'Yes', 'ASSIGNMENTS_CLAUSE_IND'] = "No"
    if "TERM_CLAUSE" in final_df:
        final_df.loc[
            final_df["TERM_CLAUSE"].str.startswith("IL-4 Termination"), "TERM_CLAUSE"
        ] = "N/A"
        final_df.loc[
            final_df["TERM_CLAUSE"].str.startswith("bonus payment shall be effective"),
            "TERM_CLAUSE",
        ] = "N/A"
        final_df.loc[
            final_df["TERM_CLAUSE"].str.contains("does not contain"), "TERM_CLAUSE"
        ] = "N/A"
        final_df.loc[
            final_df["TERM_CLAUSE"].str.contains("No term or termination"),
            "TERM_CLAUSE",
        ] = "N/A"
    # if 'CONTRACT_SIGNATORY_IND' in final_df and 'PROV_PARTICIPATION_STATUS' in final_df:
    #     final_df.loc[final_df['CONTRACT_SIGNATORY_IND'] == 'Yes', 'PROV_PARTICIPATION_STATUS'] = "Yes"
    if (
        "CONTRACT_AUTO_RENEWAL_IND" in final_df
        and "CONTRACT_TERMINATION_DT" in final_df
    ):
        final_df.loc[
            final_df["CONTRACT_AUTO_RENEWAL_IND"] == "Yes", "CONTRACT_TERMINATION_DT"
        ] = np.nan

    # check if NPI has 10 digits
    if "PROV_GROUP_NPI" in final_df:
        final_df["PROV_GROUP_NPI"] = final_df["PROV_GROUP_NPI"].map(
            lambda x: x if sum(c.isdigit() for c in str(x) + " ") == 10 else ""
        )
    # NETWORK_ACCESS_FEES_IND
    if "NETWORK_ACCESS_FEES_IND" in final_df:
        final_df.loc[
            ~final_df["NETWORK_ACCESS_FEES_IND"].isin(["N/A", "No", "", " "]),
            "NETWORK_ACCESS_FEES_IND",
        ] = "Yes"

    if "PAYER_NAME" in final_df and "HEALTH_PLAN_STATE" in final_df:
        final_df.loc[
            final_df["PAYER_NAME"].str.startswith("Illini"), "HEALTH_PLAN_STATE"
        ] = "Illinois"
    if "HEALTH_PLAN_STATE" in final_df:
        final_df["HEALTH_PLAN_STATE"] = (
            final_df["HEALTH_PLAN_STATE"]
            .map(valid.STATE_MAP)
            .fillna(final_df["HEALTH_PLAN_STATE"])
        )

    if "NOTICE_PROVIDER_NAME" in final_df and "NOTICE_PROVIDER_ADDRESS" in final_df:
        final_df.loc[
            final_df["NOTICE_PROVIDER_NAME"].str.contains(
                "Superior HealthPlan", na=False
            ),
            "NOTICE_PROVIDER_NAME",
        ] = np.nan
        final_df.loc[
            final_df["NOTICE_PROVIDER_NAME"].str.contains(
                "Superior HealthPlan", na=False
            ),
            "NOTICE_PROVIDER_ADDRESS",
        ] = np.nan

    if "Contract Name" in final_df.columns:
        final_df["temp_filename"] = (
            final_df["Contract Name"].str[:10].str.replace("-", "")
        )
        if "PROV_GROUP_TIN" in final_df and "Contract Name" in final_df:
            final_df.loc[
                (final_df["PROV_GROUP_TIN"].isin([" ", ""]))
                & (final_df["temp_filename"].str.isnumeric()),
                "PROV_GROUP_TIN",
            ] = final_df["temp_filename"]
        final_df.drop(columns=["temp_filename"], inplace=True)
    elif "Filename" in final_df.columns:
        final_df["temp_filename"] = final_df["Filename"].str[:10].str.replace("-", "")
        if "PROV_GROUP_TIN" in final_df and "Filename" in final_df:
            final_df.loc[
                (final_df["PROV_GROUP_TIN"].isin([" ", ""]))
                & (final_df["temp_filename"].str.isnumeric()),
                "PROV_GROUP_TIN",
            ] = final_df["temp_filename"]
        final_df.drop(columns=["temp_filename"], inplace=True)

    # For POLICIES_AND_PROCEDURES, filter out anything without either "policies" or "procedures".
    if "POLICIES_AND_PROCEDURES" in final_df:
        final_df.loc[
            ~final_df["POLICIES_AND_PROCEDURES"].str.contains(
                "polic", flags=re.IGNORECASE
            )
            & ~final_df["POLICIES_AND_PROCEDURES"].str.contains(
                "procedure", flags=re.IGNORECASE
            ),
            "POLICIES_AND_PROCEDURES",
        ] = "N/A"

    final_df = final_df.apply(lambda x: x.map(replace_null_terms))
    final_df = final_df.apply(lambda x: x.map(replace_quotes))

    return final_df


def replace_quotes(value):
    try:
        return str(value).replace(r"\"", '"')
    except:
        return value


def replace_null_terms(value):
    # Convert value to string and check if any term from NULL_ANSWER_TERMS is in the value
    if any(term.lower() in str(value).lower() for term in valid.NULL_ANSWER_TERMS):
        return "N/A"
    return value


def filter_dict(d, pattern):
    final_dict = {}
    for page_num, answer in d.items():
        if not pattern.search(answer):
            final_dict[page_num] = answer
    return final_dict


def filter_add_ons(df):
    pattern = r"(in no event).+(includ.?)|(forward).+(payments)"

    mask = df["ADD_ON_REIMBURSEMENT_LANGUAGE"].str.contains(
        pattern, case=False, na=False, regex=True
    )
    df.loc[mask, "ADD_ON_REIMBURSEMENT_LANGUAGE"] = "N/A"
    df.loc[mask, "ADD_ON_REIMBURSEMENT_IND"] = "N"

    def check_add_on(group):
        # Check if any 'SERVICE_TYPE' contains 'Add-on' or 'Add On'
        if (
            group["FULL_SERVICE"]
            .str.contains("Add-on|Add On", regex=True, case=False, na=False)
            .any()
        ):
            group["ADD_ON_REIMBURSEMENT_LANGUAGE"] = (
                "N/A"  # Set 'ADD_ON' to 'N/A' for the whole group
            )
            group["ADD_ON_REIMBURSEMENT_IND"] = "N"
        return group

    # Group by 'Contract Name' and 'EXHIBIT', then apply the check_add_on function
    if "Contract Name" in df.columns:
        df = df.groupby(["Contract Name", "EXHIBIT"]).apply(check_add_on)
    elif "Filename" in df.columns:
        df = df.groupby(["Filename", "EXHIBIT"]).apply(check_add_on)

    # TODO : If original add on value is actually an Exclusion, and the Exclusion value for the row is invalid, then move the Add On value to the Exclusions column
    return df


def add_scmr(df):
    def count_dollar_values(s):
        return str(s).count("$") + str(s).count("%")

    if "FLAT_FEE_STANDARD" in df.columns:
        col = "FLAT_FEE_STANDARD"
    elif "FLAT_FEE" in df.columns:
        col = "FLAT_FEE"
    elif "FLAT FEE" in df.columns:
        col = "FLAT FEE"

    df["dollar_count"] = df[col].apply(count_dollar_values)

    df["Single Code Multiple Rates (Language)"] = df.apply(
        lambda row: row[col] if row["dollar_count"] > 1 else "N/A", axis=1
    )

    df["Single Code Multiple Rates (Y/N)"] = df["dollar_count"].apply(
        lambda x: "Y" if x > 1 else "N"
    )
    df.drop("dollar_count", axis=1, inplace=True)

    return df


def clean_lob(df, filename):
    def update_contract_lob(row):
        try:
            # Check if more than one valid lob is present in the 'FULL_SERVICE' column
            if (
                sum(
                    val.lower() in row["FULL_SERVICE"].lower()
                    for val in valid.VALID_LOBS
                )
                > 1
            ):
                prompt = prompts.LOB_SWEEPER(row["EXHIBIT"])
                answer = claude_funcs.invoke_claude(
                    prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, 256
                )
                return answer  # Return the value from 'EXHIBIT' if condition is met
            return row["CONTRACT_LOB"]
        except:
            return row["CONTRACT_LOB"]

    df["CONTRACT_LOB"] = df.apply(update_contract_lob, axis=1)
    return df


def derive_indicators(results):

    for field in DERIVED_INDICATOR_FIELDS:
        indicator_field = field + "_IND"
        if field in results and results[field] and not is_empty(results[field]):
            print(f"Y - {indicator_field}")
            results[indicator_field] = "Y"
        else:
            print(f"N - {indicator_field}")
            results[indicator_field] = "N"
    return results