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

Compare commits

base: chenh:77fe6d4e779077de119ee23c3807ff6bbd0f4db4
Branches Tags
chenh:dev_na
chenh:master
..
compare: chenh:033ffaacc9807547695299a447625f31f67934aa
Branches Tags
chenh:dev_na
chenh:master

No commits in common. "77fe6d4e779077de119ee23c3807ff6bbd0f4db4" and "033ffaacc9807547695299a447625f31f67934aa" have entirely different histories.
77fe6d4e77 ... 033ffaacc9

29 changed files with 574 additions and 2187 deletions
Show Stats Expand all files Collapse all files

6
backend/pom.xml
View File

@ -177,12 +177,6 @@
<artifactId>springdoc-openapi-starter-webmvc-ui</artifactId>
<version>2.3.0</version>
</dependency>
<!-- Source: https://mvnrepository.com/artifact/com.tencentcloudapi/tencentcloud-speech-sdk-java -->
<dependency>
<groupId>com.tencentcloudapi</groupId>
<artifactId>tencentcloud-speech-sdk-java</artifactId>
<version>1.0.67</version>
</dependency>
</dependencies>
<build>

4
backend/src/main/java/com/imeeting/common/RedisKeys.java
View File

@ -71,10 +71,6 @@ public final class RedisKeys {
return "biz:meeting:realtime:state:" + meetingId;
}
public static String realtimeMeetingTranscriptCacheKey(Long meetingId) {
return "biz:meeting:realtime:transcript-cache:" + meetingId;
}
public static String realtimeMeetingResumeTimeoutKey(Long meetingId) {
return realtimeMeetingResumeTimeoutPrefix() + meetingId;
}

3
backend/src/main/java/com/imeeting/controller/biz/AiModelController.java
View File

