Katon Minhas
854 lines
28 KiB
Python
854 lines
28 KiB
Python
import ast
|
|
import hashlib
|
|
import logging
|
|
import os
|
|
import re
|
|
from datetime import datetime
|
|
import pandas as pd
|
|
from src.utils import string_utils
|
|
|
|
# Determine the base directory for the project
|
|
BASE_DIR = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
# Path to the mappings folder
|
|
MAPPINGS_DIR = os.path.join(BASE_DIR, "crosswalk", "mappings")
|
|
|
|
|
|
def rename_columns(df):
|
|
"""
|
|
Use the provided mapping to rename columns in the data
|
|
|
|
Args:
|
|
df (pd.DataFrame): The input DataFrame with columns to be renamed.
|
|
Returns:
|
|
pd.DataFrame: The DataFrame with renamed columns.
|
|
"""
|
|
# Check if the DataFrame is empty, return it as is if it is
|
|
if df.empty:
|
|
return df
|
|
|
|
# If not empty, proceed with column renaming
|
|
rename_map = {
|
|
"PROCEDURE_CD": "CPT4_PROC_CD",
|
|
"PROCEDURE_CD_DESC": "CPT4_PROC_CD_DESC",
|
|
}
|
|
# Rename columns using the mapping
|
|
df = df.rename(columns=rename_map)
|
|
return df
|
|
|
|
|
|
def normalize_indicator_field(value: str) -> str:
|
|
"""
|
|
Normalize the indicator field value to 'Y' or 'N'.
|
|
|
|
Args:
|
|
value (str): The input value from an _IND field.
|
|
Returns:
|
|
str: 'Y' if the value is 'Y', otherwise 'N'.
|
|
"""
|
|
if isinstance(value, str) and value.strip().upper() == "Y":
|
|
return "Y"
|
|
return "N"
|
|
|
|
|
|
def format_rate_fields_with_commas(value: str) -> str:
|
|
"""
|
|
Format the rate with commas and two decimal places.
|
|
|
|
Args:
|
|
value (str): A string representing a numeric value.
|
|
Returns:
|
|
str: The formatted rate string.
|
|
"""
|
|
if string_utils.is_empty(value):
|
|
return ""
|
|
|
|
try:
|
|
numeric_value = float(value)
|
|
return f"{numeric_value:,.2f}"
|
|
except ValueError:
|
|
return ""
|
|
|
|
|
|
def remove_hyphens(value):
|
|
"""
|
|
Remove hyphens from the input value.
|
|
|
|
Args:
|
|
value : The input might be list or string
|
|
Returns:
|
|
str: The cleaned value with hyphens removed.
|
|
"""
|
|
if isinstance(value, str):
|
|
return value.replace("-", "")
|
|
elif isinstance(value, list):
|
|
return [v.replace("-", "") if isinstance(v, str) else v for v in value]
|
|
return value
|
|
|
|
|
|
def flatten_singleton_string_list(list_str: str) -> str:
|
|
"""Take in a list within a string, and if it has a single element, return that element.
|
|
If it has multiple elements, return the list as a string.
|
|
|
|
Args:
|
|
list_str (str): A string representation of a list.
|
|
|
|
Returns:
|
|
str: The single element if the list has one element, otherwise the list as a string.
|
|
"""
|
|
if string_utils.is_empty(list_str):
|
|
return ""
|
|
try:
|
|
# Convert the string to a list using ast.literal_eval
|
|
list_obj = ast.literal_eval(list_str)
|
|
# Check if list_obj is actually a list
|
|
if not isinstance(list_obj, list):
|
|
return list_str
|
|
# If the list has one element, return that element
|
|
if len(list_obj) == 1:
|
|
return list_obj[0]
|
|
# Otherwise, return the list as a string
|
|
return ", ".join(list_obj)
|
|
except (ValueError, SyntaxError):
|
|
return list_str
|
|
|
|
|
|
def validate_and_reformat_date(date_str: str) -> str:
|
|
"""
|
|
Validate if a string is in YYYY/MM/DD format or reformat it to YYYY/MM/DD if possible.
|
|
|
|
Args:
|
|
date_str (str): The input date string.
|
|
|
|
Returns:
|
|
str: The reformatted date string if valid or reformatted, otherwise the original string.
|
|
"""
|
|
if not isinstance(date_str, str):
|
|
return date_str
|
|
|
|
try:
|
|
# Check if the date is already in YYYY/MM/DD format
|
|
datetime.strptime(date_str, "%Y/%m/%d")
|
|
return date_str
|
|
except ValueError:
|
|
# Attempt to reformat the date from known formats
|
|
known_formats = ["%Y-%m-%d", "%m/%d/%Y", "%d-%b-%Y", "%d/%m/%Y"]
|
|
for fmt in known_formats:
|
|
try:
|
|
return datetime.strptime(date_str, fmt).strftime("%Y/%m/%d")
|
|
except ValueError:
|
|
continue
|
|
return date_str
|
|
|
|
|
|
def normalize_auto_renewal_term(text: str) -> str:
|
|
"""
|
|
Normalize the auto-renewal term to a standard format.
|
|
|
|
Args:
|
|
text (str): The input auto-renewal term.
|
|
|
|
Returns:
|
|
str: The normalized auto-renewal term.
|
|
"""
|
|
if not text or not isinstance(text, str):
|
|
return ""
|
|
|
|
# Convert to lowercase and strip whitespace
|
|
text = text.lower().strip()
|
|
|
|
# Remove numbers enclosed in parentheses (e.g., "(1) ", "(12) ")
|
|
text = re.sub(r"\(\d+\) ", "", text)
|
|
|
|
# Define replacements
|
|
replacements = {
|
|
r"\bone\b": "1",
|
|
r"\btwo\b": "2",
|
|
r"\bthree\b": "3",
|
|
r"\btwelve\b": "12",
|
|
r"year to year": "1 year",
|
|
r"12 months": "1 year",
|
|
r"\btwelve months\b": "1 year",
|
|
r"\bone year\b": "1 year",
|
|
r"\bone-year\b": "1 year",
|
|
r"\b1 year\b": "1 year",
|
|
r"\bmonth to month\b": "1 month",
|
|
}
|
|
|
|
# Apply replacements
|
|
for pattern, replacement in replacements.items():
|
|
text = re.sub(pattern, replacement, text)
|
|
|
|
# Remove any remaining parentheses or extra spaces
|
|
text = text.strip().replace("(", "").replace(")", "")
|
|
|
|
text = text.replace("-", " ")
|
|
if text == "year":
|
|
text = "1 year"
|
|
|
|
return text
|
|
|
|
|
|
def normalize_cpt_fields(value):
|
|
"""
|
|
Normalize code values to a list format.
|
|
|
|
Args:
|
|
value (Any): Input value which could be a list, string, or other types.
|
|
|
|
Returns:
|
|
list: A normalized list of string codes.
|
|
"""
|
|
if not value or (isinstance(value, float) and pd.isna(value)): # Handle NaN values
|
|
return "" # Empty string
|
|
|
|
# First normalize to a list
|
|
if isinstance(value, (list, tuple)): # If already a list or tuple
|
|
result = [str(v).strip() for v in value]
|
|
elif isinstance(value, str): # If it's a string
|
|
value = value.strip()
|
|
if value.startswith("[") and value.endswith(
|
|
"]"
|
|
): # String representation of a list
|
|
try:
|
|
# Parse it first
|
|
parsed_list = ast.literal_eval(value)
|
|
result = []
|
|
# Process each item in the parsed list
|
|
for item in parsed_list:
|
|
item_str = str(item).strip()
|
|
# If the item contains commas, split it into multiple items
|
|
if "," in item_str:
|
|
result.extend([str(v).strip() for v in item_str.split(",")])
|
|
else:
|
|
result.append(item_str)
|
|
except (ValueError, SyntaxError):
|
|
result = [value] # If parsing fails, treat it as a single value
|
|
elif "," in value: # If it's a comma-separated string
|
|
# Split by comma and create a list
|
|
result = [str(v).strip() for v in value.split(",")]
|
|
elif "-" in value:
|
|
# Range format
|
|
result = [value]
|
|
else: # Single value
|
|
result = [value]
|
|
elif isinstance(value, (float, int)): # If it's a number
|
|
result = [str(int(value)).strip()]
|
|
else:
|
|
result = [str(value).strip()] # Default case for other types
|
|
|
|
return str(result) # Return the list directly
|
|
|
|
|
|
def generate_reimb_ids(df: pd.DataFrame) -> pd.DataFrame:
|
|
"""Generates REIMB_ID and REIMB_LESSER_OF_ID for each row in the DataFrame based on
|
|
the FILE_NAME and other fields.
|
|
The REIMB_ID is constructed using the following format:
|
|
{FILE_NAME}_exh_pg_{EXHIBIT_PAGE}_{index}_{hash_value}
|
|
|
|
the REIMB_LESSER_OF_ID is just the {hash_value} itself.
|
|
|
|
Where:
|
|
- FILE_NAME is the name of the file (without extension).
|
|
- EXHIBIT_PAGE is the page number of the exhibit.
|
|
- index is the index of the row in the group.
|
|
- hash_value is the first 8 characters of the MD5 hash of a concatenated string of relevant fields.
|
|
The relevant fields are SERVICE_TERM, REIMB_TERM, PROGRAM, PRODUCT, NETWORK, and LOB.
|
|
|
|
Args:
|
|
df (pd.DataFrame): Input DataFrame containing the columns FILE_NAME, EXHIBIT_PAGE, SERVICE_TERM, REIMB_TERM, PROGRAM, PRODUCT, NETWORK, and LOB.
|
|
The FILE_NAME column should contain the name of the file (with or without extension).
|
|
The EXHIBIT_PAGE column should contain the page number of the exhibit.
|
|
|
|
Returns:
|
|
pd.DataFrame: DataFrame with generated REIMB_IDs and REIMB_LESSER_OF_IDs for each row.
|
|
"""
|
|
# First sort to ensure consistent ordering
|
|
df = (
|
|
df.sort_values(by=["FILE_NAME", "EXHIBIT_PAGE"])
|
|
if "EXHIBIT_PAGE" in df.columns
|
|
else df.sort_values(by=["FILE_NAME"])
|
|
)
|
|
|
|
# Create a temporary copy to avoid SettingWithCopyWarning
|
|
df_temp = df.copy()
|
|
|
|
# Process each file+exhibit_page group to reset counter for each exhibit page
|
|
group_cols = (
|
|
["FILE_NAME", "EXHIBIT_PAGE"] if "EXHIBIT_PAGE" in df.columns else ["FILE_NAME"]
|
|
)
|
|
for group_key, indices in df.groupby(group_cols).groups.items():
|
|
if isinstance(group_key, tuple):
|
|
filename, exhibit_page = group_key
|
|
exhibit_page = str(exhibit_page).zfill(3)
|
|
else:
|
|
filename = group_key
|
|
exhibit_page = "000"
|
|
clean_filename = str(filename).split("/")[-1].replace(".", "_")
|
|
|
|
for i, idx in enumerate(indices):
|
|
row = df.loc[idx]
|
|
|
|
# Get fields to include in hash
|
|
service_term = str(row.get("SERVICE_TERM", ""))
|
|
reimb_term = str(row.get("REIMB_TERM", ""))
|
|
program = str(row.get("PROGRAM", ""))
|
|
product = str(row.get("PRODUCT", ""))
|
|
network = str(row.get("NETWORK", ""))
|
|
lob = str(row.get("LOB", ""))
|
|
|
|
# Create hash string
|
|
hash_string = f"{clean_filename}_{service_term}_{reimb_term}_{program}_{product}_{network}_{lob}"
|
|
hash_value = hashlib.md5(hash_string.encode()).hexdigest()[:8]
|
|
|
|
# Generate REIMB_ID and REIMB_LESSER_OF_ID
|
|
df_temp.at[idx, "REIMB_ID"] = (
|
|
f"{clean_filename}_exh_pg_{exhibit_page}_{i+1:03d}_{hash_value}"
|
|
)
|
|
df_temp.at[idx, "REIMB_LESSER_OF_ID"] = hash_value
|
|
|
|
return df_temp
|
|
|
|
|
|
def reorder_columns(df: pd.DataFrame, column_order: list[str]) -> pd.DataFrame:
|
|
"""
|
|
Reorders the columns of the DataFrame based on the given column order.
|
|
Steps:
|
|
1. Adds any missing columns from `column_order` to the DataFrame at once, filled with empty strings.
|
|
2. Reorders the columns of the DataFrame to match `column_order`.
|
|
3. Appends any columns in the DataFrame that are not in `column_order` to the end.
|
|
|
|
Args:
|
|
df (pd.DataFrame): The input DataFrame.
|
|
column_order (list[str]): The desired column order.
|
|
|
|
Returns:
|
|
pd.DataFrame: The reordered DataFrame.
|
|
"""
|
|
# Create a copy to avoid fragmentation
|
|
df_copy = df.copy()
|
|
|
|
# Find missing columns from column_order
|
|
missing_columns = [col for col in column_order if col not in df_copy.columns]
|
|
|
|
# Add all missing columns at once using a dictionary
|
|
if missing_columns:
|
|
# Create a dictionary of empty columns
|
|
empty_cols = {col: [""] * len(df_copy) for col in missing_columns}
|
|
# Add all missing columns at once with pd.concat
|
|
df_copy = pd.concat(
|
|
[df_copy, pd.DataFrame(empty_cols, index=df_copy.index)], axis=1
|
|
)
|
|
|
|
# Reorder columns to match column_order
|
|
final_column_order = [col for col in column_order if col in df_copy.columns]
|
|
|
|
# Return the DataFrame with reordered columns
|
|
return df_copy[final_column_order]
|
|
|
|
|
|
def auto_renewal(df: pd.DataFrame) -> pd.DataFrame:
|
|
# Normalize the 'AUTO_RENEWAL_TERM' column
|
|
if "AUTO_RENEWAL_TERM" in df.columns:
|
|
df["AUTO_RENEWAL_TERM"] = df["AUTO_RENEWAL_TERM"].apply(
|
|
normalize_auto_renewal_term
|
|
)
|
|
return df
|
|
|
|
|
|
def update_termination_date_for_conditions(df: pd.DataFrame) -> pd.DataFrame:
|
|
"""
|
|
Updates the 'AARETE_DERIVED_TERMINATION_DT' column based on specific conditions
|
|
involving 'TERMINATION_DT', 'AUTO_RENEWAL_IND', and 'AARETE_DERIVED_TERMINATION_DT'.
|
|
|
|
Args:
|
|
df (pd.DataFrame): The input DataFrame.
|
|
|
|
Returns:
|
|
pd.DataFrame: The updated DataFrame with modified 'AARETE_DERIVED_TERMINATION_DT' values.
|
|
"""
|
|
|
|
if (
|
|
"TERMINATION_DT" in df.columns
|
|
and "AUTO_RENEWAL_IND" in df.columns
|
|
and "AARETE_DERIVED_TERMINATION_DT" in df.columns
|
|
):
|
|
# Replace Y with 9999/12/31
|
|
df.loc[df["AUTO_RENEWAL_IND"] == "Y", "AARETE_DERIVED_TERMINATION_DT"] = (
|
|
"9999/12/31"
|
|
)
|
|
|
|
# Replace "N" and "year to year" with 9999/12/31
|
|
df.loc[
|
|
(df["TERMINATION_DT"] == "year to year")
|
|
& (df["AUTO_RENEWAL_IND"] == "N")
|
|
& (df["AARETE_DERIVED_TERMINATION_DT"] == "N/A"),
|
|
"AARETE_DERIVED_TERMINATION_DT",
|
|
] = "9999/12/31"
|
|
|
|
return df
|
|
|
|
|
|
def standardize_reimb_method_and_fee_schedule(
|
|
df: pd.DataFrame, VALID_UNIT_OF_MEASURE: list
|
|
) -> pd.DataFrame:
|
|
"""
|
|
Standardizes the 'AARETE_DERIVED_REIMB_METHOD' column and updates the
|
|
'AARETE_DERIVED_FEE_SCHEDULE' column based on specific values.
|
|
|
|
Args:
|
|
df (pd.DataFrame): The input DataFrame.
|
|
|
|
Returns:
|
|
pd.DataFrame: The updated DataFrame with standardized columns.
|
|
"""
|
|
if (
|
|
"AARETE_DERIVED_REIMB_METHOD" in df.columns
|
|
and "AARETE_DERIVED_FEE_SCHEDULE" in df.columns
|
|
):
|
|
# Standardize 'AARETE_DERIVED_REIMB_METHOD' values
|
|
# Copy AWP, ASP, WAC to AARETE_DERIVED_FEE_SCHEDULE column for the corresponding row
|
|
medrx_fee_schedule_values = ["AWP", "ASP", "WAC"]
|
|
df.loc[
|
|
df["AARETE_DERIVED_REIMB_METHOD"].isin(medrx_fee_schedule_values),
|
|
"AARETE_DERIVED_FEE_SCHEDULE",
|
|
] = df["AARETE_DERIVED_REIMB_METHOD"]
|
|
|
|
# Convert AARETE_DERIVED_REIMB_METHOD values [AWP, ASP, WAC] to 'MedRx'
|
|
df["AARETE_DERIVED_REIMB_METHOD"] = df["AARETE_DERIVED_REIMB_METHOD"].replace(
|
|
{"AWP": "MedRx", "ASP": "MedRx", "WAC": "MedRx"}
|
|
)
|
|
|
|
if "AARETE_DERIVED_REIMB_METHOD" in df.columns:
|
|
|
|
# Remove rows where 'AARETE_DERIVED_REIMB_METHOD' is 'DOFR' but 'REIMB_TERM' does not contain 'DOFR'
|
|
mask = (df["AARETE_DERIVED_REIMB_METHOD"] == "DOFR") & (
|
|
~df["REIMB_TERM"].str.contains("DOFR", na=False)
|
|
)
|
|
df = df[~mask]
|
|
|
|
# standardize the 'UNIT_OF_MEASURE' column
|
|
if "UNIT_OF_MEASURE" in df.columns:
|
|
df["UNIT_OF_MEASURE"] = df["UNIT_OF_MEASURE"].str.replace(
|
|
"Per Encounter", "Per Visit"
|
|
)
|
|
|
|
# remove cases of Per Member Per Month (PMPM)
|
|
df = df[~df["UNIT_OF_MEASURE"].isin(["Per Member Per Month (PMPM)"])]
|
|
|
|
# Enhanced logic: Set UNIT_OF_MEASURE as 'Per Unit' for Flat Rate reimbursement method with empty UNIT_OF_MEASURE
|
|
# This catches all edge cases including NaN, None, empty strings, etc.
|
|
if "AARETE_DERIVED_REIMB_METHOD" in df.columns:
|
|
flat_rate_mask = df["AARETE_DERIVED_REIMB_METHOD"] == "Flat Rate"
|
|
empty_uom_mask = (
|
|
df["UNIT_OF_MEASURE"].isna()
|
|
| df["UNIT_OF_MEASURE"].eq("")
|
|
| df["UNIT_OF_MEASURE"].apply(string_utils.is_empty)
|
|
)
|
|
df.loc[flat_rate_mask & empty_uom_mask, "UNIT_OF_MEASURE"] = "Per Unit"
|
|
|
|
# Ensure UNIT_OF_MEASURE is one of the valid values
|
|
df["UNIT_OF_MEASURE"] = df["UNIT_OF_MEASURE"].apply(
|
|
lambda x: x if x in VALID_UNIT_OF_MEASURE else ""
|
|
)
|
|
|
|
return df
|
|
|
|
|
|
def process_patient_age_range(df):
|
|
"""
|
|
Process the 'PATIENT_AGE_RANGE' column into 'PATIENT_AGE_MIN' and 'PATIENT_AGE_MAX'.
|
|
If a single number is provided, both min and max are set to that number.
|
|
Remove PATIENT_AGE_RANGE column.
|
|
|
|
Args:
|
|
df (pd.DataFrame): The input DataFrame.
|
|
|
|
Returns:
|
|
pd.DataFrame: The updated DataFrame with 'PATIENT_AGE_MIN' and 'PATIENT_AGE_MAX' columns.
|
|
"""
|
|
if "PATIENT_AGE_RANGE" in df.columns:
|
|
try:
|
|
# Create temporary columns for processing
|
|
df["PATIENT_AGE_MIN"] = df["PATIENT_AGE_RANGE"]
|
|
df["PATIENT_AGE_MAX"] = df["PATIENT_AGE_RANGE"]
|
|
|
|
# Handle hyphenated ranges
|
|
mask = df["PATIENT_AGE_RANGE"].str.contains("-", na=False)
|
|
if mask.any():
|
|
temp_split = df.loc[mask, "PATIENT_AGE_RANGE"].str.split(
|
|
"-", expand=True
|
|
)
|
|
df.loc[mask, "PATIENT_AGE_MIN"] = temp_split[0]
|
|
df.loc[mask, "PATIENT_AGE_MAX"] = temp_split[1]
|
|
|
|
except Exception as e:
|
|
# If any error occurs, keep the original values
|
|
pass
|
|
|
|
# Remove the original PATIENT_AGE_RANGE column
|
|
df.drop(columns=["PATIENT_AGE_RANGE"], inplace=True, errors="ignore")
|
|
|
|
return df
|
|
|
|
|
|
def add_aarete_derived_amendment_num(df: pd.DataFrame) -> pd.DataFrame:
|
|
"""
|
|
Add the 'AARETE_DERIVED_AMENDMENT_NUM' column to the DataFrame.
|
|
|
|
Args:
|
|
df (pd.DataFrame): The input DataFrame.
|
|
|
|
Returns:
|
|
pd.DataFrame: The updated DataFrame with the 'AARETE_DERIVED_AMENDMENT_NUM' column.
|
|
"""
|
|
|
|
if "CONTRACT_AMENDMENT_NUM" in df.columns:
|
|
df["AARETE_DERIVED_AMENDMENT_NUM"] = (
|
|
df["CONTRACT_AMENDMENT_NUM"]
|
|
.str.split("-")
|
|
.str[1]
|
|
.fillna(df["CONTRACT_AMENDMENT_NUM"])
|
|
)
|
|
|
|
return df
|
|
|
|
|
|
def add_aarete_derived_product(df):
|
|
"""
|
|
Add AARETE_DERIVED_PRODUCT column to df, based on PRODUCT column.
|
|
|
|
The new column should be Title Case of the PRODUCT column value.
|
|
"""
|
|
if "PRODUCT" in df.columns:
|
|
df["AARETE_DERIVED_PRODUCT"] = df["PRODUCT"]
|
|
return df
|
|
|
|
|
|
def fill_empty_reimb_pct_rate(df):
|
|
"""
|
|
Fill 'REIMB_PCT_RATE' based on reimbursement method:
|
|
- Set 'REIMB_PCT_RATE' to '100' for rows where:
|
|
- 'REIMB_PCT_RATE', 'REIMB_FEE_RATE', and 'REIMB_CONVERSION_FACTOR' are all empty.
|
|
- Set 'REIMB_PCT_RATE' to 'N/A' for rows where:
|
|
- 'AARETE_DERIVED_REIMB_METHOD' contains specific grouper.
|
|
"""
|
|
required_cols = [
|
|
"REIMB_PCT_RATE",
|
|
"REIMB_FEE_RATE",
|
|
"REIMB_CONVERSION_FACTOR",
|
|
"AARETE_DERIVED_REIMB_METHOD",
|
|
]
|
|
|
|
if all(col in df.columns for col in required_cols):
|
|
# Fill with '100' when all three rate fields are empty
|
|
pct_empty = string_utils.is_empty(df["REIMB_PCT_RATE"])
|
|
fee_empty = string_utils.is_empty(df["REIMB_FEE_RATE"])
|
|
conversion_empty = string_utils.is_empty(df["REIMB_CONVERSION_FACTOR"])
|
|
applicable_methods = ["Billed Charges", "Fee Schedule", "Case Rate", "Cost"]
|
|
method_applicable = df["AARETE_DERIVED_REIMB_METHOD"].isin(applicable_methods)
|
|
|
|
fill_mask = pct_empty & fee_empty & conversion_empty & method_applicable
|
|
df.loc[fill_mask, "REIMB_PCT_RATE"] = "100"
|
|
|
|
# Set to N/A for specific reimbursement methods
|
|
na_methods = ["Per Diem", "Flat Rate", "Per Visit"]
|
|
|
|
na_mask = df["AARETE_DERIVED_REIMB_METHOD"].isin(na_methods)
|
|
df.loc[na_mask, "REIMB_PCT_RATE"] = "N/A"
|
|
|
|
return df
|
|
|
|
|
|
def remove_redundant_reimb_info(df):
|
|
"""
|
|
Checks if "REIMB_" fields are the same as their corresponding 1:1 fields.
|
|
If they are the same, replace the "REIMB_" values with ""
|
|
|
|
Args:
|
|
df (pd.DataFrame): The input DataFrame.
|
|
|
|
Returns:
|
|
pd.DataFrame: The updated DataFrame with redundant reimbursement field values removed.
|
|
"""
|
|
COLUMN_PAIRS = [
|
|
("REIMB_EFFECTIVE_DT", "AARETE_DERIVED_EFFECTIVE_DT"),
|
|
("REIMB_TERMINATION_DT", "AARETE_DERIVED_TERMINATION_DT"),
|
|
("REIMB_PROV_NAME", "PROV_GROUP_NAME_FULL"),
|
|
]
|
|
for column_tuple in COLUMN_PAIRS:
|
|
reimb_col = column_tuple[0]
|
|
one_to_one_col = column_tuple[1]
|
|
if all(col in df.columns for col in [reimb_col, one_to_one_col]):
|
|
mask = df[reimb_col] == df[one_to_one_col]
|
|
df.loc[mask, reimb_col] = ""
|
|
return df
|
|
|
|
|
|
def replace_provider_name_from_hybrid_smart_chunking(df):
|
|
"""
|
|
Replaces PROV_GROUP_NAME_FULL with PROVIDER_NAME for rows where PROVIDER_NAME is not empty/N/A,
|
|
then removes the PROVIDER_NAME column.
|
|
|
|
This function is used in postprocessing after the Hybrid Smart Chunking provider name extraction.
|
|
The RAG method may produce a more accurate provider name than the initial extraction,
|
|
so we use it to override PROV_GROUP_NAME_FULL when available.
|
|
|
|
Args:
|
|
df (pd.DataFrame): DataFrame containing contract data
|
|
|
|
Returns:
|
|
pd.DataFrame: DataFrame with PROV_GROUP_NAME_FULL updated and PROVIDER_NAME removed
|
|
"""
|
|
# Check if both columns exist
|
|
if "PROVIDER_NAME" not in df.columns:
|
|
logging.debug(
|
|
"PROVIDER_NAME column not found in DataFrame, skipping replacement"
|
|
)
|
|
return df
|
|
|
|
if "PROV_GROUP_NAME_FULL" not in df.columns:
|
|
logging.debug("PROV_GROUP_NAME_FULL column not found in DataFrame")
|
|
# Just remove PROVIDER_NAME if it exists
|
|
df = df.drop(columns=["PROVIDER_NAME"])
|
|
return df
|
|
|
|
# Create a mask for rows where PROVIDER_NAME should replace PROV_GROUP_NAME_FULL
|
|
# Replace when PROVIDER_NAME is not empty
|
|
mask = ~string_utils.is_empty(df["PROVIDER_NAME"], pd_mask=True)
|
|
|
|
# Count how many rows will be updated
|
|
num_replacements = mask.sum()
|
|
|
|
if num_replacements > 0:
|
|
# Replace PROV_GROUP_NAME_FULL with PROVIDER_NAME for matching rows
|
|
df.loc[mask, "PROV_GROUP_NAME_FULL"] = df.loc[mask, "PROVIDER_NAME"]
|
|
logging.debug(
|
|
f"Replaced PROV_GROUP_NAME_FULL with PROVIDER_NAME for {num_replacements} rows"
|
|
)
|
|
else:
|
|
logging.debug("No valid PROVIDER_NAME values to replace PROV_GROUP_NAME_FULL")
|
|
|
|
# Remove the PROVIDER_NAME column
|
|
df = df.drop(columns=["PROVIDER_NAME"])
|
|
|
|
return df
|
|
|
|
|
|
def deduplicate_provider_columns(df: pd.DataFrame) -> pd.DataFrame:
|
|
"""Deduplicate provider fields in the DataFrame.
|
|
|
|
This function removes PROV_GROUP_* values from the PROV_OTHER_* columns if they are already present.
|
|
In addition, it deduplicates PROV_OTHER_* values.
|
|
|
|
E.g. If PROV_GROUP_TIN is ["123456789"] and PROV_OTHER_TIN is ["123456789", "987654321", "987654321"],
|
|
then PROV_OTHER_TIN will be updated to ["987654321"].
|
|
|
|
|
|
Args:
|
|
df (pd.DataFrame): The input DataFrame.
|
|
|
|
Returns:
|
|
pd.DataFrame: The updated DataFrame with deduplicated provider fields.
|
|
"""
|
|
provider_columns = [
|
|
"PROV_GROUP_TIN",
|
|
"PROV_GROUP_NPI",
|
|
"PROV_GROUP_NAME_FULL",
|
|
"PROV_OTHER_TIN",
|
|
"PROV_OTHER_NPI",
|
|
"PROV_OTHER_NAME_FULL",
|
|
]
|
|
|
|
if not all(col in df.columns for col in provider_columns):
|
|
return df
|
|
|
|
for index, row in df.iterrows():
|
|
group_values = {
|
|
"PROV_OTHER_TIN": row.get("PROV_GROUP_TIN"),
|
|
"PROV_OTHER_NPI": row.get("PROV_GROUP_NPI"),
|
|
"PROV_OTHER_NAME_FULL": row.get("PROV_GROUP_NAME_FULL"),
|
|
}
|
|
|
|
for other_col, group_vals in group_values.items():
|
|
other_vals = row.get(other_col)
|
|
|
|
if not other_vals:
|
|
# None, NaN, or empty list
|
|
df.at[index, other_col] = []
|
|
continue
|
|
|
|
# Normalize other_vals to a list if it's a string
|
|
if isinstance(other_vals, str):
|
|
other_vals = [other_vals]
|
|
|
|
# Normalize group values (list or empty)
|
|
group_vals = group_vals or []
|
|
|
|
# Remove GROUP values
|
|
filtered = [
|
|
v for v in other_vals if v not in group_vals and v and v != "UNKNOWN"
|
|
]
|
|
|
|
# Deduplicate while preserving order
|
|
seen = set()
|
|
deduped = []
|
|
for v in filtered:
|
|
if v not in seen:
|
|
seen.add(v)
|
|
deduped.append(v)
|
|
|
|
df.at[index, other_col] = deduped
|
|
|
|
return df
|
|
|
|
|
|
def update_grouper_base_rate_and_grouper_pct_rate(df):
|
|
"""
|
|
Update GROUPER_PCT_RATE and GROUPER_BASE_RATE columns based on grouper conditions.
|
|
|
|
Creates new columns and populates them when:
|
|
- AARETE_DERIVED_REIMB_METHOD is 'Grouper', OR
|
|
- GROUPER_TYPE is not empty (using string_utils.is_empty) AND not empty array/empty array string, OR
|
|
- GROUPER_CD is not empty (using string_utils.is_empty) AND not empty array/empty array string
|
|
|
|
Args:
|
|
df: DataFrame to update
|
|
|
|
Returns:
|
|
DataFrame with updated GROUPER_PCT_RATE and GROUPER_BASE_RATE columns
|
|
"""
|
|
# Required columns for the operation
|
|
required_cols = [
|
|
"AARETE_DERIVED_REIMB_METHOD",
|
|
"REIMB_PCT_RATE",
|
|
"REIMB_CONVERSION_FACTOR",
|
|
"GROUPER_TYPE" "GROUPER_CD",
|
|
]
|
|
|
|
# Check if all required columns exist
|
|
if not all(col in df.columns for col in required_cols):
|
|
# Initialize empty columns and return if required columns are missing
|
|
df["GROUPER_PCT_RATE"] = ""
|
|
df["GROUPER_BASE_RATE"] = ""
|
|
return df
|
|
|
|
# Create grouper condition mask
|
|
grouper_mask = (df["AARETE_DERIVED_REIMB_METHOD"] == "Grouper") & (
|
|
(~string_utils.is_empty(df["GROUPER_TYPE"]) & df["GROUPER_TYPE"].ne("[]"))
|
|
| (~string_utils.is_empty(df["GROUPER_CD"]) & df["GROUPER_CD"].ne("[]"))
|
|
)
|
|
|
|
# Initialize new columns with empty strings
|
|
df["GROUPER_PCT_RATE"] = ""
|
|
df["GROUPER_BASE_RATE"] = ""
|
|
|
|
# Update values where grouper condition is met
|
|
df.loc[grouper_mask, "GROUPER_PCT_RATE"] = df.loc[grouper_mask, "REIMB_PCT_RATE"]
|
|
df.loc[grouper_mask, "GROUPER_BASE_RATE"] = df.loc[
|
|
grouper_mask, "REIMB_CONVERSION_FACTOR"
|
|
]
|
|
|
|
return df
|
|
|
|
|
|
def attach_sid_column(df: pd.DataFrame) -> pd.DataFrame:
|
|
"""
|
|
Ensures the given AARETE_DERIVED_SID columns exist in the DataFrame, fills missing values,
|
|
and creates a unique 'AARETE_DERIVED_SID' identifier column — while keeping all original columns.
|
|
|
|
Steps:
|
|
1. Verify that all columns listed in sid_cols exist in the DataFrame.
|
|
- If any are missing, create them and fill with ''.
|
|
2. Concatenate all sid_cols values with '__' as a separator to form 'AARETE_DERIVED_SID'.
|
|
3. If any value in sid_cols is blank or NaN, replace it with 'xx' in the concatenation.
|
|
4. Add the 'AARETE_DERIVED_SID' column to the DataFrame.
|
|
|
|
Parameters
|
|
----------
|
|
df : pd.DataFrame
|
|
Input DataFrame containing the source columns.
|
|
Returns
|
|
-------
|
|
pd.DataFrame
|
|
DataFrame with all original columns plus a new 'AARETE_DERIVED_SID' column.
|
|
"""
|
|
sid_cols = [
|
|
"CLIENT_NAME",
|
|
"PAYER_NAME",
|
|
"PROV_GROUP_TIN",
|
|
"DOCUMENT_TYPE",
|
|
"AARETE_DERIVED_AMENDMENT_NUM",
|
|
"AARETE_DERIVED_CLAIM_TYPE_CD",
|
|
"AARETE_DERIVED_LOB",
|
|
"AARETE_DERIVED_EFFECTIVE_DT",
|
|
]
|
|
|
|
# Make a copy to avoid mutating the original DataFrame
|
|
df = df.copy()
|
|
# Ensure all SID columns exist; create missing ones filled with 'xx'
|
|
for col in sid_cols:
|
|
if col not in df.columns:
|
|
df[col] = ""
|
|
|
|
# Create the concatenated SID column (replace blanks/NaN only here)
|
|
if df.empty:
|
|
df["AARETE_DERIVED_SID"] = pd.Series(dtype=str)
|
|
else:
|
|
sid_df = df[sid_cols].fillna("").astype(str)
|
|
df["AARETE_DERIVED_SID"] = sid_df.apply(
|
|
lambda row: "__".join(val if val.strip() else "xx" for val in row),
|
|
axis=1,
|
|
result_type="reduce",
|
|
)
|
|
|
|
return df
|
|
|
|
|
|
def add_aarete_derived_signatory_complete_ind(df: pd.DataFrame) -> pd.DataFrame:
|
|
"""
|
|
Add a derived indicator column to the DataFrame indicating whether all signatory lines are signed.
|
|
|
|
Creates column:
|
|
AARETE_DERIVED_SIGNATORY_COMPLETE_IND = 'Y' if
|
|
NUM_AVAILABLE_SIGNATORY_LINE_COUNT == NUM_SIGNED_SIGNATORY_LINE_COUNT
|
|
else 'N'.
|
|
|
|
Parameters
|
|
----------
|
|
df : pd.DataFrame
|
|
Input DataFrame containing the required columns.
|
|
|
|
Returns
|
|
-------
|
|
pd.DataFrame
|
|
DataFrame with the new indicator column added.
|
|
"""
|
|
|
|
try:
|
|
required_cols = [
|
|
"NUM_AVAILABLE_SIGNATORY_LINE_COUNT",
|
|
"NUM_SIGNED_SIGNATORY_LINE_COUNT",
|
|
]
|
|
|
|
# Validate required columns
|
|
for col in required_cols:
|
|
if col not in df.columns:
|
|
return df
|
|
|
|
# Ensure consistent numeric types
|
|
for col in required_cols:
|
|
df[col] = pd.to_numeric(df[col], errors="coerce")
|
|
|
|
# Create indicator column
|
|
df["AARETE_DERIVED_SIGNATORY_COMPLETE_IND"] = (
|
|
df["NUM_AVAILABLE_SIGNATORY_LINE_COUNT"]
|
|
.eq(df["NUM_SIGNED_SIGNATORY_LINE_COUNT"])
|
|
.map({True: "Y", False: "N"})
|
|
)
|
|
|
|
return df
|
|
|
|
except Exception:
|
|
# Return df unchanged on any exception
|
|
return df
|