jacob-mathison/doczyai-pipelines
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Activity

Merged in refactor/hotfix_refactor (pull request #344)

Browse Source 
Refactor/hotfix refactor

* Initial hotfix refactor

* Fix references

* Remove test


Approved-by: Alex Galarce
This commit is contained in:
Katon Minhas
2025-01-06 22:39:55 +00:00
parent 48aa1d836b
commit 64e754d40f
11 changed files with 106 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fieldExtraction/src/apply_cnc_hotfix.py → fieldExtraction/scripts/cnc_hotfixes/apply_cnc_hotfix.py
View File
fieldExtraction/src/cnc_hotfix.py → fieldExtraction/scripts/cnc_hotfixes/cnc_hotfix.py
+1 -1
View File
@@ -17,7 +17,7 @@ warnings.simplefilter(action='ignore', category=FutureWarning)
import claude_funcs
import cnc_hotfix_effective_date_utils
import scripts.cnc_hotfixes.cnc_hotfix_effective_date_utils as cnc_hotfix_effective_date_utils
import config
import keywords
import postprocessing_funcs
fieldExtraction/src/cnc_hotfix_effective_date_utils.py → fieldExtraction/scripts/cnc_hotfixes/cnc_hotfix_effective_date_utils.py
View File
fieldExtraction/src/hotfix_helper_funcs.py → fieldExtraction/scripts/cnc_hotfixes/hotfix_helper_funcs.py
-107
View File
@@ -7,15 +7,6 @@ import postprocessing_funcs
from regex_funcs import find_regex_matches
from valid import STATE_MAP
keywords_to_detect_invalid_state_name = [ #This is for health plan state when the LLM talks to itself
"suggest",
"The Contract Text",
"refers",
"For ",
"Answer",
"Multiple",
"Statewide",
]
def convert_to_list(value):
if isinstance(value, str):
@@ -204,102 +195,6 @@ def irs_hotfix(filename, text_dict: dict[str, str]):
TIN_dict["PROV_GROUP_TIN"] = tin_from_title
return TIN_dict
def state_check(states: str | list[str], text_dict: dict[str, str]) -> str:
if not isinstance(states, str) and not isinstance(states,list):
states = 'N/A'
if (
isinstance(states, list) and all(isinstance(state, str) for state in states)
) or isinstance(states, str): # If it's a list of strings or a string...
if isinstance(states, str): # If it's a string, turn it to a list (even a singleton)
if "," in states:
states = states.split(",")
else:
states = [states]
result_states = []
for state in states:
state = state.strip().replace(".", "") # Remove "." in cases like "N.Y."
# fix 1
if state == "Hlorida":
state = "Florida"
# fix 2 - LLM self-talking
if any(
kw.lower() in state.lower()
for kw in keywords_to_detect_invalid_state_name
):
state = "N/A"
if state.upper() in STATE_MAP.keys(): # Check if `state` is a state abbreviation.
result_states.append(STATE_MAP[state.upper()].title())
elif state.strip().title() in STATE_MAP.values(): # Check if `state` is already an expanded name state from the validation set.
result_states.append(state.strip().title())
else:
result_states.append("N/A")
result_states = set(result_states)
result_states.discard("N/A") # Discard any N/As (so if `result_states` is ['Illinois', 'N/A'], it returns 'Illinois' instead of 'Multiple')
if len(result_states) == 1:
return list(result_states)[0]
elif len(result_states) > 1:
return "Multiple"
elif len(result_states) == 0:
state = get_health_plan_state(text_dict) # search for state(s) in contract. If it's not found, `N/A` will be returned
return state
raise TypeError(
f"Expected a string or list of strings, got {type(states).__name__}"
)
def get_health_plan_state(text_dict: dict[str, str]) -> str:
"""Go through a contract text and extract any mention of any state
Args:
text_dict (dict[str, str]): A string-keyed dictionary valued by contract text
Returns:
str: A single state extracted from the text, `multiple`, or `N/A`
"""
state_abbreviations = list(STATE_MAP.keys())
state_names = list(STATE_MAP.values())
state_names_and_abbreviations = state_abbreviations + state_names
pattern = r'\b(?:' + '|'.join(re.escape(state) for state in state_names_and_abbreviations) + r')\b'
states_set = set()
for page_num, page_content in text_dict.items():
state_matches = re.findall(pattern, page_content, flags=re.IGNORECASE)
state_matches = [state.strip().title() for state in state_matches]
states_set.update(state_matches)
states_list = list(states_set)
# Expand any abbreviations. If the state is already its full name, we're just gonna
# default to that. E.g., "CT" -> "Connecticut", "Connecticut" -> "Connecticut"
states_list = [STATE_MAP.get(state, state) for state in states_list]
if len(states_list) > 1:
state = "Multiple"
elif len(states_list) == 1:
state = states_list[0]
else:
state = "N/A"
return state
def check_field_for_matches(text, valid_lists):
"""
Check a field for matches from the valid lists using case-insensitive regex
@@ -314,8 +209,6 @@ def check_field_for_matches(text, valid_lists):
return term, list_type
return None, None
##### NPI #####
def find_10_digit_numbers(row):
matches = re.findall(r'\b\d{10}\b', row)
fieldExtraction/scripts/hotfix_stitch.py → fieldExtraction/scripts/cnc_hotfixes/hotfix_stitch.py
View File
fieldExtraction/scripts/hotfix_stitch_1A.py → fieldExtraction/scripts/cnc_hotfixes/hotfix_stitch_1A.py
View File
fieldExtraction/scripts/hotfix_stitch_1B.py → fieldExtraction/scripts/cnc_hotfixes/hotfix_stitch_1B.py
View File
fieldExtraction/src/postprocessing_funcs.py
+93 -2
View File
@@ -8,9 +8,8 @@ import claude_funcs
import config
import prompts
import valid
from hotfix_helper_funcs import state_check
from string_funcs import is_empty
from valid import DERIVED_INDICATOR_FIELDS
from valid import DERIVED_INDICATOR_FIELDS, STATE_MAP, keywords_to_detect_invalid_state_name
logging.basicConfig(
level=logging.ERROR, format="%(asctime)s - %(levelname)s - %(message)s"
@@ -335,6 +334,98 @@ def clean_health_plan_state(final_df):
final_df["HEALTH_PLAN_STATE"] = final_df["HEALTH_PLAN_STATE"].str.title()
return final_df
def get_health_plan_state(text_dict: dict[str, str]) -> str:
"""Go through a contract text and extract any mention of any state
Args:
text_dict (dict[str, str]): A string-keyed dictionary valued by contract text
Returns:
str: A single state extracted from the text, `multiple`, or `N/A`
"""
state_abbreviations = list(STATE_MAP.keys())
state_names = list(STATE_MAP.values())
state_names_and_abbreviations = state_abbreviations + state_names
pattern = r'\b(?:' + '|'.join(re.escape(state) for state in state_names_and_abbreviations) + r')\b'
states_set = set()
for page_num, page_content in text_dict.items():
state_matches = re.findall(pattern, page_content, flags=re.IGNORECASE)
state_matches = [state.strip().title() for state in state_matches]
states_set.update(state_matches)
states_list = list(states_set)
# Expand any abbreviations. If the state is already its full name, we're just gonna
# default to that. E.g., "CT" -> "Connecticut", "Connecticut" -> "Connecticut"
states_list = [STATE_MAP.get(state, state) for state in states_list]
if len(states_list) > 1:
state = "Multiple"
elif len(states_list) == 1:
state = states_list[0]
else:
state = "N/A"
return state
def state_check(states: str | list[str], text_dict: dict[str, str]) -> str:
if not isinstance(states, str) and not isinstance(states,list):
states = 'N/A'
if (
isinstance(states, list) and all(isinstance(state, str) for state in states)
) or isinstance(states, str): # If it's a list of strings or a string...
if isinstance(states, str): # If it's a string, turn it to a list (even a singleton)
if "," in states:
states = states.split(",")
else:
states = [states]
result_states = []
for state in states:
state = state.strip().replace(".", "") # Remove "." in cases like "N.Y."
# fix 1
if state == "Hlorida":
state = "Florida"
# fix 2 - LLM self-talking
if any(
kw.lower() in state.lower()
for kw in keywords_to_detect_invalid_state_name
):
state = "N/A"
if state.upper() in STATE_MAP.keys(): # Check if `state` is a state abbreviation.
result_states.append(STATE_MAP[state.upper()].title())
elif state.strip().title() in STATE_MAP.values(): # Check if `state` is already an expanded name state from the validation set.
result_states.append(state.strip().title())
else:
result_states.append("N/A")
result_states = set(result_states)
result_states.discard("N/A") # Discard any N/As (so if `result_states` is ['Illinois', 'N/A'], it returns 'Illinois' instead of 'Multiple')
if len(result_states) == 1:
return list(result_states)[0]
elif len(result_states) > 1:
return "Multiple"
elif len(result_states) == 0:
state = get_health_plan_state(text_dict) # search for state(s) in contract. If it's not found, `N/A` will be returned
return state
raise TypeError(
f"Expected a string or list of strings, got {type(states).__name__}"
)
def clean_health_plan_state_from_hotfix(
df: pd.DataFrame,
fieldExtraction/src/qa_qc_helpers.py
+1 -1
View File
@@ -10,7 +10,7 @@ import config
import io_utils
import preprocessing_funcs
import string_funcs
from hotfix_helper_funcs import keywords_to_detect_invalid_state_name
from valid import keywords_to_detect_invalid_state_name
from valid import AC_IND_LANG_COLS, B_IND_LANG_COLS, STATE_MAP
fieldExtraction/src/valid.py
+10
View File
@@ -595,6 +595,16 @@ METHODOLOGY_FILTER = [
"if there is no established payment",
]
keywords_to_detect_invalid_state_name = [ #This is for health plan state when the LLM talks to itself
"suggest",
"The Contract Text",
"refers",
"For ",
"Answer",
"Multiple",
"Statewide",
]
STATE_MAP = {
"AL": "Alabama",
"AK": "Alaska",
fieldExtraction/tests/cnc_hotfix_test.py
+1 -1
View File
@@ -3,7 +3,7 @@ import math
import pandas as pd
from pandas.testing import assert_frame_equal
from hotfix_helper_funcs import state_check
# from scripts.cnc_hotfixes.hotfix_helper_funcs import state_check
# class TestStateCheck: