# -*- coding: utf-8 -*- # @Author: Gree # @Date: 2021-05-31 18:18:57 # @Last Modified by: Gree # @Last Modified time: 2021-06-02 16:42:41 import re import configparser # 第三方库 import pandas as pd from config import conn_sql as cs from data_processing import data_deal as dd from period import get_time as gt from check import airconditioner_check as ac from check import ancient_poem_check as apc from check import fm_check as fc from check import global_control_check as gcc from check import holiday_check as hc from check import music_check as mc from check import news_check as nc from check import play_control_check as pcc from check import science_check as sc from check import sports_check as spc from check import stocks_check as stc from check import translate_check as tc from check import universal_control_check as ucc from check import weather_check as wc from check import encyclopedia_check as ec from check import chat_check as cc class AutoTest: def __init__(self): """ __init__ 函数: input: output: features: 定义初始变量 step1: 定义 self.time_before、self.time_now、self.initial_table、self.insert_table 等对象 """ # 获取 self.time_before, self.time_now self.time_before, self.time_now = gt.GetTime().get_time(days = 0) # 原始数据表 self.initial_table = "ctoc_tb" # 插入数据表 self.insert_table = "final" # sql 语句 self.sql_1 = "show tables;" # sql 语句 self.sql_2 = "select date_time, request_id, mac_wifi, user_id, query, domain, intent, response_text from %s" % self.initial_table if self.time_before == "" else "select date_time, request_id, mac_wifi, user_id, query, domain, intent, response_text from %s where date_time BETWEEN %s and %s" % (self.initial_table, self.time_before, self.time_now) # sql 语句 self.sql_3 = "CREATE TABLE %s (`id` INT PRIMARY KEY AUTO_INCREMENT,`date_time` datetime DEFAULT NULL,`request_id` varchar(50) DEFAULT NULL,`mac_wifi` varchar(50) DEFAULT NULL,`user_id` varchar(50) DEFAULT NULL,`query` varchar(255) DEFAULT NULL,`domain` varchar(50) DEFAULT NULL,`intent` varchar(50) DEFAULT NULL,`response_text` text DEFAULT NULL,`domain_is_right` varchar(50) DEFAULT NULL,`intent_is_right` varchar(50) DEFAULT NULL,`response_is_right` varchar(50) DEFAULT NULL) CHARSET=utf8;" % self.insert_table # sql 语句 self.sql_4 = "INSERT INTO " + self.insert_table + " (date_time, request_id, mac_wifi, user_id, query, domain, intent, response_text, domain_is_right, intent_is_right, response_is_right) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)" # 定义 self.result_list self.result_list = [] def integrate_data(self, df): """ integrate_data 函数: input: df output: df features: 模块的作用是整合数据 """ # 输出 log 信息 print("Loading the module of IntegrateData ...") # 定义 input_df input_df = df.copy() # 遍历 for index, row in input_df.iterrows(): # 获取 domain domain = row['domain'] # 条件判断 if domain == "Airconditioner": # 遍历 for item in ac.AirconditionerCheck().airconditioner_check(row): # 列表添加元素 self.result_list.append(item) elif domain == "ancient_poem": # 遍历 for item in apc.AncientPoemCheck().ancient_poem_check(row): # 列表添加元素 self.result_list.append(item) elif domain == "fm": # 遍历 for item in fc.FmCheck().fm_check(row): # 列表添加元素 self.result_list.append(item) elif domain == "globalctrl": # 遍历 for item in gcc.GlobalControlCheck().global_control_check(row): # 列表添加元素 self.result_list.append(item) elif domain == "holiday": # 遍历 for item in hc.HolidayCheck().holiday_check(row): # 列表添加元素 self.result_list.append(item) elif domain == "music": # 遍历 for item in mc.MusicCheck().music_check(row): # 列表添加元素 self.result_list.append(item) elif domain == "news": # 遍历 for item in nc.NewsCheck().news_check(row): # 列表添加元素 self.result_list.append(item) elif domain == "PlayControl": # 遍历 for item in pcc.PlayControlCheck().play_control_check(row): # 列表添加元素 self.result_list.append(item) elif domain == "science": # 遍历 for item in sc.ScienceCheck().science_check(row): # 列表添加元素 self.result_list.append(item) elif domain == "sports": # 遍历 for item in spc.SportsCheck().sports_check(row): # 列表添加元素 self.result_list.append(item) elif domain == "stock": # 遍历 for item in stc.StockCheck().stock_check(row): # 列表添加元素 self.result_list.append(item) elif domain == "translate": # 遍历 for item in tc.TranslateCheck().translate_check(row): # 列表添加元素 self.result_list.append(item) elif domain == "UniversalControl": # 遍历 for item in ucc.UniversalControlCheck().universal_control_check(row): # 列表添加元素 self.result_list.append(item) elif domain == "weather": # 遍历 for item in wc.WeatherCheck().weather_check(row): # 列表添加元素 self.result_list.append(item) elif domain == "baike": # 遍历 for item in ec.EncyclopediaCheck().encyclopedia_check(row): # 列表添加元素 self.result_list.append(item) elif domain == "chat": # 遍历 for item in cc.ChatCheck().chat_check(row): # 列表添加元素 self.result_list.append(item) else: # 获取 item item = { 'date_time': row['date_time'], 'request_id': row['request_id'], 'mac_wifi': row['mac_wifi'], 'user_id': row['user_id'], 'query': row['query'], 'domain': domain, 'intent': row['intent'], 'response_text': row['response_text'], 'domain_is_right': "", 'intent_is_right': "", 'response_is_right': "" } # 列表添加元素 self.result_list.append(item) # 捕获异常 try: # 将列表转为数据框 df = pd.DataFrame(self.result_list, columns = ['date_time', 'request_id', 'mac_wifi', 'user_id', 'query', 'domain', 'intent', 'response_text', 'domain_is_right', 'intent_is_right', 'response_is_right']) # 数据清洗 df = df.astype(object).where(pd.notnull(df), '') # 数据排序 df = df.sort_values(["date_time"], ascending = True) # 输出 log 信息 print('The dimension of final dataframe:', end = '') # 输出当前数据框的维度 print(df.shape) # 返回 df return df except Exception as e: print("The error of integrate_data():", e) finally: print("Exiting the module of IntegrateData ...") def main(self): """ main 函数: input: output: 数据持久化 features: 将验证过的分类数据重新插入数据库 step1: 连接数据库，获取 initial_data step2: 数据清洗， 获取 initial_df step3: 判断插入数据表是否存在，不存在则新建数据表 step4: 数据插入 """ # 实例化数据库操作类 DbRun = cs.DbRun() # 条件判断 if not DbRun.table_exists(self.initial_table, self.sql_1): # return return # 如果 initial_table 存在，获取 initial_table 中相关字段的信息 initial_data = DbRun.initial_data(self.sql_2) # 实例化一个数据框操作类 DataDeal = dd.DataDeal() # 将获取到的 initial_data 数据进行数据框操作，得到原始的 initial_df 数据框 initial_df = DataDeal.initial_df(initial_data) # 将原始的 initial_df 数据框中的数据进行自动化分类检查，返回一个 output_df 数据框 output_df = self.integrate_data(initial_df) # 条件判断 if not DbRun.table_exists(self.insert_table, self.sql_1): # 如果 insert_table 不存在，则新建 insert_table DbRun.new_table(self.insert_table, self.sql_3) # 将 output_df 数据框中数据批量插入 insert_table DbRun.insert_data(self.sql_4, output_df) else: # 将 output_df 数据框中数据批量插入 insert_table DbRun.insert_data(self.sql_4, output_df) if __name__ == '__main__': autoTest = AutoTest() autoTest.main()