Deep_match_rank_emb/match_recall/preprocess_tzld210322_gen.py

import random
import gc
import numpy as np
import pandas as pd
from tqdm import tqdm
from tensorflow.python.keras.preprocessing.sequence import pad_sequences

import time
import threadpool
from concurrent.futures import ThreadPoolExecutor, wait, ALL_COMPLETED, FIRST_COMPLETED, as_completed

import multiprocessing

def gen_data_set_bak(data, negsample=0):

    data.sort_values("logtimestamp", inplace=True)
    label_list = list(data["label"])
    label_list_b = []

    item_ids = data['videoid'].unique()
    print(item_ids)

    user_ids = data['mid'].unique()
    print("\n\nuser size is:  ", len(user_ids))

    print("********* pre data len is:  " + str(len(list(data["label"]))))
    print(data["label"].mean())
    data = data[data["label"] > 0].copy()

    user_ids = data['mid'].unique()

    train_set = []
    test_set = []

    print("pre data len is:  " + str(len(list(data["label"]))))

    for reviewerID, hist in tqdm(data.groupby('mid')):
        print("\n\nreviewerID :   ", reviewerID)
        print("\n\nhist  :  \n", hist)
        # pos_list = hist['movie_id'].tolist()
        pos_list = hist['videoid'].tolist()

        rating_list = hist['rating'].tolist()

        if negsample > 0:
            candidate_set = list(set(item_ids) - set(pos_list))
            neg_list = np.random.choice(candidate_set, size=len(pos_list)*negsample,replace=True)
        for i in range(1, len(pos_list)):
            hist = pos_list[:i]
            if i != len(pos_list) - 1:
                train_set.append((reviewerID, hist[::-1], pos_list[i], 1, len(hist[::-1]),rating_list[i]))
                # print("hist[::-1] is:  ")
                # print(hist[::-1])
                for negi in range(negsample):
                    train_set.append((reviewerID, hist[::-1], neg_list[i*negsample+negi], 0,len(hist[::-1])))
            else:
                test_set.append((reviewerID, hist[::-1], pos_list[i],1,len(hist[::-1]),rating_list[i]))

    random.shuffle(train_set)
    random.shuffle(test_set)

    print(len(train_set[0]), len(test_set[0]))
    print(len(train_set[0]))
    print(len(train_set), len(test_set))

    return train_set, test_set

all_task = list()

# pool = multiprocessing.Pool(processes=3)

def thread_func1(train_set, test_set, negsample, item_ids, reviewerID, hist):
    begin_time = time.time()

    print("\n\nreviewerID :   ", reviewerID)
    pos_list = hist['videoid'].tolist()
    rating_list = hist['rating'].tolist()

    if negsample > 0:
        candidate_set = list(set(item_ids) - set(pos_list))
        neg_list = np.random.choice(candidate_set, size=len(pos_list) * negsample, replace=True)
    for i in range(1, len(pos_list)):
        hist = pos_list[:i]
        if i != len(pos_list) - 1:
            train_set.append((reviewerID, hist[::-1], pos_list[i], 1, len(hist[::-1]), rating_list[i]))
            # print("hist[::-1] is:  ")
            # print(hist[::-1])
            for negi in range(negsample):
                train_set.append((reviewerID, hist[::-1], neg_list[i * negsample + negi], 0, len(hist[::-1])))
        else:
            test_set.append((reviewerID, hist[::-1], pos_list[i], 1, len(hist[::-1]), rating_list[i]))
    print(str(reviewerID) + "  idx process func cost time:  " + str(time.time() - begin_time))

    return train_set, test_set



def thread_func2(negsample, item_ids, reviewerID, hist):
    begin_time = time.time()
    train_set = []
    test_set = []
    print("\n\nreviewerID :   ", reviewerID)

    pos_list = hist['videoid'].tolist()

    rating_list = hist['rating'].tolist()

    if negsample > 0:
        candidate_set = list(set(item_ids) - set(pos_list))
        neg_list = np.random.choice(candidate_set, size=len(pos_list) * negsample, replace=True)
    for i in range(1, len(pos_list)):
        hist = pos_list[:i]
        print("hist[] is:  ")
        print(hist)
        if i != len(pos_list) - 1:
            train_set.append((reviewerID, hist[::-1], pos_list[i], 1, len(hist[::-1]), rating_list[i]))
            print("hist[::-1] is:  ")
            print(hist[::-1])
            for negi in range(negsample):
                train_set.append((reviewerID, hist[::-1], neg_list[i * negsample + negi], 0, len(hist[::-1])))
        else:
            test_set.append((reviewerID, hist[::-1], pos_list[i], 1, len(hist[::-1]), rating_list[i]))
    print(str(reviewerID) + "  idx process func cost time:  " + str(time.time() - begin_time))

    return train_set, test_set


