feat(推送微拌的数据)

src/main/java/com/mokamrp/privates/controller/CorpDataController.java

@@ -0,0 +1,37 @@
+package com.mokamrp.privates.controller;
+
+import com.mokamrp.privates.controller.BaseController;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.RequestMapping;
+import com.mokamrp.privates.service.CorpDataService;
+import com.mokamrp.privates.mapper.pojo.CorpData;
+
+import org.springframework.web.bind.annotation.RestController;
+
+import javax.validation.Valid;
+import java.util.List;
+
+/**
+ * <p>
+ *  前端控制器
+ * </p>
+ *
+ * @author gy
+ * @since 2021-07-19
+ */
+@RestController
+@RequestMapping("/corpData")
+public class CorpDataController extends BaseController<CorpData> {
+
+    @Autowired
+    public CorpDataService corpDataService;
+
+    /*
+     * @fast
+     * 默认生成的控制器所继承的父类中有index/add/edit/del四个基础方法,均为post json请求
+     * 如果需要定制化，需要将BaseController中对应的方法复制到当前控制器,然后进行修改
+     */
+
+
+}
+

src/main/java/com/mokamrp/privates/mapper/CorpDataMapper.java

@@ -0,0 +1,16 @@
+package com.mokamrp.privates.mapper;
+
+import com.mokamrp.privates.mapper.pojo.CorpData;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+
+/**
+ * <p>
+ *  Mapper 接口
+ * </p>
+ *
+ * @author gy
+ * @since 2021-07-19
+ */
+public interface CorpDataMapper extends BaseMapper<CorpData> {
+
+}

src/main/java/com/mokamrp/privates/mapper/CorpDataMapper.xml

@@ -0,0 +1,26 @@
+<?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="com.mokamrp.privates.mapper.CorpDataMapper">
+
+    <!-- 通用查询映射结果 -->
+    <resultMap id="BaseResultMap" type="com.mokamrp.privates.mapper.pojo.CorpData">
+        <id column="primary_key" property="primaryKey" />
+        <result column="currentdateymd" property="currentdateymd" />
+        <result column="corp_wx_id" property="corpWxId" />
+        <result column="name" property="name" />
+        <result column="increase_user_count" property="increaseUserCount" />
+        <result column="decrease_user_count" property="decreaseUserCount" />
+        <result column="total_user_count" property="totalUserCount" />
+        <result column="group_decrease_user_count" property="groupDecreaseUserCount" />
+        <result column="group_increase_user_count" property="groupIncreaseUserCount" />
+        <result column="group_total_group_count" property="groupTotalGroupCount" />
+        <result column="group_total_user_count" property="groupTotalUserCount" />
+        <result column="updatetime" property="updatetime" />
+    </resultMap>
+
+    <!-- 通用查询结果列 -->
+    <sql id="Base_Column_List">
+        primary_key, currentdateymd, corp_wx_id, name, increase_user_count, decrease_user_count, total_user_count, group_decrease_user_count, group_increase_user_count, group_total_group_count, group_total_user_count, updatetime
+    </sql>
+
+</mapper>

src/main/java/com/mokamrp/privates/mapper/pojo/CorpData.java

