LivedoorニュースコーパスをUniversal Sentence Encoderでベクトル化して、マルチクラス分類をしてみる

from lightgbm import LGBMClassifier
import matplotlib.pyplot as plt
import numpy as np
import os
from sklearn.manifold import TSNE
from sklearn.multiclass import OneVsRestClassifier
from sklearn.naive_bayes import GaussianNB # ガウシアンナイーブベイズ
import tensorflow as tf
import tensorflow_hub as hub
import tensorflow_text # 明示的な依存関係の説明がないけどtensorflow_hubのモデルに必須
import time

# グローバル変数
LIVEDOOR_TEXT_PATH = "./livedoor_text"
LIVEDOOR_DATA_LIST = ["dokujo-tsushin","it-life-hack","kaden-channel","livedoor-homme","movie-enter","peachy","smax","sports-watch","topic-news"]

def read_text_files_in_folder(folder_path, file_header="", start_line=1, end_line=None, max_files=0):
    file_contents = []  # 空のリストを作成
    file_count = 0

    # フォルダ内のファイルを走査
    for filename in os.listdir(folder_path):
        if max_files > 0 and file_count >= max_files:
            break # ファイル数上限に達したらループを終了
            
        file_path = os.path.join(folder_path, filename)
        if os.path.isfile(file_path) and filename.lower().endswith(".txt"):
            
            # ファイルヘッダーが設定されていたらフィルタ
            if file_header and filename.startswith(file_header):
                
                # テキストファイルの場合のみ読み込む
                with open(file_path, "r", encoding="utf-8") as file:
                    lines = file.readlines()
                    
                    if end_line is None:
                      	# 特定の行から最後まで読み込む
                        content = "".join(lines[start_line - 1 :])  
                    else:
                        content = "".join(lines[start_line - 1 : end_line - 1])
                        
                    file_contents.append(content)
                    file_count += 1

    return file_contents

def split_list_by_ratio(lst, ratio):
    n = int(len(lst) * ratio)
    return lst[:n], lst[n:]

def read_livedoor():
    file_label_Y = []
    file_label_test = []
    split_list_80 = []
    split_list_20 = []
    
    # Universal Sentence Encoderのモデル読み出し
    embed = hub.load("https://tfhub.dev/google/universal-sentence-encoder-multilingual/3")
    # ループで複数のフォルダの内容を結合するため、ダミーのデータで同じ形のテンソルを作っておく
    embeddings_X = embed("ダミー")
    embeddings_test = embed("ダミー")
    
    start_time = time.time()
    
    max_files = 500
    for header in LIVEDOOR_DATA_LIST:
        print(header)
        
        folder_path = LIVEDOOR_TEXT_PATH + "/" + header + "/"
        
        # フォルダ内のテキストを読み込む
        file_contents_temp = read_text_files_in_folder(folder_path,header,start_line=4,max_files=max_files)
        
        # ダミーで変数初期化
        embeddings_temp = embed("ダミー")
        for contents in file_contents_temp:
            # 読み込んだテキストをベクトル化して、これまでのテンソルにくっつける
            embeddings_temp = tf.concat([embeddings_temp, embed(contents)],0)
        
        # ダミー分を取り除く
        embeddings_temp = np.delete(embeddings_temp.numpy(),0,0)
        
        # 読み込んだデータを学習用とテスト用に分ける
        split_list_80, split_list_20 = split_list_by_ratio(embeddings_temp, 0.8)
        
        # 学習用のテンソルとテスト用のテンソルにそれぞれ追加する
        embeddings_X = tf.concat([embeddings_X, split_list_80], 0)
        embeddings_test = tf.concat([embeddings_test, split_list_20], 0)
        
        # ラベルデータも生成してくっつける
        file_label_list = [header] * len(embeddings_temp)
        split_list_80, split_list_20 = split_list_by_ratio(file_label_list, 0.8)
        file_label_Y += split_list_80
        file_label_test += split_list_20
    

    # ダミー分を取り除く
    embeddings_X = np.delete(embeddings_X.numpy(),0,0)
    embeddings_test = np.delete(embeddings_test.numpy(),0,0)
    
    interval = time.time() - start_time
    print('text read and embedding time: {}s'.format(interval))
    
    return embeddings_X, embeddings_test, file_label_Y, file_label_test