def gen_data_set(train_path="", test_path="", user_mid_uid=None):

    train_set2 = pd.read_csv(train_path)
    test_set2 = pd.read_csv(test_path)
    train_set2 = train_set2[train_set2["mid"] != "unknown"].copy()

    train_pd = pd.merge(left=train_set2,
                        right=user_mid_uid,
                        left_on="mid",
                        right_on="uidx",
                        # how="left")
                        how="inner")

    # train_pd = train_pd.sample(frac=0.2)
    test_pd = pd.merge(left=test_set2,
                         right=user_mid_uid,
                        left_on="mid",
                        right_on="uidx",
                         # how="left")
                        how="inner")

    #train_pd = train_pd.sample(frac=0.3)
    train_pd = train_pd.sample(frac=0.6)
    print("after sample train set len is:  ")
    print(len(train_pd), len(test_pd))

    return train_pd, test_pd

def gen_data_set_sdm(data, seq_short_len=5, seq_prefer_len=50):

    data.sort_values("timestamp", inplace=True)
    train_set = []
    test_set = []
    for reviewerID, hist in tqdm(data.groupby('user_id')):
        pos_list = hist['movie_id'].tolist()
        genres_list = hist['genres'].tolist()
        rating_list = hist['rating'].tolist()
        for i in range(1, len(pos_list)):
            hist = pos_list[:i]
            genres_hist = genres_list[:i]
            if i <= seq_short_len and i != len(pos_list) - 1:
                train_set.append((reviewerID, hist[::-1], [0]*seq_prefer_len, pos_list[i], 1, len(hist[::-1]), 0,
                                  rating_list[i], genres_hist[::-1], [0]*seq_prefer_len))
            elif i != len(pos_list) - 1:
                train_set.append((reviewerID, hist[::-1][:seq_short_len], hist[::-1][seq_short_len:], pos_list[i], 1, seq_short_len,
                len(hist[::-1])-seq_short_len, rating_list[i], genres_hist[::-1][:seq_short_len], genres_hist[::-1][seq_short_len:]))
            elif i <= seq_short_len and i == len(pos_list) - 1:
                test_set.append((reviewerID, hist[::-1], [0] * seq_prefer_len, pos_list[i], 1, len(hist[::-1]), 0,
                                  rating_list[i], genres_hist[::-1], [0]*seq_prefer_len))
            else:
                test_set.append((reviewerID, hist[::-1][:seq_short_len], hist[::-1][seq_short_len:], pos_list[i], 1, seq_short_len,
                len(hist[::-1])-seq_short_len, rating_list[i], genres_hist[::-1][:seq_short_len], genres_hist[::-1][seq_short_len:]))

    random.shuffle(train_set)
    random.shuffle(test_set)

    print(len(train_set[0]), len(test_set[0]))

    return train_set, test_set


def split(x):
    key_ans = x.split(' ')
    return list(map(int, key_ans))

def gen_model_input(train_set, user_profile=None, item_profile=None, seq_max_len=50):

    train_uid = np.array(list(train_set["userid"].values))
    train_seq = list(map(split, train_set["hist_video_list"].values))

    train_iid = np.array(list(train_set["pos_video_id"].values))
    train_label = np.array(list(train_set["label"].values))
    train_hist_len = np.array(list(train_set["hist_list_len"].values))


    print("\n\n train_seq")
    print(type(train_seq))
    # 补充和截断  post从后面，默认pre从前面
    train_seq_pad = pad_sequences(train_seq, maxlen=seq_max_len, padding='post', truncating='post', value=0)

    print("\n\n train_seq_pad")
    print(type(train_seq_pad))

    train_model_input = {"mid": train_uid, "videoid": train_iid, "hist_video_id": train_seq_pad,
                         "hist_len": train_hist_len}

    for key in ["userRatedVideo1", "userGenre1", "userCity", "userRealplayCount",
                "videoGenre1", "authorid", "videoRealPlayCount", "videoDuration"]:
        try:
            train_model_input[key] = user_profile.loc[train_model_input['mid']][key].values
        except Exception as ex:
            print("\n\n gen_model_input exception ", ex)
            return None, None

    return train_model_input, train_label


def gen_model_input_user_emb(train_set, user_profile=None, item_profile=None, seq_max_len=50):

    train_uid = np.array(list(train_set["userid"].values))
    train_seq = list(map(split, train_set["hist_video_list"].values))

    train_iid = np.array(list(train_set["pos_video_id"].values))
    train_label = np.array(list(train_set["label"].values))
    train_hist_len = np.array(list(train_set["hist_list_len"].values))

    # 补充和截断  post从后面，默认pre从前面
    train_seq_pad = pad_sequences(train_seq, maxlen=seq_max_len, padding='post', truncating='post', value=0)

    train_model_input = {"mid": train_uid, "videoid": train_iid, "hist_video_id": train_seq_pad,
                         "hist_len": train_hist_len}

    for key in ["userRatedVideo1", "userGenre1", "userCity", "userRealplayCount"]:
        try:
            train_model_input[key] = user_profile.loc[train_model_input['mid']][key].values
        except Exception as ex:
            print("\n\n gen_model_input_user_emb exception ", ex)
            return None, None

    return train_model_input, train_label