完成自定义长文本推流的实现

iot-common/iot-third/src/main/java/com/qiuguo/iot/third/nlp/LacWrapper.java

@@ -0,0 +1,4 @@
+package com.qiuguo.iot.third.nlp;
+
+public class LacWrapper {
+}

iot-modules/iot-box-websocket-api/src/main/java/com/qiuguo/iot/box/websocket/api/domain/BaseSession.java

@@ -45,7 +45,7 @@ public class BaseSession {
     protected MusicResp music;
 
     /**
-     * 调用千问请求序号，每次确定调用前++
+     * 问题Id，每次++
      */
     protected Long requestId = 0L;
 }

iot-modules/iot-box-websocket-api/src/main/java/com/qiuguo/iot/box/websocket/api/handler/BaseWebSocketProcess.java

@@ -47,6 +47,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.web.reactive.socket.WebSocketSession;
 import reactor.core.publisher.Mono;
+import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 
 import javax.annotation.Resource;
@@ -93,6 +94,8 @@ public class BaseWebSocketProcess {
     @Value("${tts.suanfa}")
     boolean ttsSuanfa;
 
+    private static int ONE_MAX_TEXT = 30;
+
     @Autowired
     protected SystemTalkBindU3dService systemTalkBindU3dService;
 
@@ -104,10 +107,41 @@ public class BaseWebSocketProcess {
     protected static String apiType = "api-type";
     protected static String apiToken = "api-token";
 
+    private String getSendStr(StringBuilder sb, String message){
+        String old = sb.toString() + message;
+        int d = old.lastIndexOf("，");
+        int j = old.lastIndexOf("。");
+        int a = old.lastIndexOf("：");
+        int b = old.lastIndexOf("\n");
+        int c = old.lastIndexOf("；");
+        int n = old.lastIndexOf("\\n");
+        int m = Math.max(Math.max(Math.max(Math.max(d, j), Math.max(a, b)), c), n);
+        if (m > 0) {
+            //清空
+            String msg = old.substring(0, m);
+            if(msg.replace(" ", "").length() > 0){
+                //纯空格的不推送
+                sb.setLength(0);
+                sb.append(old.substring(m));
 
 
+                return msg;
+            }
+        }
+        sb.append(message);
+        return null;
+    }
+
+
+    private void sendMoreMsg(BaseSession baseSession, StringBuilder sb, String message, int type){
+        message = getSendStr(sb, message);
+        if(StringUtils.isNotEmpty(message)){
+            normalSendMsg(baseSession, message, type, YesNo.NO.getCode());
+        }
+    }
+
     private Mono<Void> toQianWen(Action action, BaseSession baseSession, Integer type){
-        baseSession.setRequestId(baseSession.getRequestId() + 1);
+        //baseSession.setRequestId(baseSession.getRequestId() + 1);
         TongYiCommunicationRest tongYiCommunicationRest = new TongYiCommunicationRest();
         tongYiCommunicationRest.setText(action.getAsk());
         tongYiCommunicationRest.setStatus("2");
@@ -124,29 +158,8 @@ public class BaseWebSocketProcess {
                 //通知到客户端
                 MDC.put(Log4Constans.PRINT_LOG_ID, baseSession.getLogId());
                 if (tongYiCommunicationRest.getRequestId().equals(baseSession.getRequestId())) {
-                    String old = sb.toString() + message;
+                    //测试后决定是否需要
-                    int d = old.lastIndexOf("，");
+                    sendMoreMsg(baseSession, sb, message, type);
-                    int j = old.lastIndexOf("。");
-                    int a = old.lastIndexOf("：");
-                    int b = old.lastIndexOf("\n");
-                    int c = old.lastIndexOf("；");
-                    int n = old.lastIndexOf("\\n");
-                    int m = Math.max(Math.max(Math.max(Math.max(d, j), Math.max(a, b)), c), n);
-                    if (m > 0) {
-                        //清空
-                        String msg = old.substring(0, m);
-                        if(msg.replace(" ", "").length() > 0){
-                            //纯空格的不推送
-                            sb.setLength(0);
-                            sb.append(old.substring(m));
-
-                            normalSendMsg(baseSession, msg, type, YesNo.NO.getCode());
-                            return;
-                        }
-
-                    }
-                    sb.append(message);
-
                     return;
                 }
                 log.info("已经有新的请求，不在推送到客户端SN：{} userId:{}", baseSession.getSn(), baseSession.getUserId());
@@ -167,14 +180,7 @@ public class BaseWebSocketProcess {
             if(data.getCode() == 200){
                 log.info("千问正常结束");
                 //保存记录
-                DeviceUserTalkRecordEntity talkRecord = new DeviceUserTalkRecordEntity();
+                return saveTalkRecord(baseSession, action, data.getResut()).flatMap(i -> {
-                talkRecord.setAskType(AskTypeEnum.TTS.getCode());
-                talkRecord.setAskValue(action.getAsk());
-                talkRecord.setAskKey(action.getAction());
-                talkRecord.setAnswerValue(data.getResut());
-                talkRecord.setUserId(baseSession.getUserId());
-                talkRecord.setDeviceId(baseSession.getDeviceId());
-                return deviceUserTalkRecordService.insertDeviceUserTalkRecord(talkRecord).flatMap(b -> {
                     return Mono.empty();
                 });
             }else{
@@ -185,6 +191,17 @@ public class BaseWebSocketProcess {
         })/*.subscribeOn(Schedulers.boundedElastic()).subscribe()*/;
     }
 
+    private Mono<Integer> saveTalkRecord(BaseSession baseSession, Action action, String text){
+        DeviceUserTalkRecordEntity talkRecord = new DeviceUserTalkRecordEntity();
+        talkRecord.setAskType(AskTypeEnum.TTS.getCode());
+        talkRecord.setAskValue(action.getAsk());
+        talkRecord.setAskKey(action.getAction());
+        talkRecord.setAnswerValue(text);
+        talkRecord.setUserId(baseSession.getUserId());
+        talkRecord.setDeviceId(baseSession.getDeviceId());
+        return deviceUserTalkRecordService.insertDeviceUserTalkRecord(talkRecord);
+    }
+
     private Mono<Boolean> toU3DMq(Action action, SystemTalkBindDeviceEntity systemTalkBindDeviceEntity, Long metaId){
 
         U3dMsg u3dMsg = new U3dMsg();
@@ -411,6 +428,7 @@ public class BaseWebSocketProcess {
                         return sendMessage(action, baseSession, action.getSystemTalkAnswerConfigEntity().getAnswerValue(), AskTypeEnum.TTS.getCode());
                     }
                 }else if(action.getSystemTalkAnswerConfigEntity().getAnswerType().equals(AskTypeEnum.QIU_GUO.getCode())){
+
                     return sendMessage(action, baseSession, action.getSystemTalkAnswerConfigEntity().getAnswerValue(), AskTypeEnum.TTS.getCode());
                 }else if(action.getSystemTalkAnswerConfigEntity().getAnswerType().equals(AskTypeEnum.TIME.getCode())){
                     DateTimeFormatter df = DateTimeFormatter.ofPattern(action.getSystemTalkAnswerConfigEntity().getAnswerValue());
@@ -639,13 +657,7 @@ public class BaseWebSocketProcess {
             }
         }
         return deviceUserTalkRecordService.insertDeviceUserTalkRecord(talkRecord).flatMap(i ->{
-            if(this instanceof BoxWebSocketHandler){
+
-                log.info("果box聊天记录，同步到客户端");
-                BaseSession userSession = getUserSessionWithUserId(baseSession.getUserId());
-                if(userSession != null){
-                    sendMsg(userSession, resp);
-                }
-            }
             sendMsg(baseSession, resp);
 
             return Mono.just(true);
@@ -687,28 +699,103 @@ public class BaseWebSocketProcess {
     }
 
     private void sendMsg(BaseSession baseSession, BaseMessageResp baseMessageResp) {
-        if(ttsSuanfa && this instanceof BoxWebSocketHandler){
+        if(this instanceof BoxWebSocketHandler){
-            String text = baseMessageResp.getText().replace("\n", "").replace("\t", "");
+            log.info("果box聊天记录，同步到客户端");
-            if(text.startsWith("，") ||
+            BaseSession userSession = getUserSessionWithUserId(baseSession.getUserId());
-                    text.startsWith("。") ||
+            if(userSession != null){
-                    text.startsWith("：") ||
+                sendMsg(userSession, baseMessageResp);
-                    text.startsWith("；") ||
+            }
-                    text.startsWith("《") ||
+            if(ttsSuanfa){
-                    text.startsWith(" ")){
+                String text = baseMessageResp.getText().replace("\n", "").replace("\t", "");
-                //标点符号起始会导致合成的声音第一句话有杂音
+                if(text.startsWith("，") ||
-                text = text.substring(1);
+                        text.startsWith("。") ||
+                        text.startsWith("：") ||
+                        text.startsWith("；") ||
+                        text.startsWith("《") ||
+                        text.startsWith(" ")){
+                    //标点符号起始会导致合成的声音第一句话有杂音
+                    text = text.substring(1);
+                }
+                if(text.length() > ONE_MAX_TEXT){
+                    StringBuilder builder = new StringBuilder();
+                    sendAudioMessage(baseSession,
+                            baseMessageResp,
+                            builder,
+                            text,
+                            0,
+                            text.length() - 1,
+                            baseSession.getRequestId()).subscribe();
+                }else{
+                    BoxMessageResp boxMessageResp = new BoxMessageResp();
+                    BeanUtils.copyProperties(baseMessageResp, boxMessageResp);
+                    sendAudioMessage(baseSession, boxMessageResp).subscribe();
+                }
             }
-            audioService.getAudioUrl(text + "。").map(s ->{
-                BoxMessageResp boxMessageResp = new BoxMessageResp();
-                BeanUtils.copyProperties(baseMessageResp, boxMessageResp);
-                boxMessageResp.setAudio(s);
-                sendMsg(baseSession, JSONObject.toJSONString(boxMessageResp));
-                return s;
-            }).subscribeOn(Schedulers.single()).subscribe();
         }else{
             sendMsg(baseSession, JSONObject.toJSONString(baseMessageResp));
         }
+    }
 
+    /**
+     * 分批按顺序发送超长ONE_MAX_TEXT文本音频
+     * @param baseSession
+     * @param baseMessageResp
+     * @param builder
+     * @param text
+     * @param n
+     * @param length
+     * @return
+     */
+    private Mono<String> sendAudioMessage(BaseSession baseSession,
+                                          BaseMessageResp baseMessageResp,
+                                          StringBuilder builder,
+                                          String text,
+                                          int n,
+                                          int length,
+                                          Long requestId){
+        if(n < length && baseSession.getRequestId().equals(requestId)){
+            n += ONE_MAX_TEXT;
+            if(n > length){
+                n = length;
+            }
+            String message = text.substring(n - ONE_MAX_TEXT, n);
+            int status = 0;
+            if(n == length){
+                message += "。";
+                status = 1;
+                builder.setLength(0);
+            }
+            message = getSendStr(builder, message);
+            int m = n;
+            if(StringUtils.isNotEmpty(message)){
+                BoxMessageResp boxMessageResp = new BoxMessageResp();
+                BeanUtils.copyProperties(baseMessageResp, boxMessageResp);
+                boxMessageResp.setText(message);
+                boxMessageResp.getTts().setStatus(status);
+                return sendAudioMessage(baseSession, boxMessageResp).flatMap(s -> {
+                    return sendAudioMessage(baseSession, baseMessageResp, builder, text, m, length, requestId);
+                });
+            }else if(n < length){
+                return sendAudioMessage(baseSession, baseMessageResp, builder, text, m, length, requestId);
+            }
+        }
+
+        return Mono.just("");
+    }
+
+    /**
+     * 一次性发送文本
+     * @param baseSession
+     * @param boxMessageResp
+     * @return
+     */
+    private Mono<String> sendAudioMessage(BaseSession baseSession, BoxMessageResp boxMessageResp){
+        return audioService.getAudioUrl(boxMessageResp.getText() + "。").map(s ->{
+
+            boxMessageResp.setAudio(s);
+            sendMsg(baseSession, JSONObject.toJSONString(boxMessageResp));
+            return s;
+        });//.subscribeOn(Schedulers.single()).subscribe();
     }
 
     private void sendMsg(BaseSession baseSession, String msg) {

iot-modules/iot-box-websocket-api/src/main/java/com/qiuguo/iot/box/websocket/api/handler/BoxWebSocketHandler.java

@@ -133,6 +133,7 @@ public class BoxWebSocketHandler extends BaseWebSocketProcess implements WebSock
         }
         log.info("收到SN:{}，消息：{}", boxTalkMessage.getSn(), boxTalkMessage.getMessage());
         return nlpService.getActionWithLacSingle(boxSession.getUserId(), boxTalkMessage.getMessage()).defaultIfEmpty(new Actions()).flatMap(actions -> {
+            boxSession.setRequestId(boxSession.getRequestId() + 1);
             return processAction(actions, boxSession);
             //return Mono.empty();
         });

iot-modules/iot-box-websocket-api/src/main/java/com/qiuguo/iot/box/websocket/api/handler/CustomerWebSocketHandler.java

@@ -121,6 +121,7 @@ public class CustomerWebSocketHandler extends BaseWebSocketProcess implements We
         return nlpService.getActionWithLacSingle(userSession.getUserId(), userTalkMessage.getMessage())
                 .defaultIfEmpty(new Actions()).flatMap(actions -> {
                     //处理
+                    userSession.setRequestId(userSession.getRequestId() + 1);
                     return processAction(actions, userSession);
                 });