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

Merged in feature_ac_chunking_clean (pull request #265)

Browse Source 
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
This commit is contained in:
Karan Desai
2024-11-13 17:08:19 +00:00
parent 8c6867800a
commit 1098be5cf5
23 changed files with 1147 additions and 291 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2
View File
@@ -92,3 +92,5 @@ docs/
.mypy_cache/
.pytest_cache/
__pycache__/
new/
fieldExtraction/pyproject.toml
+2 -3
View File
@@ -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"
]
]
pythonpath = ["src"]
fieldExtraction/scripts/sample_smart_chunks.py
+87
View File
@@ -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()
fieldExtraction/scripts/tin_test.py
+41
View File
@@ -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()
fieldExtraction/src/ac_funcs.py
+76
View File
@@ -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
fieldExtraction/src/ac_smart_chunking.py
+173 -12
View File
@@ -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
fieldExtraction/src/conditional_funcs.py
+1 -1
View File
@@ -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()
)
fieldExtraction/src/config.py
+20 -11
View File
@@ -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)
fieldExtraction/src/consolidate_output.py
+17 -9
View File
@@ -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}")
fieldExtraction/src/file_counting.py
+5 -5
View File
@@ -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)}")
print(f"B Input Files left to be processed : {len(input_dict_b)}")
fieldExtraction/src/file_processing.py
+166 -54
View File
@@ -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):
fieldExtraction/src/keywords.py
+122 -97
View File
@@ -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' : [],
# },
}
fieldExtraction/src/main.py
+6 -6
View File
@@ -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)}")
fieldExtraction/src/postprocess.py
+13 -14
View File
@@ -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
fieldExtraction/src/postprocessing_funcs.py
+43 -26
View File
@@ -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"\"", '"')
fieldExtraction/src/preprocess.py
+4 -3
View File
@@ -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():
fieldExtraction/src/preprocessing_funcs.py
+106 -9
View File
@@ -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
fieldExtraction/src/prompts.py
+1 -4
View File
@@ -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}
fieldExtraction/src/test.py
+7 -9
View File
@@ -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)))
fieldExtraction/src/top_down_funcs.py
+29 -14
View File
@@ -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"
fieldExtraction/src/utils.py
+54 -9
View File
@@ -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
fieldExtraction/src/valid.py
+6 -5
View File
@@ -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)",
]
fieldExtraction/tests/test_chunking.py
+166
View File
@@ -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,
}
},
)