diff --git a/backend/pom.xml b/backend/pom.xml index a392a88..4996a39 100644 --- a/backend/pom.xml +++ b/backend/pom.xml @@ -183,6 +183,11 @@ tencentcloud-speech-sdk-java 1.0.67 + + com.tencentcloudapi + tencentcloud-sdk-java-asr + 3.1.1470 + diff --git a/backend/src/main/java/com/imeeting/config/RealtimeMeetingWebSocketConfig.java b/backend/src/main/java/com/imeeting/config/RealtimeMeetingWebSocketConfig.java index f940e26..e547594 100644 --- a/backend/src/main/java/com/imeeting/config/RealtimeMeetingWebSocketConfig.java +++ b/backend/src/main/java/com/imeeting/config/RealtimeMeetingWebSocketConfig.java @@ -2,6 +2,9 @@ package com.imeeting.config; import com.imeeting.websocket.RealtimeMeetingProxyWebSocketHandler; import lombok.RequiredArgsConstructor; +import org.springframework.boot.web.embedded.tomcat.TomcatServletWebServerFactory; +import org.springframework.boot.web.server.WebServerFactoryCustomizer; +import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.web.socket.config.annotation.EnableWebSocket; import org.springframework.web.socket.config.annotation.WebSocketConfigurer; @@ -19,4 +22,25 @@ public class RealtimeMeetingWebSocketConfig implements WebSocketConfigurer { registry.addHandler(realtimeMeetingProxyWebSocketHandler, "/ws/meeting/realtime") .setAllowedOriginPatterns("*"); } + + /** + * 关闭 Tomcat 内置的 WebSocket keepalive ping 检测（sessionIdleTimeout）。 + *

+ * Tomcat 默认会对 WebSocket session 设置 sessionIdleTimeout（-1 表示无限，但某些版本默认非 -1）， + * 并通过后台线程定期发送 Ping 帧，若在超时内未收到 Pong 响应，触发 + * code=1011 "keepalive ping timeout" 强制断开。 + * 实时 ASR 场景中，客户端持续发送音频帧，由前端心跳保活， + * 因此显式将 sessionIdleTimeout 设为 -1（无限）。 + *

+ */ + @Bean + public WebServerFactoryCustomizer wsSessionIdleTimeoutCustomizer() { + return factory -> factory.addConnectorCustomizers(connector -> { + // 通过系统属性通知 Tomcat WS 容器关闭 sessionIdleTimeout 检查 + // org.apache.tomcat.websocket.DEFAULT_SESSION_IDLE_TIMEOUT=-1 + if (System.getProperty("org.apache.tomcat.websocket.DEFAULT_SESSION_IDLE_TIMEOUT") == null) { + System.setProperty("org.apache.tomcat.websocket.DEFAULT_SESSION_IDLE_TIMEOUT", "-1"); + } + }); + } } diff --git a/backend/src/main/java/com/imeeting/service/android/legacy/impl/LegacyMeetingAdapterServiceImpl.java b/backend/src/main/java/com/imeeting/service/android/legacy/impl/LegacyMeetingAdapterServiceImpl.java index 9d715d2..6f3c436 100644 --- a/backend/src/main/java/com/imeeting/service/android/legacy/impl/LegacyMeetingAdapterServiceImpl.java +++ b/backend/src/main/java/com/imeeting/service/android/legacy/impl/LegacyMeetingAdapterServiceImpl.java @@ -205,7 +205,7 @@ public class LegacyMeetingAdapterServiceImpl implements LegacyMeetingAdapterServ null, null, null, - null, + true, null, null, List.of() @@ -281,7 +281,7 @@ public class LegacyMeetingAdapterServiceImpl implements LegacyMeetingAdapterServ null, null, null, - null, + true, null, null, List.of() diff --git a/backend/src/main/java/com/imeeting/service/biz/impl/AiModelServiceImpl.java b/backend/src/main/java/com/imeeting/service/biz/impl/AiModelServiceImpl.java index 36e31a6..bdfe5e8 100644 --- a/backend/src/main/java/com/imeeting/service/biz/impl/AiModelServiceImpl.java +++ b/backend/src/main/java/com/imeeting/service/biz/impl/AiModelServiceImpl.java @@ -49,6 +49,8 @@ public class AiModelServiceImpl implements AiModelService { private static final String MEDIA_TENCENT_APP_ID = "tencentAppId"; private static final String MEDIA_TENCENT_SECRET_ID = "tencentSecretId"; private static final String MEDIA_TENCENT_SECRET_KEY = "tencentSecretKey"; + private static final String MEDIA_TENCENT_OFFLINE_MODEL_CODE = "tencentOfflineModelCode"; + private static final String MEDIA_TENCENT_REALTIME_MODEL_CODE = "tencentRealtimeModelCode"; private static final int DEFAULT_SORT_ORDER = 0; private static final String DEFAULT_LLM_API_PATH = "/v1/chat/completions"; private static final String DEFAULT_ANTHROPIC_API_PATH = "/messages"; @@ -956,16 +958,19 @@ public class AiModelServiceImpl implements AiModelService { } Map mediaConfig = dto.getMediaConfig() == null ? Collections.emptyMap() : dto.getMediaConfig(); if (readConfigString(mediaConfig.get(MEDIA_TENCENT_APP_ID)) == null) { - throw new RuntimeException("腾讯实时 ASR 模型必须配置 mediaConfig.tencentAppId"); + throw new RuntimeException("腾讯 ASR 模型必须配置 mediaConfig.tencentAppId"); } if (readConfigString(mediaConfig.get(MEDIA_TENCENT_SECRET_ID)) == null) { - throw new RuntimeException("腾讯实时 ASR 模型必须配置 mediaConfig.tencentSecretId"); + throw new RuntimeException("腾讯 ASR 模型必须配置 mediaConfig.tencentSecretId"); } if (readConfigString(mediaConfig.get(MEDIA_TENCENT_SECRET_KEY)) == null) { - throw new RuntimeException("腾讯实时 ASR 模型必须配置 mediaConfig.tencentSecretKey"); + throw new RuntimeException("腾讯 ASR 模型必须配置 mediaConfig.tencentSecretKey"); } - if (dto.getModelCode() == null || dto.getModelCode().isBlank()) { - throw new RuntimeException("腾讯实时 ASR 模型必须配置 modelCode"); + if (readConfigString(mediaConfig.get(MEDIA_TENCENT_OFFLINE_MODEL_CODE)) == null) { + throw new RuntimeException("腾讯 ASR 模型必须配置 mediaConfig.tencentOfflineModelCode"); + } + if (readConfigString(mediaConfig.get(MEDIA_TENCENT_REALTIME_MODEL_CODE)) == null) { + throw new RuntimeException("腾讯 ASR 模型必须配置 mediaConfig.tencentRealtimeModelCode"); } } diff --git a/backend/src/main/java/com/imeeting/service/biz/impl/AiTaskServiceImpl.java b/backend/src/main/java/com/imeeting/service/biz/impl/AiTaskServiceImpl.java index d61460f..6e6e606 100644 --- a/backend/src/main/java/com/imeeting/service/biz/impl/AiTaskServiceImpl.java +++ b/backend/src/main/java/com/imeeting/service/biz/impl/AiTaskServiceImpl.java @@ -37,6 +37,16 @@ import com.imeeting.support.redis.MeetingLockCache; import com.unisbase.entity.SysUser; import com.unisbase.mapper.SysUserMapper; import com.unisbase.service.SysParamService; +import com.tencentcloudapi.asr.v20190614.AsrClient; +import com.tencentcloudapi.asr.v20190614.models.CreateRecTaskRequest; +import com.tencentcloudapi.asr.v20190614.models.CreateRecTaskResponse; +import com.tencentcloudapi.asr.v20190614.models.DescribeTaskStatusRequest; +import com.tencentcloudapi.asr.v20190614.models.DescribeTaskStatusResponse; +import com.tencentcloudapi.asr.v20190614.models.SentenceDetail; +import com.tencentcloudapi.asr.v20190614.models.TaskStatus; +import com.tencentcloudapi.common.Credential; +import com.tencentcloudapi.common.exception.TencentCloudSDKException; +import com.tencentcloudapi.common.profile.ClientProfile; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Qualifier; @@ -69,6 +79,13 @@ public class AiTaskServiceImpl extends ServiceImpl impleme private static final Duration ASR_QUERY_REQUEST_TIMEOUT = Duration.ofSeconds(30); private static final String DISPATCH_MODE_PARALLEL = "PARALLEL"; private static final String DISPATCH_MODE_SERIAL = "SERIAL"; + private static final String TENCENT_PROVIDER = "tencent"; + private static final String TENCENT_TASK_ID_KEY = "taskId"; + private static final String MEDIA_TENCENT_APP_ID = "tencentAppId"; + private static final String MEDIA_TENCENT_SECRET_ID = "tencentSecretId"; + private static final String MEDIA_TENCENT_SECRET_KEY = "tencentSecretKey"; + private static final String MEDIA_TENCENT_OFFLINE_MODEL_CODE = "tencentOfflineModelCode"; + private static final String TENCENT_ASR_REGION = "ap-guangzhou"; private final MeetingMapper meetingMapper; private final MeetingTranscriptMapper transcriptMapper; @@ -609,6 +626,14 @@ public class AiTaskServiceImpl extends ServiceImpl impleme log.info("[ASR-PROC] 解析ASR模型成功: meetingId={}, asrTaskId={}, asrModelId={}, baseUrl={}", meeting.getId(), taskRecord.getId(), asrModelId, asrModel.getBaseUrl()); + if ("tencent".equalsIgnoreCase(firstNonBlank(asrModel.getProvider()))) { + String transcriptText = processTencentOfflineAsr(meeting, taskRecord, asrModel); + log.info("[ASR-PROC] Tencent offline transcript persisted: meetingId={}, asrTaskId={}, transcriptLength={}", + meeting.getId(), taskRecord.getId(), transcriptText == null ? 0 : transcriptText.length()); + meetingPointsService.recordAsrSuccessCharge(meeting, taskRecord); + return transcriptText; + } + String submitUrl = appendPath(asrModel.getBaseUrl(), "api/v1/asr/transcriptions"); String taskId = taskRecord.getResponseData() != null ? String.valueOf(taskRecord.getResponseData().getOrDefault("task_id", "")) @@ -702,6 +727,124 @@ public class AiTaskServiceImpl extends ServiceImpl impleme return transcriptText; } + protected String processTencentOfflineAsr(Meeting meeting, AiTask taskRecord, AiModelVO asrModel) throws Exception { + Long taskId = readTencentOfflineTaskId(taskRecord); + if (taskId == null) { + taskId = submitTencentOfflineTask(meeting, taskRecord, asrModel); + } + + TaskStatus taskStatus = null; + for (int i = 0; i < 600; i++) { + Thread.sleep(2000); + taskStatus = queryTencentOfflineTask(asrModel, taskId); + if (taskStatus == null) { + throw new RuntimeException("腾讯离线 ASR 查询结果为空"); + } + String status = firstNonBlank(taskStatus.getStatusStr(), ""); + if ("success".equalsIgnoreCase(status)) { + updateAiTaskSuccess(taskRecord, objectMapper.valueToTree(buildTencentTaskStatusSnapshot(taskStatus))); + return saveTencentOfflineTranscripts(meeting, taskStatus.getResultDetail()); + } + if ("failed".equalsIgnoreCase(status)) { + String errorMsg = firstNonBlank(taskStatus.getErrorMsg(), "腾讯离线 ASR 识别失败"); + updateAiTaskFail(taskRecord, errorMsg); + throw new RuntimeException(errorMsg); + } + updateProgress(meeting.getId(), 5, "腾讯离线 ASR 识别中...", 0); + } + throw new RuntimeException("腾讯离线 ASR 轮询超时"); + } + + protected Map buildTencentOfflineCreateRequest(Meeting meeting, AiTask taskRecord, AiModelVO asrModel) { + Map req = new HashMap<>(); + req.put("engineModelType", resolveTencentOfflineModelCode(asrModel)); + req.put("channelNum", 1L); + req.put("resTextFormat", 2L); + req.put("sourceType", 0L); + req.put("url", resolveTencentOfflineAudioUrl(meeting)); + req.put("speakerDiarization", resolveTencentSpeakerDiarization(taskRecord)); + req.put("speakerNumber", 0L); + String hotwordList = buildTencentHotwordList(taskRecord); + if (hotwordList != null) { + req.put("hotwordList", hotwordList); + } + return req; + } + + protected Long submitTencentOfflineTask(Meeting meeting, AiTask taskRecord, AiModelVO asrModel) throws Exception { + updateProgress(meeting.getId(), 5, "提交腾讯离线 ASR 任务...", 0); + meetingPointsService.assertSufficientPointsBeforeAsrSubmit(meeting, taskRecord); + + Map reqSnapshot = buildTencentOfflineCreateRequest(meeting, taskRecord, asrModel); + taskRecord.setRequestData(reqSnapshot); + this.updateById(taskRecord); + + CreateRecTaskRequest request = new CreateRecTaskRequest(); + request.setEngineModelType(String.valueOf(reqSnapshot.get("engineModelType"))); + request.setChannelNum(longValue(reqSnapshot.get("channelNum"))); + request.setResTextFormat(longValue(reqSnapshot.get("resTextFormat"))); + request.setSourceType(longValue(reqSnapshot.get("sourceType"))); + request.setUrl(String.valueOf(reqSnapshot.get("url"))); + request.setSpeakerDiarization(longValue(reqSnapshot.get("speakerDiarization"))); + request.setSpeakerNumber(longValue(reqSnapshot.get("speakerNumber"))); + String hotwordList = stringValue(reqSnapshot.get("hotwordList")); + if (hotwordList != null && !hotwordList.isBlank()) { + request.setHotwordList(hotwordList); + } + + CreateRecTaskResponse response = buildTencentOfflineAsrClient(asrModel).CreateRecTask(request); + if (response == null || response.getData() == null || response.getData().getTaskId() == null) { + throw new RuntimeException("腾讯离线 ASR 提交失败：未返回 TaskId"); + } + Long taskId = response.getData().getTaskId(); + writeTencentOfflineTaskId(taskRecord, taskId); + this.updateById(taskRecord); + return taskId; + } + + protected TaskStatus queryTencentOfflineTask(AiModelVO asrModel, Long taskId) throws TencentCloudSDKException { + DescribeTaskStatusRequest request = new DescribeTaskStatusRequest(); + request.setTaskId(taskId); + DescribeTaskStatusResponse response = buildTencentOfflineAsrClient(asrModel).DescribeTaskStatus(request); + return response == null ? null : response.getData(); + } + + protected String saveTencentOfflineTranscripts(Meeting meeting, SentenceDetail[] resultDetail) { + transcriptMapper.delete(new LambdaQueryWrapper().eq(MeetingTranscript::getMeetingId, meeting.getId())); + + if (resultDetail == null || resultDetail.length == 0) { + return ""; + } + + StringBuilder sb = new StringBuilder(); + int order = 0; + for (SentenceDetail detail : resultDetail) { + if (detail == null) { + continue; + } + String content = firstNonBlank(detail.getFinalSentence(), detail.getWrittenText(), ""); + if (content == null || content.isBlank()) { + continue; + } + String speakerId = String.valueOf(detail.getSpeakerId() == null ? 0L : detail.getSpeakerId()); + String speakerName = "未知说话人" + speakerId; + + MeetingTranscript mt = new MeetingTranscript(); + mt.setMeetingId(meeting.getId()); + mt.setSpeakerId(speakerId); + mt.setSpeakerName(speakerName); + mt.setContent(content.trim()); + fillTencentTranscriptTime(mt, detail); + mt.setSortOrder(order++); + transcriptMapper.insert(mt); + sb.append(speakerName).append(": ").append(mt.getContent()).append("\n"); + } + if (order > 0) { + meetingTranscriptFileService.initializeTranscriptFileIfAbsent(meeting.getId()); + } + return sb.toString(); + } + private Map buildAsrRequest(Meeting meeting, AiTask taskRecord, AiModelVO asrModel) { Map req = new HashMap<>(); String rawAudioUrl = meeting.getAudioUrl(); @@ -1417,6 +1560,102 @@ public class AiTaskServiceImpl extends ServiceImpl impleme } } + private AsrClient buildTencentOfflineAsrClient(AiModelVO asrModel) { + Map mediaConfig = asrModel.getMediaConfig() == null ? Map.of() : asrModel.getMediaConfig(); + String secretId = requireTencentMediaConfig(mediaConfig, MEDIA_TENCENT_SECRET_ID); + String secretKey = requireTencentMediaConfig(mediaConfig, MEDIA_TENCENT_SECRET_KEY); + Credential credential = new Credential(secretId, secretKey); + ClientProfile profile = new ClientProfile(); + return new AsrClient(credential, TENCENT_ASR_REGION, profile); + } + + private String requireTencentMediaConfig(Map mediaConfig, String key) { + String value = stringValue(mediaConfig.get(key)); + if (value == null || value.isBlank()) { + throw new RuntimeException("腾讯离线 ASR 缺少配置: " + key); + } + return value.trim(); + } + + private Long resolveTencentSpeakerDiarization(AiTask taskRecord) { + Object useSpkObj = taskRecord.getTaskConfig() == null ? null : taskRecord.getTaskConfig().get("useSpkId"); + return "1".equals(String.valueOf(useSpkObj)) ? 1L : 0L; + } + + private String buildTencentHotwordList(AiTask taskRecord) { + if (taskRecord == null || taskRecord.getTaskConfig() == null) { + return null; + } + Object hotWordsObj = taskRecord.getTaskConfig().get("hotWords"); + if (!(hotWordsObj instanceof List words) || words.isEmpty()) { + return null; + } + return words.stream() + .filter(Objects::nonNull) + .map(String::valueOf) + .map(String::trim) + .filter(word -> !word.isBlank()) + .map(word -> word + "|5") + .collect(Collectors.joining(",")); + } + + private String resolveTencentOfflineAudioUrl(Meeting meeting) { + if (meeting == null || meeting.getAudioUrl() == null || meeting.getAudioUrl().isBlank()) { + throw new RuntimeException("腾讯离线 ASR 缺少音频地址"); + } + String audioUrl = meeting.getAudioUrl().trim(); + if (audioUrl.startsWith("http://") || audioUrl.startsWith("https://")) { + return audioUrl; + } + return serverBaseUrl + (audioUrl.startsWith("/") ? "" : "/") + audioUrl; + } + + private void writeTencentOfflineTaskId(AiTask taskRecord, Long taskId) { + Map responseData = taskRecord.getResponseData() == null + ? new HashMap<>() + : new HashMap<>(taskRecord.getResponseData()); + responseData.put(TENCENT_TASK_ID_KEY, taskId); + taskRecord.setResponseData(responseData); + } + + private Long readTencentOfflineTaskId(AiTask taskRecord) { + if (taskRecord == null || taskRecord.getResponseData() == null) { + return null; + } + Object taskId = taskRecord.getResponseData().get(TENCENT_TASK_ID_KEY); + return longValue(taskId); + } + + private Map buildTencentTaskStatusSnapshot(TaskStatus taskStatus) { + Map snapshot = new HashMap<>(); + snapshot.put("taskId", taskStatus.getTaskId()); + snapshot.put("status", taskStatus.getStatus()); + snapshot.put("statusStr", taskStatus.getStatusStr()); + snapshot.put("result", taskStatus.getResult()); + snapshot.put("errorMsg", taskStatus.getErrorMsg()); + snapshot.put("audioDuration", taskStatus.getAudioDuration()); + return snapshot; + } + + private void fillTencentTranscriptTime(MeetingTranscript transcript, SentenceDetail detail) { + if (transcript == null || detail == null) { + return; + } + int startTime = detail.getStartMs() == null ? 0 : detail.getStartMs().intValue(); + int endTime = detail.getEndMs() == null ? startTime : detail.getEndMs().intValue(); + transcript.setStartTime(startTime); + transcript.setEndTime(endTime); + } + + private String resolveTencentOfflineModelCode(AiModelVO asrModel) { + Map mediaConfig = asrModel == null || asrModel.getMediaConfig() == null ? Map.of() : asrModel.getMediaConfig(); + String offlineModelCode = stringValue(mediaConfig.get(MEDIA_TENCENT_OFFLINE_MODEL_CODE)); + if (offlineModelCode != null && !offlineModelCode.isBlank()) { + return offlineModelCode.trim(); + } + return asrModel == null ? null : asrModel.getModelCode(); + } + private AiModelVO resolveAsrModelForRevision(AiTask asrTask) { if (asrTask == null || asrTask.getTaskConfig() == null) { return null; diff --git a/backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSocketSessionServiceImpl.java b/backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSocketSessionServiceImpl.java index b15fcf7..e7603e2 100644 --- a/backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSocketSessionServiceImpl.java +++ b/backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSocketSessionServiceImpl.java @@ -26,6 +26,8 @@ import java.util.UUID; public class RealtimeMeetingSocketSessionServiceImpl implements RealtimeMeetingSocketSessionService { private static final String WS_PATH = "/ws/meeting/realtime"; + private static final String TENCENT_PROVIDER = "tencent"; + private static final String MEDIA_TENCENT_REALTIME_MODEL_CODE = "tencentRealtimeModelCode"; private final RealtimeMeetingSocketSessionCache socketSessionCache; private final MeetingAccessService meetingAccessService; @@ -87,7 +89,7 @@ public class RealtimeMeetingSocketSessionServiceImpl implements RealtimeMeetingS sessionData.setAsrModelId(asrModelId); sessionData.setProvider(realtimeAsrChannelFactory.normalizeProvider(asrModel.getProvider())); sessionData.setTargetWsUrl(targetWsUrl); - sessionData.setModelCode(asrModel.getModelCode()); + sessionData.setModelCode(resolveRealtimeModelCode(asrModel)); sessionData.setMediaConfig(asrModel.getMediaConfig()); String sessionToken = UUID.randomUUID().toString().replace("-", ""); @@ -111,6 +113,25 @@ public class RealtimeMeetingSocketSessionServiceImpl implements RealtimeMeetingS return vo; } + private String resolveRealtimeModelCode(AiModelVO asrModel) { + if (asrModel == null) { + return null; + } + if (!TENCENT_PROVIDER.equalsIgnoreCase(asrModel.getProvider())) { + return asrModel.getModelCode(); + } + Map mediaConfig = asrModel.getMediaConfig(); + if (mediaConfig == null) { + return asrModel.getModelCode(); + } + Object realtimeModelCode = mediaConfig.get(MEDIA_TENCENT_REALTIME_MODEL_CODE); + if (realtimeModelCode == null) { + return asrModel.getModelCode(); + } + String value = String.valueOf(realtimeModelCode).trim(); + return value.isEmpty() ? asrModel.getModelCode() : value; + } + @Override public RealtimeSocketSessionData getSessionData(String sessionToken) { return socketSessionCache.get(sessionToken); diff --git a/backend/src/main/java/com/imeeting/service/realtime/impl/LocalRealtimeAsrChannel.java b/backend/src/main/java/com/imeeting/service/realtime/impl/LocalRealtimeAsrChannel.java index 5b03acc..bc7f758 100644 --- a/backend/src/main/java/com/imeeting/service/realtime/impl/LocalRealtimeAsrChannel.java +++ b/backend/src/main/java/com/imeeting/service/realtime/impl/LocalRealtimeAsrChannel.java @@ -513,7 +513,6 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel { log.info("上游 ASR websocket 已关闭：meetingId={}, sessionId={}, code={}, reason={}", context.getMeetingId(), currentConnectionId(context), statusCode, reason); context.getChannelState().remove(STATE_UPSTREAM_SOCKET); - context.getCallback().removeMeetingSession(context.getMeetingId()); context.getCallback().sendFrontendError(context.getMeetingId(), "REALTIME_UPSTREAM_CLOSED", reason == null || reason.isBlank() ? "上游 ASR WebSocket 已断开" : "上游 ASR WebSocket 已断开: " + reason); @@ -526,13 +525,13 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel { log.error("上游 ASR websocket 异常：meetingId={}, sessionId={}, upstream={}", context.getMeetingId(), currentConnectionId(context), context.getTargetWsUrl(), error); context.getChannelState().remove(STATE_UPSTREAM_SOCKET); - context.getCallback().removeMeetingSession(context.getMeetingId()); context.getCallback().sendFrontendError(context.getMeetingId(), "REALTIME_UPSTREAM_ERROR", error == null || error.getMessage() == null || error.getMessage().isBlank() ? "上游 ASR WebSocket 连接异常" : "上游 ASR WebSocket 连接异常: " + error.getMessage()); context.getCallback().closeFrontend(context.getMeetingId(), CloseStatus.SERVER_ERROR); + context.getCallback().removeMeetingSession(context.getMeetingId()); } } } diff --git a/backend/src/main/java/com/imeeting/service/realtime/impl/RealtimeMeetingTranscriptCacheServiceImpl.java b/backend/src/main/java/com/imeeting/service/realtime/impl/RealtimeMeetingTranscriptCacheServiceImpl.java index 235b14a..9ca0098 100644 --- a/backend/src/main/java/com/imeeting/service/realtime/impl/RealtimeMeetingTranscriptCacheServiceImpl.java +++ b/backend/src/main/java/com/imeeting/service/realtime/impl/RealtimeMeetingTranscriptCacheServiceImpl.java @@ -206,7 +206,7 @@ public class RealtimeMeetingTranscriptCacheServiceImpl implements RealtimeMeetin return item.getSpeakerName().trim(); } String speakerId = resolveSpeakerId(item); - return speakerId == null || speakerId.isBlank() ? null : speakerId; + return speakerId == null || speakerId.isBlank() ? null : "未知说话人" + speakerId; } private String resolveSpeakerId(JsonNode sentence) { diff --git a/backend/src/main/java/com/imeeting/websocket/RealtimeMeetingProxyWebSocketHandler.java b/backend/src/main/java/com/imeeting/websocket/RealtimeMeetingProxyWebSocketHandler.java index d19ac61..c5aba29 100644 --- a/backend/src/main/java/com/imeeting/websocket/RealtimeMeetingProxyWebSocketHandler.java +++ b/backend/src/main/java/com/imeeting/websocket/RealtimeMeetingProxyWebSocketHandler.java @@ -42,25 +42,25 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl private static final String ATTR_MEETING_ID = "meetingId"; private static final String ATTR_TARGET_WS_URL = "targetWsUrl"; - private static final String ATTR_PROVIDER = "provider"; + private static final String ATTR_PROVIDER = "provider"; private static final String ATTR_FRONTEND_TEXT_COUNT = "frontendTextCount"; private static final String ATTR_FRONTEND_BINARY_COUNT = "frontendBinaryCount"; - private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); private final RealtimeMeetingSocketSessionService realtimeMeetingSocketSessionService; private final RealtimeMeetingSessionStateService realtimeMeetingSessionStateService; private final RealtimeMeetingAudioStorageService realtimeMeetingAudioStorageService; - private final RealtimeMeetingTranscriptCacheService realtimeMeetingTranscriptCacheService; - private final RealtimeAsrChannelFactory realtimeAsrChannelFactory; - private final ConcurrentMap meetingSessions = new ConcurrentHashMap<>(); - private final ConcurrentMap meetingLocks = new ConcurrentHashMap<>(); + private final RealtimeMeetingTranscriptCacheService realtimeMeetingTranscriptCacheService; + private final RealtimeAsrChannelFactory realtimeAsrChannelFactory; + private final ConcurrentMap meetingSessions = new ConcurrentHashMap<>(); + private final ConcurrentMap meetingLocks = new ConcurrentHashMap<>(); @Override public void afterConnectionEstablished(WebSocketSession session) throws Exception { String sessionToken = extractQueryParam(session.getUri(), "sessionToken"); RealtimeSocketSessionData sessionData = realtimeMeetingSocketSessionService.getSessionData(sessionToken); if (sessionData == null) { - log.warn("实时会议 websocket 拒绝连接：会话令牌无效，sessionId={}", session.getId()); + log.warn("实时会议 websocket 拒绝连接：会话令牌无效，sessionId={}", session.getId()); session.close(CloseStatus.POLICY_VIOLATION.withReason("实时 Socket 会话无效")); return; } @@ -69,65 +69,70 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl new ConcurrentWebSocketSessionDecorator(session, (int) Duration.ofSeconds(15).toMillis(), 1024 * 1024); session.getAttributes().put(ATTR_MEETING_ID, sessionData.getMeetingId()); session.getAttributes().put(ATTR_TARGET_WS_URL, sessionData.getTargetWsUrl()); - session.getAttributes().put(ATTR_PROVIDER, sessionData.getProvider()); + session.getAttributes().put(ATTR_PROVIDER, sessionData.getProvider()); session.getAttributes().put(ATTR_FRONTEND_TEXT_COUNT, new AtomicInteger()); session.getAttributes().put(ATTR_FRONTEND_BINARY_COUNT, new AtomicInteger()); realtimeMeetingAudioStorageService.openSession(sessionData.getMeetingId(), session.getId()); - log.info("实时会议 websocket 已接入：meetingId={}, sessionId={}, provider={}, upstream={}", - sessionData.getMeetingId(), session.getId(), sessionData.getProvider(), sessionData.getTargetWsUrl()); + log.info("实时会议 websocket 已接入：meetingId={}, sessionId={}, provider={}, upstream={}", + sessionData.getMeetingId(), session.getId(), sessionData.getProvider(), sessionData.getTargetWsUrl()); - attachFrontendSession(sessionData, session, frontendSession); + attachFrontendSession(sessionData, session, frontendSession); } @Override protected void handleTextMessage(WebSocketSession session, TextMessage message) { - MeetingChannelSession meetingSession = getMeetingSession(session); - if (meetingSession == null || !meetingSession.isChannelOpen()) { - log.warn("前端文本消息已忽略：上游 ASR 连接不可用，meetingId={}, sessionId={}", + // 过滤前端发来的心跳保活消息，不转发给上游 ASR 服务 + if (looksLikeKeepaliveMessage(message.getPayload())) { + log.debug("Frontend keepalive received, ignored: meetingId={}, sessionId={}", + session.getAttributes().get(ATTR_MEETING_ID), session.getId()); + return; + } + MeetingChannelSession meetingSession = getMeetingSession(session); + if (meetingSession == null || !meetingSession.isChannelOpen()) { + log.warn("前端文本消息已忽略：上游 ASR 连接不可用，meetingId={}, sessionId={}", session.getAttributes().get(ATTR_MEETING_ID), session.getId()); return; } int count = nextCount(session, ATTR_FRONTEND_TEXT_COUNT); - String payload = message.getPayload(); - log.info("前端文本 -> ASR 渠道：meetingId={}, sessionId={}, provider={}, count={}, payload={}", - session.getAttributes().get(ATTR_MEETING_ID), session.getId(), session.getAttributes().get(ATTR_PROVIDER), count, summarizeText(payload)); - meetingSession.channel.handleFrontendText(meetingSession.context, payload); + String payload = message.getPayload(); + log.info("前端文本 -> ASR 渠道：meetingId={}, sessionId={}, provider={}, count={}, payload={}", + session.getAttributes().get(ATTR_MEETING_ID), session.getId(), session.getAttributes().get(ATTR_PROVIDER), count, summarizeText(payload)); + meetingSession.channel.handleFrontendText(meetingSession.context, payload); } @Override protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) { - MeetingChannelSession meetingSession = getMeetingSession(session); - if (meetingSession == null || !meetingSession.isChannelOpen()) { - log.warn("前端音频帧已忽略：上游 ASR 连接不可用，meetingId={}, sessionId={}", + MeetingChannelSession meetingSession = getMeetingSession(session); + if (meetingSession == null || !meetingSession.isChannelOpen()) { + log.warn("前端音频帧已忽略：上游 ASR 连接不可用，meetingId={}, sessionId={}", session.getAttributes().get(ATTR_MEETING_ID), session.getId()); return; } int count = nextCount(session, ATTR_FRONTEND_BINARY_COUNT); int bytes = message.getPayloadLength(); if (shouldLogBinaryFrame(count)) { - log.info("前端音频帧 -> ASR 渠道：meetingId={}, sessionId={}, provider={}, count={}, bytes={}", - session.getAttributes().get(ATTR_MEETING_ID), session.getId(), session.getAttributes().get(ATTR_PROVIDER), count, bytes); + log.info("前端音频帧 -> ASR 渠道：meetingId={}, sessionId={}, provider={}, count={}, bytes={}", + session.getAttributes().get(ATTR_MEETING_ID), session.getId(), session.getAttributes().get(ATTR_PROVIDER), count, bytes); } byte[] payload = toByteArray(message.getPayload()); realtimeMeetingAudioStorageService.append(session.getId(), payload); - meetingSession.channel.handleFrontendBinary(meetingSession.context, payload); + meetingSession.channel.handleFrontendBinary(meetingSession.context, payload); } @Override protected void handlePongMessage(WebSocketSession session, PongMessage message) { - if (getMeetingSession(session) == null) { - return; - } - log.debug("前端 pong 已在本地忽略：meetingId={}, sessionId={}, bytes={}", - session.getAttributes().get(ATTR_MEETING_ID), session.getId(), message.getPayloadLength()); + // Pong 是浏览器对服务端（Tomcat）发出 Ping 帧的回应，代理层在此消化即可，无需转发给上游 ASR。 + // 转发 Pong 给上游没有语义意义，且可能引起上游协议混乱。 + log.debug("Frontend pong received (keepalive): meetingId={}, sessionId={}", + session.getAttributes().get(ATTR_MEETING_ID), session.getId()); } @Override public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception { - log.error("实时会议 websocket 传输异常：meetingId={}, sessionId={}, upstream={}", + log.error("实时会议 websocket 传输异常：meetingId={}, sessionId={}, upstream={}", session.getAttributes().get(ATTR_MEETING_ID), session.getId(), session.getAttributes().get(ATTR_TARGET_WS_URL), exception); - detachFrontend(session); - realtimeMeetingAudioStorageService.closeSession(session.getId()); + detachFrontend(session); + realtimeMeetingAudioStorageService.closeSession(session.getId()); if (session.isOpen()) { session.close(CloseStatus.SERVER_ERROR); } @@ -135,159 +140,159 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl @Override public void afterConnectionClosed(WebSocketSession session, CloseStatus status) { - log.info("实时会议 websocket 已关闭：meetingId={}, sessionId={}, code={}, reason={}", + log.info("实时会议 websocket 已关闭：meetingId={}, sessionId={}, code={}, reason={}", session.getAttributes().get(ATTR_MEETING_ID), session.getId(), status.getCode(), status.getReason()); Object meetingIdValue = session.getAttributes().get(ATTR_MEETING_ID); if (meetingIdValue instanceof Long meetingId) { - detachFrontend(meetingId, session.getId()); + detachFrontend(meetingId, session.getId()); realtimeMeetingSessionStateService.pauseByDisconnect(meetingId, session.getId()); } realtimeMeetingAudioStorageService.closeSession(session.getId()); } - public void closeMeetingSession(Long meetingId) { - if (meetingId == null) { - return; - } - MeetingChannelSession meetingSession = meetingSessions.get(meetingId); - if (meetingSession == null || meetingSession.channel == null) { - return; - } - meetingSession.channel.closeMeeting(meetingSession.context); - } - - private void attachFrontendSession(RealtimeSocketSessionData sessionData, - WebSocketSession rawSession, - ConcurrentWebSocketSessionDecorator frontendSession) throws Exception { - Long meetingId = sessionData.getMeetingId(); - MeetingChannelSession meetingSession; - boolean reused = false; - synchronized (lockForMeeting(meetingId)) { - meetingSession = meetingSessions.get(meetingId); - if (meetingSession != null && meetingSession.isChannelOpen()) { - String previousSessionId = meetingSession.context.getRawSession() == null - ? null - : meetingSession.context.getRawSession().getId(); - meetingSession.clearFrontendIfClosed(); - if (previousSessionId != null && !meetingSession.hasOpenFrontend()) { - realtimeMeetingSessionStateService.pauseByDisconnect(meetingId, previousSessionId); + public void closeMeetingSession(Long meetingId) { + if (meetingId == null) { + return; } - if (meetingSession.hasOpenFrontend()) { - sendFrontendError(frontendSession, "REALTIME_ACTIVE_CONNECTION_EXISTS", "当前会议已有活跃前端连接"); - frontendSession.close(CloseStatus.POLICY_VIOLATION.withReason("已存在活跃的前端连接")); - realtimeMeetingAudioStorageService.closeSession(rawSession.getId()); - return; + MeetingChannelSession meetingSession = meetingSessions.get(meetingId); + if (meetingSession == null || meetingSession.channel == null) { + return; } - if (!realtimeMeetingSessionStateService.activate(meetingId, rawSession.getId())) { - sendFrontendError(frontendSession, "REALTIME_ACTIVE_CONNECTION_REJECTED", "当前状态下无法继续会议"); - frontendSession.close(CloseStatus.POLICY_VIOLATION.withReason("当前状态下无法继续会议")); - realtimeMeetingAudioStorageService.closeSession(rawSession.getId()); - return; + meetingSession.channel.closeMeeting(meetingSession.context); + } + + private void attachFrontendSession(RealtimeSocketSessionData sessionData, + WebSocketSession rawSession, + ConcurrentWebSocketSessionDecorator frontendSession) throws Exception { + Long meetingId = sessionData.getMeetingId(); + MeetingChannelSession meetingSession; + boolean reused = false; + synchronized (lockForMeeting(meetingId)) { + meetingSession = meetingSessions.get(meetingId); + if (meetingSession != null && meetingSession.isChannelOpen()) { + String previousSessionId = meetingSession.context.getRawSession() == null + ? null + : meetingSession.context.getRawSession().getId(); + meetingSession.clearFrontendIfClosed(); + if (previousSessionId != null && !meetingSession.hasOpenFrontend()) { + realtimeMeetingSessionStateService.pauseByDisconnect(meetingId, previousSessionId); + } + if (meetingSession.hasOpenFrontend()) { + sendFrontendError(frontendSession, "REALTIME_ACTIVE_CONNECTION_EXISTS", "当前会议已有活跃前端连接"); + frontendSession.close(CloseStatus.POLICY_VIOLATION.withReason("已存在活跃的前端连接")); + realtimeMeetingAudioStorageService.closeSession(rawSession.getId()); + return; + } + if (!realtimeMeetingSessionStateService.activate(meetingId, rawSession.getId())) { + sendFrontendError(frontendSession, "REALTIME_ACTIVE_CONNECTION_REJECTED", "当前状态下无法继续会议"); + frontendSession.close(CloseStatus.POLICY_VIOLATION.withReason("当前状态下无法继续会议")); + realtimeMeetingAudioStorageService.closeSession(rawSession.getId()); + return; + } + meetingSession.bindFrontend(rawSession, frontendSession); + reused = true; + } else { + RealtimeAsrChannel channel = realtimeAsrChannelFactory.getRequired(sessionData.getProvider()); + RealtimeAsrChannelContext context = new RealtimeAsrChannelContext(); + context.setMeetingId(meetingId); + context.setProvider(realtimeAsrChannelFactory.normalizeProvider(sessionData.getProvider())); + context.setTargetWsUrl(sessionData.getTargetWsUrl()); + context.setCallback(new HandlerChannelCallback()); + context.bindFrontendSession(rawSession, frontendSession); + context.getChannelState().put("modelCode", sessionData.getModelCode()); + context.getChannelState().put("mediaConfig", sessionData.getMediaConfig()); + meetingSession = new MeetingChannelSession(meetingId, channel, context); + meetingSessions.put(meetingId, meetingSession); + } } - meetingSession.bindFrontend(rawSession, frontendSession); - reused = true; - } else { - RealtimeAsrChannel channel = realtimeAsrChannelFactory.getRequired(sessionData.getProvider()); - RealtimeAsrChannelContext context = new RealtimeAsrChannelContext(); - context.setMeetingId(meetingId); - context.setProvider(realtimeAsrChannelFactory.normalizeProvider(sessionData.getProvider())); - context.setTargetWsUrl(sessionData.getTargetWsUrl()); - context.setCallback(new HandlerChannelCallback()); - context.bindFrontendSession(rawSession, frontendSession); - context.getChannelState().put("modelCode", sessionData.getModelCode()); - context.getChannelState().put("mediaConfig", sessionData.getMediaConfig()); - meetingSession = new MeetingChannelSession(meetingId, channel, context); - meetingSessions.put(meetingId, meetingSession); - } - } - if (reused) { - sendProxyReady(frontendSession); - replayCachedMessages(meetingId, frontendSession); - return; - } - - try { - meetingSession.channel.connect(meetingSession.context); - } catch (InterruptedException ex) { - Thread.currentThread().interrupt(); - removeMeetingSession(meetingId, meetingSession); - log.error("连接上游 ASR websocket 时被中断：meetingId={}, sessionId={}", meetingId, rawSession.getId(), ex); - sendFrontendError(frontendSession, "REALTIME_UPSTREAM_CONNECT_INTERRUPTED", "连接上游 ASR 服务时被中断"); - realtimeMeetingAudioStorageService.closeSession(rawSession.getId()); - frontendSession.close(CloseStatus.SERVER_ERROR.withReason("连接上游服务时被中断")); - } catch (Exception ex) { - removeMeetingSession(meetingId, meetingSession); - log.warn("连接上游 ASR websocket 失败：meetingId={}, provider={}, target={}", - meetingId, sessionData.getProvider(), sessionData.getTargetWsUrl(), ex); - sendFrontendError(frontendSession, "REALTIME_UPSTREAM_CONNECT_FAILED", "连接上游 ASR 服务失败"); - realtimeMeetingAudioStorageService.closeSession(rawSession.getId()); - frontendSession.close(CloseStatus.SERVER_ERROR.withReason("连接 ASR WebSocket 失败")); - } - } - - private void replayCachedMessages(Long meetingId, ConcurrentWebSocketSessionDecorator frontendSession) { - try { - if (!frontendSession.isOpen()) { - return; - } - for (RealtimeMeetingTranscriptCacheItem item : realtimeMeetingTranscriptCacheService.listOrderedItems(meetingId)) { - frontendSession.sendMessage(new TextMessage(LocalRealtimeAsrChannel.buildFrontendTranscriptMessage(item))); - } - } catch (Exception ex) { - log.warn("回放缓存转写消息失败：meetingId={}", meetingId, ex); - } - } - - private void sendProxyReady(ConcurrentWebSocketSessionDecorator frontendSession) throws Exception { - if (frontendSession.isOpen()) { - frontendSession.sendMessage(new TextMessage("{\"type\":\"proxy_ready\"}")); - } - } - - private void detachFrontend(WebSocketSession session) { - Object meetingIdValue = session.getAttributes().get(ATTR_MEETING_ID); - if (meetingIdValue instanceof Long meetingId) { - detachFrontend(meetingId, session.getId()); - } - } - - private void detachFrontend(Long meetingId, String sessionId) { - MeetingChannelSession meetingSession = meetingSessions.get(meetingId); - if (meetingSession == null) { - return; + if (reused) { + sendProxyReady(frontendSession); + replayCachedMessages(meetingId, frontendSession); + return; } - synchronized (lockForMeeting(meetingId)) { - if (meetingSession.context.getRawSession() != null && meetingSession.context.getRawSession().getId().equals(sessionId)) { - meetingSession.channel.onFrontendDetached(meetingSession.context); - } - meetingSession.detachFrontend(sessionId); - } - } - private MeetingChannelSession getMeetingSession(WebSocketSession session) { - Object meetingIdValue = session.getAttributes().get(ATTR_MEETING_ID); - if (!(meetingIdValue instanceof Long meetingId)) { - return null; - } - return meetingSessions.get(meetingId); - } - - void removeMeetingSession(Long meetingId) { - synchronized (lockForMeeting(meetingId)) { - meetingSessions.remove(meetingId); - } - } - - void removeMeetingSession(Long meetingId, MeetingChannelSession meetingSession) { - synchronized (lockForMeeting(meetingId)) { - meetingSessions.remove(meetingId, meetingSession); + try { + meetingSession.channel.connect(meetingSession.context); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + removeMeetingSession(meetingId, meetingSession); + log.error("连接上游 ASR websocket 时被中断：meetingId={}, sessionId={}", meetingId, rawSession.getId(), ex); + sendFrontendError(frontendSession, "REALTIME_UPSTREAM_CONNECT_INTERRUPTED", "连接上游 ASR 服务时被中断"); + realtimeMeetingAudioStorageService.closeSession(rawSession.getId()); + frontendSession.close(CloseStatus.SERVER_ERROR.withReason("连接上游服务时被中断")); + } catch (Exception ex) { + removeMeetingSession(meetingId, meetingSession); + log.warn("连接上游 ASR websocket 失败：meetingId={}, provider={}, target={}", + meetingId, sessionData.getProvider(), sessionData.getTargetWsUrl(), ex); + sendFrontendError(frontendSession, "REALTIME_UPSTREAM_CONNECT_FAILED", "连接上游 ASR 服务失败"); + realtimeMeetingAudioStorageService.closeSession(rawSession.getId()); + frontendSession.close(CloseStatus.SERVER_ERROR.withReason("连接 ASR WebSocket 失败")); } } - private Object lockForMeeting(Long meetingId) { - return meetingLocks.computeIfAbsent(meetingId, ignored -> new Object()); - } + private void replayCachedMessages(Long meetingId, ConcurrentWebSocketSessionDecorator frontendSession) { + try { + if (!frontendSession.isOpen()) { + return; + } + for (RealtimeMeetingTranscriptCacheItem item : realtimeMeetingTranscriptCacheService.listOrderedItems(meetingId)) { + frontendSession.sendMessage(new TextMessage(LocalRealtimeAsrChannel.buildFrontendTranscriptMessage(item))); + } + } catch (Exception ex) { + log.warn("回放缓存转写消息失败：meetingId={}", meetingId, ex); + } + } + + private void sendProxyReady(ConcurrentWebSocketSessionDecorator frontendSession) throws Exception { + if (frontendSession.isOpen()) { + frontendSession.sendMessage(new TextMessage("{\"type\":\"proxy_ready\"}")); + } + } + + private void detachFrontend(WebSocketSession session) { + Object meetingIdValue = session.getAttributes().get(ATTR_MEETING_ID); + if (meetingIdValue instanceof Long meetingId) { + detachFrontend(meetingId, session.getId()); + } + } + + private void detachFrontend(Long meetingId, String sessionId) { + MeetingChannelSession meetingSession = meetingSessions.get(meetingId); + if (meetingSession == null) { + return; + } + synchronized (lockForMeeting(meetingId)) { + if (meetingSession.context.getRawSession() != null && meetingSession.context.getRawSession().getId().equals(sessionId)) { + meetingSession.channel.onFrontendDetached(meetingSession.context); + } + meetingSession.detachFrontend(sessionId); + } + } + + private MeetingChannelSession getMeetingSession(WebSocketSession session) { + Object meetingIdValue = session.getAttributes().get(ATTR_MEETING_ID); + if (!(meetingIdValue instanceof Long meetingId)) { + return null; + } + return meetingSessions.get(meetingId); + } + + void removeMeetingSession(Long meetingId) { + synchronized (lockForMeeting(meetingId)) { + meetingSessions.remove(meetingId); + } + } + + void removeMeetingSession(Long meetingId, MeetingChannelSession meetingSession) { + synchronized (lockForMeeting(meetingId)) { + meetingSessions.remove(meetingId, meetingSession); + } + } + + private Object lockForMeeting(Long meetingId) { + return meetingLocks.computeIfAbsent(meetingId, ignored -> new Object()); + } private String extractQueryParam(URI uri, String key) { if (uri == null || uri.getQuery() == null || uri.getQuery().isBlank()) { @@ -316,18 +321,18 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl return 0; } - private void sendFrontendError(ConcurrentWebSocketSessionDecorator frontendSession, String code, String message) { - try { - if (!frontendSession.isOpen()) { - return; - } - Map payload = new HashMap<>(); - payload.put("type", "error"); - payload.put("code", code); - payload.put("message", message); - frontendSession.sendMessage(new TextMessage(OBJECT_MAPPER.writeValueAsString(payload))); - } catch (Exception ex) { - log.warn("向前端发送实时代理错误消息失败：code={}", code, ex); + private void sendFrontendError(ConcurrentWebSocketSessionDecorator frontendSession, String code, String message) { + try { + if (!frontendSession.isOpen()) { + return; + } + Map payload = new HashMap<>(); + payload.put("type", "error"); + payload.put("code", code); + payload.put("message", message); + frontendSession.sendMessage(new TextMessage(OBJECT_MAPPER.writeValueAsString(payload))); + } catch (Exception ex) { + log.warn("向前端发送实时代理错误消息失败：code={}", code, ex); } } @@ -346,108 +351,116 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl return normalized.substring(0, 240) + "..."; } - static final class MeetingChannelSession { - private final Long meetingId; - private final RealtimeAsrChannel channel; - private final RealtimeAsrChannelContext context; - - private MeetingChannelSession(Long meetingId, RealtimeAsrChannel channel, RealtimeAsrChannelContext context) { - this.meetingId = meetingId; - this.channel = channel; - this.context = context; + private boolean looksLikeKeepaliveMessage(String payload) { + if (payload == null || payload.isBlank()) { + return false; + } + String normalized = payload.replaceAll("\\s+", ""); + return normalized.contains("\"type\":\"keepalive\""); } - private void bindFrontend(WebSocketSession rawSession, ConcurrentWebSocketSessionDecorator frontendSession) { - context.bindFrontendSession(rawSession, frontendSession); + static final class MeetingChannelSession { + private final Long meetingId; + private final RealtimeAsrChannel channel; + private final RealtimeAsrChannelContext context; + + private MeetingChannelSession(Long meetingId, RealtimeAsrChannel channel, RealtimeAsrChannelContext context) { + this.meetingId = meetingId; + this.channel = channel; + this.context = context; } - private void detachFrontend(String sessionId) { - if (context.getRawSession() != null && context.getRawSession().getId().equals(sessionId)) { - context.bindFrontendSession(null, null); - } + private void bindFrontend(WebSocketSession rawSession, ConcurrentWebSocketSessionDecorator frontendSession) { + context.bindFrontendSession(rawSession, frontendSession); + } + + private void detachFrontend(String sessionId) { + if (context.getRawSession() != null && context.getRawSession().getId().equals(sessionId)) { + context.bindFrontendSession(null, null); + } + } + + private void clearFrontendIfClosed() { + if (context.getRawSession() != null && !context.getRawSession().isOpen()) { + context.bindFrontendSession(null, null); + } + } + + private boolean hasOpenFrontend() { + return context.getFrontendSession() != null + && context.getFrontendSession().isOpen() + && context.getRawSession() != null + && context.getRawSession().isOpen(); + } + + private boolean isChannelOpen() { + return channel != null && channel.isOpen(context); + } } - private void clearFrontendIfClosed() { - if (context.getRawSession() != null && !context.getRawSession().isOpen()) { - context.bindFrontendSession(null, null); - } - } - - private boolean hasOpenFrontend() { - return context.getFrontendSession() != null - && context.getFrontendSession().isOpen() - && context.getRawSession() != null - && context.getRawSession().isOpen(); - } - - private boolean isChannelOpen() { - return channel != null && channel.isOpen(context); - } - } - - private final class HandlerChannelCallback implements RealtimeAsrChannelCallback { + private final class HandlerChannelCallback implements RealtimeAsrChannelCallback { @Override public void onChannelOpen(Long meetingId) throws Exception { - MeetingChannelSession meetingSession = meetingSessions.get(meetingId); - if (meetingSession == null) { + MeetingChannelSession meetingSession = meetingSessions.get(meetingId); + if (meetingSession == null) { return; } - ConcurrentWebSocketSessionDecorator frontendSession = meetingSession.context.getFrontendSession(); - if (frontendSession != null && frontendSession.isOpen()) { - sendProxyReady(frontendSession); - } + ConcurrentWebSocketSessionDecorator frontendSession = meetingSession.context.getFrontendSession(); + if (frontendSession != null && frontendSession.isOpen()) { + sendProxyReady(frontendSession); + } } - @Override - public void sendFrontendText(Long meetingId, String payload) throws Exception { - MeetingChannelSession meetingSession = meetingSessions.get(meetingId); - if (meetingSession == null) { + @Override + public void sendFrontendText(Long meetingId, String payload) throws Exception { + MeetingChannelSession meetingSession = meetingSessions.get(meetingId); + if (meetingSession == null) { return; } - ConcurrentWebSocketSessionDecorator frontendSession = meetingSession.context.getFrontendSession(); - if (frontendSession != null && frontendSession.isOpen()) { - frontendSession.sendMessage(new TextMessage(payload)); - } + ConcurrentWebSocketSessionDecorator frontendSession = meetingSession.context.getFrontendSession(); + if (frontendSession != null && frontendSession.isOpen()) { + frontendSession.sendMessage(new TextMessage(payload)); + } } @Override public void sendFrontendBinary(Long meetingId, byte[] payload) throws Exception { - MeetingChannelSession meetingSession = meetingSessions.get(meetingId); - if (meetingSession == null) { - return; - } - ConcurrentWebSocketSessionDecorator frontendSession = meetingSession.context.getFrontendSession(); - if (frontendSession != null && frontendSession.isOpen()) { - frontendSession.sendMessage(new BinaryMessage(payload)); - } + MeetingChannelSession meetingSession = meetingSessions.get(meetingId); + if (meetingSession == null) { + return; + } + ConcurrentWebSocketSessionDecorator frontendSession = meetingSession.context.getFrontendSession(); + if (frontendSession != null && frontendSession.isOpen()) { + frontendSession.sendMessage(new BinaryMessage(payload)); + } } @Override public void sendFrontendError(Long meetingId, String code, String message) { - MeetingChannelSession meetingSession = meetingSessions.get(meetingId); - if (meetingSession == null) { - return; - } - ConcurrentWebSocketSessionDecorator frontendSession = meetingSession.context.getFrontendSession(); - if (frontendSession != null) { - RealtimeMeetingProxyWebSocketHandler.this.sendFrontendError(frontendSession, code, message); - } + MeetingChannelSession meetingSession = meetingSessions.get(meetingId); + if (meetingSession == null) { + return; + } + ConcurrentWebSocketSessionDecorator frontendSession = meetingSession.context.getFrontendSession(); + if (frontendSession != null) { + RealtimeMeetingProxyWebSocketHandler.this.sendFrontendError(frontendSession, code, message); + } } @Override public void removeMeetingSession(Long meetingId) { - RealtimeMeetingProxyWebSocketHandler.this.removeMeetingSession(meetingId); + RealtimeMeetingProxyWebSocketHandler.this.removeMeetingSession(meetingId); } @Override public void closeFrontend(Long meetingId, CloseStatus status) { - MeetingChannelSession meetingSession = meetingSessions.get(meetingId); - if (meetingSession == null) { - return; - } + MeetingChannelSession meetingSession = meetingSessions.get(meetingId); + if (meetingSession == null) { + return; + } try { - WebSocketSession rawSession = meetingSession.context.getRawSession(); - if (rawSession != null && rawSession.isOpen()) { + WebSocketSession rawSession = meetingSession.context.getRawSession(); + if (rawSession != null && rawSession.isOpen()) { rawSession.close(status); } } catch (Exception ignored) { diff --git a/backend/src/test/java/com/imeeting/service/biz/impl/AiModelServiceImplTest.java b/backend/src/test/java/com/imeeting/service/biz/impl/AiModelServiceImplTest.java index 6fad4b0..4fa68a6 100644 --- a/backend/src/test/java/com/imeeting/service/biz/impl/AiModelServiceImplTest.java +++ b/backend/src/test/java/com/imeeting/service/biz/impl/AiModelServiceImplTest.java @@ -349,6 +349,30 @@ class AiModelServiceImplTest { assertNull(captor.getValue().getApiKey()); } + @Test + void saveModelShouldRejectTencentAsrWithoutAppId() { + AiModelServiceImpl service = new AiModelServiceImpl( + objectMapper, + mock(AsrModelMapper.class), + mock(LlmModelMapper.class) + ); + + AiModelDTO dto = new AiModelDTO(); + dto.setModelType("ASR"); + dto.setModelName("tencent-asr"); + dto.setProvider("tencent"); + dto.setModelCode("16k_zh"); + dto.setIsDefault(0); + dto.setStatus(1); + dto.setMediaConfig(Map.of( + "tencentSecretId", "secret-id", + "tencentSecretKey", "secret-key" + )); + + RuntimeException ex = assertThrows(RuntimeException.class, () -> service.saveModel(dto)); + assertEquals("腾讯 ASR 模型必须配置 mediaConfig.tencentAppId", ex.getMessage()); + } + @Test void saveModelShouldRejectTencentAsrWithoutSecretKey() { AiModelServiceImpl service = new AiModelServiceImpl( @@ -370,7 +394,32 @@ class AiModelServiceImplTest { )); RuntimeException ex = assertThrows(RuntimeException.class, () -> service.saveModel(dto)); - assertEquals("腾讯实时 ASR 模型必须配置 mediaConfig.tencentSecretKey", ex.getMessage()); + assertEquals("腾讯 ASR 模型必须配置 mediaConfig.tencentSecretKey", ex.getMessage()); + } + + @Test + void saveModelShouldRejectTencentAsrWithoutRealtimeModelCode() { + AiModelServiceImpl service = new AiModelServiceImpl( + objectMapper, + mock(AsrModelMapper.class), + mock(LlmModelMapper.class) + ); + + AiModelDTO dto = new AiModelDTO(); + dto.setModelType("ASR"); + dto.setModelName("tencent-asr"); + dto.setProvider("tencent"); + dto.setIsDefault(0); + dto.setStatus(1); + dto.setMediaConfig(Map.of( + "tencentAppId", "app-id", + "tencentSecretId", "secret-id", + "tencentSecretKey", "secret-key", + "tencentOfflineModelCode", "16k_zh" + )); + + RuntimeException ex = assertThrows(RuntimeException.class, () -> service.saveModel(dto)); + assertEquals("腾讯 ASR 模型必须配置 mediaConfig.tencentRealtimeModelCode", ex.getMessage()); } @Test @@ -388,13 +437,14 @@ class AiModelServiceImplTest { dto.setModelType("ASR"); dto.setModelName("tencent-asr"); dto.setProvider("tencent"); - dto.setModelCode("16k_zh"); dto.setIsDefault(0); dto.setStatus(1); dto.setMediaConfig(Map.of( "tencentAppId", "app-id", "tencentSecretId", "secret-id", - "tencentSecretKey", "secret-key" + "tencentSecretKey", "secret-key", + "tencentOfflineModelCode", "16k_zh", + "tencentRealtimeModelCode", "16k_zh_realtime" )); service.saveModel(dto); @@ -402,7 +452,9 @@ class AiModelServiceImplTest { ArgumentCaptor captor = ArgumentCaptor.forClass(AsrModel.class); verify(asrModelMapper, times(1)).insert(captor.capture()); assertEquals("tencent", captor.getValue().getProvider()); - assertEquals("16k_zh", captor.getValue().getModelCode()); + assertNull(captor.getValue().getModelCode()); + assertEquals("16k_zh", captor.getValue().getMediaConfig().get("tencentOfflineModelCode")); + assertEquals("16k_zh_realtime", captor.getValue().getMediaConfig().get("tencentRealtimeModelCode")); assertEquals("secret-key", captor.getValue().getMediaConfig().get("tencentSecretKey")); assertNull(captor.getValue().getBaseUrl()); } diff --git a/backend/src/test/java/com/imeeting/service/biz/impl/AiTaskServiceImplTest.java b/backend/src/test/java/com/imeeting/service/biz/impl/AiTaskServiceImplTest.java index 8caac30..352332c 100644 --- a/backend/src/test/java/com/imeeting/service/biz/impl/AiTaskServiceImplTest.java +++ b/backend/src/test/java/com/imeeting/service/biz/impl/AiTaskServiceImplTest.java @@ -1,152 +1,308 @@ -//package com.imeeting.service.biz.impl; -// -//import com.fasterxml.jackson.databind.ObjectMapper; -//import com.imeeting.dto.biz.AiModelVO; -//import com.imeeting.entity.biz.AiTask; -//import com.imeeting.entity.biz.HotWord; -//import com.imeeting.entity.biz.Meeting; -//import com.imeeting.mapper.biz.MeetingMapper; -//import com.imeeting.mapper.biz.MeetingTranscriptMapper; -//import com.imeeting.service.biz.AiModelService; -//import com.imeeting.service.biz.HotWordService; -//import com.imeeting.service.biz.MeetingProgressService; -//import com.imeeting.service.biz.MeetingSummaryFileService; -//import com.imeeting.service.biz.MeetingTranscriptChapterService; -//import com.imeeting.service.biz.MeetingTranscriptFileService; -//import com.imeeting.support.RedisValueSupport; -//import com.imeeting.support.TaskSecurityContextRunner; -//import com.unisbase.mapper.SysUserMapper; -//import com.unisbase.service.SysParamService; -//import org.junit.jupiter.api.Test; -//import org.springframework.test.util.ReflectionTestUtils; -// -//import java.util.HashMap; -//import java.util.List; -//import java.util.Map; -// -//import static org.junit.jupiter.api.Assertions.assertEquals; -//import static org.junit.jupiter.api.Assertions.assertFalse; -//import static org.junit.jupiter.api.Assertions.assertNull; -//import static org.junit.jupiter.api.Assertions.assertTrue; -//import static org.mockito.ArgumentMatchers.any; -//import static org.mockito.Mockito.mock; -//import static org.mockito.Mockito.when; -// -//class AiTaskServiceImplTest { -// -// @Test -// void buildAsrRequestShouldFollowCurrentOfflineAsrContract() { -// HotWordService hotWordService = mock(HotWordService.class); -// HotWord hotWord = new HotWord(); -// hotWord.setWord("汇智"); -// hotWord.setWeight(25); -// when(hotWordService.list(any())).thenReturn(List.of(hotWord)); -// -// AiTaskServiceImpl service = new AiTaskServiceImpl( -// mock(MeetingMapper.class), -// mock(MeetingTranscriptMapper.class), -// mock(AiModelService.class), -// new ObjectMapper(), -// mock(SysUserMapper.class), -// hotWordService, -// mock(RedisValueSupport.class), -// mock(MeetingProgressService.class), -// mock(MeetingSummaryFileService.class), -// mock(MeetingTranscriptFileService.class), -// mock(MeetingTranscriptChapterService.class), -// mock(MeetingSummaryPromptAssembler.class), -// mock(TaskSecurityContextRunner.class), -// mock(MeetingExternalSummaryWebhookTrigger.class), -// mock(SysParamService.class) -// ); -// ReflectionTestUtils.setField(service, "serverBaseUrl", "http://localhost:8080"); -// -// Meeting meeting = new Meeting(); -// meeting.setAudioUrl("/api/static/meetings/12/source audio.mp4"); -// -// AiTask task = new AiTask(); -// Map taskConfig = new HashMap<>(); -// taskConfig.put("useSpkId", 1); -// taskConfig.put("enableTextRefine", true); -// taskConfig.put("hotWords", List.of("汇智")); -// task.setTaskConfig(taskConfig); -// -// AiModelVO asrModel = new AiModelVO(); -// asrModel.setModelCode("legacy-model-code"); -// -// @SuppressWarnings("unchecked") -// Map request = (Map) ReflectionTestUtils.invokeMethod( -// service, -// "buildAsrRequest", -// meeting, -// task, -// asrModel -// ); -// -// assertEquals("http://localhost:8080/api/static/meetings/12/source%20audio.mp4", request.get("audio_address")); -// assertFalse(request.containsKey("file_url")); -// -// @SuppressWarnings("unchecked") -// Map config = (Map) request.get("config"); -// assertEquals(Boolean.TRUE, config.get("enable_speaker")); -// assertEquals(Boolean.TRUE, config.get("match_speaker_registry")); -// assertEquals(Boolean.TRUE, config.get("enable_text_cleanup")); -// assertFalse(config.containsKey("enable_text_refine")); -// assertFalse(config.containsKey("enable_two_pass")); -// assertFalse(config.containsKey("model")); -// -// @SuppressWarnings("unchecked") -// List> hotwords = (List>) config.get("hotwords"); -// assertEquals(1, hotwords.size()); -// assertEquals("汇智", hotwords.get(0).get("hotword")); -// assertEquals(2.5, hotwords.get(0).get("weight")); -// } -// -// @Test -// void buildAsrRequestShouldDisableRegistryMatchWhenSpeakerSplitDisabled() { -// AiTaskServiceImpl service = new AiTaskServiceImpl( -// mock(MeetingMapper.class), -// mock(MeetingTranscriptMapper.class), -// mock(AiModelService.class), -// new ObjectMapper(), -// mock(SysUserMapper.class), -// mock(HotWordService.class), -// mock(RedisValueSupport.class), -// mock(MeetingProgressService.class), -// mock(MeetingSummaryFileService.class), -// mock(MeetingTranscriptFileService.class), -// mock(MeetingTranscriptChapterService.class), -// mock(MeetingSummaryPromptAssembler.class), -// mock(TaskSecurityContextRunner.class), -// mock(MeetingExternalSummaryWebhookTrigger.class), -// mock(SysParamService.class) -// ); -// ReflectionTestUtils.setField(service, "serverBaseUrl", "http://localhost:8080"); -// -// Meeting meeting = new Meeting(); -// meeting.setAudioUrl("/api/static/audio/demo.wav"); -// -// AiTask task = new AiTask(); -// Map taskConfig = new HashMap<>(); -// taskConfig.put("useSpkId", 0); -// taskConfig.put("enableTextRefine", false); -// task.setTaskConfig(taskConfig); -// -// @SuppressWarnings("unchecked") -// Map request = (Map) ReflectionTestUtils.invokeMethod( -// service, -// "buildAsrRequest", -// meeting, -// task, -// new AiModelVO() -// ); -// -// @SuppressWarnings("unchecked") -// Map config = (Map) request.get("config"); -// assertEquals(Boolean.FALSE, config.get("enable_speaker")); -// assertEquals(Boolean.FALSE, config.get("match_speaker_registry")); -// assertEquals(Boolean.FALSE, config.get("enable_text_cleanup")); -// assertTrue(((List) config.get("hotwords")).isEmpty()); -// assertNull(request.get("file_url")); -// } -//} +package com.imeeting.service.biz.impl; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.imeeting.dto.biz.AiModelVO; +import com.imeeting.entity.biz.AiTask; +import com.imeeting.entity.biz.Meeting; +import com.imeeting.mapper.biz.AiTaskMapper; +import com.imeeting.mapper.biz.MeetingMapper; +import com.imeeting.mapper.biz.MeetingTranscriptMapper; +import com.imeeting.service.android.AndroidMeetingPushService; +import com.imeeting.service.biz.AiModelService; +import com.imeeting.service.biz.HotWordService; +import com.imeeting.service.biz.MeetingPointsService; +import com.imeeting.service.biz.MeetingProgressService; +import com.imeeting.service.biz.MeetingSummaryFileService; +import com.imeeting.service.biz.MeetingTranscriptChapterService; +import com.imeeting.service.biz.MeetingTranscriptFileService; +import com.imeeting.support.TaskSecurityContextRunner; +import com.imeeting.support.redis.MeetingAsrPermitCache; +import com.imeeting.support.redis.MeetingLockCache; +import com.unisbase.mapper.SysUserMapper; +import com.unisbase.service.SysParamService; +import com.tencentcloudapi.asr.v20190614.models.SentenceDetail; +import com.tencentcloudapi.asr.v20190614.models.TaskStatus; +import org.junit.jupiter.api.Test; +import org.springframework.test.util.ReflectionTestUtils; + +import java.util.HashMap; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class AiTaskServiceImplTest { + + @Test + void processAsrTaskShouldUseTencentOfflineBranchWhenProviderIsTencent() throws Exception { + MeetingMapper meetingMapper = mock(MeetingMapper.class); + MeetingTranscriptMapper transcriptMapper = mock(MeetingTranscriptMapper.class); + AiModelService aiModelService = mock(AiModelService.class); + HotWordService hotWordService = mock(HotWordService.class); + MeetingLockCache meetingLockCache = mock(MeetingLockCache.class); + MeetingAsrPermitCache meetingAsrPermitCache = mock(MeetingAsrPermitCache.class); + MeetingProgressService meetingProgressService = mock(MeetingProgressService.class); + MeetingPointsService meetingPointsService = mock(MeetingPointsService.class); + MeetingSummaryFileService meetingSummaryFileService = mock(MeetingSummaryFileService.class); + MeetingTranscriptFileService meetingTranscriptFileService = mock(MeetingTranscriptFileService.class); + MeetingTranscriptChapterService meetingTranscriptChapterService = mock(MeetingTranscriptChapterService.class); + MeetingSummaryPromptAssembler meetingSummaryPromptAssembler = mock(MeetingSummaryPromptAssembler.class); + TaskSecurityContextRunner taskSecurityContextRunner = mock(TaskSecurityContextRunner.class); + MeetingExternalSummaryWebhookTrigger meetingExternalSummaryWebhookTrigger = mock(MeetingExternalSummaryWebhookTrigger.class); + SysParamService sysParamService = mock(SysParamService.class); + + AiTaskServiceImpl service = spy(new AiTaskServiceImpl( + meetingMapper, + transcriptMapper, + aiModelService, + new ObjectMapper(), + mock(SysUserMapper.class), + hotWordService, + meetingLockCache, + meetingAsrPermitCache, + meetingProgressService, + meetingPointsService, + meetingSummaryFileService, + meetingTranscriptFileService, + meetingTranscriptChapterService, + meetingSummaryPromptAssembler, + taskSecurityContextRunner, + meetingExternalSummaryWebhookTrigger, + sysParamService + )); + ReflectionTestUtils.setField(service, "baseMapper", mock(AiTaskMapper.class)); + ReflectionTestUtils.setField(service, "androidMeetingPushService", mock(AndroidMeetingPushService.class)); + + Meeting meeting = new Meeting(); + meeting.setId(1L); + meeting.setAudioUrl("https://cdn.example.com/audio/demo.m4a"); + + AiTask task = new AiTask(); + task.setId(11L); + task.setMeetingId(1L); + task.setTaskType("ASR"); + task.setTaskConfig(new HashMap<>(Map.of( + "asrModelId", 101L, + "useSpkId", 1, + "enableTextRefine", true + ))); + + AiModelVO model = new AiModelVO(); + model.setId(101L); + model.setProvider("tencent"); + model.setModelCode("16k_zh"); + model.setMediaConfig(Map.of( + "tencentAppId", "123456", + "tencentSecretId", "secret-id", + "tencentSecretKey", "secret-key" + )); + when(aiModelService.getModelById(101L, "ASR")).thenReturn(model); + + doReturn(true).when(service).updateById(any(AiTask.class)); + doReturn("腾讯离线转写结果").when(service).processTencentOfflineAsr(eq(meeting), eq(task), eq(model)); + + String result = ReflectionTestUtils.invokeMethod(service, "processAsrTask", meeting, task); + + assertEquals("腾讯离线转写结果", result); + verify(service).processTencentOfflineAsr(meeting, task, model); + verify(meetingPointsService).recordAsrSuccessCharge(meeting, task); + } + + @Test + void buildTencentOfflineCreateRequestShouldMapMeetingAndTaskConfig() { + AiTaskServiceImpl service = createService(mock(MeetingPointsService.class)); + ReflectionTestUtils.setField(service, "serverBaseUrl", "https://server.example.com"); + + Meeting meeting = new Meeting(); + meeting.setId(2L); + meeting.setAudioUrl("/upload/audio/demo.m4a"); + + AiTask task = new AiTask(); + task.setTaskConfig(new HashMap<>(Map.of( + "useSpkId", 1, + "enableTextRefine", true, + "hotWords", java.util.List.of("腾讯会议", "离线转写") + ))); + + AiModelVO model = new AiModelVO(); + model.setModelCode("legacy-model-code"); + model.setMediaConfig(Map.of( + "tencentAppId", "123456", + "tencentSecretId", "secret-id", + "tencentSecretKey", "secret-key", + "tencentOfflineModelCode", "16k_zh", + "tencentRealtimeModelCode", "16k_zh_realtime" + )); + + @SuppressWarnings("unchecked") + Map request = (Map) ReflectionTestUtils.invokeMethod( + service, + "buildTencentOfflineCreateRequest", + meeting, + task, + model + ); + + assertEquals("16k_zh", request.get("engineModelType")); + assertEquals(1L, request.get("channelNum")); + assertEquals(2L, request.get("resTextFormat")); + assertEquals(0L, request.get("sourceType")); + assertEquals("https://server.example.com/upload/audio/demo.m4a", request.get("url")); + assertEquals(1L, request.get("speakerDiarization")); + assertEquals(0L, request.get("speakerNumber")); + assertEquals("腾讯会议|5,离线转写|5", request.get("hotwordList")); + } + + @Test + void buildTencentOfflineCreateRequestShouldKeepAbsoluteAudioUrl() { + AiTaskServiceImpl service = createService(mock(MeetingPointsService.class)); + ReflectionTestUtils.setField(service, "serverBaseUrl", "https://server.example.com"); + + Meeting meeting = new Meeting(); + meeting.setId(22L); + meeting.setAudioUrl("https://cdn.example.com/audio/demo.m4a"); + + AiTask task = new AiTask(); + task.setTaskConfig(new HashMap<>(Map.of("useSpkId", 0))); + + AiModelVO model = new AiModelVO(); + model.setModelCode("16k_zh"); + model.setMediaConfig(Map.of( + "tencentAppId", "123456", + "tencentSecretId", "secret-id", + "tencentSecretKey", "secret-key" + )); + + @SuppressWarnings("unchecked") + Map request = (Map) ReflectionTestUtils.invokeMethod( + service, + "buildTencentOfflineCreateRequest", + meeting, + task, + model + ); + + assertEquals("https://cdn.example.com/audio/demo.m4a", request.get("url")); + } + + @Test + void processTencentOfflineAsrShouldPollUntilSuccessAndReturnTranscriptText() throws Exception { + MeetingPointsService meetingPointsService = mock(MeetingPointsService.class); + AiTaskServiceImpl service = spy(createService(meetingPointsService)); + + Meeting meeting = new Meeting(); + meeting.setId(3L); + meeting.setAudioUrl("https://cdn.example.com/audio/demo.m4a"); + + AiTask task = new AiTask(); + task.setId(31L); + task.setMeetingId(3L); + task.setTaskType("ASR"); + task.setTaskConfig(new HashMap<>(Map.of( + "asrModelId", 101L, + "useSpkId", 1 + ))); + + AiModelVO model = new AiModelVO(); + model.setProvider("tencent"); + model.setModelCode("legacy-model-code"); + model.setMediaConfig(Map.of( + "tencentAppId", "123456", + "tencentSecretId", "secret-id", + "tencentSecretKey", "secret-key", + "tencentOfflineModelCode", "16k_zh", + "tencentRealtimeModelCode", "16k_zh_realtime" + )); + + TaskStatus doing = new TaskStatus(); + doing.setStatusStr("doing"); + + SentenceDetail sentence = new SentenceDetail(); + sentence.setSpeakerId(3L); + sentence.setFinalSentence("测试文本"); + sentence.setStartMs(1000L); + sentence.setEndMs(2500L); + + TaskStatus success = new TaskStatus(); + success.setStatusStr("success"); + success.setResultDetail(new SentenceDetail[]{sentence}); + + doReturn(90001L).when(service).submitTencentOfflineTask(meeting, task, model); + doReturn(doing).doReturn(success).when(service).queryTencentOfflineTask(model, 90001L); + doReturn("未知说话人3: 测试文本\n").when(service).saveTencentOfflineTranscripts(meeting, success.getResultDetail()); + doReturn(true).when(service).updateById(any(AiTask.class)); + + String text = service.processTencentOfflineAsr(meeting, task, model); + + assertEquals("未知说话人3: 测试文本\n", text); + } + + @Test + void saveTencentOfflineTranscriptsShouldOverwriteExistingRowsAndUseUnknownSpeakerName() { + MeetingPointsService meetingPointsService = mock(MeetingPointsService.class); + MeetingTranscriptMapper transcriptMapper = mock(MeetingTranscriptMapper.class); + MeetingTranscriptFileService transcriptFileService = mock(MeetingTranscriptFileService.class); + AiTaskServiceImpl service = createService(meetingPointsService, transcriptMapper, transcriptFileService); + + Meeting meeting = new Meeting(); + meeting.setId(5L); + + SentenceDetail first = new SentenceDetail(); + first.setSpeakerId(7L); + first.setFinalSentence("第一句"); + first.setStartMs(0L); + first.setEndMs(1000L); + + SentenceDetail second = new SentenceDetail(); + second.setSpeakerId(7L); + second.setFinalSentence("第二句"); + second.setStartMs(1000L); + second.setEndMs(2000L); + + String text = service.saveTencentOfflineTranscripts(meeting, new SentenceDetail[]{first, second}); + + assertEquals("未知说话人7: 第一句\n未知说话人7: 第二句\n", text); + verify(transcriptMapper).delete(any()); + verify(transcriptMapper, org.mockito.Mockito.times(2)).insert(any()); + verify(transcriptFileService).initializeTranscriptFileIfAbsent(5L); + } + + private AiTaskServiceImpl createService(MeetingPointsService meetingPointsService) { + return createService(meetingPointsService, mock(MeetingTranscriptMapper.class), mock(MeetingTranscriptFileService.class)); + } + + private AiTaskServiceImpl createService(MeetingPointsService meetingPointsService, + MeetingTranscriptMapper transcriptMapper, + MeetingTranscriptFileService transcriptFileService) { + AiTaskServiceImpl service = new AiTaskServiceImpl( + mock(MeetingMapper.class), + transcriptMapper, + mock(AiModelService.class), + new ObjectMapper(), + mock(SysUserMapper.class), + mock(HotWordService.class), + mock(MeetingLockCache.class), + mock(MeetingAsrPermitCache.class), + mock(MeetingProgressService.class), + meetingPointsService, + mock(MeetingSummaryFileService.class), + transcriptFileService, + mock(MeetingTranscriptChapterService.class), + mock(MeetingSummaryPromptAssembler.class), + mock(TaskSecurityContextRunner.class), + mock(MeetingExternalSummaryWebhookTrigger.class), + mock(SysParamService.class) + ); + ReflectionTestUtils.setField(service, "baseMapper", mock(AiTaskMapper.class)); + ReflectionTestUtils.setField(service, "androidMeetingPushService", mock(AndroidMeetingPushService.class)); + return service; + } +} diff --git a/frontend/src/pages/business/AiModels.tsx b/frontend/src/pages/business/AiModels.tsx index 6886584..c68cdf2 100644 --- a/frontend/src/pages/business/AiModels.tsx +++ b/frontend/src/pages/business/AiModels.tsx @@ -24,6 +24,7 @@ import { testLocalModelConnectivity, updateAiModel, } from "../../api/business/aimodel"; +import {getMeetingCreateConfig, type MeetingCreateConfig} from "../../api/business/meeting"; import AppPagination from "../../components/shared/AppPagination"; const { Option } = Select; @@ -31,6 +32,12 @@ const { Title } = Typography; type ModelType = "ASR" | "LLM"; +const DEFAULT_CREATE_CONFIG: MeetingCreateConfig = { + offlineEnabled: true, + realtimeEnabled: false, + offlineAudioMaxSizeMb: 1024, +}; + const PROVIDER_BASE_URL_MAP: Record = { openai: "https://api.openai.com", deepseek: "https://api.deepseek.com", @@ -64,6 +71,7 @@ const AiModels: React.FC = () => { const [connectivityLoading, setConnectivityLoading] = useState(false); const [remoteModels, setRemoteModels] = useState([]); const [speakerModels, setSpeakerModels] = useState([]); + const [createConfig, setCreateConfig] = useState(DEFAULT_CREATE_CONFIG); const modelNameAutoFilledRef = useRef(false); const localProfileLoadedRef = useRef(false); @@ -86,6 +94,20 @@ const AiModels: React.FC = () => { void fetchData(); }, [current, size, searchName, activeType]); + useEffect(() => { + getMeetingCreateConfig() + .then((res) => { + const config = (res as any)?.data?.data ?? (res as any); + setCreateConfig({ + ...DEFAULT_CREATE_CONFIG, + ...(config || {}), + }); + }) + .catch(() => { + setCreateConfig(DEFAULT_CREATE_CONFIG); + }); + }, []); + useEffect(() => { if (!drawerVisible || !provider) { return; @@ -137,6 +159,8 @@ const AiModels: React.FC = () => { const tencentAppId = record.mediaConfig?.tencentAppId; const tencentSecretId = record.mediaConfig?.tencentSecretId; const tencentSecretKey = record.mediaConfig?.tencentSecretKey; + const tencentOfflineModelCode = record.mediaConfig?.tencentOfflineModelCode || record.modelCode; + const tencentRealtimeModelCode = record.mediaConfig?.tencentRealtimeModelCode || record.modelCode; form.setFieldsValue({ ...record, modelType: record.modelType, @@ -145,6 +169,8 @@ const AiModels: React.FC = () => { tencentAppId, tencentSecretId, tencentSecretKey, + tencentOfflineModelCode, + tencentRealtimeModelCode, isDefaultChecked: record.isDefault === 1, statusChecked: record.status === 1, }); @@ -261,8 +287,8 @@ const AiModels: React.FC = () => { baseUrl: values.baseUrl, apiPath: values.apiPath, apiKey: values.apiKey, - modelCode: values.modelCode, - wsUrl: values.wsUrl, + modelCode: activeType === "ASR" && isTencentProvider ? values.tencentOfflineModelCode : values.modelCode, + wsUrl: activeType === "ASR" ? values.wsUrl : undefined, mediaConfig: activeType === "ASR" && isLocalProvider ? { @@ -274,6 +300,8 @@ const AiModels: React.FC = () => { tencentAppId: values.tencentAppId, tencentSecretId: values.tencentSecretId, tencentSecretKey: values.tencentSecretKey, + tencentOfflineModelCode: values.tencentOfflineModelCode, + tencentRealtimeModelCode: values.tencentRealtimeModelCode, } : undefined, temperature: values.temperature, @@ -567,13 +595,14 @@ const AiModels: React.FC = () => { - + + + + + + + + + + )} diff --git a/frontend/src/pages/business/RealtimeAsrSession.tsx b/frontend/src/pages/business/RealtimeAsrSession.tsx index e7a6cc4..ee65572 100644 --- a/frontend/src/pages/business/RealtimeAsrSession.tsx +++ b/frontend/src/pages/business/RealtimeAsrSession.tsx @@ -244,6 +244,7 @@ export function RealtimeAsrSession() { const [sessionStatus, setSessionStatus] = useState(null); const transcriptRef = useRef(null); const wsRef = useRef(null); + const wsHeartbeatRef = useRef | null>(null); const audioContextRef = useRef(null); const processorRef = useRef(null); const audioSourceRef = useRef(null); @@ -373,6 +374,33 @@ export function RealtimeAsrSession() { return () => window.removeEventListener("pagehide", handlePageHide); }, [meetingId]); + // 组件卸载（切换路由）时统一清理所有资源，防止定时器、WebSocket、音频管道泄漏 + useEffect(() => { + return () => { + // 1. 清理心跳定时器 + if (wsHeartbeatRef.current !== null) { + clearInterval(wsHeartbeatRef.current); + wsHeartbeatRef.current = null; + } + // 2. 关闭 WebSocket（移除回调避免触发不必要的状态更新） + if (wsRef.current) { + wsRef.current.onclose = null; + wsRef.current.onerror = null; + wsRef.current.onmessage = null; + wsRef.current.close(); + wsRef.current = null; + } + // 3. 关闭音频管道（同步处理，无需 await） + processorRef.current?.disconnect(); + audioSourceRef.current?.disconnect(); + streamRef.current?.getTracks().forEach((t) => t.stop()); + if (audioContextRef.current && audioContextRef.current.state !== "closed") { + void audioContextRef.current.close(); + } + }; + // eslint-disable-next-line react-hooks/exhaustive-deps + }, []); + const shutdownAudioPipeline = async () => { processorRef.current?.disconnect(); audioSourceRef.current?.disconnect(); @@ -433,6 +461,10 @@ export function RealtimeAsrSession() { setRecording(false); setStatusText("连接失败"); sessionStartedRef.current = false; + if (wsHeartbeatRef.current !== null) { + clearInterval(wsHeartbeatRef.current); + wsHeartbeatRef.current = null; + } wsRef.current?.close(); wsRef.current = null; await shutdownAudioPipeline(); @@ -442,6 +474,30 @@ export function RealtimeAsrSession() { message.error(errorMessage); }; + const handleUpstreamPauseError = async (errorMessage: string) => { + if (recording && startedAtRef.current) { + elapsedOffsetRef.current += Math.floor((Date.now() - startedAtRef.current) / 1000); + } + setConnecting(false); + setRecording(false); + setStatusText("上游识别已断开，正在暂停会议..."); + sessionStartedRef.current = false; + wsRef.current?.close(); + wsRef.current = null; + await shutdownAudioPipeline(); + startedAtRef.current = null; + try { + const pauseRes = await pauseRealtimeMeeting(meetingId); + setSessionStatus(pauseRes.data.data); + setElapsedSeconds(elapsedOffsetRef.current); + setStatusText("已暂停，可继续会议"); + message.error(errorMessage); + } catch { + setStatusText("识别已断开"); + message.error(errorMessage); + } + }; + const startAudioPipeline = async () => { if (!window.isSecureContext || !navigator.mediaDevices?.getUserMedia) { throw new Error("当前浏览器环境不支持麦克风访问。请使用 localhost 或 HTTPS 域名访问系统。"); @@ -506,6 +562,13 @@ export function RealtimeAsrSession() { } const pauseRes = await pauseRealtimeMeeting(meetingId); await closeFrontendSocket(false); + if (wsHeartbeatRef.current !== null) { + clearInterval(wsHeartbeatRef.current); + wsHeartbeatRef.current = null; + } + wsRef.current?.close(); + wsRef.current = null; + sessionStartedRef.current = false; await shutdownAudioPipeline(); setSessionStatus(pauseRes.data.data); setRecording(false); @@ -551,12 +614,41 @@ export function RealtimeAsrSession() { hotwords: sessionDraft.hotwords || [], }); const socketSession = socketSessionRes.data.data; + + // 如果已有旧的 WebSocket（比如重连），先强制关闭并清理心跳 + if (wsHeartbeatRef.current !== null) { + clearInterval(wsHeartbeatRef.current); + wsHeartbeatRef.current = null; + } + if (wsRef.current) { + wsRef.current.onclose = null; + wsRef.current.onerror = null; + wsRef.current.onmessage = null; + wsRef.current.close(); + wsRef.current = null; + } + const socket = new WebSocket(buildRealtimeProxyWsUrl(socketSession)); socket.binaryType = "arraybuffer"; wsRef.current = socket; socket.onopen = () => { setStatusText("识别服务连接中，等待第三方服务就绪..."); + // 先清除旧定时器（防止 onopen 被意外重复触发时叠加） + if (wsHeartbeatRef.current !== null) { + clearInterval(wsHeartbeatRef.current); + wsHeartbeatRef.current = null; + } + // 启动心跳保活：每 20 秒发送一次 keepalive JSON，防止服务端 ping timeout 断开 + wsHeartbeatRef.current = setInterval(() => { + if (socket.readyState === WebSocket.OPEN) { + try { + socket.send(JSON.stringify({ type: "keepalive" })); + } catch { + // ignore heartbeat send failure + } + } + }, 20000); }; socket.onmessage = (event) => { @@ -585,7 +677,11 @@ export function RealtimeAsrSession() { if ((payload.code || payload.type === "error") && payload.message) { setStatusText(payload.message); - void handleFatalRealtimeError(payload.message); + if (payload.code === "REALTIME_UPSTREAM_CLOSED" || payload.code === "REALTIME_UPSTREAM_ERROR") { + void handleUpstreamPauseError(payload.message); + } else { + void handleFatalRealtimeError(payload.message); + } return; } @@ -613,6 +709,10 @@ export function RealtimeAsrSession() { }; socket.onclose = () => { + if (wsHeartbeatRef.current !== null) { + clearInterval(wsHeartbeatRef.current); + wsHeartbeatRef.current = null; + } setConnecting(false); setRecording(false); sessionStartedRef.current = false; @@ -640,6 +740,16 @@ export function RealtimeAsrSession() { setStatusText("结束会议中..."); await closeFrontendSocket(true); + if (wsRef.current?.readyState === WebSocket.OPEN) { + wsRef.current.send(JSON.stringify({ is_speaking: false })); + } + if (wsHeartbeatRef.current !== null) { + clearInterval(wsHeartbeatRef.current); + wsHeartbeatRef.current = null; + } + wsRef.current?.close(); + wsRef.current = null; + sessionStartedRef.current = false; await shutdownAudioPipeline(); @@ -804,17 +914,17 @@ export function RealtimeAsrSession() { 字数 {totalTranscriptChars} - - 模型 {sessionDraft.asrModelName} - - - 模式 {sessionDraft.mode} - - {sessionDraft.hotwords && sessionDraft.hotwords.length > 0 && ( - - 热词 {sessionDraft.hotwords.length} - - )} + {/**/} + {/* 模型 {sessionDraft.asrModelName}*/} + {/**/} + {/**/} + {/* 模式 {sessionDraft.mode}*/} + {/**/} + {/*{sessionDraft.hotwords && sessionDraft.hotwords.length > 0 && (*/} + {/* */} + {/* 热词 {sessionDraft.hotwords.length}*/} + {/* */} + {/*)}*/} )}