提交一下

pom.xml

@@ -344,7 +344,7 @@
         <dependency>
             <groupId>cn.qg.ec.data-stream-sdk</groupId>
             <artifactId>data-stream-sdk</artifactId>
-            <version>1.0.5-SNAPSHOT</version>
+            <version>1.0.8-SNAPSHOT</version>
             <exclusions>
                 <exclusion>
                     <groupId>com.google.guava</groupId>

src/main/java/cn/quantgroup/xyqb/aspect/captcha/LoginInterceptorAspect.java

 package cn.quantgroup.xyqb.aspect.captcha;
 
+import cn.qg.ec.model.base.BusinessEventBaseInfo;
+import cn.qg.ec.model.user.UserLoginRegEvent;
+import cn.quantgroup.xyqb.entity.User;
+import cn.quantgroup.xyqb.model.AuthBean;
 import cn.quantgroup.xyqb.risk.entity.CountDevice;
 import cn.quantgroup.xyqb.entity.enums.Device;
 import cn.quantgroup.xyqb.entity.enums.KeyType;
@@ -8,7 +12,13 @@ import cn.quantgroup.xyqb.model.LoginRefuseResult;
 import cn.quantgroup.xyqb.risk.entity.LoginInfo;
 import cn.quantgroup.xyqb.risk.repository.LoginInfoRepository;
 import cn.quantgroup.xyqb.risk.repository.WhiteListRepository;
+import cn.quantgroup.xyqb.risk.uereventcollecting.UserEventCollectingUtil;
+import cn.quantgroup.xyqb.service.user.IUserService;
+import cn.quantgroup.xyqb.service.user.impl.UserServiceImpl;
+import cn.quantgroup.xyqb.util.encrypt.Md5Util;
+import com.amazonaws.util.Md5Utils;
 import com.google.common.collect.Maps;
+import com.qiniu.util.Md5;
 import lombok.extern.slf4j.Slf4j;
 import org.aspectj.lang.ProceedingJoinPoint;
 import org.aspectj.lang.annotation.Around;
@@ -42,6 +52,9 @@ public class LoginInterceptorAspect {
     @Resource
     private WhiteListRepository whiteListRepository;
 
+    @Resource
+    private IUserService iUserService;
+
     private static final String DEFAULT_CODE = "00000000-0000-0000-0000-000000000000";
     /**
      * 设备拒绝阈值
@@ -97,8 +110,13 @@ public class LoginInterceptorAspect {
              */
             deviceId = DEFAULT_CODE;
         }
+
         log.info("登录前风控策略开始deviceId:{},phone:{}", deviceId, phone);
         Device device = Device.valueOfCode(deviceCode);
+        /**
+         * 尝试登录发一下
+         */
+        sendEvent(UserLoginRegEvent.SubEventType.user_attempt_login, device, phone, deviceId, realIp, null);
 
         LoginRefuseResult result = verification(scDeviceId, phone, device);
 
