Merge branch 'feature/adv-callback'

fs-ad-new-api/src/main/java/com/fs/app/controller/LandingPageController.java

@@ -1,16 +1,22 @@
 package com.fs.app.controller;
 
 import com.fs.app.facade.CallbackProcessingFacadeService;
+import com.fs.common.core.domain.R;
+import com.fs.common.core.redis.RedisCache;
+import com.fs.common.exception.base.BusinessException;
 import com.fs.common.result.Result;
+import com.fs.common.service.ISmsService;
+import com.fs.newAdv.dto.req.FormSubmitReq;
 import com.fs.newAdv.dto.req.LandingIndexReq;
 import com.fs.newAdv.dto.req.WeChatLandingIndexReq;
 import com.fs.newAdv.dto.res.LandingIndexRes;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.bind.annotation.*;
+
+import javax.validation.Valid;
+import java.util.concurrent.TimeUnit;
 
 /**
  * 落地页控制器
@@ -26,6 +32,10 @@ public class LandingPageController {
 
     @Autowired
     private CallbackProcessingFacadeService facadeService;
+    @Autowired
+    private ISmsService smsService;
+    @Autowired
+    private RedisCache redisCache;
 
     /**
      * 落地页访问
@@ -34,7 +44,7 @@ public class LandingPageController {
     public Result<LandingIndexRes> h5Home(
             @RequestBody LandingIndexReq req) {
         // 查询落地页模板
-        return Result.success(facadeService.getLandingIndexBySiteId(req.getViewUrl(),req.getAllParams()));
+        return Result.success(facadeService.getLandingIndexBySiteId(req.getViewUrl(), req.getAllParams()));
     }
 
     /**
@@ -45,6 +55,34 @@ public class LandingPageController {
         return Result.success(facadeService.getWxLandingIndexBySiteId(req));
     }
 
+    /**
+     * 获取验证码
+     *
+     * @return
+     */
+    @GetMapping(value = "/sendSmsCode/{phone}")
+    public Result<String> sendSmsCode(@PathVariable String phone) {
+        String captcha = redisCache.getCacheObject("smsCode"+phone);
+        if (StringUtils.isNotEmpty(captcha)) {
+            throw new BusinessException("短信已发送,1分钟以内请勿重新发送验证码：");
+        }
+        int code = (int) (Math.random() * (9999 - 1000 + 1)) + 1000;// 产生1000-9999的随机数
+        redisCache.setCacheObject("smsCode"+phone,code, 2, TimeUnit.MINUTES);
+        R r = smsService.sendCaptcha(phone, code + "","验证码");
+        return Result.success(String.valueOf(r.get("msg")));
+    }
 
+    /**
+     * @return
+     */
+    @PostMapping(value = "/submit")
+    public Result<String> formSubmit(@RequestBody @Valid FormSubmitReq req) {
+        String captcha = redisCache.getCacheObject(req.getPhone());
+        if (StringUtils.isEmpty(captcha) || !captcha.equals(req.getSmsCode())) {
+            throw new BusinessException("短信验证码有误：" + req.getSmsCode());
+        }
+        facadeService.updateFromByTraceId(req);
+        return Result.success();
+    }
 }
 

fs-ad-new-api/src/main/java/com/fs/app/facade/CallbackProcessingFacadeService.java

@@ -1,10 +1,12 @@
 package com.fs.app.facade;
 
+import com.fs.newAdv.dto.req.FormSubmitReq;
 import com.fs.newAdv.dto.req.QwExternalIdBindTrackReq;
 import com.fs.newAdv.dto.req.WeChatLandingIndexReq;
 import com.fs.newAdv.dto.req.updateNickNameReq;
 import com.fs.newAdv.dto.res.LandingIndexRes;
 
