TextCategorization/data_util.py

# encoding=utf-8
import re
import pickle
import gensim
import numpy as np
import pandas as pd
from pyhanlp import *
import keras.backend as K
from keras.preprocessing.sequence import pad_sequences

def load(path):
    '''
    pickle 加载pkl 文件 
    '''
    with open(path, 'rb') as f:
        return pickle.load(f)

def get_remove_word():
    '''
    加载停用词、不重要的词
    '''
    stopwords_path = 'pickle_1/bidi_classify_stop_words.csv' #　停用词文件　
    df_stopwords = pd.read_csv(stopwords_path)
    remove_word  = df_stopwords['stopword'].values.tolist()
    return remove_word

def get_embedding():
    '''
    加载文件，返回词典、keras tokennizer对象，词向量矩阵
    '''
    word_index = load('pickle_1/word_index_955871.pk') #加载词典文件 word:id
    tokenizer = load('pickle_1/tokenizer_955871.pk')   # 加载训练后keras tokenizer对象
    w2v_model_path = 'model/thr_100_model.vector'      # 加载词向量文件 
    w2v_model = gensim.models.KeyedVectors.load_word2vec_format(w2v_model_path,binary=True)
    embedding_matrix = np.random.random((len(word_index) + 1, 100))
    count_not_in_model = 0
    count_in_model = 0
    for word, i in word_index.items():
        if word in w2v_model:
            count_in_model += 1
            embedding_matrix[i] = np.asarray(w2v_model[word], dtype='float32')
        else:
            count_not_in_model += 1
    return word_index, tokenizer, embedding_matrix

def get_label():
    '''
    加载标签字典，返回字典label_mapping {0: '安防系统', 1: '安全保护服务', 2: '安全保护设备'  ; labels10 所有类别的中文名称
    '''
    label_mapping = load('pickle_1/label_mapping_f.pk')
    labels10 = list(label_mapping.values())
    return label_mapping,labels10

def get_dic():
    '''
    加载类别字典，估计是新旧类别： 豆类、油料和薯类种植': '农业,农、林、牧、渔业', '蔬菜、食用菌及园艺作物种植': '农业,农、林、牧、渔业'
    '''
    dic_label_path = 'pickle_1/class_subclass_dic211.pk'
    dic_label = load(dic_label_path)
    return dic_label

def model_in(r1, label_mapping, id):
    '''
    获取每个文章的中文类别名称
    @Argus: r1:np.array 预测结果 ; label_mapping:分类类别字典 0: '安防系统
    @Return：中文分类名称 
    '''
    all_end = r1
    aa2 = []
    for i in range(all_end.shape[0]):
        c1 = label_mapping[np.argmax(all_end[i])]
        aa2.append(c1)
    union = []
    for x in range(len(id)):
        union.append([id[x],aa2[x]])
    return union

def convertJlistToPlist(jList):
    '''
    将javaList 转为pythonlist     
    '''

    ret = []
    if jList is None:
        return ret
    for i in range(jList.size()):
        ret.append(str(jList.get(i)))
    return ret 

def clean_RmWord(text, remove_word):
    '''
    去除没用的词语
    '''
    text_copy = text.copy()
    for i in text:
        if i in remove_word:
            text_copy.remove(i)
    text_copy = " ".join(text_copy)
    return text_copy

def handle_doc1(article_set10_1, remove_word):
    '''
    句子分词并删除单字、重复、无关词语
    @Argus: article_set10_1: 包含待处理字符串的Series
    @Return: 处理后的结果
    '''
    HanLP.Config = JClass('com.hankcs.hanlp.HanLP$Config')
    HanLP.Config.ShowTermNature = False
    article_set10_seg_1 = article_set10_1.map(lambda x: convertJlistToPlist(HanLP.segment(x)))
    article_set10_seg_1 = article_set10_seg_1.map(lambda x: ' '.join(word for word in x if len(word) > 1)) # 删除单个字
    article_set10_seg_rm = article_set10_seg_1.map(lambda x: clean_RmWord(x.split(), remove_word)) # 删除无用、重复词语
    article_set10_seg_rm = article_set10_seg_rm.map(lambda x: x.split())
    return article_set10_seg_rm

