Merge pull request #1337 from xinnan-tech/connect-server-and-api

智控台控制xiaozhi-server重启和更新配置

main/manager-api/src/main/java/xiaozhi/common/exception/RenExceptionHandler.java

@@ -71,10 +71,15 @@ public class RenExceptionHandler {
         List<ObjectError> 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("");
+                .orElse("请求参数错误！");
-        return new Result<Void>().error(400, errorMsg);
+
+        return new Result<Void>().error(ErrorCode.PARAM_VALUE_NULL, errorMsg);
     }
 
 }

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

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<List<String>> getWsServerList() {
+        String cachedWs = sysParamsService.getValue(Constant.SERVER_WEBSOCKET, true);
+        return new Result<List<String>>().ok(Arrays.asList(cachedWs.split(";")));
+    }
+
+    @Operation(summary = "通知python服务端更新配置")
+    @PostMapping("/emit-action")
+    @LogOperation("通知python服务端更新配置")
+    @RequiresPermissions("sys:role:superAdmin")
+    public Result<Boolean> 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<Boolean>().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;
+    }
+}

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<Void> delete(@RequestBody String[] ids) {
+        // 效验数据
+        AssertUtils.isArrayEmpty(ids, "id");
+
+        sysParamsService.delete(ids);
+        configService.getConfig(false);
+        return new Result<Void>();
+    }
+
     /**
      * 验证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<Void> delete(@RequestBody String[] ids) {
-        // 效验数据
-        AssertUtils.isArrayEmpty(ids, "id");
-
-        sysParamsService.delete(ids);
-        configService.getConfig(false);
-        return new Result<Void>();
-    }
 }

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;
+}

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<String, Object> content;
+
+    public static ServerActionPayloadDTO build(ServerActionEnum action, Map<String, Object> content) {
+        ServerActionPayloadDTO serverActionPayloadDTO = new ServerActionPayloadDTO();
+        serverActionPayloadDTO.setAction(action);
+        serverActionPayloadDTO.setContent(content);
+        serverActionPayloadDTO.setType("server");
+        return serverActionPayloadDTO;
+    }
+    // 私有化
+    private ServerActionPayloadDTO() {}
+}

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<String, Object> 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);
+    }
+}

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;
+    }
+}

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;
+    }
+}

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<String> textMessageQueue;
+    private final BlockingQueue<byte[]> binaryMessageQueue;
+    private final CompletableFuture<Void> errorFuture;
+    private final long maxSessionDuration;
+    private final TimeUnit maxSessionDurationUnit;
+
+    private volatile Consumer<String> onText;
+    private volatile Consumer<byte[]> onBinary;
+    private volatile Consumer<Throwable> 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<WebSocketSession> 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 <T> List<T> listenerCustom(
+            BlockingQueue<T> queue,
+            Predicate<T> predicate)
+            throws InterruptedException, TimeoutException, ExecutionException
+    {
+        List<T> 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<String> listener(Predicate<String> predicate)
+            throws InterruptedException, TimeoutException, ExecutionException
+    {
+        return listenerCustom(textMessageQueue, predicate);
+    }
+
+    public List<byte[]> listenerBinary(Predicate<byte[]> predicate)
+            throws InterruptedException, TimeoutException, ExecutionException
+    {
+        return listenerCustom(binaryMessageQueue, predicate);
+    }
+
+    /**
+     * 注册文本回调
+     */
+    public WebSocketClientManager onText(Consumer<String> c) {
+        this.onText = c;
+        return this;
+    }
+
+    /**
+     * 注册二进制回调
+     */
+    public WebSocketClientManager onBinary(Consumer<byte[]> c) {
+        this.onBinary = c;
+        return this;
+    }
+
+    /**
+     * 注册错误回调
+     */
+    public WebSocketClientManager onError(Consumer<Throwable> 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);
+        }
+
+    }
+}

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();
+    }
 
 }

main/manager-web/src/components/HeaderBar.vue

@@ -52,6 +52,9 @@
             <el-dropdown-item @click.native="goProviderManagement">
               供应器管理
             </el-dropdown-item>
+            <el-dropdown-item @click.native="goServerSideManagement">
+              服务端管理
+            </el-dropdown-item>
           </el-dropdown-menu>
         </el-dropdown>
       </div>
