去重接口

dataoperation-server/src/main/java/com/zq/dataoperation/controller/ComponentRunController.java

+package com.zq.dataoperation.controller;
+
+import com.zq.common.vo.ResultVo;
+import com.zq.dataoperation.service.CommonQueryService;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import org.springframework.web.bind.annotation.*;
+
+import javax.annotation.Resource;
+import java.util.Map;
+
+@Api(tags = "通用查询")
+@RestController
+@RequestMapping(value = "/data/execute")
+public class ComponentRunController {
+
+    @Resource
+    private CommonQueryService commonQueryService;
+
+    @ApiOperation("去重执行")
+    @PostMapping("/remove")
+    public ResultVo removeRepeat(@RequestBody Map<String, Object> body) throws Exception {
+        return ResultVo.success(commonQueryService.removeRepeat(body));
+    }
+
+}

dataoperation-server/src/main/java/com/zq/dataoperation/service/CommonQueryService.java

@@ -16,12 +16,14 @@ import com.zq.dataoperation.entity.MetaDataMapping;
 import com.zq.dataoperation.entity.QueryDb;
 import com.zq.dataoperation.utils.ConnectionUtil;
 import com.zq.dataoperation.utils.SqlUtils;
+import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.web.bind.annotation.RequestBody;
 
 import java.sql.*;
 
 import javax.annotation.Resource;
+import javax.sql.DataSource;
 import java.util.*;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -177,4 +179,13 @@ public class CommonQueryService extends ServiceImpl<CommonQuerySettingDao, Commo
         String sql="SELECT * FROM "+tableName;
         return commonQuery(sql,id);
     }
+
+    public Object removeRepeat(Map<String, Object> body) throws Exception {
+        JSONObject jsonObject = JSONUtil.parseObj(body);
+        Long datasourceId = Long.valueOf(jsonObject.get("datasourceId").toString());
+        String tableName = jsonObject.get("tableName").toString();
+        String metaData = jsonObject.get("metaData").toString();
+        String sql = "SELECT DISTINCT " + metaData + ",m.* FROM " + tableName+" m";
+        return commonQuery(sql, datasourceId);
+    }
 }

spider-flow-api/src/main/java/com/zq/spiderflow/executor/ShapeExecutor.java

 package com.zq.spiderflow.executor;
 
+import java.sql.SQLException;
 import java.util.Map;
 
 import com.zq.spiderflow.context.SpiderContext;
@@ -35,7 +36,7 @@ public interface ShapeExecutor {
 	 * @param context 爬虫上下文
 	 * @param variables 节点流程的全部变量的集合
 	 */
-	void execute(SpiderNode node, SpiderContext context, Map<String, Object> variables);
+	void execute(SpiderNode node, SpiderContext context, Map<String, Object> variables) throws SQLException;
 
 	default boolean allowExecuteNext(SpiderNode node, SpiderContext context, Map<String, Object> variables){
 		return true;

spider-flow-core/src/main/java/com/zq/spiderflow/core/executor/shape/RemoveDuplicatesExecutor.java

@@ -14,9 +14,11 @@ import org.springframework.stereotype.Component;
 
 import javax.sql.DataSource;
 import java.sql.Connection;
+import java.sql.SQLException;
 import java.util.Map;
 
 /**
+ * 去重组件
  * @author Administrator
  */
 @Component
@@ -46,7 +48,7 @@ public class RemoveDuplicatesExecutor implements ShapeExecutor {
     }
 
     @Override
-    public void execute(SpiderNode node, SpiderContext context, Map<String, Object> variables) {
+    public void execute(SpiderNode node, SpiderContext context, Map<String, Object> variables) throws SQLException {
         Long dsId = Long.valueOf(node.getStringJsonValue(DATASOURCE_ID));
         Connection connection = ConnectionUtil.get(dsId);
         JdbcTemplate template = new JdbcTemplate((DataSource)connection);
@@ -65,6 +67,7 @@ public class RemoveDuplicatesExecutor implements ShapeExecutor {
             } catch (Exception e) {
                 logger.error("获取sql出错,异常信息:{}", e.getMessage(), e);
                 ExceptionUtils.wrapAndThrow(e);
+                connection.close();
             }
         }
     }

spider-flow-core/src/main/java/com/zq/spiderflow/core/executor/shape/ReplaceStringExecutor.java

@@ -14,9 +14,11 @@ import org.springframework.stereotype.Component;
 
 import javax.sql.DataSource;
 import java.sql.Connection;
+import java.sql.SQLException;
 import java.util.Map;
 
 /**
+ * 替换字符串
  * @author Administrator
  */
 @Component
@@ -24,6 +26,9 @@ public class ReplaceStringExecutor implements ShapeExecutor {
     public static final String DATASOURCE_ID = "datasourceId";
     public static final String META_DATA = "metaData";
     public static final String TABLE_NAME = "tableName";
+    public static final String  ORIGIN_STR= "originStr";
+    public static final String REPLACE_STR = "replaceStr";
+
     private static final Logger logger = LoggerFactory.getLogger(ExecuteSQLExecutor.class);
 
     @Override
@@ -46,7 +51,35 @@ public class ReplaceStringExecutor implements ShapeExecutor {
     }
 
     @Override
-    public void execute(SpiderNode node, SpiderContext context, Map<String, Object> variables) {
+    public void execute(SpiderNode node, SpiderContext context, Map<String, Object> variables) throws SQLException {
+        String metaData = node.getStringJsonValue(META_DATA);
+        String tableName= node.getStringJsonValue(TABLE_NAME);
+        String originStr= node.getStringJsonValue(ORIGIN_STR);
+        String replaceStr= node.getStringJsonValue(REPLACE_STR);
+
+        Long dsId = Long.valueOf(node.getStringJsonValue(DATASOURCE_ID));
+        Connection connection = ConnectionUtil.get(dsId);
+        JdbcTemplate template = new JdbcTemplate((DataSource)connection);
+
+        if (dsId!=null) {
+            logger.warn("数据源id不能为空！");
+        } else if (StringUtils.isNotBlank(metaData)) {
+            logger.warn("字段不能为空！");
+        }else if (StringUtils.isNotBlank(tableName)) {
+            logger.warn("表名不能为空！");
+        }else {
+            String sql="UPDATE "+tableName+" SET "+metaData+"=REPLACE("+metaData+",'"+originStr+"','"+replaceStr+"');";
+            try {
+                template.execute(sql);
+            } catch (Exception e) {
+                logger.error("获取sql出错,异常信息:{}", e.getMessage(), e);
+                ExceptionUtils.wrapAndThrow(e);
+                connection.close();
+            }
+
+        }
+
+
 
     }