matching_dependency/ml_er/Goods Dataset.py

import sys

from py_entitymatching.debugmatcher.debug_gui_utils import _get_metric

sys.path.append('/home/w/PycharmProjects/py_entitymatching/py_entitymatching')

import py_entitymatching as em
import py_entitymatching.catalog.catalog_manager as cm
import pandas as pd
import time
import six
from md_discovery.functions.multi_process_infer_by_pairs import my_Levenshtein_ratio


def process_prediction_for_md_discovery(pred: pd.DataFrame, tp_single_tuple_path: str = "output/tp_single_tuple.csv", fn_single_tuple_path: str = "output/fn_single_tuple.csv"):
    tp = pred[(pred['gold'] == 1) & (pred['predicted'] == 1)]
    fn = pred[(pred['gold'] == 1) & (pred['predicted'] == 0)]
    # 将真阳/假阴表中左右ID调整一致
    for index, row in tp.iterrows():
        tp.loc[index, "rtable_id"] = row["ltable_id"]
    for index, row in fn.iterrows():
        fn.loc[index, "rtable_id"] = row["ltable_id"]

    pred_columns = pred.columns.values.tolist()
    l_columns = []
    r_columns = []
    columns = []  # todo 前提是两张表内对应字段名调整一致
    for _ in pred_columns:
        if _.startswith('ltable'):
            l_columns.append(_)
        elif _.startswith('rtable'):
            r_columns.append(_)
    for _ in l_columns:
        columns.append(_.replace('ltable_', ''))

    tpl = tp[l_columns]
    tpr = tp[r_columns]
    tpl.columns = columns
    tpr.columns = columns

    fnl = fn[l_columns]
    fnr = fn[r_columns]
    fnl.columns = columns
    fnr.columns = columns

    tp_single_tuple = pd.concat([tpl, tpr])
    fn_single_tuple = pd.concat([fnl, fnr])

    tp_single_tuple.to_csv(tp_single_tuple_path, sep=',', index=False, header=True)
    fn_single_tuple.to_csv(fn_single_tuple_path, sep=',', index=False, header=True)


def evaluate_prediction(df: pd.DataFrame, labeled_attr: str, predicted_attr: str, matching_number: int,
                        test_proportion: float) -> dict:
    new_df = df.reset_index(drop=False, inplace=False)
    gold = new_df[labeled_attr]
    predicted = new_df[predicted_attr]
    gold_negative = gold[gold == 0].index.values
    gold_positive = gold[gold == 1].index.values
    predicted_negative = predicted[predicted == 0].index.values
    predicted_positive = predicted[predicted == 1].index.values

    false_positive_indices = list(set(gold_negative).intersection(predicted_positive))
    true_positive_indices = list(set(gold_positive).intersection(predicted_positive))
    false_negative_indices = list(set(gold_positive).intersection(predicted_negative))

    num_true_positives = float(len(true_positive_indices))
    num_false_positives = float(len(false_positive_indices))
    num_false_negatives = float(len(false_negative_indices))

    precision_denominator = num_true_positives + num_false_positives
    recall_denominator = num_true_positives + num_false_negatives

    precision = 0.0 if precision_denominator == 0.0 else num_true_positives / precision_denominator
    recall = 0.0 if recall_denominator == 0.0 else num_true_positives / recall_denominator
    F1 = 0.0 if precision == 0.0 and recall == 0.0 else (2.0 * precision * recall) / (precision + recall)
    my_recall = num_true_positives / (matching_number * test_proportion)

    return {"precision": precision, "recall": recall, "F1": F1, "my_recall": my_recall}


def load_mds(paths: list) -> list:
    if len(paths) == 0:
        return []
    all_mds = []
    # 传入md路径列表
    for md_path in paths:
        mds = []
        # 打开每一个md文件
        with open(md_path, 'r') as f:
            # 读取每一行的md，加入该文件的md列表
            for line in f.readlines():
                md_metadata = line.strip().split('\t')
                md = eval(md_metadata[0].replace('md:', ''))
                mds.append(md)
        all_mds.extend(mds)
    return all_mds


