Преглед изворни кода

es标签;更新账户流水;

zhouzeyu@c-top.com.cn пре 3 година
родитељ
комит
186305119b
26 измењених фајлова са 1734 додато и 2 уклоњено
  1. 15 0
      jeecg-boot-module-system/pom.xml
  2. 91 0
      jeecg-boot-module-system/src/main/java/org/jeecg/config/Elasticsearch/ElasticsearchConfig.java
  3. 83 0
      jeecg-boot-module-system/src/main/java/org/jeecg/modules/elasticsearch/controller/ESController.java
  4. 15 0
      module-common/pom.xml
  5. 27 0
      module-common/src/main/java/cn/com/ctop/common/module/annotation/Document.java
  6. 16 0
      module-common/src/main/java/cn/com/ctop/common/module/annotation/EsId.java
  7. 26 0
      module-common/src/main/java/cn/com/ctop/common/module/annotation/FieId.java
  8. 24 0
      module-common/src/main/java/cn/com/ctop/common/module/entity/DataLabels.java
  9. 29 0
      module-common/src/main/java/cn/com/ctop/common/module/enums/AnalyzerType.java
  10. 40 0
      module-common/src/main/java/cn/com/ctop/common/module/enums/FieldType.java
  11. 599 0
      module-common/src/main/java/cn/com/ctop/common/module/utils/ElasticsearchUtil.java
  12. 37 0
      module-job-bytedance/src/main/java/cn/com/ctop/job/bytedance/handler/SynchronousAccountTransactionDetailsJob.java
  13. 4 2
      module-kuaishou/src/main/java/cn/com/ctop/kuaishou/modules/clean/mapper/xml/KuaishouAccountCleanTemplateMapper.xml
  14. 30 0
      module-kuaishou/src/main/java/cn/com/ctop/kuaishou/modules/label/controller/LbelController.java
  15. 25 0
      module-kuaishou/src/main/java/cn/com/ctop/kuaishou/modules/label/mapper/MaterialLabelMapper.java
  16. 25 0
      module-kuaishou/src/main/java/cn/com/ctop/kuaishou/modules/label/mapper/xml/MaterialTagMapper.xml
  17. 14 0
      module-kuaishou/src/main/java/cn/com/ctop/kuaishou/modules/label/service/MaterialTagService.java
  18. 80 0
      module-kuaishou/src/main/java/cn/com/ctop/kuaishou/modules/label/service/impl/MaterialTagServiceImpl.java
  19. 30 0
      module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/agent/controller/AgentManagementController.java
  20. 73 0
      module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/agent/entity/TransactionDetails.java
  21. 15 0
      module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/agent/mapper/CtopAccountTransactionDetailsMapper.java
  22. 16 0
      module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/agent/mapper/CtopAgentAccountMapper.java
  23. 30 0
      module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/agent/mapper/xml/CtopAccountTransactionDetailsMapper.xml
  24. 28 0
      module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/agent/mapper/xml/CtopAgentAccountMapper.xml
  25. 22 0
      module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/agent/service/AgentManagementService.java
  26. 340 0
      module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/agent/service/impl/AgentManagementServiceImpl.java

+ 15 - 0
jeecg-boot-module-system/pom.xml

@@ -174,6 +174,21 @@
             <artifactId>spring-boot-starter-jimureport</artifactId>
             <version>1.1-beta</version>
         </dependency>
+        <dependency>
+            <groupId>org.elasticsearch.client</groupId>
+            <artifactId>elasticsearch-rest-high-level-client</artifactId>
+            <version>7.1.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.elasticsearch.client</groupId>
+            <artifactId>elasticsearch-rest-client</artifactId>
+            <version>7.1.0</version>
+        </dependency>
+        <dependency>
+            <groupId>org.elasticsearch</groupId>
+            <artifactId>elasticsearch</artifactId>
+            <version>7.1.0</version>
+        </dependency>
     </dependencies>
 
     <dependencyManagement>

+ 91 - 0
jeecg-boot-module-system/src/main/java/org/jeecg/config/Elasticsearch/ElasticsearchConfig.java

@@ -0,0 +1,91 @@
+package org.jeecg.config.Elasticsearch;
+
+import org.apache.http.HttpHost;
+import org.apache.http.client.config.RequestConfig;
+import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestClientBuilder;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.ArrayList;
+
+/**
+ * @className: ElasticsearchConfig
+ * @description: elasticsearch 配置类
+ * @author: zzy
+ * @date: 2021-07-12
+ */
+@Configuration
+public class ElasticsearchConfig {
+    @Value("${elasticsearch.address}")
+    private String address;
+
+    /**
+     *  连接超时时间
+     */
+    @Value("${elasticsearch.connect-timeout}")
+    private int connectTimeOut = 1000;
+
+    /**
+     * 连接超时时间
+     */
+    @Value("${elasticsearch.socket-timeout}")
+    private int socketTimeOut = 30000;
+
+    /**
+     * 获取连接的超时时间
+     */
+    @Value("${elasticsearch.connection-request-timeout}")
+    private int connectionRequestTimeOut = 500;
+
+    /**
+     * 最大连接数
+     */
+    @Value("${elasticsearch.max-connect-num}")
+    private int maxConnectNum = 100;
+
+    /**
+     * 最大路由连接数
+     */
+    @Value("${elasticsearch.max-connect-per-route}")
+    private int maxConnectPerRoute = 100;
+
+
+    @Bean
+    RestHighLevelClient restHighLevelClient() {
+        ArrayList<HttpHost> hostList = new ArrayList<>();
+        String[] addrss = address.split(",");
+        for(String addr : addrss){
+            String[] arr = addr.split(":");
+            hostList.add(new HttpHost(arr[0], Integer.parseInt(arr[1]), "http"));
+        }
+
+        RestClientBuilder builder = RestClient.builder(hostList.toArray(new HttpHost[0]));
+        // 异步httpclient连接延时配置
+        builder.setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
+            @Override
+            public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
+                requestConfigBuilder.setConnectTimeout(connectTimeOut);
+                requestConfigBuilder.setSocketTimeout(socketTimeOut);
+                requestConfigBuilder.setConnectionRequestTimeout(connectionRequestTimeOut);
+                return requestConfigBuilder;
+            }
+        });
+        // 异步httpclient连接数配置
+        builder.setHttpClientConfigCallback(new RestClientBuilder.HttpClientConfigCallback() {
+            @Override
+            public HttpAsyncClientBuilder customizeHttpClient(HttpAsyncClientBuilder httpClientBuilder) {
+                httpClientBuilder.setMaxConnTotal(maxConnectNum);
+                httpClientBuilder.setMaxConnPerRoute(maxConnectPerRoute);
+                return httpClientBuilder;
+            }
+        });
+
+        RestHighLevelClient client = new RestHighLevelClient(builder);
+        return client;
+    }
+
+}

+ 83 - 0
jeecg-boot-module-system/src/main/java/org/jeecg/modules/elasticsearch/controller/ESController.java

