energy-data-trans/test_run_local.py

# -*- coding: utf-8 -*-
# @Time    : 2024/6/11
# @Author  : 魏志亮
import os
import sys


def get_exec_data(batch_no=None, read_type=None, run_count=1):
    if batch_no and read_type:
        data = get_data_by_batch_no_and_type(batch_no, read_type)
        if data is None:
            raise ValueError(f"未找到批次号:{batch_no},类型:{read_type}")

    else:
        data = get_batch_exec_data(run_count)
        if data is None:
            trans_print("当前有任务在执行")
            sys.exit(0)
        elif len(data.keys()) == 0:
            trans_print("当前无任务")
            sys.exit(0)

    return data


def run(data: dict = dict(), save_db=False):
    exec_process = None
    if data['transfer_type'] in ['second', 'minute']:
        exec_process = MinSecTrans(data=data, save_db=save_db)

    if data['transfer_type'] in ['fault', 'warn']:
        exec_process = FaultWarnTrans(data=data, save_db=save_db)

    if exec_process is None:
        raise Exception("No exec process")
    exec_process.run()


if __name__ == '__main__':
    env = None
    if len(sys.argv) >= 2:
        env = sys.argv[1]
    else:
        env = 'prod'
    print(sys.argv)
    if env is None:
        raise Exception("请配置运行环境")

    os.environ['env'] = env

    run_count = 1
    if len(sys.argv) >= 3:
        run_count = int(sys.argv[2])

    conf_path = '/data/config/etl_config.yaml'
    if len(sys.argv) >= 4:
        conf_path = sys.argv[3]

    os.environ['ETL_CONF'] = conf_path

    from utils.log.trans_log import trans_print
    from service.plt_service import get_batch_exec_data, get_data_by_batch_no_and_type
    from etl.wind_power.fault_warn.FaultWarnTrans import FaultWarnTrans
    from etl.wind_power.min_sec.MinSecTrans import MinSecTrans

    data = dict()
    data['batch_code'] = "test"
    data['batch_name'] = "test"
    data['transfer_type'] = "fault"
    # data['transfer_addr'] = r"D:\报警\唐珍风电2023年报警信息.xlsx"
    data['transfer_addr'] = r"D:\故障\故障数据\故障记录_20230101_20240101.csv"
    data['field_code'] = "test"
    data['field_name'] = "唐珍风电故障"

    run(data=data, save_db=False)