store归因上报

src/main/java/com/lwby/marketing/att/NovelAttributionMain.java

@@ -18,14 +18,12 @@ public class NovelAttributionMain {
     ApplicationContext ctx;
 
     public FlowExecutor<NovelAction>      executorNovel;
-    public FlowExecutor<StoryNovelAction> executorStoreNovelActive;
-    public FlowExecutor<StoryNovelAction> executorStoreNovelPay;
+    public FlowExecutor<StoryNovelAction> executorStoryNovel;
 
     @PostConstruct
     public void init() {
         executorNovel = new FlowExecutor<>(ctx, Rule.create().THEN("setup").THEN("prize").THEN("store").THEN("checkerfirst").SWITCH("cross","plan","channel"));
-        executorStoreNovelActive = new FlowExecutor<>(ctx, Rule.create().THEN("setupStoreNovel").THEN("storeNovelStore").THEN("checkerNovelStorefirst").THEN("uploadcall"));
-        executorStoreNovelPay = new FlowExecutor<>(ctx, Rule.create().THEN("setupStoreNovel").THEN("storeNovelStore").THEN("checkerNovelStorefirst").THEN("payspdu").THEN("uploadcall"));
+        executorStoryNovel = new FlowExecutor<>(ctx, Rule.create().THEN("setupStoreNovel").THEN("storeNovelStore").THEN("checkerNovelStorefirst").SWITCH("payspdu").THEN("uploadcall"));
 
     }
 

src/main/java/com/lwby/marketing/att/storynovel/CallBackType.java

+package com.lwby.marketing.att.storynovel;
+
+import com.lwby.marketing.att.novel.AttributionStatus;
+
+/**
+ * @author songxinyu
+ * @version CallBackType.java, v 0.1 2024年03月06日 17:32 songxinyu Exp $
+ */
+public enum CallBackType {
+
+    active(0,"ocpc_result", AttributionStatus.ACTIVE_CALLBACK),pay(2,"ocpc_pay",AttributionStatus.PAY);
+
+    private Integer type;
+
+    private String topic;
+
+    private AttributionStatus status;
+
+    private CallBackType(Integer type,String topic,AttributionStatus status) {
+        this.type = type;
+        this.topic = topic;
+        this.status = status;
+    }
+
+    public Integer getType() {
+        return type;
+    }
+
+    public void setType(Integer type) {
+        this.type = type;
+    }
+
+    public String getTopic() {
+        return topic;
+    }
+
+    public void setTopic(String topic) {
+        this.topic = topic;
+    }
+
+    public AttributionStatus getStatus() {
+        return status;
+    }
+
+    public void setStatus(AttributionStatus status) {
+        this.status = status;
+    }
+
+    public static CallBackType getCallBackTypeByType(Integer type) {
+        for (CallBackType callBackType:CallBackType.values()) {
+            if (callBackType.getType().equals(type)) {
+                return callBackType;
+            }
+        }
+        return null;
+    }
+}

src/main/java/com/lwby/marketing/att/storynovel/StoryNovelActiveConsumer.java

@@ -29,21 +29,21 @@ public class StoryNovelActiveConsumer implements MessageListener<String, String>
 
         long begin = System.currentTimeMillis();
         if (log.isInfoEnabled()) {
-            log.info("media.active.onMessage start, data={}", data == null ? null : data.value());
+            log.info("media.active.story.onMessage start, data={}", data == null ? null : data.value());
         }
         if (data != null) {
             try {
                 StoreUserUploadEventBO event = JSONObject.parseObject(data.value(), StoreUserUploadEventBO.class);
 
                 if (log.isInfoEnabled()) {
-                    log.info("media.active.topic={}, bookStoreEvent={}", data.topic(), JSON.toJSONString(event));
+                    log.info("media.active.story.topic={}, bookStoreEvent={}", data.topic(), JSON.toJSONString(event));
                 }
 
                 StoryNovelAction action = new StoryNovelAction(event.getClientInfo(),event.getUserId(),event.getOpenId(),event.getUserUploadEvent());
 
-                novelAttributionMain.executorStoreNovelActive.execute(action);
+                novelAttributionMain.executorStoryNovel.execute(action);
             } catch (Throwable e) {
-                log.error("novel.active.onMessage failed, data={}, costTime={} ms", data.value(),
+                log.error("story.novel.active.onMessage failed, data={}, costTime={} ms", data.value(),
                         System.currentTimeMillis() - begin, e);
             }
         }

src/main/java/com/lwby/marketing/att/storynovel/StoryNovelPayConsumer.java

@@ -29,21 +29,21 @@ public class StoryNovelPayConsumer implements MessageListener<String, String> {
 
         long begin = System.currentTimeMillis();
         if (log.isInfoEnabled()) {
-            log.info("media.active.onMessage start, data={}", data == null ? null : data.value());
+            log.info("media.pay.story.onMessage start, data={}", data == null ? null : data.value());
         }
         if (data != null) {
             try {
                 StoreUserUploadEventBO event = JSONObject.parseObject(data.value(), StoreUserUploadEventBO.class);
 
                 if (log.isInfoEnabled()) {
-                    log.info("media.active.topic={}, bookStoreEvent={}", data.topic(), JSON.toJSONString(event));
+                    log.info("media.pay.story.topic={}, bookStoreEvent={}", data.topic(), JSON.toJSONString(event));
                 }
 
                 StoryNovelAction action = new StoryNovelAction(event.getClientInfo(),event.getUserId(),event.getOpenId(),event.getUserUploadEvent(),event.getProductId());
 
-                novelAttributionMain.executorStoreNovelPay.execute(action);
+                novelAttributionMain.executorStoryNovel.execute(action);
             } catch (Throwable e) {
-                log.error("novel.active.onMessage failed, data={}, costTime={} ms", data.value(),
+                log.error("story.novel.pay.onMessage failed, data={}, costTime={} ms", data.value(),
                         System.currentTimeMillis() - begin, e);
             }
         }

src/main/java/com/lwby/marketing/att/storynovel/UniversalStoryProcess.java

@@ -11,6 +11,7 @@ import com.lwby.marketing.att.novel.AttributionType;
 import com.lwby.marketing.vo.DeliveryDeviceInfo;
 import com.lwby.marketing.vo.StoryNovelAction;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.dao.EmptyResultDataAccessException;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.kafka.core.KafkaTemplate;
@@ -41,7 +42,7 @@ public class UniversalStoryProcess {
     /**
      * 通知处理结果
      */
-    public void notifyResult(StoryNovelAction action, AttributionStatus status) {
+    public void notifyResult(StoryNovelAction action,String topic, AttributionStatus status) {
         DeliveryDeviceInfo ddi = action.getDeliveryDeviceInfo();
 
         if (Objects.isNull(ddi)) {
@@ -62,7 +63,7 @@ public class UniversalStoryProcess {
         ddi.setUserId(action.getUserId());
 
         String jsonString = JSONObject.toJSONString(ddi);
-        ListenableFuture<SendResult<String, String>> active_result = kafkaTemplate.send("ocpc_result", jsonString);
+        ListenableFuture<SendResult<String, String>> active_result = kafkaTemplate.send(topic, jsonString);
         active_result.addCallback(
                 result -> log.info("归因成功[{}],归因类型[{}]", jsonString, status.desc),
                 ex -> log.error("归因失败[{}],归因类型[{}]", jsonString, status.desc, ex)
@@ -78,7 +79,7 @@ public class UniversalStoryProcess {
     }
 
     public String getFirstCheckerKey(StoryNovelAction action) {
-        return Objects.equals(action.getType(),0)
+        return Objects.equals(action.getType(),CallBackType.active.getType())
                 ? String.format("fc_%d_%d_%s_%s", action.getUserId(), action.getPlatformId(),action.getMediaName(),action.getCurrentDateStr())
                 : String.format("fc_%d_%d_%s", action.getUserId(), action.getPlatformId(),action.getMediaName());
     }
@@ -118,7 +119,11 @@ public class UniversalStoryProcess {
     @Cached(name="attribute_rule_store_novel", cacheType = CacheType.LOCAL)
     @CacheRefresh(refresh = 300)
     public String getAttributeRuleByPlatformIdAndChannelId(int platformId,int channelId) {
+        try {
             return lwbyJdbcTemplate.queryForObject(String.format("select  shave_make_call from attribute_rule where channel_id=%d and platform_id=%d",channelId,platformId),String.class);
+        } catch (EmptyResultDataAccessException e) {
+            return null;
+        }
     }
 
 

src/main/java/com/lwby/marketing/att/storynovel/handle/ParameterSetupStoryNovelFlow.java

 package com.lwby.marketing.att.storynovel.handle;
 
+import com.alibaba.fastjson.JSONObject;
 import com.lwby.marketing.att.novel.UniversalProcess;
 import com.lwby.marketing.att.storynovel.media.StoryMedia;
 import com.lwby.marketing.flow.NodeFlow;
@@ -30,12 +31,16 @@ public class ParameterSetupStoryNovelFlow extends NodeFlow<StoryNovelAction> {
         //VO对像
         DeliveryDeviceInfo deliveryDeviceInfo = null;
 
+        //String s = "{\"creativeId\":\"112331\",\"creativeType\":\"3\",\"adid\":\"12321\","
+        //        + "\"clickId\":\"12312143232\",\"channel\":\"216011231\",\"bookId\":\"4322111\","
+        //        + "\"media\":\"jrtt\",\"clickTime\":123123123,\"code\":\"12ede3e231\"}";;
+        //StoryLogin storyLogin = JSONObject.parseObject(s,StoryLogin.class);
         //匹配OpenId
         StoryLogin storyLogin = up.get(StoryLogin.class,assembleKey(openId));
         action.setStoryLogin(storyLogin);
 
         //NovelAction对像参数填充
-        action.setPlatformId(clientInfo.getPlatformId());
+        action.setPlatformId(platformId);
         action.setUserId(action.getUserId());
         action.setCurrentDateStr(DateTimUtils.getCurrentDateString());
 
@@ -51,6 +56,7 @@ public class ParameterSetupStoryNovelFlow extends NodeFlow<StoryNovelAction> {
             deliveryDeviceInfo.setUuid(UUID.randomUUID().toString());
             deliveryDeviceInfo.setDj_channel(storyLogin.getChannel());
             deliveryDeviceInfo.setPromotion_id(storyLogin.getAdid());
+            action.setDeliveryDeviceInfo(deliveryDeviceInfo);
             action.setChannelId(NumberUtils.parseInteger(storyLogin.getChannel()));
             action.setPlanId(NumberUtils.parseInteger((storyLogin.getAdid())));
             action.setMediaName(storyLogin.getMedia());

src/main/java/com/lwby/marketing/att/storynovel/handle/PaySpduFlow.java

@@ -5,6 +5,7 @@ import com.lwby.marketing.att.novel.AttributionStatus;
 import com.lwby.marketing.att.novel.AttributionType;
 import com.lwby.marketing.att.storynovel.UniversalStoryProcess;
 import com.lwby.marketing.flow.NodeFlow;
+import com.lwby.marketing.flow.NodeSwitchFlow;
 import com.lwby.marketing.vo.StoryNovelAction;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
@@ -22,7 +23,7 @@ import java.util.concurrent.ThreadLocalRandom;
  */
 @Slf4j
 @Component("payspdu")
-public class PaySpduFlow extends NodeFlow<StoryNovelAction> {
+public class PaySpduFlow extends NodeSwitchFlow<StoryNovelAction> {
 
     @Resource
     UniversalStoryProcess up;
@@ -32,6 +33,11 @@ public class PaySpduFlow extends NodeFlow<StoryNovelAction> {
         process0(action, AttributionType.CHANNEL);
     }
 
+    @Override
+    public boolean checked(StoryNovelAction action) {
+        return action.getType() == 2;
+    }
+
     public void process0(StoryNovelAction action, AttributionType type) {
 
         int platformId = action.getPlatformId();
@@ -88,7 +94,7 @@ public class PaySpduFlow extends NodeFlow<StoryNovelAction> {
                     "PaySpduFlow.process0.deduction doing dynamic, platformId = {}, channel = {}, sprDedu = {}, goodId = {}, channelTotalCount = {}, channelCallbackCount = {}, v = {}",
                     platformId, channelId, sprDedu, goodId, channelTotalCount, channelCallbackCount, 1);
         } else {
-            up.notifyResult(action, AttributionStatus.NORMAL_DEDUCTION_CALLBACK);
+            up.notifyResult(action,"ocpc_pay", AttributionStatus.NORMAL_DEDUCTION_CALLBACK);
             up.set(up.getFirstCheckerKey(action),60 * 60 * 24,"1");
             log.info(
                     "PaySpduFlow.process0.deduction doing dynamic, platformId = {}, channel = {}, sprDedu = {}, goodId = {}, channelTotalCount = {}, channelCallbackCount = {}, v = {}",

src/main/java/com/lwby/marketing/att/storynovel/handle/StoreNovelStoryAttributionFlow.java

 package com.lwby.marketing.att.storynovel.handle;
 
 import com.lwby.marketing.att.novel.AttributionStatus;
+import com.lwby.marketing.att.storynovel.CallBackType;
 import com.lwby.marketing.att.storynovel.UniversalStoryProcess;
 import com.lwby.marketing.flow.NodeFlow;
 import com.lwby.marketing.vo.StoryNovelAction;
@@ -18,7 +19,9 @@ public class StoreNovelStoryAttributionFlow extends NodeFlow<StoryNovelAction> {
     public void process(StoryNovelAction action) {
         //商店归因通知
         if(Objects.isNull(action.getDeliveryDeviceInfo())){
-            up.notifyResult(action, AttributionStatus.STORE_CALLBACK);
+            if (action.getType().equals(CallBackType.active.getType())) {
+                up.notifyResult(action,"ocpc_result", AttributionStatus.STORE_CALLBACK);
+            }
             action.stop(true);   //结束后面所有执行流程
         }
 

src/main/java/com/lwby/marketing/att/storynovel/handle/UploadCallFlow.java

 package com.lwby.marketing.att.storynovel.handle;
 
 import com.lwby.marketing.att.novel.AttributionStatus;
+import com.lwby.marketing.att.storynovel.CallBackType;
 import com.lwby.marketing.att.storynovel.UniversalStoryProcess;
 import com.lwby.marketing.flow.NodeFlow;
 import com.lwby.marketing.vo.StoryNovelAction;
@@ -20,9 +21,16 @@ public class UploadCallFlow extends NodeFlow<StoryNovelAction> {
 
     @Override
     public void process(StoryNovelAction action) {
-        action.getMedia().callback(action);
-        up.notifyResult(action, AttributionStatus.ACTIVE_CALLBACK);
-        up.set(up.getFirstCheckerKey(action),60 * 60 * 24,"1"); //每天扣量过一次，不在重复走
-        return;
+        process0(action, CallBackType.getCallBackTypeByType(action.getType()));
+
+    }
+
+    private void process0(StoryNovelAction action, CallBackType callBackTypeByType) {
+        boolean success = action.getMedia().callback(action);
+        if (success) {
+            up.notifyResult(action, callBackTypeByType.getTopic(),
+                    callBackTypeByType.getStatus());
+            up.set(up.getFirstCheckerKey(action),60 * 60 * 24,"1");
+        }
     }
 }

src/main/java/com/lwby/marketing/att/storynovel/media/StoryMedia.java

@@ -10,7 +10,8 @@ import java.util.function.Function;
 import java.util.stream.Collectors;
 
 public enum StoryMedia {
-    JRTT("jrtt","今日头条",(c)-> c == 0 ? new JRTTActiveCallback() : new JRTTPayCallback());
+
+    jrtt("jrtt","今日头条",(c)-> c == 0 ? new JRTTActiveCallback() : new JRTTPayCallback());
 
     final String name;
     final String                            desc;

src/main/java/com/lwby/marketing/att/storynovel/media/jrtt/JRTTActiveCallback.java

@@ -7,12 +7,13 @@ import com.lwby.marketing.att.novel.media.jrtt.dto.JrttAttributeRequest;
 import com.lwby.marketing.att.storynovel.media.BaseStoryCallback;
 import com.lwby.marketing.vo.DeliveryDeviceInfo;
 import com.lwby.marketing.vo.StoryNovelAction;
-
+import lombok.extern.slf4j.Slf4j;
 
 /**
  * @author songxinyu
  * @version JRTTActiveCallback.java, v 0.1 2024年03月04日 17:24 songxinyu Exp $
  */
+@Slf4j
 public class JRTTActiveCallback extends BaseStoryCallback {
 
     public static final String ATTRIBUTE_URL = "https://analytics.oceanengine.com/api/v2/conversion";
@@ -34,6 +35,7 @@ public class JRTTActiveCallback extends BaseStoryCallback {
         try {
             String result = HttpUtil.post(ATTRIBUTE_URL, userJson);
             Integer resultCode = (Integer) JSON.parseObject(result).get("code");
+            log.info("JRTTActiveCallback.story.active.upload,result={},userId={},channel={},platform={}",JSON.toJSONString(result),na.getUserId(),na.getChannelId(),na.getPlatformId());
             return resultCode == 0;
         } catch (Exception e) {
             return false;

src/main/java/com/lwby/marketing/att/storynovel/media/jrtt/JRTTPayCallback.java

@@ -7,11 +7,13 @@ import com.lwby.marketing.att.novel.media.jrtt.dto.JrttAttributeRequest;
 import com.lwby.marketing.att.storynovel.media.BaseStoryCallback;
 import com.lwby.marketing.vo.DeliveryDeviceInfo;
 import com.lwby.marketing.vo.StoryNovelAction;
+import lombok.extern.slf4j.Slf4j;
 
 /**
  * @author songxinyu
  * @version JRTTPayCallback.java, v 0.1 2024年03月04日 17:25 songxinyu Exp $
  */
+@Slf4j
 public class JRTTPayCallback extends BaseStoryCallback {
 
     public static final String ATTRIBUTE_URL = "https://analytics.oceanengine.com/api/v2/conversion";
@@ -33,6 +35,7 @@ public class JRTTPayCallback extends BaseStoryCallback {
         try {
             String result = HttpUtil.post(ATTRIBUTE_URL, userJson);
             Integer resultCode = (Integer) JSON.parseObject(result).get("code");
+            log.info("JRTTActiveCallback.story.pay.upload,result={},userId={},channel={},platform={}",JSON.toJSONString(result),na.getUserId(),na.getChannelId(),na.getPlatformId());
             return resultCode == 0;
         } catch (Exception e) {
             return false;

src/main/java/com/lwby/marketing/controller/AttrController.java

@@ -4,6 +4,9 @@ import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson2.JSONObject;
 import com.lwby.marketing.att.NovelAttributionMain;
 import com.lwby.marketing.att.storynovel.UniversalStoryProcess;
+import com.lwby.marketing.att.storynovel.media.IStoryCallback;
+import com.lwby.marketing.att.storynovel.media.jrtt.JRTTActiveCallback;
+import com.lwby.marketing.att.storynovel.media.jrtt.JRTTPayCallback;
 import com.lwby.marketing.vo.StoryNovelAction;
 import com.lwby.marketing.vo.StoreUserUploadEventBO;
 import org.apache.commons.lang3.StringUtils;
@@ -13,6 +16,7 @@ import org.springframework.web.bind.annotation.RestController;
 
 import javax.annotation.Resource;
 import java.util.Map;
+import java.util.function.Function;
 
 /**
  * @author songxinyu
@@ -33,14 +37,15 @@ public class AttrController {
                 + " wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/117.0.0.0 Mobile Safari/537.36 aweme.lite/28.9.0 "
                 + "ToutiaoMicroApp/3.14.0 PluginVersion/28909073\",\"version\":\"2.0.0\",\"xClient\":\"version=2.0.0;platformId=412;"
                 + "equipment=android\"},\"createTime\":1709533439693,\"extraData\":{},\"id\":\"93857260-8130-442f-bc92-b549dbf28ef0\","
-                + "\"userId\":875325,\"userUploadEvent\":0,\"appId\":\"ttd3dda5604ce230b401\","
+                + "\"userId\":875325,\"userUploadEvent\":2,\"productId\":\"1\",\"appId\":\"ttd3dda5604ce230b401\","
                 + "\"openId\":\"_000HgDjWl-P5-WS9HTQIssNgTnMBRUqDHDu\"}";
 
         StoreUserUploadEventBO event = JSONObject.parseObject(msg, StoreUserUploadEventBO.class);
-        StoryNovelAction action = new StoryNovelAction(event.getClientInfo(),event.getUserId(),event.getOpenId(),event.getUserUploadEvent());
+        StoryNovelAction action = new StoryNovelAction(event.getClientInfo(),event.getUserId(),event.getOpenId(),event.getUserUploadEvent(),
+                event.getProductId());
 
         try {
-            nm.executorStoreNovelActive.execute(action);
+            nm.executorStoryNovel.execute(action);
         } catch (Exception e) {
             throw new RuntimeException(e);
         }
@@ -83,5 +88,12 @@ public class AttrController {
         }
 
         System.out.println(sprDedu);
+
+        Function<Integer, IStoryCallback> function = new Function<Integer, IStoryCallback>() {
+            @Override
+            public IStoryCallback apply(Integer c) {
+                return c == 0 ? new JRTTActiveCallback() : new JRTTPayCallback();
+            }
+        };
     }
 }

src/main/java/com/lwby/marketing/po/StoryUserProfile.java

+package com.lwby.marketing.po;
+
+import lombok.Data;
+
+import java.io.Serializable;
+import java.util.Date;
+
+@Data
+public class StoryUserProfile implements Serializable {
+	private static final long serialVersionUID = 1L;
+
+	/**
+	 * 主键ID
+	 */
+	private Long id;
+
+	/**
+	 * 昵称
+	 */
+	private String nickname;
+
+	/**
+	 * 头像
+	 */
+	private String avatarUrl;
+
+	/**
+	 * 手机号
+	 */
+	private String mobile;
+
+	/**
+	 * 金币时数
+	 */
+	private Integer goldCoinNum;
+
+	/**
+	 * 赠送金币数
+	 */
+	private Integer giftGoldCoinNum;
+
+	/**
+	 * 用户状态
+	 */
+	private Integer userStatus;
+
+	/**
+	 * 平台
+	 */
+	private Integer platform;
+
+	/**
+	 * 创建时间
+	 */
+	private Date createDate;
+
+	/**
+	 * 渠道
+	 */
+	private String channelId;
+
+	/**
+	 * 会员到期日
+	 */
+	private Date vipEndDate;
+
+}
\ No newline at end of file

src/main/java/com/lwby/marketing/vo/ClientInfo.java

 package com.lwby.marketing.vo;
 
+import com.lwby.marketing.po.StoryUserProfile;
 import lombok.Data;
 
 @Data
 public class ClientInfo {
     private String sessionid;
     private UserProfile      user;
+    private StoryUserProfile storyUserProfile;
     private String           xClient;
     private String imei;
     private String imsi;

src/main/resources/application-prod.yml

@@ -33,18 +33,18 @@ spring:
     listener:
       ack-mode: RECORD
 
-store:
+story:
   novel:
     active:
       consumer:
-        topic: storeNovelActive
+        topic: storyNovelActive
         group:
-          id: store_dy_active
+          id: story_dy_active
     pay:
       consumer:
         topic: storeNovelpay
         group:
-          id: store_dy_pay
+          id: story_dy_pay
 
 
 novel:

src/main/resources/application-test.yml

@@ -21,7 +21,7 @@ spring:
     bootstrap-servers: 172.17.255.113:9092
     producer:
       retries: 3
-      acks: 1
+      acks: all
       key-serializer: org.apache.kafka.common.serialization.StringSerializer
       value-serializer: org.apache.kafka.common.serialization.StringSerializer
     consumer:
@@ -33,18 +33,18 @@ spring:
     listener:
       ack-mode: RECORD
 
-store:
+story:
   novel:
     active:
       consumer:
-        topic: testStoreNovelActive
+        topic: testStoryNovelActive
         group:
-          id: test_store_dy_active
+          id: test_story_dy_active
     pay:
       consumer:
-        topic: testStoreNovelpay
+        topic: testStoryNovelpay
         group:
-          id: test_store_dy_pay
+          id: test_story_dy_pay
 
 
 novel: