From 1098be5cf5e0c12f8ecaf8854e493458a63bd056 Mon Sep 17 00:00:00 2001 From: Karan Desai Date: Wed, 13 Nov 2024 17:08:19 +0000 Subject: [PATCH] Merged in feature_ac_chunking_clean (pull request #265) grouped keywords and IRS regex chunking * abandon old branch and rebuild * removed print for field names, prompts * pushing as part of leftovers * IRS regex chunking * Merged in feature/chunk_term_clean (pull request #262) chunk_log_added * chunk_log_added * Merge remote-tracking branch 'remotes/origin/feature_ac_chunking_clean' into feature/chunk_term_clean Approved-by: Alex Galarce * added function for regex based IRS chunking * shifted regex_match_chunk function to utils * added helper functions for regex based chunking * replaced tin_regex function to ac_funcs * removed prompt for irs_others * added regex chunking for irs * removed irs group from smart chunking * testing functionality for regex based irs fields * updated return N/A condition * Merged main into feature_ac_chunking_clean * file_processing.py edited online with Bitbucket * added back conditional prompts * minor fixed for PR * remove install types * mering clean branch * with passed test casses * added flag for regex based tin execution Approved-by: Michael McGuinness --- .gitignore | 2 + fieldExtraction/pyproject.toml | 5 +- .../scripts/sample_smart_chunks.py | 87 +++++++ fieldExtraction/scripts/tin_test.py | 41 ++++ fieldExtraction/src/ac_funcs.py | 76 ++++++ fieldExtraction/src/ac_smart_chunking.py | 185 ++++++++++++++- fieldExtraction/src/conditional_funcs.py | 2 +- fieldExtraction/src/config.py | 31 ++- fieldExtraction/src/consolidate_output.py | 26 ++- fieldExtraction/src/file_counting.py | 10 +- fieldExtraction/src/file_processing.py | 220 +++++++++++++----- fieldExtraction/src/keywords.py | 219 +++++++++-------- fieldExtraction/src/main.py | 12 +- fieldExtraction/src/postprocess.py | 27 ++- fieldExtraction/src/postprocessing_funcs.py | 69 +++--- fieldExtraction/src/preprocess.py | 7 +- fieldExtraction/src/preprocessing_funcs.py | 115 ++++++++- fieldExtraction/src/prompts.py | 5 +- fieldExtraction/src/test.py | 16 +- fieldExtraction/src/top_down_funcs.py | 43 ++-- fieldExtraction/src/utils.py | 63 ++++- fieldExtraction/src/valid.py | 11 +- fieldExtraction/tests/test_chunking.py | 166 +++++++++++++ 23 files changed, 1147 insertions(+), 291 deletions(-) create mode 100644 fieldExtraction/scripts/sample_smart_chunks.py create mode 100644 fieldExtraction/scripts/tin_test.py create mode 100644 fieldExtraction/tests/test_chunking.py diff --git a/.gitignore b/.gitignore index 24a4a3c..dad589e 100644 --- a/.gitignore +++ b/.gitignore @@ -92,3 +92,5 @@ docs/ .mypy_cache/ .pytest_cache/ __pycache__/ + +new/ \ No newline at end of file diff --git a/fieldExtraction/pyproject.toml b/fieldExtraction/pyproject.toml index 0811440..ef9964a 100644 --- a/fieldExtraction/pyproject.toml +++ b/fieldExtraction/pyproject.toml @@ -30,10 +30,9 @@ target-version = ['py312'] [tool.mypy] disable_error_code = ["import-untyped","assignment","name-defined","call-arg","var-annotated","attr-defined","arg-type"] -install_types = true -non_interactive = true [tool.pytest.ini_options] testpaths = [ "tests" -] \ No newline at end of file +] +pythonpath = ["src"] \ No newline at end of file diff --git a/fieldExtraction/scripts/sample_smart_chunks.py b/fieldExtraction/scripts/sample_smart_chunks.py new file mode 100644 index 0000000..2faa833 --- /dev/null +++ b/fieldExtraction/scripts/sample_smart_chunks.py @@ -0,0 +1,87 @@ +from utils import read_local +import preprocess +import preprocessing_funcs +import keywords +import argparse +import os +from keywords import KEYWORD_MAPPINGS +from collections import defaultdict + +# import tqdm + + +def parse_arguments(): + parser = argparse.ArgumentParser(description="Smart Chunking Tester") + parser.add_argument( + "--input_dir", help="Input directory (local path or S3 URI)", default="src/ip2" + ) + parser.add_argument( + "--output_dir", help="Output directory (local)", default="src/output_chunks" + ) + parser.add_argument("--keyword", help="Keyword to be used for chunking") + parser.add_argument( + "--case_sensitive", + help="case sensitivity", + action=argparse.BooleanOptionalAction, + ) + return parser.parse_args() + + +def main(): + args = parse_arguments() + input_dir = args.input_dir + output_dir = args.output_dir # src/output_chunks + case_sensitive = args.case_sensitive + + if case_sensitive is None: # This shouldn't have to be assigned + case_sensitive = False + + keyword = args.keyword + keyword = list(map(str, keyword.split(","))) + keyword = [kw.strip() for kw in keyword] + print("Generating Chunks for keywords --> ", keyword) + print("Case sensitive -->", case_sensitive) + print(type(case_sensitive)) + + kws = KEYWORD_MAPPINGS + + retained_rates = [] + + for file in os.listdir(input_dir): + full_path = os.path.join(input_dir, file) + contract_text = read_local(full_path) + contract_text = preprocessing_funcs.clean_newlines(contract_text) + contract_text = preprocessing_funcs.clean_law_symbols(contract_text) + text_dict = preprocessing_funcs.split_text( + contract_text + ) # return a dictionary with keys - page_num (str), values as the page_text + + test_chunck = preprocessing_funcs.smart_chunk_ac( + text_dict=text_dict, + keyword_mappings={ + "place_holder": { + "methodology": "or", + "keywords": keyword, + "case_sensitive": case_sensitive, + } + }, + ) + + retained_rate = len(test_chunck["place_holder"]) / len(contract_text) + retained_rates.append(retained_rate) + + print( + f"\nfrom file {file} {len(contract_text)} characters were retrieved;\n{len(test_chunck['place_holder'])} were retained by smart chunking on keywords:\n{keyword}" + ) + print(f"Smart chunking reduced the document by {100*(1-retained_rate):0.2f}%") + + with open(f"{output_dir}/{file}_CHUNKED.txt", "w") as f: + f.write(test_chunck["place_holder"]) + + print("avg reduction:", 1 - (sum(retained_rates) / len(retained_rates))) + print("max retention:", max(retained_rates)) + print("min retention:", min(retained_rates)) + + +if __name__ == "__main__": + main() diff --git a/fieldExtraction/scripts/tin_test.py b/fieldExtraction/scripts/tin_test.py new file mode 100644 index 0000000..c39a1e1 --- /dev/null +++ b/fieldExtraction/scripts/tin_test.py @@ -0,0 +1,41 @@ +import argparse +import os +from utils import read_local +import preprocessing_funcs +import ac_funcs + + +def parse_arguments(): + parser = argparse.ArgumentParser(description="Smart Chunking Tester") + parser.add_argument( + "--input_dir", help="Input directory (local path or S3 URI)", default="src/new/" + ) + return parser.parse_args() + + +def main(): + args = parse_arguments() + input_dir = args.input_dir + + for file in os.listdir(input_dir): + full_path = os.path.join(input_dir, file) + print("--" * 20) + print(file) + contract_text = read_local(full_path) + contract_text = preprocessing_funcs.clean_newlines(contract_text) + contract_text = preprocessing_funcs.clean_law_symbols(contract_text) + text_dict = preprocessing_funcs.split_text( + contract_text + ) # return a dictionary with keys - page_num (str), values as the page_text + + irs_answers = ac_funcs.tin_regex(filename=str(file), text_dict=text_dict) + if irs_answers is None: + irs_answers = {} + irs_answers["PROV_GROUP_TIN"] = "N/A" + irs_answers["PROV_TIN_OTHER"] = "N/A" + + print(irs_answers) + + +if __name__ == "__main__": + main() diff --git a/fieldExtraction/src/ac_funcs.py b/fieldExtraction/src/ac_funcs.py index 1a29b51..9c1c2c4 100644 --- a/fieldExtraction/src/ac_funcs.py +++ b/fieldExtraction/src/ac_funcs.py @@ -17,6 +17,8 @@ import sys import claude_funcs import dict_operations import config +import utils +import prompts def get_ac_answer(prompt, filename, fields): @@ -314,3 +316,77 @@ def json_parsing_search(response_text, field_list): ) return dict(zip(field_l, answer_l)) + + +def IRS_prompt(filename, text_dict, page_list, matches): + """Prompts LLM to get Provider's TIN from the all the tins found via regex matches + + Args: + filename: Name of the contract + text_dict (dict): Dictionary keyed by string page-number and valued by actual textract output page + page_list(list): List of pages where the match was found + matches(list): List of regex match(s) + + Returns: + provider_tin(str): Provider TIN + other_tins(str): All other TIN(s) present in the contract + """ + questions = {} + questions["PROV_GROUP_TIN"] = prompts.AC_DICT["PROV_GROUP_TIN"] + + context = utils.chunk_selected_pages(text_dict, page_list) + prompt = prompts.AC_SINGLE_FIELD_TEMPLATE(context, questions) + provider_tin = claude_funcs.invoke_claude( + prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, 8192 + ) + matches.remove(provider_tin) + other_tins = ", ".join(matches) + + return provider_tin, other_tins + + +def tin_regex(filename, text_dict: dict[str, str]): + """Returns dictionery with Provider and other TIN(s) based on regex matches and conditional prompting + + Args: + filename: Name of the contract + text_dict (dict): Dictionary keyed by string page-number and valued by actual textract output page + + Returns: + TIN_dict(dict): A dictionery with keys 'PROV_GROUP_TIN' & 'PROV_TIN_OTHER' and values as respective TIN(s) + """ + TIN_dict = {} + + patterns = [r"\b\d{2}-\d{7}\b", r"\b\d{9}\b"] + + for pattern in patterns: + matches, page_list = utils.find_regex_matches(pattern, text_dict) + matches = list(set(matches)) + + if matches: + if len(matches) == 1: + TIN_dict["PROV_GROUP_TIN"] = matches[0] + TIN_dict["PROV_TIN_OTHER"] = "N/A" + return TIN_dict + + elif len(matches) >= 1: + irs_tin, other_tins = IRS_prompt( + filename, text_dict, page_list, matches + ) + TIN_dict["PROV_GROUP_TIN"] = irs_tin + TIN_dict["PROV_TIN_OTHER"] = other_tins + return TIN_dict + + # If no matches were found in the text_dict, check the filename + for pattern in patterns: + tin_from_title = re.findall(pattern, filename) + # TODO: What do we do when there's more than one TIN matches in the title? + if len(tin_from_title) == 1: + TIN_dict["PROV_GROUP_TIN"] = tin_from_title[0] + TIN_dict["PROV_TIN_OTHER"] = "N/A" + return TIN_dict + + # If no matches found + TIN_dict["PROV_GROUP_TIN"] = "N/A" + TIN_dict["PROV_TIN_OTHER"] = "N/A" + return TIN_dict diff --git a/fieldExtraction/src/ac_smart_chunking.py b/fieldExtraction/src/ac_smart_chunking.py index 49a62c4..101c069 100644 --- a/fieldExtraction/src/ac_smart_chunking.py +++ b/fieldExtraction/src/ac_smart_chunking.py @@ -1,10 +1,32 @@ -def chunk_hierarchical(text_dict, key_dict): - # Look for the keywords, one-by-one, and take the first that satisfies the keyword +import re + + +def chunk_hierarchical( + text_dict: dict[str, str], keywords: list[str], case_sensitive: bool = False +) -> list[str]: + """Look for the keywords, one-by-one, and take the first that satisfies the keyword. + We pad responsive pages with a one-page buffer (before and after) when possible. + + Args: + text_dict (dict[str, str]): A dictionary keyed by string page number and valued + by string page contents + keywords (list[str]): A list of string keywords + case_sensitive(bool): A flag for converting all keywords and text to lower case, + in order to ignore case altogether. + + Returns: + list[str]: The sorted list of string page numbers responsive to the first + keyword found + """ + if not case_sensitive: + keywords = [keyword.lower() for keyword in keywords] + text_dict = {key: value.lower() for key, value in text_dict.items()} + page_list = [] - for keyword in key_dict["keywords"]: + for keyword in keywords: if len(page_list) == 0: for page in text_dict.keys(): - if keyword.lower() in text_dict[page].lower(): + if keyword in text_dict[page]: if str(int(page) - 1) in text_dict.keys(): page_list.append(str(int(page) - 1)) @@ -14,17 +36,35 @@ def chunk_hierarchical(text_dict, key_dict): page_list.append(str(int(page) + 1)) page_list_sorted = sorted([int(page_str) for page_str in set(page_list)]) - return page_list_sorted + page_list_sorted_str = list(map(str, page_list_sorted)) + return page_list_sorted_str -def chunk_or(text_dict, key_dict): - # Look for any of the keywords and add any pages with any keywords to the chunk - # When we add a page, we add the previous, current, and next page +def chunk_or( + text_dict: dict[str, str], keywords: list[str], case_sensitive: bool = False +) -> list[str]: + """Look for any of the keywords and add any pages with any keywords to the chunk. + We pad responsive pages with a one-page buffer (before and after) when possible. + + Args: + text_dict (dict[str, str]): A dictionary keyed by string page number and valued + by string page contents + keywords (list[str]): A list of string keywords + case_sensitive(bool): A flag for converting all keywords and text to lower case, + in order to ignore case altogether. + + Returns: + list[str]: The sorted list of string page numbers responsive to the first + keyword found + """ + if not case_sensitive: + keywords = [keyword.lower() for keyword in keywords] + text_dict = {key: value.lower() for key, value in text_dict.items()} + page_list = [] - # for page in text_dict.values(): for page in text_dict.keys(): - for keyword in key_dict["keywords"]: - if keyword.lower() in text_dict[page].lower(): + for keyword in keywords: + if keyword in text_dict[page]: # Add buffer of previous and next page, if we're not at the beginning nor end # If we are at the beginning (or the end) just take the next (or previous) page if str(int(page) - 1) in text_dict.keys(): @@ -36,4 +76,125 @@ def chunk_or(text_dict, key_dict): page_list.append(str(int(page) + 1)) page_list_sorted = sorted([int(page_str) for page_str in set(page_list)]) - return page_list_sorted + page_list_sorted_str = list(map(str, page_list_sorted)) + return page_list_sorted_str + + +# And: run each keyword and include a page if all are present +def chunk_and( + text_dict: dict[str, str], keywords: list[str], case_sensitive: bool = False +) -> list[str]: + """Look for all of the keywords and add any pages with all keywords to the chunk. + We pad responsive pages with a one-page buffer (before and after) when possible. + + Args: + text_dict (dict[str, str]): A dictionary keyed by string page number and valued + by string page contents + keywords (list[str]): A list of string keywords + case_sensitive(bool): A flag for converting all keywords and text to lower case, + in order to ignore case altogether. + + Returns: + list[str]: The sorted list of string page numbers responsive to the first + keyword found + """ + if not case_sensitive: + keywords = [keyword.lower() for keyword in keywords] + text_dict = {key: value.lower() for key, value in text_dict.items()} + + page_list = [] + for page in text_dict.keys(): + if all([keyword in text_dict[page] for keyword in keywords]): + # Add buffer of previous and next page, if we're not at the beginning nor end + # If we are at the beginning (or the end) just take the next (or previous) page + if str(int(page) - 1) in text_dict.keys(): + page_list.append(str(int(page) - 1)) + + page_list.append(page) + + if str(int(page) + 1) in text_dict.keys(): + page_list.append(str(int(page) + 1)) + + page_list_sorted = sorted([int(page_str) for page_str in set(page_list)]) + page_list_sorted_str = list(map(str, page_list_sorted)) + return page_list_sorted_str + + +# Regex: Check for a regex and include a page if the regex matches within the page +def chunk_regex(text_dict: dict[str, str], regex: str) -> list[str]: + """Look for regex match and add any pages with regex match to the chunk. + We pad responsive pages with a one-page buffer (before and after) when possible. + + Args: + text_dict (dict[str, str]): A dictionary keyed by string page number and valued + by string page contents + regex (str): A regex expression + + Returns: + list[str]: The sorted list of string page numbers responsive to the first + keyword found + """ + + page_list = [] + for page in text_dict.keys(): + if re.search(regex, text_dict[page]): + # Add buffer of previous and next page, if we're not at the beginning nor end + # If we are at the beginning (or the end) just take the next (or previous) page + if str(int(page) - 1) in text_dict.keys(): + page_list.append(str(int(page) - 1)) + + page_list.append(page) + + if str(int(page) + 1) in text_dict.keys(): + page_list.append(str(int(page) + 1)) + + page_list_sorted = sorted([int(page_str) for page_str in set(page_list)]) + page_list_sorted_str = list(map(str, page_list_sorted)) + return page_list_sorted_str + + +def keyword_search( + text_dict: dict[str, str], + keywords: list[str], + case_sensitive: bool = False, + regex: bool = False, +) -> dict[str, list]: + """Look for keyword/regex match and add any pages with keyword/regex match to the chunk. + + Args: + text_dict (dict[str, str]): A dictionary keyed by string page number and valued + by string page contents + keywords (list[str]): A list of string keywords/regex + case_sensitive(bool): A flag for converting all keywords and text to lower case, + in order to ignore case altogether. + regex (bool): A flag to identify if the passed keyword list is regex list or string list + + Returns: + list[str]: The dictionary containing keyword as keys and sorted list of string page numbers responsive to the + keyword found + """ + if not case_sensitive and not regex: + keywords = [keyword.lower() for keyword in keywords] + text_dict = {key: value.lower() for key, value in text_dict.items()} + + page_dict = {} + for keyword in keywords: + page_dict[keyword] = [] + + for page in text_dict.keys(): + if (not regex and keyword in text_dict[page]) or ( + regex and re.search(keyword, text_dict[page]) + ): + # if str(int(page) - 1) in text_dict.keys(): + # page_dict[keyword].append(str(int(page) - 1)) + page_dict[keyword].append(page) + # if str(int(page) + 1) in text_dict.keys(): + # page_dict[keyword].append(str(int(page) + 1)) + + page_dict_sorted = {} + for keyword in page_dict: + page_dict_sorted[keyword] = sorted( + [int(page_str) for page_str in set(page_dict[keyword])] + ) + + return page_dict_sorted diff --git a/fieldExtraction/src/conditional_funcs.py b/fieldExtraction/src/conditional_funcs.py index 699c585..7ce796c 100644 --- a/fieldExtraction/src/conditional_funcs.py +++ b/fieldExtraction/src/conditional_funcs.py @@ -35,7 +35,7 @@ def run_conditional(combined_results, text_dict, filename): rbrvs_answer = claude_funcs.invoke_claude( prompt, config.MODEL_ID_CLAUDE2, filename, max_tokens=16 ) - + d['Inclusion of essential RBRVS "Fee Source" Language (Y/N)'] = ( rbrvs_answer.strip() ) diff --git a/fieldExtraction/src/config.py b/fieldExtraction/src/config.py index e58104f..c7fa306 100644 --- a/fieldExtraction/src/config.py +++ b/fieldExtraction/src/config.py @@ -11,14 +11,15 @@ import requests from requests.adapters import HTTPAdapter from requests.packages.urllib3.util.retry import Retry + ######################################## UTILS ################################################### def check_s3_bucket_exists(s3_client, bucket_name: str): try: s3_client.head_bucket(Bucket=bucket_name) print(f"{bucket_name} exists") except s3_client.exceptions.ClientError as e: - error_code = e.response['Error']['Code'] - if error_code == '404': + error_code = e.response["Error"]["Code"] + if error_code == "404": print(f"{bucket_name} does not exist") raise else: @@ -34,6 +35,7 @@ def check_s3_bucket_exists(s3_client, bucket_name: str): TEST = False # True to run test prompt - just for testing model connection TODAY = datetime.now().strftime("%Y%m%d") + ######################################## SYSTEM ARGUMENTS ######################################## # read_mode, run_mode, input_dir, output_dir, consolidated_output_dir @@ -50,7 +52,8 @@ def get_arg_value(arg_name, default): # Run config args RUN_MODE = get_arg_value("run_mode", "ec2") # Valid: local or ec2 -READ_MODE = get_arg_value("read_mode", "local") # Valid: local or s3 + +READ_MODE = get_arg_value("read_mode", "s3") # Valid: local or s3 BATCH_ID = get_arg_value("batch_id", "test_batch") # Field args @@ -70,7 +73,7 @@ OUTPUT_DIRECTORY = get_arg_value( "output_dir", "output_individual" ) # Valid: any valid directory TRACKING_FILE = get_arg_value("tracking_file", f"{BATCH_ID}_tracking.csv") -WRITE_TO_S3 = get_arg_value("write_to_s3", False) == "True" # Valid: True or False +WRITE_TO_S3 = get_arg_value("write_to_s3", False) == "True" # Valid: True or False S3_OUTPUT_BUCKET = get_arg_value("s3_output_bucket", default="doczy-output") # Filtering args @@ -82,6 +85,9 @@ FILTER_ON_CSV = get_arg_value("filter_csv", False) == "True" # Valid: True or F # Multithreading args MAX_WORKERS = int(get_arg_value("max_workers", 2)) # Valid: any integer +# default false for the regex based IRS chunking +IRS_REGEX_CHECK = get_arg_value("irs_regex_check", False) == "True" + ######################################## CLIENT SETTINGS ######################################## CLIENT_NAME = "" @@ -110,9 +116,9 @@ TABLE_ANALYSIS_NAME = f"{BATCH_ID}-Table-Analysis.csv" # Bedrock if RUN_MODE == "local": - AWS_ACCESS_KEY_ID="ASIA6GBMBVWONKTNLEQK" - AWS_SECRET_ACCESS_KEY="OJ/mV2gfHAXHA62dNpRP7HBF0AmwPhTSx9Lx7GKW" - AWS_SESSION_TOKEN="IQoJb3JpZ2luX2VjEOv//////////wEaCXVzLWVhc3QtMiJGMEQCIB52FvtJu1yJRF4m7p4Gl9APRpI4xCx8h/y+yshFMXbnAiAPktTJSCX0XF2bXAVQUftZD3VGXsZqlOedaKMNNNXHASqJAwhkEAAaDDk3NTA0OTk2MDg2MCIMI11ogWEEpCUL9KjOKuYCIyD27hNGF+jBMlWnEEn34k12WZd0/ZvYyGUqg7EYGj9DiznQ1D90+ynSHQjTiqEuTbnmhM58pHnhn5rSiB7LVVZldKHYZm2s9LXyw5lnPyytR7xI0JgiExeJGBePd12YtqCf3iDDh81hT5IPcW3kQCZ7IM/LQC6kBlT6j+dvD8DeRRYG/uH1ngkLNHj/1BUVYZohA5mwkUmle21IeVoLIArtgx2k5wWNV0QT2GM3tuDeCCCGqmZOomau5ffMJIWm+fgIJam9ckUl2IFyX0Vn/AX906tz3FZJIUdzxFcX+8wn1/yfvT1jpXXWSFekS7d/BaM6zRwAcRA2FMK4iiBITAmZm1FnxWDC0b4atdbst7KvYtuIUJODDeh7bzAotFAe6faZvIwnAYxkadMJ8rMpdamldVb+/HfwX7gYbyAs9qHjAbam2L+ZWVycMzDL68TjL2RF/gRO+Ts4uTzcihyis1bGxS6wDDDX14S5BjqnAdIuu67CAQs0ynGkmI+7Q39o6F2lg1/l0638QhjY7VugNcqhxNN2zdu8Njc0kZLwa72QsekjhUZ59cBRzV7z1zY4UdT6eGQJbR1ovNd0PsxV/BQ+PH60G+aKJuYAx1rpdz81r4cX3tal5gzktYNN9XujCA4Dts9inRl2pka6BrmuH+GyO5K9Ba9Kjrom3+bVz0QdGEsEvnQWt+AA5NMEUEXCblO9SiZc" + AWS_ACCESS_KEY_ID="ASIA6GBMBVWOC2KZPWDJ" + AWS_SECRET_ACCESS_KEY="Ee0mAnRHA03wWxFaM4NQEy1PgdXotF9OV7b5D41Y" + AWS_SESSION_TOKEN="IQoJb3JpZ2luX2VjEFAaCXVzLWVhc3QtMiJHMEUCIQDfNitr9GFdwaTb5tOSyNWsU9uuButk54yNlHGxt8FxogIgCq1IqDKHRdm1YUR+ddAFOt7RBAw2KnZZ7EOEZxgsu34qigMI2f//////////ARAAGgw5NzUwNDk5NjA4NjAiDMn0VL/snLYOd/XIPSreAlJkUy5fBZKgnhx/mmD6oC+wlCebfHCEhzmdrBJkuxPVBNgV9ebKVLUIP2fyRE+MK+/P//+RbjZzxolcLYoM4HIs6EVNO2fa/NSjTfv7IVnUAGDbS+jVUbr8ICNLz2eo8mDCm8J0SS5kTpVWhuZ6kJhxvOmN9y2U/RLCCbhxAMaIrNC1/bS8JQODSBsP25aDtdJflBazk6jSwiNry0ka6gTkmdk0R4RYXccCv7mUdmIc5V+6eH1a7qslbVdDpS+hsrU/42Jsg57CyR+Tz7a0zNixUhwXj1s8Ma8ygzkTJKevXF2W3AlJ2XK1WIwegbD6CDroFKd7USr+RrY+2Ftr+8dWqYhvt00UAHGSBpNpffkqJFcOTnUiD+6pipymwL8OYk1Pc+LB2QfNYTZgQYWkofqAx4A6NEtk4cmO0qSHpQoiBrD+sv/3TE1pShlhFnRo1Ij99byJrqSBoIUNXNS2MKyj07kGOqYBoPi0JfB2A3gmYsTmYLGMLbnfET8vr2E8WVUSr5NMrbm4fo3Yc4W3gISOVToncH254EqO7Q+gakUAi65w8JgOjQlPkM+u9Knpetw27Ym38qkxkEQ7Htk1zbEXgYkm4iQyaPXBFtD6d3Nsv2gOGQRhXgC4uLuZvqdrVPF+msYU4Cwi2rLpmsvBlbI11DIPvAapJsNz7sK+GNhXdJefP49RK/pYAH10+w==" config = Config(read_timeout=2000) BEDROCK_RUNTIME = boto3.client( service_name="bedrock-runtime", @@ -123,10 +129,13 @@ if RUN_MODE == "local": config=config, ) if WRITE_TO_S3: - S3_CLIENT = boto3.client("s3", region_name="us-east-2", - aws_access_key_id=AWS_ACCESS_KEY_ID, - aws_secret_access_key=AWS_SECRET_ACCESS_KEY, - aws_session_token=AWS_SESSION_TOKEN,) + S3_CLIENT = boto3.client( + "s3", + region_name="us-east-2", + aws_access_key_id=AWS_ACCESS_KEY_ID, + aws_secret_access_key=AWS_SECRET_ACCESS_KEY, + aws_session_token=AWS_SESSION_TOKEN, + ) elif RUN_MODE == "ec2": config = Config(read_timeout=2000, max_pool_connections=60) diff --git a/fieldExtraction/src/consolidate_output.py b/fieldExtraction/src/consolidate_output.py index 491d683..3f30b45 100644 --- a/fieldExtraction/src/consolidate_output.py +++ b/fieldExtraction/src/consolidate_output.py @@ -7,6 +7,7 @@ import valid from pathlib import Path from datetime import datetime + def consolidate_output(): Path(config.CONSOLIDATED_OUTPUT_DIRECTORY).mkdir(parents=True, exist_ok=True) @@ -50,7 +51,9 @@ def consolidate_output(): # If ABC if "a" in config.FIELDS and "c" in config.FIELDS and "b" in config.FIELDS: - b_final_df.drop(['Pages', 'Parent Agreement Code'], axis=1, inplace=True) # Drop Pages so only one in result + b_final_df.drop( + ["Pages", "Parent Agreement Code"], axis=1, inplace=True + ) # Drop Pages so only one in result abc_path = os.path.join( config.CONSOLIDATED_OUTPUT_DIRECTORY, f"{config.BATCH_ID}-ABC.csv" @@ -61,26 +64,32 @@ def consolidate_output(): abc_final_df = abc_final_df[ [col for col in valid.ABC_COLUMNS if col in abc_final_df] ] - - print("Columns not found: ", [f for f in valid.ABC_COLUMNS if f not in abc_final_df.columns]) + + print( + "Columns not found: ", + [f for f in valid.ABC_COLUMNS if f not in abc_final_df.columns], + ) abc_final_df.to_csv(abc_path) print(f"Consolidation complete. Results written to {abc_path}") - # Write to S3, if applicable # Limitataion: it writes everything in the output directory, regardless of if it's # new if config.WRITE_TO_S3: BUCKET_NAME = config.S3_OUTPUT_BUCKET - s3_prefix = datetime.now().strftime(f"run_%Y%m%d_%H:%M_{config.BATCH_ID}") # timestamped prefix + s3_prefix = datetime.now().strftime( + f"run_%Y%m%d_%H:%M_{config.BATCH_ID}" + ) # timestamped prefix print(f"Uploading to s3://{BUCKET_NAME}/{s3_prefix}") for root, _, files in os.walk(config.CONSOLIDATED_OUTPUT_DIRECTORY): for filename in files: # get full local path local_file_path = os.path.join(root, filename) - + # Get path in bucket - relative_path = os.path.relpath(local_file_path, config.CONSOLIDATED_OUTPUT_DIRECTORY) + relative_path = os.path.relpath( + local_file_path, config.CONSOLIDATED_OUTPUT_DIRECTORY + ) s3_key = os.path.join(s3_prefix, relative_path).replace("\\", "/") # upload file @@ -89,6 +98,5 @@ def consolidate_output(): config.S3_CLIENT.upload_file(local_file_path, BUCKET_NAME, s3_key) except Exception as e: print(f"Error uploading {local_file_path}: {str(e)}") - - print(f"Upload completed to s3://{BUCKET_NAME}/{s3_prefix}") + print(f"Upload completed to s3://{BUCKET_NAME}/{s3_prefix}") diff --git a/fieldExtraction/src/file_counting.py b/fieldExtraction/src/file_counting.py index 082b93b..c19f06e 100644 --- a/fieldExtraction/src/file_counting.py +++ b/fieldExtraction/src/file_counting.py @@ -16,9 +16,7 @@ print(f"Total Input Files : {len(input_dict)}") # Filter B input_dict_b = { - k: v - for k, v in input_dict.items() - if utils.contains_reimbursement(str(v)) + k: v for k, v in input_dict.items() if utils.contains_reimbursement(str(v)) } # Filter out non-contracts print( f"B Input Files after reimbursement filter: {len(input_dict_b)} | {total_files-len(input_dict_b)} files removed" @@ -26,6 +24,8 @@ print( # Filter already processed if config.FILTER_ALREADY_PROCESSED: - input_dict_ac, input_dict_b = utils.filter_already_processed(input_dict, input_dict_b) + input_dict_ac, input_dict_b = utils.filter_already_processed( + input_dict, input_dict_b + ) print(f"AC Input Files left to be processed : {len(input_dict_ac)}") - print(f"B Input Files left to be processed : {len(input_dict_b)}") \ No newline at end of file + print(f"B Input Files left to be processed : {len(input_dict_b)}") diff --git a/fieldExtraction/src/file_processing.py b/fieldExtraction/src/file_processing.py index 5f1d9d3..cb78647 100644 --- a/fieldExtraction/src/file_processing.py +++ b/fieldExtraction/src/file_processing.py @@ -2,6 +2,7 @@ import os import pandas as pd import time import csv +import json import config import keywords @@ -22,91 +23,142 @@ import claude_funcs def run_ac_prompts(file_object): + ac_answers_dict = {} + log_data = [] + chunk_data = [] + ################## INITIATE PROCESSING ################## filename, contract_text = file_object print(f"Processing AC for {filename}...") - - ################## PREPROCESS ################## + print(f"Total contract word count: {len(contract_text.split())}") + + ################## PREPROCESS - SMART CHUNK ################## text_dict, exhibit_pages, num_pages, ac_chunks = preprocess.preprocess( contract_text, filename, fields="ac" ) print(f"AC Preprocessing Complete - {filename}") - ################## RUN FULL CONTEXT PROMPTS ################## - ac_dict = {} - - full_context_fields = [ - field for field in prompts.AC_DICT if field not in keywords.KEYWORD_MAPPINGS - ] - - full_context_questions = { - field: prompts.AC_DICT.get(field) for field in full_context_fields - } - - if full_context_questions: - full_context_prompt = prompts.AC_MULTI_FIELD_TEMPLATE(contract_text[0 : min(100000, len(contract_text) - 1)], full_context_questions) - - try: - full_context_answers = ac_funcs.get_ac_answer( - full_context_prompt, - filename, - fields=list(full_context_questions.keys()), - ) - ac_dict.update(full_context_answers) - - except Exception as e: - print(f'Exception in AC Full Context : {e}') + ################## RUN REGEX FUNCTIONALITY ################## + if config.IRS_REGEX_CHECK == True: + irs_answers = ac_funcs.tin_regex(filename, text_dict) + ac_answers_dict.update(irs_answers) ################## RUN CHUNKED PROMPTS ################## - all_fields = set(prompts.AC_DICT.keys()) - chunked_fields = [field for field in keywords.KEYWORD_MAPPINGS] - for field in chunked_fields: - if field in prompts.AC_DICT: - question = prompts.AC_DICT[field] - else: - print(f"Field {field} not found in prompts. Skipping...") - continue + ## Check `ac_funcs.create_prompt()` vs. `prompts.AC_*_template()` + all_fields = set(prompts.AC_DICT.keys()) + + field_groups = [ + field_group for field_group in keywords.GROUPED_KEYWORD_MAPPINGS.keys() + ] + + for field_group in field_groups: + keyword_dict = keywords.GROUPED_KEYWORD_MAPPINGS[field_group]["keywords"] + fields = keywords.GROUPED_KEYWORD_MAPPINGS[field_group]["fields"] + questions = {} + for field in fields: + questions[field] = prompts.AC_DICT[field] # Use chunk if available and different from full context, otherwise use full context if ( - field in ac_chunks - and ac_chunks[field].strip() - and ac_chunks[field] != contract_text + field_group in ac_chunks + and ac_chunks[field_group].strip() + and ac_chunks[field_group] != contract_text ): - context = ac_chunks[field] + context = ac_chunks[field_group] context = context[0 : min(100000, len(context)) - 1] context_type = "Smart Chunking" else: context = contract_text[0 : min(100000, len(contract_text) - 1)] context_type = "Full Context" - prompt = prompts.AC_SINGLE_FIELD_TEMPLATE(context, question) + if len(fields) == 1: + question = questions[ + fields[0] + ] # Take the one question for the single field included + prompt = prompts.AC_SINGLE_FIELD_TEMPLATE(context, question) + elif len(fields) > 1: + prompt = prompts.AC_MULTI_FIELD_TEMPLATE(context, questions) + else: + raise ValueError("Field group has no defined fields") + + # TODO: Make sure prompt looks good for groups try: - field_answer = claude_funcs.invoke_claude( + # field_group_answer_raw = {"test field": "test value"} ## uncomment this and comment below line to test keyword search + field_group_answer_raw = claude_funcs.invoke_claude( prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, 8192 ) - ac_dict[field] = field_answer + + if len(fields) == 1: + field_group_answer_dict = {field: field_group_answer_raw} + elif len(fields) > 1: + field_group_answer_dict = json.loads(field_group_answer_raw) + + # TODO: Validate field names coming out of Claude, check against fields in group, make sure they're all there, make sure there aren't any extras + for field in fields: + field_answer = field_group_answer_dict[field] + ac_answers_dict[field] = field_answer + + log_data.append( + { + "Field": field, + "Prompt": f"Question: {questions}\n\nContext: {context}", # Include full context + "Context Type": context_type, + "Response": field_answer, + } + ) + + chunk_data.append( + { + "Contract name": filename, + "Field group": field_group, + "Field list": fields, + "Methodology": ac_chunks[field_group + "_methodology"], + "Case sensitivity": ac_chunks[field_group + "_case"], + "Page list": ac_chunks[field_group + "_pages"], + "Chunk size": len(context), + "% Reduction": f"{100*(1-len(context)/len(contract_text)):0.2f}%", + } + ) except Exception as e: - print(f"Error processing field {field}: {str(e)}") + print(f"Error processing field {field}: {type(e)}, {str(e)}") + ac_answers_dict[field] = f"Error: {str(e)}" + log_data.append( + { + "Field": field, + "Prompt": f"Question: {questions}\n\nContext: {context}", # Include full context + "Context Type": context_type, + "Response": f"Error: {str(e)}", + } + ) + chunk_data.append( + { + "Contract name": filename, + "Field group": field_group, + "Field list": fields, + "Methodology": ac_chunks[field_group + "_methodology"], + "Case sensitivity": ac_chunks[field_group + "_case"], + "Page list": ac_chunks[field_group + "_pages"], + "Chunk size": len(context), + "% Reduction": f"{100*(1-len(context)/len(contract_text)):0.2f}%", + } + ) ################## AC CONDITIONAL PROMPTS ################## # Non-Renewal - Days nrd_prompt = prompts.AC_SINGLE_FIELD_TEMPLATE( - ac_dict["NON_RENEWAL_LANGUAGE"], prompts.AC_DICT["NON_RENEWAL_DAYS"] + ac_answers_dict["NON_RENEWAL_LANGUAGE"], prompts.AC_DICT["NON_RENEWAL_DAYS"] ) nrd_answer = claude_funcs.invoke_claude( nrd_prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, 8192 ) - ac_dict["NON_RENEWAL_DAYS"] = nrd_answer - + ac_answers_dict["NON_RENEWAL_DAYS"] = nrd_answer ## TODO - add Notice Provider Name/Address here - # Clean Up Contract Effective Date if ( - utils.is_empty(ac_dict["CONTRACT_EFFECTIVE_DT"]) - and "meridian" in ac_dict["PAYER_NAME"].lower() + utils.is_empty(ac_answers_dict["CONTRACT_EFFECTIVE_DT"]) + and "meridian" in ac_answers_dict["PAYER_NAME"].lower() ): date_prompt = prompts.AC_EFFECTIVE_DATE_CLEANUP( contract_text[0 : min(100000, len(contract_text)) - 1] @@ -114,11 +166,53 @@ def run_ac_prompts(file_object): date_answer = claude_funcs.invoke_claude( date_prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, 124 ) - ac_dict["CONTRACT_EFFECTIVE_DT"] = date_answer + ac_answers_dict["CONTRACT_EFFECTIVE_DT"] = date_answer + ################## RUN FULL CONTEXT PROMPTS ################## + # Process 'full_context' fields together + + smartly_chunked = [] + for field_groups in keywords.GROUPED_KEYWORD_MAPPINGS.keys(): + for fields in keywords.GROUPED_KEYWORD_MAPPINGS[field_groups]["fields"]: + smartly_chunked.append(fields) + full_context_fields = [ + field for field in prompts.AC_DICT if field not in smartly_chunked + ] + full_context_fields = [ + field for field in full_context_fields if field not in ac_answers_dict.keys() + ] + + full_context_questions = { + field: prompts.AC_DICT.get(field) for field in full_context_fields + } + if full_context_questions: + full_context_prompt = ac_funcs.create_prompt(contract_text[0:min(100000, len(contract_text)-1)], question=full_context_questions) + try: + full_context_answers = ac_funcs.get_ac_answer(full_context_prompt, filename, fields=list(full_context_questions.keys())) + ac_answers_dict.update(full_context_answers) + + for field, answer in full_context_answers.items(): + log_data.append({ + 'Field': field, + 'Prompt': f"Question: {full_context_questions[field]}\n\nContext: {contract_text}", # Include full context + 'Context Type': 'Full Context (High Accuracy)', + 'Response': answer + }) + except Exception as e: + print(f"Error processing high accuracy fields: {str(e)}") + for field in full_context_questions.keys(): + ac_answers_dict[field] = f"Error: {str(e)}" + log_data.append({ + 'Field': field, + 'Prompt': f"Question: {full_context_questions[field]}\n\nContext: {contract_text}", # Include full context + 'Context Type': 'Full Context (High Accuracy)', + 'Response': f"Error: {str(e)}" + }) + + ################## ENSURE ALL FIELDS ARE PRESENT ################## for field in all_fields: - if field not in ac_dict: + if field not in ac_answers_dict: print(f"Field not found in results. {field}") ################## CREATE OUTPUT DIRECTORIES ################## @@ -127,16 +221,34 @@ def run_ac_prompts(file_object): os.makedirs(output_dir, exist_ok=True) ################## POSTPROCESS ################## - for key in ac_dict.keys(): - print(key, ac_dict[key]) + for key in ac_answers_dict.keys(): + print(key, ac_answers_dict[key]) - ac_df = pd.DataFrame([ac_dict]) + ac_df = pd.DataFrame([ac_answers_dict]) ac_df = postprocess.ac_postprocess(ac_df, filename, num_pages) ################## WRITE TO OUTPUT ################## ac_df.to_csv(os.path.join(output_dir, config.AC_RESULTS_NAME), index=False) - return ac_dict + chunk_file_path = "chunk_log.csv" + with open(chunk_file_path, "a", newline="", encoding="utf-8") as chunk_file: + fieldnames = [ + "Contract name", + "Field group", + "Field list", + "Methodology", + "Case sensitivity", + "Page list", + "Chunk size", + "% Reduction", + ] + writer = csv.DictWriter(chunk_file, fieldnames=fieldnames) + writer.writeheader() + writer.writerows(chunk_data) + + print(f"Chunk log written to {chunk_file_path}") + + return ac_answers_dict def run_b_prompts(file_object): diff --git a/fieldExtraction/src/keywords.py b/fieldExtraction/src/keywords.py index 3e7be6c..8aef71b 100644 --- a/fieldExtraction/src/keywords.py +++ b/fieldExtraction/src/keywords.py @@ -1,81 +1,84 @@ -# KEYWORD_MAPPINGS = { -# "CONTRACT_AUTO_RENEWAL_IND": ["renew", "renewal", "auto renew", "auto renewal"], -# "CONTRACT_TERMINATION_DT": ["termination", "terminate"], -# "TERMINATION_UPON_NOTICE": ["notice", "upon notice"], -# "TERMINATION_UPON_CAUSE": ["cause", "material breach"], -# "CONTRACT_EFFECTIVE_DT": ["effective date", "effective"], -# "AMEND_CONTRACT_NOTICE_IND": ["amend", "amendment"], -# "TIME_TO_OBJECT": ["object", "objection"], -# "ASSIGNMENTS_CLAUSE_IND": ["assign", "assignment"], -# "CREDENTIALING_APP_IND": ["credential", "credentialing"], -# "DELEGATED_TERMS": ["delegated"], -# "ECM": ["ecm"], -# "NATIONAL_AGREEMENT_IND": ["national", "multi-state"], -# "COST_SETTLEMENT_LANGUAGE": ["cost settlement", "settlement"], -# "LATE_PAID_CLAIMS_LANGUAGE": ["late paid", "late payment"], -# "DEEMER_AMENDMENT": ["deem", "deemed"], -# "REGULATORY_REQUIREMENTS": ["regulatory requirements", "regulatory"], -# "RECOVERY_RIGHTS": ["recovery rights"], -# "ARBITRATION_AND_DISPUTES": ["arbitration", "arbitrate", "dispute"], -# "EXCLUSIVITY_REQUIREMENT_LANGUAGE": ["exclusive"], -# "PAYOR": ["payor"], -# "PARTICIPATION_IN_PRODUCTS": ["participation", "participate"], -# "CLEAN_CLAIM": ["clean claim"], -# "INDEPENDENT_REVIEW_LANGUAGE": ["independent review"], -# "INDEMNIFICATION": ["indemnification", "indemnify"], -# "ACCESS_TO_MEDICAL_RECORDS": ["medical records", "access"], -# "MEMBER_CONFINEMENT_DAYS_LANGUAGE": ["confinement", "confined"], -# "NETWORK_ACCESS_FEES_LANGUAGE": ["network access fee", "access fee"], -# "PAYMENT_IN_ADVANCE_OF_CLAIMS_SUBMISSION_LANGUAGE": ["advance payment", "payment in advance"], -# "ELIGIBILITY_VERIFICATION": ["eligibility", "verification", "eligible"], -# "PREAUTHORIZATION": ["pre authorization", "authorization"], -# "POLICIES_AND_PROCEDURES": ["policies", "procedures"], -# "INSURANCE_REQUIREMENT": ["insurance"], -# "CARVEOUT_VENDORS": ["carve out", "carve-out"], -# "CONFLICTS_BETWEEN_CERTAIN_DOCUMENTS_LANGUAGE": ["conflict"], -# "RELATIONSHIP_OF_PARTIES_LANGUAGE": ["relationship of parties"], -# "NONSTANDARD_APPEALS_PROCESS_LANGUAGE": ["appeal"], -# "PRODUCT_REMOVAL": ["product removal"], -# "DISPARAGEMENT_PROHIBITION_LANGUAGE": ["disparage", "disparagement"], -# "CLAIMS_EDITING_LANGUAGE": ["claims editing"], -# "GUARANTEE_OF_PROVIDER_YIELD_LANGUAGE": ["provider yield", "guarantee of provider yield"], -# "HCBS_SERVICES": ["hcbs"], -# "PMPM": ["pmpm", "per member per month"], -# "INVOICE_PRICING_LANGUAGE": ["invoice pricing"], -# "TEMPLATE": ["template"], -# "PROVIDER_BASED_BILLING_EXCLUSION_LANGUAGE": ["provider based billing"], -# "NON_RENEWAL_LANGUAGE": ["not to renew", "non-renewal"], -# "NON_RENEWAL_DAYS": ["not to renew", "non-renewal"], -# 'CONTRACT_TITLE': ["agreement", "contract title", "amendment title"], -# 'PAYER_NAME': ["payer", "health plan", "insurance company"], -# 'TERM_CLAUSE': ["term", "duration", "agreement period"], -# 'CONTRACT_EFFECTIVE_DT': ["effective date", "commencement date"], -# 'PROV_GROUP_TIN': ["tax identification number", "TIN", "taxpayer id"], -# 'PROV_GROUP_NAME': ["provider group", "group name", "provider name"], -# 'PROV_GROUP_NPI': ["national provider identifier", "NPI", "provider number"], -# 'NOTICE_PROVIDER_NAME': ["notice to", "notification to", "attn:"], -# 'NOTICE_PROVIDER_ADDRESS': ["notice address", "notification address"], -# 'HEALTH_PLAN_STATE': ["state of", "jurisdiction", "governing state"], -# 'SEQUESTRATION_LANGUAGE': ["sequestration", "budget neutrality"], -# 'NPI_NAME': ["NPI", "national provider identifier"], -# 'PROV_GROUP_TIN_SIGNATORY': ["signatory TIN", "signing tax ID"], -# 'PROV_TIN_OTHER': ["additional TIN", "other tax ID"], -# 'PROV_NPI_OTHER': ["additional NPI", "other provider ID"] -# } - - # Methodology # regex = make 'keyword' arg a pattern (NPI, TIN, Dates, etc) # hierarchical and/or = Run and condition first, if nothing, then run or # hierarchy = run each keyword independently in order of listing - case sensitive # and = run each keyword if all are present, case insensitive # or = run each keyword if any are present, case insensitive -KEYWORD_MAPPINGS = { - "TERM_CLAUSE": { - "methodology": "hierarchy", - "keywords": ["Term", "term", "termination", "duration", "agreement period"], + + +# Example: {'blue_group' : {'fields' : ['TermClause', ContractAuto', ...], 'methodology' : 'or', 'keywords' : ['renew', 'term']} +# "irs_group / "Gray Group":" +# 'Tax ID,TIN,IRS,Provider Name,National Provider Identifier,NPI,Other NPIs,To Provider At:,To Provider at:' + +GROUPED_KEYWORD_MAPPINGS = { + "term_group": { + "fields": [ + "TERM_CLAUSE", + "CONTRACT_AUTO_RENEWAL_IND", + "CONTRACT_TERMINATION_DT", + "TERMINATION_UPON_NOTICE", + "TERMINATION_UPON_CAUSE", + ], + "methodology": "or", + "keywords": [ + "TERM AND TERMINATION", + "Term", + "Renew", + "upon notice", + "With Cause", + ], + "case_sensitive": True, + }, + "initial_page_group": { + "fields": ["CONTRACT_TITLE", "PAYER_NAME", "HEALTH_PLAN_STATE"], + "methodology": "or", + "keywords": ["Health Plan Name", "Health Maintenance Organization"], + "case_sensitive": False, + }, + "section_5_6_group": { + "fields": [ + "INSURANCE_REQUIREMENT", + "INDEMNIFICATION", + "ARBITRATION_AND_DISPUTES", + ], + "methodology": "or", + "keywords": [ + "INSURANCE AND INDEMNIFICATION", + "5.1", + "Insurance", + "Insurance Requirement", + "5.2", + "Indemnification", + "Demnification", + "6.1", + "6.2", + "Arbitration", + "Arbitrate", + "Dispute", + ], + "case_sensitive": False, + }, + # IRS group fields not smart chunked currently. + # "irs_group": { + # "fields": [ + # "PROV_GROUP_TIN", + # "PROV_GROUP_NAME", + # "PROV_GROUP_NPI", + # "NPI_NAME", + # "PROV_GROUP_TIN_SIGNATORY", + # ], + # "methodology": "or", + # "keywords": [], + # "case_sensitive": True + # }, + "notice_provider_group": { + "fields": ["NOTICE_PROVIDER_NAME", "NOTICE_PROVIDER_ADDRESS"], + "methodology": "or", + "keywords": ["Attn", "Inc.", "Suite", "Ste", "P.O Box", "To Provider at:"], + "case_sensitive": False, }, "LATE_PAID_CLAIMS_LANGUAGE": { + "fields": ["LATE_PAID_CLAIMS_LANGUAGE"], "methodology": "hierarchy", "keywords": [ "late paid", @@ -85,60 +88,82 @@ KEYWORD_MAPPINGS = { "interest shall be paid", "interest payment", ], + "case_sensitive": False, }, "NON_RENEWAL_LANGUAGE": { + "fields": ["NON_RENEWAL_LANGUAGE"], "methodology": "or", "keywords": ["not to renew", "non-renewal", "non renewal", "nonrenewal"], + "case_sensitive": False, }, - # "NON_RENEWAL_DAYS": {'methodology': 'or', 'keywords': ["not to renew", "non-renewal", "non renewal", "nonrenewal"]}, "HCBS_SERVICES": { + "fields": ["HCBS_SERVICES"], "methodology": "hierarchy", "keywords": ["%", "Plan shall pay", "hcbs"], + "case_sensitive": False, }, "POLICIES_AND_PROCEDURES": { + "fields": ["POLICIES_AND_PROCEDURES"], "methodology": "or", "keywords": ["2.4", "policies", "procedures"], + "case_sensitive": False, }, "REGULATORY_REQUIREMENTS": { + "fields": ["REGULATORY_REQUIREMENTS"], "methodology": "or", "keywords": ["8.3", "6.4", "regulatory requirements", "regulatory"], + "case_sensitive": False, + }, + "RECOVERY_RIGHTS": { + "fields": ["RECOVERY_RIGHTS"], + "methodology": "or", + "keywords": ["3.5", "recovery rights"], + "case_sensitive": False, }, - "RECOVERY_RIGHTS": {"methodology": "or", "keywords": ["3.5", "recovery rights"]}, "RELATIONSHIP_OF_PARTIES_LANGUAGE": { + "fields": ["RELATIONSHIP_OF_PARTIES_LANGUAGE"], "methodology": "or", "keywords": ["8.1", "relationship of parties"], + "case_sensitive": False, }, "EXCLUSIVITY_REQUIREMENT_LANGUAGE": { + "fields": ["EXCLUSIVITY_REQUIREMENT_LANGUAGE"], "methodology": "or", "keywords": ["27", "exclusiv"], + "case_sensitive": False, }, "CONTRACT_EFFECTIVE_DT": { - "methodology": "or", - "keywords": ["effective date", "effective"], - }, # TODO Switch methodology to regex for date formats + 'or' methodology is case-insensitive -remove duplicates - "TERMINATION_UPON_NOTICE": { - "methodology": "or", - "keywords": ["notice", "upon notice", "termination"], - }, - "NOTICE_PROVIDER_NAME": { - "methodology": "or", - "keywords": ["Attn", "Inc.", "Suite", "Ste", "P.O Box"], - }, # TODO - Move Name and Address together - "NOTICE_PROVIDER_ADDRESS": { - "methodology": "or", - "keywords": ["Attn", "Inc.", "Suite", "Ste", "P.O Box"], - }, # TODO - Move Name and Address together - # "TIMELY_FILING" : {'methodology': 'or', 'keywords': ["Provider claims", "claims", "reciept", "adjudication", "payment","appealed claims", "paid or denied status"]}, - "PROV_GROUP_TIN": { + "fields": ["CONTRACT_EFFECTIVE_DT"], "methodology": "or", "keywords": [ - "Tax Identification Number", - "TIN", - "Tax", - "ID", - "Tax ID Number", - "Taxpayer ID", - ], + "effective date", + "effective", + ], # TODO Switch methodology to regex for date formats + 'or' methodology is case-insensitive -remove duplicates + "case_sensitive": False, }, - # PROV_GROUP_TIN : {'methodology' : 'regex', 'keywords' : ['regex-string-here']} # TODO Add regex to match ##-####### or ######### + # "TERMINATION_UPON_NOTICE": { + # "fields": ["TERMINATION_UPON_NOTICE"], + # "methodology": "or", + # "keywords": ["notice", "upon notice", "termination"], + # "case_sensitive": False, + # }, + # "PROV_GROUP_TIN": { + # "fields": ["PROV_GROUP_TIN"], + # "methodology": "or", + # "keywords": [ + # "Tax Identification Number", + # "TIN", + # "Tax", + # "ID", + # "Tax ID Number", + # "Taxpayer ID", + # ], + # "case_sensitive": False, + # }, + # TODO Add regex to match ##-####### or ######### + # "group_name": { + # "fields" : [], + # 'methodology' : "", + # 'keywords' : [], + # }, } diff --git a/fieldExtraction/src/main.py b/fieldExtraction/src/main.py index 8647cd6..b1d4869 100644 --- a/fieldExtraction/src/main.py +++ b/fieldExtraction/src/main.py @@ -46,7 +46,7 @@ def process_ac(item): def main(): - + input_dict = ( utils.read_input() ) # keys are contract names, values are full contract text @@ -56,9 +56,7 @@ def main(): # Filter B input_dict_b = { - k: v - for k, v in input_dict.items() - if utils.contains_reimbursement(str(v)) + k: v for k, v in input_dict.items() if utils.contains_reimbursement(str(v)) } # Filter out non-contracts print( f"B Input Files after reimbursement filter: {len(input_dict_b)} | {total_files-len(input_dict_b)} files removed" @@ -66,10 +64,12 @@ def main(): # Filter already processed if config.FILTER_ALREADY_PROCESSED: - input_dict_ac, input_dict_b = utils.filter_already_processed(input_dict, input_dict_b) + input_dict_ac, input_dict_b = utils.filter_already_processed( + input_dict, input_dict_b + ) else: input_dict_ac = input_dict - + print(f"AC Input Files left to be processed : {len(input_dict_ac)}") print(f"B Input Files left to be processed : {len(input_dict_b)}") diff --git a/fieldExtraction/src/postprocess.py b/fieldExtraction/src/postprocess.py index 14cc353..d81f11d 100644 --- a/fieldExtraction/src/postprocess.py +++ b/fieldExtraction/src/postprocess.py @@ -12,12 +12,12 @@ import utils def b_postprocess(filename, df, pages): - + if df.shape[0] > 0: # Metadata fields - df.drop('Filename', axis=1, inplace=True) - - df["Contract Name"] = 'Filename: ' + filename + df.drop("Filename", axis=1, inplace=True) + + df["Contract Name"] = "Filename: " + filename df["Parent Agreement Code"] = postprocessing_funcs.get_parent_agreement_code( filename ) @@ -44,7 +44,7 @@ def b_postprocess(filename, df, pages): # Clean LOB df = postprocessing_funcs.clean_lob(df, filename) - # Rename and reorder + # Rename and reorder df.rename(columns=valid.B_MAPPING, inplace=True) column_order = [col for col in valid.B_MAPPING.values() if col in df.columns] @@ -54,18 +54,19 @@ def b_postprocess(filename, df, pages): else: return df + def ac_postprocess(df, filename, pages): - + if df.shape[0] > 0: - - df["Contract Name"] = 'Filename: ' + filename + + df["Contract Name"] = "Filename: " + filename df["Pages"] = pages df["Parent Agreement Code"] = postprocessing_funcs.get_parent_agreement_code( filename ) - + df = df.fillna("") df = postprocessing_funcs.clean_term_clause(df) @@ -94,17 +95,15 @@ def ac_postprocess(df, filename, pages): # Derive Indicators df = df.apply(postprocessing_funcs.derive_indicators, axis=1) - + # Rename and Reorder df = df.rename(columns=valid.AC_MAPPING) column_order = [col for col in valid.ABC_COLUMNS if col in df.columns] + [ col for col in df.columns if col not in valid.ABC_COLUMNS ] - + final_df = df[column_order] - + return final_df else: return df - - diff --git a/fieldExtraction/src/postprocessing_funcs.py b/fieldExtraction/src/postprocessing_funcs.py index 13d958d..3cf8e74 100644 --- a/fieldExtraction/src/postprocessing_funcs.py +++ b/fieldExtraction/src/postprocessing_funcs.py @@ -136,7 +136,7 @@ def get_parent_agreement_code(filename): try: filename = filename.split(".txt")[0] - filename = re.sub(r'\([^)]*\)', '', filename) + 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] @@ -190,17 +190,18 @@ def clean_default_term(df): 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) + 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' + 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 @@ -292,6 +293,7 @@ def clean_prov_2(df): return df + def clean_term_clause(final_df): if "TERM_CLAUSE" in final_df: final_df.loc[ @@ -323,10 +325,15 @@ def clean_auto_renewal_ind(final_df): def clean_npi(final_df): - if "PROV_GROUP_NPI" in 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) + 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 @@ -348,13 +355,13 @@ def clean_health_plan_state(final_df): 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 @@ -388,7 +395,7 @@ def get_tin_from_filename(final_df): "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: @@ -398,7 +405,7 @@ def get_tin_from_filename(final_df): "PROV_GROUP_TIN", ] = final_df["temp_filename"] final_df.drop(columns=["temp_filename"], inplace=True) - + return final_df @@ -419,36 +426,46 @@ def clean_policies_and_procedures(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) + 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)) + digits = "".join(filter(str.isdigit, x)) if len(digits) == 9: # Format as '##-#######' - return f'{digits[:2]}-{digits[2:]}' + return f"{digits[:2]}-{digits[2:]}" else: # Retain prior handling for invalid entries - if x != 'N/A': - return f'N/A - (model detected: {x})' + if x != "N/A": + return f"N/A - (model detected: {x})" else: - return 'N/A' - + 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) + 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) + 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) + 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"\"", '"') diff --git a/fieldExtraction/src/preprocess.py b/fieldExtraction/src/preprocess.py index 2613ccc..c9dfe48 100644 --- a/fieldExtraction/src/preprocess.py +++ b/fieldExtraction/src/preprocess.py @@ -4,6 +4,7 @@ import table_funcs import config import csv + def preprocess(contract_text, filename, fields=config.FIELDS): contract_text = preprocessing_funcs.clean_newlines(contract_text) @@ -12,8 +13,8 @@ def preprocess(contract_text, filename, fields=config.FIELDS): text_dict = preprocessing_funcs.split_text( contract_text ) # return a dictionary with keys - page_num (str), values as the page_text - - #text_dict = {k: v for k, v in text_dict.items() if k.isdigit() and 42 <= int(k) <= 42} + + # text_dict = {k: v for k, v in text_dict.items() if k.isdigit() and 42 <= int(k) <= 42} num_pages = len(text_dict.keys()) text_dict = preprocessing_funcs.filter_quick_review(text_dict) @@ -23,7 +24,7 @@ def preprocess(contract_text, filename, fields=config.FIELDS): ) # All pages with exhibit headers text_dict = table_funcs.align_and_format_tables(text_dict, filename) text_dict = preprocessing_funcs.chunk_consecutive(text_dict, exhibit_pages) - #write text_dict to a csv file: + # write text_dict to a csv file: with open("text_dict.csv", "w") as f: writer = csv.writer(f) for key, value in text_dict.items(): diff --git a/fieldExtraction/src/preprocessing_funcs.py b/fieldExtraction/src/preprocessing_funcs.py index 5546a51..1fb5b8f 100644 --- a/fieldExtraction/src/preprocessing_funcs.py +++ b/fieldExtraction/src/preprocessing_funcs.py @@ -4,9 +4,13 @@ from collections import defaultdict import utils import valid from valid import DERIVED_INDICATOR_FIELDS -from keywords import KEYWORD_MAPPINGS +import keywords from collections import defaultdict import ac_smart_chunking +import ac_funcs +import prompts +import claude_funcs +import config def clean_newlines(contract_text: str) -> str: @@ -183,13 +187,16 @@ def chunk_consecutive(text_dict, exhibit_pages): return final_dict -def smart_chunk_ac(text_dict: dict, keyword_mappings: dict = KEYWORD_MAPPINGS) -> dict: +def smart_chunk_ac( + text_dict: dict, keyword_mappings: dict = keywords.GROUPED_KEYWORD_MAPPINGS +) -> dict: """Performs a keyword search on textract outputs to retrieve relevant chunks for a given field. Args: text_dict (dict): Dictionary keyed by string page-number and valued by actual textract output page - keyword_mappings (dict, optional): Dictionary keyed by field, with sub-dictionary with keys `methodology` and `keywords` - `methodology` must be either `or` or `hierarchy. Defaults to keywords.KEYWORD_MAPPINGS + keyword_mappings (dict, optional): Dictionary keyed by field group, with sub-dictionary + with keys `methodology`, `keywords`, and `case_sensitive`. `methodology` must be either `or` or `hierarchy. + Defaults to keywords.GROUPED_KEYWORD_MAPPINGS Raises: ValueError: When the `key_dict` from keyword_mappings has a methodology that is not implemented. @@ -205,19 +212,48 @@ def smart_chunk_ac(text_dict: dict, keyword_mappings: dict = KEYWORD_MAPPINGS) - full_context = "\n".join( [f"Page {page}:\n{content}" for page, content in text_dict.items()] ) - # print(f"Full context word count: {len(full_context.split())}") - for field, key_dict in keyword_mappings.items(): + for field_group, key_dict in keyword_mappings.items(): if key_dict["methodology"] == "or": - page_list = ac_smart_chunking.chunk_or(text_dict, key_dict) + page_list = ac_smart_chunking.chunk_or( + text_dict, key_dict["keywords"], key_dict["case_sensitive"] + ) + keyword_page_dict = ac_smart_chunking.keyword_search( + text_dict, key_dict["keywords"], key_dict["case_sensitive"] + ) elif key_dict["methodology"] == "hierarchy": - page_list = ac_smart_chunking.chunk_hierarchical(text_dict, key_dict) + page_list = ac_smart_chunking.chunk_hierarchical( + text_dict, key_dict["keywords"], key_dict["case_sensitive"] + ) + keyword_page_dict = ac_smart_chunking.keyword_search( + text_dict, key_dict["keywords"], key_dict["case_sensitive"] + ) + elif key_dict["methodology"] == "and": + page_list = ac_smart_chunking.chunk_and( + text_dict, key_dict["keywords"], key_dict["case_sensitive"] + ) + keyword_page_dict = ac_smart_chunking.keyword_search( + text_dict, key_dict["keywords"], key_dict["case_sensitive"] + ) + elif key_dict["methodology"] == "regex": + page_list = ac_smart_chunking.chunk_regex(text_dict, key_dict["regex"]) + keyword_page_dict = ac_smart_chunking.keyword_search( + text_dict, key_dict["keywords"], key_dict["case_sensitive"], True + ) else: page_list = list(text_dict.keys()) + keyword_page_dict = {"All pages": list(text_dict.keys())} raise ValueError( "methodology must be 'hierarchy' or 'or' - using entire contract as chunk" ) - chunks[field] = "\n".join([text_dict[str(page_num)] for page_num in page_list]) + if field_group == "initial_page_group": + keyword_page_dict["initial_pages"] = [1, 2] + chunks[field_group] = "\n".join( + [text_dict[str(page_num)] for page_num in page_list] + ) + chunks[field_group + "_pages"] = keyword_page_dict + chunks[field_group + "_methodology"] = key_dict["methodology"] + chunks[field_group + "_case"] = key_dict["case_sensitive"] return chunks @@ -229,3 +265,64 @@ def filter_quick_review(text_dict): and "TOP SHEET" not in page_text.upper() and "COVER SHEET" not in page_text.upper() } + + +# Regex TIN Logic +# Regex search for all TINs (##-####### format) +# If exactly one TIN is found: that is the TIN +# If more than one TIN is found: chunk + prompt to find which of them is the correct TIN + +# If zero TINs are found, regex search for all TINs (######### format) +# Then repeat steps 2 and 3 with the new pattern +# If still zero TINs are found, check for a TIN in the filename +# If still none, return empty + + +# def tin_regex(filename, text_dict: dict[str, str]): +# pattern_1 = r'\b\d{2}-?\d{7}\b' +# matches = [] +# page_list = [] + +# for page in text_dict.keys(): +# if len(re.findall((pattern_1, text_dict[page]))) != 0: +# matches.extend(re.findall((pattern_1, text_dict[page]))) +# page_list.append(page) + +# if len(matches) == 1: +# return matches[0] + +# elif len(matches)>1: +# chunks = {} +# for pages in page_list: +# chunks[pages] = text_dict[pages] +# return chunks + +# elif len(matches) == 0: +# pattern_2 = r'\b\d{9}\b' +# matches = [] +# page_list = [] + +# for page in text_dict.keys(): +# if len(re.findall((pattern_2, text_dict[page]))) != 0: +# matches.extend(re.findall((pattern_2, text_dict[page]))) +# page_list.append(page) + + +# if len(matches) == 1: +# return matches[0] + +# elif len(matches)>1: +# chunks = {} +# for pages in page_list: +# chunks[pages] = text_dict[pages] +# return chunks + +# # find regex in file name +# tin = re.findall(pattern_1, filename) +# if len(tin) == 1: +# return tin[0] +# if len(tin) == 0: +# tin = re.findall(pattern_2, filename) +# if len(tin) == 1: +# return tin[0] +# return None diff --git a/fieldExtraction/src/prompts.py b/fieldExtraction/src/prompts.py index e0329f8..7f8f090 100644 --- a/fieldExtraction/src/prompts.py +++ b/fieldExtraction/src/prompts.py @@ -1,7 +1,5 @@ -import config import valid - ##################################################################################### ##################################### BOTTOM UP ##################################### ##################################################################################### @@ -237,7 +235,7 @@ def CONDITIONAL_PROV_2(d, page): valid_list = valid.VALID_FAC else: - valid_list = valid.VALID_PROF + valid.VALID_ANC + valid.VALID_FAC + valid_list = valid.VALID_PROF + valid.VALID_ANC + valid.VALID_FAC return f"""### PAGE START ### {page} ### PAGE END The above text contains a number of reimbursement terms. For the rest of this request, focus only on the specific term listed below: @@ -375,7 +373,6 @@ Do NOT add any additional commentary explaining the answer. Return ONLY the exac """ -# TODO: FIX PROMPT BELOW FOR JSON OUTPUT def AC_MULTI_FIELD_TEMPLATE(context, questions): return f"""## START CONTRACT TEXT ## {context} diff --git a/fieldExtraction/src/test.py b/fieldExtraction/src/test.py index abec264..092e68b 100644 --- a/fieldExtraction/src/test.py +++ b/fieldExtraction/src/test.py @@ -1,4 +1,3 @@ - import pandas as pd import os import re @@ -21,10 +20,12 @@ import postprocess # print("B Files Complete: " , len(set(b_filenames))) # print("AC Files Complete: ", len(set(ac_filenames))) - -output_dirs = ['../../complex-cnc/doczy.ai/5b-outputs', '../../complex-cnc/doczy.ai/ALL-5B-OUTPUTS'] +output_dirs = [ + "../../complex-cnc/doczy.ai/5b-outputs", + "../../complex-cnc/doczy.ai/ALL-5B-OUTPUTS", +] filename_dict = {} for output_dir in output_dirs: @@ -40,12 +41,9 @@ b_filenames, ac_filenames = [], [] for output_dir in output_dirs: for subdir in os.listdir(output_dir): for file in os.listdir(os.path.join(output_dir, subdir)): - if file == '5brun-B.csv' or file == '5brun-remaining-B.csv': + if file == "5brun-B.csv" or file == "5brun-remaining-B.csv": b_filenames.append(subdir) - elif file == '5brun-withallac-AC.csv' or file == '5brun-remaining-AC.csv': + elif file == "5brun-withallac-AC.csv" or file == "5brun-remaining-AC.csv": ac_filenames.append(subdir) -print("B Files Complete: " , len(set(b_filenames))) +print("B Files Complete: ", len(set(b_filenames))) print("AC Files Complete: ", len(set(ac_filenames))) - - - diff --git a/fieldExtraction/src/top_down_funcs.py b/fieldExtraction/src/top_down_funcs.py index 52ebc67..4a918df 100644 --- a/fieldExtraction/src/top_down_funcs.py +++ b/fieldExtraction/src/top_down_funcs.py @@ -9,22 +9,35 @@ import csv import json import re + def get_exhibit_range(text_dict, exhibit_pages, bu_page): - lower_bound = max([int(page) for page in exhibit_pages if int(page) <= int(bu_page)], default=None) - upper_bound = min([int(page) for page in exhibit_pages if int(page) > int(bu_page)], default=None) + lower_bound = max( + [int(page) for page in exhibit_pages if int(page) <= int(bu_page)], default=None + ) + upper_bound = min( + [int(page) for page in exhibit_pages if int(page) > int(bu_page)], default=None + ) if lower_bound is not None and upper_bound is not None: - pages_between = [page for page in text_dict.keys() if int(lower_bound) <= int(page) <= int(upper_bound)] + pages_between = [ + page + for page in text_dict.keys() + if int(lower_bound) <= int(page) <= int(upper_bound) + ] elif lower_bound is not None: # No upper_bound found - pages_between = [page for page in text_dict.keys() if int(lower_bound) <= int(page)] + pages_between = [ + page for page in text_dict.keys() if int(lower_bound) <= int(page) + ] elif upper_bound is not None: # No lower_bound found - pages_between = [page for page in text_dict.keys() if int(page) <= int(upper_bound)] + pages_between = [ + page for page in text_dict.keys() if int(page) <= int(upper_bound) + ] else: pages_between = [] return pages_between def run_top_down(filename, text_dict, bu_results, exhibit_pages): - bu_pages = {bu_dict['page_num'] for bu_dict in bu_results} + bu_pages = {bu_dict["page_num"] for bu_dict in bu_results} td_page_dict = {} td_answer_dict = {} @@ -42,10 +55,10 @@ def run_top_down(filename, text_dict, bu_results, exhibit_pages): answer_dict = td_answer_dict[key] td_answer_dict[bu_page] = answer_dict break - + merged_dicts = [] for bu_dict in bu_results: - answer_dict = td_answer_dict[bu_dict['page_num']] + answer_dict = td_answer_dict[bu_dict["page_num"]] bu_dict.update(answer_dict) merged_dicts.append(bu_dict) @@ -60,7 +73,7 @@ def run_top_down(filename, text_dict, bu_results, exhibit_pages): text_dict, filename, prompts.TOP_DOWN_MEDICALLY_NECESSARY, - valid.VALID_MEDICALLY_NECESSARY + valid.VALID_MEDICALLY_NECESSARY, ) medically_necessary_dict = postprocessing_funcs.filter_dict( medically_necessary_dict, @@ -72,11 +85,12 @@ def run_top_down(filename, text_dict, bu_results, exhibit_pages): ), ) final_dicts = add_medically_necessary(final_dicts, medically_necessary_dict) - + return final_dicts + def run_top_down_primary(text_dict, td_pages, filename): - page_text = '\n'.join([text_dict[page_num] for page_num in td_pages]) + page_text = "\n".join([text_dict[page_num] for page_num in td_pages]) prompt = prompts.TOP_DOWN_PRIMARY(page_text) answer = claude_funcs.invoke_claude( @@ -101,6 +115,7 @@ def run_top_down_primary(text_dict, td_pages, filename): answer_dict["PROV_TYPE"] = config.PROV_TYPE return answer_dict + def get_td_dict(text_dict, filename, PROMPT_FUNC, VALID_VALUES): td_dict = {} for page_num, page_text in text_dict.items(): @@ -120,7 +135,7 @@ def add_medically_necessary(merged_dicts, medically_necessary_dict): if not medically_necessary_dict: for d in merged_dicts: d["MEDICAL_NECESSITY_LANGUAGE"] = "N/A" - d["MEDICAL_NECESSITY_LANGUAGE_IND"] = 'N' + d["MEDICAL_NECESSITY_LANGUAGE_IND"] = "N" return merged_dicts sorted_keys = sorted(int(k) for k in medically_necessary_dict.keys()) @@ -144,9 +159,9 @@ def add_medically_necessary(merged_dicts, medically_necessary_dict): str(closest_page) ] if not utils.is_empty(d["MEDICAL_NECESSITY_LANGUAGE"]): - d["MEDICAL_NECESSITY_LANGUAGE_IND"] = 'Y' + d["MEDICAL_NECESSITY_LANGUAGE_IND"] = "Y" else: - d["MEDICAL_NECESSITY_LANGUAGE_IND"] = 'N' + d["MEDICAL_NECESSITY_LANGUAGE_IND"] = "N" except: d["MEDICAL_NECESSITY_LANGUAGE"] = "N/A" d["MEDICAL_NECESSITY_LANGUAGE_IND"] = "N" diff --git a/fieldExtraction/src/utils.py b/fieldExtraction/src/utils.py index dcf5385..b02a241 100644 --- a/fieldExtraction/src/utils.py +++ b/fieldExtraction/src/utils.py @@ -177,10 +177,12 @@ def contains_reimbursement(text, page="1"): print("contains_reimbursement - Invalid data type") -def split_and_write_csv(df, n, column_name, output_dir=config.CONSOLIDATED_OUTPUT_DIRECTORY): +def split_and_write_csv( + df, n, column_name, output_dir=config.CONSOLIDATED_OUTPUT_DIRECTORY +): """ Splits the DataFrame into N sub-DataFrames based on unique values in a specified column - and writes each sub-DataFrame to a separate CSV file. Each sub-DataFrame will contain + and writes each sub-DataFrame to a separate CSV file. Each sub-DataFrame will contain roughly the same number of unique values from the specified column. Args: @@ -192,14 +194,14 @@ def split_and_write_csv(df, n, column_name, output_dir=config.CONSOLIDATED_OUTPU # Create the directory if it does not exist if not os.path.exists(output_dir): os.makedirs(output_dir) - + # Get unique values and shuffle them to randomize distribution if needed unique_values = np.array(df[column_name].unique()) np.random.shuffle(unique_values) - + # Split the unique values into N parts value_groups = np.array_split(unique_values, n) - + # Loop through each group and filter DataFrame by the values in each group for i, values in enumerate(value_groups, 1): # Filter df for the values in this group @@ -207,7 +209,7 @@ def split_and_write_csv(df, n, column_name, output_dir=config.CONSOLIDATED_OUTPU # Create a filename for each sub-DataFrame filename = f"{config.BATCH_ID}-{i}.csv" path = os.path.join(output_dir, filename) - + # Write the sub-DataFrame to a CSV file sub_df.to_csv(path, index=False) print(f"Written {path}") @@ -217,14 +219,16 @@ def filter_already_processed(input_dict, input_dict_b): already_processed_ac, already_processed_b = [], [] for folder_name in os.listdir(config.OUTPUT_DIRECTORY): # AC - if (config.AC_RESULTS_NAME in os.listdir(os.path.join(config.OUTPUT_DIRECTORY, folder_name)) + if config.AC_RESULTS_NAME in os.listdir( + os.path.join(config.OUTPUT_DIRECTORY, folder_name) ): already_processed_ac.append(folder_name + ".txt") # B - if (config.B_RESULTS_NAME in os.listdir(os.path.join(config.OUTPUT_DIRECTORY, folder_name)) + if config.B_RESULTS_NAME in os.listdir( + os.path.join(config.OUTPUT_DIRECTORY, folder_name) ): already_processed_b.append(folder_name + ".txt") - + input_dict_ac = { key: input_dict[key] for key in input_dict.keys() @@ -244,3 +248,44 @@ def is_empty(value): else: empty_values = [None, "", "N/A", "NA", "null", "none", "NaN", np.nan, "nan"] return value in empty_values + + +def find_regex_matches( + pattern: str, text_dict: dict +) -> tuple[list, list]: # TODO: Write few unit tests + """Helper function to find matches and list of pages of a given regex pattern in text_dict. + + Args: + pattern(str): A particular regex match that with a pattern. + text_dict (dict): Dictionary keyed by string page-number and valued by actual textract output page + + Returns: + matches(list): List of match(s) + page_list(list): List of pages where the match was found + + """ + matches = [] + page_list = [] + for page, text in text_dict.items(): + found = re.findall(pattern, text) + if found: + matches.extend(found) + page_list.append(page) + return matches, page_list + + +def chunk_selected_pages(text_dict: dict[str, str], page_list: list) -> str: + """Combines chunks for pages with regex matches for IRS, NPI, etc. + + Args: + text_dict (dict[str, str]): A dictionary keyed by string page number and valued + by string page contents + page_list (str): List of pages to be chunked for context + + Returns: + context(str): corresponding chunks of text (input strings delimited by newlines). + """ + + page_list = sorted([int(page_str) for page_str in set(page_list)]) + context = "\n".join([text_dict[str(page_num)] for page_num in page_list]) + return context diff --git a/fieldExtraction/src/valid.py b/fieldExtraction/src/valid.py index 6e853ed..e8abe22 100644 --- a/fieldExtraction/src/valid.py +++ b/fieldExtraction/src/valid.py @@ -233,6 +233,7 @@ def select_valid_prov_2(provider_type): valid_list = VALID_PROF + VALID_ANC + VALID_FAC return valid_list + RATE_REPLACEMENTS = { "Medicare": "MCR", "Medicaid": "MCD", @@ -242,7 +243,7 @@ RATE_REPLACEMENTS = { "Average Sales Price": "ASP", "Average Wholesale Price": "AWP", "Amount Payable by Medicare": "MCR", - "Amount Payable by Medicaid": "MCD" + "Amount Payable by Medicaid": "MCD", } # List of fields that will have derived indicators @@ -263,7 +264,7 @@ DERIVED_INDICATOR_FIELDS = [ "LATE_PAID_CLAIMS_LANGUAGE", "COST_SETTLEMENT_LANGUAGE", "DELEGATED_TERMS", - "SEQUESTRATION_LANGUAGE" + "SEQUESTRATION_LANGUAGE", ] B_MAPPING = { @@ -306,8 +307,8 @@ B_MAPPING = { AC_MAPPING = { "Filename": "Contract Name", - "Pages" : "Pages", - "Parent Agreement Code" : "Parent Agreement Code", + "Pages": "Pages", + "Parent Agreement Code": "Parent Agreement Code", "CONTRACT_TITLE": "Agreement_Name (Contract Title)", "PAYER_NAME": "PAYER NAME", "AFFILIATION_CLAUSE_IND": "Affiliate (Y/N)", @@ -509,7 +510,7 @@ ABC_COLUMNS = [ "Medical Necessity Language (Language)", "Template", "Provider-based Billing Exclusion (Y/N)", - "Provider-based Billing Exclusion (Language)" + "Provider-based Billing Exclusion (Language)", ] diff --git a/fieldExtraction/tests/test_chunking.py b/fieldExtraction/tests/test_chunking.py new file mode 100644 index 0000000..c1bf12c --- /dev/null +++ b/fieldExtraction/tests/test_chunking.py @@ -0,0 +1,166 @@ +import pytest +from preprocessing_funcs import smart_chunk_ac + +test_chunk_dict_term = { + "1": "NA", + "2": "Termination", + "3": "NA", + "4": "NA", +} + +test_late_paid_claims = { + "1": "NA", + "2": "LATE PAYMENT", # try lowercase + "3": "NA", + "4": "NA", + "10": "NA", + "11": "LATE PAID", + "12": "NA", +} + +test_keyword_mappings = { + "or_1": { + "methodology": "or", + "keywords": ["kw1", "kw2", "kw3"], + "case_sensitive": False, + }, + "or_2": {"methodology": "or", "keywords": ["kw4"], "case_sensitive": False}, + "hierarchy_1": { + "methodology": "hierarchy", + "keywords": ["h5", "h6"], + "case_sensitive": False, + }, + "cs_or_1": { + "methodology": "or", + "keywords": ["KW1", "kw2", "kw3"], + "case_sensitive": True, + }, + "and": {"methodology": "and", "keywords": ["KW1", "KW2"], "case_sensitive": False}, + "regex": { + "methodology": "regex", + "regex": r"\b\d{2}-?\d{7}\b", + "case_sensitive": True, + "keywords": ["KW1", "KW2"], + }, +} + +case_1 = { + "1": "NA", + "2": "kw1", + "3": "NA", + "4": "h6", + "5": "NA", + "6": "kw3", +} + +case_2 = { + "1": "NA", + "2": "h6", + "3": "NA", + "4": "kw2", + "10": "NA", + "11": "h5", + "12": "NA", + "15": "kw4", +} + +case_3 = { + "1": "NA", + "2": "KW1", + "3": "NA", + "4": "h6", + "5": "NA", + "6": "kw3", +} + +case_4 = { + "1": "NA", + "2": "KW1", + "3": "NA", + "4": "KW1KW2", + "5": "NA", + "6": "kw3", +} + +case_5 = { + "1": "NA", + "2": "12-3456789", + "3": "987654321", + "4": "Next", + "5": "1234567-89", + "6": "kw3", +} + + +def test_smart_chunk_term(): + assert smart_chunk_ac(text_dict=test_chunk_dict_term)["term_group"] == "\n".join( + ["NA", "Termination", "NA"] + ) + + +def test_smart_chunk_late_paid(): + assert smart_chunk_ac(text_dict=test_late_paid_claims)[ + "LATE_PAID_CLAIMS_LANGUAGE" + ] == "\n".join(["NA", "LATE PAID", "NA"]) + + +def test_or_1(): + assert smart_chunk_ac(text_dict=case_1, keyword_mappings=test_keyword_mappings)[ + "or_1" + ] == "\n".join(["NA", "kw1", "NA", "NA", "kw3"]) + assert smart_chunk_ac(text_dict=case_2, keyword_mappings=test_keyword_mappings)[ + "or_1" + ] == "\n".join(["NA", "kw2"]) + + +def test_or_2(): + assert smart_chunk_ac(text_dict=case_1, keyword_mappings=test_keyword_mappings)[ + "or_2" + ] == "\n".join([]) + assert smart_chunk_ac(text_dict=case_2, keyword_mappings=test_keyword_mappings)[ + "or_2" + ] == "\n".join(["kw4"]) + + +def test_case_sensitive_or_1(): + assert smart_chunk_ac(text_dict=case_1, keyword_mappings=test_keyword_mappings)[ + "cs_or_1" + ] == "\n".join(["NA", "kw3"]) + assert smart_chunk_ac(text_dict=case_3, keyword_mappings=test_keyword_mappings)[ + "cs_or_1" + ] == "\n".join(["NA", "KW1", "NA", "NA", "kw3"]) + + +def test_hierarchy_1(): + assert smart_chunk_ac(text_dict=case_1, keyword_mappings=test_keyword_mappings)[ + "hierarchy_1" + ] == "\n".join(["NA", "h6", "NA"]) + assert smart_chunk_ac(text_dict=case_2, keyword_mappings=test_keyword_mappings)[ + "hierarchy_1" + ] == "\n".join(["NA", "h5", "NA"]) + + +def test_and(): + assert smart_chunk_ac(text_dict=case_4, keyword_mappings=test_keyword_mappings)[ + "and" + ] == "\n".join(["NA", "KW1KW2", "NA"]) + + +def test_regex(): + assert smart_chunk_ac(text_dict=case_5, keyword_mappings=test_keyword_mappings)[ + "regex" + ] == "\n".join(["NA", "12-3456789", "987654321", "Next"]) + + +def test_undefined_methodology(): + with pytest.raises(ValueError): + smart_chunk_ac( + text_dict=case_1, + keyword_mappings={ + "e1": { + "methodology": "Not a real methodology", + "keywords": ["NA"], + "case_sensitive": True, + } + }, + )