# main1：embedding
#テキストを文章ベクトルに変換

embeddings_X, embeddings_test, file_label_Y, file_label_test = read_livedoor()

def view_tsne_with_label(tsne_data, data_label):
    start_time = time.time()
    
    # TSNEモジュールを呼び出して、ベクトル化したデータを2次元に表現する
    embedding = TSNE(n_components=2).fit_transform(tsne_data)
    
    interval = time.time() - start_time
    print('t-sne time: {}s'.format(interval))
    
    # ラベルごとにマスクした結果を系列にして表示する
    for label in np.unique(data_label):
        mask = [ True if l == label else False for l in data_label ]
        plt.scatter(embedding[mask, 0], embedding[mask, 1], label=label)
        
    plt.legend(bbox_to_anchor=(1.05,1), loc='upper left')
    plt.show()

# main2：データ可視化
print('学習用データ件数：',len(embeddings_X))
view_tsne_with_label(embeddings_X, file_label_Y)
print('テスト用データ件数：',len(embeddings_test))
view_tsne_with_label(embeddings_test, file_label_test)

def label_to_int_def(label_def):
    replace_dict = {"none": 0}
    reverse_dict = {0: "none"}
    
    for i in range(len(label_def)):
        replace_dict[label_def[i]] = i + 1
        reverse_dict[i + 1] = label_def[i]
    
    return replace_dict, reverse_dict

# main3：学習用データ前処理

replace_dict, reverse_dict = label_to_int_def(LIVEDOOR_DATA_LIST)
file_numlabel_Y = [replace_dict[word] if word in replace_dict else 0 for word in file_label_Y]

# macro-F1 (macro-Precision)

def macro_precision(predict_list, label_list):
    harmonic_mean = []
    
    print('')
    print('hermonic mean:')
    for label in np.unique(label_list):
        
        predict_num = len([ 1 for l in predict_list if l == label ])
        label_num = len([ 1 for l in label_list if l == label ])
    
        true_num = len([ 1 for i in range(len(predict_list)) if label_list[i] == label and predict_list[i] == label_list[i] ])
        
        mean_num = 2 * (true_num / predict_num) * (true_num / label_num) / ((true_num / predict_num) + (true_num / label_num))
        
        harmonic_mean.append(mean_num)
        print(label, ': ', mean_num)
    
    macro_f1 = sum(harmonic_mean)/len(np.unique(label_list))
    
    print('')
    print('macro_f1: ',macro_f1)
    

def main3to4(clf, embeddings_X, file_numlabel_Y, embeddings_test):
    # main3：学習
    start_time = time.time()
    
    clf.fit(embeddings_X, file_numlabel_Y) # 学習をする
    
    interval = time.time() - start_time
    print('lerning time: {}s'.format(interval))
    
    # main4：推論
    start_time = time.time()

    predict_list = clf.predict(embeddings_test) # 推論
    predict_list = predict_list.tolist()

    interval = time.time() - start_time
    print('predict time: {}s'.format(interval))
    
    return predict_list

# main3to4：NaiveBayes

clf = OneVsRestClassifier(GaussianNB())
predict_list = main3to4(clf,  embeddings_X, file_numlabel_Y, embeddings_test)

# main5：評価
predict_str_list =  [reverse_dict[num] if num in reverse_dict else "none" for num in predict_list]
macro_precision(predict_str_list, file_label_test)

# main3to4：LightGBM

clf = OneVsRestClassifier(LGBMClassifier())
predict_list = main3to4(clf,  embeddings_X, file_numlabel_Y, embeddings_test)

# main5：評価
predict_str_list =  [reverse_dict[num] if num in reverse_dict else "none" for num in predict_list]
macro_precision(predict_str_list, file_label_test)