diff --git a/fieldExtraction/src/constants/investment_columns.py b/fieldExtraction/src/constants/investment_columns.py index a984ee5..e7dd39d 100644 --- a/fieldExtraction/src/constants/investment_columns.py +++ b/fieldExtraction/src/constants/investment_columns.py @@ -15,6 +15,7 @@ COLUMN_ORDER = [ "PROV_OTHER_TIN", "PROV_OTHER_NPI", "PROV_OTHER_NAME_FULL", + "PROV_INFO_JSON", "AARETE_DERIVED_EFFECTIVE_DT", "TERMINATION_DT", "AARETE_DERIVED_TERMINATION_DT", diff --git a/fieldExtraction/src/investment/file_processing.py b/fieldExtraction/src/investment/file_processing.py index 0de31fa..6ec5c75 100644 --- a/fieldExtraction/src/investment/file_processing.py +++ b/fieldExtraction/src/investment/file_processing.py @@ -1,23 +1,26 @@ import csv +import json import os import pandas as pd - -from src.utils.string_utils import datetime_str +import src.investment.aarete_derived as aarete_derived +import src.investment.dynamic_funcs as dynamic_funcs import src.investment.one_to_n_funcs as one_to_n_funcs import src.investment.one_to_one_funcs as one_to_one_funcs import src.investment.postprocess as postprocess import src.investment.preprocess as preprocess -import src.investment.aarete_derived as aarete_derived import src.investment.smart_chunking_funcs as smart_chunking_funcs -from src.prompts.investment_prompts import Field, FieldSet -from src import config -import src.utils.string_utils as string_utils import src.utils.io_utils as io_utils +import src.utils.string_utils as string_utils +from src import config from src.config import FIELD_JSON_PATH -from src.investment.one_to_n_funcs import reimbursement_level, combine_one_to_n_answers -from src.investment.tin_npi_funcs import reimbursement_tin_npi, merge_provider_info -import src.investment.dynamic_funcs as dynamic_funcs +from src.investment.one_to_n_funcs import (combine_one_to_n_answers, + reimbursement_level) +from src.investment.tin_npi_funcs import (merge_provider_info, + reimbursement_tin_npi) +from src.prompts.investment_prompts import Field, FieldSet +from src.utils.string_utils import datetime_str + ################## MERGE ################## def merge_one_to_one_into_one_to_n(one_to_n_results, one_to_one_results): @@ -95,6 +98,7 @@ def run_one_to_one_prompts(filename, contract_text, text_dict, top_sheet_dict, d ################## RUN REGEX ################## regex_answers_dict = one_to_one_funcs.run_provider_info_fields(contract_text, text_dict, filename) one_to_one_results = {} + one_to_one_results['PROV_INFO_JSON'] = json.dumps(regex_answers_dict) one_to_one_results = merge_provider_info(one_to_one_results, regex_answers_dict) ################## RUN SMART CHUNKED PROMPTS ################## diff --git a/fieldExtraction/src/investment/one_to_one_funcs.py b/fieldExtraction/src/investment/one_to_one_funcs.py index ef1e5d3..14e24aa 100644 --- a/fieldExtraction/src/investment/one_to_one_funcs.py +++ b/fieldExtraction/src/investment/one_to_one_funcs.py @@ -1,106 +1,74 @@ -from src.prompts.investment_prompts import Field, FieldSet -from src.investment.smart_chunking_funcs import field_context, group_fields, parse_chunk, stitch_chunks import src.prompts.investment_prompts as prompts -import src.regex.regex_patterns as regex_patterns import src.utils.llm_utils as llm_utils import src.utils.string_utils as string_utils -from src.enums.delimiters import Delimiter from src import config -from src.investment.tin_npi_funcs import get_all_matches, chunk_on_matches, clean_provider_info -from src.prompts.investment_prompts import Field, FieldSet, TIN_NPI_TEMPLATE -from src.utils.llm_utils import invoke_claude +from src.enums.delimiters import Delimiter +from src.investment.smart_chunking_funcs import (field_context, group_fields, + parse_chunk, stitch_chunks) +from src.investment.tin_npi_funcs import (clean_provider_info, + deduplicate_providers, + extract_identifiers, + get_provider_info, + identify_group_provider) +from src.prompts.investment_prompts import Field, FieldSet MAX_CONTEXT_LENGTH = 100000 -def run_provider_info_fields(contract_text: str, text_dict: dict, filename: str): +def run_provider_info_fields(contract_text: str, text_dict: dict, filename: str) -> list[dict]: """ Extracts provider information from a contract text using regex-based identifier extraction, processes the relevant text chunks, and cleans the results. Args: contract_text (str): The full text of the contract to process. - text_dict (dict): A dictionary containing text data, typically organized by pages or sections. + text_dict (dict): A dictionary containing text data, organized by pages filename (str): The name of the file being processed, used for logging or reference. Returns: - dict: A cleaned and standardized dictionary containing provider information extracted - from the contract text. + list[dict]: A cleaned and standardized list containing provider information extracted + from the contract text. """ all_tins, all_npis = extract_identifiers(contract_text, filename) - # Create a chunk with all pages with any TIN or NPI - provider_chunk = chunk_on_matches(matches=all_tins + all_npis, text_dict=text_dict) + # If there are no identifiers, return early with default values + if not all_tins and not all_npis: + return [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "NO_IDENTIFIERS_FOUND", "IS_GROUP": False}] + + # Find pages with TINs or NPIs + relevant_pages = {} + for page_num, page_text in text_dict.items(): + if any(tin in page_text for tin in all_tins) or any(npi in page_text for npi in all_npis): + relevant_pages[page_num] = page_text - # Call LLM on the chunk to get provider information - provider_info = get_provider_info(provider_chunk, filename) + # Process each page separately and combine results + combined_provider_info = [] + for page_num, page_text in relevant_pages.items(): + page_provider_info = get_provider_info(page_text, filename) + combined_provider_info.extend(page_provider_info) # Clean and standardize results - cleaned_provider_info = clean_provider_info(provider_info) - return cleaned_provider_info - -def extract_identifiers(contract_text: str, filename: str) -> tuple: - """ - Extracts Taxpayer Identification Numbers (TINs) and National Provider Identifiers (NPIs) - from the given contract text using predefined patterns. - - Args: - contract_text (str): The text content of the contract to search for identifiers. - filename (str): The name of the file being processed, used for logging or debugging. - - Returns: - tuple: A tuple containing two lists: - - A list of all matched TINs found in the contract text. - - A list of all matched NPIs found in the contract text. - """ - - - # Get all matches for TIN and NPI - all_tins = get_all_matches(contract_text, regex_patterns.TIN_PATTERN, filename) - all_npis = get_all_matches(contract_text, regex_patterns.NPI_PATTERN, filename) - - return all_tins, all_npis - -def get_provider_info(chunk: str, filename: str) -> list: - """ - Extracts provider information from a given text chunk using a language model. - - Args: - chunk (str): The text chunk containing provider information to be extracted. - filename (str): The name of the file being processed, used for logging or tracking purposes. - - Returns: - list: A list of extracted provider information in JSON format, parsed from the language model's response. - - Raises: - ValueError: If the response from the language model cannot be parsed as valid JSON. - """ - provider_fields = FieldSet(file_path=config.FIELD_JSON_PATH, field_type="provider_info") - # Generate prompt - prompt = TIN_NPI_TEMPLATE(chunk, provider_fields.get_prompt_dict()) - # Get response from LLM - claude_answer_raw = invoke_claude(prompt, config.MODEL_ID_CLAUDE35_SONNET, filename) - - # Defensive programming for any json parsing errors - try: - providers = string_utils.universal_json_load(claude_answer_raw) - - # Ensure we have a list of providers (not a dict or other type) - if isinstance(providers, dict): - # Single provider returned as dict - wrap in list - providers = [providers] - elif not isinstance(providers, list): - # If not a list or dict, create a default list with error info - print(f"Warning: Unexpected format in provider info response: {type(providers)}") - providers = [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "PARSING_ERROR", "IS_GROUP": False}] - - return providers - except ValueError as e: # Handle JSON parsing error - print(f"Error parsing JSON response from LLM: {str(e)}") - print(f"Raw response: {claude_answer_raw}") - # Return a default value - return [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "PARSING_ERROR", "IS_GROUP": False}] + cleaned_provider_info = clean_provider_info(combined_provider_info) + # Deduplicate providers based on TIN and NPI + deduplicated_provider_info = deduplicate_providers(cleaned_provider_info) + if len(deduplicated_provider_info) > 1: + # print(f"Identifying group provider for {filename}...") + # This function modifies the provider list in place and marks one as IS_GROUP=True + group_provider = identify_group_provider(text_dict, deduplicated_provider_info, filename) + + # The returned group_provider is the one that was marked as IS_GROUP=True - but + # we don't need to use it, as the deduplicated_provider_info list is already modified + # and contains the IS_GROUP flag. + # If you want to print the group provider, uncomment the following lines: + # if group_provider: + # print(f"Group provider identified: {group_provider.get('NAME')} with TIN {group_provider.get('TIN')} and NPI {group_provider.get('NPI')}") + # else: + # print("No group provider identified.") + elif len(deduplicated_provider_info) == 1: + # If only one provider is found, mark it as the group provider + deduplicated_provider_info[0]["IS_GROUP"] = True + return deduplicated_provider_info def run_smart_chunked_fields(one_to_one_fields: FieldSet, contract_text: str, diff --git a/fieldExtraction/src/investment/tin_npi_funcs.py b/fieldExtraction/src/investment/tin_npi_funcs.py index ffc8227..65b1d9b 100644 --- a/fieldExtraction/src/investment/tin_npi_funcs.py +++ b/fieldExtraction/src/investment/tin_npi_funcs.py @@ -1,13 +1,18 @@ +import json import re +from typing import Optional + import src.config as config +import src.prompts.investment_prompts as investment_prompts +import src.regex.regex_patterns as regex_patterns +import src.regex.regex_utils as regex_utils import src.utils.llm_utils as llm_utils import src.utils.string_utils as string_utils -import src.regex.regex_utils as regex_utils -import src.prompts.investment_prompts as investment_prompts from src.enums.delimiters import Delimiter from src.prompts.investment_prompts import Field, FieldSet - + + def get_all_matches(text, pattern, filename): """ Extracts all unique matches of a given pattern from the provided text. @@ -192,3 +197,169 @@ def reimbursement_tin_npi(exhibit_chunk: str, reimbursement_level_fields: FieldS return answer_dict, reimbursement_level_fields +def identify_group_provider(text_dict: dict, providers: list[dict], filename: str) -> Optional[dict]: + """Identifies the main contracting group provider from a list of providers. + Focuses on first/last parts of the contract, where group identification is typically found + + Args: + text_dict (dict): A dictionary containing text data, organized by pages + providers (list[dict]): A list of deduplicated provider information dictionaries + filename (str): The name of the file being processed + + Returns: + Optional[dict]: The provider that should be marked as IS_GROUP=true, + or None if no group provider is found + """ + # Extract first 3 pages and signature pages + # Sort pages to ensure consistent order + # Pages come in as strings, so convert to int for sorting + pages = list(text_dict.keys()) + pages.sort(key=string_utils.page_key_sort) + + # I want to get the first 3 pages and the signature pages. But if they overlap, I want to avoid duplicates + first_pages = pages[:3] + signature_pages = string_utils.extract_signature_page(text_dict, filename).keys() # Use textract "This page has X signature" as a marker + + # Extract text from the first 3 pages and the signature pages + first_section = "\n".join([text_dict[page] for page in first_pages if page not in signature_pages]) # Avoid duplicates + signature_section = "\n".join([text_dict[page] for page in signature_pages]) # Use textract "This page has X signature" as a marker + + # Create focused sections for group identification + key_sections = first_section + "\n\n" + signature_section + + # Format provider list for inclusion in prompt + provider_list = json.dumps(providers, indent=2) + + # Construct prompt + prompt = investment_prompts.GROUP_TIN_NPI_TEMPLATE(key_sections=key_sections, provider_list=provider_list) + + # Get response and parse + response = llm_utils.invoke_claude(prompt, config.MODEL_ID_CLAUDE35_SONNET, filename) + try: + group_info = json.loads(string_utils.extract_text_from_delimiters(response, Delimiter.PIPE)) + + # Match with an existing provider + for provider in providers: + if provider.get("TIN") == group_info.get("TIN") and provider.get("NPI") == group_info.get("NPI"): + # Always use the group-provided name when available + # This is a fallback in case the group provider name is not in the original provider list + if group_info.get("NAME") is not None and group_info.get("NAME") not in ["", "null", "None", "N/A", "UNKNOWN"]: + provider["NAME"] = group_info.get("NAME") + # Mark as group + provider["IS_GROUP"] = True + return provider + except Exception as e: + print(f"Error parsing group provider response: {response}. Exception: {e}") + return None + + # Fallback: if no match found, return None + return None + + +def deduplicate_providers(provider_info: list[dict]) -> list[dict]: # TODO: add unit test + """Deduplicates provider information based on TIN and NPI. + 1. Use TIN+NPI as primary key + 2. Merge information when the same provider appears in multiple places + + Args: + provider_info (list[dict]): A list of provider information dictionaries to be deduplicated. + + Returns: + list[dict]: A list of deduplicated provider information dictionaries. + """ + provider_map = {} # Key: (TIN, NPI), Value: provider dict + no_id_providers = [] # List of providers with no TIN or NPI + + for provider in provider_info: + key = (provider.get("TIN", ""), provider.get("NPI", "")) + + # Skip empty keys + if key == ("", ""): + no_id_providers.append(provider) + continue + + if key not in provider_map: # If the key is not in the map, add it + provider_map[key] = provider + else: # If the key is already in the map, merge the provider info + existing_provider = provider_map[key] + + # Keep non-empty NAME + if not existing_provider.get("NAME") and provider.get("NAME"): + existing_provider["NAME"] = provider.get("NAME") + + # If marked as group anywhere, keep as group + if provider.get("IS_GROUP", False): + existing_provider["IS_GROUP"] = True + + # Return both identified providers and those without IDs + return list(provider_map.values()) + no_id_providers + + +def extract_identifiers(contract_text: str, filename: str) -> tuple: + """ + Extracts Taxpayer Identification Numbers (TINs) and National Provider Identifiers (NPIs) + from the given contract text using predefined patterns. + + Args: + contract_text (str): The text content of the contract to search for identifiers. + filename (str): The name of the file being processed, used for logging or debugging. + + Returns: + tuple: A tuple containing two lists: + - A list of all matched TINs found in the contract text. + - A list of all matched NPIs found in the contract text. + """ + + + # Get all matches for TIN and NPI + all_tins = get_all_matches(contract_text, regex_patterns.TIN_PATTERN, filename) + all_npis = get_all_matches(contract_text, regex_patterns.NPI_PATTERN, filename) + + return all_tins, all_npis + + +def get_provider_info(chunk: str, filename: str) -> list: + """ + Extracts provider information from a given text chunk using a language model. + + Args: + chunk (str): The text chunk containing provider information to be extracted. + filename (str): The name of the file being processed, used for logging or tracking purposes. + + Returns: + list: A list of extracted provider information in JSON format, parsed from the language model's response. + + Raises: + ValueError: If the response from the language model cannot be parsed as valid JSON. + """ + provider_fields = FieldSet(file_path=config.FIELD_JSON_PATH, field_type="provider_info") + # Generate prompt + prompt = investment_prompts.TIN_NPI_TEMPLATE(chunk, provider_fields.get_prompt_dict()) + # Get response from LLM + claude_answer_raw = llm_utils.invoke_claude(prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, max_tokens=8192) # Sometimes rosters can have 50+ providers + + # Defensive programming for any json parsing errors + try: + providers = string_utils.universal_json_load(claude_answer_raw) + + # Ensure we have a list of providers (not a dict or other type) + if isinstance(providers, dict): + # Single provider returned as dict - wrap in list + providers = [providers] + elif not isinstance(providers, list): + # If not a list or dict, create a default list with error info + print(f"Warning: Unexpected format in provider info response: {type(providers)}") + providers = [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "PARSING_ERROR", "IS_GROUP": False}] + + # For all providers, set IS_GROUP to False + for provider in providers: + provider["IS_GROUP"] = False + + return providers + except ValueError as e: # Handle JSON parsing error + print(f"Error parsing JSON response from LLM: {str(e)}") + print(f"Raw response: {claude_answer_raw}") + # Return a default value + return [{"TIN": "UNKNOWN", "NPI": "UNKNOWN", "NAME": "PARSING_ERROR", "IS_GROUP": False}] + + diff --git a/fieldExtraction/src/prompts/investment_prompts.py b/fieldExtraction/src/prompts/investment_prompts.py index 567032c..eeafcae 100644 --- a/fieldExtraction/src/prompts/investment_prompts.py +++ b/fieldExtraction/src/prompts/investment_prompts.py @@ -1,19 +1,21 @@ -from src.constants.investment_values import VALID_LOBS, VALID_NETWORKS, VALID_PROGRAMS, VALID_CLAIM_TYPE, VALID_REIMB_TERM, VALID_CARVEOUTS, VALID_PROV_TYPE, VALID_SPECIAL, VALID_AARETE_DERIVED_FEE_SCHEDULE, VALID_AARETE_DERIVED_FEE_SCHEDULE_VERSION, VALID_PROV_SPECIALTY, VALID_PLACE_OF_SERVICE, VALID_BILL_TYPE -import src.constants.valid as valid -import pandas as pd +import importlib +import json from functools import cache + +import pandas as pd +import src.constants.valid as valid import src.utils.llm_utils as llm_utils import src.utils.string_utils as string_utils from src.config import MODEL_ID_CLAUDE35_SONNET +from src.constants.investment_values import ( + VALID_AARETE_DERIVED_FEE_SCHEDULE, + VALID_AARETE_DERIVED_FEE_SCHEDULE_VERSION, VALID_BILL_TYPE, + VALID_CARVEOUTS, VALID_CLAIM_TYPE, VALID_LOBS, VALID_NETWORKS, + VALID_PLACE_OF_SERVICE, VALID_PROGRAMS, VALID_PROV_SPECIALTY, + VALID_PROV_TYPE, VALID_REIMB_TERM, VALID_SPECIAL) from src.enums.delimiters import Delimiter from src.utils.string_utils import extract_text_from_delimiters -##################################################################################### -###################################### PROMPTS ###################################### -##################################################################################### -import json -import importlib - PIPE_FORMAT_INSTRUCTIONS = "Feel free to justify your answer, but enclose your final answer in |pipes|. If the requested information doesn't apply to this context, return |N/A|. If the information should exist but cannot be clearly identified, return |UNKNOWN|." class Field: @@ -628,66 +630,62 @@ def AARETE_DERIVED_PROMPT_TEMPLATE(contract_answer: str, valid_derived_values: l def TIN_NPI_TEMPLATE(context, questions): return f""" -Analyze the following contract text and identify ALL provider entities mentioned, including the main contracting group AND any individual practitioners or facilities associated with it, along with their identifying information. +[CRITICAL INSTRUCTION: THIS IS AN EXHAUSTIVE EXTRACTION TASK] +Analyze the following contract text and extract EVERY SINGLE provider entity mentioned - this means ALL providers without exception. -For EACH provider entity found (group or individual), extract: +For EACH provider entity found (group or individual practitioner), extract: - {'\n- '.join(f'{field} - {question}' for (field, question) in questions.items())} -- IS_GROUP - Identify whether this entity is the primary contracting party (the main group). Answer 'true' for the main group and 'false' for others (like individual doctors listed under a group practice). -Instructions: -1. Scan the entire text for any mention of provider names, Tax Identification Numbers (TINs), and National Provider Identifiers (NPIs). -2. Extract the TIN, NPI, and Full Name for every distinct provider entity you find. Do not skip any providers regardless of how many there are. -3. Determine which entity is the main contracting party (often mentioned on the first page or signature blocks) and set its IS_GROUP to 'true'. -4. Set IS_GROUP to 'false' for all other individual providers or facilities listed. -5. Your output MUST be exhaustive - length is not a concern -6. It is crucial to return information for ALL providers found, not just the main group. -7. Before finalizing, verify you have included ALL providers mentioned. +Feel free to explain your reasoning as you work through the document. After your explanation, include the heading "FINAL PROVIDER INFO:" followed by a properly formatted JSON array with your final output. -Return your answers as a JSON array of objects. Each object must have the keys: "TIN", "NPI", "NAME", "IS_GROUP". If a TIN or NPI is not found for a specific provider name, return null for that field but still include the name and set IS_GROUP appropriately. +[FORMATTING INSTRUCTIONS] +- Each provider must be an object within this array +- All provider objects must have these exact keys: "TIN", "NPI", "NAME" +- Use null (not strings like "null" or "N/A") for missing TIN or NPI values +- Your final JSON array must begin with an opening bracket '[' and end with a closing bracket ']' +- Make sure the JSON array is correctly formatted with commas between objects + +Example of correct format: +FINAL PROVIDER INFO: +[ + {{ + "TIN": "123456789", + "NPI": "1234567890", + "NAME": "Main Provider Group" + }}, + {{ + "TIN": "987654321", + "NPI": null, + "NAME": "Dr. John Smith" + }} +] + +[EXTRACTION PROCESS] +1. Search systematically for ALL providers mentioned in this text +2. For each provider, extract their TIN, NPI, and full name exactly as they appear. +3. After your extraction, include the heading "FINAL PROVIDER INFO:" followed by a properly formatted JSON array. Contract text: {context} """ +def GROUP_TIN_NPI_TEMPLATE(key_sections: str, provider_list: str) -> str: + return f""" + Identify which provider is the MAIN CONTRACTING GROUP in this contract. + + Below is a list of all providers found in the contract: + {provider_list} + + Based on the key sections of the contract (first and last parts where party definitions and signatures are usually located): + + {key_sections} + + Determine which provider is the main contracting entity/group. + Return the name, TIN, and NPI of the main group provider in JSON format like this: + |{{"NAME": "Provider Name", "TIN": "123456789", "NPI": "1234567890"}}| + + If you cannot confidently determine the main group, return null values: + |{{"NAME": null, "TIN": null, "NPI": null}}| -def PROV_GROUP_TIN_CHECK(contract_text): - # Switch to return JSON with keys PROV_GROUP_TIN, PROV_OTHER_TIN - as just the regexed TINs that are actual TINs - return f"""Examine the contract text below and determine which of the Tax Identification Numbers (TINs) seen belong to the provider group, as opposed to a specific provider. - -TINs are 9-digit numbers, often of the format '12-3456789'. They may also be found as a straight sequence of nine digits, like '123456789'. In some cases, they may be in the standard social security number format: '123-45-6789'. - -The Group TIN will most often be the one listed on the Signature Page, or the one found on the first page of the contract, if applicable. - -If there are multiple Group TINs, return them in a comma-separated list. -If TINs are present but none can be clearly identified as the Group TIN, return 'UNKNOWN'. -If there are no TINs at all or the contract doesn't involve a Provider Group, return 'N/A'. - -Here is the contract text to analyze: - -{contract_text} - -{PIPE_FORMAT_INSTRUCTIONS} -""" - -def PROV_GROUP_NPI_CHECK(contract_text, prov_group_tin): - if not string_utils.is_empty(prov_group_tin): - tin_sentence = f"We have previously identified the Group TIN as '{prov_group_tin}'. The Group NPI will be the NPI that is associated with the same Provider as the Group TIN. It will most likely be found near this TIN." - else: - tin_sentence = "The Group NPI will most often be the one listed on the Signature Page, or the one found on the first page of the contract, if applicable." - - return f"""Examine the contract text below and determine which of the National Provider Identifiers (NPIs) seen belong to the provider group, as opposed to a specific provider. - -NPIs are 10-digit numbers, usually labelled as NPI. - -{tin_sentence} - -If there are multiple Group NPIs, return them in a comma-separated list. -If NPIs are present but none can be clearly identified as the Group NPI, return 'UNKNOWN'. -If there are no NPIs at all or the contract doesn't involve a Provider Group, return 'N/A'. - -Here is the contract text to analyze: - -{contract_text} - -{PIPE_FORMAT_INSTRUCTIONS} -""" + Feel free to justify your answer, but enclose your final answer in |pipes|. + """ diff --git a/fieldExtraction/src/regex/regex_utils.py b/fieldExtraction/src/regex/regex_utils.py index 9e3fbba..cc2e698 100644 --- a/fieldExtraction/src/regex/regex_utils.py +++ b/fieldExtraction/src/regex/regex_utils.py @@ -2,16 +2,17 @@ import re from src import postprocessing_funcs +## These functions are all client-only, except `find_regex_matches` which is used for reimbursement TIN/NPI. ################################################################################################################################################# # 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. + """In case of Multiple 10-digit regex matches, this function will check if any of those matches are on the signature page. We check for the word 'Signature' or 'SIGNATURE' to determine the signature page Args: 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 10-digit regex natches + matches(list): List of 10-digit regex matches Returns: npi(str): NPI on the signature page or N/A diff --git a/fieldExtraction/src/utils/string_utils.py b/fieldExtraction/src/utils/string_utils.py index 45b77ef..1b71904 100644 --- a/fieldExtraction/src/utils/string_utils.py +++ b/fieldExtraction/src/utils/string_utils.py @@ -1,16 +1,15 @@ -from datetime import datetime import json import re +import traceback import warnings +from datetime import datetime import numpy as np import pandas as pd -import traceback - -from src.prompts import preprocessing_prompts import src.utils.llm_utils as llm_utils from src import config from src.enums.delimiters import Delimiter +from src.prompts import preprocessing_prompts from src.regex.regex_patterns import (BACKTICK_PATTERN, PIPE_PATTERN, TRIPLE_BACKTICK_PATTERN) @@ -72,6 +71,35 @@ def extract_text_from_delimiters( return matches[match_index] +def page_key_sort(page_key: str) -> tuple[int, int|str]: + """Sorts page keys, prioritizing numeric keys first. + + Args: + page_key (str): The page key to sort. + + Returns: + tuple[int, str]: A tuple where the first element is 0 for numeric keys and 1 for non-numeric keys, and the second element is the page key. + + Example: + >>> page_key_sort("1") + (0, 1) + >>> page_key_sort("A") + (1, "A") + + + Example Usage with mixed types: + >>> L = ["10", "2", "A", "1"] + >>> sort(L, key=page_key_sort) + >>> print(L) + ['1', '2', '10', 'A'] + + """ + try: + return (0, int(page_key)) # Numeric pages first, in numerical order + except ValueError: + return (1, str(page_key)) # Non-numeric pages after, in alphabetical order + + def json_parsing_search(response_text: str, field_list: list[str]) -> dict[str, str]: """ Parses a JSON-like string to extract values for fields. @@ -355,3 +383,35 @@ def get_exhibit_chunk(text_dict: dict, def datetime_str(): return datetime.now().strftime("[%Y-%m-%d %H:%M:%S]") + + +def extract_signature_page(text_dict: dict, filename: str) -> dict: + """Extracts the signature page(s) from a contract text dictionary and returns it as a dictionary. + First tries to match the textract marker "this page has N signature(s)", then falls back to looking + for the word "signature" if no matches are found. "this page has N signature(s)" is Textract's marker for signatures + + Args: + text_dict (dict): A dictionary containing text data, organized by pages. + filename (str): The name of the file being processed. + + Returns: + dict: A dictionary containing the extracted signature page(s). + """ + + signature_pages = {} + # This regex pattern matches the Textract marker for signatures + # It will look for both numeric and a few textual representations of numbers (one-ten) + textract_pattern = re.compile(r"this page has (?:\d+|one|two|three|four|five|six|seven|eight|nine|ten) signature", re.IGNORECASE) + + # First pass: Try to match the full textract pattern + for page_num, page_text in text_dict.items(): + if textract_pattern.search(page_text): + signature_pages[page_num] = page_text + + # If no pages found with textract pattern, fall back to "signature" keyword + if not signature_pages: + for page_num, page_text in text_dict.items(): + if "signature" in page_text.lower(): + signature_pages[page_num] = page_text + + return signature_pages diff --git a/fieldExtraction/tests/string_utils_test.py b/fieldExtraction/tests/string_utils_test.py index 629259d..c7b0396 100644 --- a/fieldExtraction/tests/string_utils_test.py +++ b/fieldExtraction/tests/string_utils_test.py @@ -1,6 +1,6 @@ import pytest import src.utils as utils -from src.utils.string_utils import extract_text_from_delimiters, json_parsing_search, secondary_string_to_dict, primary_string_to_dict, contains_reimbursement, is_empty, count_reimbursements_in_exhibit +from src.utils.string_utils import extract_text_from_delimiters, json_parsing_search, secondary_string_to_dict, primary_string_to_dict, contains_reimbursement, is_empty, count_reimbursements_in_exhibit, extract_signature_page, page_key_sort import src.utils.llm_utils as llm_utils import numpy as np import pandas as pd @@ -195,6 +195,69 @@ class TestStringUtils: result = is_empty(value) assert result == expected + @pytest.mark.parametrize("text_dict, filename, expected", [ + ( + {"1": "This page has 2 signatures.", "2": "None here.", "3": "Signature of the party (but ignored)"}, + "test_file.txt", + {"1": "This page has 2 signatures."} + ), + ( + {"1": "No relevant content here.", "2": "Nor over here."}, + "test_file.txt", + {} + ), + ( + {"1": "This page has a signature.", "2": "Another signature is found here."}, + "test_file.txt", + {"1": "This page has a signature.", "2": "Another signature is found here."} + ), + ( + {"1": "This page has 1 signature.", "2": "No relevant content.", "3": "Signature block here. But ignored bc of the first textract-like page"}, + "test_file.txt", + {"1": "This page has 1 signature."} + ), + ( + {"1": "No signed names at all.", "2": "Just some random text."}, + "test_file.txt", + {} + ), + ( + {"1": "This page has a signature.", "2": "Signature is mentioned here too."}, + "test_file.txt", + {"1": "This page has a signature.", "2": "Signature is mentioned here too."} + ), + ( + {}, + "test_file.txt", + {} + ), + ]) + def test_extract_signature_page(self, text_dict, filename, expected): + result = extract_signature_page(text_dict, filename) + assert result == expected + + @pytest.mark.parametrize("page_key, expected", [ + ("1", (0, 1)), # Numeric key + ("10", (0, 10)), # Larger numeric key + ("A", (1, "A")), # Alphabetic key + ("Z", (1, "Z")), # Another alphabetic key + ("1A", (1, "1A")), # Alphanumeric key + ("", (1, "")), # Empty string + ("001", (0, 1)), # Numeric key with leading zeros + ("B2", (1, "B2")), # Alphanumeric key with letter first + ("-5", (0, -5)), # Negative numeric + (" ", (1, " ")), # Space character + ]) + def test_page_key_sort(self, page_key, expected): + result = page_key_sort(page_key) + assert result == expected + + def test_page_key_sort_with_sorting(self): + keys = ["10", "IV", "2", "A", "1", "B", "Z"] + expected_sorted_keys = ["1", "2", "10", "A", "B", "IV", "Z"] + sorted_keys = sorted(keys, key=page_key_sort) + assert sorted_keys == expected_sorted_keys + class TestCountReimbursementsInExhibit: @pytest.mark.parametrize("exhibit_text, expected_count", [ ("This exhibit includes a 10% reimbursement and a $100 reimbursement.", 2),