@@ -0,0 +1,205 @@
+package com.mokamrp.privates.mapper.pojo;
+
+import com.baomidou.mybatisplus.annotation.TableName;
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.extension.activerecord.Model;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import com.baomidou.mybatisplus.annotation.TableId;
+import java.io.Serializable;
+import java.util.Date;
+
+/**
+ * <p>
+ * 
+ * </p>
+ *
+ * @author gy
+ * @since 2021-07-19
+ */
+@TableName("mc_corp_data")
+public class CorpData extends Model<CorpData> {
+
+    private static final long serialVersionUID = 1L;
+
+    /**
+     * 主键
+     */
+      @TableId(value = "primary_key", type = IdType.AUTO)
+    private String primaryKey;
+
+    /**
+     * 数据日期
+     */
+    private Date currentdateymd;
+
+    /**
+     * 主体appid
+     */
+    private String corpWxId;
+
+    /**
+     * 主体名称
+     */
+    private String name;
+
+    /**
+     * 今日新增
+     */
+    private Integer increaseUserCount;
+
+    /**
+     * 今日流失
+     */
+    private Integer decreaseUserCount;
+
+    /**
+     * 客户总数
+     */
+    private Integer totalUserCount;
+
+    /**
+     * 今日 退群
+     */
+    private Integer groupDecreaseUserCount;
+
+    /**
+     * 今日入群
+     */
+    private Integer groupIncreaseUserCount;
+
+    /**
+     * 群总数
+     */
+    private Integer groupTotalGroupCount;
+
+    /**
+     * 今日群用户数量
+     */
+    private Integer groupTotalUserCount;
+
+    /**
+     * 更新时间
+     */
+    private LocalDateTime updatetime;
+
+
+    public String getPrimaryKey() {
+        return primaryKey;
+    }
+
+    public void setPrimaryKey(String primaryKey) {
+        this.primaryKey = primaryKey;
+    }
+
+    public Date getCurrentdateymd() {
+        return currentdateymd;
+    }
+
+    public void setCurrentdateymd(Date currentdateymd) {
+        this.currentdateymd = currentdateymd;
+    }
+
+    public String getCorpWxId() {
+        return corpWxId;
+    }
+
+    public void setCorpWxId(String corpWxId) {
+        this.corpWxId = corpWxId;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+
+    public Integer getIncreaseUserCount() {
+        return increaseUserCount;
+    }
+
+    public void setIncreaseUserCount(Integer increaseUserCount) {
+        this.increaseUserCount = increaseUserCount;
+    }
+
+    public Integer getDecreaseUserCount() {
+        return decreaseUserCount;
+    }
+
+    public void setDecreaseUserCount(Integer decreaseUserCount) {
+        this.decreaseUserCount = decreaseUserCount;
+    }
+
+    public Integer getTotalUserCount() {
+        return totalUserCount;
+    }
+
+    public void setTotalUserCount(Integer totalUserCount) {
+        this.totalUserCount = totalUserCount;
+    }
+
+    public Integer getGroupDecreaseUserCount() {
+        return groupDecreaseUserCount;
+    }
+
+    public void setGroupDecreaseUserCount(Integer groupDecreaseUserCount) {
+        this.groupDecreaseUserCount = groupDecreaseUserCount;
+    }
+
+    public Integer getGroupIncreaseUserCount() {
+        return groupIncreaseUserCount;
+    }
+
+    public void setGroupIncreaseUserCount(Integer groupIncreaseUserCount) {
+        this.groupIncreaseUserCount = groupIncreaseUserCount;
+    }
+
+    public Integer getGroupTotalGroupCount() {
+        return groupTotalGroupCount;
+    }
+
+    public void setGroupTotalGroupCount(Integer groupTotalGroupCount) {
+        this.groupTotalGroupCount = groupTotalGroupCount;
+    }
+
+    public Integer getGroupTotalUserCount() {
+        return groupTotalUserCount;
+    }
+
+    public void setGroupTotalUserCount(Integer groupTotalUserCount) {
+        this.groupTotalUserCount = groupTotalUserCount;
+    }
+
+    public LocalDateTime getUpdatetime() {
+        return updatetime;
+    }
+
+    public void setUpdatetime(LocalDateTime updatetime) {
+        this.updatetime = updatetime;
+    }
+
+    @Override
+    protected Serializable pkVal() {
+        return this.primaryKey;
+    }
+
+    @Override
+    public String toString() {
+        return "CorpData{" +
+        "primaryKey=" + primaryKey +
+        ", currentdateymd=" + currentdateymd +
+        ", corpWxId=" + corpWxId +
+        ", name=" + name +
+        ", increaseUserCount=" + increaseUserCount +
+        ", decreaseUserCount=" + decreaseUserCount +
+        ", totalUserCount=" + totalUserCount +
+        ", groupDecreaseUserCount=" + groupDecreaseUserCount +
+        ", groupIncreaseUserCount=" + groupIncreaseUserCount +
+        ", groupTotalGroupCount=" + groupTotalGroupCount +
+        ", groupTotalUserCount=" + groupTotalUserCount +
+        ", updatetime=" + updatetime +
+        "}";
+    }
+}

src/main/java/com/mokamrp/privates/service/CorpDataService.java

@@ -0,0 +1,16 @@
+package com.mokamrp.privates.service;
+
+import com.mokamrp.privates.mapper.pojo.CorpData;
+import com.baomidou.mybatisplus.extension.service.IService;
+
+/**
+ * <p>
+ *  服务类
+ * </p>
+ *
+ * @author gy
+ * @since 2021-07-19
+ */
+public interface CorpDataService extends IService<CorpData> {
+
+}

src/main/java/com/mokamrp/privates/service/impl/CorpDataServiceImpl.java

@@ -0,0 +1,20 @@
+package com.mokamrp.privates.service.impl;
+
+import com.mokamrp.privates.mapper.pojo.CorpData;
+import com.mokamrp.privates.mapper.CorpDataMapper;
+import com.mokamrp.privates.service.CorpDataService;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import org.springframework.stereotype.Service;
+
+/**
+ * <p>
+ *  服务实现类
+ * </p>
+ *
+ * @author gy
+ * @since 2021-07-19
+ */
+@Service
+public class CorpDataServiceImpl extends ServiceImpl<CorpDataMapper, CorpData> implements CorpDataService {
+
+}

src/main/java/com/mokamrp/privates/tast/Datahub/DatahubCorpData.java

@@ -0,0 +1,159 @@
+package com.mokamrp.privates.tast.Datahub;
+
+import com.aliyun.datahub.client.DatahubClient;
+import com.aliyun.datahub.client.exception.DatahubClientException;
+import com.aliyun.datahub.client.model.PutRecordsResult;
+import com.aliyun.datahub.client.model.RecordEntry;
+import com.aliyun.datahub.client.model.RecordSchema;
+import com.aliyun.datahub.client.model.TupleRecordData;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.mokamrp.privates.entity.JobContactEmployee;
+import com.mokamrp.privates.mapper.pojo.CorpData;
+import com.mokamrp.privates.mapper.vo.WorkContactEmployeeHubVo;
+import com.mokamrp.privates.service.CorpDataService;
+import com.mokamrp.privates.service.WorkContactEmployeeService;
+import com.mokamrp.privates.utils.RedisUtil;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import java.text.SimpleDateFormat;
+import java.util.*;
+
+@Component
+public class DatahubCorpData {
+
+    @Autowired
+    private CorpDataService corpDataService;
+
+    @Resource
+    private RedisUtil redisUtil;
+
+    @Resource
+    private DatahubClient datahubClient;
+
+    @Value("${datahub.corpTopic}")
+    private String topic;
+
+    @Value("${datahub.project}")
+    private String project;
+
+    /**
+     * 导出任务，单进程
+     */
+    @Scheduled(fixedDelay = 300000)
+    public void execute() {
+        System.out.println("执行概览数据推送datahub任务,当前时间"+new Date());
+        job(project,topic,3);
+    }
+
+    public void job(String project,String topic,int retryTimes){
+        List<CorpData> list;
+
+        Map<String, Object> checkMap = new HashMap<>();
+        QueryWrapper<CorpData> q = new QueryWrapper<>();
+        q.ge("currentdateymd",getPastDate(7));
+        list = corpDataService.list(q);
+        if(list == null || list.size() <=0){
+            System.out.println("执行完毕CorpId"+1);
+            return ;
+        }
+
+        tupleExample(project,topic,retryTimes,list);
+    }
+
+
+    public static String getPastDate(int past) {
+        Calendar calendar = Calendar.getInstance();
+        calendar.set(Calendar.DAY_OF_YEAR, calendar.get(Calendar.DAY_OF_YEAR) - past);
+        Date today = calendar.getTime();
+        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
+        String result = format.format(today);
+        return result;
+    }
+    // 写入Tuple型数据
+    public void tupleExample(String project,String topic,int retryTimes, List<CorpData> list) {
+        Date date = new Date();
+        String timestamp = String.valueOf((new Date()).getTime()/1000);
+
+        System.out.println(timestamp);
+        Integer time =  Integer.valueOf(timestamp);
+
+        String strDateFormat = "yyyyMMdd";
+        SimpleDateFormat sdf = new SimpleDateFormat(strDateFormat);
+        String resDate = sdf.format(new Date());
+        // 获取schema
+        RecordSchema recordSchema = datahubClient.getTopic(project,topic ).getRecordSchema();
+
+        // 推送数据，生成十条数据
+        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
+        List<RecordEntry> recordEntries = new ArrayList<>();
+        for (CorpData vo:list
+             ) {
+
+            RecordEntry recordEntry = new RecordEntry();
+            // 对每条数据设置额外属性，例如ip 机器名等。可以不设置额外属性，不影响数据写入
+            //recordEntry.addAttribute("key1", "value1");
+            TupleRecordData data = new TupleRecordData(recordSchema);
+            data.setField("primary_key", vo.getPrimaryKey());
+            data.setField("current_date", format.format(vo.getCurrentdateymd()));
+            data.setField("corp_wx_id", vo.getCorpWxId());
+            data.setField("name", vo.getName());
+            data.setField("increase_user_count", vo.getIncreaseUserCount());
+            data.setField("decrease_user_count", vo.getDecreaseUserCount());
+            data.setField("total_user_count", vo.getTotalUserCount());
+            data.setField("group_decrease_user_count", vo.getGroupDecreaseUserCount());
+            data.setField("group_increase_user_count", vo.getGroupIncreaseUserCount());
+            data.setField("group_total_group_count", vo.getGroupTotalGroupCount());
+            data.setField("group_total_user_count", vo.getGroupTotalUserCount());
+            data.setField("updatetime", time);
+            data.setField("report_time", time);
+            data.setField("ds", resDate);
+            recordEntry.setRecordData(data);
+            recordEntries.add(recordEntry);
+        }
+
+        try {
+            PutRecordsResult result = datahubClient.putRecords(project, topic, recordEntries);
+            int i = result.getFailedRecordCount();
+            System.out.println("失败了"+i+"次");
+            if (i > 0) {
+                retry(datahubClient, result.getFailedRecords(), retryTimes, project, topic);
+            }
+        }  catch (DatahubClientException e) {
+            System.out.println("requestId:" + e.getRequestId() + "\tmessage:" + e.getErrorMessage());
+        }
+
+        return ;
+    }
+
+    //重试机制
+    public static void retry(DatahubClient client, List<RecordEntry> records, int retryTimes, String project, String topic) {
+        boolean suc = false;
+        while (retryTimes != 0) {
+            retryTimes = retryTimes - 1;
+            PutRecordsResult recordsResult = client.putRecords(project, topic, records);
+            if (recordsResult.getFailedRecordCount() > 0) {
+                retry(client,recordsResult.getFailedRecords(),retryTimes,project,topic);
+            }
+            suc = true;
+            break;
+        }
+        if (!suc) {
+            System.out.println("retryFailure");
+        }
+    }
+
+    public static Integer date2TimeStamp(String date_str,String format){
+        try {
+            SimpleDateFormat sdf = new SimpleDateFormat(format);
+            return Integer.valueOf(String.valueOf(sdf.parse(date_str).getTime()/1000));
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        return 0;
+    }
+
+}

src/main/resources/application-local.properties

@@ -52,4 +52,5 @@ datahub.endpoint=https://dh-cn-shanghai.aliyuncs.com
 datahub.accessid=LTAI5tEWU1j9H2poG4zWgfEP
 datahub.accesssecret=p9tVmeDeeEENKhfsBsSeYc0F0ggH5b
 datahub.topic=s_hera_api_external_user
+datahub.corpTopic=ods_hera_msv_corp_data
 datahub.project=hera

src/main/resources/application-prod.properties

@@ -57,5 +57,6 @@ datahub.endpoint=https://dh-cn-shanghai.aliyuncs.com
 datahub.accessId=LTAI5tEWU1j9H2poG4zWgfEP
 datahub.accessSecret=p9tVmeDeeEENKhfsBsSeYc0F0ggH5b
 datahub.topic=s_hera_api_external_user
+datahub.corpTopic=ods_hera_msv_corp_data
 datahub.project=hera
 

src/main/resources/application-test.properties

@@ -61,7 +61,10 @@ datahub.endpoint=https://dh-cn-shanghai.aliyuncs.com
 datahub.accessId=LTAI5tEWU1j9H2poG4zWgfEP
 datahub.accessSecret=p9tVmeDeeEENKhfsBsSeYc0F0ggH5b
 datahub.topic=s_hera_api_external_user_test
+datahub.corpTopic=ods_hera_msv_corp_data
 datahub.project=hera
 
 
 
+
+