doczyai-pipelines/fieldExtraction/src/investment/aarete_derived.py

import logging
import os
from collections import defaultdict

import pandas as pd
import src.config as config
import src.utils.string_utils as string_utils
from constants.constants import Constants
from constants.delimiters import Delimiter
from src.crosswalk.crosswalk_builder import CrosswalkBuilder
from src.prompts.fieldset import FieldSet
from src.utils.crosswalk_utils import apply_crosswalk


def fill_na_from_field(results_dict, to_field, from_field, crosswalk_path):
    from_str = results_dict.get(from_field)
    if string_utils.is_empty(from_str):
        return results_dict.get(to_field)
    MAPPING_DICT = CrosswalkBuilder().from_json(crosswalk_path).mapping
    to_value = apply_crosswalk(from_str, MAPPING_DICT)
    return to_value if to_value else results_dict.get(to_field)


def fill_na_mapping(answer_dicts):
    """
    Fill in missing values in fields using mappings from other fields.

    Args:
        results_df (pd.DataFrame): DataFrame containing the results data

    Returns:
        pd.DataFrame: DataFrame with filled missing values
    """
    for answer_dict in answer_dicts:
        # Fill AARETE_DERIVED_LOB from AARETE_DERIVED_PROGRAM
        if string_utils.is_empty(answer_dict.get("AARETE_DERIVED_LOB")):
            filled_value = fill_na_from_field(
                answer_dict,
                "AARETE_DERIVED_LOB",
                "AARETE_DERIVED_PROGRAM",
                "constants/mappings/crosswalk_program_lob.json",
            )

            answer_dict["AARETE_DERIVED_LOB"] = filled_value
            answer_dict["LOB_PROGRAM_RELATIONSHIP"] = "Exclusive"

        # Fill AARETE_DERIVED_LOB from PRODUCT
        if string_utils.is_empty(answer_dict.get("AARETE_DERIVED_LOB")):
            filled_value = fill_na_from_field(
                answer_dict,
                "AARETE_DERIVED_LOB",
                "PRODUCT",
                "constants/mappings/crosswalk_product_lob.json",
            )
            answer_dict["AARETE_DERIVED_LOB"] = filled_value
            answer_dict["LOB_PRODUCT_RELATIONSHIP"] = "Exclusive"
    return answer_dicts


def get_crosswalk_fields(answer_dicts: list, constants: Constants):
    crosswalk_fields = FieldSet(file_path=config.FIELD_JSON_PATH, crosswalk=True)
    for to_field in crosswalk_fields.fields:
        to_field_name, from_field_name = to_field.field_name, to_field.base_field
        # Find crosswalk
        crosswalk = constants.get_constant(to_field.crosswalk)
        if crosswalk:
            for answer_dict in answer_dicts:
                to_field_value, from_field_value = answer_dict.get(
                    to_field_name
                ), answer_dict.get(from_field_name)
                # If from_field_value is populated and to_field_value is not populated, perform mapping
                if not string_utils.is_empty(
                    from_field_value
                ) and string_utils.is_empty(to_field_value):
                    from_field_value_list = from_field_value.split("|")
                    to_field_answer_list = []
                    for individual_from_field_value in from_field_value_list:
                        if individual_from_field_value in crosswalk.mapping.keys():
                            to_field_answer_list.append(crosswalk.mapping.get(
                                individual_from_field_value
                            ))
                        elif from_field_value in crosswalk.mapping.values():
                            if "AARETE_DERIVED" in to_field_name:
                                to_field_answer_list.append(individual_from_field_value)
                            else:
                                to_field_answer_list.append(crosswalk.create_reverse_mapping().get(individual_from_field_value))
                    answer_dict[to_field_name] = "|".join(to_field_answer_list)

    return answer_dicts