import logging import os import re import json import numpy as np from typing import Any from rapidfuzz import fuzz from collections import defaultdict import pandas as pd import src.pipelines.shared.postprocessing.postprocessing_funcs as postprocessing_funcs import src.pipelines.saas.prompts.prompt_calls as prompt_calls import src.utils.string_utils as string_utils from src.constants.constants import Constants from src import config from src.pipelines.shared.postprocessing import postprocessing_funcs from src.pipelines.shared.extraction.vision_funcs import get_image_array_based_answer from src.prompts.fieldset import FieldSet def check_and_update_effective_date(answers_dict, text_dict, filename): """Apply two-stage postprocessing to effective date with optional vision-based fallback. This function performs effective date extraction and validation in three steps: 1. **Text postprocessing**: Normalizes existing text-extracted date using date_postprocessing() to handle common text variations (e.g., "January 1, 2024" → "2024-01-01") 2. **Format validation**: Runs validate_and_reformat_date() to ensure ISO (YYYY-MM-DD) format and valid date values 3. **Vision fallback** (conditional): If config.ENABLE_VISION is True AND steps 1-2 yield an empty result, attempts OCR extraction from the source PDF using Claude vision API on targeted pages The vision fallback targets specific pages likely to contain effective date information (identified via extract_effective_date_pages()) to minimize token usage. Args: answers_dict (dict): Extraction results containing at least "AARETE_DERIVED_EFFECTIVE_DT" key. This dictionary is **mutated in place** with the updated effective date. text_dict (dict[int|str, str]): Page-segmented contract text where keys are page numbers and values are page content. Used to identify candidate pages for vision extraction via heuristics (e.g., pages containing "effective" keyword). filename (str): Source filename (e.g., "contract_001.txt"). Used for: - Logging context - Deriving PDF path by replacing .txt extension with .pdf Returns: dict: The mutated answers_dict with "AARETE_DERIVED_EFFECTIVE_DT" updated to: - A validated ISO date string (e.g., "2024-01-15") if extraction succeeded - "N/A" if all extraction methods failed Side Effects: - Mutates answers_dict["AARETE_DERIVED_EFFECTIVE_DT"] in place - Makes vision API call to Claude if config.ENABLE_VISION=True and text extraction fails - Logs to file-specific log via logging_utils context Notes: - Vision extraction requires PDF at: /root/doczy.ai/fieldExtraction/src/investment/pdf_files/{filename}.pdf - Vision extraction is functional in local testing but the infrastructure (namely storing and accessing PDFs via S3) is not yet built out in production - S3 path support exists but is commented out - Vision API uses get_image_array_based_answer() which converts PDF pages to images before LLM processing Example: >>> answers = {"AARETE_DERIVED_EFFECTIVE_DT": "Jan 1 2024"} >>> text = {1: "Agreement effective January 1, 2024", 2: "..."} >>> result = check_and_update_effective_date(answers, text, "contract.txt") >>> result["AARETE_DERIVED_EFFECTIVE_DT"] '2024-01-01' """ # Postprocess the effective date answers_dict["AARETE_DERIVED_EFFECTIVE_DT"] = prompt_calls.prompt_date_fix( answers_dict.get("EFFECTIVE_DT", "N/A") ) answers_dict["AARETE_DERIVED_EFFECTIVE_DT"] = ( postprocessing_funcs.validate_and_reformat_date( answers_dict["AARETE_DERIVED_EFFECTIVE_DT"] ) ) # If the effective date is empty, Get the effective date using the vision-based LLM if config.ENABLE_VISION and string_utils.is_empty( answers_dict.get("AARETE_DERIVED_EFFECTIVE_DT") ): # if answers_dict.get("AARETE_DERIVED_EFFECTIVE_DT") in ["N/A", "UNKNOWN"]: effective_date_page_numbers = list( string_utils.extract_effective_date_pages(text_dict, filename).keys() ) pdf_base_path = ( "/root/doczy.ai/fieldExtraction/src/investment/pdf_files" # local path ) pdf_path = os.path.join(pdf_base_path, filename.replace("txt", "pdf")) # TODO: Uncomment if/when S3 infrastructure for PDF storage/retrieval is built out # pdf_path = os.path.join(DOCZY_PDF_FILES_BUCKET_S3_URL, filename.replace("txt", "pdf")) # s3 path image_answer_dict = get_image_array_based_answer( pdf_path, effective_date_page_numbers, filename, fields=["AARETE_DERIVED_EFFECTIVE_DT"], ) answers_dict["AARETE_DERIVED_EFFECTIVE_DT"] = image_answer_dict.get( "AARETE_DERIVED_EFFECTIVE_DT", "N/A" ) return answers_dict def get_aarete_derived_termination_dt(one_to_one_results: dict): """Derive standardized termination date from effective date and term duration. Uses LLM to calculate contract termination date by combining the validated effective date with the extracted term duration text (e.g., "3 years from effective date"). Args: one_to_one_results (dict): Field values including "AARETE_DERIVED_EFFECTIVE_DT" and "TERMINATION_DT". Mutated in place with derived termination date. Returns: dict: The mutated one_to_one_results with added: - "AARETE_DERIVED_TERMINATION_DT": ISO-formatted date (YYYY-MM-DD) or "N/A" """ required_keys = ["AARETE_DERIVED_EFFECTIVE_DT", "TERMINATION_DT"] if all(k in one_to_one_results.keys() for k in required_keys): one_to_one_results["AARETE_DERIVED_TERMINATION_DT"] = ( prompt_calls.prompt_derived_term_date( one_to_one_results["AARETE_DERIVED_EFFECTIVE_DT"], one_to_one_results["TERMINATION_DT"], ) ) return one_to_one_results def get_aarete_derived_dates( one_to_one_results: dict, text_dict: dict, filename: str ) -> dict: """Process and derive standardized contract dates from extracted one-to-one field results. Orchestrates the complete date extraction and derivation pipeline combining text-based extraction, vision-based fallback, and logical date derivation to produce reliable contract date information. Processing workflow: 1. Effective date processing via check_and_update_effective_date() - text normalization, vision fallback if enabled, and ISO format standardization 2. Termination date derivation via get_aarete_derived_termination_dt() - logical derivation from effective date and term duration Args: one_to_one_results (dict): Extracted one-to-one field values including "EFFECTIVE_DT" and "TERMINATION_DT". Mutated in place with derived date fields. text_dict (dict): Page-segmented contract text for vision extraction context. filename (str): Source filename for logging and PDF path derivation. Returns: dict: The mutated one_to_one_results with added standardized date fields: - "AARETE_DERIVED_EFFECTIVE_DT": ISO-formatted date (YYYY-MM-DD) or "N/A" - "AARETE_DERIVED_TERMINATION_DT": ISO-formatted date (YYYY-MM-DD) or "N/A" """ one_to_one_results = check_and_update_effective_date( one_to_one_results, text_dict, filename ) one_to_one_results = get_aarete_derived_termination_dt(one_to_one_results) return one_to_one_results def add_signature_count(one_to_one_results: dict, contract_txt: str) -> dict: """ Adds 'NUM_SIGNED_SIGNATORY_LINE_COUNT' to one_to_one_results based on number of signature lines found in the contract text. Args: one_to_one_results (dict): The input dictionary. contract_txt (str): Extracted full contract text. Returns: dict: Updated dictionary with 'NUM_SIGNED_SIGNATORY_LINE_COUNT'. """ # Handle empty or missing text if contract_txt is None or contract_txt.strip() == "": one_to_one_results["NUM_SIGNED_SIGNATORY_LINE_COUNT"] = 0 return one_to_one_results # Pattern: "This page has X signature(s)." pattern = r"This page has (\d+)\s+signature[s]?\." matches = re.findall(pattern, contract_txt, re.IGNORECASE) total_signatures = sum(int(m) for m in matches) one_to_one_results["NUM_SIGNED_SIGNATORY_LINE_COUNT"] = total_signatures return one_to_one_results def add_aarete_derived_payer_name(df: pd.DataFrame, state_flag: bool) -> pd.DataFrame: """ Derive canonical payer names from PAYER_NAME column. This function clusters similar payer names, standardizes them, and selects the most appropriate canonical name for each cluster. Args: df: Input DataFrame containing a PAYER_NAME column. state_flag: Controls whether payer names are treated as state-specific. Always True (confirmed with business team). True → Preserve state distinctions. Example: "Molina Healthcare of Texas" and "Molina Healthcare of Utah" remain separate. False → Collapse regional state suffixes into national brand. Example: "Molina Healthcare of Texas" and "Molina Healthcare of Utah" both become "Molina Healthcare". Leading state-based brand names (e.g., "Oklahoma Complete Health") are preserved. Returns: DataFrame with added AARETE_DERIVED_PAYER_NAME column. """ df_payer = df.copy() # Check if PAYER_NAME column exists in the DataFrame, if not add an empty AARETE_DERIVED_PAYER_NAME column and return the DataFrame if "PAYER_NAME" not in df_payer.columns: logging.warning( "PAYER_NAME column not found in DataFrame. skipping AARETE_DERIVED_PAYER_NAME column." ) return df_payer # 1. Extract unique payer names payer_names = df_payer["PAYER_NAME"].dropna().unique().tolist() unique_payer_names = [ payer for payer in payer_names if not string_utils.is_empty(payer) ] if not unique_payer_names: df_payer["AARETE_DERIVED_PAYER_NAME"] = "" return df_payer # 2. Standardize OR normalizing unique payer names (list of dicts) standardized_unique_payer_name_rows = standardize_unique_names( unique_payer_names, state_flag ) standardized_unique_payer_name_rows = propagate_dba_to_legal( standardized_unique_payer_name_rows ) # Combining Step`s 3-4: Build similarity matrix and cluster with blocking to handle large volumes efficiently clusters = cluster_with_blocking( standardized_unique_payer_name_rows, state_flag=state_flag, threshold=90, ) # 5. Build mapping using LLM placeholder mapping = {} for cluster in clusters: derived_payer_name = prompt_calls.prompt_aarete_derived_payer_name( cluster, state_flag ) for record in cluster: mapping[record["original_name"]] = derived_payer_name df_payer["AARETE_DERIVED_PAYER_NAME"] = df_payer["PAYER_NAME"].map(mapping) return df_payer def standardize_unique_names( unique_names: list[str], state_flag: bool ) -> list[dict[str, Any]]: """ Normalize payer names and enrich them with DBA and token signatures. Runs only on unique payer names for efficiency. Block key is derived from the FIRST token of the base payer name (not from sorted token signature) to preserve brand identity. """ rows = [] for name in unique_names: clean = string_utils.normalize_text(name) state = string_utils.detect_state(clean) legal, dba = string_utils.extract_dba_name(clean) base = dba if dba else clean legal = legal if legal else clean base = string_utils.remove_legal_suffixes(base) legal = string_utils.remove_legal_suffixes(legal) if not state_flag: base = string_utils.remove_states(base) legal = string_utils.remove_states(legal) signature = string_utils.token_signature(base) tokens = base.split() block_key = tokens[0] if tokens else "" rows.append( { "original_name": name, "base_name": base, "signature": signature if signature else (), "has_dba_name": dba is not None, "dba_name": dba, "legal_name": legal, "state": state, "block_key": block_key, } ) return rows def build_similarity_matrix(rows: list[dict]) -> np.ndarray: """ Create a pairwise similarity matrix for standardized entity names. Computes fuzzy similarity scores between each pair of rows using both base_name and legal_name fields. The maximum score from multiple fuzzy comparisons is used as the final similarity value. Args: rows : list[dict] List of standardized entity dictionaries containing "base_name" and "legal_name". Returns: np.ndarray An (n x n) symmetric matrix of similarity scores (0–100), where n is the number of rows. """ # Number of payer name records n = len(rows) # Initialize an empty similarity matrix sim_matrix = np.zeros((n, n)) # Compute pairwise similarities for i in range(n): for j in range(i, n): if i == j: # A string is always 100% similar to itself sim_matrix[i, j] = 100 else: # Compute similarity using multiple fuzzy matching methods and different combinations score = max( fuzz.ratio(rows[i]["base_name"], rows[j]["base_name"]), fuzz.token_set_ratio(rows[i]["base_name"], rows[j]["base_name"]), fuzz.ratio(rows[i]["legal_name"], rows[j]["base_name"]), fuzz.token_set_ratio(rows[i]["legal_name"], rows[j]["base_name"]), fuzz.ratio(rows[i]["base_name"], rows[j]["legal_name"]), fuzz.token_set_ratio(rows[i]["base_name"], rows[j]["legal_name"]), ) # Fill both symmetric positions in the matrix sim_matrix[i, j] = score sim_matrix[j, i] = score return sim_matrix def cluster_from_similarity( rows: list[dict[str, Any]], sim_matrix: np.ndarray, state_flag: bool, threshold: int = 90, ) -> list[list[dict[str, Any]]]: """ Cluster payer records using a similarity matrix and threshold. Each row is grouped with other rows whose similarity score meets or exceeds the given threshold. Once a row is assigned to a cluster, it is not reconsidered. Args: rows (list[dict]): List of row dictionaries containing payer data. sim_matrix (np.ndarray): Pairwise similarity matrix (n x n). threshold (int): Minimum similarity score required to cluster. Returns: list[list[dict]]: A list of clusters, where each cluster is a list of row dictionaries. """ n = len(rows) visited: set[int] = set() clusters: list[list[dict[str, Any]]] = [] # Iterate through each record for i in range(n): # Skip rows that have already been clustered if i in visited: continue # Start a new cluster with the current index cluster_indices = {i} # Add all rows similar enough to the current one for j in range(n): if sim_matrix[i, j] >= threshold: # --- STATE LOGIC --- if state_flag: state_i = rows[i]["state"] state_j = rows[j]["state"] # If one has state and the other doesn't → do NOT cluster if state_i != state_j: continue cluster_indices.add(j) # Mark clustered indices as visited visited.update(cluster_indices) # Build the cluster from the original rows clusters.append([rows[idx] for idx in cluster_indices]) return clusters def cluster_with_blocking( rows: list[dict[str, Any]], state_flag: bool, threshold: int = 90, ) -> list[list[dict[str, Any]]]: """ Cluster payer names using first-token blocking to scale to large volumes. """ # 1. Group rows by block key blocks = defaultdict(list) for row in rows: blocks[row["block_key"]].append(row) all_clusters = [] # 2. Process each block independently for block_rows in blocks.values(): # Small blocks don't need matrix if len(block_rows) == 1: all_clusters.append(block_rows) continue sim_matrix = build_similarity_matrix(block_rows) block_clusters = cluster_from_similarity( block_rows, sim_matrix, state_flag=state_flag, threshold=threshold, ) all_clusters.extend(block_clusters) return all_clusters def propagate_dba_to_legal(rows: list[dict]) -> list[dict]: """ If a legal entity has an associated DBA in any row, propagate that DBA to all rows sharing the same legal entity. """ legal_to_dba = {} # First pass: build mapping for row in rows: legal = row.get("legal_name") dba = row.get("dba_name") if legal and dba: legal_to_dba[legal] = dba # Second pass: propagate for row in rows: legal = row.get("legal_name") if legal in legal_to_dba: new_base = legal_to_dba[legal] row["base_name"] = new_base # Update block key tokens = new_base.split() row["block_key"] = tokens[0] if tokens else "" return rows def add_aarete_derived_provider_name( df: pd.DataFrame, state_flag: bool ) -> pd.DataFrame: """ Normalize extracted AARETE_DERIVED_PROVIDER_NAME values into canonical form. """ df_copy = df.copy() # Check if AARETE_DERIVED_PROVIDER_NAME column exists, if not add empty column and return if "AARETE_DERIVED_PROVIDER_NAME" not in df_copy.columns: logging.warning( "AARETE_DERIVED_PROVIDER_NAME column not found in DataFrame. Adding empty column." ) df_copy["AARETE_DERIVED_PROVIDER_NAME"] = "" return df_copy # 1. Extract unique provider group names raw_values = df_copy["AARETE_DERIVED_PROVIDER_NAME"].dropna().tolist() flattened_names = [] for value in raw_values: if isinstance(value, list): flattened_names.extend(value) else: flattened_names.append(value) # Remove empties unique_names = list( {name for name in flattened_names if not string_utils.is_empty(name)} ) if not unique_names: df_copy["AARETE_DERIVED_PROVIDER_NAME"] = "" return df_copy # 2. Standardize standardized_rows = standardize_unique_names(unique_names, state_flag) # 3. DBA propagation (kept separate in case rules diverge later) standardized_rows = propagate_dba_to_legal(standardized_rows) # 4. Cluster with blocking clusters = cluster_with_blocking( standardized_rows, state_flag=state_flag, threshold=90, ) # 5. Build mapping mapping = {} for cluster in clusters: derived_name = prompt_calls.prompt_aarete_derived_provider_name( cluster, state_flag ) for record in cluster: mapping[record["original_name"]] = derived_name def map_value(value): if isinstance(value, list): derived_values = {mapping.get(v) for v in value if mapping.get(v)} # If all collapse to same canonical, return single string if len(derived_values) == 1: return derived_values.pop() # If multiple canonicals (rare), return list return list(derived_values) return mapping.get(value) df_copy["AARETE_DERIVED_PROVIDER_NAME"] = df_copy[ "AARETE_DERIVED_PROVIDER_NAME" ].apply(map_value) return df_copy