@@ -244,7 +262,7 @@ public class LoginInterceptorAspect {
             if (null == loginResult) {
                 info.setIsLogin(Boolean.FALSE);
                 info.setLoginFailMsg("系统异常");
-                if(!refuseResult.isPass()){
+                if (!refuseResult.isPass()) {
                     info.setLoginFailMsg("命中风控策略");
                 }
 
@@ -252,6 +270,7 @@ public class LoginInterceptorAspect {
                 if (loginResult instanceof JsonResult) {
                     if (((JsonResult) loginResult).isSuccess()) {
                         info.setIsLogin(Boolean.TRUE);
+                        sendEvent(UserLoginRegEvent.SubEventType.user_only_login, device, phone, deviceId, ip, loginResult);
                     } else {
                         info.setIsLogin(Boolean.FALSE);
                         info.setLoginFailMsg(((JsonResult) loginResult)
@@ -278,6 +297,59 @@ public class LoginInterceptorAspect {
     private boolean isWhite(String key, KeyType type) {
         return whiteListRepository.countByKeyEqualsAndTypeEqualsAndEnableIsTrue(key, type) > 0;
     }
+
+    /**
+     * 发送登录事件
+     * @param type
+     * @param device
+     * @param phone
+     * @param deviceId
+     * @param ip
+     * @param loginResult
+     */
+    private void sendEvent(UserLoginRegEvent.SubEventType type,Device device, String phone, String deviceId, String ip, Object loginResult) {
+
+        try {
+            UserLoginRegEvent.UserLoginRegEventBuilder builder = UserLoginRegEvent.builder();
+
+            builder.maskPhoneNo(phone);
+            builder.hashPhoneNo(Md5Util.build(Md5Util.build(phone)));
+            builder.subEventType(type);
+
+            BusinessEventBaseInfo.BusinessEventBaseInfoBuilder baseInfoBuilder
+                    = BusinessEventBaseInfo.builder();
+            if (null != loginResult) {
+                if (loginResult instanceof JsonResult) {
+                    JsonResult jsonResult = ((JsonResult) loginResult);
+                    if (jsonResult.isSuccess()) {
+                        Object data = jsonResult.getData();
+                        if (data instanceof AuthBean) {
+
+                            String uuid = ((AuthBean) data).getUuid();
+                            User user = iUserService.findByUuidWithCache(uuid);
+                            baseInfoBuilder.userUuid(uuid);
+                            baseInfoBuilder.channel(String.valueOf(user.getRegisteredFrom()));
+                        } else {
+                            return;
+                        }
+
+                    } else {
+                        return;
+                    }
+                }
+            }
+
+            baseInfoBuilder.deviceId(deviceId);
+            baseInfoBuilder.ip(ip);
+            baseInfoBuilder.deviceType(device.getCode());
+            builder.businessEventBaseInfo(baseInfoBuilder.build());
+
+            UserEventCollectingUtil.addEvent(builder.build());
+        } catch (Exception e) {
+            log.error("发送用户登录事件异常phone:{}", phone);
+        }
+
+    }
     /**
      * 【登录设备反欺诈策略】
      *

src/main/java/cn/quantgroup/xyqb/event/RiskRegisteredEventListener.java

+package cn.quantgroup.xyqb.event;
+
+import cn.qg.ec.model.base.BusinessEventBaseInfo;
+import cn.qg.ec.model.user.UserLoginRegEvent;
+import cn.quantgroup.xyqb.entity.User;
+import cn.quantgroup.xyqb.model.UserRegisterParam;
+import cn.quantgroup.xyqb.risk.uereventcollecting.UserEventCollectingUtil;
+import cn.quantgroup.xyqb.util.encrypt.Md5Util;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.ApplicationListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author ：dongjianhua
+ * @date ：Created in 2020/12/23 18:07
+ * @description：风控注册事件
+ * @modified By：
+ * @version:
+ */
+@Slf4j
+@Component
+public class RiskRegisteredEventListener implements ApplicationListener<RegisterEvent> {
+    @Override
+    public void onApplicationEvent(RegisterEvent registerEvent) {
+        UserRegisterParam param =
+                registerEvent.getUserRegisterParam();
+        User user = param.getUser();
+        //拿不到user对象. 就是没注册了
+        if (user == null) {
+            return;
+        }
+        UserLoginRegEvent.UserLoginRegEventBuilder builder = UserLoginRegEvent.builder();
+
+        builder.maskPhoneNo(user.getPhoneNo());
+        builder.hashPhoneNo(Md5Util.build(Md5Util.build(user.getPhoneNo())));
+        builder.subEventType(UserLoginRegEvent.SubEventType.user_only_login);
+
+        BusinessEventBaseInfo.BusinessEventBaseInfoBuilder baseInfoBuilder
+                = BusinessEventBaseInfo.builder();
+
+        baseInfoBuilder.userUuid(user.getUuid());
+        baseInfoBuilder.channel(String.valueOf(user.getRegisteredFrom()));
+
+        builder.businessEventBaseInfo(baseInfoBuilder.build());
+
+        UserEventCollectingUtil.addEvent(builder.build());
+    }
+}

src/main/java/cn/quantgroup/xyqb/event/UserExtInfoSaveEventListener.java

 package cn.quantgroup.xyqb.event;
 
 import cn.qg.ec.model.base.BusinessEventBaseInfo;
-import cn.qg.ec.model.base.EventInfo;
 import cn.qg.ec.model.user.UserBaseInfoEvent;
 import cn.quantgroup.user.enums.EducationEnum;
 import cn.quantgroup.user.enums.IncomeRangeEnum;
@@ -49,21 +48,21 @@ public class UserExtInfoSaveEventListener implements ApplicationListener<UserExt
         }
         String uuid = user.getUuid();
         if (null != educationEnum) {
-            send(UserBaseInfoEvent.SubEventType.user_education, uuid, String.valueOf(educationEnum.ordinal()));
+            send(UserBaseInfoEvent.SubEventType.user_education, user, String.valueOf(educationEnum.ordinal()));
         }
         if (null != incomeRangeEnum) {
-            send(UserBaseInfoEvent.SubEventType.user_income, uuid, String.valueOf(incomeRangeEnum.ordinal()));
+            send(UserBaseInfoEvent.SubEventType.user_income, user, String.valueOf(incomeRangeEnum.ordinal()));
         }
         if (null != marryStatus) {
-            send(UserBaseInfoEvent.SubEventType.user_marital_status, uuid, String.valueOf(marryStatus.ordinal()));
+            send(UserBaseInfoEvent.SubEventType.user_marital_status, user, String.valueOf(marryStatus.ordinal()));
         }
         if (null != occupationEnum) {
-            send(UserBaseInfoEvent.SubEventType.user_occupation, uuid, String.valueOf(occupationEnum.ordinal()));
+            send(UserBaseInfoEvent.SubEventType.user_occupation, user, String.valueOf(occupationEnum.ordinal()));
         }
         log.info("用户保存扩展信息结束userExtInfo:{}", userExtInfo);
     }
 
-    private void send(UserBaseInfoEvent.SubEventType eventType, String uuid, String val) {
+    private void send(UserBaseInfoEvent.SubEventType eventType, User user , String val) {
 
         UserBaseInfoEvent.UserBaseInfoEventBuilder builder
                 = UserBaseInfoEvent.builder();
@@ -71,8 +70,8 @@ public class UserExtInfoSaveEventListener implements ApplicationListener<UserExt
         BusinessEventBaseInfo.BusinessEventBaseInfoBuilder baseInfoBuilder
                 = BusinessEventBaseInfo.builder();
 
-        baseInfoBuilder.userId(uuid);
-
+        baseInfoBuilder.userUuid(user.getUuid());
+        baseInfoBuilder.channel(String.valueOf(user.getRegisteredFrom()));
         builder.subEventType(eventType);
 
         switch (eventType) {

src/main/java/cn/quantgroup/xyqb/risk/uereventcollecting/UserEventCollectingUtil.java

@@ -2,7 +2,6 @@ package cn.quantgroup.xyqb.risk.uereventcollecting;
 
 import cn.qg.ec.kinesis.KinesisProducerClient;
 import cn.qg.ec.model.base.EventInfo;
-import cn.qg.ec.model.user.UserLoginRegEvent;
 import com.alibaba.fastjson.JSON;
 import com.amazonaws.services.kinesis.producer.UserRecordResult;
 import com.google.common.util.concurrent.FutureCallback;
@@ -23,13 +22,13 @@ import java.io.UnsupportedEncodingException;
 @Slf4j
 public class UserEventCollectingUtil {
 
-    private static final MyCallBack CALL_BACK = new MyCallBack();
 
     public static void addEvent(EventInfo eventInfo) {
         KinesisProducerClient client = KinesisProducerClientFactory.getClient();
         try {
+            MyCallBack<UserRecordResult> CallBack = new MyCallBack<>(eventInfo);
             ListenableFuture<UserRecordResult> future = client.SendRecord(eventInfo);
-            Futures.addCallback(future, CALL_BACK);
+            Futures.addCallback(future, CallBack);
         } catch (UnsupportedEncodingException e) {
             log.error("发送事件异常UnsupportedEncodingException:{}", e);
         } catch (Exception e) {
@@ -40,9 +39,15 @@ public class UserEventCollectingUtil {
     @Slf4j
     public static class MyCallBack<UserRecordResult> implements FutureCallback<UserRecordResult> {
 
+        private EventInfo eventInfo;
+
+        public MyCallBack(EventInfo eventInfo) {
+            this.eventInfo = eventInfo;
+        }
+
         @Override
         public void onSuccess(@Nullable UserRecordResult userRecordResult) {
-            log.info("发送事件成功 o:{}", JSON.toJSONString(userRecordResult));
+            log.info("发送事件成功 o:{},eventInfo:{}", JSON.toJSONString(userRecordResult), eventInfo);
         }
 
         @Override