# -*- 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)