脚本数据注入替换

byit-myth-core/myth-core-common/src/main/java/com/byit/job/dto/ScriptParamAndPlaceholderDto.java

+package com.byit.job.dto;
+
+import com.sun.xml.internal.ws.developer.Serialization;
+import lombok.*;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 参数和占位符的处理
+ * @author huangfu
+ */
+@Builder
+@Data
+@ToString
+@NoArgsConstructor
+@AllArgsConstructor
+public class ScriptParamAndPlaceholderDto implements Serializable {
+    /**
+     * 参数的处理
+     */
+    private Map<String,String> param;
+    /**
+     * 占位符的处理
+     */
+    private Map<String,String> Placeholder;
+
+
+}

byit-myth-core/myth-core-common/src/main/java/com/byit/job/dto/plugin/PluginNode.java

 package com.byit.job.dto.plugin;
 
+import com.alibaba.fastjson.JSON;
+import com.byit.job.dto.ScriptParamAndPlaceholderDto;
+import com.sun.istack.internal.NotNull;
 import lombok.Data;
+import lombok.NonNull;
 
 import java.io.Serializable;
 
@@ -56,5 +60,13 @@ public class PluginNode extends PluginBaseNode  implements Serializable {
      */
     private PluginNodeConfig config;
 
+    /**
+     * 脚本运行参数信息
+     * @param scriptParam 脚本
+     */
+    public void setScriptParam(@NotNull ScriptParamAndPlaceholderDto scriptParam){
+        this.runParam = JSON.toJSONString(scriptParam);
+    }
+
     private static final long serialVersionUID = 2L;
 }

byit-myth-core/myth-core-common/src/main/java/com/byit/job/utils/PlaceholderUtils.java

+package com.byit.job.utils;
+
+import lombok.extern.java.Log;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * 占位符替换工具类
+ * @author huangfu
+ */
+@Slf4j
+public class PlaceholderUtils {
+
+    /**
+     * 占位符前缀
+     */
+    private static final String PLACEHOLDER_PREFIX = "${";
+    /**
+     * 占位符后缀
+     */
+    private static final String PLACEHOLDER_SUFFIX = "}";
+
+    /**
+     * 占位符数据替换
+     * @param scriptData
+     * @param parameter
+     * @return
+     */
+    public static byte[] resolvePlaceholders(byte[] scriptData, Map<String,String> parameter){
+        //替换为null 不做操作
+        if(parameter == null || parameter.isEmpty()){
+            log.warn("------------占位符数据为空--------");
+            return scriptData;
+        }
+        //读取脚本文件
+        String scriptContent = new String(scriptData, StandardCharsets.UTF_8);
+        StringBuilder sbt = new StringBuilder(scriptContent);
+        //找到第一个需要替换的位置
+        int startIndex = sbt.indexOf(PLACEHOLDER_PREFIX);
+        //开始循环遍历
+        while (startIndex != -1){
+            //寻找这个替换符 结束的位置
+            int endIndex = sbt.indexOf(PLACEHOLDER_SUFFIX, startIndex + PLACEHOLDER_PREFIX.length());
+            //确定是否存在后缀
+            if(endIndex != -1){
+                //截取变量值
+                String placeholder = sbt.substring(startIndex + PLACEHOLDER_PREFIX.length(), endIndex);
+                //确定下一次查询的位置
+                int nextIndex = endIndex+PLACEHOLDER_SUFFIX.length();
+                //开始从  map中解析
+                if(parameter.containsKey(placeholder)){
+                    String placeholderValue = parameter.get(placeholder);
+                    //替换这个位置
+                    sbt.replace(startIndex,endIndex+PLACEHOLDER_SUFFIX.length(),placeholderValue);
+                    //再次计算替换后的查询开始位置
+                    nextIndex = startIndex+placeholderValue.length();
+                }else{
+                    throw new RuntimeException("站位解析异常");
+                }
+                startIndex = sbt.indexOf(PLACEHOLDER_PREFIX,nextIndex);
+            }
+        }
+        return sbt.toString().getBytes(StandardCharsets.UTF_8);
+    }
+
+    public static void main(String[] args) {
+        String text = "123${name},1232321${sex}123123";
+        Map<String,String> map = new HashMap<String,String>();
+        map.put("name","皇甫科星");
+        map.put("sex","'男'");
+        byte[] bytes = resolvePlaceholders(text.getBytes(StandardCharsets.UTF_8), map);
+
+        System.out.println(new String(bytes, StandardCharsets.UTF_8));
+    }
+
+
+}

byit-myth-core/myth-executor-core/src/test/java/com/test/Test1.java

