chenh
/
imeeting
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/dev_na' into dev_ymcg

dev_na
puz 2026-06-29 10:14:13 +08:00
parent 255e268138 57f343ddfd
commit 08d9024401
13 changed files with 1106 additions and 434 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
backend/pom.xml
View File

@ -183,6 +183,11 @@
<artifactId>tencentcloud-speech-sdk-java</artifactId> <artifactId>tencentcloud-speech-sdk-java</artifactId>
<version>1.0.67</version> <version>1.0.67</version>
</dependency> </dependency>
<dependency>
<groupId>com.tencentcloudapi</groupId>
<artifactId>tencentcloud-sdk-java-asr</artifactId>
<version>3.1.1470</version>
</dependency>
</dependencies> </dependencies>
<build> <build>

24
backend/src/main/java/com/imeeting/config/RealtimeMeetingWebSocketConfig.java
View File

@ -2,6 +2,9 @@ package com.imeeting.config;
import com.imeeting.websocket.RealtimeMeetingProxyWebSocketHandler; import com.imeeting.websocket.RealtimeMeetingProxyWebSocketHandler;
import lombok.RequiredArgsConstructor; 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.context.annotation.Configuration;
import org.springframework.web.socket.config.annotation.EnableWebSocket; import org.springframework.web.socket.config.annotation.EnableWebSocket;
import org.springframework.web.socket.config.annotation.WebSocketConfigurer; import org.springframework.web.socket.config.annotation.WebSocketConfigurer;
@ -19,4 +22,25 @@ public class RealtimeMeetingWebSocketConfig implements WebSocketConfigurer {
registry.addHandler(realtimeMeetingProxyWebSocketHandler, "/ws/meeting/realtime") registry.addHandler(realtimeMeetingProxyWebSocketHandler, "/ws/meeting/realtime")
.setAllowedOriginPatterns("*"); .setAllowedOriginPatterns("*");
} }
/**
* 关闭 Tomcat 内置的 WebSocket keepalive ping 检测（sessionIdleTimeout）。
* <p>
* Tomcat 默认会对 WebSocket session 设置 sessionIdleTimeout（-1 表示无限，但某些版本默认非 -1），
* 并通过后台线程定期发送 Ping 帧，若在超时内未收到 Pong 响应，触发
* code=1011 "keepalive ping timeout" 强制断开。
* 实时 ASR 场景中，客户端持续发送音频帧，由前端心跳保活，
* 因此显式将 sessionIdleTimeout 设为 -1（无限）。
* </p>
*/
@Bean
public WebServerFactoryCustomizer<TomcatServletWebServerFactory> 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");
}
});
}
} }

4
backend/src/main/java/com/imeeting/service/android/legacy/impl/LegacyMeetingAdapterServiceImpl.java
View File

@ -205,7 +205,7 @@ public class LegacyMeetingAdapterServiceImpl implements LegacyMeetingAdapterServ
null, null,
null, null,
null, null,
null, true,
null, null,
null, null,
List.of() List.of()
@ -281,7 +281,7 @@ public class LegacyMeetingAdapterServiceImpl implements LegacyMeetingAdapterServ
null, null,
null, null,
null, null,
null, true,
null, null,
null, null,
List.of() List.of()

15
backend/src/main/java/com/imeeting/service/biz/impl/AiModelServiceImpl.java
View File

@ -49,6 +49,8 @@ public class AiModelServiceImpl implements AiModelService {
private static final String MEDIA_TENCENT_APP_ID = "tencentAppId"; 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_ID = "tencentSecretId";
private static final String MEDIA_TENCENT_SECRET_KEY = "tencentSecretKey"; 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 int DEFAULT_SORT_ORDER = 0;
private static final String DEFAULT_LLM_API_PATH = "/v1/chat/completions"; private static final String DEFAULT_LLM_API_PATH = "/v1/chat/completions";
private static final String DEFAULT_ANTHROPIC_API_PATH = "/messages"; private static final String DEFAULT_ANTHROPIC_API_PATH = "/messages";
@ -956,16 +958,19 @@ public class AiModelServiceImpl implements AiModelService {
} }
Map<String, Object> mediaConfig = dto.getMediaConfig() == null ? Collections.emptyMap() : dto.getMediaConfig(); Map<String, Object> mediaConfig = dto.getMediaConfig() == null ? Collections.emptyMap() : dto.getMediaConfig();
if (readConfigString(mediaConfig.get(MEDIA_TENCENT_APP_ID)) == null) { 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) { 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) { 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()) { if (readConfigString(mediaConfig.get(MEDIA_TENCENT_OFFLINE_MODEL_CODE)) == null) {
throw new RuntimeException("腾讯实时 ASR 模型必须配置 modelCode"); throw new RuntimeException("腾讯 ASR 模型必须配置 mediaConfig.tencentOfflineModelCode");
}
if (readConfigString(mediaConfig.get(MEDIA_TENCENT_REALTIME_MODEL_CODE)) == null) {
throw new RuntimeException("腾讯 ASR 模型必须配置 mediaConfig.tencentRealtimeModelCode");
} }
} }

239
backend/src/main/java/com/imeeting/service/biz/impl/AiTaskServiceImpl.java
View File

