修改

app_run.py

@@ -42,13 +42,13 @@ def run(batch_no=None, read_type=None, save_db=True, run_count=1):
 
 if __name__ == '__main__':
 
-    conf_path = '/data/config/etl_config.yaml'
+    env = 'dev'
     if len(sys.argv) >= 2:
-        conf_path = sys.argv[1]
+        env = sys.argv[1]
 
+    conf_path = os.path.abspath(f"./conf/etl_config_{env}.yaml")
     os.environ['ETL_CONF'] = conf_path
     yaml_config = yaml_conf(conf_path)
-    env = read_conf(yaml_config, "env", "dev")
     os.environ['env'] = env
     run_count = int(read_conf(yaml_config, "run_batch_count", 1))
 

conf/etl_config_datang.yaml

@@ -21,6 +21,4 @@ log_path_dir: /data/collection_data/logs
 # 临时文件存放处,有些甲方公司隔得tmp太小,只好自己配置
 tmp_base_path: /data/collection_data/tmp
 
-run_batch_count: 1
-# 执行环境
-env: datang
+run_batch_count: 1

conf/etl_config_dev.yaml

@@ -23,6 +23,4 @@ log_path_dir: /data/logs
 # 临时文件存放处,有些甲方公司隔得tmp太小,只好自己配置
 tmp_base_path: /tmp
 
-run_batch_count: 1
-# 执行环境
-env: dev
+run_batch_count: 1

conf/etl_config_prod.yaml

@@ -23,6 +23,4 @@ log_path_dir: /data/logs
 # 临时文件存放处,有些甲方公司隔得tmp太小,只好自己配置
 tmp_base_path: /tmp
 
-run_batch_count: 2
-# 执行环境
-env: prod
+run_batch_count: 2

etl/common/PathsAndTable.py

@@ -19,7 +19,7 @@ class PathsAndTable(object):
         self.multi_pool_count = 6
         self.is_delete_db = False
 
-        yaml_config = yaml_conf(os.environ.get('ETL_CONF', r"/data/config/etl_config.yaml"))
+        yaml_config = yaml_conf(os.environ.get('ETL_CONF'))
 
         save_path_conf = read_conf(yaml_config, "save_path")
         self.tmp_base_path = read_conf(yaml_config, "tmp_base_path", "/tmp")

etl/wind_power/min_sec/StatisticsAndSaveFile.py

@@ -94,11 +94,12 @@ class StatisticsAndSaveFile(object):
         solve_time_begin = datetime.datetime.now()
         # df = df[(df['time_stamp'].str.find('-') > 0) & (df['time_stamp'].str.find(':') > 0)]
         # trans_print(wind_col_name, "去掉非法时间后大小:", df.shape[0])
-        df['time_stamp'] = pd.to_datetime(df['time_stamp'], errors="coerce")
+        df['time_stamp'] = pd.to_datetime(df['time_stamp'], errors="coerce", format='%d-%m-%Y %H:%M:%S')
         df.dropna(subset=['time_stamp'], inplace=True)
         df.sort_values(by='time_stamp', inplace=True)
         df = df[[i for i in self.trans_param.cols_tran.keys() if i in df.columns]]
 
+        df['active_power'] = df['active_power'] / 1000
         ## 做数据检测前,羡强行处理有功功率
         df = df[df['active_power'] < 5000]
 

test_run_local.py

@@ -43,18 +43,17 @@ def run(data: dict = dict(), save_db=False):
 
 
 if __name__ == '__main__':
-    conf_path = r'conf/etl_config.yaml'
+    env = 'dev'
     if len(sys.argv) >= 2:
-        conf_path = sys.argv[1]
+        env = sys.argv[1]
 
-    begin = datetime.datetime.now()
-    print("开始执行:", begin)
+    conf_path = os.path.abspath(f"./conf/etl_config_{env}.yaml")
     os.environ['ETL_CONF'] = conf_path
     yaml_config = yaml_conf(conf_path)
-    env = read_conf(yaml_config, "env", "dev")
     os.environ['env'] = env
     run_count = int(read_conf(yaml_config, "run_batch_count", 1))
 
+
     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

utils/db/ConnectMysql.py

@@ -13,7 +13,7 @@ from utils.log.trans_log import trans_print
 class ConnectMysql:
 
     def __init__(self, connet_name):
-        self.yaml_data = yaml_conf(os.environ.get('ETL_CONF', "/data/config/etl_config.yaml"))
+        self.yaml_data = yaml_conf(os.environ.get('ETL_CONF'))
         self.connet_name = connet_name
         self.config = self.yaml_data[self.connet_name]
 

utils/log/trans_log.py

@@ -35,7 +35,7 @@ stout_handle.setLevel(logging.INFO)
 stout_handle.addFilter(ContextFilter())
 logger.addHandler(stout_handle)
 
-config = yaml_conf(os.environ.get('ETL_CONF', r"/data/config/etl_config.yaml"))
+config = yaml_conf(os.environ.get('ETL_CONF'))
 log_path_dir = read_conf(config, 'log_path_dir', "/data/logs")
 
 log_path = log_path_dir + os.sep + r'etl_tools_' + (os.environ['env'] if 'env' in os.environ else 'dev')