diff --git a/src/constants/investment_columns.py b/src/constants/investment_columns.py index 0f127a3..b852798 100644 --- a/src/constants/investment_columns.py +++ b/src/constants/investment_columns.py @@ -31,6 +31,7 @@ FIELD_FORMAT_MAPPING = { "AARETE_DERIVED_AMENDMENT_NUM": "str", "CLIENT_NAME": "str", "PAYER_NAME": "str", + "AARETE_DERIVED_PAYER_NAME": "str", "PAYER_STATE": "list[str]", "PROVIDER_STATE": "list[str]", "FILENAME_TIN": "list[str]", diff --git a/src/constants/regex_patterns.py b/src/constants/regex_patterns.py index 44b0af8..a3082d3 100644 --- a/src/constants/regex_patterns.py +++ b/src/constants/regex_patterns.py @@ -73,3 +73,6 @@ OCR_SUBSTITUTIONS = { "B": "8", # Letter B to eight "b": "8", } + +# dba patterns +DBA_PATTERNS = [r"\bD/B/A\b", r"\bDBA\b", r"\bDOING BUSINESS AS\b", r"\bD B A\b"] diff --git a/src/pipelines/saas/main.py b/src/pipelines/saas/main.py index 1d103c0..45cee1e 100644 --- a/src/pipelines/saas/main.py +++ b/src/pipelines/saas/main.py @@ -11,6 +11,9 @@ import pandas as pd random.seed(42) import src.pipelines.saas.file_processing as file_processing +from src.pipelines.shared.extraction import one_to_one_funcs +from src.pipelines.shared.postprocessing import postprocessing_funcs +from src.constants.investment_columns import FIELD_FORMAT_MAPPING import src.prompts.prompt_templates as prompt_templates import src.utils.io_utils as io_utils import src.utils.llm_utils as llm_utils @@ -221,6 +224,20 @@ def main(testing=False, test_params={}): FINAL_RESULT_DF_CC = pd.DataFrame() FINAL_RESULT_DF_DASHBOARD = pd.DataFrame() + # Add derived payer name and reorder columns for both CC and dashboard results + FINAL_RESULT_DF_CC = one_to_one_funcs.add_aarete_derived_payer_name( + FINAL_RESULT_DF_CC, state_flag=True + ) + FINAL_RESULT_DF_CC = postprocessing_funcs.reorder_columns( + FINAL_RESULT_DF_CC, FIELD_FORMAT_MAPPING + ) + FINAL_RESULT_DF_DASHBOARD = one_to_one_funcs.add_aarete_derived_payer_name( + FINAL_RESULT_DF_DASHBOARD, state_flag=True + ) + FINAL_RESULT_DF_DASHBOARD = postprocessing_funcs.reorder_columns( + FINAL_RESULT_DF_DASHBOARD, FIELD_FORMAT_MAPPING + ) + if len(error_results) > 0: ERROR_RESULT_DF = pd.concat(error_results, ignore_index=True) else: diff --git a/src/pipelines/saas/prompts/prompt_calls.py b/src/pipelines/saas/prompts/prompt_calls.py index 0ff35c7..bffdbe2 100644 --- a/src/pipelines/saas/prompts/prompt_calls.py +++ b/src/pipelines/saas/prompts/prompt_calls.py @@ -1080,3 +1080,56 @@ def prompt_provider_info( logging.error(f"Raw response: {llm_answer_raw}") # Return a default value return [{"TIN": "N/A", "NPI": "N/A", "NAME": "N/A"}] + + +def prompt_aarete_derived_payer_name(cluster_rows: list[dict], state_flag: bool) -> str: + """ + Select the canonical payer name from a cluster of similar names (e.g., merging 'AvMed, Inc' and 'AvMed Health Plans'). + Expects a list of standardized payer name dictionaries. + + Args: + cluster_rows (list[dict]): A list of dictionaries, each containing standardized payer name information. + + Returns: + str: The canonical payer name selected by the LLM, or "N/A" if selection fails. + """ + + # If only one record in cluster, avoid LLM call + if len(cluster_rows) == 1: + base_name = cluster_rows[0].get("base_payer_name", "UNKNOWN") + return str(base_name).upper() + + prompt, _parser = prompt_templates.AARETE_DERIVED_PAYER_NAME( + cluster_rows, state_flag + ) + logging.debug(f"Aarete Derived Payer Name Prompt: {prompt}") + llm_answer_raw = llm_utils.invoke_claude( + prompt, + "sonnet_latest", + "all_files", + cache=True, + instruction=prompt_templates.AARETE_DERIVED_PAYER_NAME_INSTRUCTION(), + usage_label="AARETE_DERIVED_PAYER_NAME", + ) + logging.debug(f"LLM Response for Aarete Derived Payer Name: {llm_answer_raw}") + try: + llm_answer_final = _parser(llm_answer_raw) + if llm_answer_final and not string_utils.is_empty(llm_answer_final): + derived_name = ( + llm_answer_final["AARETE_DERIVED_PAYER_NAME"] + .replace("```", "") + .replace("**", "") + .strip() + .upper() + ) + return derived_name + else: + # If LLM returns N/A or empty, fallback to the programmatic base_payer_name + fallback = cluster_rows[0].get("base_payer_name", "UNKNOWN") + return str(fallback).upper() + + except Exception as e: + logging.error( + f"Error parsing LLM response for Aarete Derived Payer Name: {str(e)}" + ) + return str(cluster_rows[0].get("base_payer_name", "UNKNOWN")).upper() diff --git a/src/pipelines/shared/extraction/one_to_one_funcs.py b/src/pipelines/shared/extraction/one_to_one_funcs.py index 3e32d88..bdfef8e 100644 --- a/src/pipelines/shared/extraction/one_to_one_funcs.py +++ b/src/pipelines/shared/extraction/one_to_one_funcs.py @@ -2,6 +2,10 @@ import logging import os import re import json +import numpy as np +from typing import Any +from rapidfuzz import fuzz +from collections import defaultdict import pandas as pd import src.pipelines.shared.postprocessing.postprocessing_funcs as postprocessing_funcs @@ -395,3 +399,306 @@ def add_signature_count(one_to_one_results: dict, contract_txt: str) -> dict: one_to_one_results["NUM_SIGNED_SIGNATORY_LINE_COUNT"] = total_signatures return one_to_one_results + + +def add_aarete_derived_payer_name(df: pd.DataFrame, state_flag: bool) -> pd.DataFrame: + """ + Derive canonical payer names from PAYER_NAME column. + + This function clusters similar payer names, standardizes them, and selects + the most appropriate canonical name for each cluster. + + Args: + df: Input DataFrame containing a PAYER_NAME column. + state_flag: Controls whether payer names are treated as state-specific. + Always True (confirmed with business team). + True → Preserve state distinctions. + Example: "Molina Healthcare of Texas" and + "Molina Healthcare of Utah" remain separate. + False → Collapse regional state suffixes into national brand. + Example: "Molina Healthcare of Texas" and + "Molina Healthcare of Utah" both become "Molina Healthcare". + Leading state-based brand names (e.g., "Oklahoma Complete Health") + are preserved. + + Returns: + DataFrame with added AARETE_DERIVED_PAYER_NAME column. + """ + + if df is None or df.empty: + logging.warning("Input DataFrame is empty. Returning empty DataFrame.") + return df + + df_payer = df.copy() + # Check if PAYER_NAME column exists in the DataFrame, if not add an empty AARETE_DERIVED_PAYER_NAME column and return the DataFrame + if "PAYER_NAME" not in df_payer.columns: + logging.warning( + "PAYER_NAME column not found in DataFrame. skipping AARETE_DERIVED_PAYER_NAME column." + ) + return df_payer + # 1. Extract unique payer names + payer_names = df_payer["PAYER_NAME"].dropna().unique().tolist() + unique_payer_names = [ + payer for payer in payer_names if not string_utils.is_empty(payer) + ] + if not unique_payer_names: + df_payer["AARETE_DERIVED_PAYER_NAME"] = "" + return df_payer + + # 2. Standardize OR normalizing unique payer names (list of dicts) + standardized_unique_payer_name_rows = standardize_unique_names( + unique_payer_names, state_flag + ) + + standardized_unique_payer_name_rows = propagate_dba_to_legal( + standardized_unique_payer_name_rows + ) + + # Combining Step`s 3-4: Build similarity matrix and cluster with blocking to handle large volumes efficiently + clusters = cluster_with_blocking( + standardized_unique_payer_name_rows, + state_flag=state_flag, + threshold=90, + ) + + # 5. Build mapping using LLM placeholder + mapping = {} + for cluster in clusters: + derived_payer_name = prompt_calls.prompt_aarete_derived_payer_name( + cluster, state_flag + ) + + for record in cluster: + mapping[record["original_payer_name"]] = derived_payer_name + + df_payer["AARETE_DERIVED_PAYER_NAME"] = df_payer["PAYER_NAME"].map(mapping) + + return df_payer + + +def standardize_unique_names( + unique_names: list[str], state_flag: bool +) -> list[dict[str, Any]]: + """ + Normalize payer names and enrich them with DBA and token signatures. + Runs only on unique payer names for efficiency. + + Block key is derived from the FIRST token of the base payer name + (not from sorted token signature) to preserve brand identity. + """ + rows = [] + + for name in unique_names: + clean = string_utils.normalize_text(name) + + state = string_utils.detect_state(clean) + + legal, dba = string_utils.extract_dba_payer_name(clean) + base = dba if dba else clean + legal = legal if legal else clean + + base = string_utils.remove_legal_suffixes(base) + legal = string_utils.remove_legal_suffixes(legal) + + if not state_flag: + base = string_utils.remove_states(base) + legal = string_utils.remove_states(legal) + + signature = string_utils.token_signature(base) + tokens = base.split() + block_key = tokens[0] if tokens else "" + + rows.append( + { + "original_payer_name": name, + "base_payer_name": base, + "signature": signature if signature else (), + "has_dba_payer_name": dba is not None, + "dba_payer_name": dba, + "legal_payer_name": legal, + "state": state, + "block_key": block_key, + } + ) + + return rows + + +def build_similarity_matrix(rows: list[dict]) -> np.ndarray: + """ + Compute a pairwise similarity matrix for base payer names. + + This function calculates fuzzy string similarity scores between + all pairs of 'base_payer_name' values in the input rows using + multiple fuzzy matching strategies. The highest score is used + as the final similarity value. + + Args: + rows (list[dict]): A list of dictionaries, each containing + a 'base_payer_name' key. + + Returns: + np.ndarray: A square (n x n) matrix of similarity scores + ranging from 0 to 100. + """ + # Number of payer name records + n = len(rows) + + # Initialize an empty similarity matrix + sim_matrix = np.zeros((n, n)) + + # Compute pairwise similarities + for i in range(n): + for j in range(i, n): + if i == j: + # A string is always 100% similar to itself + sim_matrix[i, j] = 100 + else: + # Compute similarity using multiple fuzzy matching methods and different combinations + score = max( + fuzz.ratio(rows[i]["base_payer_name"], rows[j]["base_payer_name"]), + fuzz.token_set_ratio( + rows[i]["base_payer_name"], rows[j]["base_payer_name"] + ), + fuzz.ratio(rows[i]["legal_payer_name"], rows[j]["base_payer_name"]), + fuzz.token_set_ratio( + rows[i]["legal_payer_name"], rows[j]["base_payer_name"] + ), + fuzz.ratio(rows[i]["base_payer_name"], rows[j]["legal_payer_name"]), + fuzz.token_set_ratio( + rows[i]["base_payer_name"], rows[j]["legal_payer_name"] + ), + ) + + # Fill both symmetric positions in the matrix + sim_matrix[i, j] = score + sim_matrix[j, i] = score + + return sim_matrix + + +def cluster_from_similarity( + rows: list[dict[str, Any]], + sim_matrix: np.ndarray, + state_flag: bool, + threshold: int = 90, +) -> list[list[dict[str, Any]]]: + """ + Cluster payer records using a similarity matrix and threshold. + + Each row is grouped with other rows whose similarity score + meets or exceeds the given threshold. Once a row is assigned + to a cluster, it is not reconsidered. + + Args: + rows (list[dict]): List of row dictionaries containing payer data. + sim_matrix (np.ndarray): Pairwise similarity matrix (n x n). + threshold (int): Minimum similarity score required to cluster. + + Returns: + list[list[dict]]: A list of clusters, where each cluster + is a list of row dictionaries. + """ + n = len(rows) + visited: set[int] = set() + clusters: list[list[dict[str, Any]]] = [] + + # Iterate through each record + for i in range(n): + # Skip rows that have already been clustered + if i in visited: + continue + + # Start a new cluster with the current index + cluster_indices = {i} + + # Add all rows similar enough to the current one + for j in range(n): + if sim_matrix[i, j] >= threshold: + + # --- STATE LOGIC --- + if state_flag: + state_i = rows[i]["state"] + state_j = rows[j]["state"] + + # If one has state and the other doesn't → do NOT cluster + if state_i != state_j: + continue + cluster_indices.add(j) + + # Mark clustered indices as visited + visited.update(cluster_indices) + + # Build the cluster from the original rows + clusters.append([rows[idx] for idx in cluster_indices]) + + return clusters + + +def cluster_with_blocking( + rows: list[dict[str, Any]], + state_flag: bool, + threshold: int = 90, +) -> list[list[dict[str, Any]]]: + """ + Cluster payer names using first-token blocking to scale to large volumes. + """ + + # 1. Group rows by block key + blocks = defaultdict(list) + for row in rows: + blocks[row["block_key"]].append(row) + + all_clusters = [] + + # 2. Process each block independently + for block_rows in blocks.values(): + + # Small blocks don't need matrix + if len(block_rows) == 1: + all_clusters.append(block_rows) + continue + + sim_matrix = build_similarity_matrix(block_rows) + + block_clusters = cluster_from_similarity( + block_rows, + sim_matrix, + state_flag=state_flag, + threshold=threshold, + ) + + all_clusters.extend(block_clusters) + + return all_clusters + + +def propagate_dba_to_legal(rows: list[dict]) -> list[dict]: + """ + If a legal entity has an associated DBA in any row, + propagate that DBA to all rows sharing the same legal entity. + """ + + legal_to_dba = {} + + # First pass: build mapping + for row in rows: + legal = row.get("legal_payer_name") + dba = row.get("dba_payer_name") + + if legal and dba: + legal_to_dba[legal] = dba + + # Second pass: propagate + for row in rows: + legal = row.get("legal_payer_name") + + if legal in legal_to_dba: + new_base = legal_to_dba[legal] + row["base_payer_name"] = new_base + + # Update block key + tokens = new_base.split() + row["block_key"] = tokens[0] if tokens else "" + + return rows diff --git a/src/prompts/investment_prompts.json b/src/prompts/investment_prompts.json index 5b65111..30fdeb3 100644 --- a/src/prompts/investment_prompts.json +++ b/src/prompts/investment_prompts.json @@ -240,7 +240,7 @@ "field_name": "PAYER_NAME", "relationship": "one_to_one", "field_type": "smart_chunked", - "prompt": "What is the name of the payer that is a party to the contract as stated in the Preamble?", + "prompt": "What is the name of the payer that is a party to the contract as stated in the Preamble? Extract the name exactly as it appears in the contract.", "retrieval_question": "What is the name of the payer, health plan, insurance company, or payor that is a party to this contract? Payer name in the preamble, parties section, or beginning of agreement." }, { diff --git a/src/prompts/prompt_templates.py b/src/prompts/prompt_templates.py index abd8e58..3a0d5ca 100644 --- a/src/prompts/prompt_templates.py +++ b/src/prompts/prompt_templates.py @@ -2378,3 +2378,61 @@ Reimbursement: {reimbursement_dict["REIMB_TERM"]} Return the integer of the {special_case_term} that is most associated with the REIMBURSEMENT TERM.""" return (prompt, _json_list_parser) + + +def AARETE_DERIVED_PAYER_NAME_INSTRUCTION(): + return f"""You are an expert in healthcare entity data normalization. + Your goal is to take a list of similar payer names and determine the single, + most accurate canonical legal or brand name. + + Given variants string contains variations of a single Payer (insurance company) name + extracted from different contracts. They all refer to the same entity. + + TASK: + 1. Select or derive the most professional canonical name. + 2. If given STATE_FLAG is FALSE: + - Collapse regional entities into the national brand (e.g., 'Molina Healthcare'). + - DO NOT remove the word 'State' if it is part of a proper noun (e.g., 'New York State Catholic Health Plan'). + 3. If given STATE_FLAG is TRUE: + - Maintain state distinctions (e.g., 'Molina Healthcare of Texas'). + 4. Prefer the 'DBA' (Doing Business As) name if it is present. + 5. Cleaned name should be preferred over original name, but use your judgment to determine if the cleaned name is actually more accurate or if it has removed important context. + 6. Remove excessive noise like "a corporation of...", "an Illinois company", or "d/b/a" prefixes. + 7. Maintain essential legal identifiers only if they are part of the standard brand identity. + 8. Return only the cleaned canonical name without any explanation. + + Example Variants Input: + VARIANTS: + - AvMed Health Plan, Inc + - AvMED, INC., d/b/a AvMED HEALTH PLANS + Example Output: + {{"AARETE_DERIVED_PAYER_NAME":"AvMed Health Plans"}} + + [OUTPUT FORMAT] + Return ONLY the JSON dictionary with no explanatory text. Use "AARETE_DERIVED_PAYER_NAME" as the key. + {JSON_DICT_FORMAT_INSTRUCTIONS} + """ + + +def AARETE_DERIVED_PAYER_NAME( + cluster_rows: list[dict], state_flag: bool +) -> Tuple[str, Callable[[str], dict]]: + """ + Constructs the prompt using both original and cleaned names for maximum context. + """ + variant_list = [] + for r in cluster_rows: + # Using your preferred logic to show the LLM both states of the data + variant = ( + f"- Original: {r['original_payer_name']} (Cleaned: {r['base_payer_name']})" + ) + variant_list.append(variant) + + variants_string = "\n".join(variant_list) + prompt = f""" + VARIANTS: + {variants_string} + STATE_FLAG: + {state_flag} + """ + return (prompt, _json_dict_parser) diff --git a/src/tests/test_derived_payer_name.py b/src/tests/test_derived_payer_name.py new file mode 100644 index 0000000..1d273bf --- /dev/null +++ b/src/tests/test_derived_payer_name.py @@ -0,0 +1,206 @@ +import unittest +import pandas as pd +import numpy as np +from unittest.mock import patch + +from src.utils.string_utils import ( + normalize_text, + extract_dba_payer_name, + remove_legal_suffixes, + remove_states, + token_signature, +) +from src.pipelines.shared.extraction.one_to_one_funcs import ( + standardize_unique_names, + add_aarete_derived_payer_name, + cluster_from_similarity, + build_similarity_matrix, + cluster_with_blocking, + propagate_dba_to_legal, +) + + +class TestPayerNameTextUtilities(unittest.TestCase): + + def test_normalize_payer_name_text(self): + text = "Molina Healthcare, Inc." + result = normalize_text(text) + self.assertEqual(result, "molina healthcare inc") + + def test_extract_dba_payer_name_from_text(self): + text = "ABC Corp d/b/a XYZ Health" + normalized = normalize_text(text) + result = extract_dba_payer_name(normalized) + self.assertEqual(result, ("abc corp", "xyz health")) + + def test_remove_legal_suffixes_from_payer_name(self): + text = "molina healthcare inc" + result = remove_legal_suffixes(text) + self.assertEqual(result, "molina healthcare") + + def test_generate_token_signature_for_payer_name(self): + text = "molina healthcare of utah" + result = token_signature(text) + self.assertEqual(result, ("healthcare", "molina", "utah")) + + +class TestPayerNameStandardization(unittest.TestCase): + + def test_standardize_unique_payer_names(self): + names = ["Molina Healthcare of Utah, Inc."] + rows = standardize_unique_names(names, state_flag=True) + + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["original_payer_name"], names[0]) + self.assertIn("base_payer_name", rows[0]) + self.assertIn("signature", rows[0]) + self.assertIn("block_key", rows[0]) + self.assertFalse(rows[0]["has_dba_payer_name"]) + + +class TestRemoveStatesFunction(unittest.TestCase): + + def test_remove_trailing_state_suffix(self): + self.assertEqual( + remove_states("community health choice texas"), "community health choice" + ) + + self.assertEqual(remove_states("fidelis care new york"), "fidelis care") + + def test_preserve_leading_state_brand(self): + self.assertEqual( + remove_states("oklahoma complete health"), "oklahoma complete health" + ) + + self.assertEqual( + remove_states("texas childrens health plan"), "texas childrens health plan" + ) + + def test_remove_of_state_pattern(self): + self.assertEqual( + remove_states("molina healthcare of utah"), "molina healthcare of" + ) + + self.assertEqual( + remove_states("molina healthcare of texas"), "molina healthcare of" + ) + + def test_no_state_present(self): + self.assertEqual(remove_states("centene corporation"), "centene corporation") + + +class TestPropagateDbaToLegal(unittest.TestCase): + + def test_propagate_dba_to_matching_legal_entity(self): + rows = [ + { + "original_payer_name": "NY State Catholic Health Plan dba Fidelis Care New York", + "base_payer_name": "fidelis care new york", + "legal_payer_name": "new york state catholic health plan", + "dba_payer_name": "fidelis care new york", + "block_key": "fidelis", + }, + { + "original_payer_name": "New York State Catholic Health Plan", + "base_payer_name": "new york state catholic health plan", + "legal_payer_name": "new york state catholic health plan", + "dba_payer_name": None, + "block_key": "new", + }, + ] + + updated = propagate_dba_to_legal(rows) + + # Both should now share same base_payer_name + self.assertEqual(updated[0]["base_payer_name"], "fidelis care new york") + self.assertEqual(updated[1]["base_payer_name"], "fidelis care new york") + + # Block key should also update + self.assertEqual(updated[1]["block_key"], "fidelis") + + def test_no_propagation_when_no_dba_present(self): + rows = [ + { + "original_payer_name": "ODS Health Plan", + "base_payer_name": "ods health plan", + "legal_payer_name": "ods health plan", + "dba_payer_name": None, + "block_key": "ods", + }, + { + "original_payer_name": "Moda Health Plan", + "base_payer_name": "moda health plan", + "legal_payer_name": "moda health plan", + "dba_payer_name": None, + "block_key": "moda", + }, + ] + + updated = propagate_dba_to_legal(rows) + + # Should remain unchanged + self.assertEqual(updated[0]["base_payer_name"], "ods health plan") + self.assertEqual(updated[1]["base_payer_name"], "moda health plan") + + +class TestPayerNameClustering(unittest.TestCase): + + def test_cluster_payer_names_with_state_flag_true(self): + names = ["Molina Healthcare of Utah", "Molina Healthcare"] + + rows = standardize_unique_names(names, state_flag=True) + + clusters = cluster_with_blocking(rows, state_flag=True, threshold=85) + + # Should remain separate clusters + self.assertEqual(len(clusters), 2) + + def test_cluster_payer_names_with_state_flag_false(self): + names = ["Molina Healthcare of Utah", "Molina Healthcare"] + + rows = standardize_unique_names(names, state_flag=False) + + clusters = cluster_with_blocking(rows, state_flag=False, threshold=85) + + # Should cluster together + self.assertEqual(len(clusters), 1) + + +class TestPayerNameEndToEndPipeline(unittest.TestCase): + + @patch("src.pipelines.saas.prompts.prompt_calls.prompt_aarete_derived_payer_name") + def test_derive_payer_names_end_to_end(self, mock_llm): + + data = { + "PAYER_NAME": [ + "NEW YORK STATE CATHOLIC HEALTH PLAN, INC. dba FIDELIS CARE NEW YORK", + "NEW YORK STATE CATHOLIC HEALTH PLAN, INC.", + "NEW YORK STATE CATHOLIC HEALTH PLAN, INC. d/b/a FIDELIS CARE NEW YORK", + ] + } + + # Mock LLM to return canonical name + mock_llm.return_value = "FIDELIS CARE NEW YORK" + + df = pd.DataFrame(data) + + result = add_aarete_derived_payer_name(df, state_flag=True) + + self.assertIn("AARETE_DERIVED_PAYER_NAME", result.columns) + + # All rows should map to same canonical value + self.assertEqual( + result.loc[0, "AARETE_DERIVED_PAYER_NAME"], "FIDELIS CARE NEW YORK" + ) + + self.assertEqual( + result.loc[1, "AARETE_DERIVED_PAYER_NAME"], "FIDELIS CARE NEW YORK" + ) + + self.assertEqual( + result.loc[2, "AARETE_DERIVED_PAYER_NAME"], "FIDELIS CARE NEW YORK" + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/src/utils/string_utils.py b/src/utils/string_utils.py index 8515815..12f2e22 100644 --- a/src/utils/string_utils.py +++ b/src/utils/string_utils.py @@ -14,8 +14,79 @@ from src.constants.regex_patterns import ( BACKTICK_PATTERN, PIPE_PATTERN, TRIPLE_BACKTICK_PATTERN, + DBA_PATTERNS, ) +LEGAL_SUFFIXES = { + "inc", + "inc.", + "incorporated", + "corp", + "corporation", + "llc", + "l.l.c", + "ltd", + "limited", + "co", + "company", + "affiliates", +} + +STOPWORDS = {"of", "the", "and", "its"} + +US_STATES = [ + "alabama", + "alaska", + "arizona", + "arkansas", + "california", + "colorado", + "connecticut", + "delaware", + "florida", + "georgia", + "hawaii", + "idaho", + "illinois", + "indiana", + "iowa", + "kansas", + "kentucky", + "louisiana", + "maine", + "maryland", + "massachusetts", + "michigan", + "minnesota", + "mississippi", + "missouri", + "montana", + "nebraska", + "nevada", + "new hampshire", + "new jersey", + "new mexico", + "new york", + "north carolina", + "north dakota", + "ohio", + "oklahoma", + "oregon", + "pennsylvania", + "rhode island", + "south carolina", + "south dakota", + "tennessee", + "texas", + "utah", + "vermont", + "virginia", + "washington", + "west virginia", + "wisconsin", + "wyoming", +] + def extract_text_from_delimiters( raw_output: str, delimiter: Delimiter, match_index: int = -1 @@ -806,3 +877,167 @@ def normalize_to_json_list(val): result = json.dumps(expanded) return result return val + + +def normalize_text(text: str) -> str: + """ + Normalize a text string by applying common preprocessing steps. + + This function: + - Converts all characters to lowercase + - Removes punctuation and special characters + - Replaces multiple whitespace characters with a single space + - Strips leading and trailing whitespace + + Args: + text (str): The input text to normalize. + + Returns: + str: The normalized text. + """ + # Convert the text to lowercase for case-insensitive processing + text = text.lower() + + # Replace all characters that are not letters, numbers, or whitespace with a space + text = re.sub(r"[^\w\s]", " ", text) + + # Replace multiple whitespace characters with a single space + # and remove leading/trailing spaces + text = re.sub(r"\s+", " ", text).strip() + + return text + + +def remove_legal_suffixes(text: str) -> str: + """ + Remove legal entity suffixes from a text string. + + This function splits the input text into tokens (words) and + removes any tokens that match known legal suffixes + (e.g., "llc", "inc", "corp"). + + Args: + text (str): The input text containing a business name. + + Returns: + str: The text with legal suffixes removed. + """ + # Split the text into individual words + tokens = text.split() + + # Filter out any tokens that are legal suffixes + tokens = [t for t in tokens if t not in LEGAL_SUFFIXES] + + # Rejoin the remaining tokens into a single string + return " ".join(tokens) + + +def extract_dba_payer_name(text: str) -> tuple[str | None, str | None]: + """ + Extract a DBA (Doing Business As) name from a text string. + + The function checks the input text against a list of predefined + DBA patterns (e.g., "dba", "doing business as"). If a pattern is found, + it returns the text that appears after the pattern. + + Ex: + Input: NEW YORK STATE CATHOLIC HEALTH PLAN, INC. d/b/a FIDELIS CARE NEW YORK + + Output: FIDELIS CARE NEW YORK + + Args: + text (str): The input text that may contain a DBA reference. + + Returns: + tuple[str | None, str | None]: A tuple containing: + - The text before the DBA pattern (potentially the legal name) + - The text after the DBA pattern (the extracted DBA name) + If no DBA pattern is found, both elements of the tuple will be None. + """ + # Iterate through all known DBA patterns + for pattern in DBA_PATTERNS: + # Split the text using the current pattern (case-insensitive) + match = re.split(pattern, text, flags=re.I) + + # If the pattern was found, the split will produce more than one part + if len(match) > 1: + # Return the text after the matched pattern, trimmed of whitespace + return match[0].strip(), match[-1].strip() + + # Return None if no DBA pattern is found + return None, None + + +def remove_states(text: str) -> str: + """ + Remove U.S. state names or abbreviations from a text string. + + The function iterates through a list of U.S. states and removes + any exact (word-boundary matched) occurrences from the input text. + Extra whitespace created by removals is cleaned up before returning. + + Remove state names when they appear as a regional suffix + (i.e., at the end of the string), but preserve them when + they are leading brand identifiers (e.g., 'Oklahoma Complete Health'). + + Args: + text (str): The input text that may contain U.S. state names. + + Returns: + str: The text with U.S. state references removed. + """ + + tokens = text.split() + if not tokens: + return text + + for state in US_STATES: + state_tokens = state.split() + + # Check if state appears at END of string + if tokens[-len(state_tokens) :] == state_tokens: + + # Do NOT remove if state is the first token + if tokens[0] == state_tokens[0]: + return text + + # Remove trailing state tokens + new_tokens = tokens[: -len(state_tokens)] + return " ".join(new_tokens).strip() + + # Also handle "of " pattern anywhere in the string, as it often indicates a regional suffixe.g., "Health Plan of New York" should become "Health Plan" + # Remove each state using a case-insensitive regex match. It looks for the word "of" followed by the state name. + for state in US_STATES: + text = re.sub(rf"\bof\s+{state}\b", "", text, flags=re.I) + + # Normalize whitespace after removals + return re.sub(r"\s+", " ", text).strip() + + +def detect_state(text: str) -> str | None: + """Return state name if found in text.""" + for state in US_STATES: + if re.search(rf"\b{state}\b", text, re.I): + return state + return None + + +def token_signature(text: str) -> tuple: + """ + Generate a normalized token signature from a text string. + + The function removes stopwords, sorts the remaining tokens, + and returns them as an immutable tuple. This signature can be + used for fast comparison or deduplication of similar texts. + + Args: + text (str): The input text to tokenize. + + Returns: + tuple: A sorted tuple of tokens excluding stopwords. + """ + # Split the text into tokens and remove stopwords + tokens = [t for t in text.split() if t not in STOPWORDS] + + # Sort tokens to ensure order-independent comparison + return tuple(sorted(tokens))