@ -37,6 +37,16 @@ import com.imeeting.support.redis.MeetingLockCache;
import com.unisbase.entity.SysUser; import com.unisbase.entity.SysUser;
import com.unisbase.mapper.SysUserMapper; import com.unisbase.mapper.SysUserMapper;
import com.unisbase.service.SysParamService; 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 lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.beans.factory.annotation.Qualifier;
@ -69,6 +79,13 @@ public class AiTaskServiceImpl extends ServiceImpl<AiTaskMapper, AiTask> impleme
private static final Duration ASR_QUERY_REQUEST_TIMEOUT = Duration.ofSeconds(30); 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_PARALLEL = "PARALLEL";
private static final String DISPATCH_MODE_SERIAL = "SERIAL"; 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 MeetingMapper meetingMapper;
private final MeetingTranscriptMapper transcriptMapper; private final MeetingTranscriptMapper transcriptMapper;
@ -609,6 +626,14 @@ public class AiTaskServiceImpl extends ServiceImpl<AiTaskMapper, AiTask> impleme
log.info("[ASR-PROC] 解析ASR模型成功: meetingId={}, asrTaskId={}, asrModelId={}, baseUrl={}", log.info("[ASR-PROC] 解析ASR模型成功: meetingId={}, asrTaskId={}, asrModelId={}, baseUrl={}",
meeting.getId(), taskRecord.getId(), asrModelId, asrModel.getBaseUrl()); 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 submitUrl = appendPath(asrModel.getBaseUrl(), "api/v1/asr/transcriptions");
String taskId = taskRecord.getResponseData() != null String taskId = taskRecord.getResponseData() != null
? String.valueOf(taskRecord.getResponseData().getOrDefault("task_id", "")) ? String.valueOf(taskRecord.getResponseData().getOrDefault("task_id", ""))
@ -702,6 +727,124 @@ public class AiTaskServiceImpl extends ServiceImpl<AiTaskMapper, AiTask> impleme
return transcriptText; 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<String, Object> buildTencentOfflineCreateRequest(Meeting meeting, AiTask taskRecord, AiModelVO asrModel) {
Map<String, Object> 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<String, Object> 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<MeetingTranscript>().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<String, Object> buildAsrRequest(Meeting meeting, AiTask taskRecord, AiModelVO asrModel) { private Map<String, Object> buildAsrRequest(Meeting meeting, AiTask taskRecord, AiModelVO asrModel) {
Map<String, Object> req = new HashMap<>(); Map<String, Object> req = new HashMap<>();
String rawAudioUrl = meeting.getAudioUrl(); String rawAudioUrl = meeting.getAudioUrl();
@ -1417,6 +1560,102 @@ public class AiTaskServiceImpl extends ServiceImpl<AiTaskMapper, AiTask> impleme
} }
} }
private AsrClient buildTencentOfflineAsrClient(AiModelVO asrModel) {
Map<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> buildTencentTaskStatusSnapshot(TaskStatus taskStatus) {
Map<String, Object> 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<String, Object> 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) { private AiModelVO resolveAsrModelForRevision(AiTask asrTask) {
if (asrTask == null || asrTask.getTaskConfig() == null) { if (asrTask == null || asrTask.getTaskConfig() == null) {
return null; return null;

23
backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSocketSessionServiceImpl.java
View File

@ -26,6 +26,8 @@ import java.util.UUID;
public class RealtimeMeetingSocketSessionServiceImpl implements RealtimeMeetingSocketSessionService { public class RealtimeMeetingSocketSessionServiceImpl implements RealtimeMeetingSocketSessionService {
private static final String WS_PATH = "/ws/meeting/realtime"; 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 RealtimeMeetingSocketSessionCache socketSessionCache;
private final MeetingAccessService meetingAccessService; private final MeetingAccessService meetingAccessService;
@ -87,7 +89,7 @@ public class RealtimeMeetingSocketSessionServiceImpl implements RealtimeMeetingS
sessionData.setAsrModelId(asrModelId); sessionData.setAsrModelId(asrModelId);
sessionData.setProvider(realtimeAsrChannelFactory.normalizeProvider(asrModel.getProvider())); sessionData.setProvider(realtimeAsrChannelFactory.normalizeProvider(asrModel.getProvider()));
sessionData.setTargetWsUrl(targetWsUrl); sessionData.setTargetWsUrl(targetWsUrl);
sessionData.setModelCode(asrModel.getModelCode()); sessionData.setModelCode(resolveRealtimeModelCode(asrModel));
sessionData.setMediaConfig(asrModel.getMediaConfig()); sessionData.setMediaConfig(asrModel.getMediaConfig());
String sessionToken = UUID.randomUUID().toString().replace("-", ""); String sessionToken = UUID.randomUUID().toString().replace("-", "");
@ -111,6 +113,25 @@ public class RealtimeMeetingSocketSessionServiceImpl implements RealtimeMeetingS
return vo; return vo;
} }
private String resolveRealtimeModelCode(AiModelVO asrModel) {
if (asrModel == null) {
return null;
}
if (!TENCENT_PROVIDER.equalsIgnoreCase(asrModel.getProvider())) {
return asrModel.getModelCode();
}
Map<String, Object> 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 @Override
public RealtimeSocketSessionData getSessionData(String sessionToken) { public RealtimeSocketSessionData getSessionData(String sessionToken) {
return socketSessionCache.get(sessionToken); return socketSessionCache.get(sessionToken);

3
backend/src/main/java/com/imeeting/service/realtime/impl/LocalRealtimeAsrChannel.java
View File

@ -513,7 +513,6 @@ public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
log.info("上游 ASR websocket 已关闭：meetingId={}, sessionId={}, code={}, reason={}", log.info("上游 ASR websocket 已关闭：meetingId={}, sessionId={}, code={}, reason={}",
context.getMeetingId(), currentConnectionId(context), statusCode, reason); context.getMeetingId(), currentConnectionId(context), statusCode, reason);
context.getChannelState().remove(STATE_UPSTREAM_SOCKET); context.getChannelState().remove(STATE_UPSTREAM_SOCKET);
context.getCallback().removeMeetingSession(context.getMeetingId());
context.getCallback().sendFrontendError(context.getMeetingId(), context.getCallback().sendFrontendError(context.getMeetingId(),
"REALTIME_UPSTREAM_CLOSED", "REALTIME_UPSTREAM_CLOSED",
reason == null || reason.isBlank() ? "上游 ASR WebSocket 已断开" : "上游 ASR WebSocket 已断开: " + reason); 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={}", log.error("上游 ASR websocket 异常：meetingId={}, sessionId={}, upstream={}",
context.getMeetingId(), currentConnectionId(context), context.getTargetWsUrl(), error); context.getMeetingId(), currentConnectionId(context), context.getTargetWsUrl(), error);
context.getChannelState().remove(STATE_UPSTREAM_SOCKET); context.getChannelState().remove(STATE_UPSTREAM_SOCKET);
context.getCallback().removeMeetingSession(context.getMeetingId());
context.getCallback().sendFrontendError(context.getMeetingId(), context.getCallback().sendFrontendError(context.getMeetingId(),
"REALTIME_UPSTREAM_ERROR", "REALTIME_UPSTREAM_ERROR",
error == null || error.getMessage() == null || error.getMessage().isBlank() error == null || error.getMessage() == null || error.getMessage().isBlank()
? "上游 ASR WebSocket 连接异常" ? "上游 ASR WebSocket 连接异常"
: "上游 ASR WebSocket 连接异常: " + error.getMessage()); : "上游 ASR WebSocket 连接异常: " + error.getMessage());
context.getCallback().closeFrontend(context.getMeetingId(), CloseStatus.SERVER_ERROR); context.getCallback().closeFrontend(context.getMeetingId(), CloseStatus.SERVER_ERROR);
context.getCallback().removeMeetingSession(context.getMeetingId());
} }
} }
} }

2
backend/src/main/java/com/imeeting/service/realtime/impl/RealtimeMeetingTranscriptCacheServiceImpl.java
View File

@ -206,7 +206,7 @@ public class RealtimeMeetingTranscriptCacheServiceImpl implements RealtimeMeetin
return item.getSpeakerName().trim(); return item.getSpeakerName().trim();
} }
String speakerId = resolveSpeakerId(item); String speakerId = resolveSpeakerId(item);
return speakerId == null || speakerId.isBlank() ? null : speakerId; return speakerId == null || speakerId.isBlank() ? null : "未知说话人" + speakerId;
} }
private String resolveSpeakerId(JsonNode sentence) { private String resolveSpeakerId(JsonNode sentence) {

515
backend/src/main/java/com/imeeting/websocket/RealtimeMeetingProxyWebSocketHandler.java
View File

@ -42,25 +42,25 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
private static final String ATTR_MEETING_ID = "meetingId"; private static final String ATTR_MEETING_ID = "meetingId";
private static final String ATTR_TARGET_WS_URL = "targetWsUrl"; 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_TEXT_COUNT = "frontendTextCount";
private static final String ATTR_FRONTEND_BINARY_COUNT = "frontendBinaryCount"; 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 RealtimeMeetingSocketSessionService realtimeMeetingSocketSessionService;
private final RealtimeMeetingSessionStateService realtimeMeetingSessionStateService; private final RealtimeMeetingSessionStateService realtimeMeetingSessionStateService;
private final RealtimeMeetingAudioStorageService realtimeMeetingAudioStorageService; private final RealtimeMeetingAudioStorageService realtimeMeetingAudioStorageService;
private final RealtimeMeetingTranscriptCacheService realtimeMeetingTranscriptCacheService; private final RealtimeMeetingTranscriptCacheService realtimeMeetingTranscriptCacheService;
private final RealtimeAsrChannelFactory realtimeAsrChannelFactory; private final RealtimeAsrChannelFactory realtimeAsrChannelFactory;
private final ConcurrentMap<Long, MeetingChannelSession> meetingSessions = new ConcurrentHashMap<>(); private final ConcurrentMap<Long, MeetingChannelSession> meetingSessions = new ConcurrentHashMap<>();
private final ConcurrentMap<Long, Object> meetingLocks = new ConcurrentHashMap<>(); private final ConcurrentMap<Long, Object> meetingLocks = new ConcurrentHashMap<>();
@Override @Override
public void afterConnectionEstablished(WebSocketSession session) throws Exception { public void afterConnectionEstablished(WebSocketSession session) throws Exception {
String sessionToken = extractQueryParam(session.getUri(), "sessionToken"); String sessionToken = extractQueryParam(session.getUri(), "sessionToken");
RealtimeSocketSessionData sessionData = realtimeMeetingSocketSessionService.getSessionData(sessionToken); RealtimeSocketSessionData sessionData = realtimeMeetingSocketSessionService.getSessionData(sessionToken);
if (sessionData == null) { if (sessionData == null) {
log.warn("实时会议 websocket 拒绝连接：会话令牌无效，sessionId={}", session.getId()); log.warn("实时会议 websocket 拒绝连接：会话令牌无效，sessionId={}", session.getId());
session.close(CloseStatus.POLICY_VIOLATION.withReason("实时 Socket 会话无效")); session.close(CloseStatus.POLICY_VIOLATION.withReason("实时 Socket 会话无效"));
return; return;
} }
@ -69,65 +69,70 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
new ConcurrentWebSocketSessionDecorator(session, (int) Duration.ofSeconds(15).toMillis(), 1024 * 1024); new ConcurrentWebSocketSessionDecorator(session, (int) Duration.ofSeconds(15).toMillis(), 1024 * 1024);
session.getAttributes().put(ATTR_MEETING_ID, sessionData.getMeetingId()); session.getAttributes().put(ATTR_MEETING_ID, sessionData.getMeetingId());
session.getAttributes().put(ATTR_TARGET_WS_URL, sessionData.getTargetWsUrl()); 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_TEXT_COUNT, new AtomicInteger());
session.getAttributes().put(ATTR_FRONTEND_BINARY_COUNT, new AtomicInteger()); session.getAttributes().put(ATTR_FRONTEND_BINARY_COUNT, new AtomicInteger());
realtimeMeetingAudioStorageService.openSession(sessionData.getMeetingId(), session.getId()); realtimeMeetingAudioStorageService.openSession(sessionData.getMeetingId(), session.getId());
log.info("实时会议 websocket 已接入：meetingId={}, sessionId={}, provider={}, upstream={}", log.info("实时会议 websocket 已接入：meetingId={}, sessionId={}, provider={}, upstream={}",
sessionData.getMeetingId(), session.getId(), sessionData.getProvider(), sessionData.getTargetWsUrl()); sessionData.getMeetingId(), session.getId(), sessionData.getProvider(), sessionData.getTargetWsUrl());
attachFrontendSession(sessionData, session, frontendSession); attachFrontendSession(sessionData, session, frontendSession);
} }
@Override @Override
protected void handleTextMessage(WebSocketSession session, TextMessage message) { protected void handleTextMessage(WebSocketSession session, TextMessage message) {
MeetingChannelSession meetingSession = getMeetingSession(session); // 过滤前端发来的心跳保活消息，不转发给上游 ASR 服务
if (meetingSession == null || !meetingSession.isChannelOpen()) { if (looksLikeKeepaliveMessage(message.getPayload())) {
log.warn("前端文本消息已忽略：上游 ASR 连接不可用，meetingId={}, sessionId={}", 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()); session.getAttributes().get(ATTR_MEETING_ID), session.getId());
return; return;
} }
int count = nextCount(session, ATTR_FRONTEND_TEXT_COUNT); int count = nextCount(session, ATTR_FRONTEND_TEXT_COUNT);
String payload = message.getPayload(); String payload = message.getPayload();
log.info("前端文本 -> ASR 渠道：meetingId={}, sessionId={}, provider={}, count={}, payload={}", log.info("前端文本 -> ASR 渠道：meetingId={}, sessionId={}, provider={}, count={}, payload={}",
session.getAttributes().get(ATTR_MEETING_ID), session.getId(), session.getAttributes().get(ATTR_PROVIDER), count, summarizeText(payload)); session.getAttributes().get(ATTR_MEETING_ID), session.getId(), session.getAttributes().get(ATTR_PROVIDER), count, summarizeText(payload));
meetingSession.channel.handleFrontendText(meetingSession.context, payload); meetingSession.channel.handleFrontendText(meetingSession.context, payload);
} }
@Override @Override
protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) { protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) {
MeetingChannelSession meetingSession = getMeetingSession(session); MeetingChannelSession meetingSession = getMeetingSession(session);
if (meetingSession == null || !meetingSession.isChannelOpen()) { if (meetingSession == null || !meetingSession.isChannelOpen()) {
log.warn("前端音频帧已忽略：上游 ASR 连接不可用，meetingId={}, sessionId={}", log.warn("前端音频帧已忽略：上游 ASR 连接不可用，meetingId={}, sessionId={}",
session.getAttributes().get(ATTR_MEETING_ID), session.getId()); session.getAttributes().get(ATTR_MEETING_ID), session.getId());
return; return;
} }
int count = nextCount(session, ATTR_FRONTEND_BINARY_COUNT); int count = nextCount(session, ATTR_FRONTEND_BINARY_COUNT);
int bytes = message.getPayloadLength(); int bytes = message.getPayloadLength();
if (shouldLogBinaryFrame(count)) { if (shouldLogBinaryFrame(count)) {
log.info("前端音频帧 -> ASR 渠道：meetingId={}, sessionId={}, 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); session.getAttributes().get(ATTR_MEETING_ID), session.getId(), session.getAttributes().get(ATTR_PROVIDER), count, bytes);
} }
byte[] payload = toByteArray(message.getPayload()); byte[] payload = toByteArray(message.getPayload());
realtimeMeetingAudioStorageService.append(session.getId(), payload); realtimeMeetingAudioStorageService.append(session.getId(), payload);
meetingSession.channel.handleFrontendBinary(meetingSession.context, payload); meetingSession.channel.handleFrontendBinary(meetingSession.context, payload);
} }
@Override @Override
protected void handlePongMessage(WebSocketSession session, PongMessage message) { protected void handlePongMessage(WebSocketSession session, PongMessage message) {
if (getMeetingSession(session) == null) { // Pong 是浏览器对服务端（Tomcat）发出 Ping 帧的回应，代理层在此消化即可，无需转发给上游 ASR。
return; // 转发 Pong 给上游没有语义意义，且可能引起上游协议混乱。
} log.debug("Frontend pong received (keepalive): meetingId={}, sessionId={}",
log.debug("前端 pong 已在本地忽略：meetingId={}, sessionId={}, bytes={}", session.getAttributes().get(ATTR_MEETING_ID), session.getId());
session.getAttributes().get(ATTR_MEETING_ID), session.getId(), message.getPayloadLength());
} }
@Override @Override
public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception { 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); session.getAttributes().get(ATTR_MEETING_ID), session.getId(), session.getAttributes().get(ATTR_TARGET_WS_URL), exception);
detachFrontend(session); detachFrontend(session);
realtimeMeetingAudioStorageService.closeSession(session.getId()); realtimeMeetingAudioStorageService.closeSession(session.getId());
if (session.isOpen()) { if (session.isOpen()) {
session.close(CloseStatus.SERVER_ERROR); session.close(CloseStatus.SERVER_ERROR);
} }
@ -135,159 +140,159 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
@Override @Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) { 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()); session.getAttributes().get(ATTR_MEETING_ID), session.getId(), status.getCode(), status.getReason());
Object meetingIdValue = session.getAttributes().get(ATTR_MEETING_ID); Object meetingIdValue = session.getAttributes().get(ATTR_MEETING_ID);
if (meetingIdValue instanceof Long meetingId) { if (meetingIdValue instanceof Long meetingId) {
detachFrontend(meetingId, session.getId()); detachFrontend(meetingId, session.getId());
realtimeMeetingSessionStateService.pauseByDisconnect(meetingId, session.getId()); realtimeMeetingSessionStateService.pauseByDisconnect(meetingId, session.getId());
} }
realtimeMeetingAudioStorageService.closeSession(session.getId()); realtimeMeetingAudioStorageService.closeSession(session.getId());
} }
public void closeMeetingSession(Long meetingId) { public void closeMeetingSession(Long meetingId) {
if (meetingId == null) { if (meetingId == null) {
return; 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);
} }
if (meetingSession.hasOpenFrontend()) { MeetingChannelSession meetingSession = meetingSessions.get(meetingId);
sendFrontendError(frontendSession, "REALTIME_ACTIVE_CONNECTION_EXISTS", "当前会议已有活跃前端连接"); if (meetingSession == null || meetingSession.channel == null) {
frontendSession.close(CloseStatus.POLICY_VIOLATION.withReason("已存在活跃的前端连接")); return;
realtimeMeetingAudioStorageService.closeSession(rawSession.getId());
return;
} }
if (!realtimeMeetingSessionStateService.activate(meetingId, rawSession.getId())) { meetingSession.channel.closeMeeting(meetingSession.context);
sendFrontendError(frontendSession, "REALTIME_ACTIVE_CONNECTION_REJECTED", "当前状态下无法继续会议"); }
frontendSession.close(CloseStatus.POLICY_VIOLATION.withReason("当前状态下无法继续会议"));
realtimeMeetingAudioStorageService.closeSession(rawSession.getId()); private void attachFrontendSession(RealtimeSocketSessionData sessionData,
return; 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) { if (reused) {
sendProxyReady(frontendSession); sendProxyReady(frontendSession);
replayCachedMessages(meetingId, frontendSession); replayCachedMessages(meetingId, frontendSession);
return; 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;
} }
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) { try {
Object meetingIdValue = session.getAttributes().get(ATTR_MEETING_ID); meetingSession.channel.connect(meetingSession.context);
if (!(meetingIdValue instanceof Long meetingId)) { } catch (InterruptedException ex) {
return null; Thread.currentThread().interrupt();
} removeMeetingSession(meetingId, meetingSession);
return meetingSessions.get(meetingId); log.error("连接上游 ASR websocket 时被中断：meetingId={}, sessionId={}", meetingId, rawSession.getId(), ex);
} sendFrontendError(frontendSession, "REALTIME_UPSTREAM_CONNECT_INTERRUPTED", "连接上游 ASR 服务时被中断");
realtimeMeetingAudioStorageService.closeSession(rawSession.getId());
void removeMeetingSession(Long meetingId) { frontendSession.close(CloseStatus.SERVER_ERROR.withReason("连接上游服务时被中断"));
synchronized (lockForMeeting(meetingId)) { } catch (Exception ex) {
meetingSessions.remove(meetingId); removeMeetingSession(meetingId, meetingSession);
} log.warn("连接上游 ASR websocket 失败：meetingId={}, provider={}, target={}",
} meetingId, sessionData.getProvider(), sessionData.getTargetWsUrl(), ex);
sendFrontendError(frontendSession, "REALTIME_UPSTREAM_CONNECT_FAILED", "连接上游 ASR 服务失败");
void removeMeetingSession(Long meetingId, MeetingChannelSession meetingSession) { realtimeMeetingAudioStorageService.closeSession(rawSession.getId());
synchronized (lockForMeeting(meetingId)) { frontendSession.close(CloseStatus.SERVER_ERROR.withReason("连接 ASR WebSocket 失败"));
meetingSessions.remove(meetingId, meetingSession);
} }
} }
private Object lockForMeeting(Long meetingId) { private void replayCachedMessages(Long meetingId, ConcurrentWebSocketSessionDecorator frontendSession) {
return meetingLocks.computeIfAbsent(meetingId, ignored -> new Object()); 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) { private String extractQueryParam(URI uri, String key) {
if (uri == null || uri.getQuery() == null || uri.getQuery().isBlank()) { if (uri == null || uri.getQuery() == null || uri.getQuery().isBlank()) {
@ -316,18 +321,18 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
return 0; return 0;
} }
private void sendFrontendError(ConcurrentWebSocketSessionDecorator frontendSession, String code, String message) { private void sendFrontendError(ConcurrentWebSocketSessionDecorator frontendSession, String code, String message) {
try { try {
if (!frontendSession.isOpen()) { if (!frontendSession.isOpen()) {
return; return;
} }
Map<String, Object> payload = new HashMap<>(); Map<String, Object> payload = new HashMap<>();
payload.put("type", "error"); payload.put("type", "error");
payload.put("code", code); payload.put("code", code);
payload.put("message", message); payload.put("message", message);
frontendSession.sendMessage(new TextMessage(OBJECT_MAPPER.writeValueAsString(payload))); frontendSession.sendMessage(new TextMessage(OBJECT_MAPPER.writeValueAsString(payload)));
} catch (Exception ex) { } catch (Exception ex) {
log.warn("向前端发送实时代理错误消息失败：code={}", code, ex); log.warn("向前端发送实时代理错误消息失败：code={}", code, ex);
} }
} }
@ -346,108 +351,116 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
return normalized.substring(0, 240) + "..."; return normalized.substring(0, 240) + "...";
} }
static final class MeetingChannelSession { private boolean looksLikeKeepaliveMessage(String payload) {
private final Long meetingId; if (payload == null || payload.isBlank()) {
private final RealtimeAsrChannel channel; return false;
private final RealtimeAsrChannelContext context; }
String normalized = payload.replaceAll("\\s+", "");
private MeetingChannelSession(Long meetingId, RealtimeAsrChannel channel, RealtimeAsrChannelContext context) { return normalized.contains("\"type\":\"keepalive\"");
this.meetingId = meetingId;
this.channel = channel;
this.context = context;
} }
private void bindFrontend(WebSocketSession rawSession, ConcurrentWebSocketSessionDecorator frontendSession) { static final class MeetingChannelSession {
context.bindFrontendSession(rawSession, frontendSession); 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) { private void bindFrontend(WebSocketSession rawSession, ConcurrentWebSocketSessionDecorator frontendSession) {
if (context.getRawSession() != null && context.getRawSession().getId().equals(sessionId)) { context.bindFrontendSession(rawSession, frontendSession);
context.bindFrontendSession(null, null); }
}
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() { private final class HandlerChannelCallback implements RealtimeAsrChannelCallback {
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 {
@Override @Override
public void onChannelOpen(Long meetingId) throws Exception { public void onChannelOpen(Long meetingId) throws Exception {
MeetingChannelSession meetingSession = meetingSessions.get(meetingId); MeetingChannelSession meetingSession = meetingSessions.get(meetingId);
if (meetingSession == null) { if (meetingSession == null) {
return; return;
} }
ConcurrentWebSocketSessionDecorator frontendSession = meetingSession.context.getFrontendSession(); ConcurrentWebSocketSessionDecorator frontendSession = meetingSession.context.getFrontendSession();
if (frontendSession != null && frontendSession.isOpen()) { if (frontendSession != null && frontendSession.isOpen()) {
sendProxyReady(frontendSession); sendProxyReady(frontendSession);
} }
} }
@Override @Override
public void sendFrontendText(Long meetingId, String payload) throws Exception { public void sendFrontendText(Long meetingId, String payload) throws Exception {
MeetingChannelSession meetingSession = meetingSessions.get(meetingId); MeetingChannelSession meetingSession = meetingSessions.get(meetingId);
if (meetingSession == null) { if (meetingSession == null) {
return; return;
} }
ConcurrentWebSocketSessionDecorator frontendSession = meetingSession.context.getFrontendSession(); ConcurrentWebSocketSessionDecorator frontendSession = meetingSession.context.getFrontendSession();
if (frontendSession != null && frontendSession.isOpen()) { if (frontendSession != null && frontendSession.isOpen()) {
frontendSession.sendMessage(new TextMessage(payload)); frontendSession.sendMessage(new TextMessage(payload));
} }
} }
@Override @Override
public void sendFrontendBinary(Long meetingId, byte[] payload) throws Exception { public void sendFrontendBinary(Long meetingId, byte[] payload) throws Exception {
MeetingChannelSession meetingSession = meetingSessions.get(meetingId); MeetingChannelSession meetingSession = meetingSessions.get(meetingId);
if (meetingSession == null) { if (meetingSession == null) {
return; return;
} }
ConcurrentWebSocketSessionDecorator frontendSession = meetingSession.context.getFrontendSession(); ConcurrentWebSocketSessionDecorator frontendSession = meetingSession.context.getFrontendSession();
if (frontendSession != null && frontendSession.isOpen()) { if (frontendSession != null && frontendSession.isOpen()) {
frontendSession.sendMessage(new BinaryMessage(payload)); frontendSession.sendMessage(new BinaryMessage(payload));
} }
} }
@Override @Override
public void sendFrontendError(Long meetingId, String code, String message) { public void sendFrontendError(Long meetingId, String code, String message) {
MeetingChannelSession meetingSession = meetingSessions.get(meetingId); MeetingChannelSession meetingSession = meetingSessions.get(meetingId);
if (meetingSession == null) { if (meetingSession == null) {
return; return;
} }
ConcurrentWebSocketSessionDecorator frontendSession = meetingSession.context.getFrontendSession(); ConcurrentWebSocketSessionDecorator frontendSession = meetingSession.context.getFrontendSession();
if (frontendSession != null) { if (frontendSession != null) {
RealtimeMeetingProxyWebSocketHandler.this.sendFrontendError(frontendSession, code, message); RealtimeMeetingProxyWebSocketHandler.this.sendFrontendError(frontendSession, code, message);
} }
} }
@Override @Override
public void removeMeetingSession(Long meetingId) { public void removeMeetingSession(Long meetingId) {
RealtimeMeetingProxyWebSocketHandler.this.removeMeetingSession(meetingId); RealtimeMeetingProxyWebSocketHandler.this.removeMeetingSession(meetingId);
} }
@Override @Override
public void closeFrontend(Long meetingId, CloseStatus status) { public void closeFrontend(Long meetingId, CloseStatus status) {
MeetingChannelSession meetingSession = meetingSessions.get(meetingId); MeetingChannelSession meetingSession = meetingSessions.get(meetingId);
if (meetingSession == null) { if (meetingSession == null) {
return; return;
} }
try { try {
WebSocketSession rawSession = meetingSession.context.getRawSession(); WebSocketSession rawSession = meetingSession.context.getRawSession();
if (rawSession != null && rawSession.isOpen()) { if (rawSession != null && rawSession.isOpen()) {
rawSession.close(status); rawSession.close(status);
} }
} catch (Exception ignored) { } catch (Exception ignored) {

60
backend/src/test/java/com/imeeting/service/biz/impl/AiModelServiceImplTest.java
View File

@ -349,6 +349,30 @@ class AiModelServiceImplTest {
assertNull(captor.getValue().getApiKey()); 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 @Test
void saveModelShouldRejectTencentAsrWithoutSecretKey() { void saveModelShouldRejectTencentAsrWithoutSecretKey() {
AiModelServiceImpl service = new AiModelServiceImpl( AiModelServiceImpl service = new AiModelServiceImpl(
@ -370,7 +394,32 @@ class AiModelServiceImplTest {
)); ));
RuntimeException ex = assertThrows(RuntimeException.class, () -> service.saveModel(dto)); 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 @Test
@ -388,13 +437,14 @@ class AiModelServiceImplTest {
dto.setModelType("ASR"); dto.setModelType("ASR");
dto.setModelName("tencent-asr"); dto.setModelName("tencent-asr");
dto.setProvider("tencent"); dto.setProvider("tencent");
dto.setModelCode("16k_zh");
dto.setIsDefault(0); dto.setIsDefault(0);
dto.setStatus(1); dto.setStatus(1);
dto.setMediaConfig(Map.of( dto.setMediaConfig(Map.of(
"tencentAppId", "app-id", "tencentAppId", "app-id",
"tencentSecretId", "secret-id", "tencentSecretId", "secret-id",
"tencentSecretKey", "secret-key" "tencentSecretKey", "secret-key",
"tencentOfflineModelCode", "16k_zh",
"tencentRealtimeModelCode", "16k_zh_realtime"
)); ));
service.saveModel(dto); service.saveModel(dto);
@ -402,7 +452,9 @@ class AiModelServiceImplTest {
ArgumentCaptor<AsrModel> captor = ArgumentCaptor.forClass(AsrModel.class); ArgumentCaptor<AsrModel> captor = ArgumentCaptor.forClass(AsrModel.class);
verify(asrModelMapper, times(1)).insert(captor.capture()); verify(asrModelMapper, times(1)).insert(captor.capture());
assertEquals("tencent", captor.getValue().getProvider()); 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")); assertEquals("secret-key", captor.getValue().getMediaConfig().get("tencentSecretKey"));
assertNull(captor.getValue().getBaseUrl()); assertNull(captor.getValue().getBaseUrl());
} }

460
backend/src/test/java/com/imeeting/service/biz/impl/AiTaskServiceImplTest.java
View File

@ -1,152 +1,308 @@
//package com.imeeting.service.biz.impl; package com.imeeting.service.biz.impl;
//
//import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectMapper;
//import com.imeeting.dto.biz.AiModelVO; import com.imeeting.dto.biz.AiModelVO;
//import com.imeeting.entity.biz.AiTask; import com.imeeting.entity.biz.AiTask;
//import com.imeeting.entity.biz.HotWord; import com.imeeting.entity.biz.Meeting;
//import com.imeeting.entity.biz.Meeting; import com.imeeting.mapper.biz.AiTaskMapper;
//import com.imeeting.mapper.biz.MeetingMapper; import com.imeeting.mapper.biz.MeetingMapper;
//import com.imeeting.mapper.biz.MeetingTranscriptMapper; import com.imeeting.mapper.biz.MeetingTranscriptMapper;
//import com.imeeting.service.biz.AiModelService; import com.imeeting.service.android.AndroidMeetingPushService;
//import com.imeeting.service.biz.HotWordService; import com.imeeting.service.biz.AiModelService;
//import com.imeeting.service.biz.MeetingProgressService; import com.imeeting.service.biz.HotWordService;
//import com.imeeting.service.biz.MeetingSummaryFileService; import com.imeeting.service.biz.MeetingPointsService;
//import com.imeeting.service.biz.MeetingTranscriptChapterService; import com.imeeting.service.biz.MeetingProgressService;
//import com.imeeting.service.biz.MeetingTranscriptFileService; import com.imeeting.service.biz.MeetingSummaryFileService;
//import com.imeeting.support.RedisValueSupport; import com.imeeting.service.biz.MeetingTranscriptChapterService;
//import com.imeeting.support.TaskSecurityContextRunner; import com.imeeting.service.biz.MeetingTranscriptFileService;
//import com.unisbase.mapper.SysUserMapper; import com.imeeting.support.TaskSecurityContextRunner;
//import com.unisbase.service.SysParamService; import com.imeeting.support.redis.MeetingAsrPermitCache;
//import org.junit.jupiter.api.Test; import com.imeeting.support.redis.MeetingLockCache;
//import org.springframework.test.util.ReflectionTestUtils; import com.unisbase.mapper.SysUserMapper;
// import com.unisbase.service.SysParamService;
//import java.util.HashMap; import com.tencentcloudapi.asr.v20190614.models.SentenceDetail;
//import java.util.List; import com.tencentcloudapi.asr.v20190614.models.TaskStatus;
//import java.util.Map; import org.junit.jupiter.api.Test;
// import org.springframework.test.util.ReflectionTestUtils;
//import static org.junit.jupiter.api.Assertions.assertEquals;
//import static org.junit.jupiter.api.Assertions.assertFalse; import java.util.HashMap;
//import static org.junit.jupiter.api.Assertions.assertNull; import java.util.Map;
//import static org.junit.jupiter.api.Assertions.assertTrue;
//import static org.mockito.ArgumentMatchers.any; import static org.junit.jupiter.api.Assertions.assertEquals;
//import static org.mockito.Mockito.mock; import static org.mockito.ArgumentMatchers.any;
//import static org.mockito.Mockito.when; import static org.mockito.ArgumentMatchers.anyString;
// import static org.mockito.ArgumentMatchers.eq;
//class AiTaskServiceImplTest { import static org.mockito.Mockito.doReturn;
// import static org.mockito.Mockito.mock;
// @Test import static org.mockito.Mockito.never;
// void buildAsrRequestShouldFollowCurrentOfflineAsrContract() { import static org.mockito.Mockito.spy;
// HotWordService hotWordService = mock(HotWordService.class); import static org.mockito.Mockito.verify;
// HotWord hotWord = new HotWord(); import static org.mockito.Mockito.when;
// hotWord.setWord("汇智");
// hotWord.setWeight(25); class AiTaskServiceImplTest {
// when(hotWordService.list(any())).thenReturn(List.of(hotWord));
// @Test
// AiTaskServiceImpl service = new AiTaskServiceImpl( void processAsrTaskShouldUseTencentOfflineBranchWhenProviderIsTencent() throws Exception {
// mock(MeetingMapper.class), MeetingMapper meetingMapper = mock(MeetingMapper.class);
// mock(MeetingTranscriptMapper.class), MeetingTranscriptMapper transcriptMapper = mock(MeetingTranscriptMapper.class);
// mock(AiModelService.class), AiModelService aiModelService = mock(AiModelService.class);
// new ObjectMapper(), HotWordService hotWordService = mock(HotWordService.class);
// mock(SysUserMapper.class), MeetingLockCache meetingLockCache = mock(MeetingLockCache.class);
// hotWordService, MeetingAsrPermitCache meetingAsrPermitCache = mock(MeetingAsrPermitCache.class);
// mock(RedisValueSupport.class), MeetingProgressService meetingProgressService = mock(MeetingProgressService.class);
// mock(MeetingProgressService.class), MeetingPointsService meetingPointsService = mock(MeetingPointsService.class);
// mock(MeetingSummaryFileService.class), MeetingSummaryFileService meetingSummaryFileService = mock(MeetingSummaryFileService.class);
// mock(MeetingTranscriptFileService.class), MeetingTranscriptFileService meetingTranscriptFileService = mock(MeetingTranscriptFileService.class);
// mock(MeetingTranscriptChapterService.class), MeetingTranscriptChapterService meetingTranscriptChapterService = mock(MeetingTranscriptChapterService.class);
// mock(MeetingSummaryPromptAssembler.class), MeetingSummaryPromptAssembler meetingSummaryPromptAssembler = mock(MeetingSummaryPromptAssembler.class);
// mock(TaskSecurityContextRunner.class), TaskSecurityContextRunner taskSecurityContextRunner = mock(TaskSecurityContextRunner.class);
// mock(MeetingExternalSummaryWebhookTrigger.class), MeetingExternalSummaryWebhookTrigger meetingExternalSummaryWebhookTrigger = mock(MeetingExternalSummaryWebhookTrigger.class);
// mock(SysParamService.class) SysParamService sysParamService = mock(SysParamService.class);
// );
// ReflectionTestUtils.setField(service, "serverBaseUrl", "http://localhost:8080"); AiTaskServiceImpl service = spy(new AiTaskServiceImpl(
// meetingMapper,
// Meeting meeting = new Meeting(); transcriptMapper,
// meeting.setAudioUrl("/api/static/meetings/12/source audio.mp4"); aiModelService,
// new ObjectMapper(),
// AiTask task = new AiTask(); mock(SysUserMapper.class),
// Map<String, Object> taskConfig = new HashMap<>(); hotWordService,
// taskConfig.put("useSpkId", 1); meetingLockCache,
// taskConfig.put("enableTextRefine", true); meetingAsrPermitCache,
// taskConfig.put("hotWords", List.of("汇智")); meetingProgressService,
// task.setTaskConfig(taskConfig); meetingPointsService,
// meetingSummaryFileService,
// AiModelVO asrModel = new AiModelVO(); meetingTranscriptFileService,
// asrModel.setModelCode("legacy-model-code"); meetingTranscriptChapterService,
// meetingSummaryPromptAssembler,
// @SuppressWarnings("unchecked") taskSecurityContextRunner,
// Map<String, Object> request = (Map<String, Object>) ReflectionTestUtils.invokeMethod( meetingExternalSummaryWebhookTrigger,
// service, sysParamService
// "buildAsrRequest", ));
// meeting, ReflectionTestUtils.setField(service, "baseMapper", mock(AiTaskMapper.class));
// task, ReflectionTestUtils.setField(service, "androidMeetingPushService", mock(AndroidMeetingPushService.class));
// asrModel
// ); Meeting meeting = new Meeting();
// meeting.setId(1L);
// assertEquals("http://localhost:8080/api/static/meetings/12/source%20audio.mp4", request.get("audio_address")); meeting.setAudioUrl("https://cdn.example.com/audio/demo.m4a");
// assertFalse(request.containsKey("file_url"));
// AiTask task = new AiTask();
// @SuppressWarnings("unchecked") task.setId(11L);
// Map<String, Object> config = (Map<String, Object>) request.get("config"); task.setMeetingId(1L);
// assertEquals(Boolean.TRUE, config.get("enable_speaker")); task.setTaskType("ASR");
// assertEquals(Boolean.TRUE, config.get("match_speaker_registry")); task.setTaskConfig(new HashMap<>(Map.of(
// assertEquals(Boolean.TRUE, config.get("enable_text_cleanup")); "asrModelId", 101L,
// assertFalse(config.containsKey("enable_text_refine")); "useSpkId", 1,
// assertFalse(config.containsKey("enable_two_pass")); "enableTextRefine", true
// assertFalse(config.containsKey("model")); )));
//
// @SuppressWarnings("unchecked") AiModelVO model = new AiModelVO();
// List<Map<String, Object>> hotwords = (List<Map<String, Object>>) config.get("hotwords"); model.setId(101L);
// assertEquals(1, hotwords.size()); model.setProvider("tencent");
// assertEquals("汇智", hotwords.get(0).get("hotword")); model.setModelCode("16k_zh");
// assertEquals(2.5, hotwords.get(0).get("weight")); model.setMediaConfig(Map.of(
// } "tencentAppId", "123456",
// "tencentSecretId", "secret-id",
// @Test "tencentSecretKey", "secret-key"
// void buildAsrRequestShouldDisableRegistryMatchWhenSpeakerSplitDisabled() { ));
// AiTaskServiceImpl service = new AiTaskServiceImpl( when(aiModelService.getModelById(101L, "ASR")).thenReturn(model);
// mock(MeetingMapper.class),
// mock(MeetingTranscriptMapper.class), doReturn(true).when(service).updateById(any(AiTask.class));
// mock(AiModelService.class), doReturn("腾讯离线转写结果").when(service).processTencentOfflineAsr(eq(meeting), eq(task), eq(model));
// new ObjectMapper(),
// mock(SysUserMapper.class), String result = ReflectionTestUtils.invokeMethod(service, "processAsrTask", meeting, task);
// mock(HotWordService.class),
// mock(RedisValueSupport.class), assertEquals("腾讯离线转写结果", result);
// mock(MeetingProgressService.class), verify(service).processTencentOfflineAsr(meeting, task, model);
// mock(MeetingSummaryFileService.class), verify(meetingPointsService).recordAsrSuccessCharge(meeting, task);
// mock(MeetingTranscriptFileService.class), }
// mock(MeetingTranscriptChapterService.class),
// mock(MeetingSummaryPromptAssembler.class), @Test
// mock(TaskSecurityContextRunner.class), void buildTencentOfflineCreateRequestShouldMapMeetingAndTaskConfig() {
// mock(MeetingExternalSummaryWebhookTrigger.class), AiTaskServiceImpl service = createService(mock(MeetingPointsService.class));
// mock(SysParamService.class) ReflectionTestUtils.setField(service, "serverBaseUrl", "https://server.example.com");
// );
// ReflectionTestUtils.setField(service, "serverBaseUrl", "http://localhost:8080"); Meeting meeting = new Meeting();
// meeting.setId(2L);
// Meeting meeting = new Meeting(); meeting.setAudioUrl("/upload/audio/demo.m4a");
// meeting.setAudioUrl("/api/static/audio/demo.wav");
// AiTask task = new AiTask();
// AiTask task = new AiTask(); task.setTaskConfig(new HashMap<>(Map.of(
// Map<String, Object> taskConfig = new HashMap<>(); "useSpkId", 1,
// taskConfig.put("useSpkId", 0); "enableTextRefine", true,
// taskConfig.put("enableTextRefine", false); "hotWords", java.util.List.of("腾讯会议", "离线转写")
// task.setTaskConfig(taskConfig); )));
//
// @SuppressWarnings("unchecked") AiModelVO model = new AiModelVO();
// Map<String, Object> request = (Map<String, Object>) ReflectionTestUtils.invokeMethod( model.setModelCode("legacy-model-code");
// service, model.setMediaConfig(Map.of(
// "buildAsrRequest", "tencentAppId", "123456",
// meeting, "tencentSecretId", "secret-id",
// task, "tencentSecretKey", "secret-key",
// new AiModelVO() "tencentOfflineModelCode", "16k_zh",
// ); "tencentRealtimeModelCode", "16k_zh_realtime"
// ));
// @SuppressWarnings("unchecked")
// Map<String, Object> config = (Map<String, Object>) request.get("config"); @SuppressWarnings("unchecked")
// assertEquals(Boolean.FALSE, config.get("enable_speaker")); Map<String, Object> request = (Map<String, Object>) ReflectionTestUtils.invokeMethod(
// assertEquals(Boolean.FALSE, config.get("match_speaker_registry")); service,
// assertEquals(Boolean.FALSE, config.get("enable_text_cleanup")); "buildTencentOfflineCreateRequest",
// assertTrue(((List<?>) config.get("hotwords")).isEmpty()); meeting,
// assertNull(request.get("file_url")); 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<String, Object> request = (Map<String, Object>) 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;
}
}

56
frontend/src/pages/business/AiModels.tsx
View File

@ -24,6 +24,7 @@ import {
testLocalModelConnectivity, testLocalModelConnectivity,
updateAiModel, updateAiModel,
} from "../../api/business/aimodel"; } from "../../api/business/aimodel";
import {getMeetingCreateConfig, type MeetingCreateConfig} from "../../api/business/meeting";
import AppPagination from "../../components/shared/AppPagination"; import AppPagination from "../../components/shared/AppPagination";
const { Option } = Select; const { Option } = Select;
@ -31,6 +32,12 @@ const { Title } = Typography;
type ModelType = "ASR" | "LLM"; type ModelType = "ASR" | "LLM";
const DEFAULT_CREATE_CONFIG: MeetingCreateConfig = {
offlineEnabled: true,
realtimeEnabled: false,
offlineAudioMaxSizeMb: 1024,
};
const PROVIDER_BASE_URL_MAP: Record<string, string> = { const PROVIDER_BASE_URL_MAP: Record<string, string> = {
openai: "https://api.openai.com", openai: "https://api.openai.com",
deepseek: "https://api.deepseek.com", deepseek: "https://api.deepseek.com",
@ -64,6 +71,7 @@ const AiModels: React.FC = () => {
const [connectivityLoading, setConnectivityLoading] = useState(false); const [connectivityLoading, setConnectivityLoading] = useState(false);
const [remoteModels, setRemoteModels] = useState<string[]>([]); const [remoteModels, setRemoteModels] = useState<string[]>([]);
const [speakerModels, setSpeakerModels] = useState<string[]>([]); const [speakerModels, setSpeakerModels] = useState<string[]>([]);
const [createConfig, setCreateConfig] = useState<MeetingCreateConfig>(DEFAULT_CREATE_CONFIG);
const modelNameAutoFilledRef = useRef(false); const modelNameAutoFilledRef = useRef(false);
const localProfileLoadedRef = useRef(false); const localProfileLoadedRef = useRef(false);
@ -86,6 +94,20 @@ const AiModels: React.FC = () => {
void fetchData(); void fetchData();
}, [current, size, searchName, activeType]); }, [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(() => { useEffect(() => {
if (!drawerVisible || !provider) { if (!drawerVisible || !provider) {
return; return;
@ -137,6 +159,8 @@ const AiModels: React.FC = () => {
const tencentAppId = record.mediaConfig?.tencentAppId; const tencentAppId = record.mediaConfig?.tencentAppId;
const tencentSecretId = record.mediaConfig?.tencentSecretId; const tencentSecretId = record.mediaConfig?.tencentSecretId;
const tencentSecretKey = record.mediaConfig?.tencentSecretKey; const tencentSecretKey = record.mediaConfig?.tencentSecretKey;
const tencentOfflineModelCode = record.mediaConfig?.tencentOfflineModelCode || record.modelCode;
const tencentRealtimeModelCode = record.mediaConfig?.tencentRealtimeModelCode || record.modelCode;
form.setFieldsValue({ form.setFieldsValue({
...record, ...record,
modelType: record.modelType, modelType: record.modelType,
@ -145,6 +169,8 @@ const AiModels: React.FC = () => {
tencentAppId, tencentAppId,
tencentSecretId, tencentSecretId,
tencentSecretKey, tencentSecretKey,
tencentOfflineModelCode,
tencentRealtimeModelCode,
isDefaultChecked: record.isDefault === 1, isDefaultChecked: record.isDefault === 1,
statusChecked: record.status === 1, statusChecked: record.status === 1,
}); });
@ -261,8 +287,8 @@ const AiModels: React.FC = () => {
baseUrl: values.baseUrl, baseUrl: values.baseUrl,
apiPath: values.apiPath, apiPath: values.apiPath,
apiKey: values.apiKey, apiKey: values.apiKey,
modelCode: values.modelCode, modelCode: activeType === "ASR" && isTencentProvider ? values.tencentOfflineModelCode : values.modelCode,
wsUrl: values.wsUrl, wsUrl: activeType === "ASR" ? values.wsUrl : undefined,
mediaConfig: mediaConfig:
activeType === "ASR" && isLocalProvider activeType === "ASR" && isLocalProvider
? { ? {
@ -274,6 +300,8 @@ const AiModels: React.FC = () => {
tencentAppId: values.tencentAppId, tencentAppId: values.tencentAppId,
tencentSecretId: values.tencentSecretId, tencentSecretId: values.tencentSecretId,
tencentSecretKey: values.tencentSecretKey, tencentSecretKey: values.tencentSecretKey,
tencentOfflineModelCode: values.tencentOfflineModelCode,
tencentRealtimeModelCode: values.tencentRealtimeModelCode,
} }
: undefined, : undefined,
temperature: values.temperature, temperature: values.temperature,
@ -567,13 +595,14 @@ const AiModels: React.FC = () => {
<Form.Item <Form.Item
label="模型名称" label="模型名称"
required={activeType === "LLM"} required={activeType === "LLM"}
hidden={activeType === "ASR" && isTencentProvider}
tooltip="可从远程列表选择，也可手动输入；值将作为模型 code 传给后端" tooltip="可从远程列表选择，也可手动输入；值将作为模型 code 传给后端"
> >
<Space.Compact style={{ width: "100%" }}> <Space.Compact style={{ width: "100%" }}>
<Form.Item <Form.Item
name="modelCode" name="modelCode"
noStyle noStyle
rules={activeType === "LLM" || isTencentProvider ? [{ rules={activeType === "LLM" ? [{
required: true, required: true,
message: "请输入或选择模型名称" message: "请输入或选择模型名称"
}] : []} }] : []}
@ -602,7 +631,8 @@ const AiModels: React.FC = () => {
</Space.Compact> </Space.Compact>
</Form.Item> </Form.Item>
<Form.Item name="wsUrl" label="WebSocket 地址" hidden> <Form.Item name="wsUrl" label="WebSocket 地址"
hidden={!(activeType === "ASR" && createConfig.realtimeEnabled)}>
<Input placeholder="wss://api.example.com/v1/ws" /> <Input placeholder="wss://api.example.com/v1/ws" />
</Form.Item> </Form.Item>
@ -660,6 +690,24 @@ const AiModels: React.FC = () => {
<Input.Password/> <Input.Password/>
</Form.Item> </Form.Item>
</Col> </Col>
<Col span={12}>
<Form.Item
name="tencentOfflineModelCode"
label="离线识别模型名"
rules={[{required: true, message: "请输入离线识别模型名"}]}
>
<Input placeholder="例如：16k_zh"/>
</Form.Item>
</Col>
<Col span={12}>
<Form.Item
name="tencentRealtimeModelCode"
label="实时识别模型名"
rules={[{required: true, message: "请输入实时识别模型名"}]}
>
<Input placeholder="例如：16k_zh_realtime"/>
</Form.Item>
</Col>
</Row> </Row>
)} )}

134
frontend/src/pages/business/RealtimeAsrSession.tsx
View File

@ -244,6 +244,7 @@ export function RealtimeAsrSession() {
const [sessionStatus, setSessionStatus] = useState<RealtimeMeetingSessionStatus | null>(null); const [sessionStatus, setSessionStatus] = useState<RealtimeMeetingSessionStatus | null>(null);
const transcriptRef = useRef<HTMLDivElement | null>(null); const transcriptRef = useRef<HTMLDivElement | null>(null);
const wsRef = useRef<WebSocket | null>(null); const wsRef = useRef<WebSocket | null>(null);
const wsHeartbeatRef = useRef<ReturnType<typeof setInterval> | null>(null);
const audioContextRef = useRef<AudioContext | null>(null); const audioContextRef = useRef<AudioContext | null>(null);
const processorRef = useRef<ScriptProcessorNode | null>(null); const processorRef = useRef<ScriptProcessorNode | null>(null);
const audioSourceRef = useRef<MediaStreamAudioSourceNode | null>(null); const audioSourceRef = useRef<MediaStreamAudioSourceNode | null>(null);
@ -373,6 +374,33 @@ export function RealtimeAsrSession() {
return () => window.removeEventListener("pagehide", handlePageHide); return () => window.removeEventListener("pagehide", handlePageHide);
}, [meetingId]); }, [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 () => { const shutdownAudioPipeline = async () => {
processorRef.current?.disconnect(); processorRef.current?.disconnect();
audioSourceRef.current?.disconnect(); audioSourceRef.current?.disconnect();
@ -433,6 +461,10 @@ export function RealtimeAsrSession() {
setRecording(false); setRecording(false);
setStatusText("连接失败"); setStatusText("连接失败");
sessionStartedRef.current = false; sessionStartedRef.current = false;
if (wsHeartbeatRef.current !== null) {
clearInterval(wsHeartbeatRef.current);
wsHeartbeatRef.current = null;
}
wsRef.current?.close(); wsRef.current?.close();
wsRef.current = null; wsRef.current = null;
await shutdownAudioPipeline(); await shutdownAudioPipeline();
@ -442,6 +474,30 @@ export function RealtimeAsrSession() {
message.error(errorMessage); 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 () => { const startAudioPipeline = async () => {
if (!window.isSecureContext || !navigator.mediaDevices?.getUserMedia) { if (!window.isSecureContext || !navigator.mediaDevices?.getUserMedia) {
throw new Error("当前浏览器环境不支持麦克风访问。请使用 localhost 或 HTTPS 域名访问系统。"); throw new Error("当前浏览器环境不支持麦克风访问。请使用 localhost 或 HTTPS 域名访问系统。");
@ -506,6 +562,13 @@ export function RealtimeAsrSession() {
} }
const pauseRes = await pauseRealtimeMeeting(meetingId); const pauseRes = await pauseRealtimeMeeting(meetingId);
await closeFrontendSocket(false); await closeFrontendSocket(false);
if (wsHeartbeatRef.current !== null) {
clearInterval(wsHeartbeatRef.current);
wsHeartbeatRef.current = null;
}
wsRef.current?.close();
wsRef.current = null;
sessionStartedRef.current = false;
await shutdownAudioPipeline(); await shutdownAudioPipeline();
setSessionStatus(pauseRes.data.data); setSessionStatus(pauseRes.data.data);
setRecording(false); setRecording(false);
@ -551,12 +614,41 @@ export function RealtimeAsrSession() {
hotwords: sessionDraft.hotwords || [], hotwords: sessionDraft.hotwords || [],
}); });
const socketSession = socketSessionRes.data.data; 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)); const socket = new WebSocket(buildRealtimeProxyWsUrl(socketSession));
socket.binaryType = "arraybuffer"; socket.binaryType = "arraybuffer";
wsRef.current = socket; wsRef.current = socket;
socket.onopen = () => { socket.onopen = () => {
setStatusText("识别服务连接中，等待第三方服务就绪..."); 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) => { socket.onmessage = (event) => {
@ -585,7 +677,11 @@ export function RealtimeAsrSession() {
if ((payload.code || payload.type === "error") && payload.message) { if ((payload.code || payload.type === "error") && payload.message) {
setStatusText(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; return;
} }
@ -613,6 +709,10 @@ export function RealtimeAsrSession() {
}; };
socket.onclose = () => { socket.onclose = () => {
if (wsHeartbeatRef.current !== null) {
clearInterval(wsHeartbeatRef.current);
wsHeartbeatRef.current = null;
}
setConnecting(false); setConnecting(false);
setRecording(false); setRecording(false);
sessionStartedRef.current = false; sessionStartedRef.current = false;
@ -640,6 +740,16 @@ export function RealtimeAsrSession() {
setStatusText("结束会议中..."); setStatusText("结束会议中...");
await closeFrontendSocket(true); 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(); await shutdownAudioPipeline();
@ -804,17 +914,17 @@ export function RealtimeAsrSession() {
<Text type="secondary" style={{ fontSize: 13 }}> <Text type="secondary" style={{ fontSize: 13 }}>
字数 <strong style={{ color: "#334155" }}>{totalTranscriptChars}</strong> 字数 <strong style={{ color: "#334155" }}>{totalTranscriptChars}</strong>
</Text> </Text>
<Text type="secondary" style={{ fontSize: 13 }}> {/*<Text type="secondary" style={{ fontSize: 13 }}>*/}
模型 <strong style={{ color: "#334155" }}>{sessionDraft.asrModelName}</strong> {/* 模型 <strong style={{ color: "#334155" }}>{sessionDraft.asrModelName}</strong>*/}
</Text> {/*</Text>*/}
<Text type="secondary" style={{ fontSize: 13 }}> {/*<Text type="secondary" style={{ fontSize: 13 }}>*/}
模式 <strong style={{ color: "#334155" }}>{sessionDraft.mode}</strong> {/* 模式 <strong style={{ color: "#334155" }}>{sessionDraft.mode}</strong>*/}
</Text> {/*</Text>*/}
{sessionDraft.hotwords && sessionDraft.hotwords.length > 0 && ( {/*{sessionDraft.hotwords && sessionDraft.hotwords.length > 0 && (*/}
<Text type="secondary" style={{ fontSize: 13 }}> {/* <Text type="secondary" style={{ fontSize: 13 }}>*/}
热词 <strong style={{ color: "#334155" }}>{sessionDraft.hotwords.length}</strong> {/* 热词 <strong style={{ color: "#334155" }}>{sessionDraft.hotwords.length}</strong>*/}
</Text> {/* </Text>*/}
)} {/*)}*/}
</Space> </Space>
)} )}
</div> </div>