# -*- coding: utf-8 -*- # @Time : 2024/5/15 # @Author : 魏志亮 import copy import datetime import multiprocessing import sys import tempfile from base.TranseParam import TranseParam from utils.db.trans_mysql import creat_table_and_add_partition, update_trans_status, get_all_wind, \ rename_table, read_excel_and_save_to_db from utils.log.trans_log import logger from utils.trans_methods import * from utils.zip.unzip import unzip, unrar class WindFarms(object): def __init__(self, name, batch_no=None, field_code=None, params: TranseParam = None, wind_full_name=None): self.name = name self.batch_no = batch_no self.field_code = field_code self.wind_full_name = wind_full_name self.begin = datetime.datetime.now() self.save_zip = False self.trans_param = params self.__exist_wind_names = set() self.wind_col_trans = get_all_wind(self.field_code) self.batch_count = 50000 self.save_path = None def set_trans_param(self, params: TranseParam): self.trans_param = params read_path = str(params.read_path) if read_path.find(self.wind_full_name) == -1: message = "读取路径与配置路径不匹配:" + self.trans_param.read_path + ",配置文件为:" + self.wind_full_name update_trans_status(self.batch_no, self.trans_param.read_type, "error", message) raise ValueError(message) self.save_path = os.path.join(read_path[0:read_path.find(self.wind_full_name)], self.wind_full_name, "清理数据") def __params_valid(self, not_null_list=list()): for arg in not_null_list: if arg is None or arg == '': raise Exception("Invalid param set :" + arg) def __get_save_path(self): return os.path.join(self.save_path, self.batch_no, self.trans_param.read_type) def __get_save_tmp_path(self): return os.path.join(tempfile.gettempdir(), self.wind_full_name, self.batch_no, self.trans_param.read_type) def __get_excel_tmp_path(self): return os.path.join(self.__get_save_tmp_path(), 'excel_tmp' + os.sep) def __get_read_tmp_path(self): return os.path.join(self.__get_save_tmp_path(), 'read_tmp') def __df_save_to_tmp_file(self, df=pd.DataFrame(), file=None): if self.trans_param.is_vertical_table: pass else: # 转换字段 if self.trans_param.cols_tran: cols_tran = self.trans_param.cols_tran real_cols_trans = dict() for k, v in cols_tran.items(): if v and not v.startswith("$"): real_cols_trans[v] = k trans_print("包含转换字段,开始处理转换字段") df.rename(columns=real_cols_trans, inplace=True) if self.trans_param.wind_col in real_cols_trans.keys(): self.trans_param.wind_col = real_cols_trans[self.trans_param.wind_col] del_keys = set(df.columns) - set(cols_tran.keys()) for key in del_keys: df.drop(key, axis=1, inplace=True) df = del_blank(df, ['wind_turbine_number']) self.__save_to_tmp_csv(df, file) def __get_excel_files(self): if os.path.isfile(self.trans_param.read_path): all_files = [self.trans_param.read_path] else: all_files = read_files(self.trans_param.read_path) to_path = self.__get_excel_tmp_path() for file in all_files: if str(file).endswith("zip"): if str(file).endswith("csv.zip"): copy_to_new(file, file.replace(self.trans_param.read_path, to_path).replace("csv.zip", 'csv.gz')) else: is_success, e = unzip(file, file.replace(self.trans_param.read_path, to_path).split(".")[0]) self.trans_param.has_zip = True if not is_success: raise e elif str(file).endswith("rar"): is_success, e = unrar(file, file.replace(self.trans_param.read_path, to_path).split(".")[0]) self.trans_param.has_zip = True if not is_success: raise e else: copy_to_new(file, file.replace(self.trans_param.read_path, to_path)) return read_excel_files(to_path) def __read_excel_to_df(self, file): read_cols = [v for k, v in self.trans_param.cols_tran.items() if v and not v.startswith("$")] trans_dict = {} for k, v in self.trans_param.cols_tran.items(): if v and not str(v).startswith("$"): trans_dict[v] = k if self.trans_param.is_vertical_table: vertical_cols = self.trans_param.vertical_cols df = read_file_to_df(file, vertical_cols) df = df[df[self.trans_param.vertical_key].isin(read_cols)] df.rename(columns={self.trans_param.cols_tran['wind_turbine_number']: 'wind_turbine_number', self.trans_param.cols_tran['time_stamp']: 'time_stamp'}, inplace=True) df[self.trans_param.vertical_key] = df[self.trans_param.vertical_key].map(trans_dict).fillna( df[self.trans_param.vertical_key]) return df else: trans_dict = dict() for k, v in self.trans_param.cols_tran.items(): if v and v.startswith("$"): trans_dict[v] = k if self.trans_param.merge_columns: df = read_file_to_df(file) else: if self.trans_param.need_valid_cols: df = read_file_to_df(file, read_cols) else: df = read_file_to_df(file) # 处理列名前缀问题 if self.trans_param.trans_col_exec: columns_dict = dict() for column in df.columns: columns_dict[column] = eval(self.trans_param.trans_col_exec) df.rename(columns=columns_dict, inplace=True) for k, v in trans_dict.items(): if k.startswith("$file"): file_name = ".".join(os.path.basename(file).split(".")[0:-1]) if k == "$file": df[v] = str(file_name) else: datas = str(k.replace("$file", "").replace("[", "").replace("]", "")).split(":") if len(datas) != 2: raise Exception("字段映射出现错误 :" + str(trans_dict)) df[v] = str(file_name[int(datas[0]):int(datas[1])]).strip() elif k.startswith("$folder"): folder = file cengshu = int(str(k.replace("$folder", "").replace("[", "").replace("]", ""))) for i in range(cengshu): folder = os.path.dirname(folder) df[v] = str(str(folder).split(os.sep)[-1]).strip() return df def __save_to_tmp_csv(self, df, file): trans_print("开始保存", str(file), "到临时文件成功") names = set(df['wind_turbine_number'].values) for name in names: save_name = str(name) + '.csv' save_path = os.path.join(self.__get_read_tmp_path(), save_name) create_file_path(save_path, is_file_path=True) if name in self.__exist_wind_names: df[df[self.trans_param.wind_col] == name].to_csv(save_path, index=False, encoding='utf8', mode='a', header=False) else: self.__exist_wind_names.add(name) df[df[self.trans_param.wind_col] == name].to_csv(save_path, index=False, encoding='utf8') del df trans_print("保存", str(names), "到临时文件成功, 风机数量", len(names)) def save_to_csv(self, filename): df = read_file_to_df(filename) if self.trans_param.is_vertical_table: df = df.pivot_table(index=['time_stamp', 'wind_turbine_number'], columns=self.trans_param.vertical_key, values=self.trans_param.vertical_value, aggfunc='max') # 重置索引以得到普通的列 df.reset_index(inplace=True) for k in self.trans_param.cols_tran.keys(): if k not in df.columns: df[k] = None df = df[self.trans_param.cols_tran.keys()] # 添加年月日 if self.trans_param.time_col: trans_print("包含时间字段,开始处理时间字段,添加年月日", filename) df[self.trans_param.time_col] = pd.to_datetime(df[self.trans_param.time_col]) df['year'] = df[self.trans_param.time_col].dt.year df['month'] = df[self.trans_param.time_col].dt.month df['day'] = df[self.trans_param.time_col].dt.day df.sort_values(by=self.trans_param.time_col, inplace=True) df[self.trans_param.time_col] = df[self.trans_param.time_col].apply( lambda x: x.strftime('%Y-%m-%d %H:%M:%S')) trans_print("处理时间字段结束") # 转化风机名称 trans_print("开始转化风机名称") if self.trans_param.wind_name_exec: exec_str = f"df[self.trans_param.wind_col].apply(lambda wind_name: {self.trans_param.wind_name_exec} )" df[self.trans_param.wind_col] = eval(exec_str) df[self.trans_param.wind_col] = df[self.trans_param.wind_col].map( self.wind_col_trans).fillna( df[self.trans_param.wind_col]) trans_print("转化风机名称结束") wind_col_name = str(df[self.trans_param.wind_col].values[0]) if self.save_zip: save_path = os.path.join(self.__get_save_path(), str(wind_col_name) + '.csv.gz') else: save_path = os.path.join(self.__get_save_path(), str(wind_col_name) + '.csv') create_file_path(save_path, is_file_path=True) if self.save_zip: df.to_csv(save_path, compression='gzip', index=False, encoding='utf-8') else: df.to_csv(save_path, index=False, encoding='utf-8') del df trans_print("保存" + str(filename) + ".csv成功") def read_all_files(self): # 读取文件 try: all_files = self.__get_excel_files() trans_print('读取文件数量:', len(all_files)) except Exception as e: logger.exception(e) message = "读取文件列表错误:" + self.trans_param.read_path + ",系统返回错误:" + str(e) update_trans_status(self.batch_no, self.trans_param.read_type, "error", message) raise e return all_files def __read_file_and_save_tmp(self): all_files = self.read_all_files() if self.trans_param.merge_columns: # with multiprocessing.Pool(6) as pool: # dfs = pool.starmap(self.__read_excel_to_df, [(file,) for file in all_files]) dfs = list() index_keys = [self.trans_param.cols_tran['time_stamp']] wind_col = self.trans_param.cols_tran['wind_turbine_number'] if str(wind_col).startswith("$"): wind_col = 'wind_turbine_number' index_keys.append(wind_col) df_map = dict() for file in all_files: df = self.__read_excel_to_df(file) key = '-'.join(df.columns) if key in df_map.keys(): df_map[key] = pd.concat([df_map[key], df]) else: df_map[key] = df for k, df in df_map.items(): df.drop_duplicates(inplace=True) df.set_index(keys=index_keys, inplace=True) df = df[~df.index.duplicated(keep='first')] dfs.append(df) df = pd.concat(dfs, axis=1) df.reset_index(inplace=True) names = set(df[wind_col].values) try: for name in names: self.__df_save_to_tmp_file(df[df[wind_col] == name], "") except Exception as e: logger.exception(e) message = "合并列出现错误:" + str(e) update_trans_status(self.batch_no, self.trans_param.read_type, "error", message) raise e else: for file in all_files: try: self.__df_save_to_tmp_file(self.__read_excel_to_df(file), file) except Exception as e: logger.exception(e) message = "读取文件错误:" + file + ",系统返回错误:" + str(e) update_trans_status(self.batch_no, self.trans_param.read_type, "error", message) raise e def mutiprocessing_to_save_file(self): # 开始保存到正式文件 trans_print("开始保存到excel文件") all_tmp_files = read_excel_files(self.__get_read_tmp_path()) try: with multiprocessing.Pool(6) as pool: pool.starmap(self.save_to_csv, [(file,) for file in all_tmp_files]) except Exception as e: logger.exception(e) message = "保存文件错误,系统返回错误:" + str(e) update_trans_status(self.batch_no, self.trans_param.read_type, "error", message) raise e trans_print("结束保存到excel文件") def mutiprocessing_to_save_db(self): # 开始保存到SQL文件 trans_print("开始保存到数据库文件") all_saved_files = read_excel_files(self.__get_save_path()) table_name = self.batch_no + "_" + self.trans_param.read_type creat_table_and_add_partition(table_name, len(all_saved_files), self.trans_param.read_type) try: with multiprocessing.Pool(6) as pool: pool.starmap(read_excel_and_save_to_db, [(table_name, file, self.batch_count) for file in all_saved_files]) except Exception as e: logger.exception(e) message = "保存到数据库错误,系统返回错误:" + str(e) update_trans_status(self.batch_no, self.trans_param.read_type, "error", message) raise e trans_print("结束保存到数据库文件") def __rename_file(self): save_path = self.__get_save_path() files = os.listdir(save_path) files.sort(key=lambda x: int(str(x).split(os.sep)[-1].split(".")[0][1:])) for index, file in enumerate(files): file_path = os.path.join(save_path, 'F' + str(index + 1).zfill(3) + ".csv.gz") os.rename(os.path.join(save_path, file), file_path) def delete_batch_files(self): trans_print("开始删除已存在的批次文件夹") if os.path.exists(self.__get_save_path()): shutil.rmtree(self.__get_save_path()) trans_print("删除已存在的批次文件夹") def delete_tmp_files(self): trans_print("开始删除临时文件夹") if os.path.exists(self.__get_excel_tmp_path()): shutil.rmtree(self.__get_excel_tmp_path()) if os.path.exists(self.__get_read_tmp_path()): shutil.rmtree(self.__get_read_tmp_path()) if os.path.exists(self.__get_save_tmp_path()): shutil.rmtree(self.__get_save_tmp_path()) trans_print("删除临时文件夹删除成功") def delete_batch_db(self): table_name = "_".join([self.batch_no, self.trans_param.read_type]) renamed_table_name = "del_" + table_name + "_" + datetime.datetime.now().strftime('%Y%m%d%H%M%S') rename_table(table_name, renamed_table_name) def run(self): trans_print("开始执行", self.name, self.trans_param.read_type) self.delete_batch_files() self.delete_tmp_files() self.delete_batch_db() self.__params_valid([self.name, self.batch_no, self.field_code, self.save_path, self.trans_param.read_type, self.trans_param.read_path, self.trans_param.time_col, self.trans_param.wind_col, self.wind_full_name]) # 更新运行状态到运行中 update_trans_status(self.batch_no, self.trans_param.read_type, "running", "") # 开始读取数据并分类保存临时文件 self.__read_file_and_save_tmp() self.mutiprocessing_to_save_file() self.mutiprocessing_to_save_db() update_trans_status(self.batch_no, self.trans_param.read_type, "success", "", wind_count=len(read_excel_files(self.__get_read_tmp_path()))) self.delete_tmp_files()