src/info_processing.py

import pandas as pd
import numpy as np
import re
# Expected JSON schema

expected_schema = {
    'citizenshipCountry': '',
    'countryResidency': '',
    'applyingFrom': '',
    'age': np.nan,
    'gender': '',
    'visaPurpose': '',
    'activeVisa': "",
    'visaList':[{'visaName': '','visaCountry': '','visaCategory': '','visaType': ''}],
    'educationList': [{'level': '', 'degree': ''}],
    'certification': '',
    'certificationList': [{'name': '', 'description': ''}],
    'admissionInUniversity': '',
    'uniName': '',
    'uniLocation': '',
    'workList': [{'occupation': '', 'experience': np.nan, 'designation': ''}],
    'entrepreneurBusinessExp': '',
    'entrepreneurBusinessDesc': '',
    'jobOffer': '',
    'jobDescription': '',
    'businessOwner': '',
    'businessCurrentlyActive': '',
    'businessType': '',
    'langugesList': [{'language': '', 'language': ''}],
    'proficiency': '',
    'additionalLanguages': '',
    'proficienyTestsList': [{'language': '', 'test': '', 'label': '', 'score': '', 'subTest': ''}],
    'maritalStatus': '',
    'relativeDetails': [{'relativeType': '', 'relativeVisa': '', 'relativeCountry': ''}],
    'relatives': '',
    'otherRelativeAbroad': '',
    'otherRelativeType': '',
    'otherRelativeVisa': '',
    'otherRelativeLocation': '',
    'criminalRecord': '',
    'criminalRecordList': [],
    'militaryService': '',
    'militaryServiceType': '',
    'militaryServiceCountry': '',
    'immigrated': '',
    'immigrationDesc': '',
    'immigratedCountry': '',
    'deportStatus': '',
    'deportDesc': '',
    'overStayed': '',
    'appliedAsylum': '',
    'asylumResult': '',
    'asylumAttemptCountry': '',
    'relatives': '',
    'refugeeStatus': '',
    'liquidAssets': '',
    'totalInvestment': '',
    'totalAssets': '',
    'willingInvestment': '',
    'medicalExamination': '',
    'medicalDisorders': '',
    'noncontagiousDisease': [],
    'contagiousDisease': [],
    'disablities': [],
    'companyInvitation': '',
    'invitationPurpose': '',
    'invitationCountry': '',
    'specializedSkill': '',
    'skill': '',
    'innovation': '',
    'innovationDescription': '',
    'sponsorAvailable': '',
    'sponserShipvalue': '',
    'sponserDes': '',
    'anyAwards': '',
    'awards': [{'name': '', 'description': ''}],
    'otherAchivement': '',
    'email': ''
}

# Flattens a list of dictionaries into a single dictionary with keys suffixed by an index
def flatten_list_of_dicts(list_of_dicts, key_map, max_count):
    result = {}
    for i in range(max_count):
        if i < len(list_of_dicts):
            if not isinstance(list_of_dicts[i], dict):
                raise TypeError(f"Expected dict but got {type(list_of_dicts[i])} for item {i} in {list_of_dicts}")
            for original_key, new_key in key_map.items():
                result[f"{new_key}{i+1}"] = list_of_dicts[i].get(original_key, '')
        else:
            for new_key in key_map.values():
                result[f"{new_key}{i+1}"] = ''
    return result

# Concatenates list values into a single string separated by commas
def concatenate_list_values(values_list):
    return ', '.join(values_list)

# Processes the proficiency tests list and returns a dictionary where the test names are the keys and the scores are the values
def process_proficiency_tests(proficiency_tests_list):
    result = {}
    for test_result in proficiency_tests_list:
        if 'test' in test_result and 'score' in test_result:
            # Clean up the test name by removing spaces, parentheses, and content inside parentheses, trailing underscores, and colons
            clean_test_name = re.sub(r'\s+', '_', test_result['label'])
            clean_test_name = re.sub(r'\(.*?\)', '', clean_test_name)
            clean_test_name = clean_test_name.replace('(', '').replace(')', '').replace(':', '').replace('__', '_').strip('_')
            # Use the cleaned test name as the key and the score as the value in the result dictionary
            result[clean_test_name] = test_result['score']
    return result

# Combines the medical conditions into a single string
def combine_medical_conditions(item):
    combined_values = []
    for condition_key in ['noncontagiousDisease', 'contagiousDisease', 'disablities']:
        combined_values.extend(item.get(condition_key, []))
    return concatenate_list_values(combined_values)

# Processes the relative details list including the Other relative info, and returns a concatenated string
def process_relative_details(item):

    # Process the main relative details
    relative_details = '; '.join([
        f"{d.get('relativeType', '')}, {d.get('relativeVisa', '')}, {d.get('relativeCountry', '')}"
        for d in item.get('relativeDetails', [])
    ]).strip('; ')

    # Process additional relative details if all required fields are present
    if item.get('otherRelativeType') and item.get('otherRelativeVisa') and item.get('otherRelativeLocation'):
        additional_relative = f"{item['otherRelativeType']}, {item['otherRelativeVisa']}, {item['otherRelativeLocation']}"
        relative_details += f"; {additional_relative}" if relative_details else additional_relative

    return relative_details

def default_processing(value, default_value):
    return value if value is not None else default_value

def clean_dataframe(df):
    # Replace empty strings with NaN
    df.replace("", np.nan, inplace=True)
    # Replace missing values with 'N/A'
    df.fillna('N/A', inplace=True)
    return df

def convert_json_to_dataframe(json_data):
    data_list = []

    # Mapping special processing functions to keys
    special_processing_map = {
        'visaList': lambda v: flatten_list_of_dicts(v, {'visaName': 'visaName', 'visaCountry': 'visaCountry'}, 4),
        'educationList': lambda v: flatten_list_of_dicts(v, {'level': 'eduLevel', 'degree': 'eduDegree'}, 4),
        'workList': lambda v: flatten_list_of_dicts(v, {'occupation': 'occupation', 'experience': 'experience', 'designation': 'designation'}, 4),
        'certificationList': lambda v: flatten_list_of_dicts(v, {'name': 'certName', 'description': 'certDesc'}, 4),
        'langugesList': lambda v: flatten_list_of_dicts(v, {'language': 'language', 'level': 'languageLevel'}, 4),
        'awards': lambda v: flatten_list_of_dicts(v, {'name': 'awardsName', 'description': 'awardsDesc'}, 4),
        'proficienyTestsList': process_proficiency_tests,
        'relativeDetails': process_relative_details,
        'medicalConditions': combine_medical_conditions
    }

    for item in json_data:
        processed_item = {}
        for key in expected_schema:
            value = item.get(key, expected_schema[key])
            if key in special_processing_map:
                # Process keys requiring full-item processing
                if key in ['relativeDetails', 'medicalConditions']:
                    processed_item[key] = special_processing_map[key](item)
                else:
                    # Process directly using the value
                    processed_item.update(special_processing_map[key](value))
            elif key not in ['noncontagiousDisease', 'contagiousDisease', 'disablities']:
                processed_item[key] = default_processing(value, expected_schema[key])

        # Add combined medical conditions
        processed_item['medicalConditions'] = combine_medical_conditions(item)
        data_list.append(processed_item)

    df = pd.DataFrame(data_list)
    return clean_dataframe(df)