Sentiment_Analysis_DataProcess.py

# !usr/bin/env python
# -*- coding:utf-8 -*-

""" 
@Author: HsuDan
@Date: 2022-02-18 19:12:58
@Version: 1.0
@LastEditors: HsuDan
@LastEditTime: 2022-02-25 11:18:01
@Description: DataProcess
@FilePath: /Sentiment-Analysis-Chinese-pytorch/Sentiment_Analysis_DataProcess.py
"""

from __future__ import unicode_literals, print_function, division
from io import open
import torch
import re
import pandas as pd
import numpy as np
import gensim
from torch.utils.data import Dataset
from Sentiment_Analysis_Config import Config


class Data_set(Dataset):
    def __init__(self, Data, Label):
        self.Data = Data
        if Label is not None:  # 考虑对测试集的使用
            self.Label = Label

    def __len__(self):
        return len(self.Data)

    def __getitem__(self, index):
        if self.Label is not None:
            data = torch.from_numpy(self.Data[index])
            label = torch.from_numpy(self.Label[index])
            return data, label
        else:
            data = torch.from_numpy(self.Data[index])
            return data


def data_preview(file_path):
    """
    @description: preview the original data set: size, description info, etc.
    @param {*}
    - file_path: str, given the data file path.
    @return {*}
    - df: DataFrame, data in DataFrame format
    """
    with open(file_path, "r", encoding="utf-8") as f:
        data = []
        for line in f.readlines():
            sall = line.strip().split()
            data.append([int(sall[0]), sall[1:]])
    df = pd.DataFrame(data, columns=["label", "text"])
    print(
        "\n"
        + "*" * 20
        + " Original Data Description ("
        + file_path.split("/")[-1]
        + ")"
        + "*" * 20
        + "\n"
    )
    print("data length: ", len(data))
    print("data preview:\n", df.head())
    print("count label:\n", df["label"].value_counts())
    return df


def stopwordslist():  # 创建停用词表
    stopwords = [
        line.strip()
        for line in open(Config.stopword_path, encoding="UTF-8").readlines()
    ]
    return stopwords


def build_word2id(file):
    """
    @description: build the dict of 'word2id'
    @param {*}
    - file: str, word2id保存地址
    @return {*}
    返回word2id的字典
    """
    stopwords = stopwordslist()
    word2id = {"_PAD_": 0}
    path = [Config.train_path, Config.val_path]

    for _path in path:
        with open(_path, encoding="utf-8") as f:
            for line in f.readlines():
                out_list = []
                # 去停用词
                sp = line.strip().split()
                for word in sp[1:]:
                    if word not in stopwords:
                        rt = re.findall("[a-zA-Z]+", word)
                        if word != "\t":
                            # 去除长度为1的word
                            if len(rt) == 1:
                                continue
                            else:
                                out_list.append(word)
                for word in out_list:
                    if word not in word2id.keys():
                        word2id[word] = len(word2id)
    with open(file, "w", encoding="utf-8") as f:
        for w in word2id:
            f.write(w + "\t")
            f.write(str(word2id[w]))
            f.write("\n")
    return word2id


def build_id2word(word2id):
    """
    @description: 得到id2word
    @param {*}
    - word2id: dict, 字典word2id
    @return {*}
    - id2word: dict， 字典id2word
    """
    id2word = {}
    for key, val in word2id.items():
        id2word[val] = key
    return id2word


def build_word2vec(fname, word2id, save_to_path=None):
    """
    @description: 返回语料文本中词汇集对应的word2vec向量
    @param {*}
    - fname: str, 预训练的word2vec.
    - word2id: dict, 语料文本中包含的词汇集.
    - save_to_path: str, 保存训练语料库中的词组对应的word2vec到本地.
    @return {*}
    语料文本中词汇集对应的word2vec向量{id: word2vec}.
    """

    n_words = max(word2id.values()) + 1
    model = gensim.models.KeyedVectors.load_word2vec_format(fname, binary=True)
    word_vecs = np.array(np.random.uniform(-1.0, 1.0, [n_words, model.vector_size]))
    for word in word2id.keys():
        try:
            word_vecs[word2id[word]] = model[word]
        except KeyError:
            pass
    if save_to_path:
        with open(save_to_path, "w", encoding="utf-8") as f:
            for vec in word_vecs:
                vec = [str(w) for w in vec]
                f.write(" ".join(vec))
                f.write("\n")
    return word_vecs


