doczyai-pipelines/fieldExtraction/src/ac_funcs.py

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 claude_funcs
import dict_operations
import config


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 get_ac_answer_single_field_simple(prompt, filename, field):
    try:
        response_text = claude_funcs.invoke_claude(
            prompt, config.MODEL_ID_CLAUDE35_SONNET, filename, 8192
        )

        # Clean the response
        clean_response = response_text.strip()

        # Try to find a match within curly braces
        brace_match = re.search(r"\{([^}]+)\}", clean_response)
        if brace_match:
            brace_content = brace_match.group(1).strip()
            # Try to find the field within the braced content
            field_pattern = r'"{}"\s*:\s*"?([^"}}]+)"?'.format(re.escape(field))
            field_match = re.search(field_pattern, brace_content, re.IGNORECASE)
            if field_match:
                return {field: field_match.group(1).strip()}

        # If no match in braces, try to find field in the format "Field: Value"
        field_match = re.search(
            r"{}: (.+?)(?:\n|$)".format(re.escape(field)), clean_response, re.IGNORECASE
        )
        if field_match:
            return {field: field_match.group(1).strip()}

        # Try to find field in the format "Field Value"
        field_match = re.search(
            r"{}\s+(.+?)(?:\n|$)".format(re.escape(field)),
            clean_response,
            re.IGNORECASE,
        )
        if field_match:
            return {field: field_match.group(1).strip()}

        # If no specific format is found, return the entire response
        return {field: clean_response}

    except Exception as e:
        print(
            f"Error in get_ac_answer_single_field_simple for {filename} - {field}: {str(e)}"
        )
        return {field: f"Error: {str(e)}"}


def get_filename_from_path(full_path):
    return os.path.basename(full_path)


def select_context(text_dict, field):
    if (field == "RELATIONSHIP_OF_PARTIES_IND") or (
        field == "RELATIONSHIP_OF_PARTIES_LANGUAGE"
    ):
        keywords = ["Relationship of Parties"]
    if field == "REGULATORY_REQUIREMENTS":
        keywords = ["Complience with Regulatory Requirements"]
    if (field == "DISPARAGEMENT_PROHIBITION_IND") or (
        field == "DISPARAGEMENT_PROHIBITION_LANGUAGE"
    ):
        keywords = ["Disparagement Prohibition"]
    if field == "ELIGIBILITY_VERIFICATION":
        keywords = ["Eligibility Determination"]
    if field == "INSURANCE_REQUIREMENT":
        keywords = ["Insurance"]
    if field == "PARTICIPATION_IN_PRODUCTS":
        keywords = ["Participation in Products"]
    if field == "POLICIES_AND_PROCEDURES":
        keywords = ["Provider Manual", "Policies and Procedures"]
    if field == "ARBITRATION_AND_DISPUTES":
        keywords = ["Informal Dispute Resolution", "Arbitration"]
    print(field)
    page_list = []
    print(keywords)
    for keyword in keywords:
        for page in text_dict:
            if keyword.lower() in text_dict[page].lower():
                if page > 1:
                    if page < len(text_dict):
                        page_list.extend([page - 1, page, page + 1])
                    else:
                        page_list.extend([page - 1, page])
                else:
                    page_list.extend([page, page + 1])

    page_list = list(set(page_list))
    context = [text_dict[page] for page in page_list]
    context = "\n".join(context)

    return context, page_list


def create_prompt(context, question):
    return f"""
    
    Human: Use the following pieces of context to provide a concise answer to the questions at the end. You must answer in correct JSON format.
    
    {context}
    
    Question: {question}
    
    Assistant: Answer in JSON format: {{"""


def json_parsing(response_text, context):
    print("Parsing JSON response.")
    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(",") + "}"

    try:
        response_dict = json.loads(response_text)
    except:
        response_dict = {"Test field": response_text}

    field_l = list(response_dict.keys())
    answer_l = list(response_dict.values())
    field_dict = {k: v for k, v in response_dict.items() if not k.endswith("_PG")}
    page_dict = {k: v for k, v in response_dict.items() if k.endswith("_PG")}
    page_dict = {k[:-3]: v for k, v in response_dict.items()}
    field_l = list(field_dict.keys())
    answer_l = list(field_dict.values())
    page_no_l = [page_dict.get(x, "") for x in field_l]

    answer_for_snippet = [
        a.lower() if isinstance(a, str) else str(a) if isinstance(a, int) else ""
        for a in answer_l
    ]

    answer_for_snippet = [
        (
            a[:4]
            if "_DT" in f
            else (
                "renew"
                if f
                in ["CONTRACT_AUTO_RENEWAL_IND", "CONTRACT_AUTO_RENEWAL_TERM_LENGTH"]
                else (
                    "signature"
                    if f == "CONTRACT_SIGNATORY_IND"
                    else (
                        "hospital"
                        if f == "CONTRACT_TYPE"
                        else (
                            a[:10]
                            if f in ["PROV_NPI_OTHER", "PROV_TIN_OTHER"]
                            else (
                                "participat"
                                if f == "PROV_PARTICIPATION_STATUS"
                                else (
                                    "master"
                                    if f == "CONTRACT_CHARGEMASTER_PROTECTION_LANGUAGE"
                                    else a
                                )
                            )
                        )
                    )
                )
            )
        )
        for a, f in zip(answer_for_snippet, field_l)
    ]

    try:
        location_l = [
            (
                context.lower().find(a, context.find("Start of Page No. = " + str(p)))
                if isinstance(a, str) and a != ""
                else -1
            )
            for a, p in zip(answer_for_snippet, page_no_l)
        ]
    except:
        location_l = [
            (
                context.lower().find(answer)
                if isinstance(answer, str) and answer != ""
                else -1
            )
            for answer in answer_for_snippet
        ]

    words_before = [
        " ".join(context[:location].split()[-25:]) if location != -1 else " "
        for location in location_l
    ]
    words_after = [
        " ".join(context[location:].split()[:30]) if location != -1 else " "
        for location in location_l
    ]
    snippet_l = [
        (
            " ".join(context[: location - len(wb) - 1].split(".")[-1:])
            + " "
            + wb
            + " "
            + wa
            + " "
            + " ".join(context[location + len(wa) + 1 :].split(".")[:1])
            + "."
            if location != -1
            else " "
        )
        for location, wb, wa in zip(location_l, words_before, words_after)
    ]
    snippet_l = [snippet.strip() for snippet in snippet_l]

    print(f"Parsed fields: {field_l}")
    return field_l, answer_l, page_no_l, location_l, snippet_l


def json_parsing_search(response_text, field_list):
    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))