import json import ast import logging import re import traceback import warnings from datetime import datetime from typing import Literal, overload import numpy as np import pandas as pd from src.constants.delimiters import Delimiter from src.constants.regex_patterns import ( BACKTICK_PATTERN, PIPE_PATTERN, TRIPLE_BACKTICK_PATTERN, ) def extract_text_from_delimiters( raw_output: str, delimiter: Delimiter, match_index: int = -1 ) -> str: """ Extracts a match enclosed in specified delimiters from the raw output. This function searches for text enclosed by a specified delimiter in the given raw output string. It returns the match specified by the `which_match` index. If no match is found, it returns "N/A". - raw_output (str): The raw output string to search within. - which_match (int, optional): The index of the match to return. Defaults to -1, which returns the last match. If the index is out of range, the last match is returned. - str: The extracted text or "N/A" if no match is found. Raises: - TypeError: If `raw_output` is not a string or `delimiter` is not a Delimiter enum value. - ValueError: If an unsupported delimiter is provided. Returns: - str: The extracted answer or "N/A" if no match is found. .. deprecated:: 2026.02 Using Delimiter.PIPE with this function is deprecated and will be removed in a future version. Prompts should return JSON format instead of pipe-delimited strings. Use `json_utils.parse_json_list()` or `json_utils.parse_json_dict()` for parsing JSON responses. """ if not isinstance(raw_output, str): raise TypeError( "Expected a string for raw_output, got {0}.".format( type(raw_output).__name__ ) ) if not isinstance(delimiter, Delimiter): raise TypeError( "Expected a Delimiter enum value, got {0}.".format(type(delimiter).__name__) ) # Define a pattern based on the delimiter type if delimiter == Delimiter.PIPE: warnings.warn( "Delimiter.PIPE is deprecated for LLM output parsing. " "Prompts should return JSON format instead. " "Use json_utils.parse_json_list() or json_utils.parse_json_dict() instead.", DeprecationWarning, stacklevel=2, ) pattern = PIPE_PATTERN elif delimiter == Delimiter.BACKTICK: pattern = BACKTICK_PATTERN elif delimiter == Delimiter.TRIPLE_BACKTICK: pattern = TRIPLE_BACKTICK_PATTERN else: raise ValueError("Unsupported delimiter. Use one of the Delimiter enum values.") # Find all matches based on the pattern matches = re.findall(pattern=pattern, string=raw_output) if len(matches) == 0: return "N/A" # Adjust for negative indices if match_index < 0: match_index += len(matches) if not 0 <= match_index < len(matches): warnings.warn("Index out of range. Returning the last match by default.") match_index = -1 return matches[match_index] def page_key_sort(page_key: str) -> tuple[int, float | str]: """Sorts page keys, prioritizing numeric keys first. Args: page_key (str): The page key to sort. Returns: tuple[int, float|str]: A tuple where the first element is 0 for numeric keys and 1 for non-numeric keys, and the second element is the numeric value for numeric keys or the original string for non-numeric keys. Example: >>> page_key_sort("1") (0, 1.0) >>> 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, float(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. This function attempts to clean and format the input string to resemble a valid JSON structure. It then searches for the specified fields within the string and extracts their corresponding values. Parameters: response_text (str): The JSON-like string to parse. field_list (list[str]): A list of field names to search for in the response_text. Returns: dict[str, str]: A dictionary where keys are field names and values are the extracted values from the response_text. """ try: if response_text.split("{", 1)[1].strip()[0] == '"': response_text = "{" + response_text.split("{", 1)[1] else: response_text = "{" + response_text except: response_text = "{" + response_text if len(response_text.split("}", 1)) > 1: if response_text.rsplit("}", 1)[0].strip()[-1] == '"': response_text = response_text.rsplit("}", 1)[0] + "}" elif response_text.rsplit("}", 1)[0].strip()[-1] == "}": response_text = response_text.rsplit("}", 1)[0] else: response_text = response_text.rstrip(",") + "}" field_l = [] answer_l = [] position_dict = {} for f in field_list: location = response_text.find('"' + f + '"') if location != -1: position_dict[location] = f field_list = list(dict(sorted(position_dict.items())).values()) for f in field_list: if f in response_text: field_l.append(f) value = response_text.split('"' + f + '"')[0] response_text = response_text.split('"' + f + '"')[1] if f != field_list[0] and f != field_list[-1]: answer_l.append( value.strip("\n") .strip('"') .strip(":") .strip(" ") .strip("\n") .strip(" ") .strip(",") .strip('"') ) elif f == field_list[-1]: answer_l.append( value.strip("\n") .strip('"') .strip(":") .strip(" ") .strip("\n") .strip(" ") .strip(",") .strip('"') ) answer_l.append( response_text.strip("\n") .strip('"') .strip(":") .strip(" ") .strip("}") .strip("\n") .strip(" ") .strip('"') ) return dict(zip(field_l, answer_l)) def universal_json_load(s: str): """ Extract and parse all highest-level valid JSON objects from a string. Returns the last valid top-level JSON object found (dict, list, etc). Raises ValueError if no valid JSON objects are found. Does not clean up syntax. .. deprecated:: 2026.02 This function is deprecated and will be removed in a future version. Use `json_utils.parse_json_dict()` for dictionary outputs or `json_utils.parse_json_list()` for list outputs instead. The new parsers provide type-specific parsing with better error handling. """ warnings.warn( "universal_json_load is deprecated and will be removed in a future version. " "Use json_utils.parse_json_dict() or json_utils.parse_json_list() instead.", DeprecationWarning, stacklevel=2, ) if not isinstance(s, str): raise TypeError(f"Expected string input, got {type(s).__name__}") found = [] i = 0 while i < len(s): if s[i] in "{[": start = i stack = [s[i]] j = i + 1 while j < len(s): # Skip over string literals to avoid counting brackets inside strings if s[j] == '"': j += 1 while j < len(s): if s[j] == "\\" and j + 1 < len(s): j += 2 # Skip escaped character continue if s[j] == '"': break j += 1 elif s[j] in "{[": stack.append(s[j]) elif s[j] in "}]": if not stack: break open_bracket = stack.pop() if (open_bracket == "{" and s[j] != "}") or ( open_bracket == "[" and s[j] != "]" ): break if not stack: candidate = s[start : j + 1] # Only consider top-level objects (not nested) # Check if start is not inside another object # This is guaranteed by stack being empty only at top-level try: obj = json.loads(candidate) found.append(obj) except Exception: pass i = j # Move past this top-level object break j += 1 else: pass i += 1 if found: return found[-1] raise ValueError(f"No valid JSON objects found in string: {s[:100]}...") reimbursement_strings = [ # These are for `method='keyword'` "%", "$", "percent", "billed charges", "compensation", "reimbursement", "fee schedule", ] reimb_regex = r"(? int: # JUST by regex """Counts reimbursements in an exhibit text. Reimbursements are detected by a regex search as defined by `reimb_regex`. Args: exhibit_text (str): Input exhibit text Returns: int: Number of reimbursements detected """ return len(re.findall(reimb_regex, exhibit_text)) # Overloads for type checking @overload def is_empty(value: str, pd_mask: bool = True) -> bool: ... @overload def is_empty(value: list, pd_mask: bool = True) -> bool: ... @overload def is_empty(value: pd.Series, pd_mask: Literal[True] = True) -> pd.Series: ... @overload def is_empty(value: pd.Series, pd_mask: Literal[False]) -> bool: ... def is_empty(value: str | list | pd.Series, pd_mask: bool = True) -> bool | pd.Series: """ Checks if a value is considered empty or invalid. Args: value: The value to check. Can be a string, list, or pandas Series. pd_mask (bool): Only relevant for Series inputs. If True, returns a mask for empty values in a pandas Series. Otherwise, the function returns True iff the entire Series is empty. Returns: bool | pd.Series: - When the input is a string, returns True if the value is empty or invalid, False otherwise. - When the input is a list, returns True if the list is empty or all elements are empty/invalid. - When the input is a pandas Series: - If `pd_mask` is True, returns a boolean mask indicating which elements are empty or invalid. - If `pd_mask` is False, returns True iff all elements are empty/invalid, or if the entire Series is empty. """ empty_values = [None, "", "N/A", "NA", "null", "none", "NaN", np.nan, "nan", "None"] if isinstance(value, list): # Handle list inputs return not value or all(is_empty(v) for v in value) # if it's a pd.Series, return the mask (True for empty values) if isinstance(value, pd.Series): if pd_mask: return ( value.isna() | (value.isin(empty_values)) | (value.astype(str).str.strip() == "") | ( value.astype(str) .str.lower() .str.strip() .isin(["n/a", "na", "null", "none", "nan"]) ) ) else: # If the Series is empty, it's considered empty if value.empty: return True # Use the string-version check for each element in the Series return all(is_empty(str(v)) for v in value) if pd.isna(value): return True if isinstance(value, str): if not value or value.isspace(): return True lower_stripped = value.lower().strip() if lower_stripped in [ "n/a", "na", "null", "none", "nan", "no_identifiers_found", ]: return True return value in empty_values 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, ) keywords = [ "signature and information", "signature authorization", "in witness whereof", ] # 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, try with keyword if not signature_pages: # Iterate through each page in the text dictionary and check for keywords for page_num, page_text in text_dict.items(): lower_text = page_text.lower() if any(keyword in lower_text for keyword in keywords): signature_pages[page_num] = page_text # If no signature pages found, try checking for the word "signature page follows" and return the next page only if it exists if not signature_pages: for page_num, page_text in text_dict.items(): if "signature page follows" in page_text.lower(): next_page_num = str(int(page_num) + 1) if next_page_num in text_dict: signature_pages[next_page_num] = text_dict[next_page_num] return signature_pages def extract_effective_date_pages(text_dict: dict, filename: str) -> dict: """ Extracts the effective date page(s) from a contract text dictionary and returns it as a dictionary. This function identifies pages that either contain the Textract marker "this page has N signature(s)" or the keyword 'effective date:'. Additionally, it ensures that the first page (page 1) is always included in the result, as it often contains effective date. 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 effective date page(s), where keys are page numbers and values are the corresponding page text. """ effective_date_pages = {} # Regex pattern to match the Textract marker for signature pages which often has effective dates # This pattern accounts for both numeric and 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, ) # Iterate through each page in the text dictionary. for page_num, page_text in text_dict.items(): # page_num = int(float(page_num)) # Check if the page contains the Textract marker or the keyword 'effective date:'. if textract_pattern.search(page_text) or "effective date:" in page_text.lower(): effective_date_pages[page_num] = page_text # Always include the first page (page 1) as it often contains effective date. elif page_num == 1 or page_num == "1" or page_num == "1.0": effective_date_pages[page_num] = page_text return effective_date_pages def normalize_state_to_abbreviation(state_value: str) -> str: """Convert state names or abbreviations to standardized two-letter abbreviations. Handles various formats of state names and abbreviations, including full names, common abbreviations, and variations. Returns the standardized two-letter abbreviation or the original value if no match is found. Args: state_value (str): State name or abbreviation from extraction Returns: str: Two-letter state abbreviation or original value if no match found """ if not state_value or state_value in ["N/A", "UNKNOWN"]: return state_value # Dictionary mapping full state names to abbreviations state_mapping = { "alabama": "AL", "alaska": "AK", "arizona": "AZ", "arkansas": "AR", "california": "CA", "colorado": "CO", "connecticut": "CT", "delaware": "DE", "florida": "FL", "georgia": "GA", "hawaii": "HI", "idaho": "ID", "illinois": "IL", "indiana": "IN", "iowa": "IA", "kansas": "KS", "kentucky": "KY", "louisiana": "LA", "maine": "ME", "maryland": "MD", "massachusetts": "MA", "michigan": "MI", "minnesota": "MN", "mississippi": "MS", "missouri": "MO", "montana": "MT", "nebraska": "NE", "nevada": "NV", "new hampshire": "NH", "new jersey": "NJ", "new mexico": "NM", "new york": "NY", "north carolina": "NC", "north dakota": "ND", "ohio": "OH", "oklahoma": "OK", "oregon": "OR", "pennsylvania": "PA", "rhode island": "RI", "south carolina": "SC", "south dakota": "SD", "tennessee": "TN", "texas": "TX", "utah": "UT", "vermont": "VT", "virginia": "VA", "washington": "WA", "west virginia": "WV", "wisconsin": "WI", "wyoming": "WY", "district of columbia": "DC", } # Clean and normalize the input cleaned_state = state_value.strip().lower() # Check if it's already a valid 2-letter abbreviation if len(cleaned_state) == 2 and cleaned_state.upper() in state_mapping.values(): return cleaned_state.upper() # Check if it matches a full state name if cleaned_state in state_mapping: return state_mapping[cleaned_state] # Handle common variations variations = { "calif": "CA", "cal": "CA", "fla": "FL", "florida": "FL", "ny": "NY", "n.y.": "NY", "tx": "TX", "tex": "TX", "penn": "PA", "pa": "PA", "mass": "MA", "massachusetts": "MA", "wash": "WA", "washington state": "WA", "d.c.": "DC", "dc": "DC", "washington dc": "DC", } if cleaned_state in variations: return variations[cleaned_state] # If no match found, return original value return state_value def flatten_to_strings(items) -> list[str]: """Helper function to flatten and stringify any nested structure Args: items (Any): The input items; can be a list, tuple, or single value Returns: list[str]: A flattened list of stringified items """ result = [] if isinstance(items, (list, tuple)): for item in items: if isinstance(item, (list, tuple)): result.extend(flatten_to_strings(item)) else: result.append(str(item).strip()) else: result.append(str(items).strip()) return result def parse_state_field_to_list(raw_value): """ Convert PROVIDER_STATE raw input into a clean list of state strings. Accepts: - list → returned as-is - JSON list string → parsed to list - comma-separated string → split to list - single string → wrapped in list - other types → converted to string → wrapped in list Returns: list[str]: Clean list of state strings """ if raw_value is None: return [] # Case 1: Already a list if isinstance(raw_value, list): return raw_value # Case 2: String input if isinstance(raw_value, str): stripped = raw_value.strip() # JSON list string if stripped.startswith("[") and stripped.endswith("]"): try: parsed = json.loads(stripped) if isinstance(parsed, list): return parsed else: return [parsed] except Exception as e: logging.debug( f"Failed JSON parse for state list: {e}. Falling back to comma split." ) # fallback: treat as comma-separated # Comma-separated list if "," in stripped: return [s.strip() for s in stripped.split(",") if s.strip()] # Single state string return [stripped] # Case 3: Any other type (int, dict, etc.) try: return [str(raw_value).strip()] except Exception: logging.error(f"Could not convert value to string: {raw_value!r}") return [] def normalize_state_field(answers_dict: dict, field_name: str = "PROVIDER_STATE"): if field_name not in answers_dict or not answers_dict[field_name]: answers_dict[field_name] = [] return answers_dict raw_value = answers_dict[field_name] # ---- Step 1: Parse into list ---- state_list = parse_raw_state_value(raw_value) # ---- Step 2: Normalize each state ---- normalized_states = [] for state in state_list: try: normalized_states.append(normalize_state_to_abbreviation(state)) except Exception as e: logging.warning( f"Failed to normalize state '{state}': {e}. Keeping original." ) normalized_states.append(state) answers_dict[field_name] = normalized_states return answers_dict def parse_raw_state_value(raw_value): """ Parses various input formats into a flat list of state strings. Handles: strings, lists, nested lists, and various delimiters. """ if not raw_value: return [] # If it's a string that looks like a list, try to parse it if isinstance(raw_value, str): stripped = raw_value.strip() # Try to parse as Python literal (handles "[['NC', 'WA'], 'CA|UT']") if stripped.startswith("["): try: parsed = ast.literal_eval(stripped) return parse_raw_state_value(parsed) # Recursive call with parsed list except (ValueError, SyntaxError): pass # Not valid Python literal, continue with string parsing # Regular string splitting on delimiters (not whitespace - state names can have spaces like "West Virginia") states = re.split(r"[|,;]+", stripped) return [s.strip() for s in states if s.strip()] if isinstance(raw_value, list): result = [] for item in raw_value: result.extend(parse_raw_state_value(item)) # Recursive call return result return [str(raw_value)] def normalize_to_json_list(val): """ Normalizes a value to a JSON list format. Handles: pipe-delimited strings, string representations of Python lists, actual lists with pipe-delimited items, and single values. """ # Handle None and empty values first if is_empty(val): return val # Handle pandas NA/NaN - check for scalar first try: if pd.isna(val): return val except (ValueError, TypeError): # pd.isna fails on arrays/lists, which is fine - we'll handle them below pass if isinstance(val, str): # Check if it's a string representation of a Python list (e.g., "['item1', 'item2']") if val.startswith("[") and val.endswith("]"): try: # Try JSON parse first parsed = json.loads(val) except json.JSONDecodeError: # If JSON fails, try ast.literal_eval for Python list syntax try: parsed = ast.literal_eval(val) except (ValueError, SyntaxError): return val if isinstance(parsed, list): expanded = [] for item in parsed: if isinstance(item, str) and "|" in item: expanded.extend(item.split("|")) else: expanded.append(item) result = json.dumps(expanded) return result return val # Pipe-delimited format - convert to JSON list if "|" in val: result = json.dumps(val.split("|")) return result # Single value - convert to JSON list result = json.dumps([val]) return result if isinstance(val, list): # Check if list contains pipe-delimited strings and expand them expanded = [] for item in val: if isinstance(item, str) and "|" in item: expanded.extend(item.split("|")) else: expanded.append(item) result = json.dumps(expanded) return result return val def normalize_one_to_one_field_value(field_name: str, value) -> str | list: """ Normalize a 1:1 field value to ensure single-value fields are strings, not lists. This function handles the common case where LLM returns a single-value field as a list with one element (e.g., ['The Value']). It extracts the string from such lists while preserving legitimate multi-value lists. Args: field_name (str): Name of the field being normalized (for logging/debugging) value: The field value to normalize. Can be: - str: Returned as-is - list with 1 element: Extract the string - list with >1 elements: Returned unchanged (may be legitimate multi-value) - None/empty: Returned as-is Returns: str | list: Normalized value: - If input is a list with exactly 1 element: returns that element as string - If input is already a string: returns as-is - If input is a list with >1 elements: returns list unchanged - If input is None or empty: returns as-is Examples: >>> normalize_one_to_one_field_value("CONTRACT_TITLE", ["Provider Agreement"]) 'Provider Agreement' >>> normalize_one_to_one_field_value("PAYER_NAME", "Aetna") 'Aetna' >>> normalize_one_to_one_field_value("PROVIDER_NAME", ["Provider A", "Provider B"]) ['Provider A', 'Provider B'] # Multi-value preserved >>> normalize_one_to_one_field_value("LOB", ["Medicare", "Medicaid"]) ['Medicare', 'Medicaid'] # Multi-value preserved """ # Handle None and empty values if value is None or (isinstance(value, str) and is_empty(value)): return value # If it's already a string, return as-is if isinstance(value, str): return value # If it's a list if isinstance(value, list): # List with exactly 1 element: extract the string if len(value) == 1: return str(value[0]) if value[0] is not None else "N/A" # List with >1 elements: preserve (may be legitimate multi-value) # This handles cases like PROVIDER_NAME with multiple providers, # or dynamic fields (LOB, PROGRAM, PRODUCT, NETWORK) when passed to 1:1 elif len(value) > 1: return value # Empty list: return "N/A" else: return "N/A" # For any other type, convert to string return str(value) if value is not None else "N/A" def normalize_one_to_one_answers_dict( answers_dict: dict[str, str | list], ) -> dict[str, str | list]: """ Normalize all values in a 1:1 answers dictionary. Applies normalization to all fields in the dictionary, converting single-element lists to strings while preserving legitimate multi-value lists. Args: answers_dict (dict): Dictionary of field names to values (may be strings or lists) Returns: dict: Dictionary with normalized values (single-element lists converted to strings) Example: >>> answers = { ... "CONTRACT_TITLE": ["Provider Agreement"], ... "PAYER_NAME": "Aetna", ... "PROVIDER_NAME": ["Provider A", "Provider B"] ... } >>> normalize_one_to_one_answers_dict(answers) { "CONTRACT_TITLE": "Provider Agreement", "PAYER_NAME": "Aetna", "PROVIDER_NAME": ["Provider A", "Provider B"] } """ DO_NOT_NORMALIZE = ["PROV_INFO_JSON", "PAYER_STATE", "PROVIDER_STATE"] normalized = {} for field_name, value in answers_dict.items(): if field_name not in DO_NOT_NORMALIZE: normalized[field_name] = normalize_one_to_one_field_value(field_name, value) else: normalized[field_name] = value return normalized