基础镜像更新，标签为csst

L1/dockerfiles/message-router/Dockerfile

-FROM hub.cstcloud.cn/scalebox/agent
+FROM hub.cstcloud.cn/scalebox/agent:csst
 
 LABEL maintainer="Zhang Xiaoli <zhangxiaoli@cnic.cn>"
 

L1/dockerfiles/message-router/messageRoute.py

@@ -16,16 +16,20 @@ class messageRoute():
         #解析message中的dag
         dag = os.getenv("DAG_NAME", "csst-msc-l1-mbi")
         dagfile = "/dag-yaml/"+dag+".yml"
-
+       
         fixed_headers = headers.replace('\\', '')
         fixed_headers = fixed_headers.replace('"["', '["').replace('"]"', '"]')
-        print(f"fixed_headers: {headers}")
+        print(f"fixed_headers: {fixed_headers}")
 
         # 改进的JSON解析逻辑
         try:
             headersstr = json.loads(fixed_headers)
+
+            if "data_list" in headersstr:
+                data_list=headersstr["data_list"]
+
         except json.JSONDecodeError as e:
-            print(f"Invalid JSON format in headers: {e}, headers: {fixed_headers}")
+            print(f"Invalid JSON format in headers: {e}, headers: {headersstr}")
             return
 
         if headers =="null" or "from_job" not in headersstr:
@@ -44,7 +48,7 @@ class messageRoute():
                 data_list = headersstr.get("data_list")
                 print("from_job :"+from_job+" dag_run_id :"+dag_run_id+" sorted_tag :"+sorted_tag+"data_list :"+data_list)
                 messageRoute.sendsinkjobs(dagfile,sorted_tag,dag_run_id,data_list,from_job,message)
-              
+                
             except json.JSONDecodeError as e:
                 print("Invalid JSON format in headers:", e)
     @classmethod
@@ -93,7 +97,10 @@ class messageRoute():
             command = f"scalebox task add --header sorted_tag={sorted_tag} --header dag_run_id={dag_run_id} --header repeatable=yes --upsert --sink-job={job} {message}"
         else:
             print("data_list 有值")
-            command = f"scalebox task add --header sorted_tag={sorted_tag} --header dag_run_id={dag_run_id} --header repeatable=yes --header data_list={data_list} --upsert --sink-job={job} {message}"
+            # data_list 去掉空格
+            str_data_list=str(data_list)
+            cleaned_data_list = str_data_list.replace(' ', '')
+            command = f"scalebox task add --header sorted_tag={sorted_tag} --header dag_run_id={dag_run_id} --header repeatable=yes --header data_list={cleaned_data_list} --upsert --sink-job={job} {message}"
         
         print("command : "+command)
         result=subprocess.run(command, shell=True)