异步切库

fs-admin/src/main/java/com/fs/quartz/saas/TenantJobDispatcherJob.java

@@ -1,6 +1,7 @@
 package com.fs.quartz.saas;
 
 import com.alibaba.fastjson.JSONObject;
+import com.fs.common.core.domain.model.TenantPrincipal;
 import com.fs.common.enums.DataSourceType;
 import com.fs.common.utils.StringUtils;
 import com.fs.config.saas.ProjectConfig;
@@ -8,7 +9,6 @@ import com.fs.core.config.TenantConfigContext;
 import com.fs.framework.datasource.DynamicDataSourceContextHolder;
 import com.fs.framework.datasource.TenantDataSourceManager;
 import com.fs.quartz.domain.SysJob;
-import com.fs.quartz.domain.TenantPrincipal;
 import com.fs.quartz.mapper.SysJobMapper;
 import com.fs.quartz.util.CronUtils;
 import com.fs.quartz.util.JobInvokeUtil;

fs-quartz/src/main/java/com/fs/quartz/domain/TenantPrincipal.java → fs-common/src/main/java/com/fs/common/core/domain/model/TenantPrincipal.java

@@ -1,8 +1,11 @@
-package com.fs.quartz.domain;
+package com.fs.common.core.domain.model;
 
 import lombok.AllArgsConstructor;
 import lombok.Getter;
 