@@ -0,0 +1,83 @@
+package org.jeecg.modules.elasticsearch.controller;
+
+import cn.com.ctop.common.module.entity.DataLabels;
+import cn.com.ctop.common.module.utils.ElasticsearchUtil;
+import com.github.pagehelper.PageInfo;
+import lombok.extern.slf4j.Slf4j;
+import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.index.query.BoolQueryBuilder;
+import org.elasticsearch.index.query.QueryBuilders;
+import org.elasticsearch.index.query.WildcardQueryBuilder;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.UUID;
+
+/**
+ * es请求
+ *
+ * @author zzy
+ * @create: 2021-07-12
+ */
+@Slf4j
+@RestController
+@RequestMapping("/es")
+public class ESController {
+
+    @Autowired
+    ElasticsearchUtil elasticsearchUtil;
+
+    @RequestMapping("createIndex")
+    public void createIndex() {
+        DataLabels dataLabels = new DataLabels();
+        dataLabels
+                .setId("3");
+        dataLabels
+                .setContent("哦哦哦,测试es啦,I LOVE CHINA");
+        dataLabels
+                .setSize(1024);
+        try {
+            boolean indexIfNotExist = elasticsearchUtil.createIndexIfNotExist(DataLabels.class);
+            log.info("创建成功:{}", indexIfNotExist);
+        } catch (Exception e) {
+            log.info("创建错误:{}", e.toString());
+        }
+    }
+
+    @RequestMapping("addData")
+    public void addData() {
+        DataLabels dataLabels = new DataLabels();
+        dataLabels
+                .setId(UUID.randomUUID().toString());
+        dataLabels
+                .setContent("哦哦哦,测试es啦,I LOVE CHINA");
+        dataLabels
+                .setSize(1024);
+        try {
+            IndexResponse indexResponse = elasticsearchUtil.index(dataLabels);
+            log.info("添加数据成功:{}", indexResponse.toString());
+        } catch (Exception e) {
+            log.info("添加数据错误:{}", e.toString());
+            e.printStackTrace();
+        }
+    }
+
+    @RequestMapping("searchData")
+    public void searchData() {
+        SearchSourceBuilder SearchSourceBuilder = new SearchSourceBuilder();
+        WildcardQueryBuilder wildcardQueryBuilder = QueryBuilders.wildcardQuery("content", "*LOVE*");
+
+        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
+        boolQueryBuilder.must(wildcardQueryBuilder);
+        SearchSourceBuilder.query(boolQueryBuilder);
+        try {
+            PageInfo<DataLabels> indexResponse = elasticsearchUtil.search(SearchSourceBuilder, 1, 10, DataLabels.class);
+            log.info("查询数据成功:{}", indexResponse.toString());
+        } catch (Exception e) {
+            log.info("查询数据错误:{}", e.toString());
+            e.printStackTrace();
+        }
+    }
+}

+ 15 - 0
module-common/pom.xml

@@ -102,6 +102,21 @@
             <version>1.16</version>
             <scope>compile</scope>
         </dependency>
+        <dependency>
+            <groupId>org.elasticsearch.client</groupId>
+            <artifactId>elasticsearch-rest-high-level-client</artifactId>
+            <version>7.6.1</version>
+        </dependency>
+        <dependency>
+            <groupId>org.elasticsearch.client</groupId>
+            <artifactId>elasticsearch-rest-client</artifactId>
+            <version>7.6.1</version>
+        </dependency>
+        <dependency>
+            <groupId>org.elasticsearch</groupId>
+            <artifactId>elasticsearch</artifactId>
+            <version>7.6.1</version>
+        </dependency>
 
     </dependencies>
 

+ 27 - 0
module-common/src/main/java/cn/com/ctop/common/module/annotation/Document.java

@@ -0,0 +1,27 @@
+package cn.com.ctop.common.module.annotation;
+
+import java.lang.annotation.*;
+
+/**
+ * Es 文档注解,用于做索引实体映射
+ * 作用在类上
+ * @author zzy
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.TYPE)
+@Documented
+@Inherited
+public @interface  Document {
+
+    /**
+     * index : 索引名称
+     * @return
+     */
+    String index();
+
+    /**
+     * 类型名称
+     * @return
+     */
+    String type();
+}

+ 16 - 0
module-common/src/main/java/cn/com/ctop/common/module/annotation/EsId.java

@@ -0,0 +1,16 @@
+package cn.com.ctop.common.module.annotation;
+
+import java.lang.annotation.*;
+
+/**
+ * 用于标识使用 该字段作为ES数据中的id
+ * @author zzy
+ * @create: 2021-07-12
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.FIELD)
+@Documented
+@Inherited
+public @interface EsId {
+
+}

+ 26 - 0
module-common/src/main/java/cn/com/ctop/common/module/annotation/FieId.java

@@ -0,0 +1,26 @@
+package cn.com.ctop.common.module.annotation;
+
+import cn.com.ctop.common.module.enums.AnalyzerType;
+import cn.com.ctop.common.module.enums.FieldType;
+
+import java.lang.annotation.*;
+
+/**
+ * 作用在字段上,用于定义类型,映射关系
+ * @author zzy
+ * @create: 2021-07-12
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target(ElementType.FIELD)
+@Documented
+@Inherited
+public @interface FieId {
+    FieldType type() default FieldType.TEXT;
+
+    /**
+     * 指定分词器
+     * @return
+     */
+    AnalyzerType analyzer() default AnalyzerType.STANDARD;
+
+}

+ 24 - 0
module-common/src/main/java/cn/com/ctop/common/module/entity/DataLabels.java

@@ -0,0 +1,24 @@
+package cn.com.ctop.common.module.entity;
+
+import cn.com.ctop.common.module.annotation.Document;
+import cn.com.ctop.common.module.annotation.EsId;
+import cn.com.ctop.common.module.annotation.FieId;
+
+import cn.com.ctop.common.module.enums.AnalyzerType;
+import cn.com.ctop.common.module.enums.FieldType;
+import lombok.Data;
+
+import java.io.Serializable;
+
+@Data
+@Document(index = "data_labels", type = "lable")
+public class DataLabels implements Serializable {
+    private static final long serialVersionUID = 1L;
+    @EsId
+    @FieId(type = FieldType.KEYWORD)
+    private String id;
+    @FieId(type = FieldType.KEYWORD,analyzer = AnalyzerType.IK_SMART)
+    private String content;
+    @FieId(type = FieldType.KEYWORD)
+    private Integer size;
+}

+ 29 - 0
module-common/src/main/java/cn/com/ctop/common/module/enums/AnalyzerType.java

@@ -0,0 +1,29 @@
+package cn.com.ctop.common.module.enums;
+
+import lombok.Getter;
+
+@Getter
+public enum AnalyzerType {
+    NO("不使用分词"),
+    /**
+     * 标准分词,默认分词器
+     */
+    STANDARD("standard"),
+
+    /**
+     * ik_smart:会做最粗粒度的拆分;已被分出的词语将不会再次被其它词语占有
+     */
+    IK_SMART("ik_smart"),
+
+    /**
+     * ik_max_word :会将文本做最细粒度的拆分;尽可能多的拆分出词语
+     */
+    IK_MAX_WORD("ik_max_word");
+
+    private String type;
+
+    AnalyzerType(String type) {
+        this.type = type;
+    }
+
+}

+ 40 - 0
module-common/src/main/java/cn/com/ctop/common/module/enums/FieldType.java

@@ -0,0 +1,40 @@
+package cn.com.ctop.common.module.enums;
+
+import lombok.Getter;
+
+@Getter
+public enum FieldType {
+    /**
+     * text
+     */
+    TEXT("text"),
+
+    KEYWORD("keyword"),
+
+    INTEGER("integer"),
+
+    DOUBLE("double"),
+
+    DATE("date"),
+
+    /**
+     * 单条数据
+     */
+    OBJECT("object"),
+
+    /**
+     * 嵌套数组
+     */
+    NESTED("nested"),
+
+
+    ;
+
+
+    FieldType(String type) {
+        this.type = type;
+    }
+
+    private String type;
+
+}

+ 599 - 0
module-common/src/main/java/cn/com/ctop/common/module/utils/ElasticsearchUtil.java