+import javax.validation.Valid;
 import java.util.Map;
 
 public interface CallbackProcessingFacadeService {
@@ -33,4 +35,6 @@ public interface CallbackProcessingFacadeService {
     void qwExternalIdBindTrack(QwExternalIdBindTrackReq req);
 
     void updateNickName(updateNickNameReq req);
+
+    void updateFromByTraceId(@Valid FormSubmitReq req);
 }

fs-ad-new-api/src/main/java/com/fs/app/facade/CallbackProcessingFacadeServiceImpl.java

@@ -12,6 +12,7 @@ import com.fs.common.utils.SnowflakeUtil;
 import com.fs.newAdv.domain.LandingPageTemplate;
 import com.fs.newAdv.domain.Lead;
 import com.fs.newAdv.domain.Site;
+import com.fs.newAdv.dto.req.FormSubmitReq;
 import com.fs.newAdv.dto.req.QwExternalIdBindTrackReq;
 import com.fs.newAdv.dto.req.WeChatLandingIndexReq;
 import com.fs.newAdv.dto.req.updateNickNameReq;
@@ -146,16 +147,9 @@ public class CallbackProcessingFacadeServiceImpl implements CallbackProcessingFa
             lead.setAdvertiserId(advertiserId);
             lead.setSiteId(siteId);
             lead.setClickId(clickId);
+            lead.setTraceId(SnowflakeUtil.randomUUID());
             // 设置站点和落地页的关联
             setSiteByIdeaId(siteId, lead.getIdeaId());
-        } else {
-            // 检查站点和广告商信息是否异常
-            if (!Objects.equals(lead.getSiteId(), siteId)) {
-                log.info("落地页站点信息异常:{}---{}", lead.getSiteId(), siteId);
-            }
-            if (!Objects.equals(lead.getAdvertiserId(), advertiserId)) {
-                log.info("落地页广告商信息异常:{}---{}", lead.getAdvertiserId(), advertiserId);
-            }
         }
         // 模板缓存
 /*        Object ca = redisUtil.get(TEMPLATE_DATA + traceId);
@@ -186,7 +180,6 @@ public class CallbackProcessingFacadeServiceImpl implements CallbackProcessingFa
         lead.setUpdateTime(now);
         lead.setViewUrl(viewUrl);
         if (isNewLead) {
-            lead.setTraceId(SnowflakeUtil.randomUUID());
             leadService.save(lead);
         } else {
             leadService.updateById(lead);
@@ -221,12 +214,29 @@ public class CallbackProcessingFacadeServiceImpl implements CallbackProcessingFa
                 .filter(ObjectUtil::isNotEmpty)
                 .flatMap(Collection::stream)
                 .map(module -> (JSONObject) module)
-                .filter(module -> "h5-qrcode".equals(module.getStr("type")))
                 .forEach(module -> {
-                    if (StrUtil.isEmpty(qrCode.get())) {
-                        qrCode.set(getQrCodeByAllocationRuleId(site.getLaunchType(), site.getAllocationRule(), site.getAllocationRuleId(), lead));
+                    String type = module.getStr("type");
+
+                    switch (type) {
+                        case "h5-qrcode":
+                            if (StrUtil.isEmpty(qrCode.get())) {
+                                qrCode.set(getQrCodeByAllocationRuleId(
+                                        site.getLaunchType(),
+                                        site.getAllocationRule(),
+                                        site.getAllocationRuleId(),
+                                        lead));
+                            }
+                            module.set("workUrl", qrCode.get());
+                            break;
+
+                        case "h5-customer-link-button":
+                            module.set("workUrl", module.getStr("workUrl") + "?customer_channel=" + lead.getTraceId());
+                            log.info("更新获客链接参数: {}", qrCode.get());
+                            break;
+
+                        default:
+                            break;
                     }
-                    module.set("workUrl", qrCode.get());
                 });
     }
 
@@ -316,4 +326,11 @@ public class CallbackProcessingFacadeServiceImpl implements CallbackProcessingFa
         update.setWeiChatName(req.getNickName());
         leadService.updateById(update);
     }
+
+    @Override
+    public void updateFromByTraceId(FormSubmitReq req) {
+        leadService.update(new LambdaUpdateWrapper<Lead>()
+                .eq(Lead::getTraceId, req.getTraceId())
+                .set(Lead::getPhone, req.getPhone()));
+    }
 }

fs-live-app/src/main/java/com/fs/live/websocket/service/WebSocketServer.java

@@ -25,6 +25,7 @@ import com.fs.common.utils.spring.SpringUtils;
 import com.fs.live.domain.*;
 import com.fs.live.service.*;
 import com.fs.live.vo.LiveGoodsVo;
+import com.fs.newAdv.service.ILeadService;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.time.DateUtils;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -71,7 +72,7 @@ public class WebSocketServer {
     private final static long HEARTBEAT_TIMEOUT = 2 * 60 * 1000;
     // admin房间消息发送线程池（单线程，保证串行化）
     private final static ConcurrentHashMap<Long, ExecutorService> adminExecutors = new ConcurrentHashMap<>();
-    
+
     // 消息队列系统
     // 每个直播间的消息队列，使用优先级队列支持管理员消息插队
     private final static ConcurrentHashMap<Long, PriorityBlockingQueue<QueueMessage>> messageQueues = new ConcurrentHashMap<>();
@@ -104,6 +105,7 @@ public class WebSocketServer {
     private final ILiveWatchLogService liveWatchLogService = SpringUtils.getBean(ILiveWatchLogService.class);
     private final ILiveVideoService liveVideoService = SpringUtils.getBean(ILiveVideoService.class);
     private final ILiveCompletionPointsRecordService completionPointsRecordService = SpringUtils.getBean(ILiveCompletionPointsRecordService.class);
+    private final ILeadService leadService = SpringUtils.getBean(ILeadService.class);
     private static Random random = new Random();
 
     // Redis key 前缀：用户进入直播间时间
@@ -292,6 +294,8 @@ public class WebSocketServer {
                     liveUserFirstEntry.setExternalContactId(externalContactId);
                 }
                 liveUserFirstEntryService.insertLiveUserFirstEntry(liveUserFirstEntry);
+                // 第一次进入直播间 发送广告线索
+                leadService.enterLive(userId, liveId);
             }
             redisCache.setCacheObject( "live:user:first:entry:" + liveId + ":" + userId, liveUserFirstEntry, 4, TimeUnit.HOURS);
 
@@ -309,7 +313,7 @@ public class WebSocketServer {
         sessionLocks.putIfAbsent(session.getId(), new ReentrantLock());
         // 初始化心跳时间
         heartbeatCache.put(session.getId(), System.currentTimeMillis());
-        
+
         // 如果有session，启动消费者线程
         ConcurrentHashMap<Long, Session> tempRoom = getRoom(liveId);
         List<Session> tempAdminRoom = getAdminRoom(liveId);
@@ -400,7 +404,7 @@ public class WebSocketServer {
         // 清理Session相关资源
         heartbeatCache.remove(session.getId());
         sessionLocks.remove(session.getId());
-        
+
         // 检查并清理空的直播间资源
         cleanupEmptyRoom(liveId);
     }
@@ -1621,7 +1625,7 @@ public class WebSocketServer {
     private void startConsumerThread(Long liveId) {
         consumerRunningFlags.computeIfAbsent(liveId, k -> new AtomicBoolean(false));
         AtomicBoolean runningFlag = consumerRunningFlags.get(liveId);
-        
+
         // 如果线程已经在运行，直接返回
         if (runningFlag.get()) {
             return;
@@ -1633,16 +1637,16 @@ public class WebSocketServer {
                 Thread consumerThread = new Thread(() -> {
                     PriorityBlockingQueue<QueueMessage> queue = getMessageQueue(liveId);
                     log.info("[消息队列] 启动消费者线程, liveId={}", liveId);
-                    
+
                     while (runningFlag.get()) {
                         try {
                             // 检查是否还有session，如果没有则退出
                             ConcurrentHashMap<Long, Session> room = rooms.get(liveId);
                             List<Session> adminRoom = adminRooms.get(liveId);
-                            
-                            boolean hasSession = (room != null && !room.isEmpty()) || 
+
+                            boolean hasSession = (room != null && !room.isEmpty()) ||
                                                 (adminRoom != null && !adminRoom.isEmpty());
-                            
+
                             if (!hasSession) {
                                 log.info("[消息队列] 直播间无session，停止消费者线程, liveId={}", liveId);
                                 break;
@@ -1667,13 +1671,13 @@ public class WebSocketServer {
                             log.error("[消息队列] 消费消息异常, liveId={}", liveId, e);
                         }
                     }
-                    
+
                     // 清理资源
                     runningFlag.set(false);
                     consumerThreads.remove(liveId);
                     log.info("[消息队列] 消费者线程已停止, liveId={}", liveId);
                 }, "MessageConsumer-" + liveId);
-                
+
                 consumerThread.setDaemon(true);
                 consumerThread.start();
                 consumerThreads.put(liveId, consumerThread);
@@ -1705,22 +1709,22 @@ public class WebSocketServer {
     private boolean enqueueMessage(Long liveId, String message, boolean isAdmin) {
         PriorityBlockingQueue<QueueMessage> queue = getMessageQueue(liveId);
         AtomicLong currentSize = queueSizes.computeIfAbsent(liveId, k -> new AtomicLong(0));
-        
+
         // 计算新消息的大小
         long messageSize = message != null ? message.getBytes(StandardCharsets.UTF_8).length : 0;
-        
+
         // 检查队列条数限制
         if (!isAdmin && queue.size() >= MAX_QUEUE_SIZE) {
             log.warn("[消息队列] 队列条数已满，丢弃消息, liveId={}, queueSize={}", liveId, queue.size());
             return false;
         }
-        
+
         // 检查队列大小限制（200MB）
         long newTotalSize = currentSize.get() + messageSize;
         if (newTotalSize > MAX_QUEUE_SIZE_BYTES) {
             if (!isAdmin) {
                 // 普通消息超过大小限制，直接丢弃
-                log.warn("[消息队列] 队列大小超过限制，丢弃普通消息, liveId={}, currentSize={}MB, messageSize={}KB", 
+                log.warn("[消息队列] 队列大小超过限制，丢弃普通消息, liveId={}, currentSize={}MB, messageSize={}KB",
                         liveId, currentSize.get() / (1024.0 * 1024.0), messageSize / 1024.0);
                 return false;
             } else {
@@ -1728,13 +1732,13 @@ public class WebSocketServer {
                 long needToFree = newTotalSize - MAX_QUEUE_SIZE_BYTES;
                 long freedSize = removeMessagesToFreeSpace(queue, currentSize, needToFree, true);
                 if (freedSize < needToFree) {
-                    log.warn("[消息队列] 无法释放足够空间，管理员消息可能无法入队, liveId={}, needToFree={}KB, freed={}KB", 
+                    log.warn("[消息队列] 无法释放足够空间，管理员消息可能无法入队, liveId={}, needToFree={}KB, freed={}KB",
                             liveId, needToFree / 1024.0, freedSize / 1024.0);
                     // 即使空间不足，也尝试入队（可能会超过限制，但管理员消息优先级高）
                 }
             }
         }
-        
+
         // 如果是管理员消息且队列条数已满，移除一个普通消息
         if (isAdmin && queue.size() >= MAX_QUEUE_SIZE) {
             // 由于是优先级队列，普通消息（priority=0）会在队列末尾
@@ -1758,21 +1762,21 @@ public class WebSocketServer {
                 log.warn("[消息队列] 队列条数已满且无普通消息可移除, liveId={}", liveId);
             }
         }
-        
+
         QueueMessage queueMessage = new QueueMessage(message, isAdmin);
         queue.offer(queueMessage);
         currentSize.addAndGet(messageSize);
-        
+
         // 如果有session，确保消费者线程在运行
         ConcurrentHashMap<Long, Session> room = rooms.get(liveId);
         List<Session> adminRoom = adminRooms.get(liveId);
-        boolean hasSession = (room != null && !room.isEmpty()) || 
+        boolean hasSession = (room != null && !room.isEmpty()) ||
                             (adminRoom != null && !adminRoom.isEmpty());
-        
+
         if (hasSession) {
             startConsumerThread(liveId);
         }
-        
+
         return true;
     }
 
@@ -1784,13 +1788,13 @@ public class WebSocketServer {
      * @param onlyRemoveNormal 是否只移除普通消息（true=只移除普通消息，false=可以移除任何消息）
      * @return 实际释放的空间（字节数）
      */
