Mayank Aamseek
df519ea32b
term group fixed * term group fixed * Merged main into bugfix/group_chunk_json_parsing * file_processing.py edited online with Bitbucket * Merged main into bugfix/group_chunk_json_parsing Approved-by: Alex Galarce
580 lines
19 KiB
Python
580 lines
19 KiB
Python
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]
|
|
|
|
filename = re.sub(r"\([^)]*\)", "", filename)
|
|
|
|
match = re.search(r"([^\sa-zA-Z]+)(?=\.\w+$|$)", filename)
|
|
end = [i for i in match.group(1).split("_") if i]
|
|
return end[0]
|
|
except Exception as e:
|
|
print(f"Error: {e}")
|
|
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):
|
|
# Remove any invalid Defaults
|
|
pattern = "|".join(re.escape(term) for term in valid.INVALID_DEFAULT)
|
|
df["DEFAULT_TERM"] = df["DEFAULT_TERM"].astype(str)
|
|
mask = df["DEFAULT_TERM"].str.contains(pattern, case=False, na=False)
|
|
df.loc[mask, "DEFAULT_TERM"] = "N/A"
|
|
df.loc[mask, "DEFAULT_RATE"] = "N/A"
|
|
|
|
# Replace RATE with acronyms (AC, BC, etc)
|
|
df["DEFAULT_RATE"] = df["DEFAULT_RATE"].replace(valid.RATE_REPLACEMENTS, regex=True)
|
|
|
|
# Ensure Outpatient, Inpatient, etc match up
|
|
def replace_terms(row):
|
|
if "outpatient" in row["DEFAULT_TERM"].lower() and row["IP_OP"] != "OP":
|
|
row["DEFAULT_TERM"] = "N/A"
|
|
row["DEFAULT_RATE"] = "N/A"
|
|
elif "inpatient" in row["DEFAULT_TERM"].lower() and row["IP_OP"] != "IP":
|
|
row["DEFAULT_TERM"] = "N/A"
|
|
row["DEFAULT_RATE"] = "N/A"
|
|
return row
|
|
|
|
df = df.apply(replace_terms, axis=1)
|
|
|
|
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_term_clause(final_df):
|
|
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"
|
|
return final_df
|
|
|
|
|
|
def clean_auto_renewal_ind(final_df):
|
|
if (
|
|
"CONTRACT_AUTO_RENEWAL_IND" in final_df
|
|
and "CONTRACT_TERMINATION_DT" in final_df
|
|
):
|
|
final_df.loc[
|
|
final_df["CONTRACT_AUTO_RENEWAL_IND"].isin(["Yes", "Y"]), "CONTRACT_TERMINATION_DT"
|
|
] = np.nan
|
|
return final_df
|
|
|
|
|
|
def clean_npi(final_df):
|
|
if "PROV_GROUP_NPI" in final_df:
|
|
# Replace strings with a digit count not equal to 10 with "N/A"
|
|
final_df["PROV_GROUP_NPI"] = final_df["PROV_GROUP_NPI"].apply(
|
|
lambda x: (
|
|
f"N/A - (model detected: {x})"
|
|
if (len("".join(filter(str.isdigit, x))) != 10 and x != "N/A")
|
|
else x
|
|
)
|
|
)
|
|
return final_df
|
|
|
|
|
|
def clean_network_access_fees(final_df):
|
|
if "NETWORK_ACCESS_FEES_IND" in final_df:
|
|
final_df.loc[
|
|
not is_empty(final_df["NETWORK_ACCESS_FEES_IND"]),
|
|
"NETWORK_ACCESS_FEES_IND",
|
|
] = "Yes"
|
|
return final_df
|
|
|
|
|
|
def clean_health_plan_state(final_df):
|
|
|
|
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"].str.upper()
|
|
|
|
final_df["HEALTH_PLAN_STATE"] = (
|
|
final_df["HEALTH_PLAN_STATE"]
|
|
.map(valid.STATE_MAP)
|
|
.fillna(final_df["HEALTH_PLAN_STATE"])
|
|
)
|
|
|
|
final_df["HEALTH_PLAN_STATE"] = final_df["HEALTH_PLAN_STATE"].str.title()
|
|
return final_df
|
|
|
|
|
|
def clean_notice_provider_name_and_address(final_df):
|
|
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
|
|
return final_df
|
|
|
|
|
|
def get_tin_from_filename(final_df):
|
|
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)
|
|
|
|
return final_df
|
|
|
|
|
|
def clean_policies_and_procedures(final_df):
|
|
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"
|
|
return final_df
|
|
|
|
|
|
def clean_tin(final_df):
|
|
if "PROV_GROUP_TIN" in final_df:
|
|
# Replace SSN-like strings with "N/A"
|
|
final_df["PROV_GROUP_TIN"] = final_df["PROV_GROUP_TIN"].replace(
|
|
r"\b\d{3}-\d{2}-\d{4}\b", "N/A", regex=True
|
|
)
|
|
|
|
# Function to format and validate TIN numbers
|
|
def format_tin(x):
|
|
digits = "".join(filter(str.isdigit, x))
|
|
if len(digits) == 9:
|
|
# Format as '##-#######'
|
|
return f"{digits[:2]}-{digits[2:]}"
|
|
else:
|
|
# Retain prior handling for invalid entries
|
|
if x != "N/A":
|
|
return f"N/A - (model detected: {x})"
|
|
else:
|
|
return "N/A"
|
|
|
|
# Apply the formatting function to the 'PROV_GROUP_TIN' column
|
|
final_df["PROV_GROUP_TIN"] = final_df["PROV_GROUP_TIN"].apply(format_tin)
|
|
|
|
return final_df
|
|
|
|
|
|
def clean_tin_npi_other(final_df):
|
|
if "PROV_TIN_OTHER" in final_df:
|
|
final_df["PROV_TIN_OTHER"] = (
|
|
final_df["PROV_TIN_OTHER"]
|
|
.str.replace("[", "", regex=False)
|
|
.str.replace("]", "", regex=False)
|
|
)
|
|
if "PROV_NPI_OTHER" in final_df:
|
|
final_df["PROV_NPI_OTHER"] = (
|
|
final_df["PROV_NPI_OTHER"]
|
|
.str.replace("[", "", regex=False)
|
|
.str.replace("]", "", regex=False)
|
|
)
|
|
|
|
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]):
|
|
results[indicator_field] = "Y"
|
|
else:
|
|
results[indicator_field] = "N"
|
|
return results
|