测试报警功能

write_redis/alg_ad_feature_03_cid2actionv1_redis.py

@@ -158,11 +158,11 @@ def main():
     # 1 判断上游数据表是否生产完成
     project = "loghubods"
     table = "alg_ad_feature_cid_action_v1"
-    table_data_cnt = check_data(project, table, date, hour, mm)
-    run_flag = table_data_cnt == 0
+    run_flag = True
     begin_ts = int(time.time())
+    table_data_cnt = 0
     while run_flag:
-        if int(time.time()) - begin_ts >= 60*50:
+        if int(time.time()) - begin_ts >= 60 * 40:
             log_.info("等待上游数据超过50分钟了，认为失败退出:过了{}秒。".format(int(time.time()) - begin_ts))
             exit(999)
         table_data_cnt = check_data(project, table, date, hour, mm)

write_redis/alg_ad_feature_04_vidcid2actionv1_redis.py

@@ -7,6 +7,7 @@ if root_dir not in sys.path:
     print("******** sys.path ********")
     print(sys.path)
 
+from multiprocessing import Process
 from odps import ODPS
 from threading import Timer
 import threading
@@ -17,6 +18,7 @@ import json
 from datetime import datetime
 from queue import Queue
 from tqdm import tqdm
+import time
 
 
 
@@ -151,33 +153,51 @@ def main():
         mm = "00"
         log_.info("没有读取到参数，采用系统时间：{}".format(e))
     log_.info("使用时间参数-日期：{}，小时：{}".format(date, str(hour)))
-    if hour in ["00", "01"]:
+    if hour in ["00", "01", "02", "03", "04", "05", "06"]:
         log_.info(f"hour={hour}不执行，直接返回。")
         return
     # 1 判断上游数据表是否生产完成
     project = "loghubods"
     table = "alg_ad_feature_vidcid_action_v1"
-    table_data_cnt = check_data(project, table, date, hour, mm)
-    if table_data_cnt == 0:
-        log_.info("上游数据{}未就绪{}/{}，等待...".format(table, date, hour))
-        Timer(60, main).start()
-    else:
-        log_.info("上游数据就绪，count={}，开始读取数据表".format(table_data_cnt))
-        # 2 读取数据表 处理特征
-        video_list = get_sql(project, table, date, hour, mm)
-        # 3 写入redis
-        log_.info("video的数据量:{}".format(len(video_list)))
-        records_process_for_list(video_list, process_and_store, max_size=50, num_workers=8)
+    run_flag = True
+    begin_ts = int(time.time())
+    table_data_cnt = 0
+    while run_flag:
+        if int(time.time()) - begin_ts >= 60 * 40:
+            log_.info("等待上游数据超过50分钟了，认为失败退出:过了{}秒。".format(int(time.time()) - begin_ts))
+            exit(999)
+        table_data_cnt = check_data(project, table, date, hour, mm)
+        if table_data_cnt == 0:
+            log_.info("上游数据{}未就绪{}/{}，等待...".format(table, date, hour))
+            log_.info("等待2分钟")
+            time.sleep(60 * 2)
+        else:
+            run_flag = False
+
+    log_.info("上游数据就绪，count={}，开始读取数据表".format(table_data_cnt))
+    # 2 读取数据表 处理特征
+    video_list = get_sql(project, table, date, hour, mm)
+    # 3 写入redis
+    log_.info("video的数据量:{}".format(len(video_list)))
+    records_process_for_list(video_list, process_and_store, max_size=50, num_workers=8)
 
 if __name__ == '__main__':
     log_.info("开始执行:" + datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
-    main()
+    process = Process(target=main)
+    process.start()
+    # 等待子进程完成或超时
+    timeout = 3600
+    process.join(timeout=timeout)  # 设置超时为3600秒（1小时）
+    if process.is_alive():
+        print("脚本执行时间超过1小时，执行失败，经过了{}秒。".format(timeout))
+        process.terminate()  # 终止子进程
+        exit(999)  # 直接退出主进程并返回状态码999
     log_.info("完成执行:" + datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
 
 
 
 # cd /root/zhangbo/rov-offline
-# python alg_ad_feature_04_vidcid2actionv1_redis.py 20240530 20
+# python alg_ad_feature_04_vidcid2actionv1_redis.py 20240605 19
 
 """
     ！！！！！！！！！！！！！！