import datetime import traceback from etl.common.ClearData import ClearData from etl.common.PathsAndTable import PathsAndTable from etl.common.SaveToDb import SaveToDb from etl.common.UnzipAndRemove import UnzipAndRemove from service.plt_service import get_all_wind, update_trans_status_success, update_trans_status_error, \ update_trans_status_running from utils.file.trans_methods import read_excel_files from utils.log.trans_log import trans_print, set_trance_id class BaseDataTrans(object): def __init__(self, data: dict = None, save_db=True, step=0, end=4): self.batch_no = data['batch_code'] self.batch_name = data['batch_name'] self.read_type = data['transfer_type'] self.read_path = data['transfer_addr'] self.field_code = data['field_code'] self.field_name = data['field_name'] self.save_zip = False self.step = step self.end = end self.wind_col_trans, self.rated_power_and_cutout_speed_map = get_all_wind(self.field_code) self.batch_count = 100000 self.save_db = save_db self.filed_conf = self.get_filed_conf() self.pathsAndTable = PathsAndTable(self.batch_no, self.batch_name, self.read_path, self.field_name, self.read_type, save_db, self.save_zip) def get_filed_conf(self): raise NotImplementedError("需要实现 获取点检表 方法") # 第一步 清理数据 def clean_file_and_db(self): clean_data = ClearData(self.pathsAndTable) clean_data.run() # 第二步 解压 移动到临时文件 def unzip_or_remove_to_tmp_dir(self): # 解压并删除 unzip_and_remove = UnzipAndRemove(self.pathsAndTable) unzip_and_remove.run() # 第三步 读取 并 保存到临时文件 def read_and_save_tmp_file(self): raise NotImplementedError("第三步未做实现") # 第四步 统计 并 保存到正式文件 def statistics_and_save_to_file(self): raise NotImplementedError("第四步未做实现") # 第五步 保存到数据库 def save_to_db(self): save_to_db = SaveToDb(self.pathsAndTable, self.batch_count) save_to_db.run() # 最后更新执行程度 def update_exec_progress(self): update_trans_status_success(self.batch_no, self.read_type, len(read_excel_files(self.pathsAndTable.get_save_path())), None, None, None, None, self.save_db) def run(self): total_begin = datetime.datetime.now() try: trance_id = '-'.join([self.batch_no, self.field_name, self.read_type]) set_trance_id(trance_id) update_trans_status_running(self.batch_no, self.read_type, self.save_db) if self.step <= 0 and self.end >= 0: begin = datetime.datetime.now() trans_print("开始清理数据,临时文件夹:", self.pathsAndTable.get_tmp_path()) self.clean_file_and_db() trans_print("清理数据结束,耗时:", datetime.datetime.now() - begin, "总耗时:", datetime.datetime.now() - begin) if self.step <= 1 and self.end >= 1: begin = datetime.datetime.now() trans_print("开始解压移动文件") self.unzip_or_remove_to_tmp_dir() trans_print("解压移动文件结束:耗时:", datetime.datetime.now() - begin, "总耗时:", datetime.datetime.now() - begin) if self.step <= 2 and self.end >= 2: begin = datetime.datetime.now() trans_print("开始保存数据到临时文件") self.read_and_save_tmp_file() trans_print("保存数据到临时文件结束,耗时:", datetime.datetime.now() - begin, "总耗时:", datetime.datetime.now() - begin) if self.step <= 3 and self.end >= 3: begin = datetime.datetime.now() trans_print("开始保存数据到正式文件") self.statistics_and_save_to_file() trans_print("保存数据到正式文件结束,耗时:", datetime.datetime.now() - begin, "总耗时:", datetime.datetime.now() - begin) if self.step <= 4 and self.end >= 4: begin = datetime.datetime.now() trans_print("开始保存到数据库,是否存库:", self.pathsAndTable.save_db) self.save_to_db() trans_print("保存到数据结束,耗时:", datetime.datetime.now() - begin, "总耗时:", datetime.datetime.now() - begin) self.update_exec_progress() except Exception as e: trans_print(traceback.format_exc()) update_trans_status_error(self.batch_no, self.read_type, str(e), self.save_db) raise e finally: # self.pathsAndTable.delete_tmp_files() trans_print("执行结束,总耗时:", str(datetime.datetime.now() - total_begin)) if __name__ == '__main__': test = BaseDataTrans(save_db=False, batch_no="WOF053600062-WOB000010", read_type="fault") test.run()