+/**
+ * 租户身份标识
+ */
 @Getter
 @AllArgsConstructor
 public class TenantPrincipal {

fs-company/src/main/java/com/fs/company/controller/qw/QwExternalContactController.java

@@ -688,9 +688,12 @@ public class QwExternalContactController extends BaseController
     public  TableDataInfo getMiniCustomer(QwFsUserParam qwFsUserParam){
         startPage();
         List<QwFsUserVO> list = fsUserService.selectQwFsUserListVO(qwFsUserParam);
-        if (list==null||list.size()==0){
-            qwFsUserParam.setPhone(encryptPhone(qwFsUserParam.getPhone()));
-            list = fsUserService.selectQwFsUserListVO(qwFsUserParam);
+        if (list==null|| list.isEmpty()){
+            String phone = qwFsUserParam.getPhone();
+            if (StringUtils.isNotBlank(phone)){
+                qwFsUserParam.setPhone(encryptPhone(phone));
+                list = fsUserService.selectQwFsUserListVO(qwFsUserParam);
+            }
         }
         if (list != null) {
             for (QwFsUserVO vo : list) {

fs-service/src/main/java/com/fs/qw/service/impl/AsyncChatSopService.java

@@ -1,8 +1,12 @@
 package com.fs.qw.service.impl;
 
+import com.alibaba.fastjson.JSON;
 import com.fs.common.utils.PubFun;
 import com.fs.common.utils.StringUtils;
 import com.fs.common.utils.date.DateUtil;
+import com.fs.common.utils.spring.SpringUtils;
+import com.fs.core.config.TenantConfigContext;
+import com.fs.common.core.domain.model.TenantPrincipal;
 import com.fs.qw.domain.QwGroupChat;
 import com.fs.qw.service.IQwGroupChatService;
 import com.fs.qw.service.IQwUserService;
@@ -14,12 +18,18 @@ import com.fs.sop.domain.SopUserLogs;
 import com.fs.sop.mapper.QwSopMapper;
 import com.fs.sop.mapper.QwSopTempMapper;
 import com.fs.sop.mapper.SopUserLogsMapper;
+import com.fs.system.domain.SysConfig;
+import com.fs.system.mapper.SysConfigMapper;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
+import org.springframework.security.core.context.SecurityContextHolder;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 
+import java.lang.reflect.Method;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -39,13 +49,42 @@ public class AsyncChatSopService {
     private final SopUserLogsMapper sopUserLogsMapper;
     private final IQwUserService qwUserService;
     private final IQwGroupChatService qwGroupChatService;
+    private final SysConfigMapper sysConfigMapper;
 
     /**
      * 立即执行SOP任务
      */
     @Async("scheduledExecutorService")
-    public void executeChatSopByIds(String[] ids) {
+    public void executeChatSopByIds(String[] ids, Long tenantId) {
+        boolean isSwitched = false;
         try {
+            // 1. 切换到租户数据源并加载配置
+            if (tenantId != null) {
+                try {
+                    Object manager = SpringUtils.getBean("tenantDataSourceManager");
+                    Method method = manager.getClass().getMethod("ensureSwitchByTenantId", Long.class);
+                    method.invoke(manager, tenantId);
+                    isSwitched = true;
+
+                    // 加载租户 projectConfig
+                    SysConfig cfg = sysConfigMapper.selectConfigByConfigKey("projectConfig");
+                    if (cfg != null && StringUtils.isNotBlank(cfg.getConfigValue())) {
+                        TenantConfigContext.set(JSON.parseObject(cfg.getConfigValue()));
+                    }
+
+                    // 设置租户到 SecurityContext
+                    SecurityContextHolder.getContext().setAuthentication(
+                            new UsernamePasswordAuthenticationToken(
+                                    new TenantPrincipal(tenantId),
+                                    null,
+                                    Collections.emptyList()
+                            )
+                    );
+                } catch (Exception e) {
+                    log.error("SOP异步任务切换租户数据源失败: tenantId={}", tenantId, e);
+                }
+            }
+
             List<ChatSopRuleTimeVO> ruleTimeVOList = qwSopMapper.executeSopChatByIds(ids);
             List<QwSopTemp> tempList = qwSopTempMapper.selectListByIds(PubFun.listToNewList(ruleTimeVOList, ChatSopRuleTimeVO::getTempId));
             Map<String, QwSopTemp> tempMap = PubFun.listToMapByGroupObject(tempList, QwSopTemp::getId);
@@ -64,8 +103,20 @@ public class AsyncChatSopService {
             });
             qwSopMapper.updateStatusQwSopById2(PubFun.listToNewList(ruleTimeVOList, ChatSopRuleTimeVO::getId));
         } catch (Exception e) {
-            e.printStackTrace();
             log.error("立即执行执行失败", e);
+        } finally {
+            // 2. 清理租户上下文
+            if (isSwitched) {
+                try {
+                    TenantConfigContext.clear();
+                    SecurityContextHolder.clearContext();
+                    Object manager = SpringUtils.getBean("tenantDataSourceManager");
+                    Method method = manager.getClass().getMethod("clear");
+                    method.invoke(manager);
+                } catch (Exception e) {
+                    log.error("SOP异步任务清理租户数据源失败", e);
+                }
+            }
         }
     }
 

fs-service/src/main/java/com/fs/qw/service/impl/AsyncSopService.java

@@ -1,6 +1,11 @@
 package com.fs.qw.service.impl;
 
+import com.alibaba.fastjson.JSON;
 import com.fs.common.utils.PubFun;
+import com.fs.common.utils.StringUtils;
+import com.fs.common.utils.spring.SpringUtils;
+import com.fs.core.config.TenantConfigContext;
+import com.fs.common.core.domain.model.TenantPrincipal;
 import com.fs.qw.domain.QwUser;
 import com.fs.qw.mapper.QwExternalContactMapper;
 import com.fs.qw.result.QwFilterSopCustomersResult;
@@ -16,15 +21,20 @@ import com.fs.sop.params.SopUserLogsArray;
 import com.fs.sop.params.SopUserLogsList;
 import com.fs.sop.service.IQwSopTempVoiceService;
 import com.fs.sop.service.ISopUserLogsService;
+import com.fs.system.domain.SysConfig;
+import com.fs.system.mapper.SysConfigMapper;
 import com.fs.voice.utils.StringUtil;
 import com.fs.wxUser.domain.CompanyWxUser;
 import com.fs.wxUser.mapper.CompanyWxUserMapper;
 import com.fs.wxUser.param.CompanyWxUserSopParam;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
+import org.springframework.security.core.context.SecurityContextHolder;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 
+import java.lang.reflect.Method;
 import java.text.SimpleDateFormat;
 import java.util.*;
 import java.util.stream.Collectors;
@@ -43,20 +53,48 @@ public class AsyncSopService {
     private final QwExternalContactMapper qwExternalContactMapper;
     private final QwSopTempMapper qwSopTempMapper;
     private final IQwSopTempVoiceService qwSopTempVoiceService;
+    private final SysConfigMapper sysConfigMapper;
 
     /**
      * 立即执行SOP任务
      */
     @Async("scheduledExecutorService")
-    public void executeSopByIds(String[] ids) {
+    public void executeSopByIds(String[] ids, Long tenantId) {
+        boolean isSwitched = false;
+        try {
+            // 1. 切换到租户数据源并加载配置
+            if (tenantId != null) {
+                try {
+                    Object manager = SpringUtils.getBean("tenantDataSourceManager");
+                    Method method = manager.getClass().getMethod("ensureSwitchByTenantId", Long.class);
+                    method.invoke(manager, tenantId);
+                    isSwitched = true;
+
+                    // 加载租户 projectConfig
+                    SysConfig cfg = sysConfigMapper.selectConfigByConfigKey("projectConfig");
+                    if (cfg != null && StringUtils.isNotBlank(cfg.getConfigValue())) {
+                        TenantConfigContext.set(JSON.parseObject(cfg.getConfigValue()));
+                    }
+
+                    // 设置租户到 SecurityContext
+                    SecurityContextHolder.getContext().setAuthentication(
+                            new UsernamePasswordAuthenticationToken(
+                                    new TenantPrincipal(tenantId),
+                                    null,
+                                    Collections.emptyList()
+                            )
+                    );
+                } catch (Exception e) {
+                    log.error("SOP异步任务切换租户数据源失败: tenantId={}", tenantId, e);
+                }
+            }
 
-        List<QwSopRuleTimeVO> ruleTimeVOList = qwSopMapper.executeSopByIds(ids);
-        List<QwSopTemp> tempList = qwSopTempMapper.selectListByIds(PubFun.listToNewList(ruleTimeVOList, QwSopRuleTimeVO::getTempId));
-        Map<String, QwSopTemp> tempMap = PubFun.listToMapByGroupObject(tempList, QwSopTemp::getId);
-        CompanyWxUserSopParam wxUserSopParam=new CompanyWxUserSopParam();
-        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
+            List<QwSopRuleTimeVO> ruleTimeVOList = qwSopMapper.executeSopByIds(ids);
+            List<QwSopTemp> tempList = qwSopTempMapper.selectListByIds(PubFun.listToNewList(ruleTimeVOList, QwSopRuleTimeVO::getTempId));
+            Map<String, QwSopTemp> tempMap = PubFun.listToMapByGroupObject(tempList, QwSopTemp::getId);
+            CompanyWxUserSopParam wxUserSopParam=new CompanyWxUserSopParam();
+            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
 
-        try {
             ruleTimeVOList.forEach(ruleTimeVO->{
                 QwSopTemp qwSopTemp = tempMap.get(ruleTimeVO.getTempId());
                 //如果当前模板停用了，则不运行了
@@ -291,6 +329,19 @@ public class AsyncSopService {
             });
         }catch (Exception e){
             log.error("立即执行执行失败",e);
+        } finally {
+            // 2. 清理租户上下文
+            if (isSwitched) {
+                try {
+                    TenantConfigContext.clear();
+                    SecurityContextHolder.clearContext();
+                    Object manager = SpringUtils.getBean("tenantDataSourceManager");
+                    Method method = manager.getClass().getMethod("clear");
+                    method.invoke(manager);
+                } catch (Exception e) {
+                    log.error("SOP异步任务清理租户数据源失败", e);
+                }
+            }
         }
 
     }

fs-service/src/main/java/com/fs/qw/service/impl/AsyncSopTestService.java

@@ -3,6 +3,7 @@ package com.fs.qw.service.impl;
 import cn.hutool.core.util.StrUtil;
 import com.alibaba.fastjson.JSON;
 import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fs.common.core.domain.model.TenantPrincipal;
 import com.fs.common.utils.PubFun;
 import com.fs.course.domain.FsCourseSopAppLink;
 import com.fs.course.mapper.FsCourseSopAppLinkMapper;
@@ -30,12 +31,18 @@ import com.fs.voice.utils.StringUtil;
 import com.fs.wxUser.param.CompanyWxUserSopParam;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.commons.lang.StringUtils;
-import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import com.fs.common.utils.StringUtils;
+import com.fs.common.utils.spring.SpringUtils;
+import com.fs.core.config.TenantConfigContext;
+import com.fs.system.domain.SysConfig;
+import com.fs.system.mapper.SysConfigMapper;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
+import org.springframework.security.core.context.SecurityContextHolder;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
-
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import java.lang.reflect.Method;
 import java.text.SimpleDateFormat;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
@@ -62,13 +69,42 @@ public class AsyncSopTestService {
     private final SopUserLogsMapper sopUserLogsMapper;
     private final FsCourseSopAppLinkMapper fsCourseSopAppLinkMapper;
     private final uniPush2Service push2Service;
+    private final SysConfigMapper sysConfigMapper;
 
     /**
      * 立即执行SOP任务
      */
     @Async("scheduledExecutorService")
-    public void executeSopByIds(String[] ids) {
+    public void executeSopByIds(String[] ids, Long tenantId) {
+        boolean isSwitched = false;
         try {
+            // 1. 切换到租户数据源并加载配置
+            if (tenantId != null) {
+                try {
+                    Object manager = SpringUtils.getBean("tenantDataSourceManager");
+                    Method method = manager.getClass().getMethod("ensureSwitchByTenantId", Long.class);
+                    method.invoke(manager, tenantId);
+                    isSwitched = true;
+
+                    // 加载租户 projectConfig
+                    SysConfig cfg = sysConfigMapper.selectConfigByConfigKey("projectConfig");
+                    if (cfg != null && StringUtils.isNotBlank(cfg.getConfigValue())) {
+                        TenantConfigContext.set(JSON.parseObject(cfg.getConfigValue()));
+                    }
+
+                    // 设置租户到 SecurityContext，供 TenantKeyRedisSerializer 自动为 Redis Key 加 tenantid 前缀
+                    SecurityContextHolder.getContext().setAuthentication(
+                            new UsernamePasswordAuthenticationToken(
+                                    new TenantPrincipal(tenantId),
+                                    null,
+                                    Collections.emptyList()
+                            )
+                    );
+                } catch (Exception e) {
+                    log.error("SOP异步任务切换租户数据源失败: tenantId={}", tenantId, e);
+                }
+            }
+
             List<QwSopRuleTimeVO> ruleTimeVOList = qwSopMapper.executeSopByIds(ids);
             List<QwSopTemp> tempList = qwSopTempMapper.selectListByIds(PubFun.listToNewList(ruleTimeVOList, QwSopRuleTimeVO::getTempId));
             Map<String, QwSopTemp> tempMap = PubFun.listToMapByGroupObject(tempList, QwSopTemp::getId);
@@ -80,11 +116,23 @@ public class AsyncSopTestService {
                     updateQwSopStatus(ruleTimeVO.getId(), 0L);
                     return;
                 }
-                processRuleTimeVOInternal(ruleTimeVO, qwSopTemp,sdf);
+                processRuleTimeVOInternal(ruleTimeVO, qwSopTemp, sdf, tenantId);
             });
         } catch (Exception e) {
-            e.printStackTrace();
             log.error("立即执行执行失败", e);
+        } finally {
+            // 2. 清理租户上下文
+            if (isSwitched) {
+                try {
+                    TenantConfigContext.clear();
+                    SecurityContextHolder.clearContext();
+                    Object manager = SpringUtils.getBean("tenantDataSourceManager");
+                    Method method = manager.getClass().getMethod("clear");
+                    method.invoke(manager);
+                } catch (Exception e) {
+                    log.error("SOP异步任务清理租户数据源失败", e);
+                }
+            }
         }
     }
 
@@ -167,28 +215,68 @@ public class AsyncSopTestService {
     /**
      * 判断类型
      */
-    private void processRuleTimeVOInternal(QwSopRuleTimeVO ruleTimeVO, QwSopTemp qwSopTemp,SimpleDateFormat sdf) {
-        QwSop qwSop = createQwSop(ruleTimeVO, 2L);
-
-        switch (ruleTimeVO.getType()) {
-            case 1:
-                processWeChatSop(ruleTimeVO, qwSop,sdf);
-                break;
-            case 2:
-                processEnterpriseWeChatSop(ruleTimeVO, qwSop,sdf);
-                break;
-            default:
-                break;
-        }
+    private void processRuleTimeVOInternal(QwSopRuleTimeVO ruleTimeVO, QwSopTemp qwSopTemp,SimpleDateFormat sdf, Long tenantId) {
+        boolean isSwitched = false;
+        try {
+            if (tenantId != null) {
+                try {
+                    Object manager = SpringUtils.getBean("tenantDataSourceManager");
+                    Method method = manager.getClass().getMethod("ensureSwitchByTenantId", Long.class);
+                    method.invoke(manager, tenantId);
+                    isSwitched = true;
+
+                    // 加载租户 projectConfig
+                    SysConfig cfg = sysConfigMapper.selectConfigByConfigKey("projectConfig");
+                    if (cfg != null && StringUtils.isNotBlank(cfg.getConfigValue())) {
+                        TenantConfigContext.set(JSON.parseObject(cfg.getConfigValue()));
+                    }
+
+                    // 设置租户到 SecurityContext
+                    SecurityContextHolder.getContext().setAuthentication(
+                            new UsernamePasswordAuthenticationToken(
+                                    new TenantPrincipal(tenantId),
+                                    null,
+                                    Collections.emptyList()
+                            )
+                    );
+                } catch (Exception e) {
+                    log.error("processRuleTimeVOInternal 切换租户数据源失败: tenantId={}", tenantId, e);
+                }
+            }
 
-        qwSopMapper.updateQwSop(qwSop);
+            QwSop qwSop = createQwSop(ruleTimeVO, 2L);
+            switch (ruleTimeVO.getType()) {
+                case 1:
+                    processWeChatSop(ruleTimeVO, qwSop, sdf, tenantId);
+                    break;
+                case 2:
+                    processEnterpriseWeChatSop(ruleTimeVO, qwSop, sdf, tenantId);
+                    break;
+                default:
+                    break;
+            }
+
+            qwSopMapper.updateQwSop(qwSop);
+        } finally {
+            if (isSwitched) {
+                try {
+                    TenantConfigContext.clear();
+                    SecurityContextHolder.clearContext();
+                    Object manager = SpringUtils.getBean("tenantDataSourceManager");
+                    Method method = manager.getClass().getMethod("clear");
+                    method.invoke(manager);
+                } catch (Exception e) {
+                    log.error("processRuleTimeVOInternal 清理租户数据源失败", e);
+                }
+            }
+        }
     }
 
 
     /**
      * 个微
      */
-    private void processWeChatSop(QwSopRuleTimeVO ruleTimeVO, QwSop qwSop,SimpleDateFormat sdf) {
+    private void processWeChatSop(QwSopRuleTimeVO ruleTimeVO, QwSop qwSop,SimpleDateFormat sdf, Long tenantId) {
 //        CompanyWxUserSopParam wxUserSopParam = createCompanyWxUserSopParam(ruleTimeVO);
 //        List<CompanyWxUser> companyWxUserList = companyWxUserMapper.selectCompanyWxUserByCompanyUserId(wxUserSopParam);
 //
@@ -214,7 +302,7 @@ public class AsyncSopTestService {
     /**
      * 企微
      */
-    private void processEnterpriseWeChatSop(QwSopRuleTimeVO ruleTimeVO, QwSop qwSop,SimpleDateFormat sdf) {
+    private void processEnterpriseWeChatSop(QwSopRuleTimeVO ruleTimeVO, QwSop qwSop,SimpleDateFormat sdf, Long tenantId) {
         QwSopTagsParam qwSopTagsParam = createQwSopTagsParam(ruleTimeVO);
         List<QwFilterSopCustomersResult> qwFilterSopCustomersResults = qwSopMapper.selectFilterQwSopCustomers(qwSopTagsParam);
 
@@ -225,7 +313,7 @@ public class AsyncSopTestService {
         }
 
         //如果是新客对话类型的，则跳过
-        if (ruleTimeVO.getSendType()==4){
+        if (ruleTimeVO.getSendType() == 4) {
             return;
         }
 
@@ -245,8 +333,8 @@ public class AsyncSopTestService {
             rocketMQTemplate.syncSend("voice-generation", JSON.toJSONString(VoiceVo.builder().type(1).id(ruleTimeVO.getId()).build()));
 //            new Thread(() -> HttpUtils.sendGet("http://118.24.209.192:8009/qw/voice/synchronousSop", "sopId=" + ruleTimeVO.getId())).start();
 //            qwSopTempVoiceService.synchronous(ruleTimeVO.getId(), companyUserIds);
-        }catch (Exception e){
-            log.error("异步同步临时语音失败:",e);
+        } catch (Exception e) {
+            log.error("异步同步临时语音失败:", e);
         }
         groupedResults.forEach((userIdAndQwUserId, externalUserIds) -> {
             String[] keys = userIdAndQwUserId.split("\\|");
@@ -255,8 +343,8 @@ public class AsyncSopTestService {
             String companyUserId = keys[2].trim();
             String companyId = keys[3].trim();
 
-            SopUserLogs sopUserLogs = createSopUserLogs(ruleTimeVO, userId, qwUserId, companyUserId, companyId,sdf);
-            SopUserLogsList userLogsList = createSopUserLogsList(ruleTimeVO, userId, qwUserId, companyUserId, companyId,sdf);
+            SopUserLogs sopUserLogs = createSopUserLogs(ruleTimeVO, userId, qwUserId, companyUserId, companyId, sdf);
+            SopUserLogsList userLogsList = createSopUserLogsList(ruleTimeVO, userId, qwUserId, companyUserId, companyId, sdf);
 
             String unionSopId = sopUserLogsService.selectSopUserLogsByUnionSopId(sopUserLogs);
             if (!StringUtil.strIsNullOrEmpty(unionSopId)) {

fs-service/src/main/java/com/fs/qw/service/impl/AsyncWxSopService.java

@@ -1,7 +1,12 @@
 package com.fs.qw.service.impl;
 
+import com.alibaba.fastjson.JSON;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.fs.common.utils.PubFun;
+import com.fs.common.utils.StringUtils;
+import com.fs.common.utils.spring.SpringUtils;
+import com.fs.core.config.TenantConfigContext;
+import com.fs.common.core.domain.model.TenantPrincipal;
 import com.fs.qw.vo.WxSopRuleTimeVO;
 import com.fs.sop.domain.QwSop;
 import com.fs.sop.domain.QwSopTemp;
@@ -9,13 +14,19 @@ import com.fs.sop.domain.SopUserLogsWx;
 import com.fs.sop.mapper.QwSopMapper;
 import com.fs.sop.mapper.QwSopTempMapper;
 import com.fs.sop.service.ISopUserLogsWxService;
+import com.fs.system.domain.SysConfig;
+import com.fs.system.mapper.SysConfigMapper;
 import com.fs.wxUser.domain.CompanyWxUser;
 import com.fs.wxUser.mapper.CompanyWxUserMapper;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
+import org.springframework.security.core.context.SecurityContextHolder;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Service;
 
+import java.lang.reflect.Method;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
@@ -29,13 +40,42 @@ public class AsyncWxSopService {
     private final QwSopTempMapper qwSopTempMapper;
     private final CompanyWxUserMapper companyWxUserMapper;
     private final ISopUserLogsWxService sopUserLogsWxService;
+    private final SysConfigMapper sysConfigMapper;
 
     /**
      * 立即执行SOP任务
      */
     @Async("scheduledExecutorService")
-    public void scheduledWxSopService(String[] ids) {
+    public void scheduledWxSopService(String[] ids, Long tenantId) {
+        boolean isSwitched = false;
         try {
+            // 1. 切换到租户数据源并加载配置
+            if (tenantId != null) {
+                try {
+                    Object manager = SpringUtils.getBean("tenantDataSourceManager");
+                    Method method = manager.getClass().getMethod("ensureSwitchByTenantId", Long.class);
+                    method.invoke(manager, tenantId);
+                    isSwitched = true;
+
+                    // 加载租户 projectConfig
+                    SysConfig cfg = sysConfigMapper.selectConfigByConfigKey("projectConfig");
+                    if (cfg != null && StringUtils.isNotBlank(cfg.getConfigValue())) {
+                        TenantConfigContext.set(JSON.parseObject(cfg.getConfigValue()));
+                    }
+
+                    // 设置租户到 SecurityContext
+                    SecurityContextHolder.getContext().setAuthentication(
+                            new UsernamePasswordAuthenticationToken(
+                                    new TenantPrincipal(tenantId),
+                                    null,
+                                    Collections.emptyList()
+                            )
+                    );
+                } catch (Exception e) {
+                    log.error("SOP异步任务切换租户数据源失败: tenantId={}", tenantId, e);
+                }
+            }
+
             List<WxSopRuleTimeVO> ruleTimeVOList = qwSopMapper.executeSopWxByIds(ids);
             List<QwSopTemp> tempList = qwSopTempMapper.selectListByIds(PubFun.listToNewList(ruleTimeVOList, WxSopRuleTimeVO::getTempId));
             Map<String, QwSopTemp> tempMap = PubFun.listToMapByGroupObject(tempList, QwSopTemp::getId);
@@ -53,8 +93,20 @@ public class AsyncWxSopService {
                 }
             });
         } catch (Exception e) {
-            e.printStackTrace();
             log.error("立即执行执行失败", e);
+        } finally {
+            // 2. 清理租户上下文
+            if (isSwitched) {
+                try {
+                    TenantConfigContext.clear();
+                    SecurityContextHolder.clearContext();
+                    Object manager = SpringUtils.getBean("tenantDataSourceManager");
+                    Method method = manager.getClass().getMethod("clear");
+                    method.invoke(manager);
+                } catch (Exception e) {
+                    log.error("SOP异步任务清理租户数据源失败", e);
+                }
+            }
         }
     }
 

fs-service/src/main/java/com/fs/qwApi/config/OpenQwConfig.java

@@ -1,6 +1,6 @@
 package com.fs.qwApi.config;
 
 public interface OpenQwConfig {
-    String baseApi ="http://127.0.0.1:8007/open/qwapi";
-    String api ="http://127.0.0.1:8007";
+    String baseApi ="http://saasqwapi.ylrzcloud.com/open/qwapi";
+    String api ="http://saasqwapi.ylrzcloud.com";
 }

fs-service/src/main/java/com/fs/sop/mapper/QwSopMapper.java

@@ -168,7 +168,6 @@ public interface QwSopMapper extends BaseMapper<QwSop> {
             "</script>")
     List<QwSopListVO> selectSopListVO(@Param("map") QwSopParam qwSopParam);
 
-    @DataSource(DataSourceType.MASTER)
     @Select("SELECT cu.nick_name as userName , qec.* " +
             "FROM " +
             " qw_external_contact qec  left JOIN company_user cu on  qec.company_user_id=cu.user_id " +
@@ -246,7 +245,6 @@ public interface QwSopMapper extends BaseMapper<QwSop> {
 //            "</script>")
 //    public List<QwFilterSopCustomersResult> selectFilterSopCustomers(@Param("map") QwSopTagsParam qwSopTagsParam);
 
-    @DataSource(DataSourceType.MASTER)
     @Select("<script>" +
             "SELECT  qec.* , qu.company_user_id as cuCompanyUserId ,cu.company_id as cuCompanyId  " +
             "FROM " +

fs-service/src/main/java/com/fs/sop/service/impl/QwSopServiceImpl.java

@@ -7,6 +7,7 @@ import com.fs.common.core.domain.R;
 import com.fs.common.enums.DataSourceType;
 import com.fs.common.exception.base.BaseException;
 import com.fs.common.utils.PubFun;
+import com.fs.common.utils.SecurityUtils;
 import com.fs.common.utils.StringUtils;
 import com.fs.company.domain.CompanyUser;
 import com.fs.company.mapper.CompanyUserMapper;
@@ -790,8 +791,7 @@ public class QwSopServiceImpl implements IQwSopService
 
     @Override
     public R updateStatusQwSopByIds(String[] ids) {
-
-
+        Long tenantId = SecurityUtils.getTenantId();
         List<QwSop> qwSops = qwSopMapper.selectStatusQwSopById(ids);
         List<QwSop> qwSopList = qwSops.stream().filter(e -> e.getType() == 2 && e.getFilterMode() == 1).collect(Collectors.toList());
         List<QwSop> wxSopList = qwSops.stream().filter(e -> e.getType() == 1).collect(Collectors.toList());
@@ -812,7 +812,7 @@ public class QwSopServiceImpl implements IQwSopService
                     .map(QwSop::getId)
                     .collect(Collectors.toList());
 
-            asyncWxSopService.scheduledWxSopService(toBeSent);
+            asyncWxSopService.scheduledWxSopService(toBeSent, tenantId);
 
             if (toBeSent.length > 0) {
                 int i = qwSopMapper.updateStatusQwSopById(toBeSent);
@@ -839,7 +839,7 @@ public class QwSopServiceImpl implements IQwSopService
                     .collect(Collectors.toList());
 
             //异步执行
-            asyncSopTestService.executeSopByIds(toBeSent);
+            asyncSopTestService.executeSopByIds(toBeSent, tenantId);
 
             if (toBeSent.length > 0) {
                 int i = qwSopMapper.updateStatusQwSopById(toBeSent);
@@ -866,7 +866,7 @@ public class QwSopServiceImpl implements IQwSopService
                     .collect(Collectors.toList());
 
             //异步执行
-            asyncChatSopService.executeChatSopByIds(toBeSent);
+            asyncChatSopService.executeChatSopByIds(toBeSent, tenantId);
 
             if (toBeSent.length > 0) {
                 int i = qwSopMapper.updateStatusQwSopById(toBeSent);

fs-service/src/main/java/com/fs/statis/service/impl/StatisticsServiceImpl.java

@@ -1113,9 +1113,12 @@ public class StatisticsServiceImpl implements IStatisticsService {
             watchCourseStatisticsDTOS = JSONObject.parseArray(redisData, WatchCourseStatisticsResultDTO.class);
         }
         String sendType;
-        if (param.getUserType() == 1) {
+        Integer userType = param.getUserType();
+        if (userType == null){
+            return new ArrayList<>();
+        } else if (userType == 1) {
             sendType = "1";
-        } else if (param.getUserType() == 2) {
+        } else if (userType == 2) {
             sendType = "2";
         } else {
             return new ArrayList<>();