energy-data-trans/etl/common/BaseDataTrans.py

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()