@@ -13,6 +13,7 @@ import java.util.Map;
 public class Test1 {
     public static void main(String[] args) throws IOException, MyException {
         FastDfsFileSystem fds = new FastDfsFileSystem();
+        fds.fileRemove("ddmp/M00/00/00/CgB4Al5XoBSAZg1qAAAAW4eTPeg3226.py");
         byte[] bytes = FileUtils.readFileToByteArray(new File("D:\\2020project\\byit-myth-job\\byit-myth-core\\myth-executor-core\\src\\test\\java\\com\\test\\test-ex.py"));
         Map<String,String> map = new HashMap<>();
         map.put("filename","test-ex.py");

byit-myth-core/myth-executor-core/src/test/java/com/test/test-ex.py

@@ -3,5 +3,5 @@ import time
 i = 0
 while i<100:
     time.sleep(1)
-    print("--------%d" %i)
+    print("-----${name}---%d" %i)
     i+=1
\ No newline at end of file

byit-myth-executor/myth-executor-server/src/main/java/com/byit/service/ScriptExecutorServiceImpl.java

@@ -8,13 +8,14 @@ import com.byit.filesystem.FileSystem;
 import com.byit.job.dto.DispatchResponseDto;
 import com.byit.job.dto.JobRunResultDto;
 import com.byit.job.dto.ScriptDto;
+import com.byit.job.dto.ScriptParamAndPlaceholderDto;
 import com.byit.job.enums.JobResultEnum;
+import com.byit.job.utils.PlaceholderUtils;
 import com.byit.job.vo.ReturnResult;
 import com.byit.pool.RunThreadPool;
 import com.byit.rpc.remoting.provider.annotation.RpcService;
 import com.byit.utils.ServiceInfoUtil;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.io.IOUtils;
 import org.csource.common.MyException;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
@@ -63,7 +64,15 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
 
                 String remotePath = scriptDto.getRemotePath();
                 String command = scriptDto.getCommand();
-                String scriptPath = byteArrayToFile(remotePath);
+                String param = scriptDto.getParam();
+
+
+                ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = null;
+                if(null != param){
+                    scriptParamAndPlaceholderDto = JSON.parseObject(param, ScriptParamAndPlaceholderDto.class);
+                }
+
+                String scriptPath = byteArrayToFile(remotePath,scriptParamAndPlaceholderDto);
                 //开始执行脚本
                 List<String> cmdList = Arrays.asList(command,scriptPath);
                 MythJobProcess mythJobProcess = new MythJobProcess(cmdList, null, null, scriptDto.getLogId());
@@ -137,7 +146,7 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
      * 将脚本字节转换成文件
      * @return 生成文件的本地路径
      */
-    private String byteArrayToFile(String remotePath){
+    private String byteArrayToFile(String remotePath,ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto){
         //创建目录
         File rootPathMkdir = new File(rootScriptPath,formatDate());
         if(!rootPathMkdir.exists()){
@@ -150,6 +159,12 @@ public class ScriptExecutorServiceImpl implements ScriptExecutorService {
         try {
             //下载脚本文件
             byte[] scriptByteArray = fileSystem.downloaderFile(remotePath);
+            /**
+             * 替换脚本占位符
+             */
+            if(scriptParamAndPlaceholderDto != null){
+                scriptByteArray = PlaceholderUtils.resolvePlaceholders(scriptByteArray,scriptParamAndPlaceholderDto.getPlaceholder());
+            }
             //获取文件元信息
             Map<String, String> fileMate = fileSystem.getFileMate(remotePath);
             String fileName = fileMate.get("filename");

demo-client/byit-demo-client/src/main/java/com/byit/job/ScriptAddFlow.java

 package com.byit.job;
 
+import com.alibaba.fastjson.JSON;
+import com.byit.job.dto.ScriptParamAndPlaceholderDto;
 import com.byit.job.dto.plugin.*;
 import com.byit.rpc.remoting.invoker.route.LoadBalance;
 import com.byit.utils.JobUtils;
 
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
+import java.util.*;
 import java.util.concurrent.TimeUnit;
 
 public class ScriptAddFlow {
@@ -70,9 +70,15 @@ public class ScriptAddFlow {
         pluginNode2.setType("node");
         pluginNode2.setAuthor("皇甫");
         pluginNode2.setJobType("SCRIPT");
-        pluginNode2.setScriptUrls("ddmp/M00/00/00/CgB4Al5XoBSAZg1qAAAAW4eTPeg3226.py");
+        pluginNode2.setScriptUrls("ddmp/M00/00/00/CgB4Al5d3xmAdhrwAAAAYsb__Ck6673.py");
+
+        ScriptParamAndPlaceholderDto scriptParamAndPlaceholderDto = new ScriptParamAndPlaceholderDto();
+        Map<String,String> map = new HashMap<>();
+        map.put("name","皇甫科星");
+        scriptParamAndPlaceholderDto.setPlaceholder(map);
+        pluginNode2.setScriptParam(scriptParamAndPlaceholderDto);
+
         pluginNode2.setRunCommand("python");
-        pluginNode2.setRunParam("test1");
         pluginNodeConfig2.setFailedRetryCount(2);
         pluginNodeConfig2.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
         pluginNodeConfig2.setNodeCron("0 0/7 * * * ? *");