|
|
@@ -0,0 +1,309 @@
|
|
|
+package com.fs.company.service.impl.call.node;
|
|
|
+
|
|
|
+import com.alibaba.fastjson.JSONObject;
|
|
|
+import com.fs.common.constant.Constants;
|
|
|
+import com.fs.common.core.redis.RedisCacheT;
|
|
|
+import com.fs.common.exception.CustomException;
|
|
|
+import com.fs.common.utils.StringUtils;
|
|
|
+import com.fs.common.utils.http.HttpUtils;
|
|
|
+import com.fs.common.utils.spring.SpringUtils;
|
|
|
+import com.fs.company.domain.*;
|
|
|
+import com.fs.company.mapper.*;
|
|
|
+import com.fs.company.param.AddWxActionParam;
|
|
|
+import com.fs.company.param.ExecutionContext;
|
|
|
+import com.fs.company.util.ObjectPlaceholderResolver;
|
|
|
+import com.fs.company.vo.AiAddWxConfigVO;
|
|
|
+import com.fs.company.vo.AiCallWorkflowConditionVo;
|
|
|
+import com.fs.company.vo.ExecutionResult;
|
|
|
+import com.fs.crm.domain.CrmCustomer;
|
|
|
+import com.fs.crm.service.impl.CrmCustomerServiceImpl;
|
|
|
+import com.fs.enums.ExecutionStatusEnum;
|
|
|
+import com.fs.enums.NodeTypeEnum;
|
|
|
+import com.fs.system.service.ISysConfigService;
|
|
|
+import com.fs.wxcid.domain.CidIpadServer;
|
|
|
+import com.fs.wxcid.domain.WxContact;
|
|
|
+import com.fs.wxcid.mapper.CidIpadServerMapper;
|
|
|
+import com.fs.wxcid.mapper.WxContactMapper;
|
|
|
+import lombok.extern.slf4j.Slf4j;
|
|
|
+
|
|
|
+import java.util.Date;
|
|
|
+import java.util.List;
|
|
|
+import java.util.Map;
|
|
|
+import java.util.stream.Collectors;
|
|
|
+
|
|
|
+/**
|
|
|
+ * @author MixLiu
|
|
|
+ * @date 2026/04/22 13:15
|
|
|
+ * @description AI添加微信任务节点（新）
|
|
|
+ */
|
|
|
+@Slf4j
|
|
|
+public class AiAddWxTaskNewNode extends AbstractWorkflowNode {
|
|
|
+
|
|
|
+ private static final CompanyWxClientMapper companyWxClientMapper = SpringUtils.getBean(CompanyWxClientMapper.class);
|
|
|
+ @SuppressWarnings("unchecked")
|
|
|
+ private static final RedisCacheT<String> redisCache = SpringUtils.getBean(RedisCacheT.class);
|
|
|
+ private static final WxContactMapper wxContactMapper = SpringUtils.getBean(WxContactMapper.class);
|
|
|
+ private static final CompanyWxAccountMapper companyWxAccountMapper = SpringUtils.getBean(CompanyWxAccountMapper.class);
|
|
|
+ public static final String DELAY_ADD_WX_NEW_KEY = "addWxTaskNew:delay:%s:%s:%s:";
|
|
|
+ private static final CompanyWxDialogMapper companyWxDialogMapper = SpringUtils.getBean(CompanyWxDialogMapper.class);
|
|
|
+ private static final CrmCustomerServiceImpl crmCustomerService = SpringUtils.getBean(CrmCustomerServiceImpl.class);
|
|
|
+ private static final ObjectPlaceholderResolver objectPlaceholderResolver = SpringUtils.getBean(ObjectPlaceholderResolver.class);
|
|
|
+ private static final ISysConfigService sysConfigService = SpringUtils.getBean(ISysConfigService.class);
|
|
|
+ private static final CidIpadServerMapper cidIpadServerMapper = SpringUtils.getBean(CidIpadServerMapper.class);
|
|
|
+ private static final String ADDWX_POST_URL = "/app/common/addWxAction";
|
|
|
+ /**
|
|
|
+ * 默认加微超时时间（分钟）
|
|
|
+ */
|
|
|
+ private static final int DEFAULT_ADD_WX_TIMEOUT_MINUTES = 30;
|
|
|
+
|
|
|
+ public AiAddWxTaskNewNode(String nodeKey, String nodeName, Map<String, Object> properties) {
|
|
|
+ super(nodeKey, nodeName, properties);
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public NodeTypeEnum getType() {
|
|
|
+ return NodeTypeEnum.AI_ADD_WX_TASK_NEW;
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public Boolean isAsync() {
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 执行加微节点逻辑（准备数据并发起加微请求）
|
|
|
+ *
|
|
|
+ * @param context 执行上下文
|
|
|
+ * @return 执行结果
|
|
|
+ */
|
|
|
+ @Override
|
|
|
+ protected ExecutionResult doExecute(ExecutionContext context) {
|
|
|
+ if (!isAsync()) {
|
|
|
+ return ExecutionResult.failure().nextNodeKey(null).build();
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ // 设置加微话术
|
|
|
+ CompanyWorkflowNode node = context.getVariable("currentNode", CompanyWorkflowNode.class) == null
|
|
|
+ ? getNodeByKey(nodeKey)
|
|
|
+ : context.getVariable("currentNode", CompanyWorkflowNode.class);
|
|
|
+ String nodeConfig = node.getNodeConfig();
|
|
|
+ AiAddWxConfigVO addWxConfig = nodeConfig == null ? null : JSONObject.parseObject(nodeConfig, AiAddWxConfigVO.class);
|
|
|
+
|
|
|
+ if (addWxConfig == null || addWxConfig.getDialogId() == null) {
|
|
|
+ throw new CustomException("加微节点未配置加微话术，执行失败");
|
|
|
+ }
|
|
|
+
|
|
|
+ WxContact wxQuery = companyAiWorkflowExecMapper.selectWxContectByWorkflowInstanceId(context.getWorkflowInstanceId());
|
|
|
+ wxQuery.setNickName(wxQuery.getRemark());
|
|
|
+ wxQuery.setFriends(0);
|
|
|
+ wxContactMapper.insert(wxQuery);
|
|
|
+
|
|
|
+ CompanyVoiceRoboticBusiness roboticBusiness = getRoboticBusiness(context.getWorkflowInstanceId());
|
|
|
+ CompanyWxClient update = new CompanyWxClient();
|
|
|
+ update.setDialogId(addWxConfig.getDialogId());
|
|
|
+ update.setId(roboticBusiness.getWxClientId());
|
|
|
+ companyWxClientMapper.updateCompanyWxClient(update);
|
|
|
+ super.asyncWorkflowForBlockingNode(context.getWorkflowInstanceId(), context.getCurrentNodeKey(), context, ExecutionStatusEnum.PAUSED);
|
|
|
+ pendingAddWx(wxQuery.getAccountId(),wxQuery.getRemark(),wxQuery.getPhone(),addWxConfig.getDialogId(),wxQuery.getCrmUserId());
|
|
|
+ doneAddwx(context.getWorkflowInstanceId());
|
|
|
+ return ExecutionResult.paused()
|
|
|
+ .outputData(context.getVariables())
|
|
|
+ .nextNodeKey("").build();
|
|
|
+
|
|
|
+ } catch (Exception e) {
|
|
|
+ log.error("准备加微任务数据异常 流程：{}：节点:{}执行失败，", context.getWorkflowInstanceId(), nodeKey, e);
|
|
|
+ super.updateWorkflowStatus(context.getWorkflowInstanceId(), ExecutionStatusEnum.INTERRUPT);
|
|
|
+ return ExecutionResult.failure().errorMessage("准备加微任务数据异常: " + e.getMessage()).build();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 收到加微回调后，继续判定和执行下一步动作
|
|
|
+ *
|
|
|
+ * @param context 执行上下文
|
|
|
+ * @return 执行结果
|
|
|
+ */
|
|
|
+ @Override
|
|
|
+ protected ExecutionResult doContinue(ExecutionContext context) {
|
|
|
+ CompanyAiWorkflowExec exec = companyAiWorkflowExecMapper.selectByWorkflowInstanceId(context.getWorkflowInstanceId());
|
|
|
+ if (exec == null) {
|
|
|
+ log.warn("doContinue: 工作流执行实例不存在 - workflowInstanceId: {}", context.getWorkflowInstanceId());
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ List<CompanyWorkflowEdge> edges = companyWorkflowEdgeMapper.selectListByWorkflowIdAndNodeKey(exec.getWorkflowId(), nodeKey);
|
|
|
+ if (edges == null || edges.isEmpty()) {
|
|
|
+ log.warn("doContinue: 未找到出边 - workflowInstanceId: {}, nodeKey: {}", context.getWorkflowInstanceId(), nodeKey);
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 获取业务数据
|
|
|
+ CompanyVoiceRoboticBusiness business = super.getRoboticBusiness(context.getWorkflowInstanceId());
|
|
|
+ // 获取加微记录
|
|
|
+ CompanyWxClient wxClient = companyWxClientMapper.selectById(business.getWxClientId());
|
|
|
+
|
|
|
+ log.info("收到加微回调 - workflowInstanceId: {}, wxClientId: {}, isAdd: {}",
|
|
|
+ context.getWorkflowInstanceId(), business.getWxClientId(), wxClient != null ? wxClient.getIsAdd() : null);
|
|
|
+
|
|
|
+ // 判断加微是否成功 (isAdd: 0否 1是 2待添加 3作废)
|
|
|
+ boolean addSuccess = wxClient != null && Integer.valueOf(1).equals(wxClient.getIsAdd());
|
|
|
+ // 回调加微成功，走成功分支出边
|
|
|
+ if (addSuccess) {
|
|
|
+ List<CompanyWorkflowEdge> cList = edges.stream().filter(a -> {
|
|
|
+ if (StringUtils.isBlank(a.getConditionExpr())) {
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ List<AiCallWorkflowConditionVo> list = JSONObject.parseArray(a.getConditionExpr(), AiCallWorkflowConditionVo.class);
|
|
|
+ return list != null && !list.isEmpty() && list.get(0).isAdd();
|
|
|
+ }).collect(Collectors.toList());
|
|
|
+ if (null != cList && !cList.isEmpty() && nodeKey.equals(exec.getCurrentNodeKey())) {
|
|
|
+ super.runNextNode(context, cList.get(0));
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 完成加微动作，设置为 WAITING 并处理出边条件（延时/直通）
|
|
|
+ *
|
|
|
+ * @param workflowInstanceId 工作流实例ID
|
|
|
+ */
|
|
|
+ public void doneAddwx(String workflowInstanceId) {
|
|
|
+ ExecutionContext context = createExecutionContext(workflowInstanceId, nodeKey);
|
|
|
+ context.setVariable("lastNodeKey", nodeKey);
|
|
|
+ CompanyAiWorkflowExec exec = companyAiWorkflowExecMapper.selectByWorkflowInstanceId(context.getWorkflowInstanceId());
|
|
|
+ if (exec == null) {
|
|
|
+ log.error("doneAddwx: 工作流执行实例不存在 - workflowInstanceId: {}", workflowInstanceId);
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ if (!exec.getCurrentNodeKey().equals(nodeKey)) {
|
|
|
+ log.error("当前节点已流转 ,目标：{}，实际：{}", nodeKey, exec.getCurrentNodeKey());
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ // 更新加微日志执行状态
|
|
|
+ super.updateLogStatusIfExist(context, ExecutionStatusEnum.PAUSED, ExecutionStatusEnum.WAITING);
|
|
|
+ super.asyncWorkflowForBlockingNode(context.getWorkflowInstanceId(), nodeKey, context, ExecutionStatusEnum.WAITING);
|
|
|
+ List<CompanyWorkflowEdge> edges = companyWorkflowEdgeMapper.selectListByWorkflowIdAndNodeKey(exec.getWorkflowId(), nodeKey);
|
|
|
+ if (edges == null) {
|
|
|
+ return;
|
|
|
+ }
|
|
|
+ edges.forEach(edge -> {
|
|
|
+ String conditionExpr = edge.getConditionExpr();
|
|
|
+ List<AiCallWorkflowConditionVo> conditions = StringUtils.isBlank(conditionExpr) ? null : JSONObject.parseArray(conditionExpr, AiCallWorkflowConditionVo.class);
|
|
|
+ if (null == conditions || conditions.isEmpty()) {
|
|
|
+ super.runNextNode(context, edge);
|
|
|
+ } else {
|
|
|
+ AiCallWorkflowConditionVo condition = conditions.get(0);
|
|
|
+ // 节点包含延时条件
|
|
|
+ if (null != condition.getAddTime() && !condition.isAdd()) {
|
|
|
+ long l = System.currentTimeMillis() + condition.getAddTime() * 60 * 1000;
|
|
|
+ String redisKey = getDelayAddWxKeyPrefix(exec.getCidGroupNo(), l) + workflowInstanceId;
|
|
|
+ ExecutionContext nextContext = context.clone();
|
|
|
+ nextContext.setCurrentNodeKey(edge.getTargetNodeKey());
|
|
|
+ super.redisCache.setCacheObject(redisKey, nextContext);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ });
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 检查并标记已执行（互斥控制）
|
|
|
+ * 返回 true 表示当前是第一个执行的，可以继续；
|
|
|
+ * 返回 false 表示已被其他路径执行过，不再执行。
|
|
|
+ *
|
|
|
+ * @param workflowInstanceId 工作流实例ID
|
|
|
+ * @param wxClientId 加微客户ID
|
|
|
+ * @return 是否可以执行
|
|
|
+ */
|
|
|
+ public static boolean tryMarkAsExecuted(String workflowInstanceId, Long wxClientId) {
|
|
|
+ String executedKey = Constants.WORKFLOW_ADD_WX_EXECUTED + "new:" + workflowInstanceId + ":" + wxClientId;
|
|
|
+ String existingValue = redisCache.getCacheObject(executedKey);
|
|
|
+ if (existingValue != null) {
|
|
|
+ return false;
|
|
|
+ }
|
|
|
+ redisCache.setCacheObject(executedKey, "1");
|
|
|
+ redisCache.expire(executedKey, 3600);
|
|
|
+ return true;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 生成延时加微 Redis Key 前缀
|
|
|
+ *
|
|
|
+ * @param cidGroupNo CID分组号
|
|
|
+ * @param time 目标时间戳（毫秒），为 null 时取当前时间
|
|
|
+ * @return Redis Key 前缀
|
|
|
+ */
|
|
|
+ public static String getDelayAddWxKeyPrefix(Integer cidGroupNo, Long time) {
|
|
|
+ Date nowDay = new Date();
|
|
|
+ if (null != time) {
|
|
|
+ nowDay = new Date(time);
|
|
|
+ }
|
|
|
+ return String.format(DELAY_ADD_WX_NEW_KEY, cidGroupNo, nowDay.getHours(), nowDay.getMinutes());
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 从节点配置获取超时时间（分钟）
|
|
|
+ */
|
|
|
+ private int getTimeoutFromProperties() {
|
|
|
+ if (properties != null && properties.containsKey("timeout")) {
|
|
|
+ Object timeout = properties.get("timeout");
|
|
|
+ if (timeout instanceof Number) {
|
|
|
+ return ((Number) timeout).intValue();
|
|
|
+ }
|
|
|
+ if (timeout instanceof String) {
|
|
|
+ try {
|
|
|
+ return Integer.parseInt((String) timeout);
|
|
|
+ } catch (NumberFormatException e) {
|
|
|
+ log.warn("解析超时时间失败: {}, 使用默认值: {}", timeout, DEFAULT_ADD_WX_TIMEOUT_MINUTES);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ }
|
|
|
+ return DEFAULT_ADD_WX_TIMEOUT_MINUTES;
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * 发起加微请求
|
|
|
+ * @param accountId 微信账号ID
|
|
|
+ * @param remark 备注
|
|
|
+ * @param phone 手机号
|
|
|
+ * @param dialogId 话术ID
|
|
|
+ * @param crmUserId CRM用户ID
|
|
|
+ */
|
|
|
+ private void pendingAddWx(Long accountId, String remark, String phone, Long dialogId, Long crmUserId) {
|
|
|
+ // 1. 获取基础数据
|
|
|
+ CompanyWxAccount companyWxAccount = companyWxAccountMapper.selectCompanyWxAccountById(accountId);
|
|
|
+ if (companyWxAccount == null) {
|
|
|
+ throw new CustomException("未找到对应的微信账号配置, accountId: " + accountId);
|
|
|
+ }
|
|
|
+
|
|
|
+ CompanyWxDialog dialog = companyWxDialogMapper.selectCompanyWxDialogById(dialogId);
|
|
|
+ if (dialog == null) {
|
|
|
+ throw new CustomException("未找到对应的对话模板, dialogId: " + dialogId);
|
|
|
+ }
|
|
|
+
|
|
|
+ CrmCustomer crmCustomer = crmCustomerService.selectCrmCustomerById(crmUserId);
|
|
|
+
|
|
|
+ // 2. 解析话术模板
|
|
|
+ String newTxt = objectPlaceholderResolver.resolvePlaceholders(crmCustomer, dialog.getTemplateDetails());
|
|
|
+
|
|
|
+ // 3. 构建请求参数
|
|
|
+ AddWxActionParam param = new AddWxActionParam();
|
|
|
+ param.setWxId(companyWxAccount.getWxNo());
|
|
|
+ param.setRemark(remark);
|
|
|
+ param.setPhone(phone);
|
|
|
+ param.setApplyMsg(newTxt);
|
|
|
+ CidIpadServer cidIpadServer = cidIpadServerMapper.selectCidIpadServerById(companyWxAccount.getServerId());
|
|
|
+ if (null == cidIpadServer || StringUtils.isBlank(cidIpadServer.getUrl())) {
|
|
|
+ throw new CustomException("加微接口地址未配置");
|
|
|
+ }
|
|
|
+ // 4. 从系统配置获取加微接口地址
|
|
|
+ String addWxUrl = cidIpadServer.getUrl() + ADDWX_POST_URL;
|
|
|
+
|
|
|
+ // 5. 发送 HTTP 请求
|
|
|
+ try {
|
|
|
+ String result = HttpUtils.sendPost(addWxUrl, JSONObject.toJSONString(param));
|
|
|
+ log.info("pendingAddWx: 加微任务提交成功, phone: {}, result: {}", phone, result);
|
|
|
+ } catch (Exception e) {
|
|
|
+ throw new CustomException("发起加微请求异常, phone: " + phone + ", error: " + e.getMessage());
|
|
|
+ }
|
|
|
+ }
|
|
|
+}
|
