doczyai-pipelines/fieldProcessing/src/top_down_funcs.py

import prompts
import claude_funcs
import utils

import json

def run_top_down_metal_level(d, page):
    """
    Identifies and extracts the metal level of a contract within a specific line of business (LOB) from the provided page text.

    This function determines whether the contract's LOB is associated with marketplace or commercial sectors by examining the
    'CONTRACT_LOB' key in the dictionary. If relevant, it constructs and sends a specific prompt to a language model to extract the
    metal level (e.g., Bronze, Silver, Gold, Platinum). If the LOB is not relevant, it directly assigns 'N/A'.

    Parameters:
        d (dict): A dictionary containing extracted information from primary Top Down analysis, specifically the 'CONTRACT_LOB' key.
        page (str): The text content of the page that is being analyzed for metal level information.

    Returns:
        str: The metal level of the contract as determined by the analysis, or 'N/A' if the contract LOB is not applicable.
    """
    
    if 'MARKETPLACE' in str(d['CONTRACT_LOB']).upper() or 'COMMERCIAL' in str(d['CONTRACT_LOB']).upper():
        prompt = prompts.TOP_DOWN_METAL_LEVEL(d['CONTRACT_LOB'], page)
        answer = claude_funcs.invoke_claude_3(prompt, max_tokens=4000)
    else:
        answer = 'N/A'    
    return answer


def run_top_down_date(type_, d, page):
    """
    DEPRECATED - Extracts specific date-related information from a page of text using a Top Down processing approach.

    This function tailors the extraction to focus on either 'EFFECTIVE' or 'TERMINATION' dates by constructing
    a prompt that directs a language model to search for and interpret date information relevant to the provided type.
    It formats the input data to fit the processing needs, sends it along with the page text to the model, and captures
    the model's response.

    Parameters:
        type_ (str): Specifies the type of date to extract, 'EFFECTIVE' or 'TERMINATION'.
        d (dict): A dictionary containing preliminary data extracted from the text, which may include metadata like filename or page number.
        page (str): The text content of the page from which to extract the date.

    Returns:
        str: The extracted date as a string, based on the model's interpretation of the input prompt and text context.
    """
    formatted_d = utils.format_td_check([d], ['Filename', 'page_num'])
    prompt = prompts.TOP_DOWN_DATE('EFFECTIVE', formatted_d, page)
    answer = claude_funcs.invoke_claude_3(prompt, max_tokens=4000)
    return answer


def top_down_secondary(td_results, text_dict):
    """
    Conducts secondary processing on results obtained from the primary Top Down analysis of document text, enhancing detail and accuracy.

    This function iteratively enhances each dictionary result from the initial analysis by adding or refining information related to:
    - Market metal levels by running a specific function to determine the contract marketplace metal level.
    - Effective dates by analyzing the context around specific keywords in the text and extracting these dates.
    - Termination dates, similarly extracted based on the presence and context of specific keywords.

    Parameters:
        td_results (list of dict): Initial dictionaries from the primary Top Down processing containing basic extracted data.
        text_dict (dict): Dictionary keyed by page numbers, providing the full text for corresponding analysis and extraction tasks.

    Returns:
        list of dict: Updated list of dictionaries, each enhanced with additional detailed attributes like metal level and date specifics.
    """
    updated_dicts = []
    for d in td_results:
        # Run Metal Level
        d['CONTRACT_MARKETPLACE_METAL_LEVEL'] = run_top_down_metal_level(d, text_dict[d['page_num']])
        
        # Dates - Deprecated
        # d['LOB_PRICING_TERMS_EFFECTIVE_DATE'] = run_top_down_date('EFFECTIVE' , d, text_dict[d['page_num']])
        
        # if auto_renewal == N: else: 'N/A'
        # d['LOB_PRICING_TERMS_TERMINATION_DATE'] = run_top_down_date('TERMINATION', d, text_dict[d['page_num']])

        updated_dicts.append(d)
    return updated_dicts



def run_top_down(filename, text_dict):
    """
    Executes the Top Down processing strategy on a provided dictionary of text pages, extracting structured data based on specified prompts.

    The function first filters and processes each page of text that is correctly numbered, using a primary analysis prompt. The results
    from this analysis are then parsed into dictionaries, tagged with their respective page numbers and filename, and collected.
    After the initial processing, a secondary Top Down analysis is conducted to refine and possibly expand the extracted data.

    Parameters:
        filename (str): The filename associated with the text, used to tag the output data.
        text_dict (dict): A dictionary where keys are page numbers and values are the text on those pages.

    Returns:
        list of dict: A list of dictionaries where each dictionary contains structured results from both primary and secondary
                      Top Down analyses, associated with a specific page and the overall document.
    """

    all_results = []

    # Primary
    for page_num, page_text in text_dict.items():
        if not page_num.isdigit():
            continue
        prompt = prompts.TOP_DOWN_PRIMARY(page_text)
        answer = claude_funcs.invoke_claude_3(prompt, max_tokens=4000)
        answer_dict = json.loads(answer)
        answer_dict.update({'page_num' : page_num, 'Filename' : filename})
        #answer_dicts = dict_operations.primary_string_to_dict({page_num : answer}, filename) # Convert to dictionaries
        all_results.append(answer_dict)
        
    # td_primary = [i for d in all_results for i in d] # Consolidate to one list
    # print(td_primary)

    # Secondaries
    td_final = top_down_secondary(all_results, text_dict)

    return td_final # List of list of dictionaries