Browse Source

Merge remote-tracking branch 'origin/master_rule_engine' into master_rule_engine

yumeng 4 years ago
parent
commit
3169f35677

+ 10 - 0
jeecg-boot-module-system/src/test/java/org/jeecg/SampleTest.java

@@ -6,6 +6,7 @@ import cn.com.ctop.common.module.entity.UserAllocation;
 import cn.com.ctop.common.module.message.handle.impl.EmailSendMsgHandle;
 import cn.com.ctop.common.module.service.*;
 import cn.com.ctop.common.module.utils.CtopAdConstant;
+import cn.com.ctop.common.module.utils.HttpUtils;
 import cn.com.ctop.kuaishou.modules.batch.service.IKuaiShouHistoryReportTaskService;
 import cn.com.ctop.kuaishou.modules.batch.service.IKuaishouInterfaceService;
 import cn.com.ctop.kuaishou.modules.graphql.service.IKuaishouWebInterfaceService;
@@ -115,6 +116,15 @@ public class SampleTest {
     }
 
     @Test
+    public void testXxlJobDistribute(){
+        String url = "http://127.0.0.1:8090/xxl-job-admin/jobinfo/trigger?id=88&executorParam=";
+        for(int i=0;i<100;i++){
+            String result = HttpUtils.httpPostRequest(url+i,new HashMap<>(),new HashMap<>());
+            System.out.println(result);
+        }
+    }
+
+    @Test
     public void testLoadRulePlanData(){
         Date date = new Date();
         List<CtopOauthToken> tokens = tokenService.selectToutiaoToken();

+ 5 - 0
module-alarm/src/main/java/cn/com/ctop/alarm/modules/mapper/RuleAccountThresholdMapper.java

@@ -17,6 +17,11 @@ public interface RuleAccountThresholdMapper extends BaseMapper<RuleAccountThresh
 
     List<RuleAccountThreshold> selectByAccountId(Long accountId);
 
+    List<Long> selectGroupIdsByAccountId(Long accountId);
+
+    List<String> selectbatchNosByGroupIdId(Long accountId, Long groupId);
+
+    List<RuleAccountThreshold> selectCopyThresholdByAccountId(Long accountId, Long groupId, String batchNo);
     List<String> getBatchNo(@Param("accountId") Long accountId, @Param("groupId") Long groupId);
 
     List<RuleAccountThreshold> selectBatchNoByAccountId(Long accountId);

+ 16 - 3
module-alarm/src/main/java/cn/com/ctop/alarm/modules/mapper/xml/RuleAccountThresholdMapper.xml

@@ -7,11 +7,24 @@
        WHERE account_id = #{accountId}
        AND batch_no is null
     </select>
-    <select id="selectBatchNoByAccountId" resultType="cn.com.ctop.alarm.modules.entity.RuleAccountThreshold">
-       SELECT group_id,batch_no FROM `ctop_rule_account_threshold`
+    <select id="selectGroupIdsByAccountId" resultType="java.lang.Long">
+       SELECT group_id FROM `ctop_rule_account_threshold`
         WHERE batch_no is not null
         and  account_id = #{accountId}
-        group by group_id,batch_no
+        group by group_id
+    </select>
+    <select id="selectbatchNosByGroupIdId" resultType="java.lang.String">
+       SELECT batch_no FROM `ctop_rule_account_threshold`
+        WHERE batch_no is not null
+        and account_id = #{accountId}
+        and group_id = #{groupId}
+        GROUP BY batch_no
+    </select>
+    <select id="selectCopyThresholdByAccountId" resultType="cn.com.ctop.alarm.modules.entity.RuleAccountThreshold">
+       SELECT * FROM `ctop_rule_account_threshold`
+        WHERE account_id = #{accountId}
+        and group_id = #{groupId}
+        and batch_no = #{batchNo}
     </select>
 
     <select id="getBatchNo" resultType="java.lang.String">

+ 132 - 60
module-alarm/src/main/java/cn/com/ctop/alarm/modules/service/impl/RuleGroupServiceImpl.java

@@ -108,101 +108,157 @@ public class RuleGroupServiceImpl extends ServiceImpl<RuleGroupMapper, RuleGroup
             log.warn("查询匹配数据失败");
             return;
         }
