stopWordTest.py

# -*- coding: utf-8 -*-
# @Author: Gree
# @Date:   2021-05-29 15:26:57
# @Last Modified by:   Gree
# @Last Modified time: 2021-05-31 13:48:14


import json
import os
import time


# 第三方库
import csv
import matplotlib as mpl
import numpy as np
import pandas as pd
import requests
from config import readConfig as rc
from data import readData as rd
from tqdm import tqdm
from matplotlib import pyplot as plt
from matplotlib.font_manager import FontProperties


class StopWordTest:
    def __init__(self):
        """
        __init__ 函数:
        input:
        output:
        features: 定义初始变量，输出初始日志信息
        step1: 定义 self.dirPath、self.url、self.headers 等对象
        """
        # 输出 log 信息
        print("#" * 30, end = "")
        # 输出 log 信息
        print("Loading the module of stopWordTest ....", end = "")
        # 输出 log 信息
        print("#" * 30)
        # 定义 self.dirPath
        self.dirPath = os.path.split(os.path.realpath(__file__))[0] + "/result"
        # 测试接口的 url
        self.url = "https://testnlu.gree.com/regnlu/query"
        # 定义请求头
        self.headers = {
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.36"
        }
        # 定义 self.initial_list
        self.initial_list = rd.ReadData().readData()
        # 定义 initial_df
        self.initial_df = rd.ReadData().getData(self.initial_list)
        # 定义 self.stopWordList
        self.stopWordList = rc.ReadConfig().readConfig()
        # 定义 self.correctRateLst
        self.correctRateLst = []
        # 定义 self.errorRateLst
        self.errorRateLst = []


    def dealStopWord(self, stopWord):
        """
        dealStopWord 函数:
        input: stopWord
        output: df
        features: 处理停用词，去噪声
        step1: 进行停用词处理，去噪声
        step2: 返回经过处理后数据框
        """
        # 输出 log 信息
        print("==========================进入停用词处理============================")
        # 输出 log 信息
        print("正在处理停用词：", stopWord)

        # 新建 dealStopWordList
        dealStopWordList = []
        # copy self.df
        df = self.initial_df.copy()

        # 遍历
        for index, row in df.iterrows():
            # 捕获异常
            try:
                # 条件判断
                if stopWord in row["query"]:
                    # 赋值
                    initial_query = row["query"]
                    # 字符串替换
                    row["query"] = row["query"].replace(stopWord, "")
                    # 条件判断
                    if row["query"] != "":
                        # 构建字典
                        stopWordDict = {
                            "domain": row["domain"],
                            "intent": row["intent"],
                            "initial_query": initial_query,
                            "query": row["query"]
                        }

                        # 列表添加元素
                        dealStopWordList.append(stopWordDict)

                    else:
                        pass

                else:
                    pass

            except Exception as e:
                print("The error of dealing stopWord: ", e)

        # 捕获异常
        try:
            # 列表转为数据框
            df = pd.DataFrame(dealStopWordList, columns = ["domain","intent", "initial_query", "query"])
            # 数据去重
            # df = df.drop_duplicates(subset = "query")
            # 输出 log 信息
            print("停用词处理结束 ...")
            # 输出 log 信息
            print("The dimension of df after dealing stopWord: ", end = "")
            # 数据数据框的维度
            print(df.shape)
            # 输出 log 信息
            print("==========================退出停用词处理============================")
            # 返回数据框
            return df

        except Exception as e:
            print("The error of converting list to data frame: ", e)


    def dataTest(self, df):
        """
        dataTest 函数:
        input: df
        output: df
        features: 请求测试接口，返回测试数据
        step1: 请求测试接口：https://nlu.gree.com/test2/regnlu/query
        step2: 返回 df
        """
        # 输出 log 信息
        print("=========================批量发起网络请求===========================")
        # 输出 log 信息
        print("Requesting url ...")

        # copy df
        df = df.copy()

        # 捕获异常
        try:
            # for 循环
            for k in tqdm(range(0,len(df["query"]))):
                # 捕获异常
                try:
                    # 获取字段
                    query = df["query"].iloc[k]
                    # 定义 para
                    para = {"clientId":"test_zww","sessionId":"test_zww","agentId":"YTYxMjRmZDY1NGM1","token":"bb2c960a3a8e018fab0a7b1bccfab91d3855","query":query }
                    # 字典转字符串
                    para = json.dumps(para)
                    # 发起 post 请求
                    r = requests.post(url = self.url, headers = self.headers, data = para, timeout = 60)

                except Exception as e:
                    print("The error of requesting url: ", e)

                # 捕获异常
                try:
                    # 数据框操作
                    df.loc[k,"url_domain"] = json.loads(r.text).get("semantic").get("service")
                    df.loc[k,"url_intent"] = json.loads(r.text).get("semantic").get("action")

                except Exception as e:
                    print("The error of operating dataframe: ", e)

        except Exception as e:
            print("The error of dataTest(): ", e)

        else:
            # 返回 df
            return df

        finally:
            # 输出 log 信息
            print("The process of requesting url finished!")
            print("=========================结束所有网络请求===========================")


    def dataEvaluation(self, stopWord, df, countCorrect, countError):
        """
        dataEvaluation 函数:
        input: stopWord, df, countCorrect, countError
        output: df
        features: 对测试完的数据，进行评估
        step1: 对测试完的数据，进行评估，计算语料识别的正确率和错误率
        step2: 返回 correctRate, errorRate, rowCorrectDictLst, rowErrorDictLst
        """
        # copy df
        df = df.copy()
        # 新建一个 list 接收元素
        rowCorrectDictLst = []
        # 新建一个 list 接收元素
        rowErrorDictLst = []
        # for 循环
        for index, row in df.iterrows():
            # 捕获异常
            try:
                # 条件判断
                if row["domain"] == "chat" and row["url_domain"] != "":
                    # 构建字典
                    rowCorrectDict = {
                        "domain": row["domain"],
                        "intent": row["intent"],
                        "initial_query": row["initial_query"],
                        "query": row["query"],
                        "url_domain": row["url_domain"],
                        "url_intent": row["url_intent"]
                    }
                    # 列表添加元素
                    rowCorrectDictLst.append(rowCorrectDict)
                    # 计数
                    countCorrect += 1

                else:
                    pass

            except Exception as e:
                print("The error of counting correct corpus: ", e)


        # for 循环
        for index, row in df.iterrows():
            # 捕获异常
            try:
                # 条件判断
                if row["domain"] != "chat" and row["url_domain"] != "":
                    # 构建字典
                    rowErrorDict = {
                        "domain": row["domain"],
                        "intent": row["intent"],
                        "initial_query": row["initial_query"],
                        "query": row["query"],
                        "url_domain": row["url_domain"],
                        "url_intent": row["url_intent"]
                    }
                    rowErrorDictLst.append(rowErrorDict)
                    # 计数
                    countError += 1

                else:
                    pass

            except Exception as e:
                print("The error of counting error corpus: ", e)

        # 输出 log 日志
        print("=====================计算正确识别语料数量和比率=====================")

        # 计算正确识别语料比率
        correctRate = countCorrect / df.shape[0]
        # 输出 countOne
        print("正确识别语料数量：", str(countCorrect))
        # 输出 correctRate
        print("正确识别语料比率：", str(correctRate))

        # 输出 log 日志
        print("=============================计算结束===============================")


        # 输出 log 信息
        print("=====================计算错误识别语料数量和比率=====================")

        # 计算错误识别语料比率
        errorRate = countError / df.shape[0]

        # 输出 log 日志
        print("错误识别语料数量：", str(countError))
        print("错误识别语料比率：", str(errorRate))

        # 输出 log 日志
        print("=============================计算结束===============================")

        # 返回 correctRate, errorRate
        return correctRate, errorRate, rowCorrectDictLst, rowErrorDictLst


    def WriteData(self, filePath, lst):
        """
        WriteData 函数:
        input: filePath, lst
        output: 数据持久化
        features: 对最终数据进行数据持久化
        step1: 文件写入
        """
        # 输出 log 信息
        print("Writing file:", filePath)
        # 打开文件
        with open(filePath, 'w+', encoding = "utf-8") as f:
            # I/O 操作
            writer = csv.DictWriter(f, fieldnames = ("domain", "intent", "initial_query", "query", "url_domain", "url_intent"))
            # 写入头
            writer.writeheader()
            # for 循环
            for item in lst:
                # 文件写入
                writer.writerow(item)
            # 结束 I/O 操作
            f.close()


    def drawPicture(self, correctRateLst, errorRateLst):
        """
        drawPicture 函数:
        input: correctRateLst, errorRateLst
        output: 图形
        features: 将语料识别的正确率和错误率进行绘图
        step1: 图形绘制
        """
        # 输出 log 信息
        print("=============================正在绘图===============================")

        # 设置中文
        mpl.rcParams["font.sans-serif"] = ["Songti SC"]
        mpl.rcParams["axes.unicode_minus"] = False

        # 变量 x
        x = np.arange(3)
        # 变量 y1
        y1 = correctRateLst
        # 变量 y2
        y2 = errorRateLst
        # 设置宽度
        barWidth = 0.35
        # 新建列表接收 tick_label
        tickLabelLst = []
        # for 循环
        for i in range(1, len(correctRateLst) + 1):
            # 变量 item
            tick_label = "corpus{}".format(i)
            # 列表添加元素
            tickLabelLst.append(tick_label)

        # 绘制条形图
        plt.bar(x, y1, barWidth, align = "center", color = "c", label = "正确识别语料比率", alpha = 0.5)
        plt.bar(x + barWidth, y2, barWidth, color = "b", align = "center", label = "错误识别语料比率", alpha = 0.5)

        # 设置 label
        plt.xlabel("相关语料")
        plt.ylabel("比率")

        plt.xticks(x + barWidth/2, tickLabelLst)
        plt.legend()
        # 图片展示
        plt.show()

        # 输出 log 信息
        print("=============================退出绘图===============================")


    def main(self):
        """
        main 函数：
        input:
        output: 数据持久化、绘图
        features: 运行主体代码
        step1: 判断存放最终数据的目录是否存在，不存在则创建
        step2: 停用词处理、接口测试、文件写入、绘图
        step3: 将最终得到的数据持久化
        """
        # 输出 log 信息
        print("==========================判断文件存在性============================")
        # 判断文件是否存在，若不存在则创建
        if os.path.exists(self.dirPath):
            # 如果目录存在则不创建，并提示目录已存在
            print(self.dirPath + "：目录已存在")
        else:
            os.makedirs(self.dirPath)
            print(self.dirPath + "：目录创建成功")
        # 输出 log 信息
        print("=============================判断结束===============================")
        # 遍历
        for stopWord in self.stopWordList:
            # 停用词处理
            df = self.dealStopWord(stopWord = stopWord)
            # 语料测试
            test_df = self.dataTest(df = df)
            # 语料评估
            correctRate, errorRate, correctDictLst, errorDictLst = self.dataEvaluation(stopWord = stopWord, df = test_df, countCorrect = 0, countError = 0)
            # 存放正确识别语料的文件路径
            correctFilePath = os.path.join(self.dirPath, stopWord + "_correct.csv")
            # 存放错误识别语料的文件路径
            errorFilePath = os.path.join(self.dirPath, stopWord + "_error.csv")
            # 输出 log 信息
            print("=============================写入文件===============================")
            # 文件写入
            self.WriteData(correctFilePath, correctDictLst)
            # 文件写入
            self.WriteData(errorFilePath, errorDictLst)
            # 输出 log 信息
            print("=============================写入完成===============================")
            # 列表添加元素
            self.correctRateLst.append(correctRate)
            # 列表添加元素
            self.errorRateLst.append(errorRate)

        # 绘图
        self.drawPicture(self.correctRateLst, self.errorRateLst)





if __name__ == '__main__':
    stopWordTest = StopWordTest()
    stopWordTest.main()
    # 输出 log 信息
    print("#" * 30, end = "")
    # 输出 log 信息
    print("The process of stopWordTest finished!", end = "")
    # 输出 log 信息
    print("#" * 30)