冲突解决

src/main/java/com/jz/common/utils/FlowParseTool.java

@@ -13,11 +13,13 @@ import org.apache.commons.lang3.time.DateFormatUtils;
 import org.springframework.data.redis.core.RedisTemplate;
 
 import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
 import com.jz.common.enums.NodeChangeTypeEnum;
 import com.jz.dmp.modules.controller.DataIntegration.bean.flow.FlowNode;
 import com.jz.dmp.modules.controller.DataIntegration.bean.flow.FlowNodeChangeInfo;
 import com.jz.dmp.modules.controller.DataIntegration.bean.flow.FlowPro;
 import com.jz.dmp.modules.controller.projconfig.bean.DmpProjectConfigInfoDto;
+import com.jz.dmp.modules.model.DmpDevelopTask;
 import com.jz.dmp.modules.model.DmpNavigationTree;
 import com.jz.dmp.modules.model.DmpProject;
 import com.jz.dmp.modules.model.DmpProjectSystemInfo;
@@ -432,6 +434,64 @@ public class FlowParseTool {
                 //子流程
                 azkabanJobType = "flow";
                 azkabanJobCommand = generateSubprocessFile(flowNode);
+            } else if ("ftp".equalsIgnoreCase(nodeType)) {
+            	//ftp
+				//azkabanJobType = "command";
+				//azkabanJobCommand = "";
+				
+				//上传ftp下载相关参数
+            	JSONObject scriptJsonObject = JSONObject.parseObject(flowNode.getScript());
+				//FTP链接
+            	contents.add("ftpUrl=" + scriptJsonObject.getString("ftpUrl"));
+				//FTP用户名
+            	contents.add("ftpUsername=" + scriptJsonObject.getString("ftpUsername"));
+				//FTP密码
+            	contents.add("ftpPassword=" + scriptJsonObject.getString("ftpPassword"));
+				//FTP文件目录
+            	contents.add("ftpSourceFileDir=" + scriptJsonObject.getString("ftpSourceFileDir"));
+				//保存目录
+            	contents.add("ftpSaveDestDir=" + scriptJsonObject.getString("ftpSaveDestDir"));
+			}else if ("unzipFile".equalsIgnoreCase(nodeType)) {
+				//unzipFile
+				//azkabanJobType = "command";
+				//azkabanJobCommand = "";
+				
+				//上传解压文件相关参数
+            	JSONObject scriptJsonObject = JSONObject.parseObject(flowNode.getScript());
+            	//输出目录
+            	contents.add("zipOutputDir=" + scriptJsonObject.getString("zipOutputDir"));
+            	//压缩文件目录
+            	contents.add("compressedFileDir=" + scriptJsonObject.getString("compressedFileDir"));
+            	//解压格式
+            	contents.add("decompressFormat=" + scriptJsonObject.getString("decompressFormat"));
+			}else if ("docTrans".equalsIgnoreCase(nodeType)) {
+				//docTrans
+				//azkabanJobType = "command";
+				//azkabanJobCommand = "";
+				
+				//上传文件转码相关参数
+            	JSONObject scriptJsonObject = JSONObject.parseObject(flowNode.getScript());
+            	//文件编码
+            	contents.add("documentCode=" + scriptJsonObject.getString("documentCode"));
+            	//输出地址(目录)
+            	contents.add("outpuAddressUrl=" + scriptJsonObject.getString("outpuAddressUrl"));
+            	//输出编码
+            	contents.add("fileCharsetEncode=" + scriptJsonObject.getString("fileCharsetEncode"));
+            	//文件地址(目录)
+            	contents.add("fileAddressUrl=" + scriptJsonObject.getString("fileAddressUrl"));
+			}else if ("hdfs".equalsIgnoreCase(nodeType)) {
+				//docTrans
+				//azkabanJobType = "command";
+				//azkabanJobCommand = "";
+				
+				//上传HDFS相关相关参数
+            	JSONObject scriptJsonObject = JSONObject.parseObject(flowNode.getScript());
+            	//文件地址
+            	contents.add("localUploadFileDir=" + scriptJsonObject.getString("localUploadFileDir"));
+            	//文件过滤
+            	contents.add("hdfsUploadFileFilter=" + scriptJsonObject.getString("hdfsUploadFileFilter"));
+            	//HDFS目录
+            	contents.add("hdfsUploadSaveDir=" + scriptJsonObject.getString("hdfsUploadSaveDir"));
 			}
 
             //子流程类型
@@ -537,7 +597,7 @@ public class FlowParseTool {
     /**
      * 生成同步任务文件
      *
-     * @param flowNode
+     * @param flowNodeq
      * @return
      */
     private String generateSyncFile(FlowNode flowNode)throws Exception {
@@ -565,8 +625,10 @@ public class FlowParseTool {
         //暂时不上传
         //dmpDevelopTaskService.newPublishSyncing(syncTaskTreeId, publishedToProjectId);
         //获取最新版本的同步任务
-        String execXmlFileNameAndVersion = getPublishSyncTaskFileNameAndLatestVersion(taskName, syncTaskTreeId);
-        String execXmlFileName = execXmlFileNameAndVersion.split("@")[1];
+        //String execXmlFileNameAndVersion = getPublishSyncTaskFileNameAndLatestVersion(taskName, syncTaskTreeId);
+        //String execXmlFileName = execXmlFileNameAndVersion.split("@")[1];
+        //DmpDevelopTask task = dmpDevelopTaskService.
+        String execXmlFileName = "";
         //xml  执行xml的命令写到job文件中
         String command = "command=" + dmpProjectConfigInfoDto.getDmpPublicConfigInfoDto().getAzkabanExectorXmlExec() + " " + publishedToProjectId + " ${azkaban.flow.flowid} ${azkaban.job.id} " + execXmlFileName;
         return command;