def text_to_array(word2id, seq_lenth, path):
    """
    @description: 文本转为索引数字模式-将原始文本（包括标签和文本）里的每个词转为word2id对应的索引数字，并以数组返回.
    @param {*}
    - word2id: dict, 语料文本中包含的词汇集.
    - seq_length: int, 序列的限定长度.
    - path: str, 待处理的原始文本数据集.
    @return {*}
    返回原始文本转化索引数字数组后的数据集(array), 标签集(list).
    """

    label_array = []
    i = 0
    sa = []
    # 获取句子个数
    with open(path, encoding="utf-8") as f1:
        for l1 in f1.readlines():
            s = l1.strip().split()
            s1 = s[1:]
            new_s = [word2id.get(word, 0) for word in s1]  # 单词转索引数字
            sa.append(new_s)
        # print(len(sa))

    with open(path, encoding="utf-8") as f:
        sentences_array = np.zeros(shape=(len(sa), seq_lenth))  # 行：句子个数 列：句子长度
        for line in f.readlines():
            sl1 = line.strip().split()
            sen = sl1[1:]
            new_sen = [word2id.get(word, 0) for word in sen]  # 单词转索引数字,不存在则为0
            new_sen_np = np.array(new_sen).reshape(1, -1)

            # 补齐每个句子长度，多余补零，少了就直接赋值,0填在前面。
            if np.size(new_sen_np, 1) < seq_lenth:
                sentences_array[i, seq_lenth - np.size(new_sen_np, 1) :] = new_sen_np[
                    0, :
                ]
            else:
                sentences_array[i, 0:seq_lenth] = new_sen_np[0, 0:seq_lenth]

            i = i + 1
            label = int(sl1[0])  # 标签
            label_array.append(label)
    return np.array(sentences_array), label_array


def text_to_array_nolabel(word2id, seq_lenth, path):
    """
    @description: 文本转为索引数字模式-将原始文本（仅包括文本）里的每个词转为word2id对应的索引数字，并以数组返回.
    @param {*}
    - word2id: dict, 语料文本中包含的词汇集.
    - seq_length: int, 序列的限定长度.
    - path: str, 待处理的原始文本数据集.
    @return {*}
    原始文本转化索引数字数组后的数据集(array).
    """

    i = 0
    sa = []
    # 获取句子个数
    with open(path, encoding="utf-8") as f1:
        for l1 in f1.readlines():
            s = l1.strip().split()
            s1 = s[0:]  # s1=s[1:]
            new_s = [word2id.get(word, 0) for word in s1]  # 单词转索引数字
            sa.append(new_s)
    with open(path, encoding="utf-8") as f:
        sentences_array = np.zeros(shape=(len(sa), seq_lenth))  # 行：句子个数 列：句子长度
        for line in f.readlines():
            sl1 = line.strip().split()
            sen = sl1[0:]  # sen=sl1[1:]
            new_sen = [word2id.get(word, 0) for word in sen]  # 单词转索引数字,不存在则为0
            new_sen_np = np.array(new_sen).reshape(1, -1)
            if np.size(new_sen_np, 1) < seq_lenth:
                sentences_array[i, seq_lenth - np.size(new_sen_np, 1) :] = new_sen_np[
                    0, :
                ]
            else:
                sentences_array[i, 0:seq_lenth] = new_sen_np[0, 0:seq_lenth]
            i = i + 1
    return np.array(sentences_array)


