diff --git a/main/manager-api/src/main/java/xiaozhi/common/exception/RenExceptionHandler.java b/main/manager-api/src/main/java/xiaozhi/common/exception/RenExceptionHandler.java index f6b19d31..e8eaf462 100644 --- a/main/manager-api/src/main/java/xiaozhi/common/exception/RenExceptionHandler.java +++ b/main/manager-api/src/main/java/xiaozhi/common/exception/RenExceptionHandler.java @@ -71,10 +71,15 @@ public class RenExceptionHandler { List allErrors = ex.getBindingResult().getAllErrors(); String errorMsg = allErrors.stream() .filter(Objects::nonNull) - .map(DefaultMessageSourceResolvable::getDefaultMessage) + .map(err -> { + String msg = err.getDefaultMessage(); + return (msg != null && !msg.trim().isEmpty()) ? msg : null; + }) + .filter(Objects::nonNull) .findFirst() - .orElse(""); - return new Result().error(400, errorMsg); + .orElse("请求参数错误！"); + + return new Result().error(ErrorCode.PARAM_VALUE_NULL, errorMsg); } } \ No newline at end of file diff --git a/main/manager-api/src/main/java/xiaozhi/common/utils/DateUtils.java b/main/manager-api/src/main/java/xiaozhi/common/utils/DateUtils.java index bfa7afed..800733c1 100644 --- a/main/manager-api/src/main/java/xiaozhi/common/utils/DateUtils.java +++ b/main/manager-api/src/main/java/xiaozhi/common/utils/DateUtils.java @@ -21,6 +21,8 @@ public class DateUtils { * 时间格式(yyyy-MM-dd HH:mm:ss) */ public final static String DATE_TIME_PATTERN = "yyyy-MM-dd HH:mm:ss"; + public final static String DATE_TIME_MILLIS_PATTERN = "yyyy-MM-dd HH:mm:ss.SSS"; + /** * 日期格式化 日期格式为：yyyy-MM-dd @@ -63,6 +65,19 @@ public class DateUtils { return null; } + + public static String getDateTimeNow() { + return getDateTimeNow(DATE_TIME_PATTERN); + } + + public static String getDateTimeNow(String pattern) { + return format(new Date(), pattern); + } + + public static String millsToSecond(long mills) { + return String.format("%.3f", mills / 1000.0); + } + /** * 获取简短的时间字符串：10秒前返回刚刚，多少秒前，几小时前，超过一周返回年月日时分秒 * @param date diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/ServerSideManageController.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/ServerSideManageController.java new file mode 100644 index 00000000..a9664ced --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/ServerSideManageController.java @@ -0,0 +1,120 @@ +package xiaozhi.modules.sys.controller; + +import java.util.Arrays; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.TimeUnit; + +import org.apache.commons.lang3.StringUtils; +import org.apache.shiro.authz.annotation.RequiresPermissions; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.socket.WebSocketHttpHeaders; + +import com.fasterxml.jackson.core.JsonProcessingException; +import com.fasterxml.jackson.databind.DeserializationFeature; +import com.fasterxml.jackson.databind.ObjectMapper; + +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.tags.Tag; +import jakarta.validation.Valid; +import lombok.AllArgsConstructor; +import xiaozhi.common.annotation.LogOperation; +import xiaozhi.common.constant.Constant; +import xiaozhi.common.exception.RenException; +import xiaozhi.common.utils.Result; +import xiaozhi.modules.sys.dto.EmitSeverActionDTO; +import xiaozhi.modules.sys.dto.ServerActionPayloadDTO; +import xiaozhi.modules.sys.dto.ServerActionResponseDTO; +import xiaozhi.modules.sys.enums.ServerActionEnum; +import xiaozhi.modules.sys.service.SysParamsService; +import xiaozhi.modules.sys.utils.WebSocketClientManager; + +/** + * 服务端管理控制器 + */ +@RestController +@RequestMapping("admin/server") +@Tag(name = "服务端管理") +@AllArgsConstructor +public class ServerSideManageController { + private final SysParamsService sysParamsService; + private static final ObjectMapper objectMapper; + static { + objectMapper = new ObjectMapper(); + objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false); + } + + @Operation(summary = "获取Ws服务端列表") + @GetMapping("/server-list") + @RequiresPermissions("sys:role:superAdmin") + public Result> getWsServerList() { + String cachedWs = sysParamsService.getValue(Constant.SERVER_WEBSOCKET, true); + return new Result>().ok(Arrays.asList(cachedWs.split(";"))); + } + + @Operation(summary = "通知python服务端更新配置") + @PostMapping("/emit-action") + @LogOperation("通知python服务端更新配置") + @RequiresPermissions("sys:role:superAdmin") + public Result emitServerAction(@RequestBody @Valid EmitSeverActionDTO emitSeverActionDTO) { + if (emitSeverActionDTO.getAction() == null) { + throw new RenException("无效服务端操作"); + } + String wsText = sysParamsService.getValue(Constant.SERVER_WEBSOCKET, true); + if (StringUtils.isBlank(wsText)) { + throw new RenException("未配置服务端WebSocket地址"); + } + String targetWs = emitSeverActionDTO.getTargetWs(); + String[] wsList = wsText.split(";"); + // 找到需要发起的 + if (StringUtils.isBlank(targetWs) || !Arrays.asList(wsList).contains(targetWs)) { + throw new RenException("目标WebSocket地址不存在"); + } + return new Result().ok(emitServerActionByWs(targetWs, emitSeverActionDTO.getAction())); + } + + private Boolean emitServerActionByWs(String targetWsUri, ServerActionEnum actionEnum) { + if (StringUtils.isBlank(targetWsUri) || actionEnum == null) { + return false; + } + String serverSK = sysParamsService.getValue(Constant.SERVER_SECRET, true); + WebSocketHttpHeaders headers = new WebSocketHttpHeaders(); + headers.add("device-id", UUID.randomUUID().toString()); + headers.add("client-id", UUID.randomUUID().toString()); + + try (WebSocketClientManager client = new WebSocketClientManager.Builder() + .connectTimeout(3, TimeUnit.SECONDS) + .maxSessionDuration(120, TimeUnit.SECONDS) + .uri(targetWsUri) + .headers(headers) + .build()) { + // 如果连接成功则发送一个json数据包并等待服务端响应 + client.sendJson( + ServerActionPayloadDTO.build( + actionEnum, + Map.of("secret", serverSK))); + // 等待服务端响应并持续监听信息 + client.listener((jsonText) -> { + if (StringUtils.isBlank(jsonText)) { + return false; + } + try { + ServerActionResponseDTO response = objectMapper.readValue(jsonText, ServerActionResponseDTO.class); + Boolean isSuccess = ServerActionResponseDTO.isSuccess(response); + return isSuccess; + } catch (JsonProcessingException e) { + return false; + } + }); + } catch (Exception e) { + // 捕获全部错误，由全局异常处理器返回 + throw new RenException("WebSocket连接失败或连接超时"); + } + return true; + } +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/SysParamsController.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/SysParamsController.java index e1774280..4860cf1c 100644 --- a/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/SysParamsController.java +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/controller/SysParamsController.java @@ -111,7 +111,7 @@ public class SysParamsController { /** * 验证WebSocket地址列表 - * + * * @param urls WebSocket地址列表，以分号分隔 * @return 验证结果 */ @@ -143,6 +143,19 @@ public class SysParamsController { } } + @PostMapping("/delete") + @Operation(summary = "删除") + @LogOperation("删除") + @RequiresPermissions("sys:role:superAdmin") + public Result delete(@RequestBody String[] ids) { + // 效验数据 + AssertUtils.isArrayEmpty(ids, "id"); + + sysParamsService.delete(ids); + configService.getConfig(false); + return new Result(); + } + /** * 验证OTA地址 */ @@ -182,17 +195,4 @@ public class SysParamsController { throw new RenException("OTA接口验证失败：" + e.getMessage()); } } - - @PostMapping("/delete") - @Operation(summary = "删除") - @LogOperation("删除") - @RequiresPermissions("sys:role:superAdmin") - public Result delete(@RequestBody String[] ids) { - // 效验数据 - AssertUtils.isArrayEmpty(ids, "id"); - - sysParamsService.delete(ids); - configService.getConfig(false); - return new Result(); - } } diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/EmitSeverActionDTO.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/EmitSeverActionDTO.java new file mode 100644 index 00000000..888ef4ec --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/EmitSeverActionDTO.java @@ -0,0 +1,29 @@ +package xiaozhi.modules.sys.dto; + +import io.swagger.v3.oas.annotations.media.Schema; +import jakarta.validation.constraints.NotEmpty; +import jakarta.validation.constraints.NotNull; +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; +import xiaozhi.modules.sys.enums.ServerActionEnum; + +/** + * 发送python服务端操作DTO + * + * @Author CAIXYPROMISE + * @since 2025/5/15 4:55 + */ +@Data +@NoArgsConstructor +@AllArgsConstructor +public class EmitSeverActionDTO +{ + @Schema(description = "目标ws地址") + @NotEmpty(message = "目标ws地址不能为空") + private String targetWs; + + @Schema(description = "指定操作") + @NotNull(message = "操作不能为空") + private ServerActionEnum action; +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/ServerActionPayloadDTO.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/ServerActionPayloadDTO.java new file mode 100644 index 00000000..cb158d25 --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/ServerActionPayloadDTO.java @@ -0,0 +1,39 @@ +package xiaozhi.modules.sys.dto; + +import lombok.Data; +import xiaozhi.modules.sys.enums.ServerActionEnum; + +import java.util.Map; + +/** + * 服务端动作DTO + * + * @Author CAIXYPROMISE + * @since 2025/5/15 5:10 + */ +@Data +public class ServerActionPayloadDTO +{ + /** + * 类型（智控台发往服务端的都是server） + */ + private String type; + /** + * 动作 + */ + private ServerActionEnum action; + /** + * 内容 + */ + private Map content; + + public static ServerActionPayloadDTO build(ServerActionEnum action, Map content) { + ServerActionPayloadDTO serverActionPayloadDTO = new ServerActionPayloadDTO(); + serverActionPayloadDTO.setAction(action); + serverActionPayloadDTO.setContent(content); + serverActionPayloadDTO.setType("server"); + return serverActionPayloadDTO; + } + // 私有化 + private ServerActionPayloadDTO() {} +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/ServerActionResponseDTO.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/ServerActionResponseDTO.java new file mode 100644 index 00000000..6286b74a --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/dto/ServerActionResponseDTO.java @@ -0,0 +1,34 @@ +package xiaozhi.modules.sys.dto; + +import lombok.Data; +import xiaozhi.modules.sys.enums.ServerActionResponseEnum; + +import java.util.Map; + +/** + * 服务端动作响应体 + */ +@Data +public class ServerActionResponseDTO +{ + private ServerActionResponseEnum status; + private String message; + private String type; + private Map content; // 后续这个字段可以移除，并把这个类作为基类，针对业务写自己的content类型 + public static final String DEFAULT_TYPE_FORM_SERVER = "server"; + + public static Boolean isSuccess(ServerActionResponseDTO actionResponseDTO) { + System.out.println(actionResponseDTO); + if (actionResponseDTO == null) { + return false; + } + if (actionResponseDTO.getStatus() == null || !actionResponseDTO.getStatus().equals(ServerActionResponseEnum.SUCCESS)) { + return false; + } + Object actionType = actionResponseDTO.getContent().get("action"); + if (actionType == null) { + return false; + } + return actionResponseDTO.getType() != null && actionResponseDTO.getType().equals(DEFAULT_TYPE_FORM_SERVER); + } +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/enums/ServerActionEnum.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/enums/ServerActionEnum.java new file mode 100644 index 00000000..33babf45 --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/enums/ServerActionEnum.java @@ -0,0 +1,43 @@ +package xiaozhi.modules.sys.enums; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonValue; +import xiaozhi.common.exception.RenException; + +/** + * 服务端动作枚举 + * + * @Author CAIXYPROMISE + * @since 2025/5/15 4:57 + */ +public enum ServerActionEnum +{ + RESTART("restart"), + UPDATE_CONFIG("update_config"); + + private final String value; + + ServerActionEnum(String value) + { + this.value = value; + } + + @JsonValue + public String getValue() + { + return value; + } + + @JsonCreator + public static ServerActionEnum fromValue(String value) + { + for (ServerActionEnum action : ServerActionEnum.values()) + { + if (action.value.equalsIgnoreCase(value)) + { + return action; + } + } + return null; + } +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/enums/ServerActionResponseEnum.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/enums/ServerActionResponseEnum.java new file mode 100644 index 00000000..e1b54a66 --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/enums/ServerActionResponseEnum.java @@ -0,0 +1,49 @@ +package xiaozhi.modules.sys.enums; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonValue; +import lombok.Getter; +import org.apache.commons.lang3.StringUtils; + +/** + * 服务端调用响应枚举 + * + * @Author CAIXYPROMISE + * @since 2025/5/15 5:50 + */ +public enum ServerActionResponseEnum +{ + SUCCESS("success"), FAIL("fail"); + private final String value; + + ServerActionResponseEnum(String value) { + this.value = value; + } + + @JsonValue + public String getValue() + { + return value; + } + + @JsonCreator + public static ServerActionResponseEnum fromValue(String value) { + ServerActionResponseEnum byValue = getByValue(value); + if (byValue == null) { + throw new IllegalArgumentException("Unknown enum value: " + value); + } + return byValue; + } + + public static ServerActionResponseEnum getByValue(String value) { + if (StringUtils.isBlank(value)) { + return null; + } + for (ServerActionResponseEnum action : ServerActionResponseEnum.values()) { + if (action.value.equals(value)) { + return action; + } + } + return null; + } +} diff --git a/main/manager-api/src/main/java/xiaozhi/modules/sys/utils/WebSocketClientManager.java b/main/manager-api/src/main/java/xiaozhi/modules/sys/utils/WebSocketClientManager.java new file mode 100644 index 00000000..0c49aead --- /dev/null +++ b/main/manager-api/src/main/java/xiaozhi/modules/sys/utils/WebSocketClientManager.java @@ -0,0 +1,315 @@ +package xiaozhi.modules.sys.utils; + +import com.fasterxml.jackson.databind.ObjectMapper; +import lombok.extern.slf4j.Slf4j; +import org.springframework.util.StopWatch; +import org.springframework.web.socket.*; +import org.springframework.web.socket.client.standard.StandardWebSocketClient; +import org.springframework.web.socket.handler.AbstractWebSocketHandler; +import xiaozhi.common.utils.DateUtils; + +import java.io.Closeable; +import java.io.IOException; +import java.net.URI; +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; +import java.util.function.Predicate; + +/** + * WebSocketClientResource：支持 try-with-resources 模式 + */ +@Slf4j +public class WebSocketClientManager implements Closeable +{ + private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper(); + + // 全局回调线程池 + private static final ExecutorService CALLBACK_EXECUTOR = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors(), new ThreadFactory() { + private final AtomicInteger cnt = new AtomicInteger(); + + public Thread newThread(Runnable r) + { + Thread t = new Thread(r, "ws-callback-" + cnt.getAndIncrement()); + t.setDaemon(true); + return t; + } + }); + + private volatile WebSocketSession session; + private final BlockingQueue textMessageQueue; + private final BlockingQueue binaryMessageQueue; + private final CompletableFuture errorFuture; + private final long maxSessionDuration; + private final TimeUnit maxSessionDurationUnit; + + private volatile Consumer onText; + private volatile Consumer onBinary; + private volatile Consumer onError; + + private final String uri; + private final WebSocketHttpHeaders headers; + private final long connectTimeout; + private final TimeUnit connectUnit; + private final int queueCapacity; + + // 私有构造，仅由 Builder 调用 + private WebSocketClientManager(Builder b) { + this.uri = b.uri; + this.headers = b.headers != null ? b.headers : new WebSocketHttpHeaders(); + this.connectTimeout = b.connectTimeout; + this.connectUnit = b.connectUnit; + this.maxSessionDuration = b.maxSessionDuration; + this.maxSessionDurationUnit = b.maxSessionDurationUnit; + this.queueCapacity = b.queueCapacity; + this.textMessageQueue = new LinkedBlockingQueue<>(queueCapacity); + this.binaryMessageQueue = new LinkedBlockingQueue<>(queueCapacity); + this.errorFuture = new CompletableFuture<>(); + } + + public static WebSocketClientManager build(Builder b) throws InterruptedException, ExecutionException, TimeoutException, IOException { + WebSocketClientManager ws = new WebSocketClientManager(b); + StandardWebSocketClient client = new StandardWebSocketClient(); + CompletableFuture future = client.execute(ws.new InternalHandler(b.uri), b.headers, URI.create(b.uri)); + WebSocketSession sess = future.get(b.connectTimeout, b.connectUnit); + if (sess == null || !sess.isOpen()) + { + throw new IOException("握手失败或会话未打开"); + } + ws.session = sess; + return ws; + } + + /** + * 发送 Text + */ + public void sendText(String text) throws IOException { + session.sendMessage(new TextMessage(text)); + } + + public void sendBinary(byte[] data) throws IOException { + session.sendMessage(new BinaryMessage(data)); + } + + public void sendJson(Object payload) throws IOException { + String json = OBJECT_MAPPER.writeValueAsString(payload); + session.sendMessage(new TextMessage(json)); + } + + + + private List listenerCustom( + BlockingQueue queue, + Predicate predicate) + throws InterruptedException, TimeoutException, ExecutionException + { + List collected = new ArrayList<>(); + long deadline = System.currentTimeMillis() + maxSessionDurationUnit.toMillis(maxSessionDuration); + + while (true) { + if (errorFuture.isDone()) { + errorFuture.get(); + } + + long remaining = deadline - System.currentTimeMillis(); + if (remaining <= 0) { + throw new TimeoutException("等待批量消息超时"); + } + + T msg = queue.poll(remaining, TimeUnit.MILLISECONDS); + if (msg == null) { + throw new TimeoutException("等待批量消息超时"); + } + + collected.add(msg); + if (predicate.test(msg)) { + break; + } + } + close(); + return collected; + } + + /** + * 同步接收多条消息，直到 predicate 为 true 或超时抛异常； + * @return 返回监听期间的所有消息列表 + */ + public List listener(Predicate predicate) + throws InterruptedException, TimeoutException, ExecutionException + { + return listenerCustom(textMessageQueue, predicate); + } + + public List listenerBinary(Predicate predicate) + throws InterruptedException, TimeoutException, ExecutionException + { + return listenerCustom(binaryMessageQueue, predicate); + } + + /** + * 注册文本回调 + */ + public WebSocketClientManager onText(Consumer c) { + this.onText = c; + return this; + } + + /** + * 注册二进制回调 + */ + public WebSocketClientManager onBinary(Consumer c) { + this.onBinary = c; + return this; + } + + /** + * 注册错误回调 + */ + public WebSocketClientManager onError(Consumer c) { + this.onError = c; + return this; + } + + /** + * 关闭会话，try-with-resources / finally 自动调用 + */ + @Override + public void close() { + try { + if (session != null && session.isOpen()) { + session.close(CloseStatus.NORMAL); + } + } + catch (IOException ignored) {} + textMessageQueue.clear(); + binaryMessageQueue.clear(); + errorFuture.completeExceptionally(new IOException("WebSocket 已关闭")); + } + + private class InternalHandler extends AbstractWebSocketHandler { + private final String targetUri; + private final StopWatch stopWatch; + + InternalHandler(String targetUri) { + this.targetUri = targetUri; + this.stopWatch = new StopWatch(); + } + + /** + * 连接建立时回调 + */ + @Override + public void afterConnectionEstablished(WebSocketSession session) { + // 保存会话 + WebSocketClientManager.this.session = session; + this.stopWatch.start(); + log.info("ws连接成功, 目标URI: {}, 连接时间: {}", targetUri, DateUtils.getDateTimeNow(DateUtils.DATE_TIME_MILLIS_PATTERN)); + } + + /** + * 处理文本消息 + */ + @Override + protected void handleTextMessage(WebSocketSession session, TextMessage message) throws Exception { + String payload = message.getPayload(); + // 入队 + textMessageQueue.offer(payload); + // 回调用户注册的 onText + if (onText != null) { + CALLBACK_EXECUTOR.submit(() -> onText.accept(payload)); + } + } + + /** + * 处理二进制消息 + */ + @Override + protected void handleBinaryMessage(WebSocketSession session, BinaryMessage message) throws Exception { + ByteBuffer buf = message.getPayload(); + byte[] data = new byte[buf.remaining()]; + buf.get(data); + // 入队 + binaryMessageQueue.offer(data); + // 回调用户注册的 onBinary + if (onBinary != null) { + CALLBACK_EXECUTOR.submit(() -> onBinary.accept(data)); + } + } + + /** + * 传输错误时回调 + */ + @Override + public void handleTransportError(WebSocketSession session, Throwable exception) throws Exception { + super.handleTransportError(session, exception); + // 保持原有逻辑：完成 errorFuture、回调 onError、关闭会话、异步通知连接失败 + errorFuture.completeExceptionally(exception); + if (onError != null) { + CALLBACK_EXECUTOR.submit(() -> onError.accept(exception)); + } + session.close(CloseStatus.SERVER_ERROR); + } + + /** + * 连接关闭时回调 + */ + @Override + public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception { + super.afterConnectionClosed(session, status); + if (stopWatch.isRunning()) { + stopWatch.stop(); + } + log.info("ws连接关闭, 目标URI: {}, 关闭时间: {}, 连接总时长: {}s", + targetUri, DateUtils.getDateTimeNow(DateUtils.DATE_TIME_MILLIS_PATTERN), DateUtils.millsToSecond(stopWatch.getTotalTimeMillis())); + } + } + + public static class Builder { + private String uri; // 目标 WS URI + private long connectTimeout = 3; // 请求连接等待时间 + private TimeUnit connectUnit = TimeUnit.SECONDS; // 请求连接等待时间单位 + private long maxSessionDuration = 5; // 最大连线时间，默认5秒 + private TimeUnit maxSessionDurationUnit = TimeUnit.SECONDS; // 最大连线时间单位 + private int queueCapacity = 100; // 消息队列容量 + private WebSocketHttpHeaders headers; // 请求头 + + /** + * 目标 WS URI + */ + public Builder uri(String uri) { + this.uri = Objects.requireNonNull(uri); + return this; + } + + public Builder headers(WebSocketHttpHeaders h) { + this.headers = h; + return this; + } + + public Builder connectTimeout(long t, TimeUnit u) { + this.connectTimeout = t; + this.connectUnit = u; + return this; + } + + public Builder maxSessionDuration(long t, TimeUnit u) { + this.maxSessionDuration = t; + this.maxSessionDurationUnit = u; + return this; + } + + public Builder queueCapacity(int c) { + this.queueCapacity = c; + return this; + } + + public WebSocketClientManager build() throws InterruptedException, ExecutionException, TimeoutException, IOException { + return WebSocketClientManager.build(this); + } + + } +} diff --git a/main/manager-web/src/apis/module/admin.js b/main/manager-web/src/apis/module/admin.js index 92c344e2..27c1bc3e 100644 --- a/main/manager-web/src/apis/module/admin.js +++ b/main/manager-web/src/apis/module/admin.js @@ -130,5 +130,37 @@ export default { }) }).send() }, + // 获取ws服务端列表 + getWsServerList(params, callback) { + RequestService.sendRequest() + .url(`${getServiceUrl()}/admin/server/server-list`) + .method('GET') + .success((res) => { + RequestService.clearRequestTime() + callback(res) + }) + .networkFail((err) => { + console.error('获取ws服务端列表失败:', err) + RequestService.reAjaxFun(() => { + this.getWsServerList(params, callback) + }) + }).send(); + }, + // 发送ws服务器动作指令 + sendWsServerAction(data, callback) { + RequestService.sendRequest() + .url(`${getServiceUrl()}/admin/server/emit-action`) + .method('POST') + .data(data) + .success((res) => { + RequestService.clearRequestTime() + callback(res) + }) + .networkFail((err) => { + RequestService.reAjaxFun(() => { + this.sendWsServerAction(data, callback) + }) + }).send(); + } } diff --git a/main/manager-web/src/components/HeaderBar.vue b/main/manager-web/src/components/HeaderBar.vue index 842441ce..3843151a 100644 --- a/main/manager-web/src/components/HeaderBar.vue +++ b/main/manager-web/src/components/HeaderBar.vue @@ -52,6 +52,9 @@ 供应器管理 + + 服务端管理 + @@ -145,6 +148,9 @@ export default { goProviderManagement() { this.$router.push('/provider-management') }, + goServerSideManagement() { + this.$router.push('/server-side-management') + }, // 获取用户信息 fetchUserInfo() { userApi.getUserInfo(({ data }) => { diff --git a/main/manager-web/src/router/index.js b/main/manager-web/src/router/index.js index 49ee7e4c..7e304f76 100644 --- a/main/manager-web/src/router/index.js +++ b/main/manager-web/src/router/index.js @@ -80,6 +80,18 @@ const routes = [ title: '参数管理' } }, + + { + path: '/server-side-management', + name: 'ServerSideManager', + component: function () { + return import('../views/ServerSideManager.vue') + }, + meta: { + requiresAuth: true, + title: '服务端管理' + } + }, { path: '/ota-management', name: 'OtaManagement', diff --git a/main/manager-web/src/views/ServerSideManager.vue b/main/manager-web/src/views/ServerSideManager.vue new file mode 100644 index 00000000..973fcd6f --- /dev/null +++ b/main/manager-web/src/views/ServerSideManager.vue @@ -0,0 +1,475 @@ + + + + + diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 0abb3cec..74b612fe 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -22,6 +22,7 @@ from core.utils.util import ( initialize_modules, check_vad_update, check_asr_update, + filter_sensitive_info, ) from concurrent.futures import ThreadPoolExecutor, TimeoutError from core.handle.sendAudioHandle import sendAudioMessage @@ -273,9 +274,10 @@ class ConnectionHandler: await self.websocket.send( json.dumps( { - "type": "server_response", + "type": "server", "status": "success", "message": "服务器重启中...", + "content": {"action": "restart"}, } ) ) @@ -302,9 +304,10 @@ class ConnectionHandler: await self.websocket.send( json.dumps( { - "type": "server_response", + "type": "server", "status": "error", "message": f"Restart failed: {str(e)}", + "content": {"action": "restart"}, } ) ) @@ -1087,37 +1090,3 @@ class ConnectionHandler: break except Exception as e: self.logger.bind(tag=TAG).error(f"超时检查任务出错: {e}") - - -def filter_sensitive_info(config: dict) -> dict: - """ - 过滤配置中的敏感信息 - Args: - config: 原始配置字典 - Returns: - 过滤后的配置字典 - """ - sensitive_keys = [ - "api_key", - "personal_access_token", - "access_token", - "token", - "secret", - "access_key_secret", - "secret_key", - ] - - def _filter_dict(d: dict) -> dict: - filtered = {} - for k, v in d.items(): - if any(sensitive in k.lower() for sensitive in sensitive_keys): - filtered[k] = "***" - elif isinstance(v, dict): - filtered[k] = _filter_dict(v) - elif isinstance(v, list): - filtered[k] = [_filter_dict(i) if isinstance(i, dict) else i for i in v] - else: - filtered[k] = v - return filtered - - return _filter_dict(copy.deepcopy(config)) diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index 79874f1c..9b4981a1 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -1,7 +1,7 @@ import json from core.handle.abortHandle import handleAbortMessage from core.handle.helloHandle import handleHelloMessage -from core.utils.util import remove_punctuation_and_length +from core.utils.util import remove_punctuation_and_length, filter_sensitive_info from core.handle.receiveAudioHandle import startToChat, handleAudioMessage from core.handle.sendAudioHandle import send_stt_message, send_tts_message from core.handle.iotHandle import handleIotDescriptors, handleIotStatus @@ -13,17 +13,20 @@ TAG = __name__ async def handleTextMessage(conn, message): """处理文本消息""" - conn.logger.bind(tag=TAG).info(f"收到文本消息：{message}") try: msg_json = json.loads(message) if isinstance(msg_json, int): + conn.logger.bind(tag=TAG).info(f"收到文本消息：{message}") await conn.websocket.send(message) return if msg_json["type"] == "hello": + conn.logger.bind(tag=TAG).info(f"收到hello消息：{message}") await handleHelloMessage(conn, msg_json) elif msg_json["type"] == "abort": + conn.logger.bind(tag=TAG).info(f"收到abort消息：{message}") await handleAbortMessage(conn) elif msg_json["type"] == "listen": + conn.logger.bind(tag=TAG).info(f"收到listen消息：{message}") if "mode" in msg_json: conn.client_listen_mode = msg_json["mode"] conn.logger.bind(tag=TAG).debug( @@ -64,11 +67,16 @@ async def handleTextMessage(conn, message): # 否则需要LLM对文字内容进行答复 await startToChat(conn, text) elif msg_json["type"] == "iot": + conn.logger.bind(tag=TAG).info(f"收到iot消息：{message}") if "descriptors" in msg_json: asyncio.create_task(handleIotDescriptors(conn, msg_json["descriptors"])) if "states" in msg_json: asyncio.create_task(handleIotStatus(conn, msg_json["states"])) elif msg_json["type"] == "server": + # 记录日志时过滤敏感信息 + conn.logger.bind(tag=TAG).info( + f"收到服务器消息：{filter_sensitive_info(msg_json)}" + ) # 如果配置是从API读取的，则需要验证secret if not conn.read_config_from_api: return @@ -95,9 +103,10 @@ async def handleTextMessage(conn, message): await conn.websocket.send( json.dumps( { - "type": "config_update_response", + "type": "server", "status": "error", "message": "无法获取服务器实例", + "content": {"action": "update_config"}, } ) ) @@ -107,9 +116,10 @@ async def handleTextMessage(conn, message): await conn.websocket.send( json.dumps( { - "type": "config_update_response", + "type": "server", "status": "error", "message": "更新服务器配置失败", + "content": {"action": "update_config"}, } ) ) @@ -119,9 +129,10 @@ async def handleTextMessage(conn, message): await conn.websocket.send( json.dumps( { - "type": "config_update_response", + "type": "server", "status": "success", "message": "配置更新成功", + "content": {"action": "update_config"}, } ) ) @@ -130,9 +141,10 @@ async def handleTextMessage(conn, message): await conn.websocket.send( json.dumps( { - "type": "config_update_response", + "type": "server", "status": "error", "message": f"更新配置失败: {str(e)}", + "content": {"action": "update_config"}, } ) ) diff --git a/main/xiaozhi-server/core/utils/util.py b/main/xiaozhi-server/core/utils/util.py index 9ab77a5a..0b7e46a1 100644 --- a/main/xiaozhi-server/core/utils/util.py +++ b/main/xiaozhi-server/core/utils/util.py @@ -9,6 +9,7 @@ import opuslib_next from pydub import AudioSegment from typing import Dict, Any from core.utils import tts, llm, intent, memory, vad, asr +import copy TAG = __name__ emoji_map = { @@ -319,7 +320,7 @@ def initialize_modules( modules["memory"] = memory.create_instance( memory_type, config["Memory"][select_memory_module], - config.get('summaryMemory', None), + config.get("summaryMemory", None), ) logger.bind(tag=TAG).info(f"初始化组件: memory成功 {select_memory_module}") @@ -956,3 +957,37 @@ def check_asr_update(before_config, new_config): ) update_asr = current_asr_type != new_asr_type return update_asr + + +def filter_sensitive_info(config: dict) -> dict: + """ + 过滤配置中的敏感信息 + Args: + config: 原始配置字典 + Returns: + 过滤后的配置字典 + """ + sensitive_keys = [ + "api_key", + "personal_access_token", + "access_token", + "token", + "secret", + "access_key_secret", + "secret_key", + ] + + def _filter_dict(d: dict) -> dict: + filtered = {} + for k, v in d.items(): + if any(sensitive in k.lower() for sensitive in sensitive_keys): + filtered[k] = "***" + elif isinstance(v, dict): + filtered[k] = _filter_dict(v) + elif isinstance(v, list): + filtered[k] = [_filter_dict(i) if isinstance(i, dict) else i for i in v] + else: + filtered[k] = v + return filtered + + return _filter_dict(copy.deepcopy(config)) diff --git a/main/xiaozhi-server/core/websocket_server.py b/main/xiaozhi-server/core/websocket_server.py index 441fd4e1..3f9e2fd7 100644 --- a/main/xiaozhi-server/core/websocket_server.py +++ b/main/xiaozhi-server/core/websocket_server.py @@ -82,11 +82,13 @@ class WebSocketServer: if new_config is None: self.logger.bind(tag=TAG).error("获取新配置失败") return False - + self.logger.bind(tag=TAG).info(f"获取新配置成功") # 检查 VAD 和 ASR 类型是否需要更新 update_vad = check_vad_update(self.config, new_config) update_asr = check_asr_update(self.config, new_config) - + self.logger.bind(tag=TAG).info( + f"检查VAD和ASR类型是否需要更新: {update_vad} {update_asr}" + ) # 更新配置 self.config = new_config # 重新初始化组件 @@ -114,7 +116,7 @@ class WebSocketServer: self._intent = modules["intent"] if "memory" in modules: self._memory = modules["memory"] - + self.logger.bind(tag=TAG).info(f"更新配置任务执行完毕") return True except Exception as e: self.logger.bind(tag=TAG).error(f"更新服务器配置失败: {str(e)}")