modules/mapping.py

import json
from rapidfuzz import process, fuzz

# Initialized Modules
from modules.normalization import normalization

# Mapping merchant
def mapping_merchant(information, json_path, normalization_rule):
    # Load from merchant JSON file
    with open(json_path, "r", encoding="utf-8") as f:
        name_id_dict = json.load(f)

    # Create cached normalized dictionary
    normalized_cached_map = {normalization(text= k, normalization_rule= normalization_rule): v for k, v in name_id_dict.items()}
    name_list = list(normalized_cached_map.keys())

    for item in information:
        # Match seller
        seller_name = normalization(text= item["seller"], normalization_rule= normalization_rule)
        seller_match, seller_score, _ = process.extractOne(seller_name, name_list, scorer= fuzz.token_sort_ratio)
        if seller_score >= 80:
            item["seller_id"] = normalized_cached_map[seller_match]
        else:
            item["seller_id"] = None

        # Match buyer
        buyer_name = normalization(text= item["buyer"], normalization_rule= normalization_rule)
        buyer_match, buyer_score, _ = process.extractOne(buyer_name, name_list, scorer= fuzz.token_sort_ratio)
        if buyer_score >= 80:
            item["buyer_id"] = normalized_cached_map[buyer_match]
        else:
            item["buyer_id"] = None
    
    return information

# Mapping unit
def mapping_unit(information, json_path, normalization_rule):
    # Load from unit JSON file
    with open(json_path, "r", encoding="utf-8") as f:
        unit_id_dict = json.load(f)

    # Create cached normalized dictionary
    normalized_cached_map = {normalization(text= k, normalization_rule= normalization_rule): v for k, v in unit_id_dict.items()}
    unit_list = list(normalized_cached_map.keys())

    for item in information:
        # Match unit
        unit = normalization(text= item["unit"], normalization_rule= normalization_rule)
        unit_match, unit_score, _ = process.extractOne(unit, unit_list, scorer= fuzz.token_sort_ratio)
        if unit_score >= 80:
            item["unit_id"] = normalized_cached_map[unit_match]
        else:
            item["unit_id"] = None
    
    return information

# Mapping employee
def mapping_employee(information, json_path, normalization_rule):
    # Load from employee JSON file
    with open(json_path, "r", encoding="utf-8") as f:
        employee_id_dict = json.load(f)

# Mapping product
def mapping_product(information, json_path, normalization_rule):
    # Load from product JSON file
    with open(json_path, "r", encoding="utf-8") as f:
        product_id_dict = json.load(f)