def to_categorical(y, num_classes=None):
    """
    @description: 将类别转化为one-hot编码
    @param {*}
    - y: list, 类别特征的列表
    - num_class: int, 类别个数
    @return {*}
    返回one-hot编码数组,shape:（len(y), num_classes）
    """
    y = np.array(y, dtype="int")
    input_shape = y.shape
    if input_shape and input_shape[-1] == 1 and len(input_shape) > 1:
        input_shape = tuple(input_shape[:-1])
    y = y.ravel()
    if not num_classes:
        num_classes = np.max(y) + 1
    n = y.shape[0]
    categorical = np.zeros((n, num_classes))
    categorical[np.arange(n), y] = 1
    output_shape = input_shape + (num_classes,)
    categorical = np.reshape(categorical, output_shape)
    return categorical


def prepare_data(w2id, train_path, val_path, test_path, seq_lenth):
    """
    @description: 得到数字索引表示的句子和标签
    @param {*}
    - w2id: dict, 语料文本中包含的词汇集.
    - train_path: str, 训练数据集所在路径
    - val_path: str, 验证数据集所在路径
    - test_path: str, 测试数据集所在路径
    - seq_lenth: int, 序列的固定长度
    @return {*}
    - train_array: array, 训练集文本数组, shape:(len(train), seq_len)
    - train_label: array, 训练集标签数组, shape:(len(train), 1)
    - val_array: array, 验证集文本数组, shape:(len(val), seq_len)
    - val_label: array, 验证集标签数组, shape:(len(val), 1)
    - test_array: array, 测试集文本数组, shape:(len(test), seq_len)
    - test_label: array, 测试集标签数组, shape:(len(test), 1)
    """

    train_array, train_label = text_to_array(w2id, seq_lenth=seq_lenth, path=train_path)
    val_array, val_label = text_to_array(w2id, seq_lenth=seq_lenth, path=val_path)
    test_array, test_label = text_to_array(w2id, seq_lenth=seq_lenth, path=test_path)

    # 标签为[1, 1, 1, 1, 1, 1, 1, 1, 0, 0...]将标签转为onehot
    # train_label=to_categorical(train_label,num_classes=2)
    # val_label=to_categorical(val_label,num_classes=2)
    # print(train_lab,"\nval\n",val_lab)
    # """转换后标签
    #         [[0. 1.]
    #         [0. 1.]
    #         [0. 1.]
    #         ...
    #         [1. 0.]
    #         [1. 0.]
    #         [1. 0.]]"""

    train_label = np.array([train_label]).T
    val_label = np.array([val_label]).T
    test_label = np.array([test_label]).T

    return train_array, train_label, val_array, val_label, test_array, test_label


if __name__ == "__main__":
    # preview data
    train_df = data_preview(Config.train_path)
    test_df = data_preview(Config.test_path)
    val_df = data_preview(Config.val_path)

    # 建立word2id
    word2id = build_word2id(Config.word2id_path)

    # 建立id2word
    id2word = build_id2word(word2id)

    # 建立word2vec
    w2vec = build_word2vec(
        Config.pre_word2vec_path, word2id, Config.corpus_word2vec_path
    )

    # 得到句子id表示和标签
    (
        train_array,
        train_label,
        val_array,
        val_label,
        test_array,
        test_label,
    ) = prepare_data(
        word2id,
        train_path=Config.train_path,
        val_path=Config.val_path,
        test_path=Config.test_path,
        seq_lenth=Config.max_sen_len,
    )

    np.savetxt("./word2vec/train_data.txt", train_array, fmt="%d")
    np.savetxt("./word2vec/val_data.txt", val_array, fmt="%d")
    np.savetxt("./word2vec/test_data.txt", test_array, fmt="%d")
    np.savetxt("./word2vec/train_label.txt", train_label, fmt="%d")
    np.savetxt("./word2vec/val_label.txt", val_label, fmt="%d")
    np.savetxt("./word2vec/test_label.txt", test_label, fmt="%d")
    print("Data processing is complete.")