Alex Galarce
5b88185611
[daip2-9] basic code refactor - unreferenced functions and reorganize existing files * remove unreferenced functions in ac_funcs.py * move compare_output.py * move file_counting.py * move merge_issue_regex.py * remove unreferenced functions in postprocessing_funcs.py * clean up preprocessing_funcs.py * move prerun.py * move table_analysis.py and textract_template.py to scripts * remove unreferenced functions in table_funcs.py * moved tin_pull.py to scripts * move detect_complex.py and remove merge_funcs.py * clean up dict operations and utils, mark functions in utils for destinations * removed errant import Approved-by: Katon Minhas
645 lines
21 KiB
Python
645 lines
21 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
|
|
from hotfix_helper_funcs import state_check
|
|
|
|
import logging
|
|
import qa_qc_helpers
|
|
|
|
logging.basicConfig(
|
|
level=logging.ERROR, format="%(asctime)s - %(levelname)s - %(message)s"
|
|
)
|
|
|
|
|
|
def filter_service_column(answer_dicts):
|
|
"""
|
|
Filters the answer_dicts based on the presence of specific keywords.
|
|
"""
|
|
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
|
|
] # an error is happening here. This function needs a docstring.
|
|
return end[0]
|
|
except Exception as e:
|
|
print(f"Error in `get_parent_agreement_code`: {e}; returning `N/A`")
|
|
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):
|
|
"""this function is to ensure the following relation between term, auto-renew and termination date is correct:
|
|
- automatically renew|coterminous cases are marked as autorenewal Yes
|
|
- null term clause should have autorenewal as N and termination date as blank
|
|
- autorenewal Y should have termination date blank
|
|
- If autorenewal is not Y it should be N. there should not be any blank value.
|
|
"""
|
|
if "TERM_CLAUSE" in final_df and "CONTRACT_AUTO_RENEWAL_IND" in final_df:
|
|
final_df.loc[
|
|
final_df["TERM_CLAUSE"].str.contains("automatically renew|coterminous"),
|
|
"CONTRACT_AUTO_RENEWAL_IND",
|
|
] = "Y"
|
|
final_df.loc[
|
|
(final_df["TERM_CLAUSE"].isin(["N/A", "", " "]))
|
|
| (final_df["TERM_CLAUSE"].isna()),
|
|
"CONTRACT_AUTO_RENEWAL_IND",
|
|
] = "N"
|
|
if "TERM_CLAUSE" in final_df and "CONTRACT_TERMINATION_DT" in final_df:
|
|
final_df.loc[
|
|
(final_df["TERM_CLAUSE"].isin(["N/A", "", " "]))
|
|
| (final_df["TERM_CLAUSE"].isna()),
|
|
"CONTRACT_TERMINATION_DT",
|
|
] = np.nan
|
|
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
|
|
if "CONTRACT_AUTO_RENEWAL_IND" in final_df:
|
|
final_df.loc[
|
|
~(final_df["CONTRACT_AUTO_RENEWAL_IND"].isin(["Yes", "Y"])),
|
|
"CONTRACT_AUTO_RENEWAL_IND",
|
|
] = "N"
|
|
else:
|
|
final_df["CONTRACT_AUTO_RENEWAL_IND"] = "N"
|
|
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_health_plan_state_from_hotfix(
|
|
df: pd.DataFrame,
|
|
contract_name: str,
|
|
text_dict: dict[str, str],
|
|
) -> pd.DataFrame:
|
|
|
|
if not isinstance(df, pd.DataFrame):
|
|
# raise TypeError(f"Expected a dataframe, got {type(df).__name__}")
|
|
return df
|
|
|
|
if contract_name is None:
|
|
# raise ValueError(
|
|
# f"Expected 'contract_name' parameter, got {contract_name}"
|
|
# )
|
|
return df
|
|
|
|
if "HEALTH_PLAN_STATE" not in df.columns:
|
|
# raise KeyError(
|
|
# f"Column 'Health Plan State' not found in DataFrame. Available columns are: {list(df.columns)}"
|
|
# )
|
|
return df
|
|
|
|
health_plan_state = df["HEALTH_PLAN_STATE"].unique().tolist()[0]
|
|
|
|
answer = state_check(health_plan_state, text_dict)
|
|
|
|
df["HEALTH_PLAN_STATE"] = answer
|
|
|
|
return 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
|
|
|
|
|
|
def icm_number_fix(df: pd.DataFrame) -> pd.DataFrame:
|
|
|
|
def clean_icm(payer_name, parent_agreement_code):
|
|
|
|
pattern = r"\d{4,6}"
|
|
|
|
if (
|
|
(
|
|
re.search(pattern, payer_name.split("ICMProviderAgreementAmendment")[0])
|
|
is not None
|
|
)
|
|
and (is_empty(parent_agreement_code))
|
|
and ("ICMProviderAgreementAmendment" in payer_name)
|
|
):
|
|
|
|
parent_agreement_code = re.search(
|
|
pattern, payer_name.split("ICMProviderAgreementAmendment")[0]
|
|
).group(0)
|
|
|
|
return parent_agreement_code
|
|
|
|
else:
|
|
|
|
return parent_agreement_code
|
|
|
|
df["Parent Agreement Code"] = df.apply(
|
|
lambda row: clean_icm(row["PAYER_NAME"], row["Parent Agreement Code"]), axis=1
|
|
)
|
|
|
|
return df
|
|
|
|
|
|
def methodology_short_fix(df: pd.DataFrame) -> pd.DataFrame:
|
|
|
|
df["RATE_SHORT"] = df["RATE_SHORT"].apply(
|
|
lambda x: re.sub(r"^\$?\d*\.?\d*", "", x) if not is_empty(x) else x
|
|
)
|
|
|
|
return df
|
|
|
|
|
|
def yn_fixes(df: pd.DataFrame) -> pd.DataFrame:
|
|
|
|
for col in df.columns:
|
|
if "(Y/N)" in col:
|
|
df.loc[df[col] == "Yes", col] = "Y"
|
|
df.loc[df[col] == "No", col] = "N"
|
|
df.loc[(df[col] != "Y") | (df[col] != "N"), col] = "N"
|
|
|
|
for yn in valid.AC_IND_LANG_COLS.keys():
|
|
if yn in df.columns and valid.AC_IND_LANG_COLS[yn] in df.columns:
|
|
lang = valid.AC_IND_LANG_COLS[yn]
|
|
df.loc[df[yn] != "Y", lang] = ""
|
|
|
|
for yn in valid.B_IND_LANG_COLS.keys():
|
|
if yn in df.columns and valid.B_IND_LANG_COLS[yn] in df.columns:
|
|
lang = valid.B_IND_LANG_COLS[yn]
|
|
df.loc[df[yn] != "Y", lang] = ""
|
|
|
|
return df
|