def is_explicable(row, all_mds: list) -> bool:
    attrs = all_mds[0].keys()  # 从第一条md中读取所有字段
    for md in all_mds:
        explicable = True  # 假设这条md能解释当前元组
        for a in attrs:
            threshold = md[a]
            if my_Levenshtein_ratio(str(getattr(row, 'ltable_'+a)), str(getattr(row, 'rtable_'+a))) < threshold:
                explicable = False  # 任意一个字段的相似度达不到阈值，这条md就不能解释当前元组
                break  # 不再与当前md的其他相似度阈值比较，跳转到下一条md
        if explicable:
            return True  # 任意一条md能解释，直接返回
    return False  # 遍历结束，不能解释


def load_data(left_path: str, right_path: str, mapping_path: str):
    left = pd.read_csv(left_path, encoding='ISO-8859-1')
    cm.set_key(left, left.columns.values.tolist()[0])
    left.fillna("", inplace=True)
    left = left.astype(str)

    right = pd.read_csv(right_path, encoding='ISO-8859-1')
    cm.set_key(right, right.columns.values.tolist()[0])
    right.fillna("", inplace=True)
    right = right.astype(str)

    mapping = pd.read_csv(mapping_path)
    mapping = mapping.astype(str)
    return left, right, mapping


if __name__ == '__main__':
    # 读入公开数据，注册并填充空值
    path_Amazon = '/home/w/PycharmProjects/py_entitymatching/py_entitymatching/datasets/end-to-end/Amazon-GoogleProducts/Amazon.csv'
    path_Google = '/home/w/PycharmProjects/py_entitymatching/py_entitymatching/datasets/end-to-end/Amazon-GoogleProducts/GoogleProducts.csv'
    path_Mappings = '/home/w/PycharmProjects/py_entitymatching/py_entitymatching/datasets/end-to-end/Amazon-GoogleProducts/Amzon_GoogleProducts_perfectMapping.csv'
    Amazon = pd.read_csv(path_Amazon, encoding='ISO-8859-1')
    cm.set_key(Amazon, 'id')
    Amazon.fillna("", inplace=True)
    Google = pd.read_csv(path_Google, encoding='ISO-8859-1')
    cm.set_key(Google, 'id')
    Google.fillna("", inplace=True)
    Mappings = pd.read_csv(path_Mappings)

    # 仅保留两表中出现在映射表中的行，增大正样本比例
    l_id_list = []
    r_id_list = []
    # 全部转为字符串
    Amazon = Amazon.astype(str)
    Google = Google.astype(str)
    Mappings = Mappings.astype(str)
    matching_number = len(Mappings)  # 所有阳性样本数，商品数据集应为1300

    for index, row in Mappings.iterrows():
        l_id_list.append(row["idAmazon"])
        r_id_list.append(row["idGoogleBase"])
    selected_Amazon = Amazon[Amazon['id'].isin(l_id_list)]
    selected_Amazon = selected_Amazon.rename(columns={'title': 'name'})
    selected_Google = Google[Google['id'].isin(r_id_list)]
    cm.set_key(selected_Amazon, 'id')
    cm.set_key(selected_Google, 'id')

    # block 并将gold标记为0
    blocker = em.OverlapBlocker()
    candidate = blocker.block_tables(selected_Amazon, selected_Google, 'name', 'name',
                                     l_output_attrs=['id', 'name', 'description', 'manufacturer', 'price'],
                                     r_output_attrs=['id', 'name', 'description', 'manufacturer', 'price'],
                                     overlap_size=1, show_progress=False)
    candidate['gold'] = 0

    start = time.time()
    candidate_match_rows = []
    for index, row in candidate.iterrows():
        l_id = row["ltable_id"]
        map_row = Mappings[Mappings['idAmazon'] == l_id]

        if map_row is not None:
            r_id = map_row["idGoogleBase"]
            for value in r_id:
                if value == row["rtable_id"]:
                    candidate_match_rows.append(row["_id"])
        else:
            continue
    for row in candidate_match_rows:
        candidate.loc[row, 'gold'] = 1

    # 裁剪负样本，保持正负样本数量一致
    candidate_mismatch = candidate[candidate['gold'] == 0]
    candidate_match = candidate[candidate['gold'] == 1]
    candidate_mismatch = candidate_mismatch.sample(n=len(candidate_match))
    # 拼接正负样本
    candidate_for_train_test = pd.concat([candidate_mismatch, candidate_match])
    cm.set_key(candidate_for_train_test, '_id')
    cm.set_fk_ltable(candidate_for_train_test, 'ltable_id')
    cm.set_fk_rtable(candidate_for_train_test, 'rtable_id')
    cm.set_ltable(candidate_for_train_test, selected_Amazon)
    cm.set_rtable(candidate_for_train_test, selected_Google)

    # 分为训练测试集
    train_proportion = 0.7
    test_proportion = 0.3
    sets = em.split_train_test(candidate_for_train_test, train_proportion=0.7, random_state=0)
    train_set = sets['train']
    test_set = sets['test']

    dt = em.DTMatcher(name='DecisionTree', random_state=0)
    svm = em.SVMMatcher(name='SVM', random_state=0)
    rf = em.RFMatcher(name='RF', random_state=0)
    lg = em.LogRegMatcher(name='LogReg', random_state=0)
    ln = em.LinRegMatcher(name='LinReg')
    nb = em.NBMatcher(name='NaiveBayes')
    feature_table = em.get_features_for_matching(selected_Amazon, selected_Google, validate_inferred_attr_types=False)

    train_feature_vecs = em.extract_feature_vecs(train_set,
                                                 feature_table=feature_table,
                                                 attrs_after=['gold'],
                                                 show_progress=False)

    result = em.select_matcher([dt, rf, svm, ln, lg, nb], table=train_feature_vecs,
                               exclude_attrs=['_id', 'ltable_id', 'rtable_id', 'gold'],
                               k=5,
                               target_attr='gold', metric_to_select_matcher='f1', random_state=0)

    test_feature_vecs = em.extract_feature_vecs(test_set, feature_table=feature_table,
                                                attrs_after=['ltable_name', 'ltable_description', 'ltable_manufacturer',
                                                             'ltable_price', 'rtable_name', 'rtable_description',
                                                             'rtable_manufacturer', 'rtable_price', 'gold'],
                                                show_progress=False)

    rf.fit(table=train_feature_vecs,
           exclude_attrs=['_id', 'ltable_id', 'rtable_id', 'gold'],
           target_attr='gold')
    predictions = rf.predict(table=test_feature_vecs, exclude_attrs=['_id', 'ltable_id', 'rtable_id', 'ltable_name',
                                                                     'ltable_description', 'ltable_manufacturer',
                                                                     'ltable_price', 'rtable_name',
                                                                     'rtable_description',
                                                                     'rtable_manufacturer', 'rtable_price', 'gold'],
                             append=True, target_attr='predicted', inplace=False)
    eval_result = em.eval_matches(predictions, 'gold', 'predicted')
    em.print_eval_summary(eval_result)
    indicators = evaluate_prediction(predictions, 'gold', 'predicted', matching_number, test_proportion)
    print(indicators)

    # 计算可解释性
    ################################################################################################################
    predictions = predictions[
        ['ltable_id', 'rtable_id', 'ltable_name', 'ltable_description', 'ltable_manufacturer', 'ltable_price',
         'rtable_name', 'rtable_description', 'rtable_manufacturer', 'rtable_price', 'gold', 'predicted']]
    epl_match = 0  # 可解释，预测match
    nepl_mismatch = 0  # 不可解释，预测mismatch
    p_md = "/home/w/A-New Folder/8.14/Goods Dataset/TP_md_list.txt"
    p_vio = "/home/w/A-New Folder/8.14/Goods Dataset/TP_vio_list.txt"
    md_paths: list = [p_md, p_vio]
    md_list = load_mds(md_paths)  # 从全局变量中读取所有的md
    for row in predictions.itertuples():
        if is_explicable(row, md_list):
            if getattr(row, 'predicted') == 1:
                epl_match += 1
        else:
            if getattr(row, 'predicted') == 0:
                nepl_mismatch += 1

    epl_ability = (epl_match + nepl_mismatch) / len(predictions)
    ################################################################################################################
    process_prediction_for_md_discovery(predictions)
    # todo 将prediction表处理成真阳/假阴表提供给挖掘算法


    output_path = "output/eval_result" + str(time.time()) + ".txt"
    with open(output_path, 'w') as f:
        for key, value in six.iteritems(_get_metric(eval_result)):
            f.write(key + " : " + value)
            f.write('\n')
        f.write('my_recall:' + str(indicators["my_recall"]))
        f.write('\n')