Переглянути джерело

头条账户日流水 定时任务重构

zhaoxian 1 рік тому
батько
коміт
5734e4c3ba

+ 2 - 0
jeecg-boot-module-system/src/main/java/org/jeecg/config/ShiroConfig.java

@@ -202,6 +202,8 @@ public class ShiroConfig {
         filterChainDefinitionMap.put("/kuaishou/batch/*", "anon");
         filterChainDefinitionMap.put("/dimension/kuaiShouDimensionAccount/*", "anon");
         filterChainDefinitionMap.put("/ai/projectCreate/*", "anon");
+        filterChainDefinitionMap.put("/agent/**", "anon");
+        filterChainDefinitionMap.put("/etl/reportAccountHourly/*", "anon");
         //获取快手实时组数据信息
         filterChainDefinitionMap.put("/kuaishou/realTime/*", "anon");
 

+ 4 - 6
module-job-kuaishou/src/main/java/cn/com/ctop/job/kuaishou/handler/SynchronousAccountTransactionJob.java

@@ -51,19 +51,17 @@ public class SynchronousAccountTransactionJob {
      */
     @XxlJob("synchronousAccountTransactionNew")
     public void synchronousAccountTransactionNew() throws Exception {
-
         String nowDate = DateUtils.getNowDate("yyyy-MM-dd");
-
         String yesterda1 = DateUtils.getAnotherDay("yyyy-MM-dd", nowDate, -1);
         String yesterda2 = DateUtils.getAnotherDay("yyyy-MM-dd", nowDate, -2);
         String yesterda3 = DateUtils.getAnotherDay("yyyy-MM-dd", nowDate, -3);
 
         try {
-            agentManagementService.synchronousAccount();
+//            agentManagementService.synchronousAccount();
 
-            agentManagementService.synchronousAccountTransactionDayNew(yesterda1,yesterda1,1,1000);
-            agentManagementService.synchronousAccountTransactionDayNew(yesterda2,yesterda2,1,1000);
-            agentManagementService.synchronousAccountTransactionDayNew(yesterda3,yesterda3,1,1000);
+            agentManagementService.insertAccountTransactionDay(yesterda1,yesterda1);
+            agentManagementService.insertAccountTransactionDay(yesterda2,yesterda2);
+            agentManagementService.insertAccountTransactionDay(yesterda3,yesterda3);
 
         } catch (Exception e) {
             log.info("同步完成");

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

@@ -37,6 +37,17 @@ public class AgentManagementController {
         Result recharge = agentManagementService.synchronousAccountTransactionDay(startDate, endDate);
     }
 
+    @RequestMapping("/insertAccountTransactionDay")
+    public void insertAccountTransactionDay(String startDate, String endDate) {
+        agentManagementService.insertAccountTransactionDay(startDate, endDate);
+    }
+
+
+    @RequestMapping("/insertAccountTransactionDayByProjectId")
+    public void insertAccountTransactionDayByProjectId(Long projectId, String startDate, String endDate) {
+        agentManagementService.insertAccountTransactionDayByProjectId(projectId, startDate, endDate);
+    }
+
 
     @RequestMapping("/synchronous")
     public void synchronous() throws ParseException {

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

@@ -11,6 +11,10 @@ public interface CtopAgentAccountMapper {
     //获取所有账户
     public List<String> getAllAccountId();
 
+    public List<String> getAllAccountId2();
+
+    List<String> getAccountIdsByProjectId(@Param("projectId") Long projectId);
+
     //根据账户id修改账户状态
     public void updateAccountState(@Param("list") List<String> list, @Param("state") int state);
 
@@ -24,4 +28,5 @@ public interface CtopAgentAccountMapper {
      * @return
      */
     CtopOauthToken selectByAccountId(Long accountId);
+
 }

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

@@ -17,19 +17,31 @@
     </resultMap>
 
     <sql id="Base_Column_List">
-        id,media_id,advertiser_id,account_id,access_token,refresh_token,access_token_expires_in,refresh_token_expires_in,agent_type,app_id,create_time,update_time
-   </sql>
+        id
+        ,media_id,advertiser_id,account_id,access_token,refresh_token,access_token_expires_in,refresh_token_expires_in,agent_type,app_id,create_time,update_time
+    </sql>
     <!--获取所有账户-->
     <select id="getAllAccountId" resultType="String">
-        SELECT
-            a.account_id
-        FROM
-            ctop_agent_account a
-                LEFT JOIN ctop_user_allocation u ON a.account_id = u.account_id
-        WHERE
-            a.state = 0
+        SELECT a.account_id
+        FROM ctop_agent_account a
+                 LEFT JOIN ctop_user_allocation u ON a.account_id = u.account_id
+        WHERE a.state = 0
           AND u.account_status = 0
     </select>
+    <!--获取所有账户-->
+    <select id="getAllAccountId2" resultType="String">
+        SELECT account_id
+        FROM ctop_user_allocation
+        WHERE account_status = 0
+          AND media_id = 1
+    </select>
+
+    <select id="getAccountIdsByProjectId" resultType="String">
+        SELECT account_id
+        FROM ctop_user_allocation
+        WHERE project_id = #{projectId}
+          AND media_id = 1
+    </select>
     <!-- 根据账户id修改账户状态-->
     <update id="updateAccountState">
         update ctop_agent_account set state=#{state} where account_id in

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

@@ -26,4 +26,8 @@ public interface AgentManagementService {
     //同步账户日流水
     public Result synchronousAccountTransactionDay(String startDate, String endDate);
     public void synchronousAccountTransactionDayNew(String startDate, String endDate,int pageNum, int pageSize);
+
+    void insertAccountTransactionDay(String startDate, String endDate);
+
+    void insertAccountTransactionDayByProjectId(Long projectId, String startDate, String endDate);
 }

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

@@ -1,7 +1,7 @@
 package cn.com.ctop.toutiao.modules.agent.service.impl;
 
 import cn.com.ctop.common.module.entity.CtopOauthToken;
-import cn.com.ctop.common.module.utils.HttpUtils;
+import cn.com.ctop.common.module.utils.Check;
 import cn.com.ctop.toutiao.modules.agent.entity.TransactionDay;
 import cn.com.ctop.toutiao.modules.agent.entity.TransactionDetails;
 import cn.com.ctop.toutiao.modules.agent.mapper.CtopAccountTransactionDayMapper;
@@ -11,7 +11,6 @@ import cn.com.ctop.toutiao.modules.agent.service.AgentManagementService;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
-import javafx.beans.binding.When;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.http.client.ClientProtocolException;
 import org.apache.http.client.methods.CloseableHttpResponse;
@@ -29,7 +28,14 @@ import java.io.IOException;
 import java.io.InputStreamReader;
 import java.net.URI;
 import java.text.SimpleDateFormat;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 import static java.util.stream.Collectors.toList;
 
@@ -51,6 +57,9 @@ public class AgentManagementServiceImpl implements AgentManagementService {
     private final String RECHARGE = "RECHARGE";
     private final String TRANSFER = "TRANSFER";
 
+    static ExecutorService executorService = Executors.newFixedThreadPool(3);
+
+
     @Resource
     CtopAgentAccountMapper ctopAgentAccountMapper;
     @Resource
@@ -80,9 +89,10 @@ public class AgentManagementServiceImpl implements AgentManagementService {
                 for (String account : split) {
                     //分页查询第一页数据
                     JSONObject agentAccountList = selectAgentAdvertiser(1, 2000, account);
-                    if (!Integer.valueOf(agentAccountList.get("code").toString()).equals(0)){
-                        log.info("代理商:{}拉取账户异常--{}",account,agentAccountList);
-                    };
+                    if (!Integer.valueOf(agentAccountList.get("code").toString()).equals(0)) {
+                        log.info("代理商:{}拉取账户异常--{}", account, agentAccountList);
+                    }
+                    ;
 
                     JSONObject data = (JSONObject) agentAccountList.get("data");
                     //dataList.addAll((List) data.get("list"));
@@ -263,7 +273,7 @@ public class AgentManagementServiceImpl implements AgentManagementService {
      */
     @Override
     public Result synchronousAccountTransactionDay(String startDate, String endDate) {
-        log.info("开始同步账号日流水---日期{}---{}",startDate,endDate);
+        log.info("开始同步账号日流水---日期{}---{}", startDate, endDate);
         Result result = new Result();
         SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd");//设置日期格式
         Calendar ca = Calendar.getInstance();
@@ -301,9 +311,9 @@ public class AgentManagementServiceImpl implements AgentManagementService {
                                     TransactionDay transactionDay = JSONObject.toJavaObject((JSON) ob, TransactionDay.class);
                                     List<String> advertiserIdList = ctopAccountTransactionDayMapper.getAdvertiserIdByDate(transactionDay.getDate(), transactionDay.getAdvertiserId());
 
-                                    if (advertiserIdList.size() > 1){
+                                    if (advertiserIdList.size() > 1) {
                                         ctopAccountTransactionDayMapper.deleteAccountTransactionById(advertiserIdList.get(0));
-                                        log.info("----------->删除重复数据。accountId:{},date:{}",transactionDay.getAdvertiserId(),transactionDay.getDate());
+                                        log.info("----------->删除重复数据。accountId:{},date:{}", transactionDay.getAdvertiserId(), transactionDay.getDate());
                                     }
                                     //订单号不存在
                                     if (advertiserIdList.isEmpty() || advertiserIdList.size() <= 0) {
@@ -323,7 +333,7 @@ public class AgentManagementServiceImpl implements AgentManagementService {
                                             TransactionDay transactionDay = JSONObject.toJavaObject((JSON) obj, TransactionDay.class);
                                             List<String> advertiserIdList = ctopAccountTransactionDayMapper.getAdvertiserIdByDate(transactionDay.getDate(), transactionDay.getAdvertiserId());
 
-                                            if (advertiserIdList.size() > 1){
+                                            if (advertiserIdList.size() > 1) {
                                                 ctopAccountTransactionDayMapper.deleteAccountTransactionById(advertiserIdList.get(0));
                                             }
 
@@ -356,14 +366,59 @@ public class AgentManagementServiceImpl implements AgentManagementService {
 
 
         return result;
-    } /**
+    }
+
+
+    @Override
+    public void insertAccountTransactionDay(String startDate, String endDate) {
+        List<String> accountIdList = ctopAgentAccountMapper.getAllAccountId2();
+        if (!Check.isNull(accountIdList)) {
+            accountIdList.forEach(accountId -> executorService.submit(() -> this.insertAccountTransactionDay(accountId, startDate, endDate, 1)));
+        }
+    }
+
+    @Override
+    public void insertAccountTransactionDayByProjectId(Long projectId, String startDate, String endDate) {
+        List<String> accountIdList = ctopAgentAccountMapper.getAccountIdsByProjectId(projectId);
+        if (!Check.isNull(accountIdList)) {
+            accountIdList.forEach(accountId -> executorService.submit(() -> this.insertAccountTransactionDay(accountId, startDate, endDate, 1)));
+        }
+    }
+
+    public void insertAccountTransactionDay(String accountId, String startDate, String endDate, Integer page) {
+        try {
+            JSONObject obj = queryAccountTransactionDay(accountId, page, 1000, startDate, endDate);
+            if (!Check.isNull(obj) && obj.getInteger("code") == 0) {
+                JSONObject data = obj.getJSONObject("data");
+                JSONArray list = data.getJSONArray("list");
+                List<TransactionDay> transactionDays = JSONArray.parseArray(list.toJSONString(), TransactionDay.class);
+                if (!Check.isNull(transactionDays)) {
+                    ctopAccountTransactionDayMapper.replaceAccountTransactionDay(transactionDays);
+                }
+
+                JSONObject pageInfo = data.getJSONObject("page_info");
+                Integer totalPage = pageInfo.getInteger("total_page");
+                if (totalPage > page) {
+                    insertAccountTransactionDay(accountId, startDate, endDate, page + 1);
+                }
+            }
+
+        } catch (Exception e) {
+            e.printStackTrace();
+            log.info("同步失败:{}", e.toString());
+        }
+
+
+    }
+
+    /**
      * 同步账户日流水
      *
      * @author zzy
      * @create: 2021-07-15
      */
     @Override
-    public void synchronousAccountTransactionDayNew(String startDate, String endDate,int page, int pageSize) {
+    public void synchronousAccountTransactionDayNew(String startDate, String endDate, int page, int pageSize) {
 
         List<TransactionDay> list = new ArrayList<>();
         List<String> accountIdList = ctopAgentAccountMapper.getAllAccountId();
@@ -375,7 +430,7 @@ public class AgentManagementServiceImpl implements AgentManagementService {
 
         for (String accountId : accountIdList) {
             page = 1;
-            log.info("同步头条账户流水信息{}:开始时间{},结束时间{},第{}页" ,accountId,startDate,endDate,page);
+            log.info("同步头条账户流水信息{}:开始时间{},结束时间{},第{}页", accountId, startDate, endDate, page);
 
             JSONObject jsonObject = queryAccountTransactionDay(accountId, page, 1000, startDate, endDate);
             JSONObject data = null;
@@ -401,7 +456,7 @@ public class AgentManagementServiceImpl implements AgentManagementService {
                 }
             }
             ctopAccountTransactionDayMapper.replaceAccountTransactionDay(list);
-            log.info("获取头条账户:{}流水信息{}条" ,accountId,list.size());
+            log.info("获取头条账户:{}流水信息{}条", accountId, list.size());
             page++;
             int totalPage = jsonObject.getJSONObject("data").getJSONObject("page_info").getInteger("total_page");
             if (page <= totalPage) {