-    private long removeMessagesToFreeSpace(PriorityBlockingQueue<QueueMessage> queue, 
-                                          AtomicLong currentSize, 
-                                          long needToFree, 
+    private long removeMessagesToFreeSpace(PriorityBlockingQueue<QueueMessage> queue,
+                                          AtomicLong currentSize,
+                                          long needToFree,
                                           boolean onlyRemoveNormal) {
         long freedSize = 0;
         List<QueueMessage> toRemove = new ArrayList<>();
-        
+
         // 收集需要移除的消息（优先移除普通消息）
         Iterator<QueueMessage> iterator = queue.iterator();
         while (iterator.hasNext() && freedSize < needToFree) {
@@ -1800,7 +1804,7 @@ public class WebSocketServer {
                 freedSize += msg.getSizeBytes();
             }
         }
-        
+
         // 如果只移除普通消息但空间还不够，可以移除管理员消息
         if (onlyRemoveNormal && freedSize < needToFree) {
             iterator = queue.iterator();
@@ -1812,19 +1816,19 @@ public class WebSocketServer {
                 }
             }
         }
-        
+
         // 移除消息并更新大小
         for (QueueMessage msg : toRemove) {
             if (queue.remove(msg)) {
                 currentSize.addAndGet(-msg.getSizeBytes());
             }
         }
-        
+
         if (freedSize > 0) {
-            log.info("[消息队列] 释放队列空间, removedCount={}, freedSize={}KB", 
+            log.info("[消息队列] 释放队列空间, removedCount={}, freedSize={}KB",
                     toRemove.size(), freedSize / 1024.0);
         }
-        
+
         return freedSize;
     }
 
@@ -1841,10 +1845,10 @@ public class WebSocketServer {
     private void cleanupEmptyRoom(Long liveId) {
         ConcurrentHashMap<Long, Session> room = rooms.get(liveId);
         List<Session> adminRoom = adminRooms.get(liveId);
-        
-        boolean hasSession = (room != null && !room.isEmpty()) || 
+
+        boolean hasSession = (room != null && !room.isEmpty()) ||
                             (adminRoom != null && !adminRoom.isEmpty());
-        
+
         if (!hasSession) {
             // 停止消费者线程
             stopConsumerThread(liveId);

fs-qw-api/src/main/java/com/fs/app/service/QwDataCallbackService.java

@@ -206,6 +206,7 @@ public class QwDataCallbackService {
                             if(StateList.getLength() > 0) {
                                 State = StateList.item(0).getTextContent();
                             }
+                            log.info("添加企微State参数: {}",State);
                             String WelcomeCode =null;
                             NodeList WelcomeCodeList = root.getElementsByTagName("WelcomeCode");
                             if(WelcomeCodeList.getLength() > 0) {

fs-qw-api/src/main/java/com/fs/framework/config/SecurityConfig.java

@@ -108,7 +108,7 @@ public class SecurityConfig extends WebSecurityConfigurerAdapter
                         "/**/*.js",
                         "/profile/**"
                 ).permitAll()
-
+                .antMatchers("/zentao/**").denyAll() // 拒绝访问
                 .antMatchers("/**").anonymous()
                 .antMatchers("/msg/**").anonymous()
                 .antMatchers("/msg/**/**").anonymous()

fs-service/src/main/java/com/fs/newAdv/domain/Lead.java

@@ -127,6 +127,14 @@ public class Lead implements Serializable {
      * 发起进入小程序 1是 0否
      */
     private Integer miniLaunchIndexCount;
+    /**
+     * 进入直播间1是 0否
+     */
+    private Integer enterLive;
+    /**
+     * 是否看课 1是 0否
+     */
+    private Integer completeCourse;
     /**
      * /**
      * 创建时间

fs-service/src/main/java/com/fs/newAdv/dto/req/FormSubmitReq.java

@@ -0,0 +1,14 @@
+package com.fs.newAdv.dto.req;
+
+import lombok.Data;
+
+import javax.validation.constraints.NotBlank;
+
+@Data
+public class FormSubmitReq {
+    private String phone;
+    private String name;
+    @NotBlank
+    private String smsCode;
+    private String traceId;
+}

fs-service/src/main/java/com/fs/newAdv/enums/SystemEventTypeEnum.java

@@ -12,7 +12,10 @@ public enum SystemEventTypeEnum {
     BUY_ORDER("event5", "商品购买订单"),
     AUTH_TODAY_CREATE("event6", "微信授权且当日创建"),
     COMPLETE_CLASS_AND_GROUP_TODAY("event7", "直播完课且当日加群"),
-    COMPLETE_CLASS_AND_WEI_CHAT_TODAY("event8", "直播完课且当日加微");
+    COMPLETE_CLASS_AND_WEI_CHAT_TODAY("event8", "直播完课且当日加微"),
+    ENTER_LIVE("event9", "进入直播间"),
+    COMPLETE_COURSE("event10", "看课记录"),
+    ;
 
     private final String code;
     private final String description;

fs-service/src/main/java/com/fs/newAdv/service/ILeadService.java

@@ -37,6 +37,15 @@ public interface ILeadService extends IService<Lead> {
      */
     void updateAddMemberLead(String externalUserID,String userID,String corpId,String State);
 
+    /**
+     * 第一次进入直播间
+     */
+    void enterLive(Long userId, Long liveId);
+
+    /**
+     * 第一次完成看课
+     */
+    void completeCourse(Long userId, Long courseId);
     /**
      * 用户删除企业微信线索处理
      */

fs-service/src/main/java/com/fs/newAdv/service/impl/LeadServiceImpl.java

@@ -5,6 +5,8 @@ import cn.hutool.core.util.StrUtil;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.fs.his.domain.FsUser;
+import com.fs.his.service.IFsUserService;
 import com.fs.newAdv.domain.Lead;
 import com.fs.newAdv.enums.SystemEventTypeEnum;
 import com.fs.newAdv.event.ConversionEventPublisher;
@@ -39,6 +41,8 @@ public class LeadServiceImpl extends ServiceImpl<LeadMapper, Lead> implements IL
 
     @Autowired
     private QwExternalContactMapper qwExternalContactMapper;
+    @Autowired
+    private IFsUserService fsUserService;
 
     @Override
     public Lead getByTraceId(String traceId) {
@@ -85,7 +89,7 @@ public class LeadServiceImpl extends ServiceImpl<LeadMapper, Lead> implements IL
         Lead lead = this.getOne(last);
         if (lead != null) {
             lead.setChatId(chatId);
-            lead.setAddContactQw(1);
+            lead.setAddContactQwGroup(1);
             lead.setCorpId(corpId);
             this.updateById(lead);
             if (ObjectUtil.isNotEmpty(lead.getLandingPageTs()) && lead.getLandingPageTs().toLocalDate().isEqual(LocalDate.now())) {
@@ -108,17 +112,62 @@ public class LeadServiceImpl extends ServiceImpl<LeadMapper, Lead> implements IL
             return;
         }
         qwExternalContact.setUnionid(unionid);
-        this.updateAddMemberLead(qwExternalContact);
+        this.updateAddMemberLead(qwExternalContact, null);
     }
 
     @Override
+    @Async
     public void updateAddMemberLead(String externalUserID, String userID, String corpId, String State) {
         QwExternalContact qwExternalContact = qwExternalContactMapper.selectQwExternalByExternalIdAndCompanyIdToIdAndFs(externalUserID, userID, corpId);
         if (qwExternalContact == null) {
             log.info("外部联系人信息不存在:{} {} {}", externalUserID, userID, corpId);
             return;
         }
-        this.updateAddMemberLead(qwExternalContact);
+        this.updateAddMemberLead(qwExternalContact, State);
+    }
+
+    @Override
+    @Async
+    public void enterLive(Long userId, Long liveId) {
+        FsUser fsUser = fsUserService.selectFsUserById(userId);
+        if (fsUser == null || ObjectUtil.isEmpty(fsUser.getQwExtId())) {
+            return;
+        }
+        QwExternalContact qwExternalContact = qwExternalContactMapper.selectQwExternalContactById(fsUser.getQwExtId());
+        if (qwExternalContact == null || ObjectUtil.isEmpty(qwExternalContact.getTraceId())) {
+            return;
+        }
+        String traceId = qwExternalContact.getTraceId();
+        Lead lead = this.getByTraceId(traceId);
+        if (lead == null || lead.getEnterLive() == 1) {
+            log.info("用户进入直播间线索已经完成:{}", lead);
+            return;
+        }
+        lead.setEnterLive(1);
+        this.updateById(lead);
+        conversionEventPublisher.publishConversionEvent(traceId, SystemEventTypeEnum.ENTER_LIVE);
+    }
+
+    @Override
+    @Async
+    public void completeCourse(Long userId, Long course) {
+        FsUser fsUser = fsUserService.selectFsUserById(userId);
+        if (fsUser == null || ObjectUtil.isEmpty(fsUser.getQwExtId())) {
+            return;
+        }
+        QwExternalContact qwExternalContact = qwExternalContactMapper.selectQwExternalContactById(fsUser.getQwExtId());
+        if (qwExternalContact == null || ObjectUtil.isEmpty(qwExternalContact.getTraceId())) {
+            return;
+        }
+        String traceId = qwExternalContact.getTraceId();
+        Lead lead = this.getByTraceId(traceId);
+        if (lead == null || lead.getCompleteCourse() == 1) {
+            log.info("用户看课线索已经完成:{}", lead);
+            return;
+        }
+        lead.setCompleteCourse(1);
+        this.updateById(lead);
+        conversionEventPublisher.publishConversionEvent(traceId, SystemEventTypeEnum.COMPLETE_COURSE);
     }
 
     @Override
@@ -138,8 +187,8 @@ public class LeadServiceImpl extends ServiceImpl<LeadMapper, Lead> implements IL
 
     }
 
-    private void updateAddMemberLead(QwExternalContact qwExternalContact) {
-        log.info("用户加微线索信息:{}", qwExternalContact);
+    private void updateAddMemberLead(QwExternalContact qwExternalContact, String state) {
+        log.info("用户加微线索信息:{} {}", qwExternalContact,state);
         LambdaQueryWrapper<Lead> last = new LambdaQueryWrapper<Lead>();
         if (StrUtil.isNotEmpty(qwExternalContact.getUnionid())) {
             last.eq(Lead::getUnionid, qwExternalContact.getUnionid());
@@ -149,9 +198,10 @@ public class LeadServiceImpl extends ServiceImpl<LeadMapper, Lead> implements IL
         last.eq(Lead::getAddContactQwGroup, 0).last("LIMIT 1");
         // 末次归因逻辑
         Lead lead = this.getOne(last);
+        lead = lead == null ? this.getByTraceId(state) : lead;
         if (lead != null) {
             lead.setExternalId(qwExternalContact.getId());
-            lead.setAddContactQwGroup(1);
+            lead.setAddContactQw(1);
             this.updateById(lead);
             // 绑定企微用户线索关系
             QwExternalContact temp = new QwExternalContact();

fs-service/src/main/resources/application-dev.yml

@@ -201,15 +201,14 @@ spring:
 
 # RocketMQ配置
 rocketmq:
-    name-server: 127.0.0.1:9876
+    name-server: rmq-19op47padq.rocketmq.cq.public.tencenttdmq.com:8080
     producer:
-        group: event-feedback-producer
-        send-message-timeout: 3000
-        retry-times-when-send-failed: 2
-        retry-times-when-send-async-failed: 2
-        max-message-size: 4194304
-        compress-message-body-threshold: 4096
-        retry-next-server: true
+        group: conversion-tracking-group
+        access-key: ak19op47padq83a6882e2edb   # 替换为实际的 accessKey
+        secret-key: sk4ca8a0dbdac7ed75 # 替换为实际的 secretKey
+    consumer:
+        access-key: ak19op47padq83a6882e2edb  # 替换为实际的 accessKey
+        secret-key: sk4ca8a0dbdac7ed75 # 替换为实际的 secretKey
 custom:
     token: "1o62d3YxvdHd4LEUiltnu7sK"
     encoding-aes-key: "UJfTQ5qKTKlegjkXtp1YuzJzxeHlUKvq5GyFbERN1iU"