@ -3,7 +3,6 @@ package com.imeeting.controller.biz;
import com.imeeting.dto.biz.AiLocalProfileVO;
import com.imeeting.dto.biz.AiModelDTO;
import com.imeeting.dto.biz.AiModelVO;
import com.imeeting.enums.ModelProviderEnum;
import com.imeeting.service.biz.AiModelService;
import com.unisbase.common.ApiResponse;
import com.unisbase.common.annotation.Log;
@ -122,7 +121,7 @@ public class AiModelController {
@PostMapping("/llm-connectivity-test")
@PreAuthorize("isAuthenticated()")
public ApiResponse<Boolean> testLlmConnectivity(@RequestBody AiModelDTO dto) {
if (ModelProviderEnum.LOCAL.getCode().equalsIgnoreCase(dto.getProvider()) && (dto.getBaseUrl() == null || dto.getBaseUrl().isBlank())) {
if ("custom".equalsIgnoreCase(dto.getProvider()) && (dto.getBaseUrl() == null || dto.getBaseUrl().isBlank())) {
return ApiResponse.error("基础地址不能为空");
}
if (dto.getModelCode() == null || dto.getModelCode().isBlank()) {

11
backend/src/main/java/com/imeeting/controller/biz/MeetingController.java
View File

@ -399,6 +399,17 @@ public class MeetingController {
return ApiResponse.ok(result);
}
@Operation(summary = "追加实时转写片段")
@PostMapping("/{id}/realtime/transcripts")
@PreAuthorize("isAuthenticated()")
public ApiResponse<Boolean> appendRealtimeTranscripts(@PathVariable Long id, @RequestBody List<RealtimeTranscriptItemDTO> items) {
LoginUser loginUser = currentLoginUser();
Meeting meeting = meetingAccessService.requireMeeting(id);
meetingAccessService.assertCanControlRealtimeMeeting(meeting, loginUser, MeetingConstants.SOURCE_WEB);
meetingCommandService.appendRealtimeTranscripts(id, items);
return ApiResponse.ok(true);
}
@Operation(summary = "暂停实时会议")
@PostMapping("/{id}/realtime/pause")
@PreAuthorize("isAuthenticated()")

2
backend/src/main/java/com/imeeting/dto/biz/RealtimeMeetingResumeConfig.java
View File

@ -27,6 +27,4 @@ public class RealtimeMeetingResumeConfig {
private Boolean saveAudio;
@Schema(description = "热词列表")
private List<Map<String, Object>> hotwords;
@Schema(description = "腾讯说话人上下文 ID")
private String speakerContextId;
}

22
backend/src/main/java/com/imeeting/dto/biz/RealtimeMeetingTranscriptCacheItem.java
View File

@ -1,22 +0,0 @@
package com.imeeting.dto.biz;
import lombok.Data;
@Data
public class RealtimeMeetingTranscriptCacheItem {
private String sentenceKey;
private String sentenceGroupKey;
private Integer sentenceId;
private Integer sentenceType;
private String speakerId;
private String speakerName;
private String userId;
private Integer startTime;
private Integer endTime;
private String content;
private Integer sortOrder;
private Boolean finalResult;
private Long transcriptId;
private Long firstReceivedAt;
private Long updatedAt;
}

15
backend/src/main/java/com/imeeting/dto/biz/RealtimeMeetingTranscriptCacheState.java
View File

@ -1,15 +0,0 @@
package com.imeeting.dto.biz;
import lombok.Data;
import java.util.ArrayList;
import java.util.List;
@Data
public class RealtimeMeetingTranscriptCacheState {
private Long meetingId;
private Integer nextSortOrder;
private Integer nextLegacySequence;
private List<RealtimeMeetingTranscriptCacheItem> items = new ArrayList<>();
private Long updatedAt;
}

5
backend/src/main/java/com/imeeting/dto/biz/RealtimeSocketSessionData.java
View File

@ -2,16 +2,11 @@ package com.imeeting.dto.biz;
import lombok.Data;
import java.util.Map;
@Data
public class RealtimeSocketSessionData {
private Long meetingId;
private Long userId;
private Long tenantId;
private Long asrModelId;
private String provider;
private String targetWsUrl;
private String modelCode;
private Map<String, Object> mediaConfig;
}

17
backend/src/main/java/com/imeeting/enums/ModelProviderEnum.java
View File

@ -1,17 +0,0 @@
package com.imeeting.enums;
import lombok.Getter;
@Getter
public enum ModelProviderEnum {
LOCAL("local", "本地"),
TENCENT("tencent", "腾讯云");
private final String code;
private final String description;
ModelProviderEnum(String code, String description) {
this.code = code;
this.description = description;
}
}

2
backend/src/main/java/com/imeeting/service/biz/MeetingCommandService.java
View File

@ -28,6 +28,8 @@ public interface MeetingCommandService {
void deleteMeeting(Long id);
void appendRealtimeTranscripts(Long meetingId, List<RealtimeTranscriptItemDTO> items);
void saveRealtimeTranscriptSnapshot(Long meetingId, RealtimeTranscriptItemDTO item, boolean finalResult);
void completeRealtimeMeeting(Long meetingId, String audioUrl, boolean overwriteAudio);

4
backend/src/main/java/com/imeeting/service/biz/RealtimeMeetingSessionStateService.java
View File

@ -11,8 +11,6 @@ public interface RealtimeMeetingSessionStateService {
void rememberResumeConfig(Long meetingId, RealtimeMeetingResumeConfig resumeConfig);
void rememberSpeakerContext(Long meetingId, String speakerContextId);
void assertCanOpenSession(Long meetingId);
boolean activate(Long meetingId, String connectionId);
@ -27,8 +25,6 @@ public interface RealtimeMeetingSessionStateService {
void refreshAfterTranscript(Long meetingId);
void refreshAfterTranscriptCapture(Long meetingId, long transcriptCount);
boolean markCompletingIfResumeExpired(Long meetingId);
void expireEmptySession(Long meetingId);

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

@ -11,7 +11,6 @@ import com.imeeting.dto.biz.AiLocalProfileVO;
import com.imeeting.dto.biz.AiModelVO;
import com.imeeting.entity.biz.AsrModel;
import com.imeeting.entity.biz.LlmModel;
import com.imeeting.enums.ModelProviderEnum;
import com.imeeting.mapper.biz.AsrModelMapper;
import com.imeeting.mapper.biz.LlmModelMapper;
import com.imeeting.service.biz.AiModelService;
@ -45,10 +44,6 @@ public class AiModelServiceImpl implements AiModelService {
private static final String TYPE_ASR = "ASR";
private static final String TYPE_LLM = "LLM";
private static final String TENCENT_PROVIDER = "tencent";
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 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";
@ -174,7 +169,7 @@ public class AiModelServiceImpl implements AiModelService {
if (resolvedBaseUrl == null || resolvedBaseUrl.isBlank()) {
return Collections.emptyList();
}
if (ModelProviderEnum.LOCAL.getCode().equals(providerKey)) {
if ("custom".equals(providerKey)) {
return fetchLocalProfile(resolvedBaseUrl, apiKey).getAsrModels();
}
String targetUrl = resolveModelListUrl(providerKey, resolvedBaseUrl, apiKey);
@ -469,7 +464,7 @@ public class AiModelServiceImpl implements AiModelService {
}
private String resolveModelListUrl(String providerKey, String baseUrl, String apiKey) {
if (ModelProviderEnum.LOCAL.getCode().equalsIgnoreCase(providerKey)) {
if ("Custom".equalsIgnoreCase(providerKey)) {
return baseUrl+"/api/asrconfig";
}
if ("gemini".equals(providerKey) || "google".equals(providerKey)) {
@ -745,7 +740,6 @@ public class AiModelServiceImpl implements AiModelService {
if (Integer.valueOf(1).equals(dto.getIsDefault()) && !Integer.valueOf(1).equals(dto.getStatus())) {
throw new RuntimeException("默认模型必须为启用状态");
}
validateTencentAsrConfig(dto);
// if ("custom".equals(normalizeProvider(dto.getProvider()))) {
// if (TYPE_ASR.equals(normalizeType(dto.getModelType()))) {
// Map<String, Object> mediaConfig = dto.getMediaConfig() == null ? Collections.emptyMap() : dto.getMediaConfig();
@ -838,20 +832,13 @@ public class AiModelServiceImpl implements AiModelService {
}
private void pushAsrConfig(AsrModel entity) {
String provider = normalizeProvider(entity.getProvider());
if (ModelProviderEnum.LOCAL.getCode().equals(provider)) {
if ("custom".equals(normalizeProvider(entity.getProvider()))) {
if (entity.getApiKey() == null || entity.getApiKey().isBlank()) {
log.info("Skip syncing local ASR profile because apiKey is blank, modelName={}", entity.getModelName());
return;
}
updateLocalProfile(entity);
return;
}
if ("custom".equals(provider)) {
return;
}
if (TENCENT_PROVIDER.equals(provider)) {
return;
}
if (entity.getBaseUrl() == null || entity.getBaseUrl().isBlank()) {
throw new RuntimeException("ASR 模型必须配置 baseUrl");
@ -947,28 +934,6 @@ public class AiModelServiceImpl implements AiModelService {
return text.isEmpty() ? null : text;
}
private void validateTencentAsrConfig(AiModelDTO dto) {
if (!TYPE_ASR.equals(normalizeType(dto.getModelType()))) {
return;
}
if (!TENCENT_PROVIDER.equals(normalizeProvider(dto.getProvider()))) {
return;
}
Map<String, Object> mediaConfig = dto.getMediaConfig() == null ? Collections.emptyMap() : dto.getMediaConfig();
if (readConfigString(mediaConfig.get(MEDIA_TENCENT_APP_ID)) == null) {
throw new RuntimeException("腾讯实时 ASR 模型必须配置 mediaConfig.tencentAppId");
}
if (readConfigString(mediaConfig.get(MEDIA_TENCENT_SECRET_ID)) == null) {
throw new RuntimeException("腾讯实时 ASR 模型必须配置 mediaConfig.tencentSecretId");
}
if (readConfigString(mediaConfig.get(MEDIA_TENCENT_SECRET_KEY)) == null) {
throw new RuntimeException("腾讯实时 ASR 模型必须配置 mediaConfig.tencentSecretKey");
}
if (dto.getModelCode() == null || dto.getModelCode().isBlank()) {
throw new RuntimeException("腾讯实时 ASR 模型必须配置 modelCode");
}
}
private BigDecimal readConfigDecimal(Object value) {
if (value == null) {
return null;

65
backend/src/main/java/com/imeeting/service/biz/impl/MeetingCommandServiceImpl.java
View File

@ -49,7 +49,6 @@ import com.imeeting.support.redis.MeetingAsrPermitCache;
import com.imeeting.support.redis.MeetingLockCache;
import com.unisbase.common.exception.BusinessException;
import com.unisbase.common.exception.ErrorCodeEnum;
import com.imeeting.websocket.RealtimeMeetingProxyWebSocketHandler;
import com.unisbase.service.SysParamService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
@ -83,8 +82,7 @@ public class MeetingCommandServiceImpl implements MeetingCommandService {
private final MeetingDomainSupport meetingDomainSupport;
private final MeetingRuntimeProfileResolver meetingRuntimeProfileResolver;
private final RealtimeMeetingSessionStateService realtimeMeetingSessionStateService;
private final RealtimeMeetingAudioStorageService realtimeMeetingAudioStorageService;
private final RealtimeMeetingProxyWebSocketHandler realtimeMeetingProxyWebSocketHandler;
private final RealtimeMeetingAudioStorageService realtimeMeetingAudioStorageService;
private final MeetingProgressService meetingProgressService;
private final MeetingPointsService meetingPointsService;
private final MeetingSummaryPromptAssembler meetingSummaryPromptAssembler;
@ -111,7 +109,7 @@ public class MeetingCommandServiceImpl implements MeetingCommandService {
MeetingDomainSupport meetingDomainSupport,
MeetingRuntimeProfileResolver meetingRuntimeProfileResolver,
RealtimeMeetingSessionStateService realtimeMeetingSessionStateService,
RealtimeMeetingAudioStorageService realtimeMeetingAudioStorageService, RealtimeMeetingProxyWebSocketHandler realtimeMeetingProxyWebSocketHandler,
RealtimeMeetingAudioStorageService realtimeMeetingAudioStorageService,
MeetingProgressService meetingProgressService,
MeetingPointsService meetingPointsService,
MeetingSummaryPromptAssembler meetingSummaryPromptAssembler,
@ -133,8 +131,7 @@ public class MeetingCommandServiceImpl implements MeetingCommandService {
this.meetingDomainSupport = meetingDomainSupport;
this.meetingRuntimeProfileResolver = meetingRuntimeProfileResolver;
this.realtimeMeetingSessionStateService = realtimeMeetingSessionStateService;
this.realtimeMeetingAudioStorageService = realtimeMeetingAudioStorageService;
this.realtimeMeetingProxyWebSocketHandler = realtimeMeetingProxyWebSocketHandler;
this.realtimeMeetingAudioStorageService = realtimeMeetingAudioStorageService;
this.meetingProgressService = meetingProgressService;
this.meetingPointsService = meetingPointsService;
this.meetingSummaryPromptAssembler = meetingSummaryPromptAssembler;
@ -350,6 +347,58 @@ public class MeetingCommandServiceImpl implements MeetingCommandService {
deleteMeetingArtifactsAfterCommit(id);
}
@Override
@Transactional(rollbackFor = Exception.class)
public void appendRealtimeTranscripts(Long meetingId, List<RealtimeTranscriptItemDTO> items) {
if (items == null || items.isEmpty()) {
return;
}
Integer maxSortOrder = transcriptMapper.selectList(new LambdaQueryWrapper<MeetingTranscript>()
.eq(MeetingTranscript::getMeetingId, meetingId)
.orderByDesc(MeetingTranscript::getSortOrder)
.last("LIMIT 1"))
.stream()
.findFirst()
.map(MeetingTranscript::getSortOrder)
.orElse(0);
int nextSortOrder = maxSortOrder == null ? 0 : maxSortOrder + 1;
boolean inserted = false;
for (RealtimeTranscriptItemDTO item : items) {
if (item.getContent() == null || item.getContent().isBlank()) {
continue;
}
MeetingTranscript existing = transcriptMapper.selectOne(new LambdaQueryWrapper<MeetingTranscript>()
.eq(MeetingTranscript::getMeetingId, meetingId)
.eq(MeetingTranscript::getContent, item.getContent().trim())
.eq(item.getSpeakerId() != null && !item.getSpeakerId().isBlank(), MeetingTranscript::getSpeakerId, item.getSpeakerId())
.eq(item.getStartTime() != null, MeetingTranscript::getStartTime, item.getStartTime())
.eq(item.getEndTime() != null, MeetingTranscript::getEndTime, item.getEndTime())
.last("LIMIT 1"));
if (existing != null) {
continue;
}
MeetingTranscript transcript = new MeetingTranscript();
transcript.setMeetingId(meetingId);
transcript.setSpeakerId(meetingDomainSupport.resolveSpeakerId(item.getSpeakerId()));
transcript.setSpeakerName(meetingDomainSupport.resolveSpeakerName(item.getSpeakerId(), item.getSpeakerName()));
transcript.setContent(item.getContent().trim());
transcript.setStartTime(item.getStartTime());
transcript.setEndTime(item.getEndTime());
transcript.setSortOrder(nextSortOrder++);
transcriptMapper.insert(transcript);
inserted = true;
}
if (inserted) {
meetingTranscriptFileService.initializeTranscriptFileIfAbsent(meetingId);
realtimeMeetingSessionStateService.refreshAfterTranscript(meetingId);
}
}
@Override
@Transactional(rollbackFor = Exception.class)
public void saveRealtimeTranscriptSnapshot(Long meetingId, RealtimeTranscriptItemDTO item, boolean finalResult) {
@ -408,11 +457,11 @@ public class MeetingCommandServiceImpl implements MeetingCommandService {
if (meeting == null) {
throw new RuntimeException("会议不存在");
}
realtimeMeetingProxyWebSocketHandler.closeMeetingSession(meetingId);
RealtimeMeetingSessionStatusVO currentStatus = realtimeMeetingSessionStateService.getStatus(meetingId);
if (overwriteAudio) {
if (audioUrl == null || audioUrl.isBlank()) {
throw new RuntimeException("overwriteAudio=true requires audioUrl");
throw new RuntimeException("overwriteAudio 为 true 时必须提供音频地址");
}
meetingDomainSupport.applyMeetingAudioMetadata(
meeting,

68
backend/src/main/java/com/imeeting/service/biz/impl/RealtimeMeetingSessionStateServiceImpl.java
View File

@ -4,7 +4,6 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.imeeting.dto.biz.RealtimeMeetingResumeConfig;
import com.imeeting.dto.biz.RealtimeMeetingSessionState;
import com.imeeting.dto.biz.RealtimeMeetingSessionStatusVO;
import com.imeeting.dto.biz.RealtimeMeetingTranscriptCacheState;
import com.imeeting.entity.biz.Meeting;
import com.imeeting.entity.biz.MeetingTranscript;
import com.imeeting.enums.MeetingStatusEnum;
@ -13,7 +12,6 @@ import com.imeeting.mapper.biz.MeetingTranscriptMapper;
import com.imeeting.service.biz.RealtimeMeetingSessionStateService;
import com.imeeting.support.redis.MeetingLockCache;
import com.imeeting.support.redis.RealtimeMeetingSessionCache;
import com.imeeting.support.redis.RealtimeMeetingTranscriptCache;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
@ -33,7 +31,6 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
private final MeetingLockCache meetingLockCache;
private final MeetingTranscriptMapper transcriptMapper;
private final MeetingMapper meetingMapper;
private final RealtimeMeetingTranscriptCache realtimeMeetingTranscriptCache;
@Value("${imeeting.realtime.resume-window-minutes:30}")
private String resumeWindowMinutesValue;
@ -52,9 +49,8 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
next.setTenantId(tenantId);
next.setUserId(userId);
next.setStatus("IDLE");
long transcriptCount = countCapturedTranscripts(meetingId);
next.setHasTranscript(transcriptCount > 0);
next.setTranscriptCountSnapshot(transcriptCount);
next.setHasTranscript(countTranscripts(meetingId) > 0);
next.setTranscriptCountSnapshot(countTranscripts(meetingId));
next.setUpdatedAt(System.currentTimeMillis());
writeState(next);
}
@ -63,22 +59,6 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
public void rememberResumeConfig(Long meetingId, RealtimeMeetingResumeConfig resumeConfig) {
RealtimeMeetingSessionState state = getOrCreateState(meetingId);
state.setResumeConfig(resumeConfig);
state.setUpdatedAt(System.currentTimeMillis());
writeState(state);
}
@Override
public void rememberSpeakerContext(Long meetingId, String speakerContextId) {
if (meetingId == null || speakerContextId == null || speakerContextId.isBlank()) {
return;
}
RealtimeMeetingSessionState state = getOrCreateState(meetingId);
RealtimeMeetingResumeConfig resumeConfig = state.getResumeConfig();
if (resumeConfig == null) {
resumeConfig = new RealtimeMeetingResumeConfig();
state.setResumeConfig(resumeConfig);
}
resumeConfig.setSpeakerContextId(speakerContextId.trim());
state.setUpdatedAt(System.currentTimeMillis());
writeState(state);
}
@ -92,16 +72,16 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
String currentStatus = status.getStatus();
if ("COMPLETING".equals(currentStatus)) {
throw new RuntimeException("Realtime meeting is completing");
throw new RuntimeException("实时会议正在结束处理中");
}
if ("COMPLETED".equals(currentStatus)) {
throw new RuntimeException("Realtime meeting is already completed");
throw new RuntimeException("实时会议已结束");
}
if ("ACTIVE".equals(currentStatus) || Boolean.TRUE.equals(status.getActiveConnection())) {
throw new RuntimeException("Realtime meeting already has an active connection");
throw new RuntimeException("实时会议已存在活动连接");
}
if ("PAUSED_RESUMABLE".equals(currentStatus) && !Boolean.TRUE.equals(status.getCanResume())) {
throw new RuntimeException("Realtime meeting resume window has expired");
throw new RuntimeException("实时会议恢复窗口已过期");
}
}
@ -122,7 +102,7 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
}
long now = System.currentTimeMillis();
long transcriptCount = countCapturedTranscripts(meetingId);
long transcriptCount = countTranscripts(meetingId);
state.setStatus("ACTIVE");
state.setHasTranscript(transcriptCount > 0);
state.setTranscriptCountSnapshot(transcriptCount);
@ -202,13 +182,9 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
@Override
public void refreshAfterTranscript(Long meetingId) {
refreshAfterTranscriptCapture(meetingId, countTranscripts(meetingId));
}
@Override
public void refreshAfterTranscriptCapture(Long meetingId, long transcriptCount) {
RealtimeMeetingSessionState state = getOrCreateState(meetingId);
long now = System.currentTimeMillis();
long transcriptCount = countTranscripts(meetingId);
state.setHasTranscript(transcriptCount > 0);
state.setTranscriptCountSnapshot(transcriptCount);
@ -238,7 +214,7 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
return false;
}
long transcriptCount = countCapturedTranscripts(meetingId);
long transcriptCount = countTranscripts(meetingId);
if (transcriptCount <= 0) {
clear(meetingId);
return false;
@ -280,7 +256,7 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
}
private RealtimeMeetingSessionStatusVO pauseState(Long meetingId, RealtimeMeetingSessionState state) {
long transcriptCount = countCapturedTranscripts(meetingId);
long transcriptCount = countTranscripts(meetingId);
long now = System.currentTimeMillis();
state.setHasTranscript(transcriptCount > 0);
@ -331,7 +307,7 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
} else {
vo.setStatus("IDLE");
}
vo.setHasTranscript(countCapturedTranscripts(meetingId) > 0);
vo.setHasTranscript(countTranscripts(meetingId) > 0);
vo.setCanResume(false);
vo.setRemainingSeconds(0L);
vo.setActiveConnection(false);
@ -386,9 +362,8 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
RealtimeMeetingSessionState next = new RealtimeMeetingSessionState();
next.setMeetingId(meetingId);
next.setStatus("IDLE");
long transcriptCount = countCapturedTranscripts(meetingId);
next.setHasTranscript(transcriptCount > 0);
next.setTranscriptCountSnapshot(transcriptCount);
next.setHasTranscript(countTranscripts(meetingId) > 0);
next.setTranscriptCountSnapshot(countTranscripts(meetingId));
next.setUpdatedAt(System.currentTimeMillis());
return next;
}
@ -425,21 +400,4 @@ public class RealtimeMeetingSessionStateServiceImpl implements RealtimeMeetingSe
return transcriptMapper.selectCount(new LambdaQueryWrapper<MeetingTranscript>()
.eq(MeetingTranscript::getMeetingId, meetingId));
}
private long countCapturedTranscripts(Long meetingId) {
return Math.max(countTranscripts(meetingId), countCachedTranscripts(meetingId));
}
private long countCachedTranscripts(Long meetingId) {
if (meetingId == null) {
return 0L;
}
RealtimeMeetingTranscriptCacheState state = realtimeMeetingTranscriptCache.getState(meetingId);
if (state == null || state.getItems() == null || state.getItems().isEmpty()) {
return 0L;
}
return state.getItems().stream()
.filter(item -> item.getContent() != null && !item.getContent().isBlank())
.count();
}
}

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

@ -10,13 +10,12 @@ import com.imeeting.service.biz.AiModelService;
import com.imeeting.service.biz.MeetingAccessService;
import com.imeeting.service.biz.RealtimeMeetingSessionStateService;
import com.imeeting.service.biz.RealtimeMeetingSocketSessionService;
import com.imeeting.service.realtime.RealtimeAsrChannel;
import com.imeeting.service.realtime.RealtimeAsrChannelFactory;
import com.imeeting.support.redis.RealtimeMeetingSocketSessionCache;
import com.unisbase.security.LoginUser;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@ -31,7 +30,6 @@ public class RealtimeMeetingSocketSessionServiceImpl implements RealtimeMeetingS
private final MeetingAccessService meetingAccessService;
private final AiModelService aiModelService;
private final RealtimeMeetingSessionStateService realtimeMeetingSessionStateService;
private final RealtimeAsrChannelFactory realtimeAsrChannelFactory;
@Override
public RealtimeSocketSessionVO createSession(Long meetingId, Long asrModelId, String mode, String language,
@ -39,10 +37,10 @@ public class RealtimeMeetingSocketSessionServiceImpl implements RealtimeMeetingS
Boolean enableTextRefine, Boolean saveAudio,
List<Map<String, Object>> hotwords, LoginUser loginUser) {
if (meetingId == null) {
throw new RuntimeException("会议 ID 不能为空");
throw new RuntimeException("会议 ID 不能为空");
}
if (asrModelId == null) {
throw new RuntimeException("ASR 模型 ID 不能为空");
throw new RuntimeException("ASR 模型 ID 不能为空");
}
Meeting meeting = meetingAccessService.requireMeeting(meetingId);
@ -53,13 +51,12 @@ public class RealtimeMeetingSocketSessionServiceImpl implements RealtimeMeetingS
AiModelVO asrModel = aiModelService.getModelById(asrModelId, "ASR");
if (asrModel == null) {
throw new RuntimeException("ASR 模型不存在");
throw new RuntimeException("ASR 模型不存在");
}
RealtimeAsrChannel realtimeAsrChannel = realtimeAsrChannelFactory.getRequired(asrModel.getProvider());
String targetWsUrl = realtimeAsrChannel.resolveTargetWsUrl(asrModel);
String targetWsUrl = resolveWsUrl(asrModel);
if (targetWsUrl == null || targetWsUrl.isBlank()) {
throw new RuntimeException("ASR 模型未配置 WebSocket 地址");
throw new RuntimeException("ASR 模型未配置 WebSocket 地址");
}
RealtimeMeetingResumeConfig resumeConfig = new RealtimeMeetingResumeConfig();
@ -85,10 +82,7 @@ public class RealtimeMeetingSocketSessionServiceImpl implements RealtimeMeetingS
sessionData.setUserId(loginUser.getUserId());
sessionData.setTenantId(loginUser.getTenantId());
sessionData.setAsrModelId(asrModelId);
sessionData.setProvider(realtimeAsrChannelFactory.normalizeProvider(asrModel.getProvider()));
sessionData.setTargetWsUrl(targetWsUrl);
sessionData.setModelCode(asrModel.getModelCode());
sessionData.setMediaConfig(asrModel.getMediaConfig());
String sessionToken = UUID.randomUUID().toString().replace("-", "");
socketSessionCache.save(sessionToken, sessionData);
@ -97,8 +91,9 @@ public class RealtimeMeetingSocketSessionServiceImpl implements RealtimeMeetingS
vo.setSessionToken(sessionToken);
vo.setPath(WS_PATH);
vo.setExpiresInSeconds(socketSessionCache.getSessionTtlSeconds());
vo.setStartMessage(realtimeAsrChannel.buildStartMessage(
vo.setStartMessage(buildStartMessage(
asrModel,
meetingId,
mode,
language,
useSpkId,
@ -115,4 +110,74 @@ public class RealtimeMeetingSocketSessionServiceImpl implements RealtimeMeetingS
public RealtimeSocketSessionData getSessionData(String sessionToken) {
return socketSessionCache.get(sessionToken);
}
private String resolveWsUrl(AiModelVO model) {
if (model.getWsUrl() != null && !model.getWsUrl().isBlank()) {
return model.getWsUrl();
}
if (model.getBaseUrl() == null || model.getBaseUrl().isBlank()) {
return "";
}
return model.getBaseUrl()
.replaceFirst("^http://", "ws://")
.replaceFirst("^https://", "wss://");
}
private Map<String, Object> buildStartMessage(AiModelVO model, Long meetingId, String mode, String language,
Integer useSpkId, Boolean enablePunctuation, Boolean enableItn,
Boolean enableTextRefine, Boolean saveAudio,
List<Map<String, Object>> hotwords) {
Map<String, Object> root = new HashMap<>();
root.put("type", "start");
root.put("request_id", "web_" + System.currentTimeMillis() + "_" + meetingId);
root.put("authorization", buildAuthorization(model.getApiKey()));
Map<String, Object> config = new HashMap<>();
Map<String, Object> audio = new HashMap<>();
audio.put("format", "pcm");
audio.put("sample_rate", 16000);
audio.put("channels", 1);
config.put("audio", audio);
Map<String, Object> recognition = new HashMap<>();
recognition.put("language", normalizeLanguage(language));
recognition.put("enable_punctuation", boolOrDefault(enablePunctuation, true));
recognition.put("enable_itn", boolOrDefault(enableItn, true));
recognition.put("enable_speaker", Integer.valueOf(1).equals(useSpkId));
recognition.put("enable_two_pass", !"online".equalsIgnoreCase(mode));
recognition.put("enable_text_refine", boolOrDefault(enableTextRefine, false));
recognition.put("speaker_threshold", readSpeakerThreshold(model.getMediaConfig()));
recognition.put("hotwords", hotwords == null ? List.of() : hotwords);
config.put("recognition", recognition);
config.put("model", model.getModelCode());
config.put("save_audio", boolOrDefault(saveAudio, false));
root.put("config", config);
return root;
}
private String buildAuthorization(String apiKey) {
if (apiKey == null || apiKey.isBlank()) {
return "";
}
return apiKey.startsWith("Bearer ") ? apiKey : "Bearer " + apiKey;
}
private Object readSpeakerThreshold(Map<String, Object> mediaConfig) {
if (mediaConfig == null) {
return null;
}
return mediaConfig.get("svThreshold");
}
private String normalizeLanguage(String language) {
if (language == null || language.isBlank()) {
return "auto";
}
return language.trim();
}
private boolean boolOrDefault(Boolean value, boolean defaultValue) {
return value != null ? value : defaultValue;
}
}

36
backend/src/main/java/com/imeeting/service/realtime/RealtimeAsrChannel.java
View File

@ -1,36 +0,0 @@
package com.imeeting.service.realtime;
import com.imeeting.dto.biz.AiModelVO;
import java.util.List;
import java.util.Map;
public interface RealtimeAsrChannel {
boolean supports(String provider);
String resolveTargetWsUrl(AiModelVO model);
Map<String, Object> buildStartMessage(AiModelVO model,
String mode,
String language,
Integer useSpkId,
Boolean enablePunctuation,
Boolean enableItn,
Boolean enableTextRefine,
Boolean saveAudio,
List<Map<String, Object>> hotwords);
void connect(RealtimeAsrChannelContext context) throws Exception;
void handleFrontendText(RealtimeAsrChannelContext context, String payload);
void handleFrontendBinary(RealtimeAsrChannelContext context, byte[] payload);
default void onFrontendDetached(RealtimeAsrChannelContext context) {
// default no-op
}
void closeMeeting(RealtimeAsrChannelContext context);
boolean isOpen(RealtimeAsrChannelContext context);
}

17
backend/src/main/java/com/imeeting/service/realtime/RealtimeAsrChannelCallback.java
View File

@ -1,17 +0,0 @@
package com.imeeting.service.realtime;
import org.springframework.web.socket.CloseStatus;
public interface RealtimeAsrChannelCallback {
void onChannelOpen(Long meetingId) throws Exception;
void sendFrontendText(Long meetingId, String payload) throws Exception;
void sendFrontendBinary(Long meetingId, byte[] payload) throws Exception;
void sendFrontendError(Long meetingId, String code, String message);
void removeMeetingSession(Long meetingId);
void closeFrontend(Long meetingId, CloseStatus status);
}

26
backend/src/main/java/com/imeeting/service/realtime/RealtimeAsrChannelContext.java
View File

@ -1,26 +0,0 @@
package com.imeeting.service.realtime;
import lombok.Data;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
@Data
public class RealtimeAsrChannelContext {
private Long meetingId;
private String provider;
private String targetWsUrl;
private WebSocketSession rawSession;
private ConcurrentWebSocketSessionDecorator frontendSession;
private RealtimeAsrChannelCallback callback;
private final ConcurrentMap<String, Object> channelState = new ConcurrentHashMap<>();
private volatile ConcurrentMap<String, Object> frontendState = new ConcurrentHashMap<>();
public void bindFrontendSession(WebSocketSession rawSession, ConcurrentWebSocketSessionDecorator frontendSession) {
this.rawSession = rawSession;
this.frontendSession = frontendSession;
this.frontendState = new ConcurrentHashMap<>();
}
}

29
backend/src/main/java/com/imeeting/service/realtime/RealtimeAsrChannelFactory.java
View File

@ -1,29 +0,0 @@
package com.imeeting.service.realtime;
import com.imeeting.enums.ModelProviderEnum;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;
import java.util.List;
@Component
@RequiredArgsConstructor
public class RealtimeAsrChannelFactory {
private final List<RealtimeAsrChannel> channels;
public RealtimeAsrChannel getRequired(String provider) {
String normalizedProvider = normalizeProvider(provider);
return channels.stream()
.filter(channel -> channel.supports(normalizedProvider))
.findFirst()
.orElseThrow(() -> new RuntimeException("暂不支持的实时 ASR 渠道: " + provider));
}
public String normalizeProvider(String provider) {
if (provider == null || provider.isBlank()) {
return ModelProviderEnum.LOCAL.getCode();
}
return provider.trim().toLowerCase();
}
}

13
backend/src/main/java/com/imeeting/service/realtime/RealtimeMeetingTranscriptCacheService.java
View File

@ -1,13 +0,0 @@
package com.imeeting.service.realtime;
import com.imeeting.dto.biz.RealtimeMeetingTranscriptCacheItem;
import java.util.List;
public interface RealtimeMeetingTranscriptCacheService {
void mergeUpstreamMessage(Long meetingId, String payload);
List<RealtimeMeetingTranscriptCacheItem> listOrderedItems(Long meetingId);
void clear(Long meetingId);
}

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

@ -1,538 +0,0 @@
package com.imeeting.service.realtime.impl;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.imeeting.dto.biz.AiModelVO;
import com.imeeting.dto.biz.RealtimeMeetingTranscriptCacheItem;
import com.imeeting.enums.ModelProviderEnum;
import com.imeeting.service.biz.RealtimeMeetingSessionStateService;
import com.imeeting.service.realtime.RealtimeAsrChannel;
import com.imeeting.service.realtime.RealtimeAsrChannelContext;
import com.imeeting.service.realtime.RealtimeMeetingTranscriptCacheService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import java.io.ByteArrayOutputStream;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
@Slf4j
@Component
@RequiredArgsConstructor
public class LocalRealtimeAsrChannel implements RealtimeAsrChannel {
private static final String STATE_UPSTREAM_SOCKET = "upstreamSocket";
private static final String STATE_CLOSE_AFTER_END = "closeAfterEnd";
private static final String STATE_START_MESSAGE_FORWARDED = "startMessageForwarded";
private static final String STATE_UPSTREAM_SEND_CHAIN = "upstreamSendChain";
private static final String STATE_START_MESSAGE_SENT = "startMessageSent";
private static final String STATE_PENDING_AUDIO_FRAMES = "pendingAudioFrames";
private static final CompletableFuture<Void> COMPLETED = CompletableFuture.completedFuture(null);
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
private static final String STOP_MESSAGE = "{\"type\":\"stop\"}";
private final RealtimeMeetingSessionStateService realtimeMeetingSessionStateService;
private final RealtimeMeetingTranscriptCacheService realtimeMeetingTranscriptCacheService;
@Override
public boolean supports(String provider) {
return ModelProviderEnum.LOCAL.getCode().equalsIgnoreCase(provider);
}
@Override
public String resolveTargetWsUrl(AiModelVO model) {
if (model.getWsUrl() != null && !model.getWsUrl().isBlank()) {
return model.getWsUrl();
}
if (model.getBaseUrl() == null || model.getBaseUrl().isBlank()) {
return "";
}
return model.getBaseUrl()
.replaceFirst("^http://", "ws://")
.replaceFirst("^https://", "wss://");
}
@Override
public Map<String, Object> buildStartMessage(AiModelVO model,
String mode,
String language,
Integer useSpkId,
Boolean enablePunctuation,
Boolean enableItn,
Boolean enableTextRefine,
Boolean saveAudio,
List<Map<String, Object>> hotwords) {
Map<String, Object> root = new HashMap<>();
root.put("type", "start");
Map<String, Object> payload = new HashMap<>();
payload.put("format", "pcm");
payload.put("sample_rate", 16000);
payload.put("language", normalizeProtocolLanguage(language));
payload.put("context", "");
payload.put("enable_inverse_text_normalization", boolOrDefault(enableItn, true));
payload.put("unfixed_token_num", 3);
payload.put("silence_duration_ms", 800);
payload.put("min_partial_sec", 0.3D);
payload.put("pre_roll_ms", 240);
payload.put("max_sentence_count", 8);
payload.put("partial_holdback_chars", 2);
payload.put("enable_native_partial_stream", false);
payload.put("enable_speaker", Integer.valueOf(1).equals(useSpkId));
payload.put("match_speaker_registry", false);
payload.put("speaker_threshold", readSpeakerThreshold(model.getMediaConfig()));
payload.put("enable_realtime_longform", false);
payload.put("enable_realtime_vad_split", false);
payload.put("force_stable_segment_sec", 6);
payload.put("force_stable_min_chars", 24);
payload.put("max_segment_sec", 12);
payload.put("hotwords", hotwords == null ? List.of() : hotwords);
root.put("payload", payload);
return root;
}
@Override
public void connect(RealtimeAsrChannelContext context) throws Exception {
initializeFrontendState(context);
java.net.http.WebSocket upstreamSocket = java.net.http.HttpClient.newHttpClient()
.newWebSocketBuilder()
.buildAsync(URI.create(context.getTargetWsUrl()), new UpstreamListener(context))
.get();
context.getChannelState().put(STATE_UPSTREAM_SOCKET, upstreamSocket);
}
@Override
public void handleFrontendText(RealtimeAsrChannelContext context, String payload) {
java.net.http.WebSocket upstreamSocket = getUpstreamSocket(context);
if (upstreamSocket == null) {
return;
}
initializeFrontendState(context);
if (looksLikeStartMessage(payload)) {
context.getFrontendState().put(STATE_START_MESSAGE_SENT, Boolean.TRUE);
if (!Boolean.TRUE.equals(context.getChannelState().get(STATE_START_MESSAGE_FORWARDED))) {
context.getChannelState().put(STATE_START_MESSAGE_FORWARDED, Boolean.TRUE);
sendUpstreamOrdered(context, () -> upstreamSocket.sendText(payload, true), "text-start");
}
flushPendingAudioFrames(context, upstreamSocket);
return;
}
if (looksLikeStopMessage(payload)) {
context.getChannelState().put(STATE_CLOSE_AFTER_END, Boolean.TRUE);
}
sendUpstreamOrdered(context, () -> upstreamSocket.sendText(payload, true), "text");
}
@Override
public void handleFrontendBinary(RealtimeAsrChannelContext context, byte[] payload) {
java.net.http.WebSocket upstreamSocket = getUpstreamSocket(context);
if (upstreamSocket == null) {
return;
}
initializeFrontendState(context);
if (!Boolean.TRUE.equals(context.getFrontendState().get(STATE_START_MESSAGE_SENT))) {
queuePendingAudioFrame(context, payload);
return;
}
sendUpstreamOrdered(context, () -> upstreamSocket.sendBinary(ByteBuffer.wrap(payload), true), "binary");
}
@Override
public void closeMeeting(RealtimeAsrChannelContext context) {
if (context == null) {
return;
}
java.net.http.WebSocket upstreamSocket = getUpstreamSocket(context);
if (upstreamSocket == null) {
return;
}
context.getChannelState().put(STATE_CLOSE_AFTER_END, Boolean.TRUE);
upstreamSocket.sendText(STOP_MESSAGE, true);
}
@Override
public boolean isOpen(RealtimeAsrChannelContext context) {
return getUpstreamSocket(context) != null;
}
public static String buildFrontendTranscriptMessage(RealtimeMeetingTranscriptCacheItem item) throws JsonProcessingException {
ObjectNode root = OBJECT_MAPPER.createObjectNode();
boolean isFinal = Boolean.TRUE.equals(item.getFinalResult());
root.put("type", isFinal ? "segment" : "partial");
ObjectNode data = root.putObject("data");
data.put("text", item.getContent());
data.put("is_final", isFinal);
if (item.getSentenceId() != null) {
data.put("sentence_id", item.getSentenceId());
}
if (item.getStartTime() != null) {
data.put("start", item.getStartTime() / 1000D);
}
if (item.getEndTime() != null) {
data.put("end", item.getEndTime() / 1000D);
}
if (item.getSpeakerId() != null && !item.getSpeakerId().isBlank()) {
data.put("speaker_id", item.getSpeakerId());
}
if (item.getSpeakerName() != null && !item.getSpeakerName().isBlank()) {
data.put("speaker_name", item.getSpeakerName());
}
if (item.getUserId() != null && !item.getUserId().isBlank()) {
data.put("user_id", item.getUserId());
}
return OBJECT_MAPPER.writeValueAsString(root);
}
private Object readSpeakerThreshold(Map<String, Object> mediaConfig) {
if (mediaConfig == null) {
return null;
}
return mediaConfig.get("svThreshold");
}
private String normalizeProtocolLanguage(String language) {
if (language == null || language.isBlank()) {
return null;
}
String normalized = language.trim();
if ("auto".equalsIgnoreCase(normalized)) {
return null;
}
return normalized;
}
private boolean boolOrDefault(Boolean value, boolean defaultValue) {
return value != null ? value : defaultValue;
}
private void initializeFrontendState(RealtimeAsrChannelContext context) {
context.getFrontendState().putIfAbsent(STATE_UPSTREAM_SEND_CHAIN, COMPLETED);
context.getFrontendState().putIfAbsent(STATE_START_MESSAGE_SENT, Boolean.FALSE);
context.getFrontendState().putIfAbsent(STATE_PENDING_AUDIO_FRAMES, new ArrayList<byte[]>());
}
private java.net.http.WebSocket getUpstreamSocket(RealtimeAsrChannelContext context) {
Object value = context.getChannelState().get(STATE_UPSTREAM_SOCKET);
return value instanceof java.net.http.WebSocket socket ? socket : null;
}
@SuppressWarnings("unchecked")
private void sendUpstreamOrdered(RealtimeAsrChannelContext context,
Supplier<CompletableFuture<?>> sendAction,
String messageType) {
ConcurrentMap<String, Object> frontendState = context.getFrontendState();
synchronized (frontendState) {
CompletableFuture<Void> chain = (CompletableFuture<Void>) frontendState.getOrDefault(STATE_UPSTREAM_SEND_CHAIN, COMPLETED);
CompletableFuture<Void> nextChain = chain
.exceptionally(ex -> null)
.thenCompose(ignored -> sendAction.get().thenApply(ignoredResult -> null));
nextChain = nextChain.whenComplete((ignored, ex) -> {
if (ex != null) {
log.error("顺序发送上游消息失败：meetingId={}, sessionId={}, type={}",
context.getMeetingId(), currentConnectionId(context), messageType, ex);
}
});
frontendState.put(STATE_UPSTREAM_SEND_CHAIN, nextChain);
}
}
@SuppressWarnings("unchecked")
private void queuePendingAudioFrame(RealtimeAsrChannelContext context, byte[] payload) {
ConcurrentMap<String, Object> frontendState = context.getFrontendState();
synchronized (frontendState) {
List<byte[]> pendingFrames = (List<byte[]>) frontendState.get(STATE_PENDING_AUDIO_FRAMES);
if (pendingFrames == null) {
pendingFrames = new ArrayList<>();
frontendState.put(STATE_PENDING_AUDIO_FRAMES, pendingFrames);
}
pendingFrames.add(payload);
}
}
@SuppressWarnings("unchecked")
private void flushPendingAudioFrames(RealtimeAsrChannelContext context, java.net.http.WebSocket upstreamSocket) {
List<byte[]> pendingFrames;
ConcurrentMap<String, Object> frontendState = context.getFrontendState();
synchronized (frontendState) {
pendingFrames = (List<byte[]>) frontendState.get(STATE_PENDING_AUDIO_FRAMES);
if (pendingFrames == null || pendingFrames.isEmpty()) {
return;
}
frontendState.put(STATE_PENDING_AUDIO_FRAMES, new ArrayList<byte[]>());
}
log.info("start 后开始补发排队音频帧：meetingId={}, sessionId={}, frameCount={}",
context.getMeetingId(), currentConnectionId(context), pendingFrames.size());
for (byte[] frame : pendingFrames) {
sendUpstreamOrdered(context, () -> upstreamSocket.sendBinary(ByteBuffer.wrap(frame), true), "binary-flush");
}
}
private String currentConnectionId(RealtimeAsrChannelContext context) {
return context.getRawSession() == null ? null : context.getRawSession().getId();
}
private static ByteBuffer copyBuffer(ByteBuffer source) {
ByteBuffer duplicate = source.asReadOnlyBuffer();
byte[] bytes = new byte[duplicate.remaining()];
duplicate.get(bytes);
return ByteBuffer.wrap(bytes);
}
private static boolean shouldLogBinaryFrame(int count) {
return count <= 3 || count % 25 == 0;
}
private static String summarizeText(String payload) {
if (payload == null) {
return "";
}
String normalized = payload.replaceAll("\\s+", " ").trim();
if (normalized.length() <= 240) {
return normalized;
}
return normalized.substring(0, 240) + "...";
}
private static boolean looksLikeStartMessage(String payload) {
if (payload == null || payload.isBlank()) {
return false;
}
String normalized = payload.replaceAll("\\s+", "");
return normalized.contains("\"type\":\"start\"");
}
private static boolean looksLikeStopMessage(String payload) {
if (payload == null || payload.isBlank()) {
return false;
}
String normalized = payload.replaceAll("\\s+", "");
return normalized.contains("\"type\":\"stop\"");
}
private static boolean looksLikeEndMessage(String payload) {
if (payload == null || payload.isBlank()) {
return false;
}
try {
JsonNode root = OBJECT_MAPPER.readTree(payload);
return "end".equals(root.path("type").asText(""));
} catch (Exception ex) {
return false;
}
}
private static List<String> normalizeFrontendMessages(String upstreamPayload) {
try {
JsonNode root = OBJECT_MAPPER.readTree(upstreamPayload);
String type = root.path("type").asText("");
if (!"sentences".equals(type) && !"end".equals(type)) {
return List.of(upstreamPayload);
}
List<String> normalizedMessages = new ArrayList<>();
JsonNode sentences = root.path("sentences");
if (sentences.isArray()) {
for (JsonNode sentence : sentences) {
String text = sentence.path("sentence").asText("").trim();
if (text.isEmpty()) {
continue;
}
boolean isFinal = sentence.path("sentence_type").asInt(0) != 0 || "end".equals(type);
normalizedMessages.add(buildFrontendTranscriptMessage(sentence, text, isFinal));
}
}
if (normalizedMessages.isEmpty()) {
String fallbackText = root.path("result").path("voice_text_str").asText("").trim();
if (!fallbackText.isEmpty()) {
normalizedMessages.add(buildFrontendTranscriptMessage((JsonNode) null, fallbackText, true));
}
}
return normalizedMessages.isEmpty() ? List.of(upstreamPayload) : normalizedMessages;
} catch (Exception ex) {
return List.of(upstreamPayload);
}
}
private static String buildFrontendTranscriptMessage(JsonNode sentence, String text, boolean isFinal) throws JsonProcessingException {
ObjectNode root = OBJECT_MAPPER.createObjectNode();
root.put("type", isFinal ? "segment" : "partial");
ObjectNode data = root.putObject("data");
data.put("text", text);
data.put("is_final", isFinal);
if (sentence != null && sentence.has("sentence_id") && sentence.get("sentence_id").canConvertToInt()) {
data.put("sentence_id", sentence.get("sentence_id").asInt());
}
if (sentence != null) {
copyOptionalTimeSeconds(sentence, data, "start_time", "start");
copyOptionalTimeSeconds(sentence, data, "end_time", "end");
copyOptionalText(sentence, data, "speaker_id");
copyOptionalText(sentence, data, "speaker_name");
copyOptionalText(sentence, data, "user_id");
}
return OBJECT_MAPPER.writeValueAsString(root);
}
private static void copyOptionalTimeSeconds(JsonNode source, ObjectNode target, String sourceFieldName, String targetFieldName) {
if (source == null) {
return;
}
if (source.has(sourceFieldName) && source.get(sourceFieldName).isNumber()) {
target.put(targetFieldName, source.get(sourceFieldName).asDouble() / 1000D);
return;
}
if (source.has(targetFieldName) && source.get(targetFieldName).isNumber()) {
target.put(targetFieldName, source.get(targetFieldName).asDouble());
}
}
private static void copyOptionalText(JsonNode source, ObjectNode target, String fieldName) {
if (source == null || !source.has(fieldName) || source.get(fieldName).isNull()) {
return;
}
String value = source.get(fieldName).asText("").trim();
if (!value.isEmpty()) {
target.put(fieldName, value);
}
}
private final class UpstreamListener implements java.net.http.WebSocket.Listener {
private final RealtimeAsrChannelContext context;
private final StringBuilder textBuffer = new StringBuilder();
private final ByteArrayOutputStream binaryBuffer = new ByteArrayOutputStream();
private final AtomicInteger upstreamTextCount = new AtomicInteger();
private final AtomicInteger upstreamBinaryCount = new AtomicInteger();
private UpstreamListener(RealtimeAsrChannelContext context) {
this.context = context;
}
@Override
public void onOpen(java.net.http.WebSocket webSocket) {
context.getChannelState().put(STATE_UPSTREAM_SOCKET, webSocket);
log.info("上游 ASR websocket 已打开：meetingId={}, sessionId={}, upstream={}",
context.getMeetingId(), currentConnectionId(context), context.getTargetWsUrl());
String connectionId = currentConnectionId(context);
if (connectionId == null || !realtimeMeetingSessionStateService.activate(context.getMeetingId(), connectionId)) {
context.getCallback().sendFrontendError(context.getMeetingId(), "REALTIME_ACTIVE_CONNECTION_EXISTS", "当前会议无法激活这条前端连接");
webSocket.sendClose(CloseStatus.POLICY_VIOLATION.getCode(), "当前会议无法激活这条前端连接");
context.getCallback().closeFrontend(context.getMeetingId(), CloseStatus.POLICY_VIOLATION.withReason("当前会议无法激活这条前端连接"));
return;
}
try {
context.getCallback().onChannelOpen(context.getMeetingId());
} catch (Exception ex) {
log.error("通知前端上游就绪失败：meetingId={}, sessionId={}", context.getMeetingId(), currentConnectionId(context), ex);
context.getCallback().closeFrontend(context.getMeetingId(), CloseStatus.SERVER_ERROR);
return;
}
webSocket.request(1);
}
@Override
public java.util.concurrent.CompletionStage<?> onText(java.net.http.WebSocket webSocket, CharSequence data, boolean last) {
textBuffer.append(data);
if (last) {
int count = upstreamTextCount.incrementAndGet();
String upstreamPayload = textBuffer.toString();
realtimeMeetingTranscriptCacheService.mergeUpstreamMessage(context.getMeetingId(), upstreamPayload);
try {
for (String frontendPayload : normalizeFrontendMessages(upstreamPayload)) {
context.getCallback().sendFrontendText(context.getMeetingId(), frontendPayload);
}
log.info("上游 ASR 文本 -> 前端：meetingId={}, sessionId={}, count={}, payload={}",
context.getMeetingId(), currentConnectionId(context), count, summarizeText(upstreamPayload));
if (Boolean.TRUE.equals(context.getChannelState().get(STATE_CLOSE_AFTER_END))
&& looksLikeEndMessage(upstreamPayload)) {
webSocket.sendClose(CloseStatus.NORMAL.getCode(), "meeting-complete");
}
} catch (Exception ex) {
log.error("转发上游 ASR 文本失败：meetingId={}, sessionId={}", context.getMeetingId(), currentConnectionId(context), ex);
context.getCallback().closeFrontend(context.getMeetingId(), CloseStatus.SERVER_ERROR);
} finally {
textBuffer.setLength(0);
}
}
webSocket.request(1);
return COMPLETED;
}
@Override
public java.util.concurrent.CompletionStage<?> onBinary(java.net.http.WebSocket webSocket, ByteBuffer data, boolean last) {
byte[] chunk = new byte[data.remaining()];
data.get(chunk);
binaryBuffer.writeBytes(chunk);
if (last) {
int count = upstreamBinaryCount.incrementAndGet();
try {
context.getCallback().sendFrontendBinary(context.getMeetingId(), binaryBuffer.toByteArray());
if (shouldLogBinaryFrame(count)) {
log.info("上游 ASR 二进制消息 -> 前端：meetingId={}, sessionId={}, count={}, bytes={}",
context.getMeetingId(), currentConnectionId(context), count, binaryBuffer.size());
}
} catch (Exception ex) {
log.error("转发上游 ASR 二进制消息失败：meetingId={}, sessionId={}", context.getMeetingId(), currentConnectionId(context), ex);
context.getCallback().closeFrontend(context.getMeetingId(), CloseStatus.SERVER_ERROR);
} finally {
binaryBuffer.reset();
}
}
webSocket.request(1);
return COMPLETED;
}
@Override
public java.util.concurrent.CompletionStage<?> onPing(java.net.http.WebSocket webSocket, ByteBuffer message) {
webSocket.sendPong(copyBuffer(message));
log.info("上游 ASR ping 已本地响应：meetingId={}, sessionId={}, bytes={}",
context.getMeetingId(), currentConnectionId(context), message.remaining());
webSocket.request(1);
return COMPLETED;
}
@Override
public java.util.concurrent.CompletionStage<?> onPong(java.net.http.WebSocket webSocket, ByteBuffer message) {
log.debug("上游 ASR pong 已本地忽略：meetingId={}, sessionId={}, bytes={}",
context.getMeetingId(), currentConnectionId(context), message.remaining());
webSocket.request(1);
return COMPLETED;
}
@Override
public java.util.concurrent.CompletionStage<?> onClose(java.net.http.WebSocket webSocket, int statusCode, String reason) {
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);
context.getCallback().closeFrontend(context.getMeetingId(), new CloseStatus(statusCode, reason));
return COMPLETED;
}
@Override
public void onError(java.net.http.WebSocket webSocket, Throwable error) {
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);
}
}
}

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

@ -1,291 +0,0 @@
package com.imeeting.service.realtime.impl;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.imeeting.dto.biz.RealtimeMeetingTranscriptCacheItem;
import com.imeeting.dto.biz.RealtimeMeetingTranscriptCacheState;
import com.imeeting.entity.biz.MeetingTranscript;
import com.imeeting.mapper.biz.MeetingTranscriptMapper;
import com.imeeting.service.biz.MeetingTranscriptFileService;
import com.imeeting.service.biz.RealtimeMeetingSessionStateService;
import com.imeeting.service.realtime.RealtimeMeetingTranscriptCacheService;
import com.imeeting.support.redis.RealtimeMeetingTranscriptCache;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
@Service
public class RealtimeMeetingTranscriptCacheServiceImpl implements RealtimeMeetingTranscriptCacheService {
private final RealtimeMeetingTranscriptCache transcriptCache;
private final RealtimeMeetingSessionStateService realtimeMeetingSessionStateService;
private final MeetingTranscriptMapper transcriptMapper;
private final MeetingTranscriptFileService meetingTranscriptFileService;
private final ObjectMapper objectMapper = new ObjectMapper();
private final Map<Long, Object> meetingLocks = new ConcurrentHashMap<>();
@Autowired
public RealtimeMeetingTranscriptCacheServiceImpl(RealtimeMeetingTranscriptCache transcriptCache,
RealtimeMeetingSessionStateService realtimeMeetingSessionStateService,
MeetingTranscriptMapper transcriptMapper,
MeetingTranscriptFileService meetingTranscriptFileService) {
this.transcriptCache = transcriptCache;
this.realtimeMeetingSessionStateService = realtimeMeetingSessionStateService;
this.transcriptMapper = transcriptMapper;
this.meetingTranscriptFileService = meetingTranscriptFileService;
}
@Override
public void mergeUpstreamMessage(Long meetingId, String payload) {
if (meetingId == null || payload == null || payload.isBlank()) {
return;
}
synchronized (lockForMeeting(meetingId)) {
RealtimeMeetingTranscriptCacheState state = getOrCreateState(meetingId);
try {
JsonNode root = objectMapper.readTree(payload);
String type = root.path("type").asText("");
if (!"sentences".equals(type) && !"end".equals(type)) {
return;
}
JsonNode sentences = root.path("sentences");
if (!sentences.isArray()) {
return;
}
for (JsonNode sentence : sentences) {
RealtimeMeetingTranscriptCacheItem item = mergeSentenceNode(state, sentence, "end".equals(type));
persistFinalSentence(meetingId, item);
}
state.setUpdatedAt(System.currentTimeMillis());
transcriptCache.saveState(state);
realtimeMeetingSessionStateService.refreshAfterTranscriptCapture(meetingId, countNonEmptyItems(state));
} catch (Exception ignored) {
// ignore malformed upstream payload
}
}
}
@Override
public List<RealtimeMeetingTranscriptCacheItem> listOrderedItems(Long meetingId) {
RealtimeMeetingTranscriptCacheState state = transcriptCache.getState(meetingId);
if (state == null || state.getItems() == null || state.getItems().isEmpty()) {
return List.of();
}
return state.getItems().stream()
.filter(item -> item.getContent() != null && !item.getContent().isBlank())
.sorted(Comparator.comparing(item -> item.getSortOrder() == null ? Integer.MAX_VALUE : item.getSortOrder()))
.toList();
}
@Override
public void clear(Long meetingId) {
transcriptCache.clear(meetingId);
}
private RealtimeMeetingTranscriptCacheItem mergeSentenceNode(RealtimeMeetingTranscriptCacheState state,
JsonNode sentence,
boolean fromEndMessage) {
String text = sentence.path("sentence").asText("").trim();
if (text.isEmpty()) {
return null;
}
Integer sentenceId = sentence.has("sentence_id") && sentence.get("sentence_id").canConvertToInt()
? sentence.get("sentence_id").asInt()
: null;
String upstreamSentenceKey = readText(sentence, "sentence_key");
String sentenceKey = upstreamSentenceKey != null && !upstreamSentenceKey.isBlank()
? upstreamSentenceKey
: sentenceId == null ? "sentence-" + nextLegacySequence(state) : "sentence-" + sentenceId;
RealtimeMeetingTranscriptCacheItem item = findBySentenceKey(state, sentenceKey);
long now = System.currentTimeMillis();
if (item == null) {
item = new RealtimeMeetingTranscriptCacheItem();
item.setSentenceKey(sentenceKey);
item.setSentenceGroupKey(upstreamSentenceKey);
item.setSentenceId(sentenceId);
item.setSortOrder(nextSortOrder(state));
item.setFirstReceivedAt(now);
state.getItems().add(item);
}
item.setSentenceGroupKey(upstreamSentenceKey);
item.setSentenceType(readInteger(sentence, "sentence_type"));
item.setSpeakerId(resolveSpeakerId(sentence));
item.setSpeakerName(readText(sentence, "speaker_name"));
item.setUserId(readText(sentence, "user_id"));
item.setStartTime(readTimeMilliseconds(sentence, "start_time", "start"));
item.setEndTime(readTimeMilliseconds(sentence, "end_time", "end"));
item.setContent(text);
item.setFinalResult(fromEndMessage || Objects.equals(item.getSentenceType(), 1));
item.setUpdatedAt(now);
return item;
}
private void persistFinalSentence(Long meetingId, RealtimeMeetingTranscriptCacheItem item) {
if (meetingId == null || item == null || !Boolean.TRUE.equals(item.getFinalResult())) {
return;
}
String content = item.getContent() == null ? null : item.getContent().trim();
if (content == null || content.isBlank()) {
return;
}
String speakerId = resolveSpeakerId(item);
String speakerName = resolveSpeakerName(item);
if (item.getTranscriptId() != null) {
transcriptMapper.update(null, new LambdaUpdateWrapper<MeetingTranscript>()
.eq(MeetingTranscript::getId, item.getTranscriptId())
.set(MeetingTranscript::getSpeakerId, speakerId)
.set(MeetingTranscript::getSpeakerName, speakerName)
.set(MeetingTranscript::getContent, content)
.set(item.getStartTime() != null, MeetingTranscript::getStartTime, item.getStartTime())
.set(item.getEndTime() != null, MeetingTranscript::getEndTime, item.getEndTime()));
meetingTranscriptFileService.initializeTranscriptFileIfAbsent(meetingId);
return;
}
MeetingTranscript transcript = new MeetingTranscript();
transcript.setMeetingId(meetingId);
transcript.setSpeakerId(speakerId);
transcript.setSpeakerName(speakerName);
transcript.setContent(content);
transcript.setStartTime(item.getStartTime());
transcript.setEndTime(item.getEndTime());
transcript.setSortOrder(nextPersistedSortOrder(meetingId));
transcriptMapper.insert(transcript);
item.setTranscriptId(transcript.getId());
meetingTranscriptFileService.initializeTranscriptFileIfAbsent(meetingId);
}
private int nextPersistedSortOrder(Long meetingId) {
Integer maxSortOrder = transcriptMapper.selectList(new LambdaQueryWrapper<MeetingTranscript>()
.eq(MeetingTranscript::getMeetingId, meetingId)
.orderByDesc(MeetingTranscript::getSortOrder)
.last("LIMIT 1"))
.stream()
.findFirst()
.map(MeetingTranscript::getSortOrder)
.orElse(0);
return maxSortOrder == null ? 0 : maxSortOrder + 1;
}
private RealtimeMeetingTranscriptCacheItem findBySentenceKey(RealtimeMeetingTranscriptCacheState state, String sentenceKey) {
if (state.getItems() == null || state.getItems().isEmpty()) {
return null;
}
return state.getItems().stream()
.filter(item -> sentenceKey.equals(item.getSentenceKey()))
.findFirst()
.orElse(null);
}
private String resolveSpeakerId(RealtimeMeetingTranscriptCacheItem item) {
if (item == null) {
return null;
}
if (item.getUserId() != null && !item.getUserId().isBlank()) {
return item.getUserId().trim();
}
if (item.getSpeakerId() == null || item.getSpeakerId().isBlank() || "-1".equals(item.getSpeakerId().trim())) {
return null;
}
return item.getSpeakerId().trim();
}
private String resolveSpeakerName(RealtimeMeetingTranscriptCacheItem item) {
if (item == null) {
return null;
}
if (item.getSpeakerName() != null && !item.getSpeakerName().isBlank()) {
return item.getSpeakerName().trim();
}
String speakerId = resolveSpeakerId(item);
return speakerId == null || speakerId.isBlank() ? null : speakerId;
}
private String resolveSpeakerId(JsonNode sentence) {
String userId = readText(sentence, "user_id");
if (userId != null && !userId.isBlank()) {
return userId;
}
return readText(sentence, "speaker_id");
}
private Integer readInteger(JsonNode node, String fieldName) {
if (node == null || !node.has(fieldName) || !node.get(fieldName).canConvertToInt()) {
return null;
}
return node.get(fieldName).asInt();
}
private Integer readTimeMilliseconds(JsonNode node, String primaryField, String fallbackField) {
if (node == null) {
return null;
}
if (node.has(primaryField) && node.get(primaryField).canConvertToInt()) {
return node.get(primaryField).asInt();
}
if (node.has(fallbackField) && node.get(fallbackField).isNumber()) {
return Math.round((float) (node.get(fallbackField).asDouble() * 1000));
}
return null;
}
private String readText(JsonNode node, String fieldName) {
if (node == null || !node.has(fieldName) || node.get(fieldName).isNull()) {
return null;
}
String value = node.get(fieldName).asText("");
return value == null ? null : value.trim();
}
private long countNonEmptyItems(RealtimeMeetingTranscriptCacheState state) {
if (state.getItems() == null || state.getItems().isEmpty()) {
return 0L;
}
return state.getItems().stream()
.filter(item -> item.getContent() != null && !item.getContent().isBlank())
.count();
}
private int nextSortOrder(RealtimeMeetingTranscriptCacheState state) {
Integer current = state.getNextSortOrder();
int next = current == null ? 0 : current;
state.setNextSortOrder(next + 1);
return next;
}
private int nextLegacySequence(RealtimeMeetingTranscriptCacheState state) {
Integer current = state.getNextLegacySequence();
int next = current == null ? 0 : current;
state.setNextLegacySequence(next + 1);
return next;
}
private RealtimeMeetingTranscriptCacheState getOrCreateState(Long meetingId) {
RealtimeMeetingTranscriptCacheState state = transcriptCache.getState(meetingId);
if (state != null) {
if (state.getItems() == null) {
state.setItems(new ArrayList<>());
}
return state;
}
RealtimeMeetingTranscriptCacheState next = new RealtimeMeetingTranscriptCacheState();
next.setMeetingId(meetingId);
next.setItems(new ArrayList<>());
next.setNextSortOrder(0);
next.setNextLegacySequence(0);
next.setUpdatedAt(System.currentTimeMillis());
return next;
}
private Object lockForMeeting(Long meetingId) {
return meetingLocks.computeIfAbsent(meetingId, ignored -> new Object());
}
}

506
backend/src/main/java/com/imeeting/service/realtime/impl/TencentRealtimeAsrChannel.java
View File

@ -1,506 +0,0 @@
package com.imeeting.service.realtime.impl;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.imeeting.dto.biz.AiModelVO;
import com.imeeting.enums.ModelProviderEnum;
import com.imeeting.service.biz.RealtimeMeetingSessionStateService;
import com.imeeting.service.realtime.RealtimeAsrChannel;
import com.imeeting.service.realtime.RealtimeAsrChannelContext;
import com.imeeting.service.realtime.RealtimeMeetingTranscriptCacheService;
import com.tencent.asrspeaker.SpeakerConstant;
import com.tencent.asrspeaker.SpeakerRecognitionListener;
import com.tencent.asrspeaker.SpeakerRecognitionResponse;
import com.tencent.asrspeaker.SpeakerRecognizer;
import com.tencent.asrspeaker.SpeakerRecognizerRequest;
import com.tencent.asrspeaker.SpeakerSentenceItem;
import com.tencent.core.ws.Credential;
import com.tencent.core.ws.SpeechClient;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.CloseStatus;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@Slf4j
@Component
@RequiredArgsConstructor
public class TencentRealtimeAsrChannel implements RealtimeAsrChannel {
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
private static final String TARGET_WS_URL = "tencent-sdk://speaker-recognizer";
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 STATE_CONNECTED = "tencentConnected";
private static final String STATE_STARTED = "tencentStarted";
private static final String STATE_RECOGNIZER = "tencentRecognizer";
private static final String STATE_SPEECH_CLIENT = "tencentSpeechClient";
private static final String STATE_STOP_REQUESTED = "tencentStopRequested";
private static final String STATE_MEETING_COMPLETE_REQUESTED = "tencentMeetingCompleteRequested";
private static final String STATE_FRONTEND_DETACHED = "tencentFrontendDetached";
private static final String STATE_SPEAKER_CONTEXT_ID = "speakerContextId";
private static final String STATE_VOICE_ID = "voiceId";
private static final String STATE_PENDING_AUDIO_FRAMES = "pendingAudioFrames";
private static final String STATE_MODEL_CODE = "modelCode";
private static final String STATE_MEDIA_CONFIG = "mediaConfig";
private final RealtimeMeetingSessionStateService realtimeMeetingSessionStateService;
private final RealtimeMeetingTranscriptCacheService realtimeMeetingTranscriptCacheService;
@Override
public boolean supports(String provider) {
return ModelProviderEnum.TENCENT.getCode().equalsIgnoreCase(provider);
}
@Override
public String resolveTargetWsUrl(AiModelVO model) {
return TARGET_WS_URL;
}
@Override
public Map<String, Object> buildStartMessage(AiModelVO model,
String mode,
String language,
Integer useSpkId,
Boolean enablePunctuation,
Boolean enableItn,
Boolean enableTextRefine,
Boolean saveAudio,
List<Map<String, Object>> hotwords) {
Map<String, Object> payload = new HashMap<>();
payload.put("provider", ModelProviderEnum.TENCENT.getCode());
payload.put("engine_model_type", model.getModelCode());
payload.put("language", language);
Map<String, Object> root = new HashMap<>();
root.put("type", "start");
root.put("payload", payload);
return root;
}
@Override
public void connect(RealtimeAsrChannelContext context) throws Exception {
String connectionId = currentConnectionId(context);
if (connectionId == null || !realtimeMeetingSessionStateService.activate(context.getMeetingId(), connectionId)) {
context.getCallback().sendFrontendError(context.getMeetingId(), "REALTIME_ACTIVE_CONNECTION_EXISTS", "当前会议无法激活这条前端连接");
context.getCallback().closeFrontend(context.getMeetingId(), CloseStatus.POLICY_VIOLATION.withReason("当前会议无法激活这条前端连接"));
return;
}
context.getChannelState().put(STATE_CONNECTED, Boolean.TRUE);
context.getChannelState().put(STATE_STARTED, Boolean.FALSE);
context.getChannelState().put(STATE_STOP_REQUESTED, Boolean.FALSE);
context.getChannelState().put(STATE_MEETING_COMPLETE_REQUESTED, Boolean.FALSE);
context.getChannelState().put(STATE_FRONTEND_DETACHED, Boolean.FALSE);
context.getChannelState().putIfAbsent(STATE_PENDING_AUDIO_FRAMES, new java.util.ArrayList<byte[]>());
context.getCallback().onChannelOpen(context.getMeetingId());
}
@Override
public void handleFrontendText(RealtimeAsrChannelContext context, String payload) {
if (looksLikeStartMessage(payload)) {
startRecognizerIfNecessary(context);
return;
}
if (looksLikeStopMessage(payload)) {
context.getChannelState().put(STATE_STOP_REQUESTED, Boolean.TRUE);
stopRecognizer(context);
}
}
@Override
public void handleFrontendBinary(RealtimeAsrChannelContext context, byte[] payload) {
SpeakerRecognizer recognizer = getRecognizer(context);
if (payload == null || payload.length == 0) {
return;
}
if (recognizer == null) {
queuePendingAudioFrame(context, payload);
return;
}
try {
recognizer.write(payload);
} catch (Exception ex) {
handleChannelFailure(context, "REALTIME_UPSTREAM_ERROR", "腾讯实时 ASR 音频发送失败", ex);
}
}
@Override
public void closeMeeting(RealtimeAsrChannelContext context) {
context.getChannelState().put(STATE_STOP_REQUESTED, Boolean.TRUE);
context.getChannelState().put(STATE_MEETING_COMPLETE_REQUESTED, Boolean.TRUE);
stopRecognizer(context);
}
@Override
public boolean isOpen(RealtimeAsrChannelContext context) {
return !Boolean.TRUE.equals(context.getChannelState().get(STATE_MEETING_COMPLETE_REQUESTED));
}
@Override
public void onFrontendDetached(RealtimeAsrChannelContext context) {
context.getChannelState().put(STATE_FRONTEND_DETACHED, Boolean.TRUE);
context.getChannelState().put(STATE_STOP_REQUESTED, Boolean.TRUE);
stopRecognizer(context);
}
static String buildFrontendTranscriptMessage(String sentenceKey,
String text,
boolean isFinal,
Integer sentenceId,
Long startTime,
Long endTime,
Integer speakerId) throws JsonProcessingException {
ObjectNode root = OBJECT_MAPPER.createObjectNode();
root.put("type", isFinal ? "segment" : "partial");
ObjectNode data = root.putObject("data");
data.put("text", text);
data.put("is_final", isFinal);
if (sentenceId != null) {
data.put("sentence_id", sentenceId);
}
if (sentenceKey != null && !sentenceKey.isBlank()) {
data.put("sentence_key", sentenceKey);
}
if (startTime != null) {
data.put("start", startTime / 1000D);
}
if (endTime != null) {
data.put("end", endTime / 1000D);
}
if (speakerId != null) {
data.put("speaker_id", String.valueOf(speakerId));
}
return OBJECT_MAPPER.writeValueAsString(root);
}
private void startRecognizerIfNecessary(RealtimeAsrChannelContext context) {
synchronized (context.getChannelState()) {
if (Boolean.TRUE.equals(context.getChannelState().get(STATE_STARTED))) {
return;
}
try {
SpeechClient speechClient = createSpeechClient();
SpeakerRecognizerRequest request = createRecognizerRequest(context);
SpeakerRecognizer recognizer = createRecognizer(context, speechClient, request);
context.getChannelState().put(STATE_SPEECH_CLIENT, speechClient);
context.getChannelState().put(STATE_RECOGNIZER, recognizer);
context.getChannelState().put(STATE_VOICE_ID, request.getVoiceId());
recognizer.start();
context.getChannelState().put(STATE_STARTED, Boolean.TRUE);
flushPendingAudioFrames(context, recognizer);
} catch (Exception ex) {
handleChannelFailure(context, "REALTIME_UPSTREAM_CONNECT_FAILED", "腾讯实时 ASR 启动失败", ex);
}
}
}
protected SpeechClient createSpeechClient() {
return new SpeechClient(SpeakerConstant.DEFAULT_RT_REQ_URL);
}
protected SpeakerRecognizer createRecognizer(RealtimeAsrChannelContext context,
SpeechClient speechClient,
SpeakerRecognizerRequest request) {
return new SpeakerRecognizer(
speechClient,
buildCredential(context),
request,
new TencentRecognitionListener(context)
);
}
private Credential buildCredential(RealtimeAsrChannelContext context) {
Map<String, Object> mediaConfig = getMediaConfig(context);
String appId = readConfigString(mediaConfig, MEDIA_TENCENT_APP_ID);
String secretId = readConfigString(mediaConfig, MEDIA_TENCENT_SECRET_ID);
String secretKey = readConfigString(mediaConfig, MEDIA_TENCENT_SECRET_KEY);
if (appId == null || secretId == null || secretKey == null) {
throw new RuntimeException("腾讯实时 ASR 会话缺少鉴权配置");
}
return new Credential(appId, secretId, secretKey);
}
SpeakerRecognizerRequest createRecognizerRequest(RealtimeAsrChannelContext context) {
SpeakerRecognizerRequest request = SpeakerRecognizerRequest.init();
request.setEngineModelType(resolveEngineModelType(context));
request.setVoiceFormat(SpeakerConstant.AUDIO_FORMAT_PCM);
request.setVoiceId(UUID.randomUUID().toString());
//是否需要vad
request.setNeedVad(1);
//vad静默时间
request.setVadSilenceTime(1000);
// 分句策略参数 0小1大
request.setSentenceStrategy(0);
//是否进行阿拉伯数字智能转换 0否1智能 23:打开数学相关转化
request.setConvertNumMode(1);
request.setSpeakerDiarization(1);
//启动断点续传
request.setEnableSpeakerContext(1);
String speakerContextId = resolveSpeakerContextId(context);
if (speakerContextId != null) {
request.setSpeakerContextId(speakerContextId);
}
return request;
}
private String resolveEngineModelType(RealtimeAsrChannelContext context) {
Object modelCode = context.getChannelState().get(STATE_MODEL_CODE);
if (modelCode instanceof String value && !value.isBlank()) {
return value;
}
return "16k_zh";
}
private String resolveSpeakerContextId(RealtimeAsrChannelContext context) {
Object speakerContextId = context.getChannelState().get(STATE_SPEAKER_CONTEXT_ID);
if (speakerContextId instanceof String value && !value.isBlank()) {
return value;
}
var status = realtimeMeetingSessionStateService.getStatus(context.getMeetingId());
if (status == null || status.getResumeConfig() == null) {
return null;
}
String value = status.getResumeConfig().getSpeakerContextId();
if (value == null || value.isBlank()) {
return null;
}
context.getChannelState().put(STATE_SPEAKER_CONTEXT_ID, value);
return value;
}
@SuppressWarnings("unchecked")
private Map<String, Object> getMediaConfig(RealtimeAsrChannelContext context) {
Object mediaConfig = context.getChannelState().get(STATE_MEDIA_CONFIG);
if (mediaConfig instanceof Map<?, ?> map) {
return (Map<String, Object>) map;
}
return Map.of();
}
private String readConfigString(Map<String, Object> mediaConfig, String key) {
Object value = mediaConfig.get(key);
if (value == null) {
return null;
}
String text = String.valueOf(value).trim();
return text.isEmpty() ? null : text;
}
private SpeakerRecognizer getRecognizer(RealtimeAsrChannelContext context) {
Object recognizer = context.getChannelState().get(STATE_RECOGNIZER);
return recognizer instanceof SpeakerRecognizer value ? value : null;
}
private SpeechClient getSpeechClient(RealtimeAsrChannelContext context) {
Object speechClient = context.getChannelState().get(STATE_SPEECH_CLIENT);
return speechClient instanceof SpeechClient value ? value : null;
}
private void stopRecognizer(RealtimeAsrChannelContext context) {
SpeakerRecognizer recognizer = getRecognizer(context);
if (recognizer == null) {
shutdownSdkResources(context);
return;
}
try {
recognizer.stop();
} catch (Exception ex) {
log.warn("Tencent realtime ASR stop failed, meetingId={}, sessionId={}",
context.getMeetingId(), currentConnectionId(context), ex);
shutdownSdkResources(context);
}
}
private void shutdownSdkResources(RealtimeAsrChannelContext context) {
SpeakerRecognizer recognizer = getRecognizer(context);
if (recognizer != null) {
try {
recognizer.close();
} catch (Exception ignored) {
// ignore
}
}
SpeechClient speechClient = getSpeechClient(context);
if (speechClient != null) {
try {
speechClient.shutdown();
} catch (Exception ignored) {
// ignore
}
}
context.getChannelState().remove(STATE_RECOGNIZER);
context.getChannelState().remove(STATE_SPEECH_CLIENT);
context.getChannelState().put(STATE_STARTED, Boolean.FALSE);
}
private void forwardResponse(RealtimeAsrChannelContext context,
SpeakerRecognitionResponse response,
boolean forceFinal) {
if (response == null || response.getSentences() == null || response.getSentences().getSentenceList() == null) {
return;
}
try {
rememberSpeakerContext(context, response);
String cachePayload = buildCachePayload(response, forceFinal);
realtimeMeetingTranscriptCacheService.mergeUpstreamMessage(context.getMeetingId(), cachePayload);
for (SpeakerSentenceItem item : response.getSentences().getSentenceList()) {
if (item == null || item.getSentence() == null || item.getSentence().trim().isEmpty()) {
continue;
}
boolean isFinal = forceFinal || item.getSentenceType() == 1;
context.getCallback().sendFrontendText(
context.getMeetingId(),
buildFrontendTranscriptMessage(
buildSentenceKey(context, item.getSentenceId()),
item.getSentence().trim(),
isFinal,
item.getSentenceId(),
item.getStartTime(),
item.getEndTime(),
item.getSpeakerId()
)
);
}
} catch (Exception ex) {
handleChannelFailure(context, "REALTIME_UPSTREAM_ERROR", "腾讯实时 ASR 结果转发失败", ex);
}
}
private void rememberSpeakerContext(RealtimeAsrChannelContext context, SpeakerRecognitionResponse response) {
if (response == null || response.getSpeakerContextId() == null || response.getSpeakerContextId().isBlank()) {
return;
}
String speakerContextId = response.getSpeakerContextId().trim();
context.getChannelState().put(STATE_SPEAKER_CONTEXT_ID, speakerContextId);
realtimeMeetingSessionStateService.rememberSpeakerContext(context.getMeetingId(), speakerContextId);
}
private String buildSentenceKey(RealtimeAsrChannelContext context, Integer sentenceId) {
Object voiceId = context.getChannelState().get(STATE_VOICE_ID);
if (!(voiceId instanceof String value) || value.isBlank() || sentenceId == null) {
return null;
}
return value + "-" + sentenceId;
}
@SuppressWarnings("unchecked")
private void queuePendingAudioFrame(RealtimeAsrChannelContext context, byte[] payload) {
Object frames = context.getChannelState().get(STATE_PENDING_AUDIO_FRAMES);
if (frames instanceof List<?> list) {
((List<byte[]>) list).add(payload.clone());
return;
}
List<byte[]> next = new java.util.ArrayList<>();
next.add(payload.clone());
context.getChannelState().put(STATE_PENDING_AUDIO_FRAMES, next);
}
@SuppressWarnings("unchecked")
private void flushPendingAudioFrames(RealtimeAsrChannelContext context, SpeakerRecognizer recognizer) {
Object frames = context.getChannelState().get(STATE_PENDING_AUDIO_FRAMES);
if (!(frames instanceof List<?> list) || list.isEmpty()) {
return;
}
List<byte[]> pendingFrames = (List<byte[]>) list;
for (byte[] frame : pendingFrames) {
if (frame != null && frame.length > 0) {
recognizer.write(frame);
}
}
pendingFrames.clear();
}
private String buildCachePayload(SpeakerRecognitionResponse response, boolean forceFinal) throws JsonProcessingException {
ObjectNode root = OBJECT_MAPPER.createObjectNode();
root.put("type", forceFinal ? "end" : "sentences");
ArrayNode sentences = root.putArray("sentences");
for (SpeakerSentenceItem item : response.getSentences().getSentenceList()) {
if (item == null || item.getSentence() == null || item.getSentence().trim().isEmpty()) {
continue;
}
ObjectNode sentenceNode = sentences.addObject();
sentenceNode.put("sentence", item.getSentence().trim());
sentenceNode.put("sentence_type", forceFinal ? 1 : item.getSentenceType());
sentenceNode.put("sentence_id", item.getSentenceId());
sentenceNode.put("speaker_id", String.valueOf(item.getSpeakerId()));
sentenceNode.put("start_time", item.getStartTime());
sentenceNode.put("end_time", item.getEndTime());
}
return OBJECT_MAPPER.writeValueAsString(root);
}
private void handleChannelFailure(RealtimeAsrChannelContext context, String code, String message, Exception ex) {
log.error("Tencent realtime ASR channel failed, meetingId={}, sessionId={}",
context.getMeetingId(), currentConnectionId(context), ex);
shutdownSdkResources(context);
context.getCallback().sendFrontendError(context.getMeetingId(), code, message);
CompletableFuture.delayedExecutor(200, TimeUnit.MILLISECONDS).execute(
() -> context.getCallback().closeFrontend(context.getMeetingId(), CloseStatus.SERVER_ERROR)
);
}
private String currentConnectionId(RealtimeAsrChannelContext context) {
return context.getRawSession() == null ? null : context.getRawSession().getId();
}
private static boolean looksLikeStartMessage(String payload) {
if (payload == null || payload.isBlank()) {
return false;
}
String normalized = payload.replaceAll("\\s+", "");
return normalized.contains("\"type\":\"start\"");
}
private static boolean looksLikeStopMessage(String payload) {
if (payload == null || payload.isBlank()) {
return false;
}
String normalized = payload.replaceAll("\\s+", "");
return normalized.contains("\"type\":\"stop\"");
}
private final class TencentRecognitionListener extends SpeakerRecognitionListener {
private final RealtimeAsrChannelContext context;
private TencentRecognitionListener(RealtimeAsrChannelContext context) {
this.context = context;
}
@Override
public void onRecognitionStart(SpeakerRecognitionResponse response) {
rememberSpeakerContext(context, response);
log.info("Tencent realtime ASR started, meetingId={}, sessionId={}",
context.getMeetingId(), currentConnectionId(context));
}
@Override
public void onRecognitionSentences(SpeakerRecognitionResponse response) {
forwardResponse(context, response, false);
}
@Override
public void onSentenceEnd(SpeakerRecognitionResponse response) {
forwardResponse(context, response, true);
shutdownSdkResources(context);
if (Boolean.TRUE.equals(context.getChannelState().get(STATE_MEETING_COMPLETE_REQUESTED))) {
context.getCallback().removeMeetingSession(context.getMeetingId());
}
}
@Override
public void onFail(SpeakerRecognitionResponse response, Exception error) {
handleChannelFailure(context, "REALTIME_UPSTREAM_ERROR", "腾讯实时 ASR 识别失败", error);
}
}
}

39
backend/src/main/java/com/imeeting/support/redis/RealtimeMeetingTranscriptCache.java
View File

@ -1,39 +0,0 @@
package com.imeeting.support.redis;
import com.imeeting.common.RedisKeys;
import com.imeeting.dto.biz.RealtimeMeetingTranscriptCacheState;
import com.imeeting.support.RedisSupport;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;
import java.time.Duration;
@Component
@RequiredArgsConstructor
public class RealtimeMeetingTranscriptCache {
private static final Duration CACHE_TTL = Duration.ofHours(12);
private final RedisSupport redisSupport;
public RealtimeMeetingTranscriptCacheState getState(Long meetingId) {
if (meetingId == null) {
return null;
}
return redisSupport.getJsonQuietly(RedisKeys.realtimeMeetingTranscriptCacheKey(meetingId), RealtimeMeetingTranscriptCacheState.class);
}
public void saveState(RealtimeMeetingTranscriptCacheState state) {
if (state == null || state.getMeetingId() == null) {
return;
}
redisSupport.setJson(RedisKeys.realtimeMeetingTranscriptCacheKey(state.getMeetingId()), state, CACHE_TTL);
}
public void clear(Long meetingId) {
if (meetingId == null) {
return;
}
redisSupport.deleteQuietly(RedisKeys.realtimeMeetingTranscriptCacheKey(meetingId));
}
}

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

@ -1,133 +1,173 @@
package com.imeeting.websocket;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.imeeting.dto.biz.RealtimeMeetingTranscriptCacheItem;
import com.imeeting.dto.biz.RealtimeSocketSessionData;
import com.imeeting.service.biz.RealtimeMeetingSessionStateService;
import com.imeeting.service.biz.RealtimeMeetingSocketSessionService;
import com.imeeting.service.realtime.RealtimeAsrChannel;
import com.imeeting.service.realtime.RealtimeAsrChannelCallback;
import com.imeeting.service.realtime.RealtimeAsrChannelContext;
import com.imeeting.service.realtime.RealtimeAsrChannelFactory;
import com.imeeting.service.realtime.RealtimeMeetingAudioStorageService;
import com.imeeting.service.realtime.RealtimeMeetingTranscriptCacheService;
import com.imeeting.service.realtime.impl.LocalRealtimeAsrChannel;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.BinaryMessage;
import org.springframework.web.socket.CloseStatus;
import org.springframework.web.socket.PingMessage;
import org.springframework.web.socket.PongMessage;
import org.springframework.web.socket.TextMessage;
import org.springframework.web.socket.WebSocketSession;
import org.springframework.web.socket.handler.AbstractWebSocketHandler;
import org.springframework.web.socket.handler.ConcurrentWebSocketSessionDecorator;
import java.io.ByteArrayOutputStream;
import java.net.URI;
import java.net.URLDecoder;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Supplier;
@Slf4j
@Component
@RequiredArgsConstructor
public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandler {
private static final String ATTR_FRONTEND_SESSION = "frontendSession";
private static final String ATTR_UPSTREAM_SOCKET = "upstreamSocket";
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_FRONTEND_TEXT_COUNT = "frontendTextCount";
private static final String ATTR_FRONTEND_BINARY_COUNT = "frontendBinaryCount";
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
private static final String ATTR_UPSTREAM_SEND_CHAIN = "upstreamSendChain";
private static final String ATTR_START_MESSAGE_SENT = "startMessageSent";
private static final String ATTR_PENDING_AUDIO_FRAMES = "pendingAudioFrames";
private static final CompletableFuture<Void> COMPLETED = CompletableFuture.completedFuture(null);
private final RealtimeMeetingSocketSessionService realtimeMeetingSocketSessionService;
private final RealtimeMeetingSessionStateService realtimeMeetingSessionStateService;
private final RealtimeMeetingAudioStorageService realtimeMeetingAudioStorageService;
private final RealtimeMeetingTranscriptCacheService realtimeMeetingTranscriptCacheService;
private final RealtimeAsrChannelFactory realtimeAsrChannelFactory;
private final ConcurrentMap<Long, MeetingChannelSession> meetingSessions = new ConcurrentHashMap<>();
private final ConcurrentMap<Long, Object> 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("Realtime websocket rejected: invalid session token, sessionId={}", session.getId());
session.close(CloseStatus.POLICY_VIOLATION.withReason("实时 Socket 会话无效"));
return;
}
ConcurrentWebSocketSessionDecorator frontendSession =
new ConcurrentWebSocketSessionDecorator(session, (int) Duration.ofSeconds(15).toMillis(), 1024 * 1024);
session.getAttributes().put(ATTR_FRONTEND_SESSION, frontendSession);
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_FRONTEND_TEXT_COUNT, new AtomicInteger());
session.getAttributes().put(ATTR_FRONTEND_BINARY_COUNT, new AtomicInteger());
session.getAttributes().put(ATTR_UPSTREAM_SEND_CHAIN, COMPLETED);
session.getAttributes().put(ATTR_START_MESSAGE_SENT, Boolean.FALSE);
session.getAttributes().put(ATTR_PENDING_AUDIO_FRAMES, new ArrayList<byte[]>());
realtimeMeetingAudioStorageService.openSession(sessionData.getMeetingId(), session.getId());
log.info("实时会议 websocket 已接入：meetingId={}, sessionId={}, provider={}, upstream={}",
sessionData.getMeetingId(), session.getId(), sessionData.getProvider(), sessionData.getTargetWsUrl());
log.info("Realtime websocket accepted: meetingId={}, sessionId={}, upstream={}",
sessionData.getMeetingId(), session.getId(), sessionData.getTargetWsUrl());
attachFrontendSession(sessionData, session, frontendSession);
java.net.http.WebSocket upstreamSocket;
try {
upstreamSocket = java.net.http.HttpClient.newHttpClient()
.newWebSocketBuilder()
.buildAsync(URI.create(sessionData.getTargetWsUrl()),
new UpstreamListener(
frontendSession,
session,
sessionData.getMeetingId(),
sessionData.getTargetWsUrl(),
realtimeMeetingSessionStateService
))
.get();
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
log.error("Realtime websocket upstream connect interrupted: meetingId={}, sessionId={}",
sessionData.getMeetingId(), session.getId(), ex);
sendFrontendError(frontendSession, "REALTIME_UPSTREAM_CONNECT_INTERRUPTED", "连接第三方识别服务时被中断");
realtimeMeetingAudioStorageService.closeSession(session.getId());
frontendSession.close(CloseStatus.SERVER_ERROR.withReason("连接上游服务时被中断"));
return;
} catch (ExecutionException | CompletionException ex) {
log.warn("Failed to connect upstream websocket, meetingId={}, target={}", sessionData.getMeetingId(), sessionData.getTargetWsUrl(), ex);
sendFrontendError(frontendSession, "REALTIME_UPSTREAM_CONNECT_FAILED", "连接第三方识别服务失败，请检查模型 WebSocket 配置或服务状态");
realtimeMeetingAudioStorageService.closeSession(session.getId());
frontendSession.close(CloseStatus.SERVER_ERROR.withReason("连接 ASR WebSocket 失败"));
return;
}
session.getAttributes().put(ATTR_UPSTREAM_SOCKET, upstreamSocket);
}
@Override
protected void handleTextMessage(WebSocketSession session, TextMessage message) {
MeetingChannelSession meetingSession = getMeetingSession(session);
if (meetingSession == null || !meetingSession.isChannelOpen()) {
log.warn("前端文本消息已忽略：上游 ASR 连接不可用，meetingId={}, sessionId={}",
java.net.http.WebSocket upstreamSocket = getUpstreamSocket(session);
if (upstreamSocket == null) {
log.warn("Frontend text ignored because upstream socket is unavailable, 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);
log.info("Frontend text -> upstream: meetingId={}, sessionId={}, count={}, payload={}",
session.getAttributes().get(ATTR_MEETING_ID), session.getId(), count, summarizeText(message.getPayload()));
sendUpstreamOrdered(session, () -> upstreamSocket.sendText(message.getPayload(), true), "text");
if (looksLikeStartMessage(message.getPayload())) {
session.getAttributes().put(ATTR_START_MESSAGE_SENT, Boolean.TRUE);
flushPendingAudioFrames(session, upstreamSocket);
}
}
@Override
protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) {
MeetingChannelSession meetingSession = getMeetingSession(session);
if (meetingSession == null || !meetingSession.isChannelOpen()) {
log.warn("前端音频帧已忽略：上游 ASR 连接不可用，meetingId={}, sessionId={}",
java.net.http.WebSocket upstreamSocket = getUpstreamSocket(session);
if (upstreamSocket == null) {
log.warn("Frontend binary ignored because upstream socket is unavailable, 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("Frontend binary -> upstream: meetingId={}, sessionId={}, count={}, bytes={}",
session.getAttributes().get(ATTR_MEETING_ID), session.getId(), count, bytes);
}
byte[] payload = toByteArray(message.getPayload());
realtimeMeetingAudioStorageService.append(session.getId(), payload);
meetingSession.channel.handleFrontendBinary(meetingSession.context, payload);
if (!Boolean.TRUE.equals(session.getAttributes().get(ATTR_START_MESSAGE_SENT))) {
queuePendingAudioFrame(session, payload);
if (shouldLogBinaryFrame(count)) {
log.warn("Frontend binary queued before start message: meetingId={}, sessionId={}, count={}, bytes={}",
session.getAttributes().get(ATTR_MEETING_ID), session.getId(), count, bytes);
}
return;
}
sendUpstreamOrdered(session, () -> upstreamSocket.sendBinary(ByteBuffer.wrap(payload), true), "binary");
}
@Override
protected void handlePongMessage(WebSocketSession session, PongMessage message) {
if (getMeetingSession(session) == null) {
java.net.http.WebSocket upstreamSocket = getUpstreamSocket(session);
if (upstreamSocket == null) {
return;
}
log.debug("前端 pong 已在本地忽略：meetingId={}, sessionId={}, bytes={}",
session.getAttributes().get(ATTR_MEETING_ID), session.getId(), message.getPayloadLength());
sendUpstreamOrdered(session, () -> upstreamSocket.sendPong(copyBuffer(message.getPayload())), "pong");
}
@Override
public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception {
log.error("实时会议 websocket 传输异常：meetingId={}, sessionId={}, upstream={}",
log.error("Realtime websocket transport error: 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());
closeUpstreamSocket(session, CloseStatus.SERVER_ERROR);
if (session.isOpen()) {
session.close(CloseStatus.SERVER_ERROR);
}
@ -135,160 +175,32 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
@Override
public void afterConnectionClosed(WebSocketSession session, CloseStatus status) {
log.info("实时会议 websocket 已关闭：meetingId={}, sessionId={}, code={}, reason={}",
log.info("Realtime websocket closed: 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());
realtimeMeetingSessionStateService.pauseByDisconnect(meetingId, session.getId());
}
realtimeMeetingAudioStorageService.closeSession(session.getId());
closeUpstreamSocket(session, status);
}
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);
private java.net.http.WebSocket getUpstreamSocket(WebSocketSession session) {
Object socket = session.getAttributes().get(ATTR_UPSTREAM_SOCKET);
if (socket instanceof java.net.http.WebSocket webSocket) {
return webSocket;
}
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);
}
return null;
}
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;
}
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 void closeUpstreamSocket(WebSocketSession session, CloseStatus status) {
java.net.http.WebSocket upstreamSocket = getUpstreamSocket(session);
if (upstreamSocket != null) {
upstreamSocket.sendClose(status.getCode(), status.getReason() == null ? "" : status.getReason());
session.getAttributes().remove(ATTR_UPSTREAM_SOCKET);
}
}
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()) {
return null;
@ -301,6 +213,13 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
.orElse(null);
}
private ByteBuffer copyBuffer(ByteBuffer source) {
ByteBuffer duplicate = source.asReadOnlyBuffer();
byte[] bytes = new byte[duplicate.remaining()];
duplicate.get(bytes);
return ByteBuffer.wrap(bytes);
}
private byte[] toByteArray(ByteBuffer source) {
ByteBuffer duplicate = source.asReadOnlyBuffer();
byte[] bytes = new byte[duplicate.remaining()];
@ -316,18 +235,21 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
return 0;
}
private void sendFrontendError(ConcurrentWebSocketSessionDecorator frontendSession, String code, String message) {
try {
if (!frontendSession.isOpen()) {
return;
}
Map<String, Object> 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);
@SuppressWarnings("unchecked")
private void sendUpstreamOrdered(WebSocketSession session, Supplier<CompletableFuture<?>> sendAction, String messageType) {
synchronized (session) {
CompletableFuture<Void> chain = (CompletableFuture<Void>) session.getAttributes()
.getOrDefault(ATTR_UPSTREAM_SEND_CHAIN, COMPLETED);
CompletableFuture<Void> nextChain = chain
.exceptionally(ex -> null)
.thenCompose(ignored -> sendAction.get().thenApply(ignoredResult -> null));
nextChain = nextChain.whenComplete((ignored, ex) -> {
if (ex != null) {
log.error("Ordered upstream send failed: meetingId={}, sessionId={}, type={}",
session.getAttributes().get(ATTR_MEETING_ID), session.getId(), messageType, ex);
}
});
session.getAttributes().put(ATTR_UPSTREAM_SEND_CHAIN, nextChain);
}
}
@ -346,113 +268,237 @@ public class RealtimeMeetingProxyWebSocketHandler extends AbstractWebSocketHandl
return normalized.substring(0, 240) + "...";
}
static final class MeetingChannelSession {
private boolean looksLikeStartMessage(String payload) {
if (payload == null || payload.isBlank()) {
return false;
}
String normalized = payload.replaceAll("\\s+", "");
return normalized.contains("\"type\":\"start\"");
}
private void sendFrontendError(ConcurrentWebSocketSessionDecorator frontendSession, String code, String message) {
try {
if (!frontendSession.isOpen()) {
return;
}
Map<String, Object> payload = new HashMap<>();
payload.put("type", "error");
payload.put("code", code);
payload.put("message", message);
frontendSession.sendMessage(new TextMessage(new com.fasterxml.jackson.databind.ObjectMapper().writeValueAsString(payload)));
} catch (Exception ex) {
log.warn("Failed to send realtime proxy error to frontend: code={}", code, ex);
}
}
@SuppressWarnings("unchecked")
private void queuePendingAudioFrame(WebSocketSession session, byte[] payload) {
synchronized (session) {
List<byte[]> pendingFrames = (List<byte[]>) session.getAttributes().get(ATTR_PENDING_AUDIO_FRAMES);
if (pendingFrames == null) {
pendingFrames = new ArrayList<>();
session.getAttributes().put(ATTR_PENDING_AUDIO_FRAMES, pendingFrames);
}
pendingFrames.add(payload);
}
}
@SuppressWarnings("unchecked")
private void flushPendingAudioFrames(WebSocketSession session, java.net.http.WebSocket upstreamSocket) {
List<byte[]> pendingFrames;
synchronized (session) {
pendingFrames = (List<byte[]>) session.getAttributes().get(ATTR_PENDING_AUDIO_FRAMES);
if (pendingFrames == null || pendingFrames.isEmpty()) {
return;
}
session.getAttributes().put(ATTR_PENDING_AUDIO_FRAMES, new ArrayList<byte[]>());
}
log.info("Flushing queued audio frames after start message: meetingId={}, sessionId={}, frameCount={}",
session.getAttributes().get(ATTR_MEETING_ID), session.getId(), pendingFrames.size());
for (byte[] frame : pendingFrames) {
sendUpstreamOrdered(session, () -> upstreamSocket.sendBinary(ByteBuffer.wrap(frame), true), "binary-flush");
}
}
private static final class UpstreamListener implements java.net.http.WebSocket.Listener {
private final ConcurrentWebSocketSessionDecorator frontendSession;
private final WebSocketSession rawSession;
private final Long meetingId;
private final RealtimeAsrChannel channel;
private final RealtimeAsrChannelContext context;
private final String targetWsUrl;
private final RealtimeMeetingSessionStateService realtimeMeetingSessionStateService;
private final StringBuilder textBuffer = new StringBuilder();
private final ByteArrayOutputStream binaryBuffer = new ByteArrayOutputStream();
private final AtomicInteger upstreamTextCount = new AtomicInteger();
private final AtomicInteger upstreamBinaryCount = new AtomicInteger();
private MeetingChannelSession(Long meetingId, RealtimeAsrChannel channel, RealtimeAsrChannelContext context) {
private UpstreamListener(ConcurrentWebSocketSessionDecorator frontendSession, WebSocketSession rawSession,
Long meetingId, String targetWsUrl,
RealtimeMeetingSessionStateService realtimeMeetingSessionStateService) {
this.frontendSession = frontendSession;
this.rawSession = rawSession;
this.meetingId = meetingId;
this.channel = channel;
this.context = context;
}
private void bindFrontend(WebSocketSession rawSession, ConcurrentWebSocketSessionDecorator frontendSession) {
context.bindFrontendSession(rawSession, frontendSession);
this.targetWsUrl = targetWsUrl;
this.realtimeMeetingSessionStateService = realtimeMeetingSessionStateService;
}
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 final class HandlerChannelCallback implements RealtimeAsrChannelCallback {
@Override
public void onChannelOpen(Long meetingId) throws Exception {
MeetingChannelSession meetingSession = meetingSessions.get(meetingId);
if (meetingSession == null) {
public void onOpen(java.net.http.WebSocket webSocket) {
log.info("Upstream websocket opened: meetingId={}, sessionId={}, upstream={}",
meetingId, rawSession.getId(), targetWsUrl);
if (!realtimeMeetingSessionStateService.activate(meetingId, rawSession.getId())) {
sendFrontendError("REALTIME_ACTIVE_CONNECTION_EXISTS", "当前会议已有活跃实时连接，请先关闭旧连接后再继续");
webSocket.sendClose(CloseStatus.POLICY_VIOLATION.getCode(), "Active realtime connection already exists");
closeFrontend(CloseStatus.POLICY_VIOLATION.withReason("已存在活动的实时连接"));
return;
}
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) {
return;
}
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));
}
}
@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);
}
}
@Override
public void removeMeetingSession(Long meetingId) {
RealtimeMeetingProxyWebSocketHandler.this.removeMeetingSession(meetingId);
}
@Override
public void closeFrontend(Long meetingId, CloseStatus status) {
MeetingChannelSession meetingSession = meetingSessions.get(meetingId);
if (meetingSession == null) {
return;
}
try {
WebSocketSession rawSession = meetingSession.context.getRawSession();
if (rawSession != null && rawSession.isOpen()) {
if (frontendSession.isOpen()) {
frontendSession.sendMessage(new TextMessage("{\"type\":\"proxy_ready\"}"));
}
} catch (Exception ex) {
log.error("Failed to notify frontend that upstream websocket is ready: meetingId={}, sessionId={}", meetingId, rawSession.getId(), ex);
closeFrontend(CloseStatus.SERVER_ERROR);
return;
}
webSocket.request(1);
}
@Override
public java.util.concurrent.CompletionStage<?> onText(java.net.http.WebSocket webSocket, CharSequence data, boolean last) {
textBuffer.append(data);
if (last) {
int count = upstreamTextCount.incrementAndGet();
try {
if (frontendSession.isOpen()) {
frontendSession.sendMessage(new TextMessage(textBuffer.toString()));
}
log.info("Upstream text -> frontend: meetingId={}, sessionId={}, count={}, payload={}",
meetingId, rawSession.getId(), count, summarizeText(textBuffer.toString()));
} catch (Exception ex) {
log.error("Failed to forward upstream text: meetingId={}, sessionId={}", meetingId, rawSession.getId(), ex);
closeFrontend(CloseStatus.SERVER_ERROR);
} finally {
textBuffer.setLength(0);
}
}
webSocket.request(1);
return COMPLETED;
}
@Override
public java.util.concurrent.CompletionStage<?> onBinary(java.net.http.WebSocket webSocket, ByteBuffer data, boolean last) {
byte[] chunk = new byte[data.remaining()];
data.get(chunk);
binaryBuffer.writeBytes(chunk);
if (last) {
int count = upstreamBinaryCount.incrementAndGet();
try {
if (frontendSession.isOpen()) {
frontendSession.sendMessage(new BinaryMessage(binaryBuffer.toByteArray()));
}
if (shouldLogBinaryFrame(count)) {
log.info("Upstream binary -> frontend: meetingId={}, sessionId={}, count={}, bytes={}",
meetingId, rawSession.getId(), count, binaryBuffer.size());
}
} catch (Exception ex) {
log.error("Failed to forward upstream binary: meetingId={}, sessionId={}", meetingId, rawSession.getId(), ex);
closeFrontend(CloseStatus.SERVER_ERROR);
} finally {
binaryBuffer.reset();
}
}
webSocket.request(1);
return COMPLETED;
}
@Override
public java.util.concurrent.CompletionStage<?> onPing(java.net.http.WebSocket webSocket, ByteBuffer message) {
try {
if (frontendSession.isOpen()) {
frontendSession.sendMessage(new PingMessage(copyBuffer(message)));
}
log.info("Upstream ping -> frontend: meetingId={}, sessionId={}, bytes={}",
meetingId, rawSession.getId(), message.remaining());
} catch (Exception ex) {
log.error("Failed to forward upstream ping: meetingId={}, sessionId={}", meetingId, rawSession.getId(), ex);
closeFrontend(CloseStatus.SERVER_ERROR);
}
webSocket.request(1);
return COMPLETED;
}
@Override
public java.util.concurrent.CompletionStage<?> onPong(java.net.http.WebSocket webSocket, ByteBuffer message) {
try {
if (frontendSession.isOpen()) {
frontendSession.sendMessage(new PongMessage(copyBuffer(message)));
}
log.info("Upstream pong -> frontend: meetingId={}, sessionId={}, bytes={}",
meetingId, rawSession.getId(), message.remaining());
} catch (Exception ex) {
log.error("Failed to forward upstream pong: meetingId={}, sessionId={}", meetingId, rawSession.getId(), ex);
closeFrontend(CloseStatus.SERVER_ERROR);
}
webSocket.request(1);
return COMPLETED;
}
@Override
public java.util.concurrent.CompletionStage<?> onClose(java.net.http.WebSocket webSocket, int statusCode, String reason) {
log.info("Upstream websocket closed: meetingId={}, sessionId={}, code={}, reason={}",
meetingId, rawSession.getId(), statusCode, reason);
sendFrontendError("REALTIME_UPSTREAM_CLOSED", reason == null || reason.isBlank() ? "第三方识别服务已断开连接" : "第三方识别服务已断开: " + reason);
closeFrontend(new CloseStatus(statusCode, reason));
return COMPLETED;
}
@Override
public void onError(java.net.http.WebSocket webSocket, Throwable error) {
log.error("Upstream websocket error: meetingId={}, sessionId={}, upstream={}",
meetingId, rawSession.getId(), targetWsUrl, error);
sendFrontendError("REALTIME_UPSTREAM_ERROR", error == null || error.getMessage() == null || error.getMessage().isBlank()
? "第三方识别服务连接异常"
: "第三方识别服务连接异常: " + error.getMessage());
closeFrontend(CloseStatus.SERVER_ERROR);
}
private void sendFrontendError(String code, String message) {
try {
if (!frontendSession.isOpen()) {
return;
}
frontendSession.sendMessage(new TextMessage("{\"type\":\"error\",\"code\":\"" + code + "\",\"message\":\"" + escapeJson(message) + "\"}"));
} catch (Exception ex) {
log.warn("Failed to send upstream error to frontend: meetingId={}, sessionId={}, code={}", meetingId, rawSession.getId(), code, ex);
}
}
private String escapeJson(String value) {
if (value == null) {
return "";
}
return value
.replace("\\", "\\\\")
.replace("\"", "\\\"")
.replace("\r", "\\r")
.replace("\n", "\\n");
}
private void closeFrontend(CloseStatus status) {
try {
if (rawSession.isOpen()) {
rawSession.close(status);
}
} catch (Exception ignored) {
// ignore close failure
}
}
private ByteBuffer copyBuffer(ByteBuffer source) {
ByteBuffer duplicate = source.asReadOnlyBuffer();
byte[] bytes = new byte[duplicate.remaining()];
duplicate.get(bytes);
return ByteBuffer.wrap(bytes);
}
}
}

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

@ -349,64 +349,6 @@ class AiModelServiceImplTest {
assertNull(captor.getValue().getApiKey());
}
@Test
void saveModelShouldRejectTencentAsrWithoutSecretKey() {
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(
"tencentAppId", "app-id",
"tencentSecretId", "secret-id"
));
RuntimeException ex = assertThrows(RuntimeException.class, () -> service.saveModel(dto));
assertEquals("腾讯实时 ASR 模型必须配置 mediaConfig.tencentSecretKey", ex.getMessage());
}
@Test
void saveModelShouldPersistTencentAsrWithoutBaseUrl() {
AsrModelMapper asrModelMapper = mock(AsrModelMapper.class);
when(asrModelMapper.insert(any(AsrModel.class))).thenReturn(1);
AiModelServiceImpl service = new AiModelServiceImpl(
objectMapper,
asrModelMapper,
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(
"tencentAppId", "app-id",
"tencentSecretId", "secret-id",
"tencentSecretKey", "secret-key"
));
service.saveModel(dto);
ArgumentCaptor<AsrModel> captor = ArgumentCaptor.forClass(AsrModel.class);
verify(asrModelMapper, times(1)).insert(captor.capture());
assertEquals("tencent", captor.getValue().getProvider());
assertEquals("16k_zh", captor.getValue().getModelCode());
assertEquals("secret-key", captor.getValue().getMediaConfig().get("tencentSecretKey"));
assertNull(captor.getValue().getBaseUrl());
}
private void captureRequest(HttpExchange exchange,
AtomicReference<String> requestPath,
AtomicReference<String> authorization,

6
frontend/src/api/business/meeting.ts
View File

@ -249,6 +249,12 @@ export const createRealtimeMeeting = (data: CreateRealtimeMeetingCommand) => {
);
};
export const appendRealtimeTranscripts = (meetingId: number, data: RealtimeTranscriptItemDTO[]) => {
return http.post<{ code: string; data: boolean; msg: string }>(
`/api/biz/meeting/${meetingId}/realtime/transcripts`,
data
);
};
export const getRealtimeMeetingSessionStatus = (meetingId: number) => {
return http.get<{ code: string; data: RealtimeMeetingSessionStatus; msg: string }>(

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

@ -70,7 +70,6 @@ const AiModels: React.FC = () => {
const provider = Form.useWatch("provider", form);
const isDefaultChecked = Form.useWatch("isDefaultChecked", form);
const isLocalProvider = String(provider || "").toLowerCase() === "custom";
const isTencentProvider = String(provider || "").toLowerCase() === "tencent";
const isPlatformAdmin = useMemo(() => {
const profileStr = sessionStorage.getItem("userProfile");
@ -134,17 +133,11 @@ const AiModels: React.FC = () => {
setEditingId(record.id);
const speakerModel = record.mediaConfig?.speakerModel;
const svThreshold = record.mediaConfig?.svThreshold;
const tencentAppId = record.mediaConfig?.tencentAppId;
const tencentSecretId = record.mediaConfig?.tencentSecretId;
const tencentSecretKey = record.mediaConfig?.tencentSecretKey;
form.setFieldsValue({
...record,
modelType: record.modelType,
speakerModel,
svThreshold,
tencentAppId,
tencentSecretId,
tencentSecretKey,
isDefaultChecked: record.isDefault === 1,
statusChecked: record.status === 1,
});
@ -269,12 +262,6 @@ const AiModels: React.FC = () => {
speakerModel: values.speakerModel,
svThreshold: values.svThreshold,
}
: activeType === "ASR" && isTencentProvider
? {
tencentAppId: values.tencentAppId,
tencentSecretId: values.tencentSecretId,
tencentSecretKey: values.tencentSecretKey,
}
: undefined,
temperature: values.temperature,
topP: values.topP,
@ -537,20 +524,16 @@ const AiModels: React.FC = () => {
</Col>
</Row>
{!isTencentProvider && (
<>
<Form.Item name="baseUrl" label="Base URL" rules={[{required: true, message: "请输入 Base URL"}]}>
<Input placeholder="https://api.example.com"/>
</Form.Item>
<Form.Item name="baseUrl" label="Base URL" rules={[{ required: true, message: "请输入 Base URL" }]}>
<Input placeholder="https://api.example.com" />
</Form.Item>
<Form.Item
name="apiKey"
label="API Key"
>
<Input.Password/>
</Form.Item>
</>
)}
<Form.Item
name="apiKey"
label="API Key"
>
<Input.Password />
</Form.Item>
{(activeType === "LLM" || isLocalProvider) && (
<Form.Item label="连通性测试">
@ -573,10 +556,7 @@ const AiModels: React.FC = () => {
<Form.Item
name="modelCode"
noStyle
rules={activeType === "LLM" || isTencentProvider ? [{
required: true,
message: "请输入或选择模型名称"
}] : []}
rules={activeType === "LLM" ? [{ required: true, message: "请输入或选择模型名称" }] : []}
>
<AutoComplete
style={{ width: "calc(100% - 100px)" }}
@ -594,11 +574,9 @@ const AiModels: React.FC = () => {
<Input allowClear placeholder="可选择或自定义输入模型名称" />
</AutoComplete>
</Form.Item>
{!isTencentProvider && (
<Button icon={<SyncOutlined spin={fetchLoading}/>} onClick={handleFetchRemote} style={{width: 100}}>
刷新
</Button>
)}
<Button icon={<SyncOutlined spin={fetchLoading} />} onClick={handleFetchRemote} style={{ width: 100 }}>
刷新
</Button>
</Space.Compact>
</Form.Item>
@ -631,38 +609,6 @@ const AiModels: React.FC = () => {
</Row>
)}
{activeType === "ASR" && isTencentProvider && (
<Row gutter={16}>
<Col span={12}>
<Form.Item
name="tencentAppId"
label="App ID"
rules={[{required: true, message: "请输入 App ID"}]}
>
<Input/>
</Form.Item>
</Col>
<Col span={12}>
<Form.Item
name="tencentSecretId"
label="Secret ID"
rules={[{required: true, message: "请输入 Secret ID"}]}
>
<Input/>
</Form.Item>
</Col>
<Col span={24}>
<Form.Item
name="tencentSecretKey"
label="Secret Key"
rules={[{required: true, message: "请输入 Secret Key"}]}
>
<Input.Password/>
</Form.Item>
</Col>
</Row>
)}
{activeType === "LLM" && (
<>
<Form.Item name="apiPath" label="API 路径" initialValue="/v1/chat/completions">

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

@ -18,6 +18,7 @@ import dayjs from "dayjs";
import PageHeader from "../../components/shared/PageHeader";
import PageContainer from "@/components/shared/PageContainer";
import {
appendRealtimeTranscripts,
completeRealtimeMeeting,
getMeetingDetail,
getRealtimeMeetingSessionStatus,
@ -27,6 +28,7 @@ import {
type MeetingTranscriptVO,
type MeetingVO,
type RealtimeMeetingSessionStatus,
type RealtimeTranscriptItemDTO,
type RealtimeSocketSessionVO,
} from "../../api/business/meeting";
const { Text, Title } = Typography;
@ -42,8 +44,6 @@ type WsMessage = {
data?: {
text?: string;
is_final?: boolean;
sentence_id?: number;
sentence_key?: string;
start?: number;
end?: number;
speaker_id?: string;
@ -58,8 +58,6 @@ type WsMessage = {
type TranscriptCard = {
id: string;
sentenceKey?: string;
sentenceId?: number;
speakerName: string;
userId?: string | number;
text: string;
@ -68,16 +66,6 @@ type TranscriptCard = {
final: boolean;
};
type NormalizedWsMessage = {
sentenceKey?: string;
text: string;
isFinal: boolean;
sentenceId?: number;
speaker?: WsSpeaker;
startTime?: number;
endTime?: number;
};
type RealtimeMeetingSessionDraft = {
meetingId: number;
meetingTitle: string;
@ -135,18 +123,12 @@ function resolveSpeaker(speaker?: WsSpeaker) {
return { speakerId: "spk_0", speakerName: "Unknown", userId: undefined };
}
if (typeof speaker === "string") {
const normalized = speaker.trim();
if (!normalized || normalized === "-1") {
return {speakerId: "spk_0", speakerName: "Unknown", userId: undefined};
}
return {speakerId: normalized, speakerName: normalized, userId: undefined};
return { speakerId: speaker, speakerName: speaker, userId: undefined };
}
const rawUserId = speaker.user_id === null || speaker.user_id === undefined ? undefined : String(speaker.user_id).trim();
const speakerId = rawUserId && rawUserId !== "-1" ? rawUserId : "spk_0";
return {
speakerId,
speakerName: speaker.name?.trim() || (speakerId !== "spk_0" ? speakerId : "Unknown"),
userId: speakerId === "spk_0" ? undefined : rawUserId,
speakerId: speaker.user_id ? String(speaker.user_id) : "spk_0",
speakerName: speaker.name || (speaker.user_id ? String(speaker.user_id) : "Unknown"),
userId: speaker.user_id,
};
}
@ -177,29 +159,17 @@ function toMs(value?: number) {
return Math.round(value * 1000);
}
function buildTranscriptCardId(sentenceKey?: string, sentenceId?: number) {
if (sentenceKey) {
return sentenceKey;
}
if (sentenceId === undefined || sentenceId === null) {
return `live-${Date.now()}-${Math.random()}`;
}
return `sentence-${sentenceId}`;
}
function buildRealtimeProxyWsUrl(socketSession: RealtimeSocketSessionVO) {
const protocol = window.location.protocol === "https:" ? "wss" : "ws";
return `${protocol}://${window.location.host}${socketSession.path}?sessionToken=${encodeURIComponent(socketSession.sessionToken)}`;
}
function normalizeWsMessage(payload: WsMessage): NormalizedWsMessage | null {
function normalizeWsMessage(payload: WsMessage) {
if (payload.type === "partial" || payload.type === "segment") {
const data = payload.data || {};
return {
text: data.text || "",
isFinal: payload.type === "segment" || !!data.is_final,
sentenceKey: data.sentence_key,
sentenceId: data.sentence_id,
speaker: {
name: data.speaker_name,
user_id: data.user_id ?? data.speaker_id,
@ -242,6 +212,7 @@ export function RealtimeAsrSession() {
const [audioLevel, setAudioLevel] = useState(0);
const [elapsedSeconds, setElapsedSeconds] = useState(0);
const [sessionStatus, setSessionStatus] = useState<RealtimeMeetingSessionStatus | null>(null);
const transcriptRef = useRef<HTMLDivElement | null>(null);
const wsRef = useRef<WebSocket | null>(null);
const audioContextRef = useRef<AudioContext | null>(null);
@ -252,7 +223,6 @@ export function RealtimeAsrSession() {
const completeOnceRef = useRef(false);
const startedAtRef = useRef<number | null>(null);
const sessionStartedRef = useRef(false);
const elapsedOffsetRef = useRef(0);
const finalTranscriptCount = transcripts.length;
const totalTranscriptChars = useMemo(
@ -311,7 +281,7 @@ export function RealtimeAsrSession() {
}
setTranscripts(
(transcriptRes.data.data || []).map((item: MeetingTranscriptVO) => ({
id: `persisted-${item.id}`,
id: String(item.id),
speakerName: item.speakerName || item.speakerId || "发言人",
text: item.content,
startTime: item.startTime,
@ -331,12 +301,12 @@ export function RealtimeAsrSession() {
useEffect(() => {
if (!recording) {
setElapsedSeconds(elapsedOffsetRef.current);
setElapsedSeconds(0);
return;
}
const timer = window.setInterval(() => {
if (startedAtRef.current) {
setElapsedSeconds(elapsedOffsetRef.current + Math.floor((Date.now() - startedAtRef.current) / 1000));
setElapsedSeconds(Math.floor((Date.now() - startedAtRef.current) / 1000));
}
}, 1000);
return () => window.clearInterval(timer);
@ -356,7 +326,7 @@ export function RealtimeAsrSession() {
}
const token = localStorage.getItem("accessToken");
if (wsRef.current?.readyState === WebSocket.OPEN) {
wsRef.current.close();
wsRef.current.send(JSON.stringify({ is_speaking: false }));
}
fetch(`/api/biz/meeting/${meetingId}/realtime/pause`, {
method: "POST",
@ -390,44 +360,6 @@ export function RealtimeAsrSession() {
setAudioLevel(0);
};
const closeFrontendSocket = async (sendStop: boolean) => {
const socket = wsRef.current;
if (sendStop && socket?.readyState === WebSocket.OPEN) {
socket.send(JSON.stringify({type: "stop"}));
await new Promise((resolve) => window.setTimeout(resolve, 150));
}
socket?.close();
wsRef.current = null;
sessionStartedRef.current = false;
};
const upsertTranscriptCard = (normalized: NormalizedWsMessage, speaker: ReturnType<typeof resolveSpeaker>) => {
setTranscripts((prev) => {
const next = [...prev];
const cardId = buildTranscriptCardId(normalized.sentenceKey, normalized.sentenceId);
const nextCard: TranscriptCard = {
id: cardId,
sentenceKey: normalized.sentenceKey,
sentenceId: normalized.sentenceId,
speakerName: speaker.speakerName,
userId: speaker.userId,
text: normalized.text,
startTime: normalized.startTime,
endTime: normalized.endTime,
final: true,
};
if (normalized.sentenceKey || (normalized.sentenceId !== undefined && normalized.sentenceId !== null)) {
const index = next.findIndex((item) => item.id === cardId);
if (index >= 0) {
next[index] = {...next[index], ...nextCard};
return next;
}
}
next.push(nextCard);
return next;
});
};
const handleFatalRealtimeError = async (errorMessage: string) => {
setConnecting(false);
setRecording(false);
@ -437,8 +369,6 @@ export function RealtimeAsrSession() {
wsRef.current = null;
await shutdownAudioPipeline();
startedAtRef.current = null;
elapsedOffsetRef.current = 0;
setElapsedSeconds(0);
message.error(errorMessage);
};
@ -489,6 +419,26 @@ export function RealtimeAsrSession() {
processor.connect(audioContext.destination);
};
const saveFinalTranscript = async (normalized: {
text: string;
speaker?: WsSpeaker;
startTime?: number;
endTime?: number;
}) => {
if (!normalized.text || !meetingId) {
return;
}
const speaker = resolveSpeaker(normalized.speaker);
const item: RealtimeTranscriptItemDTO = {
speakerId: speaker.speakerId,
speakerName: speaker.speakerName,
content: normalized.text,
startTime: normalized.startTime,
endTime: normalized.endTime,
};
await appendRealtimeTranscripts(meetingId, [item]);
};
const handlePause = async () => {
if (!meetingId || pausing || finishing || (!recording && !connecting)) {
return;
@ -501,19 +451,20 @@ export function RealtimeAsrSession() {
setPausing(true);
setStatusText("暂停识别中...");
try {
if (recording && startedAtRef.current) {
elapsedOffsetRef.current += Math.floor((Date.now() - startedAtRef.current) / 1000);
if (wsRef.current?.readyState === WebSocket.OPEN) {
wsRef.current.send(JSON.stringify({ is_speaking: false }));
}
const pauseRes = await pauseRealtimeMeeting(meetingId);
await closeFrontendSocket(false);
wsRef.current?.close();
wsRef.current = null;
sessionStartedRef.current = false;
await shutdownAudioPipeline();
const pauseRes = await pauseRealtimeMeeting(meetingId);
setSessionStatus(pauseRes.data.data);
setRecording(false);
setConnecting(false);
startedAtRef.current = null;
setElapsedSeconds(elapsedOffsetRef.current);
setStatusText("已暂停，可继续会议并等待说话人修正");
message.success("实时会议已暂停");
setStatusText(pauseRes.data.data?.hasTranscript ? "已暂停，可继续识别" : "已暂停，当前还没有转录内容");
message.success("实时识别已暂停");
} catch (error) {
setStatusText("暂停失败");
message.error(error instanceof Error ? error.message : "暂停实时识别失败");
@ -596,9 +547,21 @@ export function RealtimeAsrSession() {
const speaker = resolveSpeaker(normalized.speaker);
if (normalized.isFinal) {
upsertTranscriptCard(normalized, speaker);
setTranscripts((prev) => [
...prev,
{
id: `${Date.now()}-${Math.random()}`,
speakerName: speaker.speakerName,
userId: speaker.userId,
text: normalized.text,
startTime: normalized.startTime,
endTime: normalized.endTime,
final: true,
},
]);
setStreamingText("");
setStreamingSpeaker("Unknown");
void saveFinalTranscript(normalized);
} else {
setStreamingText(normalized.text);
setStreamingSpeaker(speaker.speakerName);
@ -639,7 +602,12 @@ export function RealtimeAsrSession() {
setFinishing(true);
setStatusText("结束会议中...");
await closeFrontendSocket(true);
if (wsRef.current?.readyState === WebSocket.OPEN) {
wsRef.current.send(JSON.stringify({ is_speaking: false }));
}
wsRef.current?.close();
wsRef.current = null;
sessionStartedRef.current = false;
await shutdownAudioPipeline();