+        boolean isCopy = false;
         Map<Long, Object> map = (Map<Long, Object>) ruleGroupMap.get(templates.getTemplateId());
         //媒体类型 1:头条2:快手
         Integer type = (Integer) map.get("type");
         //常规规则+阈值
         List<RuleGroup> notCopyGroups = (List<RuleGroup>) map.get("notCopy");
-        JSONObject notCopythreshold = getThreshold(templates.getAccountId(), false);
+        JSONObject notCopythreshold = getThreshold(templates.getAccountId(), isCopy);
         //可复制规则+阈值
         List<RuleGroup> isCopyGroups = (List<RuleGroup>) map.get("isCopy");
+        JSONObject isCopythreshold = new JSONObject();
         if (!Check.isNull(isCopyGroups)) {
-            JSONObject isCopythreshold = getThreshold(templates.getAccountId(), true);
+            isCopy = true;
+            isCopythreshold = getThreshold(templates.getAccountId(), isCopy);
         }
         log.info("已获取账户({})的数据,开始匹配规则", templates.getAccountId());
         try {
             if (!Check.isNull(notCopyGroups)) {
-                checkRuleGroups(notCopyGroups, notCopythreshold, indicators, matchData, type);
+                checkRuleGroups(notCopyGroups, notCopythreshold, indicators, matchData, type, false);
             }
+            if (isCopy && !Check.isNull(isCopythreshold)) {
+                checkRuleGroups(isCopyGroups, isCopythreshold, indicators, matchData, type, isCopy);
+            }
+
         } catch (Exception e) {
             log.error("匹配规则逻辑异常", e);
         }
     }
 
