init

pom.xml

@@ -0,0 +1,139 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <version>2.7.0</version>
+        <relativePath/>
+    </parent>
+
+    <groupId>com.winhc</groupId>
+    <artifactId>winhc_data_push</artifactId>
+    <version>1.0-SNAPSHOT</version>
+
+    <properties>
+        <java.version>1.8</java.version>
+        <mybatis.starter.version>1.2.0</mybatis.starter.version>
+        <druid.version>1.1.5</druid.version>
+        <sdk.version>0.33.7-public</sdk.version>
+    </properties>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>com.aliyun.openservices</groupId>
+            <artifactId>ons-client</artifactId>
+            <version>1.8.4.Final</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-data-mongodb</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.aliyun.odps</groupId>
+            <artifactId>odps-sdk-core</artifactId>
+            <version>${sdk.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>io.springfox</groupId>
+            <artifactId>springfox-boot-starter</artifactId>
+            <version>3.0.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <optional>true</optional>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>dns-cache-manipulator</artifactId>
+            <version>1.5.1</version>
+        </dependency>
+
+        <!-- swagger -->
+        <dependency>
+            <groupId>io.springfox</groupId>
+            <artifactId>springfox-boot-starter</artifactId>
+            <version>3.0.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-core</artifactId>
+            <version>2.7</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-api</artifactId>
+            <version>2.7</version>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework.kafka</groupId>
+            <artifactId>spring-kafka</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>cn.hutool</groupId>
+            <artifactId>hutool-all</artifactId>
+            <version>5.3.5</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.aliyun.oss</groupId>
+            <artifactId>aliyun-sdk-oss</artifactId>
+            <version>3.13.1</version>
+        </dependency>
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.aliyun.odps</groupId>
+            <artifactId>odps-sdk-core</artifactId>
+            <version>${sdk.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>com.aliyun</groupId>
+            <artifactId>aliyun-java-sdk-dataworks-public</artifactId>
+            <version>1.8.3</version>
+        </dependency>
+        <dependency>
+            <groupId>com.aliyun</groupId>
+            <artifactId>aliyun-java-sdk-core</artifactId>
+            <version>4.0.3</version>
+        </dependency>
+        <dependency>
+            <groupId>cn.cocowwy</groupId>
+            <artifactId>cocowwy-dingtalk-robot</artifactId>
+            <version>1.0.3-RELEASE</version>
+        </dependency>
+
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+    </build>
+</project>

src/main/java/com/winhc/data/push/WinhcDataPushApplication.java

@@ -0,0 +1,21 @@
+package com.winhc.data.push;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.scheduling.annotation.EnableScheduling;
+
+/**
+ * @author π
+ * @Description:
+ * @date 2022/7/21 14:04
+ */
+
+
+@SpringBootApplication
+@EnableScheduling
+public class WinhcDataPushApplication {
+
+    public static void main(String[] args) {
+        SpringApplication.run(WinhcDataPushApplication.class, args);
+    }
+}

src/main/java/com/winhc/data/push/bean/DataWorksFlowJob.java

@@ -0,0 +1,27 @@
+package com.winhc.data.push.bean;
+
+import com.winhc.data.push.utils.JsonUtils;
+import lombok.*;
+
+import java.util.List;
+
+/**
+ * @Author: XuJiakai
+ * @Date: 2020/6/24 08:49
+ * @Description:
+ */
+@Getter
+@Setter
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class DataWorksFlowJob {
+    private String project;
+    private String flow;
+    private List<DataWorksFlowTask> task;
+
+    @Override
+    public String toString() {
+        return JsonUtils.jsonObjToString(this);
+    }
+}

src/main/java/com/winhc/data/push/bean/DataWorksFlowTask.java

@@ -0,0 +1,47 @@
+package com.winhc.data.push.bean;
+
+import com.winhc.data.push.utils.JsonUtils;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * @Author: XuJiakai
+ * @Date: 2020/6/24 09:01
+ * @Description:
+ */
+@Getter
+@Setter
+@NoArgsConstructor
+@AllArgsConstructor
+public class DataWorksFlowTask {
+    private String taskName;
+    private List<NodeParam> params;
+
+    public Map<String, String> toNodeParam(Map<String, String> otherParams) {
+        return params.stream()
+                .collect(Collectors.toMap(NodeParam::getNodeId, n -> n.toNodeParam(otherParams), (o1, o2) -> o1));
+    }
+
+    public Map<String, String> toNodeParam(String bizDate) {
+        HashMap<String, String> map = new HashMap<>();
+        map.put("bizdate", bizDate.replace("-", ""));
+        return toNodeParam(map);
+    }
+
+    public Map<String, String> toNodeParam(String bizDate, Map<String, String> otherParams) {
+        otherParams.put("bizdate", bizDate.replace("-", ""));
+        return toNodeParam(otherParams);
+    }
+
+    @Override
+    public String toString() {
+        return JsonUtils.jsonObjToString(this);
+    }
+}

src/main/java/com/winhc/data/push/bean/Entry.java

@@ -0,0 +1,50 @@
+package com.winhc.data.push.bean;
+
+import cn.hutool.core.builder.EqualsBuilder;
+import cn.hutool.core.builder.HashCodeBuilder;
+import com.winhc.data.push.utils.JsonUtils;
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+/**
+ * @Author: XuJiakai
+ * @Date: 2020/6/24 14:05
+ * @Description:
+ */
+@Getter
+@Setter
+@AllArgsConstructor
+@NoArgsConstructor
+public class Entry<K, V> {
+    private K key;
+    private V value;
+
+    @Override
+    public String toString() {
+        return JsonUtils.jsonObjToString(this);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+
+        if (o == null || getClass() != o.getClass()) return false;
+
+        Entry<?, ?> entry = (Entry<?, ?>) o;
+
+        return new EqualsBuilder()
+                .append(key, entry.key)
+                .append(value, entry.value)
+                .isEquals();
+    }
+
+    @Override
+    public int hashCode() {
+        return new HashCodeBuilder(17, 37)
+                .append(key)
+                .append(value)
+                .toHashCode();
+    }
+}

src/main/java/com/winhc/data/push/bean/NodeParam.java

@@ -0,0 +1,41 @@
+package com.winhc.data.push.bean;
+
+import com.aliyun.openservices.shade.org.apache.commons.lang3.StringUtils;
+import com.winhc.data.push.utils.JsonUtils;
+import lombok.*;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * @Author: XuJiakai
+ * @Date: 2020/6/24 08:50
+ * @Description:
+ */
+@Setter
+@Getter
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class NodeParam {
+    private String nodeId;
+    private Map<String, String> param;
+
+    public String toNodeParam(Map<String, String> otherParams) {
+        HashMap<String, String> map = new HashMap<>(param);
+        for (Map.Entry<String, String> entry : otherParams.entrySet()) {
+            map.put(entry.getKey(), entry.getValue());
+        }
+        return map.entrySet().stream()
+                .filter(e -> !e.getKey().startsWith("_"))
+                .filter(e -> StringUtils.isNotEmpty(e.getValue()))
+                .map(e -> e.getKey() + "=" + e.getValue())
+                .collect(Collectors.joining(" "));
+    }
+
+    @Override
+    public String toString() {
+        return JsonUtils.jsonObjToString(this);
+    }
+}

src/main/java/com/winhc/data/push/bean/TaskFlowEnum.java

@@ -0,0 +1,43 @@
+package com.winhc.data.push.bean;
+
+import lombok.Getter;
+
+import java.util.Arrays;
+
+/**
+ * @Author: XuJiakai
+ * @Date: 2020/6/23 10:58
+ * @Description:
+ */
+@Getter
+@SuppressWarnings("all")
+public enum TaskFlowEnum {
+    NOT_RUN(1, "未运行"),
+    WAIT_TIME(2, "等待时间"),
+    WAIT_RESOURCE(3, "等待资源"),
+    RUNNING(4, "运行中"),
+    FAILURE(5, "运行失败"),
+    SUCCESS(6, "运行成功"),
+    CHECKING(7, "校检中"),
+
+    ;
+    private final Integer code;
+    private final String msg;
+
+    TaskFlowEnum(int code, String msg) {
+        this.code = code;
+        this.msg = msg;
+    }
+
+    public static TaskFlowEnum getTaskFlowEnumByCode(Integer code) {
+        return Arrays.stream(TaskFlowEnum.values())
+                .filter(taskFlowEnum -> taskFlowEnum.getCode().equals(code))
+                .findFirst()
+                .orElse(null);
+    }
+
+    @Override
+    public String toString() {
+        return msg;
+    }
+}

src/main/java/com/winhc/data/push/bean/TaskInfo.java

@@ -0,0 +1,52 @@
+package com.winhc.data.push.bean;
+
+import cn.hutool.core.builder.EqualsBuilder;
+import cn.hutool.core.builder.HashCodeBuilder;
+import lombok.*;
+
+/**
+ * @Author: XuJiakai
+ * @Date: 2020/6/24 14:53
+ * @Description:
+ */
+@Getter
+@Setter
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class TaskInfo {
+    private String project;
+    private String flow;
+    private String taskName;
+    private Long dagId;
+
+    public String getKey() {
+        return project + ":" + flow + ":" + taskName;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+
+        if (o == null || getClass() != o.getClass()) return false;
+
+        TaskInfo taskInfo = (TaskInfo) o;
+
+        return new EqualsBuilder()
+                .append(project, taskInfo.project)
+                .append(flow, taskInfo.flow)
+                .append(taskName, taskInfo.taskName)
+                .append(dagId, taskInfo.dagId)
+                .isEquals();
+    }
+
+    @Override
+    public int hashCode() {
+        return new HashCodeBuilder(17, 37)
+                .append(project)
+                .append(flow)
+                .append(taskName)
+                .append(dagId)
+                .toHashCode();
+    }
+}

src/main/java/com/winhc/data/push/bean/TaskParam.java

@@ -0,0 +1,29 @@
+package com.winhc.data.push.bean;
+
+import com.aliyun.openservices.shade.com.alibaba.fastjson.JSONObject;
+import com.aliyun.openservices.shade.com.alibaba.fastjson.serializer.SerializerFeature;
+import lombok.*;
+
+import java.util.Map;
+
+/**
+ * @Author: XuJiakai
+ * @Date: 2020/6/22 14:45
+ * @Description:
+ */
+@Getter
+@Setter
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class TaskParam {
+    private String projectName;
+    private String flowName;
+    private String bizDate;
+    private Map<String, String> nodeParam;
+
+    @Override
+    public String toString() {
+        return JSONObject.toJSONString(this, SerializerFeature.WriteMapNullValue);
+    }
+}

src/main/java/com/winhc/data/push/configuration/CORSConfiguration.java

@@ -0,0 +1,42 @@
+package com.winhc.data.push.configuration;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.web.cors.CorsConfiguration;
+import org.springframework.web.cors.CorsConfigurationSource;
+import org.springframework.web.cors.UrlBasedCorsConfigurationSource;
+import org.springframework.web.servlet.config.annotation.CorsRegistry;
+import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+
+@Configuration
+public class CORSConfiguration implements WebMvcConfigurer {
+    @Override
+    public void addCorsMappings(CorsRegistry registry) {
+        registry.addMapping("/**")
+                //.allowedOrigins("*")
+                .allowedOriginPatterns("*")
+                .allowCredentials(true)
+                .allowedMethods("*")
+                .maxAge(3600);
+    }
+
+    /**
+     * 配置允许跨域
+     * @return
+     */
+//    @Bean
+//    protected CorsConfigurationSource corsConfigurationSource() {
+//        CorsConfiguration configuration = new CorsConfiguration();
+//        configuration.setAllowedOrigins(Collections.singletonList("*"));
+//        configuration.setAllowedMethods(Arrays.asList("GET", "POST", "PUT", "DELETE", "HEAD", "OPTION"));
+//        configuration.setAllowedHeaders(Collections.singletonList("*"));
+//        configuration.addExposedHeader("Authorization");
+//        UrlBasedCorsConfigurationSource source = new UrlBasedCorsConfigurationSource();
+//        source.registerCorsConfiguration("/**", configuration);
+//        return source;
+//    }
+}

src/main/java/com/winhc/data/push/configuration/DataWorksAccessProperties.java

@@ -0,0 +1,21 @@
+package com.winhc.data.push.configuration;
+
+import lombok.Data;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @Author: XuJiakai
+ * @Date: 2020/9/17 14:56
+ * @Description:
+ */
+@Data
+@Configuration
+public class DataWorksAccessProperties {
+    @Value("${odps.access-key-id}")
+    private String accessKeyId;
+    @Value("${odps.access-key-secret}")
+    private String accessKeySecret;
+    @Value("${odps.region-id}")
+    private String regionId;
+}

src/main/java/com/winhc/data/push/configuration/DataWorksProjectConfiguration.java

@@ -0,0 +1,37 @@
+package com.winhc.data.push.configuration;
+
+import com.aliyuncs.DefaultAcsClient;
+import com.aliyuncs.IAcsClient;
+import com.aliyuncs.profile.DefaultProfile;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @Author: XuJiakai
+ * @Date: 2020/6/8 15:41
+ * @Description:
+ */
+@Configuration
+@Data
+@Slf4j
+@AllArgsConstructor
+public class DataWorksProjectConfiguration {
+    private final DataWorksAccessProperties dataWorksAccessProperties;
+
+    @SneakyThrows
+    @Bean
+    public IAcsClient client() {
+        DefaultProfile.addEndpoint(
+                "winhc_ng",
+                dataWorksAccessProperties.getRegionId(),
+                "dataworks-public",
+                "dataworks." + dataWorksAccessProperties.getRegionId() + ".aliyuncs.com");
+        DefaultProfile profile = DefaultProfile.getProfile(dataWorksAccessProperties.getRegionId(), dataWorksAccessProperties.getAccessKeyId(), dataWorksAccessProperties.getAccessKeySecret());
+        IAcsClient client = new DefaultAcsClient(profile);
+        return client;
+    }
+}

src/main/java/com/winhc/data/push/configuration/GlobalExceptionHandler.java

@@ -0,0 +1,28 @@
+package com.winhc.data.push.configuration;
+
+import com.winhc.data.push.vo.ResponseVo;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.http.HttpStatus;
+import org.springframework.web.bind.annotation.ExceptionHandler;
+import org.springframework.web.bind.annotation.ResponseStatus;
+import org.springframework.web.bind.annotation.RestControllerAdvice;
+
+/**
+ * @author: XuJiakai
+ * 2022/5/25 15:38
+ */
+@Slf4j
+@RestControllerAdvice
+@AllArgsConstructor
+public class GlobalExceptionHandler {
+
+    @ResponseStatus(HttpStatus.OK)
+    @ExceptionHandler(value = Exception.class)
+    public ResponseVo handleBadRequest(Exception e) {
+        log.error(e.getMessage(), e);
+        return ResponseVo.failure(e.getMessage());
+    }
+
+
+}

src/main/java/com/winhc/data/push/configuration/OSSAccessProperties.java

@@ -0,0 +1,26 @@
+package com.winhc.data.push.configuration;
+
+import lombok.Data;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+
+
+@Data
+@Configuration
+public class OSSAccessProperties {
+
+    @Value("${oss.endpoint}")
+    private String endpoint;
+    @Value("${oss.access_key_id}")
+    private String accessKeyId;
+    @Value("${oss.access_key_secret}")
+    private String accessKeySecret;
+    @Value("${oss.source_bucket_name}")
+    private String sourceBucketName;
+    @Value("${oss.target_bucket_name}")
+    private String targetBucketName;
+    @Value("${oss.source_path_prefix}")
+    private String sourcePathPrefix;
+    @Value("${oss.target_path_prefix}")
+    private String targetPathprefix;
+}

src/main/java/com/winhc/data/push/configuration/OSSConfiguration.java

@@ -0,0 +1,23 @@
+package com.winhc.data.push.configuration;
+
+import com.aliyun.oss.OSS;
+import com.aliyun.oss.OSSClientBuilder;
+import lombok.AllArgsConstructor;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author π
+ * @Description:
+ * @date 2022/7/20 17:23
+ */
+@Configuration
+@AllArgsConstructor
+public class OSSConfiguration {
+    private final OSSAccessProperties project;
+
+    @Bean
+    public OSS init2() {
+        return new OSSClientBuilder().build(project.getEndpoint(), project.getAccessKeyId(), project.getAccessKeySecret());
+    }
+}

src/main/java/com/winhc/data/push/configuration/OdpsConfiguration.java

@@ -0,0 +1,36 @@
+package com.winhc.data.push.configuration;
+
+import com.aliyun.odps.Odps;
+import com.aliyun.odps.account.Account;
+import com.aliyun.odps.account.AliyunAccount;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * @author: XuJiakai
+ * 2021/7/1 15:49
+ */
+@Configuration
+public class OdpsConfiguration {
+    @Value("${odps.access-key-id}")
+    private String accessKeyId;
+
+    @Value("${odps.access-key-secret}")
+    private String accessKeySecret;
+
+    @Value("${odps.default-project}")
+    private String defaultProject;
+
+    @Value("${odps.end-point}")
+    private String endPoint;
+
+    @Bean
+    public Odps odps() {
+        Account account = new AliyunAccount(accessKeyId, accessKeySecret);
+        Odps odps = new Odps(account);
+        odps.setEndpoint(endPoint);
+        odps.setDefaultProject(defaultProject);
+        return odps;
+    }
+}

src/main/java/com/winhc/data/push/configuration/ProjectParamInit.java

@@ -0,0 +1,48 @@
+package com.winhc.data.push.configuration;
+
+import com.winhc.data.push.bean.DataWorksFlowJob;
+import com.winhc.data.push.bean.DataWorksFlowTask;
+import com.winhc.data.push.utils.DataWorksParamUtils;
+import lombok.Getter;
+import org.jetbrains.annotations.NotNull;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.List;
+
+/**
+ * @author: XuJiakai
+ * 2021/4/14 14:42
+ */
+@Getter
+@Configuration
+public class ProjectParamInit implements BeanFactoryPostProcessor {
+
+    public static final String project = "winhc_ng";
+
+    public static final String exportMongoFlowName = "push_data_as_to_mongo";
+    public static final String exportMongoBeanName = "winhc_ng:push_data_as_to_mongo:mongo_auction_tracking_push_ads";
+    public static final String exportOssFlowName = "push_data_as_to_oss";
+    public static final String exportOssBeanName = "winhc_ng:push_data_as_to_oss:oss_auction_tracking_push_ads";
+
+    /**
+     * @param beanFactory
+     * @throws BeansException
+     */
+    @Override
+    public void postProcessBeanFactory(@NotNull ConfigurableListableBeanFactory beanFactory) throws BeansException {
+        DataWorksParamUtils bean = new DataWorksParamUtils();
+        List<DataWorksFlowJob> jobs = bean.getJobs();
+        for (DataWorksFlowJob job : jobs) {
+            String project = job.getProject();
+            String flow = job.getFlow();
+            List<DataWorksFlowTask> task = job.getTask();
+            for (DataWorksFlowTask dataWorksFlowTask : task) {
+                String beanName = project + ":" + flow + ":" + dataWorksFlowTask.getTaskName();
+                beanFactory.registerSingleton(beanName, dataWorksFlowTask);
+            }
+        }
+    }
+}

src/main/java/com/winhc/data/push/configuration/SwaggerConfiguration.java

@@ -0,0 +1,37 @@
+package com.winhc.data.push.configuration;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import springfox.documentation.builders.ApiInfoBuilder;
+import springfox.documentation.builders.PathSelectors;
+import springfox.documentation.builders.RequestHandlerSelectors;
+import springfox.documentation.service.ApiInfo;
+import springfox.documentation.spi.DocumentationType;
+import springfox.documentation.spring.web.plugins.Docket;
+
+/**
+ * @Author: π
+ * @Date: 2022/7/20 16:40
+ * @Description:
+ */
+@Configuration
+public class SwaggerConfiguration {
+
+    @Bean
+    public Docket createRestApi() {
+        return new Docket(DocumentationType.OAS_30).apiInfo(apiInfo())
+                .select()
+                .apis(RequestHandlerSelectors.basePackage("com.winhc.data.push.controller"))
+                .paths(PathSelectors.any()).build();
+    }
+
+    private ApiInfo apiInfo() {
+        return new ApiInfoBuilder()
+                .title("push data API")
+                .description("同步交易数据")
+                .version("1.0")
+                .build();
+    }
+
+}
+

src/main/java/com/winhc/data/push/constant/BaseParam.java

@@ -0,0 +1,15 @@
+package com.winhc.data.push.constant;
+
+/**
+ * @author π
+ * @Description:
+ * @date 2022/7/21 15:35
+ */
+public class BaseParam {
+    public static final String sign = "da742d3391af7d14acf1539e72f98323";
+    //安硕任务状态表
+    public static final String SYN_ANSHUO_TASKS = "syn_anshuo_tasks";
+    //每日url集合
+    public static final String AUCTION_TRACKING_ANSHUO_URL = "auction_tracking_anshuo_url";
+
+}

src/main/java/com/winhc/data/push/controller/PushDataController.java

@@ -0,0 +1,64 @@
+package com.winhc.data.push.controller;
+
+import cn.hutool.core.lang.Assert;
+import com.aliyun.openservices.shade.org.apache.commons.lang3.StringUtils;
+import com.winhc.data.push.constant.BaseParam;
+import com.winhc.data.push.service.SynDataService;
+import com.winhc.data.push.utils.BaseUtils;
+import com.winhc.data.push.vo.ResponseVo;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.web.bind.annotation.*;
+
+
+/**
+ * @author π
+ * @Description:推送数据
+ * @date 2022/7/20 16:24
+ */
+@Slf4j
+@AllArgsConstructor
+@RestController
+@Api(tags = "任务触发", value = "task")
+@RequestMapping("task")
+public class PushDataController {
+
+    private final SynDataService synDataService;
+
+
+    @ApiOperation(value = "触发任务")
+    @GetMapping("push/{ds}")
+    public ResponseVo touch(@PathVariable String ds, @RequestParam String tn, @RequestParam String sign) {
+        //日期是昨天之前数据
+        Assert.isTrue(BaseUtils.verifyDate(ds), "ds 不合法");
+        Assert.isTrue(BaseParam.sign.equals(sign), "验签不通过！");
+        Assert.isTrue(StringUtils.isNotBlank(tn), "参数不能为空!");
+        long start = System.currentTimeMillis();
+        try {
+            String re = synDataService.saveTask(tn, ds);
+            return ResponseVo.success(start, re);
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
+            return ResponseVo.failure(start, e.getMessage());
+        }
+    }
+
+    @ApiOperation(value = "强制触发任务")
+    @GetMapping("repush/{ds}")
+    public ResponseVo reTouch(@PathVariable String ds, @RequestParam String tn, @RequestParam String sign) {
+        Assert.isTrue(BaseUtils.verifyDate(ds), "ds 不合法");
+        Assert.isTrue(BaseParam.sign.equals(sign), "验签不通过！");
+        Assert.isTrue(StringUtils.isNotBlank(tn), "参数不能为空!");
+        long start = System.currentTimeMillis();
+        try {
+            Boolean re = synDataService.updateTask(tn, ds, "create");
+            return ResponseVo.success(start, re ? "restart" : "error");
+        } catch (Exception e) {
+            log.error(e.getMessage(), e);
+            return ResponseVo.failure(start, e.getMessage());
+        }
+    }
+
+}

src/main/java/com/winhc/data/push/framework/MongoDbFastScan.java

@@ -0,0 +1,89 @@
+package com.winhc.data.push.framework;
+
+import com.mongodb.client.MongoDatabase;
+import lombok.extern.slf4j.Slf4j;
+import org.bson.Document;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Vector;
+import java.util.function.Consumer;
+
+
+@Slf4j
+public class MongoDbFastScan {
+    int thread_num = 1;
+    int data_num;
+    int batchSize = 500;
+    int skip = 0;
+    String scanTable;
+    Consumer<List<Document>> func;
+    protected MongoDatabase mongoDatabase;
+
+    public MongoDbFastScan(String scanTable, Consumer<List<Document>> func, MongoDatabase mongoDatabase) {
+        this.scanTable = scanTable;
+        this.func = func;
+        this.mongoDatabase = mongoDatabase;
+    }
+
+    public MongoDbFastScan threadNum(int thread_num) {
+        this.thread_num = thread_num;
+        return this;
+    }
+
+    public MongoDbFastScan skip(int skip) {
+        this.skip = skip;
+        return this;
+    }
+
+    public MongoDbFastScan batchSize(int batchSize) {
+        this.batchSize = batchSize;
+        return this;
+    }
+
+    private void scan(int skip, int limit) {
+        List<Document> list = new ArrayList<>();
+        for (Document document : mongoDatabase.getCollection(scanTable).find().skip(skip).limit(limit).noCursorTimeout(true).batchSize(batchSize)) {
+            list.add(document);
+            if (list.size() >= batchSize) {
+                func.accept(list);
+                list.clear();
+            }
+        }
+        if (!list.isEmpty()) {
+            func.accept(list);
+            list.clear();
+        }
+        log.info("start:{} end:{} success", skip, skip + limit);
+    }
+
+    public void scan() {
+        double count = new Long(mongoDatabase.getCollection(scanTable).countDocuments()).doubleValue();
+        data_num = new Double(Math.ceil(count / (double) thread_num) * thread_num).intValue();
+        if (count == 0d) {
+            return;
+        }
+        Vector<Thread> ts = new Vector<>();
+        int skipValue = skip % (data_num / thread_num);
+        int skipContinue = skip / (data_num / thread_num);
+
+        for (int i = 0; i < thread_num; i++) {
+            final int a = i;
+            if (a < skipContinue) {
+                continue;
+            } else {
+                Thread t = new Thread(() -> scan(a * (data_num / thread_num) + skipValue, data_num / thread_num - skipValue));
+                ts.add(t);
+                t.start();
+            }
+        }
+
+        for (Thread t : ts) {
+            try {
+                t.join();
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+}

src/main/java/com/winhc/data/push/service/DataWorksService.java

@@ -0,0 +1,27 @@
+package com.winhc.data.push.service;
+
+
+import com.aliyuncs.dataworks_public.model.v20180601.CreateManualDagResponse;
+import com.winhc.data.push.bean.DataWorksFlowTask;
+import com.winhc.data.push.bean.TaskFlowEnum;
+import com.winhc.data.push.bean.TaskParam;
+
+import java.util.Map;
+
+/**
+ * @author: XuJiakai
+ * 2020/10/31 10:54
+ */
+public interface DataWorksService {
+
+     CreateManualDagResponse export2Mongo(String ds);
+
+     CreateManualDagResponse export2OSS(String ds);
+
+     CreateManualDagResponse touch(TaskParam task);
+
+     Map<String, TaskFlowEnum> query(String project, Long returnValue);
+
+     CreateManualDagResponse touch(String project, String flowName, String bizDate, DataWorksFlowTask dataWorksFlowTask, Map<String, String> otherParams);
+
+}

src/main/java/com/winhc/data/push/service/DingTalkService.java

@@ -0,0 +1,10 @@
+package com.winhc.data.push.service;
+
+/**
+ * @author π
+ * @Description:
+ * @date 2022/7/24 19:07
+ */
+public interface DingTalkService {
+    void sendMessage(String message);
+}

src/main/java/com/winhc/data/push/service/MongoDbService.java

@@ -0,0 +1,12 @@
+package com.winhc.data.push.service;
+
+/**
+ * @author π
+ * @Description:
+ * @date 2022/7/22 18:17
+ */
+public interface MongoDbService {
+
+    Boolean find(String id);
+
+}

src/main/java/com/winhc/data/push/service/SynDataService.java

@@ -0,0 +1,25 @@
+package com.winhc.data.push.service;
+
+import org.bson.Document;
+
+import java.util.List;
+
+/**
+ * @author π
+ * @Description:
+ * @date 2022/7/20 16:52
+ */
+public interface SynDataService {
+    void push(String tn, String ds);
+
+    void sendMessage(String message) throws Exception;
+
+    String saveTask(String tn, String ds);
+
+    Boolean findTaskExists(Document doc);
+
+    List<Document> findAllTask(Document id);
+
+    Boolean updateTask(String tn, String ds, String status);
+
+}

src/main/java/com/winhc/data/push/service/TouchService.java

@@ -0,0 +1,17 @@
+package com.winhc.data.push.service;
+
+import com.aliyuncs.dataworks_public.model.v20180601.CreateManualDagResponse;
+
+/**
+ * @author π
+ * @Description:
+ * @date 2022/7/21 18:17
+ */
+public interface TouchService {
+    Boolean query(String projectName, Long dagId);
+
+    Boolean export2Mongo(String  project,String  ds);
+
+    Boolean export2OSS(String  project,String  ds);
+
+}

src/main/java/com/winhc/data/push/service/impl/DataWorksServiceImpl.java

@@ -0,0 +1,107 @@
+package com.winhc.data.push.service.impl;
+
+import com.aliyuncs.IAcsClient;
+import com.aliyuncs.dataworks_public.model.v20180601.CreateManualDagRequest;
+import com.aliyuncs.dataworks_public.model.v20180601.CreateManualDagResponse;
+import com.aliyuncs.dataworks_public.model.v20180601.SearchManualDagNodeInstanceRequest;
+import com.aliyuncs.dataworks_public.model.v20180601.SearchManualDagNodeInstanceResponse;
+import com.aliyuncs.http.ProtocolType;
+import com.winhc.data.push.bean.DataWorksFlowTask;
+import com.winhc.data.push.bean.TaskFlowEnum;
+import com.winhc.data.push.bean.TaskParam;
+import com.winhc.data.push.configuration.DataWorksAccessProperties;
+import com.winhc.data.push.configuration.ProjectParamInit;
+import com.winhc.data.push.service.DataWorksService;
+import com.winhc.data.push.utils.DateUtils;
+import com.winhc.data.push.utils.JsonUtils;
+import lombok.AllArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * @Author: XuJiakai
+ * @Date: 2020/6/22 11:28
+ * @Description:
+ */
+@Slf4j
+@Service
+@AllArgsConstructor
+public class DataWorksServiceImpl implements DataWorksService {
+
+    private final DataWorksAccessProperties dataWorksAccessProperties;
+    private final Map<String, DataWorksFlowTask> dataWorksFlowTaskMap;
+    private final IAcsClient client;
+
+    @Override
+    public CreateManualDagResponse export2Mongo(String ds) {
+        return touch(ProjectParamInit.project, ProjectParamInit.exportMongoFlowName, DateUtils.parseDate(ds), dataWorksFlowTaskMap.get(ProjectParamInit.exportMongoBeanName), new HashMap<>());
+    }
+
+    @Override
+    public CreateManualDagResponse export2OSS(String ds) {
+        return touch(ProjectParamInit.project, ProjectParamInit.exportOssFlowName, DateUtils.parseDate(ds), dataWorksFlowTaskMap.get(ProjectParamInit.exportOssBeanName), new HashMap<>());
+    }
+
+    @Override
+    public CreateManualDagResponse touch(String project, String flowName, String bizDate, DataWorksFlowTask dataWorksFlowTask, Map<String, String> otherParams) {
+        TaskParam build = TaskParam.builder()
+                .projectName(project)
+                .bizDate(bizDate)
+                .flowName(flowName)
+                .nodeParam(dataWorksFlowTask.toNodeParam(bizDate, otherParams))
+                .build();
+        return touch(build);
+    }
+
+    @SneakyThrows
+    public CreateManualDagResponse triggerWithParam(String projectName
+            , String flowName, String bizDate, String dagPara, String nodePara) {
+        log.info("触发任务：{}.{} {}", projectName, flowName, bizDate);
+        CreateManualDagRequest request = new CreateManualDagRequest();
+        request.setProjectName(projectName);
+        request.setFlowName(flowName);
+        request.setBizdate(bizDate + " 00:00:00");
+        request.setRegionId(dataWorksAccessProperties.getRegionId());
+        request.setProtocol(ProtocolType.HTTP);
+        request.setDagPara(dagPara);
+        request.setNodePara(nodePara);
+        CreateManualDagResponse response = client.getAcsResponse(request);
+        log.info("\n任务结果：\n\trequestId:{},\n\treturnCode:{},\n\treturnErrorSolution:{},\n\treturnMessage:{},\n\treturnValue:{}",
+                response.getRequestId()
+                , response.getReturnCode()
+                , response.getReturnErrorSolution()
+                , response.getReturnMessage()
+                , response.getReturnValue());
+        return response;
+    }
+
+    @SneakyThrows
+    public CreateManualDagResponse touch(TaskParam taskParam) {
+        return triggerWithParam(taskParam.getProjectName(), taskParam.getFlowName(), taskParam.getBizDate(), null, JsonUtils.jsonObjToString(taskParam.getNodeParam()));
+    }
+
+
+    @SneakyThrows
+    public Map<String, TaskFlowEnum> query(String projectName, Long dagId) {
+        SearchManualDagNodeInstanceRequest searchNodeInstanceListRequest
+                = new SearchManualDagNodeInstanceRequest();
+        searchNodeInstanceListRequest.setDagId(dagId);
+        searchNodeInstanceListRequest.setProjectName(projectName);
+        searchNodeInstanceListRequest.setProtocol(ProtocolType.HTTP);
+        SearchManualDagNodeInstanceResponse searchResponse = client
+                .getAcsResponse(searchNodeInstanceListRequest);
+        java.util.List<SearchManualDagNodeInstanceResponse.NodeInsInfo> nodeInsfos = searchResponse.getData();
+        for (SearchManualDagNodeInstanceResponse.NodeInsInfo nodeInsInfo : nodeInsfos) {
+            TaskFlowEnum code = TaskFlowEnum.getTaskFlowEnumByCode(nodeInsInfo.getStatus());
+            log.info("{}:{} {}", nodeInsInfo.getNodeName(), nodeInsInfo.getStatus(), code);
+        }
+        return nodeInsfos.stream()
+                .collect(Collectors.toMap(SearchManualDagNodeInstanceResponse.NodeInsInfo::getNodeName, node -> TaskFlowEnum.getTaskFlowEnumByCode(node.getStatus()), (o1, o2) -> o1));
+    }
+
+}

src/main/java/com/winhc/data/push/service/impl/DingTalkServiceImpl.java

@@ -0,0 +1,32 @@
+package com.winhc.data.push.service.impl;
+
+import cn.cocowwy.dingtalk.DingTalkGroupApi;
+import com.winhc.data.push.service.DingTalkService;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.util.Collections;
+
+/**
+ * @author π
+ * @Description:
+ * @date 2022/7/24 19:08
+ */
+@Slf4j
+@Service
+@AllArgsConstructor
+public class DingTalkServiceImpl implements DingTalkService {
+
+    private final DingTalkGroupApi dingTalkApi;
+
+    @Override
+    public void sendMessage(String message) {
+        try {
+            dingTalkApi.sendTextByPhones("data_push_message", message, Collections.singletonList("17602140784"));
+        } catch (Exception e) {
+            log.error("send message error : {}", e.getMessage());
+        }
+
+    }
+}

src/main/java/com/winhc/data/push/service/impl/MongoServiceImpl.java

@@ -0,0 +1,16 @@
+package com.winhc.data.push.service.impl;
+
+
+import com.winhc.data.push.service.MongoDbService;
+
+/**
+ * @author π
+ * @Description:
+ * @date 2022/7/22 18:18
+ */
+public class MongoServiceImpl implements MongoDbService {
+    @Override
+    public Boolean find(String id) {
+        return null;
+    }
+}

src/main/java/com/winhc/data/push/service/impl/SynDataServiceImpl.java

@@ -0,0 +1,166 @@
+package com.winhc.data.push.service.impl;
+
+import cn.cocowwy.dingtalk.DingTalkGroupApi;
+import com.aliyun.openservices.shade.org.apache.commons.lang3.StringUtils;
+import com.aliyun.oss.OSS;
+import com.aliyun.oss.model.PutObjectRequest;
+import com.mongodb.client.FindIterable;
+import com.mongodb.client.MongoDatabase;
+import com.mongodb.client.model.Filters;
+import com.mongodb.client.model.FindOneAndUpdateOptions;
+import com.mongodb.client.model.UpdateOneModel;
+import com.mongodb.client.model.UpdateOptions;
+import com.winhc.data.push.configuration.OSSAccessProperties;
+import com.winhc.data.push.framework.MongoDbFastScan;
+import com.winhc.data.push.service.SynDataService;
+import com.winhc.data.push.service.TouchService;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.bson.Document;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.stereotype.Service;
+
+import java.io.ByteArrayInputStream;
+import java.util.Collections;
+import java.util.Date;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+import java.util.stream.StreamSupport;
+
+import static com.winhc.data.push.constant.BaseParam.AUCTION_TRACKING_ANSHUO_URL;
+import static com.winhc.data.push.constant.BaseParam.SYN_ANSHUO_TASKS;
+
+/**
+ * @author π
+ * @Description:
+ * @date 2022/7/20 17:02
+ */
+@Slf4j
+@Service
+@AllArgsConstructor
+public class SynDataServiceImpl implements SynDataService {
+
+    private final MongoTemplate mongoTemplate;
+    private final OSS ossClient;
+    private final OSSAccessProperties project;
+    private final TouchService touchService;
+    private final DingTalkGroupApi dingTalkApi;
+
+
+    @Override
+    public void push(String tn, String ds) {
+        String id = tn + "_" + ds;
+        try {
+            updateTask(tn, ds, "running");
+            sendMessage("task running !!! " + id);
+            String project = "winhc_ng";
+            if (touchService.export2Mongo(project, ds)) {
+                copyOSS(ds);
+                if (touchService.export2OSS(project, ds)) {
+                    saveObject(ds);
+                    updateTask(tn, ds, "success");
+                    sendMessage("导出 任务 成功 !!! " + id);
+                } else {
+                    log.error("导出OSS 失败 !!!");
+                    sendMessage("导出 OSS 失败!!! " + id);
+                    updateTask(tn, ds, "oss copy error");
+                }
+            } else {
+                log.error("导出Mongo 失败!!!");
+                sendMessage("导出 Mongo 失败!!! " + id);
+                updateTask(tn, ds, "export mongo error");
+            }
+
+        } catch (Exception e) {
+            sendMessage("任务失败!!! " + id + " |  error " + e.getMessage());
+            updateTask(tn, ds, "task error");
+            e.printStackTrace();
+            log.error("export error: {}", e.getMessage());
+        }
+    }
+
+    @Override
+    public String saveTask(String tn, String ds) {
+        String id = tn + "_" + ds;
+        if (findTaskExists(new Document("_id", id))) {
+            return "task repeat";
+        } else {
+            Document d = new Document();
+            d.put("_id", id);
+            d.put("create_time", new Date());
+            d.put("update_time", new Date());
+            d.put("status", "create");
+            d.put("tn", tn);
+            d.put("ds", ds);
+            mongoTemplate.getCollection(SYN_ANSHUO_TASKS).insertOne(d);
+            return "task create success";
+        }
+
+    }
+
+    @Override
+    public void sendMessage(String message) {
+        try {
+            dingTalkApi.sendTextByPhones("data_push_message", message, Collections.singletonList("17602140784"));
+        } catch (Exception e) {
+            log.error("send message error : {}", e.getMessage());
+        }
+
+    }
+
+    @Override
+    public Boolean findTaskExists(Document doc) {
+        return mongoTemplate.getCollection(SYN_ANSHUO_TASKS).find(doc).iterator().hasNext();
+    }
+
+    @Override
+    public List<Document> findAllTask(Document doc) {
+        FindIterable<Document> it = mongoTemplate.getCollection(SYN_ANSHUO_TASKS).find(doc);
+        return StreamSupport.stream(it.spliterator(), false).collect(Collectors.toList());
+    }
+
+
+    @Override
+    public Boolean updateTask(String tn, String ds, String status) {
+        String _id = tn + "_" + ds;
+        Document d = new Document();
+        d.put("update_time", new Date());
+        d.put("status", status);
+        d.put("tn", tn);
+        d.put("ds", ds);
+        mongoTemplate.getCollection(SYN_ANSHUO_TASKS)
+                .findOneAndUpdate(Filters.eq("_id", _id), new Document("$set", d), new FindOneAndUpdateOptions().upsert(true));
+        return true;
+    }
+
+    private void copyOSS(String ds) {
+        MongoDatabase db = mongoTemplate.getDb();
+        Consumer<List<Document>> func = list -> {
+            list.stream().map(d -> d.getString("file_path"))
+                    .filter(StringUtils::isNotEmpty)
+                    .forEach(keySuffix -> {
+                        copyObject2OSS(keySuffix, ds);
+                    });
+        };
+
+        MongoDbFastScan mongoDbFastScan = new MongoDbFastScan(AUCTION_TRACKING_ANSHUO_URL, func, db)
+                .batchSize(200).threadNum(10);
+        mongoDbFastScan.scan();
+    }
+
+    public void copyObject2OSS(String keySuffix, String ds) {
+        String sourceKey = project.getSourcePathPrefix() + keySuffix;
+        String targetKey = project.getTargetPathprefix() + ds + "/file/" + keySuffix;
+        ossClient.copyObject(project.getSourceBucketName(), sourceKey, project.getTargetBucketName(), targetKey);
+        log.info(targetKey);
+    }
+
+    public void saveObject(String ds) {
+        String targetKey = project.getTargetPathprefix() + ds + "/anshuo.txt";
+        PutObjectRequest putObjectRequest = new PutObjectRequest(project.getTargetBucketName(), targetKey, new ByteArrayInputStream("".getBytes()));
+        ossClient.putObject(putObjectRequest);
+
+    }
+}

src/main/java/com/winhc/data/push/service/impl/TouchServiceImpl.java

@@ -0,0 +1,104 @@
+package com.winhc.data.push.service.impl;
+
+import com.winhc.data.push.bean.*;
+import com.winhc.data.push.service.DataWorksService;
+import com.winhc.data.push.service.DingTalkService;
+import com.winhc.data.push.service.TouchService;
+import com.winhc.data.push.utils.DateUtils;
+import lombok.AllArgsConstructor;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.util.*;
+import java.util.stream.Collectors;
+
+/**
+ * @author π
+ * @Description:
+ * @date 2022/7/21 18:18
+ */
+@Service
+@Slf4j
+@AllArgsConstructor
+public class TouchServiceImpl implements TouchService {
+
+    private final DataWorksService dataWorksService;
+    private final DingTalkService dingTalkService;
+
+    @Override
+    public Boolean export2Mongo(String project, String ds) {
+        return query(project, dataWorksService.export2Mongo(ds).getReturnValue());
+    }
+
+    @Override
+    public Boolean export2OSS(String project, String ds) {
+        return query(project, dataWorksService.export2OSS(ds).getReturnValue());
+    }
+
+
+    @SneakyThrows
+    @Override
+    public Boolean query(String projectName, Long dagId) {
+        LocalDateTime start = LocalDateTime.now();
+        int totalTask = 1;
+        int i = 0;
+        int successTask = 0;
+        int failedTask = 0;
+        ArrayList<Long> hours = new ArrayList<>();
+        while (true) {
+            int awaitTask = 0;
+            List<String> empty = new ArrayList<>();
+            Map<String, TaskFlowEnum> query = dataWorksService.query(projectName, dagId);
+            Map<String, List<String>> status = query.entrySet().stream()
+                    .map(e -> {
+                        Entry<String, String> entry = new Entry<>();
+                        if (e.getValue().equals(TaskFlowEnum.SUCCESS) || e.getValue().equals(TaskFlowEnum.FAILURE)) {
+                            entry.setKey(e.getValue().getMsg());
+                        } else {
+                            entry.setKey("运行中");
+                        }
+                        entry.setValue(e.getKey());
+                        return entry;
+                    }).collect(Collectors.groupingBy(Entry::getKey, Collectors.mapping(Entry::getValue, Collectors.toList())));
+            List<String> failure = status.getOrDefault(TaskFlowEnum.FAILURE.getMsg(), empty);
+            List<String> await = status.getOrDefault(TaskFlowEnum.RUNNING.getMsg(), empty);
+
+            if (failure.size() != 0) {
+                log.error("任务失败 node : {}", dagId);
+                return false;
+            } else {
+                if (await.size() != 0) {
+                    awaitTask++;
+                    LocalDateTime now = LocalDateTime.now();
+                    Duration duration = Duration.between(start, now);
+                    long hour = duration.toHours();
+                    if (hour >= 1 && !hours.contains(hour)) {
+                        i++;
+                        hours.add(hour);
+                        log.info("导出数据阻塞告警 !!!!!!");
+                        dingTalkService.sendMessage("导出数据阻塞告警 !!!!!!");
+                    }
+                } else {
+                    return true;
+                }
+            }
+
+            //超时退出
+            if (DateUtils.getCurrentHours() >= 22 || i > 6) {
+                log.info("超时程序主动退出 !!!!!!");
+                return false;
+            }
+            log.info("\nawait task:{}\ntotal task:{}\nsuccess task:{}\nfailure task:{}\n"
+                    , awaitTask
+                    , totalTask
+                    , successTask
+                    , failedTask
+            );
+            Thread.sleep(10000);
+        }
+    }
+
+}

src/main/java/com/winhc/data/push/task/PushDataTask.java

@@ -0,0 +1,50 @@
+package com.winhc.data.push.task;
+
+
+import com.aliyun.openservices.shade.org.apache.commons.lang3.StringUtils;
+import com.winhc.data.push.service.DataWorksService;
+import com.winhc.data.push.service.SynDataService;
+import com.winhc.data.push.utils.BaseUtils;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.bson.Document;
+import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+/**
+ * @author π
+ * @Description:触发任务
+ * @date 2021/6/22 17:07
+ */
+
+@Component
+@Slf4j
+@EnableScheduling
+@AllArgsConstructor
+public class PushDataTask {
+
+
+    private final SynDataService synDataService;
+
+
+    @Scheduled(cron = "*/10 * * * * ?")
+    //@Scheduled(cron = "0 /2 * * * ? ")
+    public synchronized void start() {
+        log.info("start PushDataTask !!! ");
+        //if (BaseUtils.isWindows()) return;
+        //获取运行任务
+        List<Document> allTask = synDataService.findAllTask(new Document("status", "create"));
+        allTask.forEach(task -> {
+            String ds = task.getString("ds");
+            String tn = task.getString("tn");
+            if (StringUtils.isNotBlank(ds) && StringUtils.isNotBlank(tn)) {
+                synDataService.push(tn, ds);
+            }
+        });
+        log.info("stop PushDataTask !!! ");
+
+    }
+}

src/main/java/com/winhc/data/push/utils/BaseUtils.java

@@ -0,0 +1,31 @@
+package com.winhc.data.push.utils;
+
+import com.aliyun.openservices.shade.org.apache.commons.lang3.StringUtils;
+
+import static com.winhc.data.push.utils.DateUtils.*;
+
+/**
+ * @author π
+ * @Description:
+ * @date 2022/7/24 13:38
+ */
+public class BaseUtils {
+
+    public static Boolean isWindows() {
+        return System.getProperty("os.name").contains("Windows");
+    }
+
+    public static Boolean verifyDate(String ds) {
+        if (StringUtils.isBlank(ds) || ds.length() != 8) return false;
+        if (StringUtils.isBlank(parseDate(ds))) return false;
+        String now = nowDate(YYYYMMDD);
+        return ds.compareTo(now) < 0;
+    }
+
+    public static void main(String[] args) {
+        System.out.println(verifyDate("20220920"));
+        System.out.println(verifyDate("20220919"));
+        System.out.println(verifyDate("2022-09-19"));
+        System.out.println(verifyDate("2022-0919"));
+    }
+}

src/main/java/com/winhc/data/push/utils/BigDecimalUtil.java

@@ -0,0 +1,113 @@
+package com.winhc.data.push.utils;
+
+import java.math.BigDecimal;
+
+/**
+ * @Author: XuJiakai
+ * @Date: 2020/9/2 17:57
+ * @Description:
+ */
+public class BigDecimalUtil {
+    // 进行加法运算
+    public static double add(String v1, String v2) {
+
+        BigDecimal b1 = new BigDecimal(v1);
+        BigDecimal b2 = new BigDecimal(v2);
+        return b1.add(b2).doubleValue();
+    }
+
+    public static double add(double v1, double v2) {
+        BigDecimal b1 = new BigDecimal(v1);
+        BigDecimal b2 = new BigDecimal(v2);
+        return b1.add(b2).doubleValue();
+    }
+
+    // 进行减法运算
+    public static double subtract(String v1, String v2) {
+        BigDecimal b1 = new BigDecimal(v1);
+        BigDecimal b2 = new BigDecimal(v2);
+        return b1.subtract(b2).doubleValue();
+    }
+
+    public static double subtract(double v1, double v2) {
+        BigDecimal b1 = new BigDecimal(v1);
+        BigDecimal b2 = new BigDecimal(v2);
+        return b1.subtract(b2).doubleValue();
+    }
+
+    // 进行乘法运算
+    public static double mul(String d1, String d2) {
+        BigDecimal b1 = new BigDecimal(d1);
+        BigDecimal b2 = new BigDecimal(d2);
+        return b1.multiply(b2).doubleValue();
+    }
+
+    public static double mul(double d1, double d2) {
+        BigDecimal b1 = new BigDecimal(d1);
+        BigDecimal b2 = new BigDecimal(d2);
+        return b1.multiply(b2).doubleValue();
+    }
+
+    // 进行除法运算
+    public static double div(String d1, String d2, int len) {
+        BigDecimal b1 = new BigDecimal(d1);
+        BigDecimal b2 = new BigDecimal(d2);
+        if (b2.doubleValue() == 0) {
+            return 0;
+        }
+        return b1.divide(b2, len, BigDecimal.ROUND_HALF_UP).doubleValue();
+    }
+
+    public static double div(double d1, double d2, int len) {
+        if (d2 == 0) {
+            return 0;
+        }
+        BigDecimal b1 = new BigDecimal(d1);
+        BigDecimal b2 = new BigDecimal(d2);
+        return b1.divide(b2, len, BigDecimal.ROUND_HALF_UP).doubleValue();
+    }
+
+    public static double div(long d1, long d2, int len) {
+        if (d2 == 0) {
+            return 0;
+        }
+        BigDecimal b1 = new BigDecimal(d1);
+        BigDecimal b2 = new BigDecimal(d2);
+        return b1.divide(b2, len, BigDecimal.ROUND_HALF_UP).doubleValue();
+    }
+
+    // 进行四舍五入操作-len 小数点后保留几位
+    public static double round(double d, int len) {
+        BigDecimal b1 = new BigDecimal(d);
+        BigDecimal b2 = new BigDecimal(1);
+        // 任何一个数字除以1都是原数字
+        // ROUND_HALF_UP是BigDecimal的一个常量，表示进行四舍五入的操作
+        return b1.divide(b2, len, BigDecimal.ROUND_HALF_UP).doubleValue();
+    }
+
+
+    /**
+     * 进行四舍五入操作 -并保留指定位数
+     *
+     * @param @param  d1		处理数据
+     * @param @param  d2		如果为1 则自己保留多少位，100为除以100之后的结果再保留位数
+     * @param @param  len  小数点保留位数
+     * @param @return 参数
+     * @return String    返回类型
+     * @throws
+     * @Title: round
+     * @Description:
+     */
+    public static double round(String d1, String d2, int len) {
+        BigDecimal b1 = new BigDecimal(d1);
+        BigDecimal b2 = new BigDecimal(d2);
+        // 任何一个数字除以1都是原数字
+        // ROUND_HALF_UP是BigDecimal的一个常量，表示进行四舍五入的操作
+        double res = b1.divide(b2, len, BigDecimal.ROUND_HALF_UP).doubleValue();
+        return res;
+    }
+
+    public static Long double2Long(Double d) {
+        return Math.round(d);
+    }
+}

src/main/java/com/winhc/data/push/utils/DataWorksParamUtils.java

@@ -0,0 +1,114 @@
+package com.winhc.data.push.utils;
+
+import com.winhc.data.push.bean.DataWorksFlowJob;
+import com.winhc.data.push.bean.DataWorksFlowTask;
+import com.winhc.data.push.bean.NodeParam;
+import lombok.AllArgsConstructor;
+import lombok.SneakyThrows;
+import org.springframework.core.io.ClassPathResource;
+import org.springframework.core.io.Resource;
+import org.springframework.stereotype.Component;
+import org.yaml.snakeyaml.Yaml;
+
+import java.io.*;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * @Author: XuJiakai
+ * @Date: 2020/6/23 14:23
+ * @Description:
+ */
+@SuppressWarnings("all")
+@Component
+@AllArgsConstructor
+public class DataWorksParamUtils {
+//    @Autowired
+//    ResourceLoader resourceLoader;
+
+    private static final String defaultFileName = "data-works-param.yaml";
+
+    @SneakyThrows
+    public List<DataWorksFlowJob> getJobs() {
+        InputStream is = null;
+        InputStreamReader isr = null;
+        BufferedReader br = null;
+        try {
+//            File file =ResourceUtils.getFile("classpath:data-works-param.yaml");
+            Resource resource = new ClassPathResource("data-works-param.yaml");
+            is = resource.getInputStream();
+
+//            Resource resource = resourceLoader.getResource("classpath:data-works-param.yaml");
+//            is = resource.getInputStream();
+            isr = new InputStreamReader(is);
+            br = new BufferedReader(isr);
+            return parseJobs(br);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        } finally {
+            if (br != null) {
+                br.close();
+            }
+            if (isr != null) {
+                isr.close();
+            }
+            if (is != null) {
+                is.close();
+            }
+        }
+    }
+
+    public static List<DataWorksFlowJob> getJobs(String fileName) throws FileNotFoundException {
+        return parseJobs(fileName);
+    }
+
+    private static List<DataWorksFlowJob> parseJobs(String path) throws FileNotFoundException {
+        Reader reader = new FileReader(new File(path));
+        return parseJobs(reader);
+    }
+
+    private static List<DataWorksFlowJob> parseJobs(Reader reader) throws FileNotFoundException {
+        Yaml yml = new Yaml();
+        Map map = yml.loadAs(reader, Map.class);
+
+        List j = ((List) map.get("job"));
+        List<DataWorksFlowJob> jobs = (List<DataWorksFlowJob>) j
+                .stream()
+
+                .map(m -> {
+                    String project = ((String) ((Map<String, Object>) m).get("project"));
+                    String flow = ((String) ((Map<String, Object>) m).get("flow"));
+                    List<DataWorksFlowTask> collect1 = (List<DataWorksFlowTask>) ((List) ((Map<String, Object>) m).get("task"))
+                            .stream()
+                            .map(t -> {
+                                String taskName = ((String) ((Map<String, Object>) t).get("taskName"));
+                                List<Map<String, Object>> ll = ((List<Map<String, Object>>) ((Map<String, Object>) t).get("param"));
+                                List<NodeParam> nodeParamList = ll.stream()
+                                        .map(mm -> {
+                                            String nodeId = String.valueOf(mm.remove("_nodeId"));
+                                            Map<String, String> collect = mm.entrySet().stream().collect(Collectors.toMap(Map.Entry::getKey, e -> String.valueOf(e.getValue()), (o1, o2) -> o1));
+                                            return NodeParam.builder().nodeId(nodeId).param(collect).build();
+                                        }).collect(Collectors.toList());
+                                return new DataWorksFlowTask(taskName, nodeParamList);
+                            })
+                            .collect(Collectors.toList());
+                    DataWorksFlowJob build = DataWorksFlowJob.builder()
+                            .project(project)
+                            .flow(flow)
+                            .task(collect1)
+                            .build();
+                    return build;
+                })
+                .collect(Collectors.toList());
+
+        List<DataWorksFlowJob> collect = jobs.stream().collect(Collectors.groupingBy(m -> m.getFlow() + m.getProject())).values().stream().map(list -> new DataWorksFlowJob(list.get(0).getProject(), list.get(0).getFlow(), list.stream().flatMap(l -> l.getTask().stream()).collect(Collectors.toList()))).collect(Collectors.toList());
+        return collect;
+    }
+
+    public static void main(String[] args) {
+        DataWorksParamUtils bean = new DataWorksParamUtils();
+        List<DataWorksFlowJob> jobs = bean.getJobs();
+        System.out.println(jobs.toString());
+    }
+}

src/main/java/com/winhc/data/push/utils/DateUtils.java

@@ -0,0 +1,104 @@
+package com.winhc.data.push.utils;
+
+import java.text.SimpleDateFormat;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.GregorianCalendar;
+
+
+public class DateUtils {
+    public static final DateTimeFormatter YYYYMMDD = DateTimeFormatter.ofPattern("yyyyMMdd");
+    public static final DateTimeFormatter YYYYMMDDHHMMSS = DateTimeFormatter.ofPattern("yyyyMMdd HH:mm:ss");
+    public static final DateTimeFormatter YYYY_MM_DDHHMMSS = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+    public static final String FORMAT_YYYYMMDD = "yyyyMMdd";
+    public static final String FORMAT_YYYY_MM_DD = "yyyy-MM-dd";
+
+    public static final String formatDate(Date date, String pattern) {
+        String v = null;
+        try {
+            if (date == null)
+                return null;
+            SimpleDateFormat dateFormat = new SimpleDateFormat(pattern);
+            v = dateFormat.format(date);
+        } catch (Exception e) {
+            // do nothing
+        }
+        return v;
+    }
+
+    public static final String parseDate(String d) {
+        String v = null;
+        try {
+            if (d == null)
+                return null;
+            SimpleDateFormat dateFormat = new SimpleDateFormat(FORMAT_YYYYMMDD);
+            Date v1 = dateFormat.parse(d);
+            v = formatDate(v1, FORMAT_YYYY_MM_DD);
+        } catch (Exception e) {
+            // do nothing
+        }
+        return v;
+    }
+
+    public static String nowDate() {
+        return nowDate(null);
+    }
+
+    public static String nowDate(DateTimeFormatter pattern) {
+        if (pattern == null) {
+            pattern = DateTimeFormatter.ISO_DATE;
+        }
+        return LocalDateTime.now().format(pattern);
+    }
+    public static String getDate(int num) {
+        return LocalDate.now().plusDays(num).format(DateTimeFormatter.BASIC_ISO_DATE);
+    }
+
+    public static String nowDateTime() {
+        return nowDateTime(null);
+    }
+
+    public static String nowDateTime(DateTimeFormatter pattern) {
+        if (pattern == null) {
+            pattern = DateTimeFormatter.ISO_LOCAL_DATE_TIME;
+        }
+        return LocalDateTime.now().format(pattern);
+    }
+
+    public static String getYesterday() {
+        return LocalDate.now().plusDays(-1).format(DateTimeFormatter.BASIC_ISO_DATE);
+    }
+
+    public static final String FORMAT_YYYY_MM_DD_HH_MM_SS = "yyyy-MM-dd HH:mm:ss";
+
+    public static final String formatDate_YYYY_MM_DD_HH_MM_SS(Date date) {
+        String v = null;
+        try {
+            if (date == null)
+                return null;
+            SimpleDateFormat dateFormat = new SimpleDateFormat(FORMAT_YYYY_MM_DD_HH_MM_SS);
+            v = dateFormat.format(date);
+        } catch (Exception e) {
+            // do nothing
+        }
+        return v;
+    }
+
+    public static int getCurrentHours() {
+        return new GregorianCalendar().get(Calendar.HOUR_OF_DAY);
+    }
+    public static final String formatDate_YYYY_MM_DD_HH_MM_SS() {
+        return formatDate_YYYY_MM_DD_HH_MM_SS(new Date());
+    }
+
+    public static void main(String[] args) {
+        System.out.println(parseDate("20220724"));
+        System.out.println(parseDate("2022-07-21"));
+        System.out.println(getDate(-1));
+        System.out.println(nowDate(YYYYMMDD));
+        System.out.println(nowDate(YYYYMMDDHHMMSS));
+    }
+}

src/main/java/com/winhc/data/push/utils/JsonUtils.java

@@ -0,0 +1,20 @@
+package com.winhc.data.push.utils;
+
+
+import com.aliyun.openservices.shade.com.alibaba.fastjson.JSONObject;
+import com.aliyun.openservices.shade.com.alibaba.fastjson.serializer.SerializerFeature;
+
+/**
+ * @Author: XuJiakai
+ * @Date: 2020/6/22 14:18
+ * @Description:
+ */
+public class JsonUtils {
+    public static String jsonObjToString(Object jsonObject) {
+        return JSONObject.toJSONString(jsonObject, SerializerFeature.WriteMapNullValue);
+    }
+
+    public static String jsonObjToStringNotWriteMapNull(Object jsonObject) {
+        return JSONObject.toJSONString(jsonObject, SerializerFeature.IgnoreNonFieldGetter);
+    }
+}

src/main/java/com/winhc/data/push/vo/ResponseVo.java

@@ -0,0 +1,43 @@
+package com.winhc.data.push.vo;
+
+import lombok.Getter;
+import com.winhc.data.push.utils.BigDecimalUtil;
+
+/**
+ * @Author: XuJiakai
+ * @Date: 2020/9/2 16:55
+ * @Description:
+ */
+@Getter
+public class ResponseVo<T> {
+    private Boolean success;
+    private String msg;
+    private T data;
+    private String duration;
+
+    public static <T> ResponseVo success(long startTime, T data) {
+        long endTime = System.currentTimeMillis();
+        ResponseVo<T> responseVo = new ResponseVo<T>();
+        responseVo.success = true;
+        responseVo.msg = "success";
+        responseVo.data = data;
+        responseVo.duration = BigDecimalUtil.div(endTime - startTime, 1000L, 3) + "s";
+        return responseVo;
+    }
+
+    public static <T> ResponseVo failure(long startTime, String errorMsg) {
+        long endTime = System.currentTimeMillis();
+        ResponseVo<T> responseVo = new ResponseVo<T>();
+        responseVo.success = false;
+        responseVo.msg = errorMsg;
+        responseVo.duration = BigDecimalUtil.div(endTime - startTime, 1000L, 3) + "s";
+        return responseVo;
+    }
+
+    public static <T> ResponseVo failure(String errorMsg) {
+        ResponseVo<T> responseVo = new ResponseVo<T>();
+        responseVo.success = false;
+        responseVo.msg = errorMsg;
+        return responseVo;
+    }
+}

src/main/resources/application-dev.yml

@@ -0,0 +1,17 @@
+spring:
+  data:
+    mongodb:
+      uri: mongodb://itslaw:itslaw_168@dds-uf6ff5dfd9aef3641601-pub.mongodb.rds.aliyuncs.com:3717,dds-uf6ff5dfd9aef3642555-pub.mongodb.rds.aliyuncs.com:3717/itslaw?replicaSet=mgset-6501997
+
+oss:
+  endpoint: oss-cn-shanghai.aliyuncs.com
+  access_key_id: LTAI5tFzWLXJyXNPAAzdGCag
+  access_key_secret: cbvFayBxVDCjM2o5FvUgk6ZJhfpvO3
+  source_bucket_name: data-exchange-as
+  target_bucket_name: data-exchange-as
+  source_path_prefix: bigdata/auction_tracking/
+  target_path_prefix: anshuo/auction_tracking/
+
+
+scheduling:
+  enabled: true

src/main/resources/application-prod.yml

@@ -0,0 +1,16 @@
+spring:
+  data:
+    mongodb:
+      uri: mongodb://itslaw:itslaw_168@dds-uf6ff5dfd9aef3641.mongodb.rds.aliyuncs.com:3717,dds-uf6ff5dfd9aef3642.mongodb.rds.aliyuncs.com:3717/itslaw?replicaSet=mgset-6501997
+
+oss:
+  endpoint: oss-cn-shanghai-internal.aliyuncs.com
+  access_key_id: LTAI5tFzWLXJyXNPAAzdGCag
+  access_key_secret: cbvFayBxVDCjM2o5FvUgk6ZJhfpvO3
+  source_bucket_name: data-exchange-as
+  target_bucket_name: data-exchange-as
+  source_path_prefix: bigdata/auction_tracking/
+  target_path_prefix: anshuo/auction_tracking/
+
+scheduling:
+  enabled: true

src/main/resources/application.yml

@@ -0,0 +1,32 @@
+spring:
+  profiles:
+    active: ${SPRING_PROFILES_ACTIVE:prod}
+  jackson:
+    date-format: yyyy-MM-dd HH:mm:ss
+    time-zone: GMT+8
+
+  mvc:
+    pathmatch:
+      matching-strategy: ant_path_matcher
+
+server:
+  port: 9999
+
+data-works:
+  access-key-id: LTAI4G4n7pAW8tUbJVkkZQPD
+  access-key-secret: uNJOBskzcDqHq1TYG3m2rebR4c1009
+  region-id: cn-shanghai
+
+odps:
+  access-key-id: LTAI4G4n7pAW8tUbJVkkZQPD
+  access-key-secret: uNJOBskzcDqHq1TYG3m2rebR4c1009
+  region-id: cn-shanghai
+  default-project: winhc_ng
+  end-point: http://service.odps.aliyun.com/api
+
+dingding:
+  robots:
+    hooks:
+      - label: data_push_message
+        signature: SEC70a379808313f7fb97355a48f8f5c554d6862994d9b3c2f71942626502ef3ffa
+        webhook: https://oapi.dingtalk.com/robot/send?access_token=b35fc350bafbf907f01c556a484c399d4e65f3a9db943c2a8670ef201a872905

src/main/resources/data-works-param.yaml

@@ -0,0 +1,20 @@
+job:
+  - project: winhc_ng
+    flow: push_data_as_to_mongo
+    task:
+      - taskName: mongo_auction_tracking_push_ads
+        param:
+          - _nodeId: 700005473716
+            project: winhc_ng
+
+  - project: winhc_ng
+    flow: push_data_as_to_oss
+    task:
+      - taskName: oss_auction_tracking_push_ads
+        param:
+          - _nodeId: 700005473717
+            project: winhc_ng
+            sourceTable: inc_ads_auction_tracking_anshuo_push
+            ossPrefix: anshuo/auction_tracking/
+
+

src/test/java/com/winhc/data/push/Demo.java

@@ -0,0 +1,71 @@
+package com.winhc.data.push;
+
+import com.aliyun.oss.ClientException;
+import com.aliyun.oss.OSS;
+import com.aliyun.oss.OSSClientBuilder;
+import com.aliyun.oss.OSSException;
+import com.aliyun.oss.model.*;
+
+
+public class Demo {
+    public static void main(String[] args) throws Exception {
+        // Endpoint以华东1（杭州）为例，其它Region请按实际情况填写。
+        String endpoint = "https://oss-cn-shanghai.aliyuncs.com";
+        // 阿里云账号AccessKey拥有所有API的访问权限，风险很高。强烈建议您创建并使用RAM用户进行API访问或日常运维，请登录RAM控制台创建RAM用户。
+        String accessKeyId = "LTAI5tDrbWcjKkzTnJXCrwkC";
+        String accessKeySecret = "AWDvBsQc16AYMszNVRga5ANbzuLowy";
+
+
+
+//        accessKeyId = "LTAI6HKo33DbtiBI";
+//        accessKeySecret = "8FmSBhHyaz4jtyJhiimK0NGF5rBPMl";
+        // 填写Bucket名称，例如examplebucket。
+        //String bucketName = "data-exchange-as";
+        String bucketName = "bigdata-rt";
+
+        // 创建OSSClient实例。
+        OSS ossClient = new OSSClientBuilder().build(endpoint, accessKeyId, accessKeySecret);
+
+        try {
+            // 构造ListObjectsV2Request请求。
+            ListObjectsV2Request listObjectsV2Request = new ListObjectsV2Request(bucketName);
+
+            // 设置prefix参数来获取fun目录下的文件。
+            //listObjectsV2Request.setPrefix("anshuo/auction_tracking/20220721/file/");
+            //listObjectsV2Request.setPrefix("anshuo/");
+            //listObjectsV2Request.setPrefix("bigdata/auction_tracking/667169871995/");
+            listObjectsV2Request.setPrefix("business-schema/");
+
+            // 发起列举请求。
+            ListObjectsV2Result result = ossClient.listObjectsV2(listObjectsV2Request);
+
+            // 遍历文件。
+            System.out.println("Objects:");
+            for (OSSObjectSummary objectSummary : result.getObjectSummaries()) {
+                System.out.println(objectSummary.getKey());
+            }
+
+            // 遍历commonPrefix。
+            System.out.println("\nCommonPrefixes:");
+            for (String commonPrefix : result.getCommonPrefixes()) {
+                System.out.println(commonPrefix);
+            }
+        } catch (OSSException oe) {
+            System.out.println("Caught an OSSException, which means your request made it to OSS, "
+                    + "but was rejected with an error response for some reason.");
+            System.out.println("Error Message:" + oe.getErrorMessage());
+            System.out.println("Error Code:" + oe.getErrorCode());
+            System.out.println("Request ID:" + oe.getRequestId());
+            System.out.println("Host ID:" + oe.getHostId());
+        } catch (ClientException ce) {
+            System.out.println("Caught an ClientException, which means the client encountered "
+                    + "a serious internal problem while trying to communicate with OSS, "
+                    + "such as not being able to access the network.");
+            System.out.println("Error Message:" + ce.getMessage());
+        } finally {
+            if (ossClient != null) {
+                ossClient.shutdown();
+            }
+        }
+    }
+}

src/test/java/com/winhc/data/push/TestPush.java

@@ -0,0 +1,42 @@
+package com.winhc.data.push;
+
+import com.winhc.data.push.service.SynDataService;
+import com.winhc.data.push.service.TouchService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+
+/**
+ * @author π
+ * @Description:
+ * @date 2022/7/21 14:19
+ */
+@RunWith(SpringRunner.class)
+@SpringBootTest
+public class TestPush {
+
+    @Autowired
+    SynDataService synDataService;
+
+    @Autowired
+    TouchService touchService;
+
+    @Test
+    public void sendKafka() {
+        String ds = "20220721";
+        synDataService.push("","");
+    }
+
+    @Test
+    public void touchTask() {
+        touchService.export2OSS("winhc_ng","20220721");
+    }
+
+    @Test
+    public void touchTask1() throws Exception {
+        synDataService.sendMessage("运行中...");
+    }
+}