diff --git a/fieldExtraction/pyproject.toml b/fieldExtraction/pyproject.toml index 8f37735..401d97b 100644 --- a/fieldExtraction/pyproject.toml +++ b/fieldExtraction/pyproject.toml @@ -36,6 +36,7 @@ target-version = ['py312'] [tool.mypy] disable_error_code = ["import-untyped","assignment","name-defined","call-arg","var-annotated","attr-defined","arg-type"] +exclude = ["scripts/"] [tool.pytest.ini_options] pythonpath=["src"] diff --git a/fieldExtraction/tests/adhoc/cnc_adhoc_stitching_1.py b/fieldExtraction/scripts/adhoc/cnc_adhoc_stitching_1.py similarity index 100% rename from fieldExtraction/tests/adhoc/cnc_adhoc_stitching_1.py rename to fieldExtraction/scripts/adhoc/cnc_adhoc_stitching_1.py diff --git a/fieldExtraction/tests/adhoc/cnc_adhoc_stitching_2.py b/fieldExtraction/scripts/adhoc/cnc_adhoc_stitching_2.py similarity index 100% rename from fieldExtraction/tests/adhoc/cnc_adhoc_stitching_2.py rename to fieldExtraction/scripts/adhoc/cnc_adhoc_stitching_2.py diff --git a/fieldExtraction/tests/adhoc/cnc_adhoc_stitching_3.py b/fieldExtraction/scripts/adhoc/cnc_adhoc_stitching_3.py similarity index 100% rename from fieldExtraction/tests/adhoc/cnc_adhoc_stitching_3.py rename to fieldExtraction/scripts/adhoc/cnc_adhoc_stitching_3.py diff --git a/fieldExtraction/tests/adhoc/cnc_prov2_effdate.py b/fieldExtraction/scripts/adhoc/cnc_prov2_effdate.py similarity index 100% rename from fieldExtraction/tests/adhoc/cnc_prov2_effdate.py rename to fieldExtraction/scripts/adhoc/cnc_prov2_effdate.py diff --git a/fieldExtraction/tests/adhoc/combine_ac2_and_b.py b/fieldExtraction/scripts/adhoc/combine_ac2_and_b.py similarity index 100% rename from fieldExtraction/tests/adhoc/combine_ac2_and_b.py rename to fieldExtraction/scripts/adhoc/combine_ac2_and_b.py diff --git a/fieldExtraction/tests/adhoc/consolidate_ac2.py b/fieldExtraction/scripts/adhoc/consolidate_ac2.py similarity index 100% rename from fieldExtraction/tests/adhoc/consolidate_ac2.py rename to fieldExtraction/scripts/adhoc/consolidate_ac2.py diff --git a/fieldExtraction/tests/adhoc/tx_ac_no_reimbursement.py b/fieldExtraction/scripts/adhoc/tx_ac_no_reimbursement.py similarity index 100% rename from fieldExtraction/tests/adhoc/tx_ac_no_reimbursement.py rename to fieldExtraction/scripts/adhoc/tx_ac_no_reimbursement.py diff --git a/fieldExtraction/tests/adhoc/unit_test.py b/fieldExtraction/scripts/adhoc/unit_test.py similarity index 100% rename from fieldExtraction/tests/adhoc/unit_test.py rename to fieldExtraction/scripts/adhoc/unit_test.py diff --git a/fieldExtraction/src/ac_smart_chunking_funcs.py b/fieldExtraction/src/ac_smart_chunking_funcs.py deleted file mode 100644 index 21826e4..0000000 --- a/fieldExtraction/src/ac_smart_chunking_funcs.py +++ /dev/null @@ -1,297 +0,0 @@ -import prompts -import keywords -import claude_funcs -import config -import ac_funcs -import utils -from enums.delimiters import Delimiter - -chunk_data = [] -log_data = [] - - -def run_smart_chunk_prompts( - ac_answers_dict: dict[str, str], - ac_chunks: dict, - contract_text: str, - filename: str, - text_dict: dict[str, str], -): - - ## 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() - ] - - # list to append the fields with no matching keywords - no_keyword_matches = [] - - for field_group in field_groups: - keyword_dict = keywords.GROUPED_KEYWORD_MAPPINGS[field_group].get( - "keywords", [] - ) - fields = keywords.GROUPED_KEYWORD_MAPPINGS[field_group].get("fields", []) - fields = [field for field in fields] - fields.sort() - questions = {} - for field in fields: - questions[field] = prompts.AC_DICT.get(field, "") - - # Use chunk if available and different from full context, otherwise use full context - ############# Important Note ############# - # the above logic is changed - # New Logic - chunk if available and different from full context, otherwise add the field to full context list - if ( - field_group in ac_chunks - and ac_chunks[field_group].strip() - and ac_chunks[field_group] != contract_text - ): - - context = ac_chunks[field_group] - context = context[0 : min(100000, len(context)) - 1] - context_type = "Smart Chunking" - - if len(fields) == 1: - question = questions[ - fields[0] - ] # Take the one question for the single field included - if fields[0] == "CONTRACT_EFFECTIVE_DT": - prompt = prompts.get_effective_date_prompt(context) - else: - 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") - - try: - if field_group == "AGREEMENT_NAME": - - # context, prompt and answer based on first 5 pages - context_agreement = "\n".join( - [ - text_dict[str(page_num)] - for page_num in text_dict.keys() - if str(page_num) in ["1", "2", "3", "4", "5"] - ] - ) - prompt_agreement = prompts.AC_SINGLE_FIELD_TEMPLATE( - context_agreement, question - ) - field_group_answer_raw = claude_funcs.invoke_claude( - prompt_agreement, - config.MODEL_ID_CLAUDE35_SONNET, - filename, - 8192, - ) - - # if answer is N/A use the smart-chunked context - if field_group_answer_raw == "N/A": - field_group_answer_raw = claude_funcs.invoke_claude( - prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, 8192 - ) - else: - field_group_answer_raw = claude_funcs.invoke_claude( - prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, 8192 - ) - - if len(fields) == 1: - if fields[0] == "CONTRACT_EFFECTIVE_DT": - # print(f'Effective Date Answer obtained from LLM: {field_group_answer_raw}') - field_group_answer_raw = ac_funcs.extract_text_from_delimiters( - field_group_answer_raw, Delimiter.PIPE - ) - # print(f'Parsed Effective Date Answer (extract_text_from_delimiters): {field_group_answer_raw}') - try: - field_group_answer_raw = utils.convert_to_us_date_format( - field_group_answer_raw - ) # Fix date formatting - # print(f'Effective Date Answer (convert_to_us_date_format): {field_group_answer_raw}') - except utils.InvalidDateException as e: - field_group_answer_raw = "N/A" - # print(f'Got Invalid Date Exception, using contract effective date as N/A for contract {filename}') - - field_group_answer_dict = {fields[0]: field_group_answer_raw} - - elif len(fields) > 1: - field_group_answer_raw = field_group_answer_raw.replace( - "_DATE", "_DT" - ) - field_group_answer_dict = ac_funcs.json_parsing_search( - field_group_answer_raw, fields - ) - - # 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}: {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}%", - } - ) - else: # If no chunk available - if ( - field_group == "CONTRACT_EFFECTIVE_DT" - ): # Special case for effective date - ac_answers_dict["CONTRACT_EFFECTIVE_DT"] = "N/A" - # print('Got no chunk for effective date, so setting it to N/A', ac_answers_dict) - else: # Add to no_keyword_matches list to run using full context - no_keyword_matches.extend(fields) - - return ac_answers_dict, no_keyword_matches - - -def get_full_context_fields( - ac_answers_dict: dict[str, str], no_keyword_matches: list -) -> list: - - 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) - - # Instantiate full context fields from those that weren't smart chunked - full_context_fields = [ - field for field in prompts.AC_DICT if field not in smartly_chunked - ] - - # Add fields with no keyword matches to full context list - if len(no_keyword_matches) > 0: - full_context_fields.extend(no_keyword_matches) - - # Filter out fields that already have answers - # e.g., some have already been filled out by regex chunking - full_context_fields = [ - field for field in full_context_fields if field not in ac_answers_dict.keys() - ] - return full_context_fields - - -def run_full_context_prompts( - ac_answers_dict: dict[str, str], - full_context_fields: list, - contract_text: str, - filename: str, -) -> dict[str, str]: - 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( - context=contract_text[0 : min(100000, len(contract_text) - 1)], - questions=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 AC fields have a key, populated with N/A if not from prompt - for key in prompts.AC_DICT.keys(): - if key not in ac_answers_dict: - ac_answers_dict[key] = "N/A" - - return ac_answers_dict - - -def run_ac_conditional_prompts( - ac_answers_dict: dict[str, str], contract_text: str, filename: str -) -> dict[str, str]: - - # Non-Renewal - Days - if ac_answers_dict["NON_RENEWAL_LANGUAGE"] != "N/A": - nrd_prompt = prompts.AC_SINGLE_FIELD_TEMPLATE( - 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_answers_dict["NON_RENEWAL_DAYS"] = nrd_answer - else: - ac_answers_dict["NON_RENEWAL_DAYS"] = "N/A" - - # Clean Up Contract Effective Date - if ( - 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] - ) - date_answer = claude_funcs.invoke_claude( - date_prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, 124 - ) - ac_answers_dict["CONTRACT_EFFECTIVE_DT"] = date_answer - - return ac_answers_dict diff --git a/fieldExtraction/src/apply_cnc_hotfix.py b/fieldExtraction/src/apply_cnc_hotfix.py index 2a503ba..146d695 100644 --- a/fieldExtraction/src/apply_cnc_hotfix.py +++ b/fieldExtraction/src/apply_cnc_hotfix.py @@ -1,16 +1,16 @@ -import utils -import config -import cnc_hotfix -import preprocessing_funcs -import keywords -import valid - +import concurrent.futures +import os import pandas as pd -import os from rapidfuzz import fuzz, process -import concurrent.futures + +import cnc_hotfix +import config +import io_utils +import keywords +import preprocessing_funcs +import valid clean_file_name = config.get_arg_value('clean_file', '') ABC_PATH = f'{clean_file_name}' @@ -152,7 +152,7 @@ def main(): elif '.csv' in ABC_PATH: abc_clean = pd.read_csv(ABC_PATH) elif '.xlsb' in ABC_PATH: - abc_clean = utils.read_xlsb(ABC_PATH) + abc_clean = io_utils.read_xlsb(ABC_PATH) # Print current columns and identify invalid ones @@ -182,7 +182,7 @@ def main(): # Read input dict print("Reading input .txt files...") - input_dict = utils.read_input() + input_dict = io_utils.read_input() input_dict = {'Filename: ' + filename.replace('.txt', '').strip() : contract_text for filename, contract_text in input_dict.items()} if config.FILTER_ALREADY_PROCESSED: input_dict = {filename : contract_text for filename, contract_text in input_dict.items() if filename not in already_ran} diff --git a/fieldExtraction/src/batch_tracking.py b/fieldExtraction/src/batch_tracking.py index 39ed452..53f372c 100644 --- a/fieldExtraction/src/batch_tracking.py +++ b/fieldExtraction/src/batch_tracking.py @@ -1,11 +1,14 @@ -import pandas as pd -import boto3 import csv +import os +import subprocess from datetime import datetime from pathlib import Path -import subprocess + +import boto3 +import pandas as pd + import config -import os + class BatchTracker: def __init__(self): diff --git a/fieldExtraction/src/bottom_up_funcs.py b/fieldExtraction/src/bottom_up_funcs.py deleted file mode 100644 index ab75a3e..0000000 --- a/fieldExtraction/src/bottom_up_funcs.py +++ /dev/null @@ -1,157 +0,0 @@ -import dict_operations -import postprocessing_funcs -import prompts -import config -import claude_funcs -import utils - -import difflib -import re - - -def run_bottom_up(filename: str, text_dict: dict) -> list[dict]: - """Processes the text of a document using a two-tiered Bottom Up approach to extract key financial and operational information. - - This function first runs BOTTOM_UP_PRIMARY and BOTTOM_UP_SECONDARY prompts, then processes these initial results to further - refine and structure them into a dictionary form. - - Args: - filename (str): The name of the file being processed, used to tag output data. - text_dict (dict): A dictionary of text keyed by page numbers that contains the content to be analyzed. - - Returns: - list[dict]: A list of dictionaries with each dictionary containing refined and structured information - from both primary and secondary Bottom Up analyses, all tagged with the filename. - """ - - # Bottom Up Primary - FULL_SERVICE, FULL_METHODOLOGY - bu_primary_results = run_bottom_up_primary(text_dict, 8000, filename) - - # Bottom Up Secondary - bu_secondary_results = run_bottom_up_secondary( - bu_primary_results, text_dict, 8000, filename - ) - - for d in bu_secondary_results: - d["Filename"] = filename - return bu_secondary_results # List of dictionaries - - -def run_bottom_up_primary( - text_dict: dict, tokens: int, filename: str -) -> dict[str, str]: - """Executes the primary Bottom Up processing for pages that contain reimbursement terms. - - This function scans through a dictionary of page texts, identifies pages containing reimbursement terms, - and processes those pages using Claude to extract relevant information. - - Args: - text_dict (dict): A dictionary containing text content of documents keyed by page numbers. - tokens (int): The maximum number of tokens to use in the language model invocation for generating the response. Deprecated. - filename (str): The filename of the contract in question. Used for logging purposes. - - Returns: - dict[str, str]: A dictionary where each key is a page number (string) and the value is the response from the language model. - """ - - # chunk_dict = preprocess.chunk_text(text_dict) - answer_strings = {} - # for page_number in chunk_dict.keys(): - # if '%' in chunk_dict[page_number] or '$' in chunk_dict[page_number]: - # prompt = prompts.BOTTOM_UP_PRIMARY(chunk_dict[page_number], config.CLIENT_NAME) - for page_number in text_dict.keys(): - if utils.contains_reimbursement(text_dict, page_number): - # Run Primary - prompt = prompts.BOTTOM_UP_PRIMARY( - text_dict[page_number], config.CLIENT_NAME - ) - answer = claude_funcs.invoke_claude( - prompt, config.MODEL_ID_CLAUDE35_SONNET, filename - ) - answer_strings[page_number] = answer - - answer_dicts = dict_operations.primary_string_to_dict(answer_strings, filename) - - answer_dicts = postprocessing_funcs.consolidate_subheader(answer_dicts) - answer_dicts = postprocessing_funcs.filter_service_column( - answer_dicts - ) # Filter on entire Service + Subheader - answer_dicts = postprocessing_funcs.filter_methodology_column(answer_dicts) - - return answer_dicts - - -def run_bottom_up_secondary( - answer_dicts: list[dict], text_dict: dict, tokens: int, filename: str -) -> list[dict]: - """Executes the secondary Bottom Up processing phase on the results obtained from the primary Bottom Up analysis. - - This function enhances the primary results with additional analyses based on configured conditions. Invokes - Claude 3 with tailored prompts to generate structured information that complements the initial results. - - Each piece of data processed possibly undergoes several rounds of checks and transformations, ensuring detailed and - comprehensive output. - - Args: - answer_dicts (list[dict]): Initial processed data from the primary Bottom Up analysis. - text_dict (dict): Dictionary containing the original document text keyed by page numbers. - tokens (int): Token limit for language model invocations. - filename (str): The filename of the contract in question. Used for logging purposes. - - Returns: - list[dict]: A list of dictionaries containing enriched and finalized structured data from both the primary and - secondary analyses. - """ - # Add additional fields to each row - final_dicts = [] - for d in answer_dicts: - if d is not None: - page_num = d["page_num"] - - # if '%' in d['FULL_METHODOLOGY'] or '$' in d['FULL_METHODOLOGY']: - prompt = prompts.BOTTOM_UP_METHODOLOGY_BREAKOUT(d) - answer = claude_funcs.invoke_claude( - prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, max_tokens=512 - ) - answer_dict = dict_operations.secondary_string_to_dict(answer, filename) - d.update(answer_dict) - - # Ensure all bottom up methodology keys are present - for key in [ - "LESSER", - "RATE_STANDARD", - "RATE_SHORT", - "FLAT_FEE_STANDARD", - "LESSER_RATE", - "NOT_TO_EXCEED", - ]: - if key not in d.keys(): - d[key] = "" - - # SHORT_METHODOLOGY - if "%" in d["RATE_STANDARD"]: - d["SHORT_METHODOLOGY"] = get_short_methodology(d["RATE_STANDARD"]) - elif "$" in d["FLAT_FEE_STANDARD"]: - d["SHORT_METHODOLOGY"] = "Flat Fee" - else: - d["SHORT_METHODOLOGY"] = "N/A" - - # Escalator - prompt = prompts.BOTTOM_UP_ESCALATOR(d, text_dict[page_num]) - escalator_answer = claude_funcs.invoke_claude( - prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, max_tokens=1024 - ) - escalator_dict = dict_operations.secondary_string_to_dict( - escalator_answer, filename - ) - d.update(escalator_dict) - - final_dicts.append(d) - - return final_dicts - - -def get_short_methodology(s): - pattern = r"\d+\.*\d*%" - short_string = re.sub(pattern, "%", s) - return short_string diff --git a/fieldExtraction/src/claude_funcs.py b/fieldExtraction/src/claude_funcs.py index 9964975..3966c96 100644 --- a/fieldExtraction/src/claude_funcs.py +++ b/fieldExtraction/src/claude_funcs.py @@ -9,6 +9,7 @@ import anthropic from botocore.exceptions import ClientError, ReadTimeoutError import config +import string_funcs def update_stats(filename, tokens_sent, tokens_received, elapsed_time, model_type): @@ -425,3 +426,13 @@ def ec2_claude_35( raise raise Exception("Max retries reached. Unable to get a response from the model.") + + +def get_ac_answer(prompt, filename, fields): + response_text = invoke_claude( + prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, 8192 + ) + response_text = response_text.replace("_DATE", "_DT") + # response_dict = dict_operations.secondary_string_to_dict(response_text, filename) + response_dict = string_funcs.json_parsing_search(response_text, fields) + return response_dict diff --git a/fieldExtraction/src/cnc_hotfix.py b/fieldExtraction/src/cnc_hotfix.py index 33f78aa..db01861 100644 --- a/fieldExtraction/src/cnc_hotfix.py +++ b/fieldExtraction/src/cnc_hotfix.py @@ -1,32 +1,33 @@ -import utils -import pandas as pd import itertools -import numpy as np import re -import pandas as pd -from valid import STATE_MAP -from prompts import BOTTOM_UP_METHODOLOGY_BREAKOUT -from bottom_up_funcs import get_short_methodology -import claude_funcs -from utils import is_empty -import dict_operations -import config import warnings + +import numpy as np +import pandas as pd + +import claude_funcs +import config +import postprocessing_funcs +import string_funcs +from one_to_n_funcs import get_short_methodology +from prompts import BOTTOM_UP_METHODOLOGY_BREAKOUT +from valid import STATE_MAP + warnings.simplefilter(action='ignore', category=FutureWarning) -import ac_smart_chunking import claude_funcs import cnc_hotfix_effective_date_utils import config -import utils -import valid -import table_funcs -from hotfix_helper_funcs import irs_hotfix, clean_output_postprocess, state_check, check_field_for_matches, npi_hotfix, npi_post_process import keywords -import prompts -import ac_funcs import postprocessing_funcs +import prompts +import smart_chunking_funcs +import string_funcs +import valid +from hotfix_helper_funcs import (check_field_for_matches, + clean_output_postprocess, irs_hotfix, + npi_hotfix, npi_post_process, state_check) ####### Exhibit ####### @@ -305,7 +306,7 @@ def clean_lob(df, text_dict): continue # Issue 1: Missing - if utils.is_empty(row['Line of Business']) and pd.notna(row['Page_Num']): # Check if 'Line of Business' is missing + if string_funcs.is_empty(row['Line of Business']) and pd.notna(row['Page_Num']): # Check if 'Line of Business' is missing # Prepare to search text page_text = text_dict.get(str(int(row['Page_Num'])), "") @@ -361,7 +362,7 @@ def reimb_methodology_fix(rm: str): answer = claude_funcs.invoke_claude( prompt, config.MODEL_ID_CLAUDE35_SONNET, '', max_tokens=512 ) - answer_dict = dict_operations.secondary_string_to_dict(answer, '') + answer_dict = string_funcs.secondary_string_to_dict(answer, '') d.update(answer_dict) # Ensure all bottom up methodology keys are present @@ -621,15 +622,15 @@ def clean_contract_effective_date( is_meridian = any(re.search(r'\b' + keyword + r'\b', payer_name, re.IGNORECASE) for keyword in ['meridian']) - if utils.is_empty(contract_effective_date): # Try prompting (with smart chunking) when the contract effective date is missing + if string_funcs.is_empty(contract_effective_date): # Try prompting (with smart chunking) when the contract effective date is missing try: answer,d,page_list,claude_answer_raw = contract_effective_date_fix(file_name,text_dict,is_meridian) except Exception as e: # print(f"Error processing {file_name}, got error: {e}") answer,d,page_list,claude_answer_raw = "N/A",{},[],"" try: - corrected_date = utils.convert_to_us_date_format(str(answer)) # apply date formatting fix - except utils.InvalidDateException as e: + corrected_date = postprocessing_funcs.convert_to_us_date_format(str(answer)) # apply date formatting fix + except postprocessing_funcs.InvalidDateException as e: corrected_date = "N/A" else: corrected_date = contract_effective_date # don't correct format when we already have an answer @@ -701,7 +702,7 @@ def clean_term_group(df: pd.DataFrame, prompt, config.MODEL_ID_CLAUDE35_SONNET, contract_name, 8192 ) field_group_answer_raw = field_group_answer_raw.replace("_DATE", "_DT") - field_group_answer_dict = ac_funcs.json_parsing_search(field_group_answer_raw, fields) + field_group_answer_dict = string_funcs.json_parsing_search(field_group_answer_raw, fields) for field in fields: if field in field_group_answer_dict: @@ -753,7 +754,7 @@ def clean_agreement_name(df: pd.DataFrame, if prompt_answer_raw == "N/A": agreement_keywords = ["AGREEMENT", "AMENDMENT", "ADDENDUM", "PROVIDER", "CONTRACT", "MEMORANDUM", "LETTER", "REQUEST", "EFFECTIVE"] - page_list = ac_smart_chunking.chunk_hierarchical( + page_list = smart_chunking_funcs.chunk_hierarchical( text_dict, agreement_keywords, False ) context = "\n".join( diff --git a/fieldExtraction/src/cnc_hotfix_effective_date_utils.py b/fieldExtraction/src/cnc_hotfix_effective_date_utils.py index 5938c48..f40c046 100644 --- a/fieldExtraction/src/cnc_hotfix_effective_date_utils.py +++ b/fieldExtraction/src/cnc_hotfix_effective_date_utils.py @@ -1,6 +1,6 @@ import re -import ac_smart_chunking +import smart_chunking_funcs regex_backticks = r"\|([^|]+)\|" @@ -66,7 +66,7 @@ def chunk_with_include_exclude_keywords( pages_containing_date = set() for page_num, page_content in text_dict.items(): - dates_list = re.findall(ac_smart_chunking.DATE_PATTERN, page_content) + dates_list = re.findall(smart_chunking_funcs.DATE_PATTERN, page_content) if len(dates_list) > 0: pages_containing_date.add(page_num) diff --git a/fieldExtraction/src/conditional_funcs.py b/fieldExtraction/src/conditional_funcs.py index 7ce796c..6dd4f8e 100644 --- a/fieldExtraction/src/conditional_funcs.py +++ b/fieldExtraction/src/conditional_funcs.py @@ -1,7 +1,6 @@ -import prompts import claude_funcs -import dict_operations import config +import prompts def run_conditional(combined_results, text_dict, filename): @@ -19,8 +18,6 @@ def run_conditional(combined_results, text_dict, filename): prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, max_tokens=128 ) d["CONTRACT_LOB"] = lob_answer - # lob_dict = dict_operations.secondary_string_to_dict(lob_answer, filename) - # d.update(lob_dict) # Prov Type Level 2 prompt = prompts.CONDITIONAL_PROV_2(d, text_dict[page_num]) diff --git a/fieldExtraction/src/config.py b/fieldExtraction/src/config.py index c0a3a38..4e40784 100644 --- a/fieldExtraction/src/config.py +++ b/fieldExtraction/src/config.py @@ -5,7 +5,6 @@ from collections import defaultdict from datetime import datetime import boto3 -from boto3.session import Session from botocore.config import Config from dotenv import load_dotenv @@ -35,6 +34,7 @@ load_dotenv() from botocore.exceptions import NoCredentialsError + def check_s3_bucket_exists(s3_client, bucket_name: str): try: s3_client.head_bucket(Bucket=bucket_name) diff --git a/fieldExtraction/src/consolidate_output.py b/fieldExtraction/src/consolidate_output.py index 1c3a2da..0bce333 100644 --- a/fieldExtraction/src/consolidate_output.py +++ b/fieldExtraction/src/consolidate_output.py @@ -1,15 +1,16 @@ -import os -import pandas as pd +import os from pathlib import Path -from datetime import datetime + +import pandas as pd import config import costlog_funcs -import valid import tracking -from qa_qc_helpers import (table_stats_qaqc, perform_qc_qa_tests_both, - perform_qc_qa_tests_ac, - perform_qc_qa_tests_b, perform_qc_qa, save_qcqa_wb) +import valid +from qa_qc_helpers import (perform_qc_qa, perform_qc_qa_tests_ac, + perform_qc_qa_tests_b, perform_qc_qa_tests_both, + save_qcqa_wb, table_stats_qaqc) + def upload_tracking_files(run_timestamp): """Upload tracking files and aggregated cost logs to S3 during batch processing.""" diff --git a/fieldExtraction/src/costlog_funcs.py b/fieldExtraction/src/costlog_funcs.py index 9687aa1..e8eeea0 100644 --- a/fieldExtraction/src/costlog_funcs.py +++ b/fieldExtraction/src/costlog_funcs.py @@ -1,4 +1,5 @@ import pandas as pd + import config diff --git a/fieldExtraction/src/dict_operations.py b/fieldExtraction/src/dict_operations.py deleted file mode 100644 index 4907d88..0000000 --- a/fieldExtraction/src/dict_operations.py +++ /dev/null @@ -1,79 +0,0 @@ -import re -import json -from collections import defaultdict - -import config -import prompts -import claude_funcs - - -def secondary_string_to_dict(dict_string, filename): - """ - Converts a string representation of a dictionary into an actual dictionary object, handling potential formatting issues. - - This function cleans the string by removing specific unwanted characters and markers that may interfere with JSON parsing. - It then locates the substring that correctly forms a dictionary format, attempts to parse it as JSON, and handles common - parsing errors by replacing problematic single quotes with double quotes before re-parsing. - - Parameters: - dict_string (str): The string containing the dictionary-like content, potentially surrounded by extra text or characters. - - Returns: - dict: The dictionary obtained from parsing the cleaned and corrected string. - """ - start_index = dict_string.find("{") - end_index = dict_string.rfind("}") + 1 - dict_substring = dict_string[start_index:end_index] - try: - result_dict = json.loads(dict_substring) - except: - try: - dict_substring = dict_substring.replace("'", '"') - result_dict = json.loads(dict_substring) - except: - try: - prompt = prompts.FIX_JSON(dict_substring) - dict_substring = claude_funcs.invoke_claude( - prompt, config.MODEL_ID_CLAUDE3_HAIKU, filename - ) - result_dict = json.loads(dict_substring) - except Exception as e: - print(e) - result_dict = {} - return result_dict - - -def primary_string_to_dict(string_dict, filename): - """ - Converts a dictionary of strings, where each string represents multiple dictionary entries, into a list of dictionaries, - augmenting each with metadata such as page number and filename. - - This function iterates over each page number's string data, extracting and converting string representations of - dictionaries into actual dictionary objects. It handles and cleans the string format to properly parse it into dictionaries. - All dictionaries are then augmented with their respective page number and the filename before being compiled into a list. - - Parameters: - string_dict (dict): A dictionary where keys are page numbers and values are strings containing multiple dictionary entries. - filename (str): The filename associated with these entries, used to tag each resulting dictionary. - - Returns: - list of dict: A list of dictionaries, each representing data extracted and converted from the input string, tagged with - their page number and filename. - """ - data = [] - pattern = r"\{.*?\}" - for page_num in string_dict.keys(): - primary_list = string_dict[page_num] - dicts = primary_list.split("[")[1] # Strip front - dicts = dicts.split("]")[0] # Strip back - dicts = dicts.replace("\n", "") # Remove new lines - dicts = dicts.replace("<<<", "") - dicts = dicts.replace(">>>", "") - dicts = re.sub(r"(? 0 else 0 - b_progress = min((completed_b / total_b * 100), 100.0) if total_b > 0 else 0 - return ac_progress, b_progress - -def contains_reimbursement(text, page="1"): # string_funcs.py - if isinstance(text, dict): - return page.isdigit() and ( - "%" in text[page] - or "$" in text[page] - or "percent " in text[page] - or "compensation schedule" in text[page].lower() - or "reimbursement schedule" in text[page].lower() - ) - elif isinstance(text, str): - return ( - "%" in text - or "$" in text - or "percent " in text - or "compensation schedule" in text.lower() - or "reimbursement schedule" in text.lower() - ) - else: - print("contains_reimbursement - Invalid data type") - def filter_already_processed(input_dict, input_dict_b): # io_utils.py """Filter out already processed files based on either local files or S3 master tracking""" if config.WRITE_TO_S3: # Get processed files from S3 master tracking for this specific batch already_processed_ac, already_processed_b = tracking.get_already_processed_files(config.BATCH_ID) - + # Keep files that still need either AC or B processing input_dict_ac = { - k: v for k, v in input_dict.items() + k: v for k, v in input_dict.items() if k not in already_processed_ac } - + input_dict_b = { - k: v for k, v in input_dict_b.items() + k: v for k, v in input_dict_b.items() if k not in already_processed_b } - + print(f"Filtering based on S3 master tracking for batch {config.BATCH_ID}:") print(f"AC: Found {len(already_processed_ac)} already processed, {len(input_dict_ac)} remaining") print(f"B: Found {len(already_processed_b)} already processed, {len(input_dict_b)} remaining") - + else: # Original local directory checking logic already_processed_ac, already_processed_b = [], [] @@ -192,17 +164,17 @@ def filter_already_processed(input_dict, input_dict_b): # io_utils.py already_processed_ac.append(folder_name + ".txt") if config.B_RESULTS_NAME in os.listdir(folder_path): already_processed_b.append(folder_name + ".txt") - + input_dict_ac = { k: v for k, v in input_dict.items() if k not in already_processed_ac } - + input_dict_b = { k: v for k, v in input_dict_b.items() if k not in already_processed_b } - + print(f"Filtering based on local output directory:") print(f"AC: Found {len(already_processed_ac)} already processed, {len(input_dict_ac)} remaining") print(f"B: Found {len(already_processed_b)} already processed, {len(input_dict_b)} remaining") @@ -210,140 +182,12 @@ def filter_already_processed(input_dict, input_dict_b): # io_utils.py return input_dict_ac, input_dict_b - - -def is_empty(value): # string_funcs.py - if pd.isna(value): - return True - else: - empty_values = [None, "", "N/A", "NA", "null", "none", "NaN", np.nan, "nan"] - return value in empty_values - - -def find_regex_matches( # regex_funcs.py - 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) # make it append - page_list.append(page) - - return matches, page_list - - - -def add_hyphen_if_needed(input_str): # postprocessing_funcs.py - """Checks for specific patterns in the input string and modifies it by adding - a '0' at the start and a hyphen at the 3rd index, or just adds a hyphen at - the 3rd index if it's a 9-digit number. - - If the input string is exactly 8 characters long, it adds a '0' at the start - and inserts a hyphen at the 3rd position. If the string matches the pattern - of a 9-digit number (exactly 9 digits), it inserts a hyphen at the 3rd index. - If the input string matches the pattern 'x-xxxxxxx', it adds '0' at the start. - - All conditions will return a string in the format 'xx-xxxxxxx'. - - Args: - - input_str (str): The input string (IRS). - - Returns: - - str: The modified string in the format 'xx-xxxxxxx'. - - Example: - - '1-2345678' → '01-2345678' - - '12345678' → '01-2345678' - - '123456789' → '12-3456789' - """ - # Check if the string matches the pattern 'x-xxxxxxx' (1 character, hyphen, 7 characters) - if re.fullmatch(r"\d-\d{7}", input_str): - # Add '0' at the start - input_str = '0' + input_str - - # Check if the string is exactly 'xxxxxxxx' (8 characters) - elif bool(re.fullmatch(r"\b\d{8}\b", input_str)): - # Add '0' at the start and insert a hyphen at the 3rd index - input_str = '0' + input_str - input_str = input_str[:2] + '-' + input_str[2:] - - # If the string matches the pattern of a 9-digit number, add a hyphen at the 3rd index - elif bool(re.fullmatch(r"\b\d{9}\b", input_str)): - input_str = input_str[:2] + '-' + input_str[2:] - - # Ensure the output is in the format 'xx-xxxxxxx' - if not bool(re.fullmatch(r"\b\d{2}-\d{7}\b", input_str)): - input_str = None - - return input_str - -def convert_to_us_date_format(date_str: str) -> str: # postprocessing_funcs.py - """ - Input date string is expected to be in the format YYYY-MM-DD - - Ensure the date is in MM/DD/YYYY format. If the month is greater than 12, swap the month and day. - If the input is `N/A` or some variant (defined by is_empty), InvalidDateException will be raised. - Args: - date_str (str): The date string to be formatted. We expect it to come in as YYYY-MM-DD format (note the hyphens as well) - - Returns: - str: The date string formatted as MM/DD/YYYY. - - Raises: - InvalidDateException: If the date string is not in the expected format. - """ - - if not isinstance(date_str, str): - raise TypeError(f"Date must be a string, got: {type(date_str)}") - - if is_empty(date_str) or re.fullmatch(r"\d{4}-\d{1,2}-\d{1,2}", date_str) is None: - raise InvalidDateException(f"Invalid date format: {date_str}") - - # Split the date string into year, month, and day - year, month, day = date_str.split("-") - - year_int = int(year) - month_int = int(month) - day_int = int(day) - - if year_int == 0 or month_int == 0 or day_int == 0: - raise InvalidDateException(f"Invalid date format: {date_str}") - - if month_int > 12 and 0 < day_int <= 31: - # Swap the month and day if the month is greater than 12 - month_int, day_int = day_int, month_int - - if 0 < month_int <= 12 and 0 < day_int <= 31: - # Return the formatted date in MM/DD/YYYY format - year, month, day = str(year_int).zfill(4), str(month_int).zfill(2), str(day_int).zfill(2) - return "/".join([month, day, year]) - - else: - # Raise an exception if the month or day is out of range - raise InvalidDateException(f"Invalid date format: {date_str}") - - -def remove_unnamed_columns(df): # postprocessing_funcs.py - return df[[col for col in df.columns if 'Unnamed' not in col]] - def remove_txt_extension(filename): # io_utils.py if isinstance(filename, str) and filename.lower().endswith(".txt"): return filename[:-4] return filename + def read_xlsb(path): # io_utils.py with open_workbook(path) as wb: with wb.get_sheet(1) as sheet: @@ -351,5 +195,4 @@ def read_xlsb(path): # io_utils.py for row in sheet.rows(): rows.append([item.v for item in row]) # Extracting cell values df = pd.DataFrame(rows[1:], columns=rows[0]) # Converting to DataFrame, assuming first row is header - return df - + return df \ No newline at end of file diff --git a/fieldExtraction/src/main.py b/fieldExtraction/src/main.py index 59d515e..9a5e388 100644 --- a/fieldExtraction/src/main.py +++ b/fieldExtraction/src/main.py @@ -1,30 +1,27 @@ import concurrent.futures -import traceback import os -import csv -import time -import sys import random -import pandas as pd +import traceback from datetime import datetime -import random + random.seed(42) -import utils import config -import claude_funcs -import file_processing -import tracking import consolidate_output -from batch_tracking import BatchTracker +import file_processing +import io_utils import qa_qc_helpers +import string_funcs +import tracking +from batch_tracking import BatchTracker + def process_b(item): filename, contract_text = item start_time = datetime.now() - if utils.contains_reimbursement(contract_text, 0): + if string_funcs.contains_reimbursement(contract_text, 0): try: file_processing.run_b_prompts(item) end_time = datetime.now() @@ -228,13 +225,13 @@ def main(): run_timestamp = datetime.now().strftime(f"run_%Y%m%d_%H:%M_{config.BATCH_ID}") # Read and process input - input_dict = utils.read_input() + input_dict = io_utils.read_input() print(f"Total Input Files : {len(input_dict)}") # Filter B - files with reimbursement input_dict_b = { k: v for k, v in input_dict.items() - if utils.contains_reimbursement(str(v)) + if string_funcs.contains_reimbursement(str(v)) } batch_tracker.set_input_file_counts( @@ -313,7 +310,7 @@ def main(): stats = process_results(result, filename, input_dict_b, stats) # Calculate and display progress - ac_progress, b_progress = utils.calculate_progress_stats( + ac_progress, b_progress = tracking.calculate_progress_stats( stats['completed_ac'], total_ac, stats['completed_b'], total_b ) diff --git a/fieldExtraction/src/top_down_funcs.py b/fieldExtraction/src/one_to_n_funcs.py similarity index 56% rename from fieldExtraction/src/top_down_funcs.py rename to fieldExtraction/src/one_to_n_funcs.py index 4a918df..7c42cf0 100644 --- a/fieldExtraction/src/top_down_funcs.py +++ b/fieldExtraction/src/one_to_n_funcs.py @@ -1,14 +1,162 @@ -import prompts -import claude_funcs -import utils -import config -import dict_operations -import valid -import postprocessing_funcs -import csv -import json + + import re +import claude_funcs +import config +import postprocessing_funcs +import prompts +import string_funcs +import valid + + +def run_bottom_up_primary( + text_dict: dict, tokens: int, filename: str +) -> dict[str, str]: + """Executes the primary Bottom Up processing for pages that contain reimbursement terms. + + This function scans through a dictionary of page texts, identifies pages containing reimbursement terms, + and processes those pages using Claude to extract relevant information. + + Args: + text_dict (dict): A dictionary containing text content of documents keyed by page numbers. + tokens (int): The maximum number of tokens to use in the language model invocation for generating the response. Deprecated. + filename (str): The filename of the contract in question. Used for logging purposes. + + Returns: + dict[str, str]: A dictionary where each key is a page number (string) and the value is the response from the language model. + """ + + # chunk_dict = preprocess.chunk_text(text_dict) + answer_strings = {} + # for page_number in chunk_dict.keys(): + # if '%' in chunk_dict[page_number] or '$' in chunk_dict[page_number]: + # prompt = prompts.BOTTOM_UP_PRIMARY(chunk_dict[page_number], config.CLIENT_NAME) + for page_number in text_dict.keys(): + if string_funcs.contains_reimbursement(text_dict, page_number): + # Run Primary + prompt = prompts.BOTTOM_UP_PRIMARY( + text_dict[page_number], config.CLIENT_NAME + ) + answer = claude_funcs.invoke_claude( + prompt, config.MODEL_ID_CLAUDE35_SONNET, filename + ) + answer_strings[page_number] = answer + + answer_dicts = string_funcs.primary_string_to_dict(answer_strings, filename) + + answer_dicts = postprocessing_funcs.consolidate_subheader(answer_dicts) + answer_dicts = postprocessing_funcs.filter_service_column( + answer_dicts + ) # Filter on entire Service + Subheader + answer_dicts = postprocessing_funcs.filter_methodology_column(answer_dicts) + + return answer_dicts + + +def get_short_methodology(s): + pattern = r"\d+\.*\d*%" + short_string = re.sub(pattern, "%", s) + return short_string + + +def run_bottom_up_secondary( + answer_dicts: list[dict], text_dict: dict, tokens: int, filename: str +) -> list[dict]: + """Executes the secondary Bottom Up processing phase on the results obtained from the primary Bottom Up analysis. + + This function enhances the primary results with additional analyses based on configured conditions. Invokes + Claude 3 with tailored prompts to generate structured information that complements the initial results. + + Each piece of data processed possibly undergoes several rounds of checks and transformations, ensuring detailed and + comprehensive output. + + Args: + answer_dicts (list[dict]): Initial processed data from the primary Bottom Up analysis. + text_dict (dict): Dictionary containing the original document text keyed by page numbers. + tokens (int): Token limit for language model invocations. + filename (str): The filename of the contract in question. Used for logging purposes. + + Returns: + list[dict]: A list of dictionaries containing enriched and finalized structured data from both the primary and + secondary analyses. + """ + # Add additional fields to each row + final_dicts = [] + for d in answer_dicts: + if d is not None: + page_num = d["page_num"] + + # if '%' in d['FULL_METHODOLOGY'] or '$' in d['FULL_METHODOLOGY']: + prompt = prompts.BOTTOM_UP_METHODOLOGY_BREAKOUT(d) + answer = claude_funcs.invoke_claude( + prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, max_tokens=512 + ) + answer_dict = string_funcs.secondary_string_to_dict(answer, filename) + d.update(answer_dict) + + # Ensure all bottom up methodology keys are present + for key in [ + "LESSER", + "RATE_STANDARD", + "RATE_SHORT", + "FLAT_FEE_STANDARD", + "LESSER_RATE", + "NOT_TO_EXCEED", + ]: + if key not in d.keys(): + d[key] = "" + + # SHORT_METHODOLOGY + if "%" in d["RATE_STANDARD"]: + d["SHORT_METHODOLOGY"] = get_short_methodology(d["RATE_STANDARD"]) + elif "$" in d["FLAT_FEE_STANDARD"]: + d["SHORT_METHODOLOGY"] = "Flat Fee" + else: + d["SHORT_METHODOLOGY"] = "N/A" + + # Escalator + prompt = prompts.BOTTOM_UP_ESCALATOR(d, text_dict[page_num]) + escalator_answer = claude_funcs.invoke_claude( + prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, max_tokens=1024 + ) + escalator_dict = string_funcs.secondary_string_to_dict( + escalator_answer, filename + ) + d.update(escalator_dict) + + final_dicts.append(d) + + return final_dicts + + +def run_bottom_up(filename: str, text_dict: dict) -> list[dict]: + """Processes the text of a document using a two-tiered Bottom Up approach to extract key financial and operational information. + + This function first runs BOTTOM_UP_PRIMARY and BOTTOM_UP_SECONDARY prompts, then processes these initial results to further + refine and structure them into a dictionary form. + + Args: + filename (str): The name of the file being processed, used to tag output data. + text_dict (dict): A dictionary of text keyed by page numbers that contains the content to be analyzed. + + Returns: + list[dict]: A list of dictionaries with each dictionary containing refined and structured information + from both primary and secondary Bottom Up analyses, all tagged with the filename. + """ + + # Bottom Up Primary - FULL_SERVICE, FULL_METHODOLOGY + bu_primary_results = run_bottom_up_primary(text_dict, 8000, filename) + + # Bottom Up Secondary + bu_secondary_results = run_bottom_up_secondary( + bu_primary_results, text_dict, 8000, filename + ) + + for d in bu_secondary_results: + d["Filename"] = filename + return bu_secondary_results # List of dictionaries + def get_exhibit_range(text_dict, exhibit_pages, bu_page): lower_bound = max( @@ -36,59 +184,6 @@ def get_exhibit_range(text_dict, exhibit_pages, bu_page): 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} - - td_page_dict = {} - td_answer_dict = {} - - for bu_page in bu_pages: - td_pages = get_exhibit_range(text_dict, exhibit_pages, bu_page) - if td_pages not in td_page_dict.values(): - answer_dict = run_top_down_primary(text_dict, td_pages, filename) - td_page_dict[bu_page] = td_pages - td_answer_dict[bu_page] = answer_dict - else: - td_page_dict[bu_page] = td_pages - for key, value in td_page_dict.items(): - if value == td_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"]] - bu_dict.update(answer_dict) - merged_dicts.append(bu_dict) - - # Run and Merge Exclusions - exclusion_dict = get_td_dict( - text_dict, filename, prompts.TOP_DOWN_EXCLUSIONS, valid.EXCLUSION_TERMS - ) - final_dicts = add_exclusions(merged_dicts, exclusion_dict) - - # Run and Merge Medically Necessary - medically_necessary_dict = get_td_dict( - text_dict, - filename, - prompts.TOP_DOWN_MEDICALLY_NECESSARY, - valid.VALID_MEDICALLY_NECESSARY, - ) - medically_necessary_dict = postprocessing_funcs.filter_dict( - medically_necessary_dict, - pattern=re.compile( - "|".join( - re.escape(keyword) for keyword in valid.INVALID_MEDICALLY_NECESSARY - ), - re.IGNORECASE, - ), - ) - 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]) @@ -96,7 +191,7 @@ def run_top_down_primary(text_dict, td_pages, filename): answer = claude_funcs.invoke_claude( prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, max_tokens=4096 ) - answer_dict = dict_operations.secondary_string_to_dict(answer, filename) + answer_dict = string_funcs.secondary_string_to_dict(answer, filename) for field in prompts.TD_PRIMARY_FIELDS: if field not in answer_dict: answer_dict[field] = "N/A" @@ -158,7 +253,7 @@ def add_medically_necessary(merged_dicts, medically_necessary_dict): d["MEDICAL_NECESSITY_LANGUAGE"] = medically_necessary_dict[ str(closest_page) ] - if not utils.is_empty(d["MEDICAL_NECESSITY_LANGUAGE"]): + if not string_funcs.is_empty(d["MEDICAL_NECESSITY_LANGUAGE"]): d["MEDICAL_NECESSITY_LANGUAGE_IND"] = "Y" else: d["MEDICAL_NECESSITY_LANGUAGE_IND"] = "N" @@ -223,6 +318,59 @@ def add_exclusions(merged_dicts, exclusions_dict): return merged_dicts +def run_top_down(filename, text_dict, bu_results, exhibit_pages): + bu_pages = {bu_dict["page_num"] for bu_dict in bu_results} + + td_page_dict = {} + td_answer_dict = {} + + for bu_page in bu_pages: + td_pages = get_exhibit_range(text_dict, exhibit_pages, bu_page) + if td_pages not in td_page_dict.values(): + answer_dict = run_top_down_primary(text_dict, td_pages, filename) + td_page_dict[bu_page] = td_pages + td_answer_dict[bu_page] = answer_dict + else: + td_page_dict[bu_page] = td_pages + for key, value in td_page_dict.items(): + if value == td_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"]] + bu_dict.update(answer_dict) + merged_dicts.append(bu_dict) + + # Run and Merge Exclusions + exclusion_dict = get_td_dict( + text_dict, filename, prompts.TOP_DOWN_EXCLUSIONS, valid.EXCLUSION_TERMS + ) + final_dicts = add_exclusions(merged_dicts, exclusion_dict) + + # Run and Merge Medically Necessary + medically_necessary_dict = get_td_dict( + text_dict, + filename, + prompts.TOP_DOWN_MEDICALLY_NECESSARY, + valid.VALID_MEDICALLY_NECESSARY, + ) + medically_necessary_dict = postprocessing_funcs.filter_dict( + medically_necessary_dict, + pattern=re.compile( + "|".join( + re.escape(keyword) for keyword in valid.INVALID_MEDICALLY_NECESSARY + ), + re.IGNORECASE, + ), + ) + final_dicts = add_medically_necessary(final_dicts, medically_necessary_dict) + + return final_dicts + + def get_exhibit_pages(text_dict, filename): exhibit_pages = [] for page_num, page in text_dict.items(): @@ -232,4 +380,4 @@ def get_exhibit_pages(text_dict, filename): ) if "Y" in answer: exhibit_pages.append(page_num) - return exhibit_pages + return exhibit_pages \ No newline at end of file diff --git a/fieldExtraction/src/postprocess.py b/fieldExtraction/src/postprocess.py index e63feca..cc3f5c6 100644 --- a/fieldExtraction/src/postprocess.py +++ b/fieldExtraction/src/postprocess.py @@ -1,14 +1,5 @@ -import pandas as pd -import re -import os -import numpy as np - import postprocessing_funcs import valid -import prompts -import claude_funcs -import config -import utils def b_postprocess(filename, df, pages): diff --git a/fieldExtraction/src/postprocessing_funcs.py b/fieldExtraction/src/postprocessing_funcs.py index f88a85f..8e32eb8 100644 --- a/fieldExtraction/src/postprocessing_funcs.py +++ b/fieldExtraction/src/postprocessing_funcs.py @@ -1,18 +1,16 @@ -import pandas as pd -import numpy as np +import logging import re -import difflib +import numpy as np +import pandas as pd + +import claude_funcs import config import prompts import valid -from valid import DERIVED_INDICATOR_FIELDS -import claude_funcs -from utils import is_empty from hotfix_helper_funcs import state_check - -import logging -import qa_qc_helpers +from string_funcs import is_empty +from valid import DERIVED_INDICATOR_FIELDS logging.basicConfig( level=logging.ERROR, format="%(asctime)s - %(levelname)s - %(message)s" @@ -642,3 +640,102 @@ def yn_fixes(df: pd.DataFrame) -> pd.DataFrame: df.loc[df[yn] != "Y", lang] = "" return df + + +def add_hyphen_if_needed(input_str): # postprocessing_funcs.py + """Checks for specific patterns in the input string and modifies it by adding + a '0' at the start and a hyphen at the 3rd index, or just adds a hyphen at + the 3rd index if it's a 9-digit number. + + If the input string is exactly 8 characters long, it adds a '0' at the start + and inserts a hyphen at the 3rd position. If the string matches the pattern + of a 9-digit number (exactly 9 digits), it inserts a hyphen at the 3rd index. + If the input string matches the pattern 'x-xxxxxxx', it adds '0' at the start. + + All conditions will return a string in the format 'xx-xxxxxxx'. + + Args: + - input_str (str): The input string (IRS). + + Returns: + - str: The modified string in the format 'xx-xxxxxxx'. + + Example: + - '1-2345678' → '01-2345678' + - '12345678' → '01-2345678' + - '123456789' → '12-3456789' + """ + # Check if the string matches the pattern 'x-xxxxxxx' (1 character, hyphen, 7 characters) + if re.fullmatch(r"\d-\d{7}", input_str): + # Add '0' at the start + input_str = '0' + input_str + + # Check if the string is exactly 'xxxxxxxx' (8 characters) + elif bool(re.fullmatch(r"\b\d{8}\b", input_str)): + # Add '0' at the start and insert a hyphen at the 3rd index + input_str = '0' + input_str + input_str = input_str[:2] + '-' + input_str[2:] + + # If the string matches the pattern of a 9-digit number, add a hyphen at the 3rd index + elif bool(re.fullmatch(r"\b\d{9}\b", input_str)): + input_str = input_str[:2] + '-' + input_str[2:] + + # Ensure the output is in the format 'xx-xxxxxxx' + if not bool(re.fullmatch(r"\b\d{2}-\d{7}\b", input_str)): + input_str = None + + return input_str + + +class InvalidDateException(Exception): + pass + + +def convert_to_us_date_format(date_str: str) -> str: # postprocessing_funcs.py + """ + Input date string is expected to be in the format YYYY-MM-DD + + Ensure the date is in MM/DD/YYYY format. If the month is greater than 12, swap the month and day. + If the input is `N/A` or some variant (defined by is_empty), InvalidDateException will be raised. + Args: + date_str (str): The date string to be formatted. We expect it to come in as YYYY-MM-DD format (note the hyphens as well) + + Returns: + str: The date string formatted as MM/DD/YYYY. + + Raises: + InvalidDateException: If the date string is not in the expected format. + """ + + if not isinstance(date_str, str): + raise TypeError(f"Date must be a string, got: {type(date_str)}") + + if is_empty(date_str) or re.fullmatch(r"\d{4}-\d{1,2}-\d{1,2}", date_str) is None: + raise InvalidDateException(f"Invalid date format: {date_str}") + + # Split the date string into year, month, and day + year, month, day = date_str.split("-") + + year_int = int(year) + month_int = int(month) + day_int = int(day) + + if year_int == 0 or month_int == 0 or day_int == 0: + raise InvalidDateException(f"Invalid date format: {date_str}") + + if month_int > 12 and 0 < day_int <= 31: + # Swap the month and day if the month is greater than 12 + month_int, day_int = day_int, month_int + + if 0 < month_int <= 12 and 0 < day_int <= 31: + # Return the formatted date in MM/DD/YYYY format + year, month, day = str(year_int).zfill(4), str(month_int).zfill(2), str(day_int).zfill(2) + return "/".join([month, day, year]) + + else: + # Raise an exception if the month or day is out of range + raise InvalidDateException(f"Invalid date format: {date_str}") + + +def remove_unnamed_columns(df): # postprocessing_funcs.py + return df[[col for col in df.columns if 'Unnamed' not in col]] diff --git a/fieldExtraction/src/preprocess.py b/fieldExtraction/src/preprocess.py index 91199d7..a0fbdf2 100644 --- a/fieldExtraction/src/preprocess.py +++ b/fieldExtraction/src/preprocess.py @@ -1,10 +1,11 @@ -import preprocessing_funcs -import top_down_funcs -import keywords -import table_funcs -import config import csv +import config +import keywords +import one_to_n_funcs +import preprocessing_funcs +import table_funcs + def preprocess(contract_text, filename, fields=config.FIELDS): @@ -20,7 +21,7 @@ def preprocess(contract_text, filename, fields=config.FIELDS): text_dict, top_sheet_dict = preprocessing_funcs.filter_quick_review(text_dict) if fields == "b": - exhibit_pages = top_down_funcs.get_exhibit_pages( + exhibit_pages = one_to_n_funcs.get_exhibit_pages( text_dict, filename ) # All pages with exhibit headers text_dict = table_funcs.align_and_format_tables(text_dict, filename) diff --git a/fieldExtraction/src/preprocessing_funcs.py b/fieldExtraction/src/preprocessing_funcs.py index cb47f33..0f43adc 100644 --- a/fieldExtraction/src/preprocessing_funcs.py +++ b/fieldExtraction/src/preprocessing_funcs.py @@ -1,16 +1,15 @@ -from itertools import groupby, count, chain -import re +import re from collections import defaultdict -import utils -import valid -from valid import DERIVED_INDICATOR_FIELDS -import keywords -from collections import defaultdict -import ac_smart_chunking -import ac_funcs -import prompts +from itertools import chain, count, groupby + import claude_funcs import config +import keywords +import prompts +import smart_chunking_funcs +import string_funcs +import valid +from valid import DERIVED_INDICATOR_FIELDS def clean_newlines(contract_text: str) -> str: @@ -66,7 +65,7 @@ def clean_law_symbols(contract_text): def chunk_consecutive_og(text_dict, exhibit_pages): # If needed - extend page 1 to page 1 and 2, then cut chunking off after (edge case: compensation terms not found on first page of exhibit) - reimbursement_pages = [page_num for page_num in text_dict.keys() if utils.contains_reimbursement(text_dict, page_num)] + reimbursement_pages = [page_num for page_num in text_dict.keys() if string_funcs.contains_reimbursement(text_dict, page_num)] page_dict = {} current_exhibit = None @@ -106,7 +105,7 @@ def chunk_consecutive(text_dict, exhibit_pages): reimbursement_pages = { page_num for page_num in text_dict.keys() - if utils.contains_reimbursement(text_dict, page_num) + if string_funcs.contains_reimbursement(text_dict, page_num) } page_dict = {} @@ -203,34 +202,34 @@ def smart_chunk_ac( for field_group, key_dict in keyword_mappings.items(): if key_dict["methodology"] == "or": - page_list = ac_smart_chunking.chunk_or( + page_list = smart_chunking_funcs.chunk_or( text_dict, key_dict["keywords"], key_dict["case_sensitive"] ) - keyword_page_dict = ac_smart_chunking.keyword_search( + keyword_page_dict = smart_chunking_funcs.keyword_search( text_dict, key_dict["keywords"], key_dict["case_sensitive"] ) elif key_dict["methodology"] == "hierarchy": - page_list = ac_smart_chunking.chunk_hierarchical( + page_list = smart_chunking_funcs.chunk_hierarchical( text_dict, key_dict["keywords"], key_dict["case_sensitive"] ) - keyword_page_dict = ac_smart_chunking.keyword_search( + keyword_page_dict = smart_chunking_funcs.keyword_search( text_dict, key_dict["keywords"], key_dict["case_sensitive"] ) elif key_dict["methodology"] == "and": - page_list = ac_smart_chunking.chunk_and( + page_list = smart_chunking_funcs.chunk_and( text_dict, key_dict["keywords"], key_dict["case_sensitive"] ) - keyword_page_dict = ac_smart_chunking.keyword_search( + keyword_page_dict = smart_chunking_funcs.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( + page_list = smart_chunking_funcs.chunk_regex(text_dict, key_dict["regex"]) + keyword_page_dict = smart_chunking_funcs.keyword_search( text_dict, key_dict["keywords"], key_dict["case_sensitive"], True ) elif key_dict["methodology"] == "include_exclude": # Chunk pages based on included and excluded keywords - page_list = ac_smart_chunking.chunk_with_include_exclude_keywords( + page_list = smart_chunking_funcs.chunk_with_include_exclude_keywords( text_dict, key_dict["included_keywords"], key_dict["excluded_keywords"] ) @@ -242,7 +241,7 @@ def smart_chunk_ac( # Perform keyword search on the text dictionary - this returns a dictionary keyed by keyword and values are list of pages containing that keyword # This is not currently used but is available for debugging purposes (or future use cases) - keyword_page_dict = ac_smart_chunking.keyword_search( + keyword_page_dict = smart_chunking_funcs.keyword_search( text_dict, keywords, key_dict["case_sensitive"] ) else: diff --git a/fieldExtraction/src/prompts.py b/fieldExtraction/src/prompts.py index 9d479ce..9d67b34 100644 --- a/fieldExtraction/src/prompts.py +++ b/fieldExtraction/src/prompts.py @@ -1,4 +1,3 @@ -import config import valid ##################################################################################### diff --git a/fieldExtraction/src/qa_qc_helpers.py b/fieldExtraction/src/qa_qc_helpers.py index 3bdd6b3..437eef1 100644 --- a/fieldExtraction/src/qa_qc_helpers.py +++ b/fieldExtraction/src/qa_qc_helpers.py @@ -1,16 +1,18 @@ import os -import pandas as pd -import boto3 -from botocore.exceptions import ClientError import re -from datetime import datetime -from valid import STATE_MAP, AC_IND_LANG_COLS, B_IND_LANG_COLS -import utils -import preprocessing_funcs -import config +from datetime import datetime + +import pandas as pd from openpyxl import Workbook from openpyxl.utils.dataframe import dataframe_to_rows + +import config +import io_utils +import preprocessing_funcs +import string_funcs from hotfix_helper_funcs import keywords_to_detect_invalid_state_name +from valid import AC_IND_LANG_COLS, B_IND_LANG_COLS, STATE_MAP + def detect_date_anomalies(series, field_name, is_merged=False): total_count = len(series) @@ -162,7 +164,7 @@ def blanks_check(df: pd.DataFrame, threshold=config.BLANKS_THRESHOLD) -> pd.Data highly_blank_cols = [] for col in df.columns: - if df[col].apply(lambda x: utils.is_empty(x)).sum() / len(df[col]) > threshold: + if df[col].apply(lambda x: string_funcs.is_empty(x)).sum() / len(df[col]) > threshold: highly_blank_cols.append(col) return '\n'.join(highly_blank_cols) @@ -175,7 +177,7 @@ def self_talk_check(df: pd.DataFrame): for col in lang_cols: - df['selftalk'] = df[col].apply(lambda x: any([(kw in x) for kw in keywords_to_detect_invalid_state_name]) if not utils.is_empty(x) else False) + df['selftalk'] = df[col].apply(lambda x: any([(kw in x) for kw in keywords_to_detect_invalid_state_name]) if not string_funcs.is_empty(x) else False) if df['selftalk'].any(): @@ -238,18 +240,18 @@ def perform_qc_qa_tests_both(ac_df, b_df, merged_df, input_dict, processed_files "# of fields for B run" : len(b_df.columns), "Highly Blank Fields in merged DataFrame" : blanks_check(merged_df), "Potential 'Self-Talk' columns in merged DataFrame" : self_talk_check(merged_df), - "# of incorrectly formatted Provider Group TINs" : ac_df["IRS #"].apply(lambda x: not tin_check(x)[1] if not utils.is_empty(x) else False).sum(), - "# of incorrectly formatted Provider Group NPIs" : ac_df["NPI (10-digits)"].apply(lambda x: not npi_check(x)[1] if not utils.is_empty(x) else False).sum(), - "# of incorrectly formatted Provider Group Signatory TINs": ac_df["PROV_GROUP_TIN_SIGNATORY"].apply(lambda x: not tin_check(x)[1] if not utils.is_empty(x) else False).sum(), - "# of incorrectly formatted Provider TINs (other)" : ac_df["PROV_TIN_OTHER"].apply(lambda x: not tin_check(x)[1] if not utils.is_empty(x) else False).sum(), - "# of incorrectly formatted Provider NPIs (other)": ac_df["PROV_NPI_OTHER"].apply(lambda x: not npi_check(x)[1] if not utils.is_empty(x) else False).sum(), - "# of incorrect Health Plan States" : ac_df['Health Plan State'].apply(lambda x: not state_check(x)[1] if not utils.is_empty(x) else False).sum(), - "# of rows with Pages greater than Page Num" : merged_df.apply(lambda row: pages_pagenum_check(row['Pages'], row['Page_Num']) if (not utils.is_empty(row['Pages']) and utils.is_empty(row['Page_Num'])) else False, axis=1).sum(), - "# of Payer Names found in Notice to Provider Name" : ac_df.apply(lambda row: row['PAYER NAME'] in row['Notice to Provider Name'] if not utils.is_empty(row['Notice to Provider Name']) and not utils.is_empty(row['PAYER NAME']) else False, axis = 1).sum(), + "# of incorrectly formatted Provider Group TINs" : ac_df["IRS #"].apply(lambda x: not tin_check(x)[1] if not string_funcs.is_empty(x) else False).sum(), + "# of incorrectly formatted Provider Group NPIs" : ac_df["NPI (10-digits)"].apply(lambda x: not npi_check(x)[1] if not string_funcs.is_empty(x) else False).sum(), + "# of incorrectly formatted Provider Group Signatory TINs": ac_df["PROV_GROUP_TIN_SIGNATORY"].apply(lambda x: not tin_check(x)[1] if not string_funcs.is_empty(x) else False).sum(), + "# of incorrectly formatted Provider TINs (other)" : ac_df["PROV_TIN_OTHER"].apply(lambda x: not tin_check(x)[1] if not string_funcs.is_empty(x) else False).sum(), + "# of incorrectly formatted Provider NPIs (other)": ac_df["PROV_NPI_OTHER"].apply(lambda x: not npi_check(x)[1] if not string_funcs.is_empty(x) else False).sum(), + "# of incorrect Health Plan States" : ac_df['Health Plan State'].apply(lambda x: not state_check(x)[1] if not string_funcs.is_empty(x) else False).sum(), + "# of rows with Pages greater than Page Num" : merged_df.apply(lambda row: pages_pagenum_check(row['Pages'], row['Page_Num']) if (not string_funcs.is_empty(row['Pages']) and string_funcs.is_empty(row['Page_Num'])) else False, axis=1).sum(), + "# of Payer Names found in Notice to Provider Name" : ac_df.apply(lambda row: row['PAYER NAME'] in row['Notice to Provider Name'] if not string_funcs.is_empty(row['Notice to Provider Name']) and not string_funcs.is_empty(row['PAYER NAME']) else False, axis = 1).sum(), "# of invalid Y/N values for B run" : b_df[yn_cols_b].apply(lambda x: yn_check(x)[1], axis = 1).sum().sum(), "# of invalid Y/N values for AC run" : ac_df[yn_cols_ac].apply(lambda x: yn_check(x)[1], axis = 1).sum().sum(), - "# of invalid Termination Dates" : ac_df['Termination Date'].apply(lambda x: not date_format_check(x) if not utils.is_empty(x) else x).sum(), - "# of invalid Conract Effective Dates" : ac_df['Contract Effective Date'].apply(lambda x: not date_format_check(x) if not utils.is_empty(x) else x).sum() + "# of invalid Termination Dates" : ac_df['Termination Date'].apply(lambda x: not date_format_check(x) if not string_funcs.is_empty(x) else x).sum(), + "# of invalid Conract Effective Dates" : ac_df['Contract Effective Date'].apply(lambda x: not date_format_check(x) if not string_funcs.is_empty(x) else x).sum() } ac_stats = {} @@ -317,16 +319,16 @@ def perform_qc_qa_tests_ac(ac_df, input_dict, processed_files): "# of fields for AC run" : len(ac_df.columns), "Highly Blank Fields in AC DataFrame" : blanks_check(ac_df), "Potential 'Self-Talk' columns in merged DataFrame" : self_talk_check(ac_df), - "# of incorrectly formatted Provider Group TINs" : ac_df["IRS #"].apply(lambda x: not tin_check(x)[1] if not utils.is_empty(x) else False).sum(), - "# of incorrectly formatted Provider Group NPIs" : ac_df["NPI (10-digits)"].apply(lambda x: not npi_check(x)[1] if not utils.is_empty(x) else False).sum(), - "# of incorrectly formatted Provider Group Signatory TINs": ac_df["PROV_GROUP_TIN_SIGNATORY"].apply(lambda x: not tin_check(x)[1] if not utils.is_empty(x) else False).sum(), - "# of incorrectly formatted Provider TINs (other)" : ac_df["PROV_TIN_OTHER"].apply(lambda x: not tin_check(x)[1] if not utils.is_empty(x) else False).sum(), - "# of incorrectly formatted Provider NPIs (other)": ac_df["PROV_NPI_OTHER"].apply(lambda x: not npi_check(x)[1] if not utils.is_empty(x) else False).sum(), - "# of incorrect Health Plan States" : ac_df['Health Plan State'].apply(lambda x: not state_check(x)[1] if not utils.is_empty(x) else False).sum(), - "# of Payer Names found in Notice to Provider Name" : ac_df.apply(lambda row: row['PAYER NAME'] in row['Notice to Provider Name'] if not utils.is_empty(row['Notice to Provider Name']) else False, axis = 1).sum(), + "# of incorrectly formatted Provider Group TINs" : ac_df["IRS #"].apply(lambda x: not tin_check(x)[1] if not string_funcs.is_empty(x) else False).sum(), + "# of incorrectly formatted Provider Group NPIs" : ac_df["NPI (10-digits)"].apply(lambda x: not npi_check(x)[1] if not string_funcs.is_empty(x) else False).sum(), + "# of incorrectly formatted Provider Group Signatory TINs": ac_df["PROV_GROUP_TIN_SIGNATORY"].apply(lambda x: not tin_check(x)[1] if not string_funcs.is_empty(x) else False).sum(), + "# of incorrectly formatted Provider TINs (other)" : ac_df["PROV_TIN_OTHER"].apply(lambda x: not tin_check(x)[1] if not string_funcs.is_empty(x) else False).sum(), + "# of incorrectly formatted Provider NPIs (other)": ac_df["PROV_NPI_OTHER"].apply(lambda x: not npi_check(x)[1] if not string_funcs.is_empty(x) else False).sum(), + "# of incorrect Health Plan States" : ac_df['Health Plan State'].apply(lambda x: not state_check(x)[1] if not string_funcs.is_empty(x) else False).sum(), + "# of Payer Names found in Notice to Provider Name" : ac_df.apply(lambda row: row['PAYER NAME'] in row['Notice to Provider Name'] if not string_funcs.is_empty(row['Notice to Provider Name']) else False, axis = 1).sum(), "# of invalid Y/N values for AC run" : ac_df[yn_cols_ac].apply(lambda x: yn_check(x)[1], axis = 1).sum().sum(), - "# of invalid Termination Dates" : ac_df['Termination Date'].apply(lambda x: not date_format_check(x) if not utils.is_empty(x) else x).sum(), - "# of invalid Conract Effective Dates" : ac_df['Contract Effective Date'].apply(lambda x: not date_format_check(x) if not utils.is_empty(x) else x).sum() + "# of invalid Termination Dates" : ac_df['Termination Date'].apply(lambda x: not date_format_check(x) if not string_funcs.is_empty(x) else x).sum(), + "# of invalid Conract Effective Dates" : ac_df['Contract Effective Date'].apply(lambda x: not date_format_check(x) if not string_funcs.is_empty(x) else x).sum() } ac_stats = {} @@ -402,7 +404,7 @@ def get_table_stats(text_dict): table_count, num_table_pages, rate_count = 0, 0, 0 table_pages, rate_pages, rates_more_than_10 = [], [], [] for page_num, page_text in text_dict.items(): - if utils.contains_reimbursement(page_text, page_num): + if string_funcs.contains_reimbursement(page_text, page_num): rate_pages.append(page_num) rates_on_page = sum(1 for char in page_text if char in {"$", "%"}) rate_count += rates_on_page @@ -436,7 +438,7 @@ def table_stats_qaqc(input_dict, ac_df, b_df, file_name_column) -> list[dict]: table_stats = [] if input_dict: for filename, content in input_dict.items(): - filename_without_ext = utils.remove_txt_extension(filename).lower() + filename_without_ext = io_utils.remove_txt_extension(filename).lower() text_dict = preprocessing_funcs.split_text(content) stats = get_table_stats(text_dict) table_stats.append( diff --git a/fieldExtraction/src/ac_regex_chunking.py b/fieldExtraction/src/regex_funcs.py similarity index 89% rename from fieldExtraction/src/ac_regex_chunking.py rename to fieldExtraction/src/regex_funcs.py index bc2dfcc..f75b093 100644 --- a/fieldExtraction/src/ac_regex_chunking.py +++ b/fieldExtraction/src/regex_funcs.py @@ -1,10 +1,10 @@ import re -from utils import find_regex_matches, add_hyphen_if_needed + +import postprocessing_funcs + ################################################################################################################################################# # NPI - - def signature_page_npi(text_dict, page_list, matches): """In case of Multiple 10-digit regex matches, this function will check if anyog those matches are on the signature page. We check for the word 'Signature' or 'SIGNATURE' to determine the signature page @@ -51,6 +51,31 @@ def signature_page_npi(text_dict, page_list, matches): return npi, npi_other +def find_regex_matches( # regex_funcs.py + 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) # make it append + page_list.append(page) + + return matches, page_list + + def npi_pattern_match(text_dict: dict[str, str]) -> dict[str, str]: """Returns dictionery with Provider and other TIN(s) based on regex matches and presence on Signatory page @@ -119,7 +144,7 @@ def check_tin_in_filename(filename): for pattern in patterns: tin_from_title = re.findall(pattern, filename) if len(tin_from_title) == 1: - tin_from_title = [add_hyphen_if_needed(s) for s in tin_from_title] + tin_from_title = [postprocessing_funcs.add_hyphen_if_needed(s) for s in tin_from_title] return tin_from_title[0] return "empty" @@ -147,8 +172,8 @@ def signature_page_irs(text_dict, page_list, matches, filename): tin_on_signature_page = list(set(tin_on_signature_page)) # adds hyphen if needed - matches = [add_hyphen_if_needed(s) for s in matches] - tin_on_signature_page = [add_hyphen_if_needed(s) for s in tin_on_signature_page] + matches = [postprocessing_funcs.add_hyphen_if_needed(s) for s in matches] + tin_on_signature_page = [postprocessing_funcs.add_hyphen_if_needed(s) for s in tin_on_signature_page] # case1 - If 1 TIN found on signature page, return it, put the remaining ones in "TIN Others" have signature_tin = N/A if len(tin_on_signature_page) == 1: @@ -215,7 +240,7 @@ def irs_pattern_match(filename: str, text_dict: dict[str, str]): matches = list(set(matches)) if matches: if len(matches) == 1: - matches = [add_hyphen_if_needed(s) for s in matches] + matches = [postprocessing_funcs.add_hyphen_if_needed(s) for s in matches] TIN_dict["PROV_GROUP_TIN"] = matches[0] TIN_dict["PROV_TIN_OTHER"] = "N/A" TIN_dict["PROV_GROUP_TIN_SIGNATORY"] = "N/A" @@ -264,3 +289,4 @@ def irs_hotfix( ): irs_answers = irs_pattern_match(filename, top_sheet_dict) return irs_answers + diff --git a/fieldExtraction/src/ac_smart_chunking.py b/fieldExtraction/src/smart_chunking_funcs.py similarity index 56% rename from fieldExtraction/src/ac_smart_chunking.py rename to fieldExtraction/src/smart_chunking_funcs.py index 3ae2dd2..58534c5 100644 --- a/fieldExtraction/src/ac_smart_chunking.py +++ b/fieldExtraction/src/smart_chunking_funcs.py @@ -1,5 +1,13 @@ import re +import claude_funcs +import config +import keywords +import postprocessing_funcs +import prompts +import string_funcs +from enums.delimiters import Delimiter + """ Date pattern regex: Matches for "day of", @@ -21,6 +29,237 @@ Case insensitive DATE_PATTERN = r"(?i)\bday of\b|\b(?:jan(?:uary)?|feb(?:ruary)?|mar(?:ch)?|apr(?:il)?|may|jun(?:e)?|jul(?:y)?|aug(?:ust)?|sep(?:tember)?|oct(?:ober)?|nov(?:ember)?|dec(?:ember)?)\b|\b(?:\d{1,2}[\/\.-]\d{1,2}[\/\.-](?:\d{2}|\d{4})|\d{4}[\/\.-]\d{1,2}[\/\.-]\d{1,2}|\d{8}|\d{2}\d{2}|\d{4}[\/\.-]\d{1,2}[\/\.-]\d{1,2}|\d{4}[\.]\d{1,2}[\.]\d{1,2})\b" + +def run_smart_chunk_prompts( + ac_answers_dict: dict[str, str], + ac_chunks: dict, + contract_text: str, + filename: str, + text_dict: dict[str, str], +): + + ## 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() + ] + + # list to append the fields with no matching keywords + no_keyword_matches = [] + + for field_group in field_groups: + keyword_dict = keywords.GROUPED_KEYWORD_MAPPINGS[field_group].get( + "keywords", [] + ) + fields = keywords.GROUPED_KEYWORD_MAPPINGS[field_group].get("fields", []) + fields = [field for field in fields] + fields.sort() + questions = {} + for field in fields: + questions[field] = prompts.AC_DICT.get(field, "") + + # Use chunk if available and different from full context, otherwise use full context + ############# Important Note ############# + # the above logic is changed + # New Logic - chunk if available and different from full context, otherwise add the field to full context list + if ( + field_group in ac_chunks + and ac_chunks[field_group].strip() + and ac_chunks[field_group] != contract_text + ): + + context = ac_chunks[field_group] + context = context[0 : min(100000, len(context)) - 1] + context_type = "Smart Chunking" + + if len(fields) == 1: + question = questions[ + fields[0] + ] # Take the one question for the single field included + if fields[0] == "CONTRACT_EFFECTIVE_DT": + prompt = prompts.get_effective_date_prompt(context) + else: + 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") + + try: + if field_group == "AGREEMENT_NAME": + + # context, prompt and answer based on first 5 pages + context_agreement = "\n".join( + [ + text_dict[str(page_num)] + for page_num in text_dict.keys() + if str(page_num) in ["1", "2", "3", "4", "5"] + ] + ) + prompt_agreement = prompts.AC_SINGLE_FIELD_TEMPLATE( + context_agreement, question + ) + field_group_answer_raw = claude_funcs.invoke_claude( + prompt_agreement, + config.MODEL_ID_CLAUDE35_SONNET, + filename, + 8192, + ) + + # if answer is N/A use the smart-chunked context + if field_group_answer_raw == "N/A": + field_group_answer_raw = claude_funcs.invoke_claude( + prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, 8192 + ) + else: + field_group_answer_raw = claude_funcs.invoke_claude( + prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, 8192 + ) + + if len(fields) == 1: + if fields[0] == "CONTRACT_EFFECTIVE_DT": + # print(f'Effective Date Answer obtained from LLM: {field_group_answer_raw}') + field_group_answer_raw = string_funcs.extract_text_from_delimiters( + field_group_answer_raw, Delimiter.PIPE + ) + # print(f'Parsed Effective Date Answer (extract_text_from_delimiters): {field_group_answer_raw}') + try: + field_group_answer_raw = postprocessing_funcs.convert_to_us_date_format( + field_group_answer_raw + ) # Fix date formatting + # print(f'Effective Date Answer (convert_to_us_date_format): {field_group_answer_raw}') + except postprocessing_funcs.InvalidDateException as e: + field_group_answer_raw = "N/A" + # print(f'Got Invalid Date Exception, using contract effective date as N/A for contract {filename}') + + field_group_answer_dict = {fields[0]: field_group_answer_raw} + + elif len(fields) > 1: + field_group_answer_raw = field_group_answer_raw.replace( + "_DATE", "_DT" + ) + field_group_answer_dict = string_funcs.json_parsing_search( + field_group_answer_raw, fields + ) + + # 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 + + except Exception as e: + print(f"Error processing field {field}: {type(e)}, {str(e)}") + ac_answers_dict[field] = f"Error: {str(e)}" + + else: # If no chunk available + if ( + field_group == "CONTRACT_EFFECTIVE_DT" + ): # Special case for effective date + ac_answers_dict["CONTRACT_EFFECTIVE_DT"] = "N/A" + # print('Got no chunk for effective date, so setting it to N/A', ac_answers_dict) + else: # Add to no_keyword_matches list to run using full context + no_keyword_matches.extend(fields) + + return ac_answers_dict, no_keyword_matches + + +def get_full_context_fields( + ac_answers_dict: dict[str, str], no_keyword_matches: list +) -> list: + + 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) + + # Instantiate full context fields from those that weren't smart chunked + full_context_fields = [ + field for field in prompts.AC_DICT if field not in smartly_chunked + ] + + # Add fields with no keyword matches to full context list + if len(no_keyword_matches) > 0: + full_context_fields.extend(no_keyword_matches) + + # Filter out fields that already have answers + # e.g., some have already been filled out by regex chunking + full_context_fields = [ + field for field in full_context_fields if field not in ac_answers_dict.keys() + ] + return full_context_fields + + +def run_full_context_prompts( + ac_answers_dict: dict[str, str], + full_context_fields: list, + contract_text: str, + filename: str, +) -> dict[str, str]: + 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( + context=contract_text[0 : min(100000, len(contract_text) - 1)], + questions=full_context_questions, + ) + + try: + full_context_answers = claude_funcs.get_ac_answer( + full_context_prompt, + filename, + fields=list(full_context_questions.keys()), + ) + + ac_answers_dict.update(full_context_answers) + + 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)}" + + # Ensure all AC fields have a key, populated with N/A if not from prompt + for key in prompts.AC_DICT.keys(): + if key not in ac_answers_dict: + ac_answers_dict[key] = "N/A" + + return ac_answers_dict + + +def run_ac_conditional_prompts( + ac_answers_dict: dict[str, str], contract_text: str, filename: str +) -> dict[str, str]: + + # Non-Renewal - Days + if ac_answers_dict["NON_RENEWAL_LANGUAGE"] != "N/A": + nrd_prompt = prompts.AC_SINGLE_FIELD_TEMPLATE( + 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_answers_dict["NON_RENEWAL_DAYS"] = nrd_answer + else: + ac_answers_dict["NON_RENEWAL_DAYS"] = "N/A" + + # Clean Up Contract Effective Date + if ( + string_funcs.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] + ) + date_answer = claude_funcs.invoke_claude( + date_prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, 124 + ) + ac_answers_dict["CONTRACT_EFFECTIVE_DT"] = date_answer + + return ac_answers_dict + + def chunk_hierarchical( text_dict: dict[str, str], keywords: list[str], case_sensitive: bool = False ) -> list[str]: @@ -219,16 +458,17 @@ def keyword_search( return page_dict_sorted + def chunk_with_include_exclude_keywords( - text_dict: dict[str, str], + text_dict: dict[str, str], include_keywords: list[str], exclude_keywords: list[str], case_sensitive: bool = False, apply_date_filtering: bool = True ) -> list[str]: - + """Look for pages that include any of the include_keywords and exclude any pages that contain any of the exclude_keywords. - + Args: text_dict (dict[str, str]): A dictionary keyed by string page number and valued by string page contents @@ -247,18 +487,18 @@ def chunk_with_include_exclude_keywords( exclude_keywords = [keyword.lower() for keyword in exclude_keywords] text_dict = {key: value.lower() for key, value in text_dict.items()} - if apply_date_filtering: + if apply_date_filtering: pages_containing_date = set() for page_num, page_content in text_dict.items(): dates_list = re.findall(DATE_PATTERN, page_content) - + if len(dates_list) > 0: - pages_containing_date.add(page_num) + pages_containing_date.add(page_num) pages_containing_date = sorted(pages_containing_date,key=int) # debug lines for inclusion and exclusion keywords found on each page - + # print(f"Pages containing date: {pages_containing_date}") # for page_num, page_content in text_dict.items(): @@ -270,13 +510,13 @@ def chunk_with_include_exclude_keywords( # for keyword in include_keywords: # if keyword in page_content: # inclusion_keywords.add(keyword) - + # for keyword in exclude_keywords: # if keyword in page_content: # exclusion_keywords.add(keyword) - + # print(f"Page #: {page_num} contains inclusion keywords: {inclusion_keywords} and contain exclusion keywords: {exclusion_keywords}") - + page_list = [] for page in text_dict.keys(): @@ -299,7 +539,7 @@ def chunk_with_include_exclude_keywords( page_list_sorted_str = list(map(str, page_list_sorted)) # print(", ".join([page for page in page_list_sorted_str])) - + # print(f"Obtained total pages: {len(page_list_sorted_str)}") return page_list_sorted_str \ No newline at end of file diff --git a/fieldExtraction/src/ac_funcs.py b/fieldExtraction/src/string_funcs.py similarity index 51% rename from fieldExtraction/src/ac_funcs.py rename to fieldExtraction/src/string_funcs.py index fd37901..10c19b8 100644 --- a/fieldExtraction/src/ac_funcs.py +++ b/fieldExtraction/src/string_funcs.py @@ -1,38 +1,18 @@ import json -import boto3 -import pandas as pd -import numpy as np -from datetime import datetime -import os -import anthropic import re -import logging -from urllib.parse import unquote_plus -from botocore.exceptions import ClientError -from botocore.config import Config -import traceback -import time -import sys import warnings -from enums.delimiters import Delimiter + +import numpy as np +import pandas as pd import claude_funcs -import dict_operations import config -import utils import prompts -from regex_patterns import PIPE_PATTERN, BACKTICK_PATTERN, TRIPLE_BACKTICK_PATTERN +from enums.delimiters import Delimiter +from regex_patterns import (BACKTICK_PATTERN, PIPE_PATTERN, + TRIPLE_BACKTICK_PATTERN) -def get_ac_answer(prompt, filename, fields): - response_text = claude_funcs.invoke_claude( - prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, 8192 - ) - response_text = response_text.replace("_DATE", "_DT") - # response_dict = dict_operations.secondary_string_to_dict(response_text, filename) - response_dict = json_parsing_search(response_text, fields) - return response_dict - def extract_text_from_delimiters( raw_output: str, delimiter: Delimiter, match_index: int = -1 ) -> str: @@ -154,3 +134,104 @@ def json_parsing_search(response_text, field_list): ) return dict(zip(field_l, answer_l)) + + +def secondary_string_to_dict(dict_string, filename): + """ + Converts a string representation of a dictionary into an actual dictionary object, handling potential formatting issues. + + This function cleans the string by removing specific unwanted characters and markers that may interfere with JSON parsing. + It then locates the substring that correctly forms a dictionary format, attempts to parse it as JSON, and handles common + parsing errors by replacing problematic single quotes with double quotes before re-parsing. + + Parameters: + dict_string (str): The string containing the dictionary-like content, potentially surrounded by extra text or characters. + + Returns: + dict: The dictionary obtained from parsing the cleaned and corrected string. + """ + start_index = dict_string.find("{") + end_index = dict_string.rfind("}") + 1 + dict_substring = dict_string[start_index:end_index] + try: + result_dict = json.loads(dict_substring) + except: + try: + dict_substring = dict_substring.replace("'", '"') + result_dict = json.loads(dict_substring) + except: + try: + prompt = prompts.FIX_JSON(dict_substring) + dict_substring = claude_funcs.invoke_claude( + prompt, config.MODEL_ID_CLAUDE3_HAIKU, filename + ) + result_dict = json.loads(dict_substring) + except Exception as e: + print(e) + result_dict = {} + return result_dict + + +def primary_string_to_dict(string_dict, filename): + """ + Converts a dictionary of strings, where each string represents multiple dictionary entries, into a list of dictionaries, + augmenting each with metadata such as page number and filename. + + This function iterates over each page number's string data, extracting and converting string representations of + dictionaries into actual dictionary objects. It handles and cleans the string format to properly parse it into dictionaries. + All dictionaries are then augmented with their respective page number and the filename before being compiled into a list. + + Parameters: + string_dict (dict): A dictionary where keys are page numbers and values are strings containing multiple dictionary entries. + filename (str): The filename associated with these entries, used to tag each resulting dictionary. + + Returns: + list of dict: A list of dictionaries, each representing data extracted and converted from the input string, tagged with + their page number and filename. + """ + data = [] + pattern = r"\{.*?\}" + for page_num in string_dict.keys(): + primary_list = string_dict[page_num] + dicts = primary_list.split("[")[1] # Strip front + dicts = dicts.split("]")[0] # Strip back + dicts = dicts.replace("\n", "") # Remove new lines + dicts = dicts.replace("<<<", "") + dicts = dicts.replace(">>>", "") + dicts = re.sub(r"(? 0 else 0 + b_progress = min((completed_b / total_b * 100), 100.0) if total_b > 0 else 0 + return ac_progress, b_progress \ No newline at end of file diff --git a/fieldExtraction/src/valid.py b/fieldExtraction/src/valid.py index 563456c..1038478 100644 --- a/fieldExtraction/src/valid.py +++ b/fieldExtraction/src/valid.py @@ -1,6 +1,6 @@ -import config import pandas as pd +import config VALID_LOBS = [ "MEDICARE", diff --git a/fieldExtraction/tests/ac_smart_chunking_test.py b/fieldExtraction/tests/ac_smart_chunking_test.py index c3bb08b..56a35ee 100644 --- a/fieldExtraction/tests/ac_smart_chunking_test.py +++ b/fieldExtraction/tests/ac_smart_chunking_test.py @@ -1,7 +1,7 @@ import pytest import re -from ac_smart_chunking import DATE_PATTERN +from smart_chunking_funcs import DATE_PATTERN legal_traditional_dates = [ "This 12th day of December, 2024", diff --git a/fieldExtraction/tests/calculate_progress_stats_test.py b/fieldExtraction/tests/calculate_progress_stats_test.py index 85b30f2..eb25b89 100644 --- a/fieldExtraction/tests/calculate_progress_stats_test.py +++ b/fieldExtraction/tests/calculate_progress_stats_test.py @@ -1,7 +1,7 @@ import os import sys import unittest -from utils import calculate_progress_stats +from tracking import calculate_progress_stats class TestCalculateProgressStats(unittest.TestCase): def test_normal_case(self): diff --git a/fieldExtraction/tests/utils_test.py b/fieldExtraction/tests/postprocessing_funcs_test.py similarity index 95% rename from fieldExtraction/tests/utils_test.py rename to fieldExtraction/tests/postprocessing_funcs_test.py index 28ffd3d..942aa3e 100644 --- a/fieldExtraction/tests/utils_test.py +++ b/fieldExtraction/tests/postprocessing_funcs_test.py @@ -1,5 +1,5 @@ import pytest -from utils import convert_to_us_date_format, InvalidDateException, is_empty +from postprocessing_funcs import convert_to_us_date_format, InvalidDateException class TestConvertToUSDateFormat: diff --git a/fieldExtraction/tests/ac_funcs_test.py b/fieldExtraction/tests/string_funcs_test.py similarity index 96% rename from fieldExtraction/tests/ac_funcs_test.py rename to fieldExtraction/tests/string_funcs_test.py index 122e80d..3cf83eb 100644 --- a/fieldExtraction/tests/ac_funcs_test.py +++ b/fieldExtraction/tests/string_funcs_test.py @@ -1,6 +1,6 @@ import pytest from enums.delimiters import Delimiter -from ac_funcs import extract_text_from_delimiters +from string_funcs import extract_text_from_delimiters class TestExtractTextFromDelimiters: