blocks-transformer/block.py

import pandas as pd
import json
import jmespath
import xgboost as xgb
import math
import joblib
import logging
from typing import Optional, List, Dict
from pre_processing import pre_processing
from processing import processing
from post_processing import post_processing

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s - %(message)s",
)
logger = logging.getLogger(__name__)


def __main__(application_key: str, application_timestamp: str, application_source_name: str, application_date_of_birth: str, ownhome: str, employmentstatus: str, lengthatjob: float, payfrequency: str, lengthatbank: str, inputipaddress: str, deviceid: str, fuzzydeviceid: str, trueip: str, dnsip: str, requestid: str, riskrating: str, tmxsummaryreasoncode: str, digitalidconfidence: str, results: Optional[List[Dict]] = None) -> dict:
    
    data = {
        "application_key": application_key,
        "application_timestamp": application_timestamp,
        "application_source_name": application_source_name,
        "application_date_of_birth": application_date_of_birth,
        "ownhome": ownhome,
        "employmentstatus": employmentstatus,
        "lengthatjob": lengthatjob,
        "payfrequency": payfrequency,
        "lengthatbank": lengthatbank,
        "inputipaddress": inputipaddress,
        "deviceid": deviceid,
        "fuzzydeviceid": fuzzydeviceid,
        "trueip": trueip,
        "dnsip": dnsip,
        "requestid": requestid,
        "riskrating": riskrating,
        "tmxsummaryreasoncode": tmxsummaryreasoncode,
        "digitalidconfidence": digitalidconfidence
    }

    data = pd.DataFrame([data])

    expected_schema = {
        "ea_score": str,                     
        "ip_net_speed_cell": str,            
        "ip_country_confidence": str,        
        "ip_region_confidence": str,         
        "fraud_risk": str,                   
        "first_seen_days": str,              
        "domain_creation_days": str          
    }

    expressions = {
        "ea_score": "EAScore",
        "ip_net_speed_cell": "ip_netSpeedCell",
        "ip_country_confidence": "ip_countryconf",
        "ip_region_confidence": "ip_regionconf",
        "fraud_risk": "fraudRisk",
        "first_seen_days": "first_seen_days",
        "domain_creation_days": "domain_creation_days",
    }


    if results:
        first_result = results[0]
        for column, expression in expressions.items():
            try:
                extracted_value = jmespath.search(expression, first_result)
                expected_type = expected_schema[column]

                if extracted_value is not None and not isinstance(extracted_value, expected_type):
                    try:
                        extracted_value = expected_type(extracted_value)
                    except (ValueError, TypeError) as cast_error:
                        logger.error(f"Failed to cast {column} value to {expected_type.__name__}: {cast_error}")
                        extracted_value = None
                
                data[column] = extracted_value
            except Exception as e:
                logger.error(f"Error extracting value for {column}: {e}")
                data[column] = None
    else:
        for column in expressions.keys():
            data[column] = None


    logger.info(f"pre_pre_processed_data: {data.to_dict(orient='records')}")

    pre_processed_data = pre_processing(data)

    logger.info(f"pre_processed_data: {pre_processed_data}")

    prediction = processing(pre_processed_data)

    logger.info("prediction: %.8f", float(prediction))

    result = post_processing(prediction[0])

    logger.info("Score: %.0f", float(result["score"]))
    logger.info("Action: %s", result["action"])
    logger.info("Description: %s", result["description"])

    return {
        # 'prediction': prediction,
        'score': result["score"],
        'action': result["action"],
        'description': result["description"]
    }