blocks-transformer/block.py

import logging
from typing import Dict
from pre_processing import pre_processing
from processing import processing
from post_processing import post_processing
from parse_report import extract_model_variables

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


def __main__(application_id: str, creditBureau: dict) -> Dict:
    """
    Process a single application record via the scoring pipeline.
    Returns:
        A dict containing:
          - prediction: float
          - grade: str
          - reason_codes: list or None
    """
    record = extract_model_variables(creditBureau)

    if not record or all(v is None for v in record.values()):
        final_result = {
            'application_id': application_id,
            'prediction':        0.99,
            'grade':             'M14',
            "reason_codes": [{
                "code": None,
                "rank": None,
                "description": "Lack of account information"
            }]
        }
        logger.info(
            f"final_result (early exit due to missing or empty extracted variables): {final_result}")
        return final_result

    processed = pre_processing(record)

    out = processing(processed)

    final = post_processing(out, record)

    final_result = {
        'application_id': application_id,
        'prediction':        out['prediction'],
        'grade':             final['grade'],
        'reason_codes': final['reason_codes'],
    }

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

    return final_result


if __name__ == "__main__":
    import json
    import sys
    with open(sys.argv[1]) as f:
        data = json.load(f)
    __main__(application_id=data["application_id"], creditBureau=data)