RelationalIntentionGraph/util.py at main · HKUST-KnowComp/RelationalIntentionGraph · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
import pandas as pd
import os
import random
import time
import json
import pickle
import numpy as np


all_uk_session_file = r'.\data\uk_session_train.txt'
all_item_file = r'.\data\uk_products_train.csv'
session_count = 1182180

def sample_from_all_session(sample_num = 100):
    sample_file = rf'.\data\uk_sample_{sample_num}.txt'
    fw = open(sample_file,'w',encoding='utf-8')

    sample_list = set()
    while len(sample_list)<sample_num:
        sample_list.add(random.randint(0,session_count))
    sample_list = list(sample_list)

    with open(all_uk_session_file,'r',encoding='utf-8') as f:
        tmp = f.readlines()
        for item in sample_list:
            all_list = tmp[item]
            train_list = all_list.strip().rsplit(' ',1)
            fw.write(train_list[0]+'\n')
    fw.close()

def find_item_info_for_sample(sample_num = 100):
    filter_item_path = all_item_file
    sample_path = rf'.\data\uk_sample_{sample_num}.txt'
    sample_item_path=rf'.\data\uk_item_sample_{sample_num}.txt'
    file_product = pd.read_csv(filter_item_path,encoding='utf-8')
    item_set = set()
    with open(sample_path,'r') as f:
        for lines in f.readlines():
            item= lines.strip().split(' ')
            item_set.update(item)
    filter_item = file_product[file_product['id'].isin(item_set)]
    filter_item = filter_item.fillna("NaN")
    filter_item.to_csv(sample_item_path,index=False,sep='\t')

def get_message(sample_session_path = 'uk_sample_1000.txt',sample_item_path='uk_item_sample_1000.txt'):
    item_dict = {}
    # title = ['id','locale','title','price','brand','color','size','model','material','author','desc']
    with open(sample_item_path,'r',encoding='utf-8') as f:
        title = f.readline().strip().split('\t')
        del title[1] #### locale is same for all

        for line in f.readlines():
            item = line.strip().split('\t')
            del item[1]  ### locale info
            if item[0] not in item_dict:
                tmp = dict(zip(title,item))
                item_dict[item[0]] = {key:val for key,val in tmp.items() if val != 'NaN'}  ### clear Nan value for prompt

    with open(sample_session_path,'r',encoding='utf-8') as f:
        for sessions in f.readlines():
            sessions = sessions.strip().split(' ')
            sessions = sessions[:-1]  ## the last one is test instance.
            session_temp = [item_dict[x] for x in sessions]
            message = "Below is a user chronological record list:\n"+str(session_temp)+'\n'+\
            "Explain the basic intentions of this user exactly, and output your answers in the following format concisely:\nUser buy these items because:\nintention1: {a simplest description within 10 words}\n....\nAnalyze the user intention belongs to which levels of Maslow's Hierarchy of Needs. Only Return level names in the following format:\nNeeds:\nintention1: {levelname,...}...."

            yield sessions, message


def get_simple_message(sample_session_path = 'uk_sample_1000.txt',sample_item_path='uk_item_sample_1000.txt'):
    item_dict = {}
    # title = ['id','locale','title','price','brand','color','size','model','material','author','desc']
    with open(sample_item_path,'r',encoding='utf-8') as f:
        title = f.readline().strip().split('\t')
        del title[1] #### locale is same for all

        for line in f.readlines():
            item = line.strip().split('\t')
            del item[1]  ### locale info
            if item[0] not in item_dict:
                tmp = dict(zip(title,item))
                item_dict[item[0]] = {key:val for key,val in tmp.items() if val != 'NaN'}  ### clear Nan value for prompt

    with open(sample_session_path,'r',encoding='utf-8') as f:
        for sessions in f.readlines():
            sessions = sessions.strip().split(' ')
            sessions = sessions[:-1]  ## the last one is test instance.
            session_temp = [item_dict[x] for x in sessions]
            # message = "Below is a user chronological record list:\n"+str(session_temp)+'\n'+\
            # "Explain the basic intentions of this user exactly, and output your answers in the following format concisely:\nUser buy these items because:\nintention1: {a simplest description within 10 words}\n...."

            message = "Below is a user chronological record list:\n"+str(session_temp)+'\n'+\
            "Explain the basic intentions of this user exactly. Output several different intentions one by one to answer the following question: User buy these items because they want to:\nintention 1: {a simple verb phrase within 10 words}\nintention 2: {a simple verb phrase within 10 words}....\n"


            yield sessions, message