def cleanSeg(text):
    '''
    清除干扰字符（英文、日期、数字、标点符号）
    '''
    text = re.sub('[a-zA-Z]', '', text)
    text = text.replace('\n', ' ')
    text = re.sub(r"-", " ", text)
    text = re.sub(r"\d+/\d/\d+", "", text)
    text = re.sub(r"[0-2]?[0-9]:[0-6][0-9]", "", text)
    text = re.sub(r"[\w]+@[\.\w]+", "", text)
    text = re.sub(r"/[a-zA-Z]*[:\//\]*[A-Za-z0-9\-_]+\.+[A-Za-z0-9\.\/%&=\?\-_]+/i", "", text)
    pure_text = ''
    for letter in text:
        if letter.isalpha() or letter == ' ':
            pure_text += letter
    text = ' '.join(word for word in pure_text.split() if len(word) > 1)
    text = text.replace(' ', '')
    return text 

def fetch_sub_data_1(data, num):
    '''
    获取文本前N个字符
    '''
    return data[:num]

def data_set(text):
    '''
    保持顺序词语去重
    '''
    l2 = []
    [l2.append(i) for i in text if i not in l2]
    return l2

def clean_word(article_set10,remove_word):
    """
    清理数据，清除符号、字母、数字等，统一文章长度，对句子进行分词，删除单字、重复、无关词语、停用词
    :param article_set10: 原数据，list
    :param remove_word: 停用词表，list
    :return: Series
    """
    article_set10_1 = pd.Series(article_set10)
    article_set10_1 = article_set10_1.map(lambda x: cleanSeg(x))  # 清除干扰字符（英文、日期、数字、标点符号）
    article_set10_1 = article_set10_1.map(lambda x: fetch_sub_data_1(x, 500))  # 获取文本前N个字符
    # test
    article_set10_seg_rm = handle_doc1(article_set10_1, remove_word) # 句子分词并删除单字、重复、无关词语
    # test
    x_train_df_10 = article_set10_seg_rm.copy()
    x_train_df_10 = x_train_df_10.map(lambda x: data_set(x))  #  保持顺序词语去重
    return x_train_df_10

def clean_word_with_tokenizer(article_set10,remove_word,tokenizer):
    """
    清理数据，清除符号、字母、数字、停用词，分词
    :param article_set10: 原数据，list
    :param remove_word: 停用词表，list
    :return: Series
    """
    id = [i[0] for i in article_set10]
    article_set10 = [i[1] for i in article_set10]
    article_set10_1 = pd.Series(article_set10)
    article_set10_1 = article_set10_1.map(lambda x: cleanSeg(x))
    article_set10_1 = article_set10_1.map(lambda x: fetch_sub_data_1(x, 500))
    # test
    article_set10_seg_rm = handle_doc1(article_set10_1, remove_word)
    # print(article_set10_seg_rm)
    # test
    x_train_df_10 = article_set10_seg_rm.copy()
    sequences = tokenizer.texts_to_sequences(x_train_df_10)
    padded_sequences = pad_sequences(sequences, maxlen=100, padding='post', truncating='post',value=0.0)
    # left_word = [x[:-1] for x in padded_sequences]
    # right_word = [x[1:] for x in padded_sequences]
    # left_pad = pad_sequences(left_word, maxlen=100, value=0.0)
    # right_pad = pad_sequences(right_word, maxlen=100, padding='post', truncating='post', value=0.0)
    return padded_sequences, id

def recall(y_true, y_pred):
    '''
    计算召回率

    @Argus:
        y_true: 正确的标签
        y_pred: 模型预测的标签

    @Return
        召回率
    '''
    c1 = K.sum(K.round(K.clip(y_true * y_pred, 0, 1)))
    c3 = K.sum(K.round(K.clip(y_true, 0, 1)))
    if c3 == 0:
        return 0
    recall = c1 / c3
    return recall


def f1_score(y_true, y_pred):
    '''
    计算F1

    @Argus:
        y_true: 正确的标签
        y_pred: 模型预测的标签

    @Return
        F1值
    '''

    c1 = K.sum(K.round(K.clip(y_true * y_pred, 0, 1)))
    c2 = K.sum(K.round(K.clip(y_pred, 0, 1)))
    c3 = K.sum(K.round(K.clip(y_true, 0, 1)))
    precision = c1 / c2
    if c3 == 0:
        recall = 0
    else:
        recall = c1 / c3
    f1_score = 2 * (precision * recall) / (precision + recall)
    return f1_score


def precision(y_true, y_pred):
    '''
    计算精确率

    @Argus:
        y_true: 正确的标签
        y_pred: 模型预测的标签

    @Return
        精确率
    '''
    c1 = K.sum(K.round(K.clip(y_true * y_pred, 0, 1)))
    c2 = K.sum(K.round(K.clip(y_pred, 0, 1)))
    precision = c1 / c2
    return precision

if __name__ == '__main__':
    dic_label = get_dic()
    print(dic_label)