@@ -0,0 +1,599 @@
+package cn.com.ctop.common.module.utils;
+
+
+import cn.com.ctop.common.module.annotation.Document;
+import cn.com.ctop.common.module.annotation.EsId;
+import cn.com.ctop.common.module.annotation.FieId;
+import cn.com.ctop.common.module.enums.FieldType;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import com.github.pagehelper.PageInfo;
+import lombok.extern.slf4j.Slf4j;
+import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
+import org.elasticsearch.action.bulk.BulkItemResponse;
+import org.elasticsearch.action.bulk.BulkRequest;
+import org.elasticsearch.action.bulk.BulkResponse;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.delete.DeleteResponse;
+import org.elasticsearch.action.get.GetRequest;
+import org.elasticsearch.action.get.GetResponse;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.index.IndexResponse;
+import org.elasticsearch.action.search.SearchRequest;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.action.support.IndicesOptions;
+import org.elasticsearch.action.support.master.AcknowledgedResponse;
+import org.elasticsearch.action.support.replication.ReplicationResponse;
+import org.elasticsearch.action.update.UpdateRequest;
+import org.elasticsearch.action.update.UpdateResponse;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.client.indices.CreateIndexRequest;
+import org.elasticsearch.client.indices.CreateIndexResponse;
+import org.elasticsearch.client.indices.GetIndexRequest;
+import org.elasticsearch.client.indices.PutMappingRequest;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.search.SearchHit;
+import org.elasticsearch.search.SearchHits;
+import org.elasticsearch.search.builder.SearchSourceBuilder;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @className: EsUtil
+ * @description: es 操作工具类;
+ * 这里均采用同步调用的方式
+ * @author: zzy
+ * @create: 2021-07-12
+ */
+@Component
+@Slf4j
+public class ElasticsearchUtil {
+
+    @Resource
+    private RestHighLevelClient restHighLevelClient;
+
+    /**
+     * 创建索引(默认分片数为5和副本数为1)
+     *
+     * @param clazz 根据实体自动映射es索引
+     * @throws IOException
+     */
+    public boolean createIndex(Class clazz) throws Exception {
+        Document declaredAnnotation = (Document) clazz.getDeclaredAnnotation(Document.class);
+        if (declaredAnnotation == null) {
+            throw new Exception(String.format("class name: %s can not find Annotation [Document], please check", clazz.getName()));
+        }
+        String indexName = declaredAnnotation.index();
+        CreateIndexRequest request = new CreateIndexRequest(indexName);
+        request.settings(Settings.builder()
+                // 设置分片数为3, 副本为2
+                .put("index.number_of_shards", 3)
+                .put("index.number_of_replicas", 2)
+        );
+        request.mapping(generateBuilder(clazz));
+        CreateIndexResponse response = restHighLevelClient.indices().create(request, RequestOptions.DEFAULT);
+        // 指示是否所有节点都已确认请求
+        boolean acknowledged = response.isAcknowledged();
+        // 指示是否在超时之前为索引中的每个分片启动了必需的分片副本数
+        boolean shardsAcknowledged = response.isShardsAcknowledged();
+        if (acknowledged || shardsAcknowledged) {
+            log.info("创建索引成功!索引名称为{}", indexName);
+            return true;
+        }
+        return false;
+    }
+
+    /**
+     * 创建索引(默认分片数为5和副本数为1)
+     *
+     * @param clazz 根据实体自动映射es索引
+     * @throws IOException
+     */
+    public boolean createIndexIfNotExist(Class clazz) throws Exception {
+        Document declaredAnnotation = (Document) clazz.getDeclaredAnnotation(Document.class);
+        if (declaredAnnotation == null) {
+            throw new Exception(String.format("class name: %s can not find Annotation [Document], please check", clazz.getName()));
+        }
+        String indexName = declaredAnnotation.index();
+
+        boolean indexExists = isIndexExists(indexName);
+        if (!indexExists) {
+            CreateIndexRequest request = new CreateIndexRequest(indexName);
+            request.settings(Settings.builder()
+                    // 设置分片数为3, 副本为2
+                    .put("index.number_of_shards", 3)
+                    .put("index.number_of_replicas", 2)
+            );
+            request.mapping(generateBuilder(clazz));
+            CreateIndexResponse response = restHighLevelClient.indices().create(request, RequestOptions.DEFAULT);
+            // 指示是否所有节点都已确认请求
+            boolean acknowledged = response.isAcknowledged();
+            // 指示是否在超时之前为索引中的每个分片启动了必需的分片副本数
+            boolean shardsAcknowledged = response.isShardsAcknowledged();
+            if (acknowledged || shardsAcknowledged) {
+                log.info("创建索引成功!索引名称为{}", indexName);
+                return true;
+            }
+        }
+        return false;
+    }
+
+    /**
+     * 更新索引(默认分片数为5和副本数为1):
+     * 只能给索引上添加一些不存在的字段
+     * 已经存在的映射不能改
+     *
+     * @param clazz 根据实体自动映射es索引
+     * @throws IOException
+     */
+    public boolean updateIndex(Class clazz) throws Exception {
+        Document declaredAnnotation = (Document) clazz.getDeclaredAnnotation(Document.class);
+        if (declaredAnnotation == null) {
+            throw new Exception(String.format("class name: %s can not find Annotation [Document], please check", clazz.getName()));
+        }
+        String indexName = declaredAnnotation.index();
+        PutMappingRequest request = new PutMappingRequest(indexName);
+
+        request.source(generateBuilder(clazz));
+        AcknowledgedResponse response = restHighLevelClient.indices().putMapping(request, RequestOptions.DEFAULT);
+        // 指示是否所有节点都已确认请求
+        boolean acknowledged = response.isAcknowledged();
+
+        if (acknowledged) {
+            log.info("更新索引索引成功!索引名称为{}", indexName);
+            return true;
+        }
+        return false;
+    }
+
+    /**
+     * 删除索引
+     *
+     * @param indexName
+     * @return
+     */
+    public boolean delIndex(String indexName) {
+        boolean acknowledged = false;
+        try {
+            DeleteIndexRequest deleteIndexRequest = new DeleteIndexRequest(indexName);
+            deleteIndexRequest.indicesOptions(IndicesOptions.LENIENT_EXPAND_OPEN);
+            AcknowledgedResponse delete = restHighLevelClient.indices().delete(deleteIndexRequest, RequestOptions.DEFAULT);
+            acknowledged = delete.isAcknowledged();
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        return acknowledged;
+    }
+
+    /**
+     * 判断索引是否存在
+     *
+     * @param indexName
+     * @return
+     */
+    public boolean isIndexExists(String indexName) {
+        boolean exists = false;
+        try {
+            GetIndexRequest getIndexRequest = new GetIndexRequest(indexName);
+            getIndexRequest.humanReadable(true);
+            exists = restHighLevelClient.indices().exists(getIndexRequest, RequestOptions.DEFAULT);
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+        return exists;
+    }
+
+
+    /**
+     * 添加单条数据
+     * 提供多种方式:
+     * 1. json
+     * 2. map
+     * Map<String, Object> jsonMap = new HashMap<>();
+     * jsonMap.put("user", "kimchy");
+     * jsonMap.put("postDate", new Date());
+     * jsonMap.put("message", "trying out Elasticsearch");
+     * IndexRequest indexRequest = new IndexRequest("posts")
+     * .id("1").source(jsonMap);
+     * 3. builder
+     * XContentBuilder builder = XContentFactory.jsonBuilder();
+     * builder.startObject();
+     * {
+     * builder.field("user", "kimchy");
+     * builder.timeField("postDate", new Date());
+     * builder.field("message", "trying out Elasticsearch");
+     * }
+     * builder.endObject();
+     * IndexRequest indexRequest = new IndexRequest("posts")
+     * .id("1").source(builder);
+     * 4. source:
+     * IndexRequest indexRequest = new IndexRequest("posts")
+     * .id("1")
+     * .source("user", "kimchy",
+     * "postDate", new Date(),
+     * "message", "trying out Elasticsearch");
+     * <p>
+     * 报错:  Validation Failed: 1: type is missing;
+     * 加入两个jar包解决
+     * <p>
+     * 提供新增或修改的功能
+     *
+     * @return
+     */
+    public IndexResponse index(Object o) throws Exception {
+        Document declaredAnnotation = (Document) o.getClass().getDeclaredAnnotation(Document.class);
+        if (declaredAnnotation == null) {
+            throw new Exception(String.format("class name: %s can not find Annotation [Document], please check", o.getClass().getName()));
+        }
+        String indexName = declaredAnnotation.index();
+
+        IndexRequest request = new IndexRequest(indexName);
+        Field fieldByAnnotation = getFieldByAnnotation(o, EsId.class);
+        if (fieldByAnnotation != null) {
+            fieldByAnnotation.setAccessible(true);
+            try {
+                Object id = fieldByAnnotation.get(o);
+                request = request.id(id.toString());
+            } catch (IllegalAccessException e) {
+                log.error("获取id字段出错:{}", e);
+            }
+        }
+
+        String userJson = JSON.toJSONString(o);
+        request.source(userJson, XContentType.JSON);
+        IndexResponse indexResponse = restHighLevelClient.index(request, RequestOptions.DEFAULT);
+        return indexResponse;
+    }
+
+
+    /**
+     * 根据id查询
+     *
+     * @return
+     */
+    public String queryById(String indexName, String id) throws IOException {
+        GetRequest getRequest = new GetRequest(indexName, id);
+        // getRequest.fetchSourceContext(FetchSourceContext.DO_NOT_FETCH_SOURCE);
+
+        GetResponse getResponse = restHighLevelClient.get(getRequest, RequestOptions.DEFAULT);
+        String jsonStr = getResponse.getSourceAsString();
+        return jsonStr;
+    }
+
+    /**
+     * 查询封装返回json字符串
+     *
+     * @param indexName
+     * @param searchSourceBuilder
+     * @return
+     * @throws IOException
+     */
+    public String search(String indexName, SearchSourceBuilder searchSourceBuilder) throws IOException {
+        SearchRequest searchRequest = new SearchRequest(indexName);
+        searchRequest.source(searchSourceBuilder);
+        searchRequest.scroll(TimeValue.timeValueMinutes(1L));
+        SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
+        String scrollId = searchResponse.getScrollId();
+        SearchHits hits = searchResponse.getHits();
+        JSONArray jsonArray = new JSONArray();
+        for (SearchHit hit : hits) {
+            String sourceAsString = hit.getSourceAsString();
+            JSONObject jsonObject = JSON.parseObject(sourceAsString);
+            jsonArray.add(jsonObject);
+        }
+        log.info("返回总数为:" + hits.getTotalHits());
+        return jsonArray.toJSONString();
+    }
+
+    /**
+     * 查询封装,带分页
+     *
+     * @param searchSourceBuilder
+     * @param pageNum
+     * @param pageSize
+     * @param s
+     * @param <T>
+     * @return
+     * @throws IOException
+     */
+    public <T> PageInfo<T> search(SearchSourceBuilder searchSourceBuilder, int pageNum, int pageSize, Class<T> s) throws Exception {
+        Document declaredAnnotation = (Document) s.getDeclaredAnnotation(Document.class);
+        if (declaredAnnotation == null) {
+            throw new Exception(String.format("class name: %s can not find Annotation [Document], please check", s.getName()));
+        }
+        String indexName = declaredAnnotation.index();
+        SearchRequest searchRequest = new SearchRequest(indexName);
+        searchRequest.source(searchSourceBuilder);
+        SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
+        SearchHits hits = searchResponse.getHits();
+        JSONArray jsonArray = new JSONArray();
+        for (SearchHit hit : hits) {
+            String sourceAsString = hit.getSourceAsString();
+            JSONObject jsonObject = JSON.parseObject(sourceAsString);
+            jsonArray.add(jsonObject);
+        }
+        log.info("返回总数为:" + hits.getTotalHits());
+        int total = (int) hits.getTotalHits().value;
+
+        // 封装分页
+        List<T> list = jsonArray.toJavaList(s);
+        PageInfo<T> page = new PageInfo<>();
+        page.setList(list);
+        page.setPageNum(pageNum);
+        page.setPageSize(pageSize);
+        page.setTotal(total);
+        page.setPages(total == 0 ? 0 : (total % pageSize == 0 ? total / pageSize : (total / pageSize) + 1));
+        page.setHasNextPage(page.getPageNum() < page.getPages());
+        return page;
+    }
+
+    /**
+     * 查询封装,返回集合
+     *
+     * @param searchSourceBuilder
+     * @param s
+     * @param <T>
+     * @return
+     * @throws IOException
+     */
+    public <T> List<T> search(SearchSourceBuilder searchSourceBuilder, Class<T> s) throws Exception {
+        Document declaredAnnotation = (Document) s.getDeclaredAnnotation(Document.class);
+        if (declaredAnnotation == null) {
+            throw new Exception(String.format("class name: %s can not find Annotation [Document], please check", s.getName()));
+        }
+        String indexName = declaredAnnotation.index();
+        SearchRequest searchRequest = new SearchRequest(indexName);
+        searchRequest.source(searchSourceBuilder);
+        searchRequest.scroll(TimeValue.timeValueMinutes(1L));
+        SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
+        String scrollId = searchResponse.getScrollId();
+        SearchHits hits = searchResponse.getHits();
+        JSONArray jsonArray = new JSONArray();
+        for (SearchHit hit : hits) {
+            String sourceAsString = hit.getSourceAsString();
+            JSONObject jsonObject = JSON.parseObject(sourceAsString);
+            jsonArray.add(jsonObject);
+        }
+        // 封装分页
+        List<T> list = jsonArray.toJavaList(s);
+        return list;
+    }
+
+
+    /**
+     * 批量插入文档
+     * 文档存在 则插入
+     * 文档不存在 则更新
+     *
+     * @param list
+     * @return
+     */
+    public <T> boolean batchSaveOrUpdate(List<T> list) throws Exception {
+        Object o1 = list.get(0);
+        Document declaredAnnotation = (Document) o1.getClass().getDeclaredAnnotation(Document.class);
+        if (declaredAnnotation == null) {
+            throw new Exception(String.format("class name: %s can not find Annotation [@Document], please check", o1.getClass().getName()));
+        }
+        String indexName = declaredAnnotation.index();
+
+        BulkRequest request = new BulkRequest(indexName);
+        for (Object o : list) {
+            String jsonStr = JSON.toJSONString(o);
+            IndexRequest indexReq = new IndexRequest().source(jsonStr, XContentType.JSON);
+
+            Field fieldByAnnotation = getFieldByAnnotation(o, EsId.class);
+            if (fieldByAnnotation != null) {
+                fieldByAnnotation.setAccessible(true);
+                try {
+                    Object id = fieldByAnnotation.get(o);
+                    indexReq = indexReq.id(id.toString());
+                } catch (IllegalAccessException e) {
+                    log.error("获取id字段出错:{}", e);
+                }
+            }
+            request.add(indexReq);
+        }
+        BulkResponse bulkResponse = restHighLevelClient.bulk(request, RequestOptions.DEFAULT);
+
+        for (BulkItemResponse bulkItemResponse : bulkResponse) {
+            DocWriteResponse itemResponse = bulkItemResponse.getResponse();
+            IndexResponse indexResponse = (IndexResponse) itemResponse;
+            log.info("单条返回结果:{}", indexResponse);
+            if (bulkItemResponse.isFailed()) {
+                log.error("es 返回错误{}", bulkItemResponse.getFailureMessage());
+                return false;
+            }
+        }
+        return true;
+    }
+
+    /**
+     * 删除文档
+     *
+     * @param indexName: 索引名称
+     * @param docId:     文档id
+     */
+    public boolean deleteDoc(String indexName, String docId) throws IOException {
+        DeleteRequest request = new DeleteRequest(indexName, docId);
+        DeleteResponse deleteResponse = restHighLevelClient.delete(request, RequestOptions.DEFAULT);
+        // 解析response
+        String index = deleteResponse.getIndex();
+        String id = deleteResponse.getId();
+        long version = deleteResponse.getVersion();
+        ReplicationResponse.ShardInfo shardInfo = deleteResponse.getShardInfo();
+        if (shardInfo.getFailed() > 0) {
+            for (ReplicationResponse.ShardInfo.Failure failure :
+                    shardInfo.getFailures()) {
+                String reason = failure.reason();
+                log.info("删除失败,原因为 {}", reason);
+            }
+        }
+        return true;
+    }
+
+    /**
+     * 根据json类型更新文档
+     *
+     * @param indexName
+     * @param docId
+     * @param o
+     * @return
+     * @throws IOException
+     */
+    public boolean updateDoc(String indexName, String docId, Object o) throws IOException {
+        UpdateRequest request = new UpdateRequest(indexName, docId);
+        request.doc(JSON.toJSONString(o), XContentType.JSON);
+        UpdateResponse updateResponse = restHighLevelClient.update(request, RequestOptions.DEFAULT);
+        String index = updateResponse.getIndex();
+        String id = updateResponse.getId();
+        long version = updateResponse.getVersion();
+        if (updateResponse.getResult() == DocWriteResponse.Result.CREATED) {
+            return true;
+        } else if (updateResponse.getResult() == DocWriteResponse.Result.UPDATED) {
+            return true;
+        } else if (updateResponse.getResult() == DocWriteResponse.Result.DELETED) {
+        } else if (updateResponse.getResult() == DocWriteResponse.Result.NOOP) {
+
+        }
+        return false;
+    }
+
+    /**
+     * 根据Map类型更新文档
+     *
+     * @param indexName
+     * @param docId
+     * @param map
+     * @return
+     * @throws IOException
+     */
+    public boolean updateDoc(String indexName, String docId, Map<String, Object> map) throws IOException {
+        UpdateRequest request = new UpdateRequest(indexName, docId);
+        request.doc(map);
+        UpdateResponse updateResponse = restHighLevelClient.update(request, RequestOptions.DEFAULT);
+        String index = updateResponse.getIndex();
+        String id = updateResponse.getId();
+        long version = updateResponse.getVersion();
+        if (updateResponse.getResult() == DocWriteResponse.Result.CREATED) {
+            return true;
+        } else if (updateResponse.getResult() == DocWriteResponse.Result.UPDATED) {
+            return true;
+        } else if (updateResponse.getResult() == DocWriteResponse.Result.DELETED) {
+        } else if (updateResponse.getResult() == DocWriteResponse.Result.NOOP) {
+
+        }
+        return false;
+    }
+
+
+    public XContentBuilder generateBuilder(Class clazz) throws IOException {
+        // 获取索引名称及类型
+        Document doc = (Document) clazz.getAnnotation(Document.class);
+        log.info("docIndex:{}", doc.index());
+        log.info("docType:{}", doc.type());
+
+        XContentBuilder builder = XContentFactory.jsonBuilder();
+        builder.startObject();
+        builder.startObject("properties");
+        Field[] declaredFields = clazz.getDeclaredFields();
+        for (Field f : declaredFields) {
+            if (f.isAnnotationPresent(FieId.class)) {
+                // 获取注解
+                FieId declaredAnnotation = f.getDeclaredAnnotation(FieId.class);
+
+                // 如果嵌套对象:
+                /**
+                 * {
+                 *   "mappings": {
+                 *     "properties": {
+                 *       "region": {
+                 *         "type": "keyword"
+                 *       },
+                 *       "manager": {
+                 *         "properties": {
+                 *           "age":  { "type": "integer" },
+                 *           "name": {
+                 *             "properties": {
+                 *               "first": { "type": "text" },
+                 *               "last":  { "type": "text" }
+                 *             }
+                 *           }
+                 *         }
+                 *       }
+                 *     }
+                 *   }
+                 * }
+                 */
+                if (declaredAnnotation.type() == FieldType.OBJECT) {
+                    // 获取当前类的对象-- Action
+                    Class<?> type = f.getType();
+                    Field[] df2 = type.getDeclaredFields();
+                    builder.startObject(f.getName());
+                    builder.startObject("properties");
+                    // 遍历该对象中的所有属性
+                    for (Field f2 : df2) {
+                        if (f2.isAnnotationPresent(FieId.class)) {
+                            // 获取注解
+                            FieId declaredAnnotation2 = f2.getDeclaredAnnotation(FieId.class);
+                            builder.startObject(f2.getName());
+                            builder.field("type", declaredAnnotation2.type().getType());
+                            // keyword不需要分词
+                            if (declaredAnnotation2.type() == FieldType.TEXT) {
+                                builder.field("analyzer", declaredAnnotation2.analyzer().getType());
+                            }
+                            if (declaredAnnotation2.type() == FieldType.DATE) {
+                                builder.field("format", "yyyy-MM-dd HH:mm:ss");
+                            }
+                            builder.endObject();
+                        }
+                    }
+                    builder.endObject();
+                    builder.endObject();
+
+                } else {
+                    builder.startObject(f.getName());
+                    builder.field("type", declaredAnnotation.type().getType());
+                    // keyword不需要分词
+                    if (declaredAnnotation.type() == FieldType.TEXT) {
+                        builder.field("analyzer", declaredAnnotation.analyzer().getType());
+                    }
+                    if (declaredAnnotation.type() == FieldType.DATE) {
+                        builder.field("format", "yyyy-MM-dd HH:mm:ss");
+                    }
+                    builder.endObject();
+                }
+            }
+        }
+        // 对应property
+        builder.endObject();
+        builder.endObject();
+        return builder;
+    }
+
+
+    public static Field getFieldByAnnotation(Object o, Class annotationClass) {
+        Field[] declaredFields = o.getClass().getDeclaredFields();
+        if (declaredFields != null && declaredFields.length > 0) {
+            for (Field f : declaredFields) {
+                if (f.isAnnotationPresent(annotationClass)) {
+                    return f;
+                }
+            }
+        }
+        return null;
+    }
+
+}
+

+ 37 - 0
module-job-bytedance/src/main/java/cn/com/ctop/job/bytedance/handler/SynchronousAccountTransactionDetailsJob.java

@@ -0,0 +1,37 @@
+package cn.com.ctop.job.bytedance.handler;
+
+import cn.com.ctop.toutiao.modules.agent.service.AgentManagementService;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+@Component
+@Slf4j
+public class SynchronousAccountTransactionDetailsJob {
+
+    @Resource
+    AgentManagementService agentManagementService;
+
+    /**
+     * 同步账号交易流水
+     *
+     * @param
+     * @return
+     * @throws
+     * @author ZZY
+     */
+    @XxlJob("synchronousAccountTransactionDetails")
+    public void execute() throws Exception {
+        try {
+            //同步账号信息
+            agentManagementService.synchronousAccount();
+            //同步每个账号交易流水
+            agentManagementService.synchronousAccountTransactionDetails();
+        } catch (Exception e) {
+            log.info("同步完成");
+        }
+
+    }
+}

+ 4 - 2
module-kuaishou/src/main/java/cn/com/ctop/kuaishou/modules/clean/mapper/xml/KuaishouAccountCleanTemplateMapper.xml

@@ -45,7 +45,7 @@
       </if >
 
       <if test="createUser!=null and createUser!=''">
-        and t.createUser=#{createUser}
+        and t.create_user=#{createUser}
       </if >
       <if test="createId!=null and createId!=''">
         and t.create_id=#{createId}
@@ -105,8 +105,9 @@
     LEFT JOIN ctop_kuaishou_account_clean_center t2 on t.id=t2.template_id
     LEFT JOIN ctop_user_allocation t3 on t2.account_id=t3.account_id
     <where>
+        t2.account_id is not null
       <if test="accountId!=null and accountId!=''">
-        t2.account_id=#{accountId}
+        and t2.account_id=#{accountId}
       </if >
 
       <if test="userName!=null and userName!=''">
@@ -116,6 +117,7 @@
         and t.create_id=#{createId}
       </if >
     </where>
+
   </select>
 
   <select id="getTemplateListBystatus" resultMap="templateResultMap">

+ 30 - 0
module-kuaishou/src/main/java/cn/com/ctop/kuaishou/modules/label/controller/LbelController.java

@@ -0,0 +1,30 @@
+package cn.com.ctop.kuaishou.modules.label.controller;
+
+import cn.com.ctop.kuaishou.modules.clean.entity.KuaiShouAccountCleanTemplateInfo;
+import org.jeecg.common.api.vo.Result;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @Description: 标签管理
+ * @Author: zzy
+ * @Date: 2021-07-30
+ * @Version: V1.0
+ */
+@RestController
+@RequestMapping("/label")
+public class LbelController {
+
+    /**
+     * 查询标签
+     *
+     * @param kuaiShouAccountCleanTemplateInfo
+     * @return
+     */
+    @PostMapping("/createTemplate")
+    public Result createCleanTemplate(@RequestBody KuaiShouAccountCleanTemplateInfo kuaiShouAccountCleanTemplateInfo){
+        return   null;
+    }
+}

+ 25 - 0
module-kuaishou/src/main/java/cn/com/ctop/kuaishou/modules/label/mapper/MaterialLabelMapper.java

@@ -0,0 +1,25 @@
+package cn.com.ctop.kuaishou.modules.label.mapper;
+
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.List;
+
+/**
+ * @Description: 素材标签
+ * @Author: zzy
+ * @Date: 2021-07-30
+ * @Version: V1.0
+ */
+@Mapper
+public interface MaterialLabelMapper {
+    /**
+     * 查询所有素材
+     */
+    List<String> getMateriaSignature();
+
+    /**
+     * 获取素材产生消耗的最早日期
+     */
+    String getMaterialFirstDate(@Param("signature")String signature);
+}

+ 25 - 0
module-kuaishou/src/main/java/cn/com/ctop/kuaishou/modules/label/mapper/xml/MaterialTagMapper.xml

@@ -0,0 +1,25 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="cn.com.ctop.kuaishou.modules.label.mapper.MaterialLabelMapper">
+
+
+  <resultMap id="logResultMap" type="cn.com.ctop.kuaishou.modules.clean.entity.KuaiShouAccountCleanLogInfo">
+    <id column="id" property="id"/>
+    <result column="account_id" property="accountId"/>
+    <result column="user_name" property="userName"/>
+    <result column="operation_type" property="operationType"/>
+    <result column="create_time" property="createTime"/>
+    <result column="executive_information" property="executiveInformation"/>
+  </resultMap>
+
+    <select id="getMateriaSignature" resultType="String">
+        SELECT  signature from ctop_etl_kuaishou_video_base_info t  GROUP BY t.signature
+    </select>
+
+
+    <select id="getMaterialFirstDate" parameterType="String" resultType="String">
+        SELECT  MIN(stat_date) from ctop_kuaishou_report_daily_material t where t.charge>0 and  t.signature=#{signature}
+    </select>
+
+
+</mapper>

+ 14 - 0
module-kuaishou/src/main/java/cn/com/ctop/kuaishou/modules/label/service/MaterialTagService.java

@@ -0,0 +1,14 @@
+package cn.com.ctop.kuaishou.modules.label.service;
+
+import org.jeecg.common.api.vo.Result;
+
+/**
+ * @Description: 素材标签
+ * @Author: zzy
+ * @Date: 2021-07-30
+ * @Version: V1.0
+ */
+public interface MaterialTagService {
+
+    public Result filterMaterial ();
+}

+ 80 - 0
module-kuaishou/src/main/java/cn/com/ctop/kuaishou/modules/label/service/impl/MaterialTagServiceImpl.java

@@ -0,0 +1,80 @@
+package cn.com.ctop.kuaishou.modules.label.service.impl;
+
+import cn.com.ctop.kuaishou.modules.label.mapper.MaterialLabelMapper;
+import cn.com.ctop.kuaishou.modules.label.service.MaterialTagService;
+import lombok.extern.slf4j.Slf4j;
+import org.jeecg.common.api.vo.Result;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.text.SimpleDateFormat;
+import java.util.Calendar;
+import java.util.List;
+
+/**
+ * @Description: 素材标签
+ * @Author: zzy
+ * @Date: 2021-07-30
+ * @Version: V1.0
+ */
+@Service
+@Slf4j
+public class MaterialTagServiceImpl implements MaterialTagService {
+
+
+    @Resource
+    MaterialLabelMapper materialLabelMapper;
+
+
+    /**
+     * 清洗各素材指标入库
+     *
+     * @param
+     * @return
+     */
+    @Override
+    public Result filterMaterial() {
+
+        try {
+            //查询所有素材标签
+            List<String> materiaSignature = materialLabelMapper.getMateriaSignature();
+            if(!materiaSignature.isEmpty()){
+                materiaSignature.stream().forEach(str->{
+                    if(!str.isEmpty()){
+                        //获取素材产生消耗的最早日期
+                        String materialFirstDate = materialLabelMapper.getMaterialFirstDate(str);
+                        //获取30天之后的日期
+                        String afterDate = getDateByNumber(materialFirstDate, -30);
+                        //查询素材各项指标入库
+                        
+                    }
+
+                });
+            }
+        }catch (Exception e){
+
+        }
+
+
+        return null;
+    }
+    /**
+     * 获取指定日期number天之后之前的日期
+     *
+     * @param
+     * @return
+     */
+    public String getDateByNumber(String date,int number){
+        Calendar calc = Calendar.getInstance();
+        SimpleDateFormat sdf  = new SimpleDateFormat("yyyy-MM-dd");
+        String AfterDate=null;
+        try {
+            calc.setTime(sdf.parse(date));
+            calc.add(calc.DATE, number);
+            AfterDate=sdf.format(calc.getTime());
+        }catch (Exception e){
+            log.info("获取日期失败:{}",e.toString());
+        }
+        return AfterDate;
+    }
+}

+ 30 - 0
module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/agent/controller/AgentManagementController.java

@@ -0,0 +1,30 @@
+package cn.com.ctop.toutiao.modules.agent.controller;
+
+import cn.com.ctop.toutiao.modules.agent.service.AgentManagementService;
+import com.alibaba.fastjson.JSONObject;
+import lombok.extern.slf4j.Slf4j;
+import org.jeecg.common.api.vo.Result;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.ArrayList;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+@RestController
+@RequestMapping("/agent")
+@Slf4j
+public class AgentManagementController {
+
+    @Autowired
+    AgentManagementService agentManagementService;
+
+    @RequestMapping("/queryAccountTransactionDetails")
+    public void  getList(){
+        Result recharge = agentManagementService.synchronousAccountTransactionDetails();
+        log.info("123");
+
+    }
+}

+ 73 - 0
module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/agent/entity/TransactionDetails.java

@@ -0,0 +1,73 @@
+package cn.com.ctop.toutiao.modules.agent.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+
+import java.math.BigDecimal;
+
+@Data
+@TableName("ctop_account_transaction_details")
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+@ApiModel(value = "ctop_account_transaction_details对象", description = "账户交易流水")
+public class TransactionDetails {
+    @TableId(type = IdType.AUTO)
+    @ApiModelProperty(value = "id")
+    private String id;
+
+    //广告主ID
+    @ApiModelProperty(value = "advertiser_id")
+    private String advertiser_id;
+
+    //流水类型  RECHARGE:充值      TRANSFER:转账
+    @ApiModelProperty(value = "transaction_type")
+    private String transaction_type;
+
+    //流水产生时间
+    @ApiModelProperty(value = "data_time")
+    private String data_time;
+
+    //交易总金额(单位元)
+    @ApiModelProperty(value = "amount")
+    private BigDecimal amount;
+
+    //现金总金额(单位元)
+    @ApiModelProperty(value = "cash")
+    private BigDecimal cash;
+
+    //赠款总金额(单位元)
+    @ApiModelProperty(value = "grant")
+    private BigDecimal grant;
+
+    //返货总金额(单位元)
+    @ApiModelProperty(value = "return_goods")
+    private BigDecimal return_goods;
+
+    //交易流水号
+    @ApiModelProperty(value = "transaction_seq")
+    private String transaction_seq;
+
+    //付款方,即广告主id
+    @ApiModelProperty(value = "remitter")
+    private String remitter;
+
+    //收款方,即广告主id。
+    @ApiModelProperty(value = "payee")
+    private String payee;
+
+    //返点
+    @ApiModelProperty(value = "dealbase")
+    private String dealbase;
+
+    //数据创建时间
+    @ApiModelProperty(value = "create_time")
+    private String create_time;
+
+
+}

+ 15 - 0
module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/agent/mapper/CtopAccountTransactionDetailsMapper.java

@@ -0,0 +1,15 @@
+package cn.com.ctop.toutiao.modules.agent.mapper;
+
+import cn.com.ctop.toutiao.modules.agent.entity.TransactionDetails;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.List;
+
+@Mapper
+public interface CtopAccountTransactionDetailsMapper {
+    //保存交易记录
+    public void saveAccountTransactionDetails(@Param("list") List<TransactionDetails> list, @Param("date")String date);
+    //根据日期查询当日所有交易订单号
+    public List<String> getTransactionSeq( @Param("date")String date);
+}

+ 16 - 0
module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/agent/mapper/CtopAgentAccountMapper.java

@@ -0,0 +1,16 @@
+package cn.com.ctop.toutiao.modules.agent.mapper;
+
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.List;
+
+@Mapper
+public interface CtopAgentAccountMapper {
+    //获取所有账户
+    public List<String> getAllAccountId();
+    //根据账户id修改账户状态
+    public void updateAccountState(@Param("list")List<String> list,@Param("state")int state);
+    //批量插入新增数据
+    public void insertAccountData(@Param("list") List<String> list,@Param("agent_id")String agent_id,@Param("create_time")String create_time);
+}

+ 30 - 0
module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/agent/mapper/xml/CtopAccountTransactionDetailsMapper.xml

@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="cn.com.ctop.toutiao.modules.agent.mapper.CtopAccountTransactionDetailsMapper">
+   <!-- 批量插入新增数据-->
+    <insert id="saveAccountTransactionDetails">
+        insert  into ctop_account_transaction_details (advertiser_id,transaction_type,data_time,amount,cash,grant_amount,return_goods,transaction_seq,
+        remitter,payee,dealbase,create_time) values
+        <foreach collection="list" item="item" index="index" separator=",">
+            (
+            #{item.advertiser_id},
+            #{item.transaction_type},
+            #{item.create_time},
+            #{item.amount},
+            #{item.cash},
+            #{item.grant},
+            #{item.return_goods},
+            #{item.transaction_seq},
+            #{item.remitter},
+            #{item.payee},
+            #{item.dealbase},
+            #{date}
+            )
+        </foreach>
+    </insert>
+
+    <select id="getTransactionSeq" parameterType="String" resultType="String">
+        select t.transaction_seq from ctop_account_transaction_details t where t.create_time=#{date}
+    </select>
+
+</mapper>

+ 28 - 0
module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/agent/mapper/xml/CtopAgentAccountMapper.xml

@@ -0,0 +1,28 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="cn.com.ctop.toutiao.modules.agent.mapper.CtopAgentAccountMapper">
+    <!--获取所有账户-->
+    <select id="getAllAccountId" resultType="String">
+        select account_id from ctop_agent_account where state=0
+    </select>
+   <!-- 根据账户id修改账户状态-->
+    <update id="updateAccountState">
+        update ctop_agent_account set state=#{state} where account_id in
+        <foreach item="item" index="index" collection="list" open="("  close=")" separator=",">
+            #{item}
+        </foreach>
+    </update>
+   <!-- 批量插入新增数据-->
+    <insert id="insertAccountData">
+        insert  into ctop_agent_account (account_id,agent_id,state,create_time) values
+        <foreach collection="list" item="item" index="index" separator=",">
+            (
+            #{item},
+            #{agent_id},
+            0,
+            #{create_time}
+            )
+        </foreach>
+    </insert>
+
+</mapper>

+ 22 - 0
module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/agent/service/AgentManagementService.java

@@ -0,0 +1,22 @@
+package cn.com.ctop.toutiao.modules.agent.service;
+
+import com.alibaba.fastjson.JSONObject;
+import org.jeecg.common.api.vo.Result;
+
+public interface AgentManagementService {
+    //获取TOKEN
+    public String getToken();
+
+    //查询代理商下账户
+    public  JSONObject selectAgentAdvertiser(int pageNum,int pageSize);
+
+
+    //查询账号流水
+    public JSONObject queryAccountTransactionDetails(String AccountId,String transaction_type, int pageNum, int pageSize);
+
+    //同步代理商账户入库
+    public Result synchronousAccount();
+
+    //同步账户交易流水
+    public Result synchronousAccountTransactionDetails();
+}

+ 340 - 0
module-toutiao/src/main/java/cn/com/ctop/toutiao/modules/agent/service/impl/AgentManagementServiceImpl.java

@@ -0,0 +1,340 @@
+package cn.com.ctop.toutiao.modules.agent.service.impl;
+
+import cn.com.ctop.common.module.entity.CtopOauthToken;
+import cn.com.ctop.common.module.mapper.CtopOauthTokenMapper;
+import cn.com.ctop.toutiao.modules.agent.entity.TransactionDetails;
+import cn.com.ctop.toutiao.modules.agent.mapper.CtopAccountTransactionDetailsMapper;
+import cn.com.ctop.toutiao.modules.agent.mapper.CtopAgentAccountMapper;
+import cn.com.ctop.toutiao.modules.agent.service.AgentManagementService;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.http.client.ClientProtocolException;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpEntityEnclosingRequestBase;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.jeecg.common.api.vo.Result;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.Resource;
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.URI;
+import java.text.SimpleDateFormat;
+import java.util.*;
+
+import static java.util.stream.Collectors.toList;
+
+/**
+ * 代理商管理
+ *
+ * @author zzy
+ * @create: 2021-07-14
+ */
+@Service
+@Slf4j
+public class AgentManagementServiceImpl implements AgentManagementService {
+
+    private final String OPEN_API_URL_PREFIX = "https://ad.oceanengine.com/open_api/2/";
+    private final String ACCOUNT_LIST_URL = "agent/advertiser/select/";
+    private final String ACCOUNT_TRANSACTION_DETAILS_URL = "advertiser/fund/transaction/get/";
+    private final String ADVERTISER_ID = "73970348172";
+    private final String RECHARGE = "RECHARGE";
+    private final String TRANSFER = "TRANSFER";
+    @Resource
+    CtopOauthTokenMapper ctopOauthTokenMapper;
+
+    @Resource
+    CtopAgentAccountMapper ctopAgentAccountMapper;
+    @Resource
+    CtopAccountTransactionDetailsMapper ctopAccountTransactionDetailsMapper;
+
+    /**
+     * 同步代理商下账号
+     *
+     * @author zzy
+     * @create: 2021-07-14
+     */
+
+    @Override
+    public Result synchronousAccount() {
+        log.info("开始同步账号信息");
+        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd");//设置日期格式
+        List<String> dataList = new ArrayList();
+        Result result = new Result();
+        try {
+            //分页查询第一页数据
+            JSONObject agentAccountList = selectAgentAdvertiser(1, 2000);
+            JSONObject data = (JSONObject) agentAccountList.get("data");
+            //dataList.addAll((List) data.get("list"));
+            if (data.size() > 0) {
+                JSONObject page_info = (JSONObject) data.get("page_info");
+                int total_page = (int) page_info.get("total_page");
+                //查询其他页数据
+                for (int i = 0; i < total_page; i++) {
+                    JSONObject agenttList = selectAgentAdvertiser(i + 1, 2000);
+                    JSONObject agentData = (JSONObject) agenttList.get("data");
+                    List<Long> list = (List<Long>) agentData.get("list");
+                    list.stream().forEach(str -> {
+                        dataList.add(str.toString());
+                    });
+
+                }
+                //查询数据库已存在id
+                List<String> accountIdList = ctopAgentAccountMapper.getAllAccountId();
+                if (!accountIdList.isEmpty()) {
+                    //筛选巨量引擎剔除的用户
+                    List<String> removeList = accountIdList.stream().filter(item -> !dataList.contains(item)).collect(toList());
+                    //筛选巨量引擎新增的用户
+                    List<String> addList = dataList.stream().filter(item -> !accountIdList.contains(item)).collect(toList());
+                    //修改剔除的用户状态
+                    if (!removeList.isEmpty()) {
+                        ctopAgentAccountMapper.updateAccountState(removeList, 1);
+                    }
+                    //保存新增的用户
+                    if (!addList.isEmpty()) {
+                        ctopAgentAccountMapper.insertAccountData(addList, ADVERTISER_ID, df.format(new Date()));
+                    }
+                } else {
+                    ctopAgentAccountMapper.insertAccountData(dataList, ADVERTISER_ID, df.format(new Date()));
+                }
+
+            }
+            result.setSuccess(true);
+            result.setMessage("success");
+            return result;
+        } catch (Exception e) {
+            log.info("同步失败:{}", e.toString());
+            result.error500("同步失败");
+            return result;
+        }
+    }
+
+    /**
+     * 同步账户交易流水
+     *
+     * @author zzy
+     * @create: 2021-07-15
+     */
+    @Override
+    public Result synchronousAccountTransactionDetails() {
+        log.info("开始同步账号交易流水");
+        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd");//设置日期格式
+        List<TransactionDetails> transactionDetailsList = new ArrayList<>();
+        try {
+            List<String> accountIdList = ctopAgentAccountMapper.getAllAccountId();
+            if (!accountIdList.isEmpty()) {
+                List<String> transactionSeqList = ctopAccountTransactionDetailsMapper.getTransactionSeq(df.format(new Date()));
+                accountIdList.stream().forEach(str -> {
+                    if (!str.isEmpty()) {
+                        //查询账号下充值的交易流水
+                        JSONObject jsonObject = queryAccountTransactionDetails(str, RECHARGE, 1, 1000);
+                        JSONObject data = (JSONObject) jsonObject.get("data");
+                        List<Object> dataList = (List<Object>) data.get("list");
+                        if (!dataList.isEmpty()) {
+                            //充值类存在记录
+                            //保存第一页数据
+                            dataList.stream().forEach(ob -> {
+                                TransactionDetails transactionDetails = JSONObject.toJavaObject((JSON) ob, TransactionDetails.class);
+                                //订单号不存在
+                                if (transactionSeqList.isEmpty() || !transactionSeqList.contains(transactionDetails.getTransaction_seq())) {
+                                    transactionDetailsList.add(transactionDetails);
+                                }
+
+                            });
+                            JSONObject page_info = (JSONObject) data.get("page_info");
+                            int total_page = (int) page_info.get("total_page");
+                            //查询其他页数据
+                            if (total_page > 1) {
+                                for (int i = 2; i <= total_page; i++) {
+                                    JSONObject jo = queryAccountTransactionDetails(str, RECHARGE, i, 1000);
+                                    JSONObject accountData = (JSONObject) jo.get("data");
+                                    List<Object> rechargeList = (List<Object>) accountData.get("list");
+                                    rechargeList.stream().forEach(transactionDetails -> {
+                                        TransactionDetails transactionDetailsOther = JSONObject.toJavaObject((JSON) transactionDetails, TransactionDetails.class);
+                                        //订单号不存在
+                                        if (transactionSeqList.isEmpty() || !transactionSeqList.contains(transactionDetailsOther.getTransaction_seq())) {
+                                            transactionDetailsList.add(transactionDetailsOther);
+                                        }
+                                    });
+                                }
+                            }
+                        }
+                        //查询账号下转账的交易流水
+                        JSONObject transferJsonObject = queryAccountTransactionDetails(str, TRANSFER, 1, 1000);
+                        JSONObject transferData = (JSONObject) transferJsonObject.get("data");
+                        List<Object> transferDataList = (List<Object>) transferData.get("list");
+                        if (!transferDataList.isEmpty()) {
+                            //转账类存在记录
+                            //保存第一页数据
+                            transferDataList.stream().forEach(ob -> {
+                                TransactionDetails transactionDetails = JSONObject.toJavaObject((JSON) ob, TransactionDetails.class);
+                                //订单号不存在
+                                if (transactionSeqList.isEmpty() || !transactionSeqList.contains(transactionDetails.getTransaction_seq())) {
+                                    transactionDetailsList.add(transactionDetails);
+                                }
+                            });
+                            JSONObject transferPageInfo = (JSONObject) transferData.get("page_info");
+                            int transferTotalPage = (int) transferPageInfo.get("total_page");
+                            //查询其他页数据
+                            if (transferTotalPage > 1) {
+                                for (int i = 2; i <= transferTotalPage; i++) {
+                                    JSONObject jo = queryAccountTransactionDetails(str, TRANSFER, i, 1000);
+                                    JSONObject accountData = (JSONObject) jo.get("data");
+                                    List<Object> rechargeList = (List<Object>) accountData.get("list");
+                                    rechargeList.stream().forEach(transactionDetails -> {
+                                        TransactionDetails transactionDetailsOther = JSONObject.toJavaObject((JSON) transactionDetails, TransactionDetails.class);
+                                        //订单号不存在
+                                        if (transactionSeqList.isEmpty() || !transactionSeqList.contains(transactionDetailsOther.getTransaction_seq())) {
+                                            transactionDetailsList.add(transactionDetailsOther);
+                                        }
+                                    });
+                                }
+                            }
+                        }
+                    }
+                });
+            }
+            //保存数据
+            if (!transactionDetailsList.isEmpty()) {
+                ctopAccountTransactionDetailsMapper.saveAccountTransactionDetails(transactionDetailsList, df.format(new Date()));
+                log.info("新增交易记录:{}条", transactionDetailsList.size());
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+            log.info("同步失败:{}", e.toString());
+        }
+
+
+        return null;
+    }
+
+    /**
+     * 查询账号流水
+     *
+     * @return
+     * @author zzy
+     * @Param AccountId 账户id    transaction_type 流水类型  RECHARGE:充值  TRANSFER:转账     pageNum 当前页    pageSize 每页显示数量
+     * @create: 2021-07-14
+     */
+    @Override
+    public JSONObject queryAccountTransactionDetails(String AccountId, String transaction_type, int pageNum, int pageSize) {
+        SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd");//设置日期格式
+        Calendar ca = Calendar.getInstance();
+        ca.setTime(new Date());
+        ca.add(Calendar.DATE, -1);
+        // 请求参数
+        Map data = new HashMap() {
+            {
+                put("advertiser_id", AccountId);
+                put("start_date", df.format(ca.getTime()));
+                put("end_date", df.format(ca.getTime()));
+                put("transaction_type", transaction_type);
+                put("page", pageNum);
+                put("page_size", pageSize);
+            }
+        };
+        return sendHttpRequest(data, ACCOUNT_TRANSACTION_DETAILS_URL);
+    }
+
+
+    /**
+     * 获取token
+     *
+     * @author zzy
+     * @create: 2021-07-14
+     */
+    @Override
+    public String getToken() {
+        CtopOauthToken ctopOauthToken = ctopOauthTokenMapper.selectByAccountId(Long.parseLong(ADVERTISER_ID));
+        if (ctopOauthToken != null) {
+            return ctopOauthToken.getAccessToken();
+        } else {
+            log.info("未查询到Token");
+            return null;
+        }
+
+    }
+
+    /**
+     * 获取账户列表
+     *
+     * @author zzy
+     * @Param pageNum 当前页    pageSize 每页显示数量
+     * @create: 2021-07-14
+     */
+    @Override
+    public JSONObject selectAgentAdvertiser(int pageNum, int pageSize) {
+        // 请求参数
+        Map data = new HashMap() {
+            {
+                put("advertiser_id", ADVERTISER_ID);
+                put("page", pageNum);
+                put("page_size", pageSize);
+            }
+        };
+        return sendHttpRequest(data, ACCOUNT_LIST_URL);
+    }
+
+
+    /**
+     * 根据地址请求巨量殷勤,返回响应结果
+     *
+     * @author zzy
+     * @Param data 请求参数    apiUrl请求路径
+     * @create: 2021-07-14
+     */
+    public JSONObject sendHttpRequest(Map data, String apiUrl) {
+        String access_token = getToken();
+        // 构造请求
+        HttpEntityEnclosingRequestBase httpEntity = new HttpEntityEnclosingRequestBase() {
+            @Override
+            public String getMethod() {
+                return "GET";
+            }
+        };
+
+        httpEntity.setHeader("Access-Token", access_token);
+
+        CloseableHttpResponse response = null;
+        CloseableHttpClient client = null;
+
+        try {
+            client = HttpClientBuilder.create().build();
+            httpEntity.setURI(URI.create(OPEN_API_URL_PREFIX + apiUrl));
+            httpEntity.setEntity(new StringEntity(JSONObject.toJSONString(data), ContentType.APPLICATION_JSON));
+
+            response = client.execute(httpEntity);
+            if (response != null && response.getStatusLine().getStatusCode() == 200) {
+                BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(response.getEntity().getContent()));
+                StringBuffer result = new StringBuffer();
+                String line = "";
+                while ((line = bufferedReader.readLine()) != null) {
+                    result.append(line);
+                }
+                bufferedReader.close();
+                return JSONObject.parseObject(result.toString());
+            }
+
+        } catch (ClientProtocolException e) {
+            e.printStackTrace();
+        } catch (IOException e) {
+            e.printStackTrace();
+        } finally {
+            try {
+                if (response != null) {
+                    response.close();
+                }
+                client.close();
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+        return null;
+    }
+
+}