-    private void checkRuleGroups(List<RuleGroup> ruleGroups, JSONObject thresholdObj, JSONObject indicators, JSONObject matchData, Integer type) {
+    private void checkRuleGroups(List<RuleGroup> ruleGroups, JSONObject thresholdObj, JSONObject indicators, JSONObject matchData, Integer type, boolean isCopy) {
         for (RuleGroup ruleGroup : ruleGroups) {
             List<RuleBase> ruleBaseList = ruleGroup.getRuleBaseList();
             if (Check.isNull(ruleBaseList)) {
                 continue;
             }
-            boolean isBase = "base".equals(ruleGroup.getRuleType());
-            //匹配单规则
-            if (isBase) {
-                RuleBase ruleBase = ruleBaseList.get(0);
-                //指标对象
-                JSONObject indicator = indicators.getJSONObject(ruleBase.getIndicatorCode());
-                //指标阈值
-                String threshold = thresholdObj.getString(ruleBase.getId().toString());
-                //阈值为空时,或者阈值为“unlimited”(不限),不执行该规则
-                if (Check.isNull(threshold) || "unlimited".equals(threshold)) {
-                    log.warn("阈值为空/不限,该规则({})不执行", ruleBase.getId());
-                    continue;
-                }
-                //维度数据
-                JSONArray dimensionData = matchData.getJSONArray(ruleBase.getRuleDimension());
-                if (!Check.isNull(dimensionData)) {
-                    for (int i = 0; i < dimensionData.size(); i++) {
-                        JSONObject obj = dimensionData.getJSONObject(i);
-                        String value = obj.getString(ruleBase.getIndicatorCode());
-                        if (MatchLogic.matchCondition(indicator.getString("dataType"), ruleBase.getRuleCondition(), threshold, value)) {
-                            sendMsg(ruleGroup, obj);
+            //复制逻辑
+            if (isCopy) {
+                JSONObject batchNo = thresholdObj.getJSONObject(ruleGroup.getId().toString());
+                for (Map.Entry<String, Object> entry : batchNo.entrySet()) {
+                    Map<String, String> thresholdMap = (Map<String, String>) entry.getValue();
+                    //符合规则的数据集
+                    JSONArray targetDatas = new JSONArray();
+                    JSONArray planDatas = new JSONArray();
+                    JSONArray creativeDatas = new JSONArray();
+                    JSONArray accountDatas = new JSONArray();
+                    for (RuleBase ruleBase : ruleBaseList) {
+                        //指标对象
+                        JSONObject indicator = indicators.getJSONObject(ruleBase.getIndicatorCode());
+                        //阈值类型
+                        String dataType = indicator.getString("dataType");
+                        //指标阈值
+                        String threshold = thresholdMap.get(ruleBase.getId());
+                        //阈值为空时,或者阈值为“unlimited”(不限),不执行该规则
+                        if (Check.isNull(threshold) || "unlimited".equals(threshold)) {
+                            continue;
+                        }
+                        JSONArray dimensionData = matchData.getJSONArray(ruleBase.getRuleDimension());
+                        //账户维度数据
+                        if ("account".equals(ruleBase.getRuleDimension())) {
+                            JSONObject accountEntity = dimensionData.getJSONObject(0);
+                            if (MatchLogic.matchCondition(dataType, ruleBase.getRuleCondition(), threshold, accountEntity.getString(ruleBase.getIndicatorCode()))) {
+                                accountDatas.add(accountEntity);
+                            } else {
+                                log.warn("判断规则组({}),账户规则不符合", ruleGroup.getId());
+                                break;
+                            }
+                        } else if ("plan".equals(ruleBase.getRuleDimension())) {
+                            planDatas = getOKData(planDatas, dimensionData, ruleBase, dataType, threshold);
+                        } else if ("creative".equals(ruleBase.getRuleDimension())) {
+                            creativeDatas = getOKData(creativeDatas, dimensionData, ruleBase, dataType, threshold);
+                        } else if ("target".equals(ruleBase.getRuleDimension())) {
+                            targetDatas = getOKData(targetDatas, dimensionData, ruleBase, dataType, threshold);
+                        }
+                    }
+                    //获取达标可发送的数据
+                    JSONArray sendData = MatchLogic.getSendData(accountDatas, planDatas, creativeDatas, targetDatas);
+                    if (!Check.isNull(sendData)) {
+                        //执行发送
+                        for (int i = 0; i < sendData.size(); i++) {
+                            sendMsg(ruleGroup, sendData.getJSONObject(i));
                         }
                     }
                 }
-                //匹配组合规则
             } else {
-                boolean flag = false;
-                //符合规则的数据集
-                JSONArray targetDatas = new JSONArray();
-                JSONArray planDatas = new JSONArray();
-                JSONArray creativeDatas = new JSONArray();
-                JSONArray accountDatas = new JSONArray();
-                for (RuleBase ruleBase : ruleBaseList) {
+                boolean isBase = "base".equals(ruleGroup.getRuleType());
+                //匹配单规则
+                if (isBase) {
+                    RuleBase ruleBase = ruleBaseList.get(0);
                     //指标对象
                     JSONObject indicator = indicators.getJSONObject(ruleBase.getIndicatorCode());
-                    //阈值类型
-                    String dataType = indicator.getString("dataType");
                     //指标阈值
                     String threshold = thresholdObj.getString(ruleBase.getId().toString());
                     //阈值为空时,或者阈值为“unlimited”(不限),不执行该规则
                     if (Check.isNull(threshold) || "unlimited".equals(threshold)) {
+                        log.warn("阈值为空/不限,该规则({})不执行", ruleBase.getId());
                         continue;
                     }
+                    //维度数据
                     JSONArray dimensionData = matchData.getJSONArray(ruleBase.getRuleDimension());
-                    //账户维度数据
-                    if ("account".equals(ruleBase.getRuleDimension())) {
-                        JSONObject accountEntity = dimensionData.getJSONObject(0);
-                        if (MatchLogic.matchCondition(dataType, ruleBase.getRuleCondition(), threshold, accountEntity.getString(ruleBase.getIndicatorCode()))) {
-                            accountDatas.add(accountEntity);
-                        } else {
-                            log.warn("判断规则组({}),账户规则不符合", ruleGroup.getId());
-                            break;
+                    if (!Check.isNull(dimensionData)) {
+                        for (int i = 0; i < dimensionData.size(); i++) {
+                            JSONObject obj = dimensionData.getJSONObject(i);
+                            String value = obj.getString(ruleBase.getIndicatorCode());
+                            if (MatchLogic.matchCondition(indicator.getString("dataType"), ruleBase.getRuleCondition(), threshold, value)) {
+                                sendMsg(ruleGroup, obj);
+                            }
                         }
-                    } else if ("plan".equals(ruleBase.getRuleDimension())) {
-                        planDatas = getOKData(planDatas, dimensionData, ruleBase, dataType, threshold);
-                    } else if ("creative".equals(ruleBase.getRuleDimension())) {
-                        creativeDatas = getOKData(creativeDatas, dimensionData, ruleBase, dataType, threshold);
-                    } else if ("target".equals(ruleBase.getRuleDimension())) {
-                        targetDatas = getOKData(targetDatas, dimensionData, ruleBase, dataType, threshold);
                     }
+                    //匹配组合规则
+                } else {
+                    //符合规则的数据集
+                    JSONArray targetDatas = new JSONArray();
+                    JSONArray planDatas = new JSONArray();
+                    JSONArray creativeDatas = new JSONArray();
+                    JSONArray accountDatas = new JSONArray();
+                    for (RuleBase ruleBase : ruleBaseList) {
+                        //指标对象
+                        JSONObject indicator = indicators.getJSONObject(ruleBase.getIndicatorCode());
+                        //阈值类型
+                        String dataType = indicator.getString("dataType");
+                        //指标阈值
+                        String threshold = thresholdObj.getString(ruleBase.getId().toString());
+                        //阈值为空时,或者阈值为“unlimited”(不限),不执行该规则
+                        if (Check.isNull(threshold) || "unlimited".equals(threshold)) {
+                            continue;
+                        }
+                        JSONArray dimensionData = matchData.getJSONArray(ruleBase.getRuleDimension());
+                        //账户维度数据
+                        if ("account".equals(ruleBase.getRuleDimension())) {
+                            JSONObject accountEntity = dimensionData.getJSONObject(0);
+                            if (MatchLogic.matchCondition(dataType, ruleBase.getRuleCondition(), threshold, accountEntity.getString(ruleBase.getIndicatorCode()))) {
+                                accountDatas.add(accountEntity);
+                            } else {
+                                log.warn("判断规则组({}),账户规则不符合", ruleGroup.getId());
+                                break;
+                            }
+                        } else if ("plan".equals(ruleBase.getRuleDimension())) {
+                            planDatas = getOKData(planDatas, dimensionData, ruleBase, dataType, threshold);
+                        } else if ("creative".equals(ruleBase.getRuleDimension())) {
+                            creativeDatas = getOKData(creativeDatas, dimensionData, ruleBase, dataType, threshold);
+                        } else if ("target".equals(ruleBase.getRuleDimension())) {
+                            targetDatas = getOKData(targetDatas, dimensionData, ruleBase, dataType, threshold);
+                        }
 
-                }
-                //获取达标可发送的数据
-                JSONArray sendData = MatchLogic.getSendData(accountDatas, planDatas, creativeDatas, targetDatas);
-                if (!Check.isNull(sendData)) {
-                    //执行发送
-                    for (int i = 0; i < sendData.size(); i++) {
-                        sendMsg(ruleGroup, sendData.getJSONObject(i));
+                    }
+                    //获取达标可发送的数据
+                    JSONArray sendData = MatchLogic.getSendData(accountDatas, planDatas, creativeDatas, targetDatas);
+                    if (!Check.isNull(sendData)) {
+                        //执行发送
+                        for (int i = 0; i < sendData.size(); i++) {
+                            sendMsg(ruleGroup, sendData.getJSONObject(i));
+                        }
                     }
                 }
             }
@@ -322,13 +378,29 @@ public class RuleGroupServiceImpl extends ServiceImpl<RuleGroupMapper, RuleGroup
         JSONObject obj = new JSONObject();
         try {
             if (isCopy) {
-                List<RuleAccountThreshold> copyThresholdList = ruleAccountThresholdMapper.selectBatchNoByAccountId(accountId);
-                if (Check.isNull(copyThresholdList)) {
+                List<Long> groupList = ruleAccountThresholdMapper.selectGroupIdsByAccountId(accountId);
+                if (Check.isNull(groupList)) {
                     return null;
                 }
-                for (RuleAccountThreshold threshold : copyThresholdList) {
+                for (Long groupId : groupList) {
+                    JSONObject groupIdObj = new JSONObject();
+                    List<String> batchNos = ruleAccountThresholdMapper.selectbatchNosByGroupIdId(accountId, groupId);
+                    if (Check.isNull(batchNos)) {
+                        return null;
+                    }
+                    for (String batchNo : batchNos) {
+                        JSONObject batchNoObj = new JSONObject();
+                        List<RuleAccountThreshold> thresholdList = ruleAccountThresholdMapper.selectCopyThresholdByAccountId(accountId, groupId, batchNo);
+                        if (Check.isNull(thresholdList)) {
+                            return null;
+                        }
+                        for (RuleAccountThreshold threshold : thresholdList) {
+                            batchNoObj.put(String.valueOf(threshold.getRuleId()), threshold.getThreshold());
+                        }
+                        groupIdObj.put(batchNo.toString(), batchNoObj);
+                    }
+                    obj.put(groupId.toString(), groupIdObj);
                 }
-
             } else {
                 List<RuleAccountThreshold> thresholdList = ruleAccountThresholdMapper.selectByAccountId(accountId);
                 if (Check.isNull(thresholdList)) {

+ 1 - 8
module-job-bytedance/src/main/java/cn/com/ctop/job/bytedance/handler/ByteDancePlanRuleDataJob.java

@@ -11,9 +11,6 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import java.util.Date;
-import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 
 @Component
 public class ByteDancePlanRuleDataJob {
@@ -21,7 +18,6 @@ public class ByteDancePlanRuleDataJob {
     private ICtopOauthTokenService tokenService;
     @Autowired
     private IReportService reportService;
-    static ExecutorService executorService = Executors.newFixedThreadPool(10);
 
     @XxlJob("bytedancePlanLoadJob")
     public ReturnT<String> execute(String param) throws Exception {
@@ -30,11 +26,8 @@ public class ByteDancePlanRuleDataJob {
             Long accountId = Long.parseLong(param);
             CtopOauthToken token = tokenService.getTokenByAccountId(accountId);
             reportService.getAdvertiserPlanRuleData(token,date,date,CtopAdConstant.BYTEDANCE_REPORT_TYPE_DAILY);
-        }else{
-            List<CtopOauthToken> tokens = tokenService.selectToutiaoToken();
-            tokens.forEach(token -> executorService.submit(() -> reportService.getAdvertiserPlanRuleData(token,date,date, CtopAdConstant.BYTEDANCE_REPORT_TYPE_DAILY)));
         }
-        XxlJobLogger.log("头条全量计划数据获取完成");
+        XxlJobLogger.log("头条计划数据清洗完成:accountId=>{}",param);
         return ReturnT.SUCCESS;
     }
 }

+ 46 - 0
module-job-bytedance/src/main/java/cn/com/ctop/job/bytedance/handler/BytedanceRuleDataJobDistribute.java

@@ -0,0 +1,46 @@
+package cn.com.ctop.job.bytedance.handler;
+
+import cn.com.ctop.common.module.entity.CtopOauthToken;
+import cn.com.ctop.common.module.service.ICtopOauthTokenService;
+import cn.com.ctop.common.module.utils.HttpUtils;
+import com.xxl.job.core.biz.model.ReturnT;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import com.xxl.job.core.log.XxlJobLogger;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.HashMap;
+import java.util.List;
+
+@Component
+public class BytedanceRuleDataJobDistribute {
+    @Autowired
+    private ICtopOauthTokenService tokenService;
+    @XxlJob("bytedancePlanRuleDataJobDistribute")
+    public ReturnT<String> bytedancePlanRuleDataJobDistribute(String params) throws Exception {
+        List<CtopOauthToken> tokens = tokenService.selectToutiaoToken();
+        if(null!=tokens&&!tokens.isEmpty()){
+            //TODO 指定相应的任务id
+            String planUrl = "http://139.186.29.76:8090/jobinfo/trigger?id=87&executorParam=";
+            for(CtopOauthToken token:tokens){
+                String planResult = HttpUtils.httpPostRequest(planUrl+token.getAccountId(),new HashMap<>(),new HashMap<>());
+                XxlJobLogger.log("accountId:{};message:{}",token.getAccountId(),planResult);
+            }
+        }
+        return ReturnT.SUCCESS;
+    }
+
+    @XxlJob("bytedanceAccountRuleDataJobDistribute")
+    public ReturnT<String> bytedanceAccountRuleDataJobDistribute(String params) throws Exception {
+        List<CtopOauthToken> tokens = tokenService.selectToutiaoToken();
+        if(null!=tokens&&!tokens.isEmpty()){
+            //TODO
+            String accountUrl = "http://139.186.29.76:8090/jobinfo/trigger?id=87&executorParam=";
+            for(CtopOauthToken token:tokens){
+                String accountResult = HttpUtils.httpPostRequest(accountUrl+token.getAccountId(),new HashMap<>(),new HashMap<>());
+                XxlJobLogger.log("accountId:{};message:{}",token.getAccountId(),accountResult);
+            }
+        }
+        return ReturnT.SUCCESS;
+    }
+}

+ 4 - 13
module-job-bytedance/src/main/java/cn/com/ctop/job/bytedance/handler/RuleDataAccountCleanJob.java

@@ -3,7 +3,6 @@ package cn.com.ctop.job.bytedance.handler;
 /**
  *  Created by JQ.bi on 2020.11.16
  */
-import cn.com.ctop.common.module.entity.CtopOauthToken;
 import cn.com.ctop.common.module.service.ICtopOauthTokenService;
 import cn.com.ctop.toutiao.modules.report.service.IRuleByteDanceAccountService;
 import com.xxl.job.core.biz.model.ReturnT;
@@ -12,8 +11,6 @@ import com.xxl.job.core.log.XxlJobLogger;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
-
-import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
@@ -31,18 +28,12 @@ public class RuleDataAccountCleanJob {
     @XxlJob("ruleDataAccountCleanJob")
     public ReturnT<String> execute(String param) throws Exception {
         XxlJobLogger.log("规则预警账户维度数据开始清洗");
-        //查询所有头条账户
-        List<CtopOauthToken> tokens = oauthTokenService.selectToutiaoToken();
-        if (null == tokens || tokens.isEmpty()) {
-            XxlJobLogger.log("未获取到可用的token");
+        if (null == param || "".equals(param.trim())) {
+            XxlJobLogger.log("参数异常");
             return ReturnT.FAIL;
         }
-        tokens.forEach(token->executorPool.submit(() -> {
-            ruleByteDanceAccountService.cleanRuleDataAccount(token.getAccountId());
-        }));
-        if(!executorPool.isShutdown()){
-            executorPool.shutdown();
-        }
+        Long accountId = Long.parseLong(param);
+        ruleByteDanceAccountService.cleanRuleDataAccount(accountId);
         XxlJobLogger.log("规则预警账户维度数据清洗结束");
         return ReturnT.SUCCESS;
     }