@@ -145,6 +148,9 @@ export default {
     goProviderManagement() {
       this.$router.push('/provider-management')
     },
+    goServerSideManagement() {
+      this.$router.push('/server-side-management')
+    },
     // 获取用户信息
     fetchUserInfo() {
       userApi.getUserInfo(({ data }) => {

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',

main/manager-web/src/views/ServerSideManager.vue

@@ -0,0 +1,475 @@
+<template>
+  <div class="welcome">
+    <HeaderBar />
+
+    <div class="operation-bar">
+      <h2 class="page-title">服务端管理</h2>
+    </div>
+
+    <div class="main-wrapper">
+      <div class="content-panel">
+        <div class="content-area">
+          <el-card class="params-card" shadow="never">
+            <el-table ref="paramsTable" :data="paramsList" class="transparent-table" v-loading="loading"
+              element-loading-text="拼命加载中" element-loading-spinner="el-icon-loading"
+              element-loading-background="rgba(255, 255, 255, 0.7)" :header-cell-class-name="headerCellClassName">
+              <el-table-column label="选择" align="center" width="120">
+                <template slot-scope="scope">
+                  <el-checkbox v-model="scope.row.selected"></el-checkbox>
+                </template>
+              </el-table-column>
+              <el-table-column label="ws地址" prop="address" align="center"></el-table-column>
+              <el-table-column label="操作" prop="operator" align="center" show-overflow-tooltip>
+                <template slot-scope="scope">
+                  <el-button size="medium" type="text" @click="emitAction(scope.row, actionMap.restart)">重启</el-button>
+                  <el-button size="medium" type="text"
+                    @click="emitAction(scope.row, actionMap.update_config)">更新配置</el-button>
+                </template>
+              </el-table-column>
+            </el-table>
+          </el-card>
+        </div>
+      </div>
+    </div>
+
+
+    <el-footer>
+      <version-footer />
+    </el-footer>
+  </div>
+</template>
+
+<script>
+import Api from "@/apis/api";
+import HeaderBar from "@/components/HeaderBar.vue";
+import ParamDialog from "@/components/ParamDialog.vue";
+import VersionFooter from "@/components/VersionFooter.vue";
+
+export default {
+  components: { HeaderBar, ParamDialog, VersionFooter },
+  data() {
+    return {
+      paramsList: [],
+      actionMap: {
+        restart: {
+          value: 'restart',
+          title: "重启服务端",
+          message: "确定要重启服务端吗？",
+          confirmText: "重启",
+        },
+        update_config: {
+          value: 'update_config',
+          title: "更新配置",
+          message: "确定要更新配置吗？",
+          confirmText: "更新",
+        }
+      },
+      currentPage: 1,
+      loading: false,
+      pageSize: 10,
+      pageSizeOptions: [10, 20, 50, 100],
+      total: 0,
+      dialogVisible: false,
+      dialogTitle: "新增参数",
+      isAllSelected: false,
+      sensitive_keys: ["api_key", "personal_access_token", "access_token", "token", "secret", "access_key_secret", "secret_key"],
+      paramForm: {
+        id: null,
+        paramCode: "",
+        paramValue: "",
+        remark: ""
+      },
+    };
+  },
+  created() {
+    this.fetchParams();
+  },
+
+  computed: {
+    pageCount() {
+      return Math.ceil(this.total / this.pageSize);
+    },
+    visiblePages() {
+      const pages = [];
+      const maxVisible = 3;
+      let start = Math.max(1, this.currentPage - 1);
+      let end = Math.min(this.pageCount, start + maxVisible - 1);
+
+      if (end - start + 1 < maxVisible) {
+        start = Math.max(1, end - maxVisible + 1);
+      }
+
+      for (let i = start; i <= end; i++) {
+        pages.push(i);
+      }
+      return pages;
+    },
+  },
+  methods: {
+    handlePageSizeChange(val) {
+      this.pageSize = val;
+      this.currentPage = 1;
+      this.fetchParams();
+    },
+    fetchParams() {
+      this.loading = true;
+      Api.admin.getWsServerList(
+        {},
+        ({ data }) => {
+          this.loading = false;
+          if (data.code === 0) {
+            this.paramsList = data.data.map(item => ({ address: item }));
+            this.total = data.data.length;
+          } else {
+            this.$message.error({
+              message: data.msg || '获取参数列表失败',
+              showClose: true
+            });
+          }
+        }
+      );
+    },
+    emitAction(rowItem, actionItem) {
+      if (actionItem === undefined || rowItem.address === undefined) {
+        return;
+      }
+      // 弹开询问框
+      this.$confirm(actionItem.message, actionItem.title, {
+        confirmButtonText: actionItem.confirmText, // 确认按钮文本
+      }).then(() => {
+        // 用户点击了确认按钮
+        Api.admin.sendWsServerAction({
+          targetWs: rowItem.address,
+          action: actionItem.value
+        }, ({ data }) => {
+          if (data.code !== 0) {
+            this.$message.error({
+              message: data.msg || '操作失败',
+              showClose: true
+            });
+            return;
+          }
+          this.$message.success({
+            message: `${actionItem.title}成功`,
+            showClose: true
+          })
+        })
+      })
+    },
+    headerCellClassName({ columnIndex }) {
+      if (columnIndex === 0) {
+        return "custom-selection-header";
+      }
+      return "";
+    }
+  },
+};
+</script>
+
+<style lang="scss" scoped>
+.welcome {
+  min-width: 900px;
+  min-height: 506px;
+  height: 100vh;
+  display: flex;
+  position: relative;
+  flex-direction: column;
+  background-size: cover;
+  background: linear-gradient(to bottom right, #dce8ff, #e4eeff, #e6cbfd) center;
+  -webkit-background-size: cover;
+  -o-background-size: cover;
+  overflow: hidden;
+}
+
+.main-wrapper {
+  margin: 5px 22px;
+  border-radius: 15px;
+  min-height: calc(100vh - 24vh);
+  height: auto;
+  max-height: 80vh;
+  box-shadow: 0 2px 12px rgba(0, 0, 0, 0.1);
+  position: relative;
+  background: rgba(237, 242, 255, 0.5);
+  display: flex;
+  flex-direction: column;
+}
+
+.operation-bar {
+  display: flex;
+  justify-content: space-between;
+  align-items: center;
+  padding: 16px 24px;
+}
+
+.page-title {
+  font-size: 24px;
+  margin: 0;
+}
+
+.right-operations {
+  display: flex;
+  gap: 10px;
+  margin-left: auto;
+}
+
+.search-input {
+  width: 240px;
+}
+
+.btn-search {
+  background: linear-gradient(135deg, #6b8cff, #a966ff);
+  border: none;
+  color: white;
+}
+
+.content-panel {
+  flex: 1;
+  display: flex;
+  overflow: hidden;
+  height: 100%;
+  border-radius: 15px;
+  background: transparent;
+  border: 1px solid #fff;
+}
+
+.content-area {
+  flex: 1;
+  height: 100%;
+  min-width: 600px;
+  overflow: auto;
+  background-color: white;
+  display: flex;
+  flex-direction: column;
+}
+
+.params-card {
+  background: white;
+  flex: 1;
+  display: flex;
+  flex-direction: column;
+  border: none;
+  box-shadow: none;
+  overflow: hidden;
+
+  ::v-deep .el-card__body {
+    padding: 15px;
+    display: flex;
+    flex-direction: column;
+    flex: 1;
+    overflow: hidden;
+  }
+}
+
+.table_bottom {
+  display: flex;
+  justify-content: space-between;
+  align-items: center;
+  margin-top: 10px;
+  padding-bottom: 10px;
+}
+
+.ctrl_btn {
+  display: flex;
+  gap: 8px;
+  padding-left: 26px;
+
+  .el-button {
+    min-width: 72px;
+    height: 32px;
+    padding: 7px 12px 7px 10px;
+    font-size: 12px;
+    border-radius: 4px;
+    line-height: 1;
+    font-weight: 500;
+    border: none;
+    transition: all 0.3s ease;
+    box-shadow: 0 2px 6px rgba(0, 0, 0, 0.1);
+
+    &:hover {
+      transform: translateY(-1px);
+      box-shadow: 0 4px 8px rgba(0, 0, 0, 0.15);
+    }
+  }
+
+  .el-button--primary {
+    background: #5f70f3;
+    color: white;
+  }
+
+  .el-button--danger {
+    background: #fd5b63;
+    color: white;
+  }
+}
+
+:deep(.transparent-table) {
+  background: white;
+  flex: 1;
+  width: 100%;
+  display: flex;
+  flex-direction: column;
+
+  .el-table__body-wrapper {
+    flex: 1;
+    overflow-y: auto;
+    max-height: none !important;
+  }
+
+  .el-table__header-wrapper {
+    flex-shrink: 0;
+  }
+
+  .el-table__header th {
+    background: white !important;
+    color: black;
+  }
+
+  &::before {
+    display: none;
+  }
+
+  .el-table__body tr {
+    background-color: white;
+
+    td {
+      border-top: 1px solid rgba(0, 0, 0, 0.04);
+      border-bottom: 1px solid rgba(0, 0, 0, 0.04);
+    }
+  }
+}
+
+
+:deep(.el-checkbox__inner) {
+  background-color: #eeeeee !important;
+  border-color: #cccccc !important;
+}
+
+:deep(.el-checkbox__inner:hover) {
+  border-color: #cccccc !important;
+}
+
+:deep(.el-checkbox__input.is-checked .el-checkbox__inner) {
+  background-color: #5f70f3 !important;
+  border-color: #5f70f3 !important;
+}
+
+@media (min-width: 1144px) {
+  .table_bottom {
+    display: flex;
+    justify-content: space-between;
+    align-items: center;
+    margin-top: 40px;
+  }
+
+  :deep(.transparent-table) {
+    .el-table__body tr {
+      td {
+        padding-top: 16px;
+        padding-bottom: 16px;
+      }
+
+      &+tr {
+        margin-top: 10px;
+      }
+    }
+  }
+}
+
+:deep(.el-table .el-button--text) {
+  color: #7079aa;
+}
+
+:deep(.el-table .el-button--text:hover) {
+  color: #5a64b5;
+}
+
+.el-button--success {
+  background: #5bc98c;
+  color: white;
+}
+
+:deep(.el-table .cell) {
+  white-space: nowrap;
+  overflow: hidden;
+  text-overflow: ellipsis;
+}
+
+.page-size-select {
+  width: 100px;
+  margin-right: 10px;
+
+  :deep(.el-input__inner) {
+    height: 32px;
+    line-height: 32px;
+    border-radius: 4px;
+    border: 1px solid #e4e7ed;
+    background: #dee7ff;
+    color: #606266;
+    font-size: 14px;
+  }
+
+  :deep(.el-input__suffix) {
+    right: 6px;
+    width: 15px;
+    height: 20px;
+    display: flex;
+    justify-content: center;
+    align-items: center;
+    top: 6px;
+    border-radius: 4px;
+  }
+
+  :deep(.el-input__suffix-inner) {
+    display: flex;
+    align-items: center;
+    justify-content: center;
+    width: 100%;
+  }
+
+  :deep(.el-icon-arrow-up:before) {
+    content: "";
+    display: inline-block;
+    border-left: 6px solid transparent;
+    border-right: 6px solid transparent;
+    border-top: 9px solid #606266;
+    position: relative;
+    transform: rotate(0deg);
+    transition: transform 0.3s;
+  }
+}
+
+:deep(.el-table) {
+  .el-table__body-wrapper {
+    transition: height 0.3s ease;
+  }
+}
+
+.el-table {
+  --table-max-height: calc(100vh - 40vh);
+  max-height: var(--table-max-height);
+
+  .el-table__body-wrapper {
+    max-height: calc(var(--table-max-height) - 40px);
+  }
+}
+
+:deep(.el-loading-mask) {
+  background-color: rgba(255, 255, 255, 0.6) !important;
+  backdrop-filter: blur(2px);
+}
+
+:deep(.el-loading-spinner .circular) {
+  width: 28px;
+  height: 28px;
+}
+
+:deep(.el-loading-spinner .path) {
+  stroke: #6b8cff;
+}
+
+:deep(.el-loading-text) {
+  color: #6b8cff !important;
+  font-size: 14px;
+  margin-top: 8px;
+}
+</style>

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))

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"},
                             }
                         )
                     )

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))

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)}")