diff --git a/plugins/maibot_adapter/maibot_maintenance.py b/plugins/maibot_adapter/maibot_maintenance.py new file mode 100644 index 0000000..5fb31ec --- /dev/null +++ b/plugins/maibot_adapter/maibot_maintenance.py @@ -0,0 +1,219 @@ +#!/usr/bin/env python +# -*- coding: utf-8 -*- +""" +MaiBot 维护脚本（放在 maibot_adapter 插件目录下）。 + +用途： +1. 一键查看远端 MaiBot 容器状态； +2. 一键拉取关键日志（重点看 timing_gate / no_reply / 发送失败）； +3. 一键重启 MaiBot 容器； +4. 一键查看远端核心配置片段； +5. 一键检查 WebUI 健康接口。 + +示例： +python plugins/maibot_adapter/maibot_maintenance.py status +python plugins/maibot_adapter/maibot_maintenance.py logs --since 20m +python plugins/maibot_adapter/maibot_maintenance.py restart +python plugins/maibot_adapter/maibot_maintenance.py health +python plugins/maibot_adapter/maibot_maintenance.py config +""" + +from __future__ import annotations + +import argparse +import sys +from typing import Iterable + +import paramiko + + +def _run_ssh_cmd( + host: str, + username: str, + password: str, + command: str, + timeout: int = 60, +) -> tuple[int, str, str]: + """ + 在远端通过 SSH 执行命令，并返回退出码、stdout、stderr。 + + 说明： + 1. 统一封装 SSH 调用，方便所有子命令复用； + 2. 使用 utf-8 ignore 解码，避免日志中出现异常字符时脚本直接崩溃； + 3. 返回退出码用于调用方判断是否成功并决定是否继续后续步骤。 + """ + client = paramiko.SSHClient() + client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + try: + client.connect(hostname=host, username=username, password=password, timeout=20) + stdin, stdout, stderr = client.exec_command(command, timeout=timeout) + out_text = stdout.read().decode("utf-8", "ignore") + err_text = stderr.read().decode("utf-8", "ignore") + exit_status = stdout.channel.recv_exit_status() + return exit_status, out_text, err_text + finally: + client.close() + + +def _print_header(title: str) -> None: + """输出分隔标题，方便终端快速定位每段结果。""" + print(f"\n===== {title} =====") + + +def _safe_print_lines(lines: Iterable[str]) -> None: + """ + 安全逐行输出，兼容 Windows 控制台编码。 + + 说明： + 1. 某些日志包含 emoji 或控制字符，直接 print 可能触发编码异常； + 2. 这里做 gbk/backslashreplace 兜底，确保维护脚本不会因打印失败中断。 + """ + for line in lines: + try: + print(line) + except UnicodeEncodeError: + print(line.encode("gbk", "backslashreplace").decode("gbk", "ignore")) + + +def cmd_status(args: argparse.Namespace) -> int: + """查看远端容器运行状态。""" + command = ( + "docker ps --format " + "'table {{.Names}}\\t{{.Image}}\\t{{.Status}}' " + f"| (head -n 1; grep '{args.container}' || true)" + ) + code, out_text, err_text = _run_ssh_cmd(args.host, args.user, args.password, command) + _print_header("MaiBot 容器状态") + print(out_text.strip() or "(无输出)") + if err_text.strip(): + _print_header("stderr") + print(err_text.strip()) + return code + + +def cmd_logs(args: argparse.Namespace) -> int: + """查看远端关键日志，默认聚焦回复决策与发送链路。""" + command = f"docker logs --since {args.since} {args.container} 2>&1" + code, out_text, err_text = _run_ssh_cmd(args.host, args.user, args.password, command, timeout=120) + text = f"{out_text}\n{err_text}".strip() + if not text: + _print_header("关键日志") + print("(无输出)") + return code + + # 这里聚焦你当前排障最常用的关键词，尽量减少噪声。 + keywords = ( + "Timing Gate", + "no_reply", + "continue", + "wait", + "reply", + "回复", + "SendService", + "无法发送", + "发送成功", + "发送失败", + "Updated platform_map", + "Bridge received", + ) + + _print_header(f"关键日志（since={args.since}）") + filtered = [line for line in text.splitlines() if any(k in line for k in keywords)] + if not filtered: + print("(未匹配到关键日志，可尝试增大 --since)") + return code + + _safe_print_lines(filtered) + return code + + +def cmd_restart(args: argparse.Namespace) -> int: + """重启远端容器并回显最新状态。""" + _print_header("执行重启") + code, out_text, err_text = _run_ssh_cmd( + args.host, + args.user, + args.password, + f"docker restart {args.container}", + timeout=120, + ) + print(out_text.strip() or "(无输出)") + if err_text.strip(): + _print_header("stderr") + print(err_text.strip()) + # 重启后再补一段状态，避免用户还要再跑一次 status。 + cmd_status(args) + return code + + +def cmd_health(args: argparse.Namespace) -> int: + """检查远端 WebUI 健康接口。""" + url = f"http://{args.host}:{args.webui_port}/api/webui/health" + command = f"curl -sS --max-time 8 '{url}'" + code, out_text, err_text = _run_ssh_cmd(args.host, args.user, args.password, command, timeout=20) + _print_header(f"WebUI 健康检查: {url}") + print(out_text.strip() or "(无输出)") + if err_text.strip(): + _print_header("stderr") + print(err_text.strip()) + return code + + +def cmd_config(args: argparse.Namespace) -> int: + """查看远端核心配置片段，便于快速确认关键参数。""" + command = ( + "docker exec {container} sh -lc " + "\"echo '[bot_config.toml]'; " + "sed -n '1,220p' /MaiMBot/config/bot_config.toml; " + "echo; " + "echo '[model_config.toml]'; " + "sed -n '1,220p' /MaiMBot/config/model_config.toml\"" + ).format(container=args.container) + code, out_text, err_text = _run_ssh_cmd(args.host, args.user, args.password, command, timeout=120) + _print_header("远端配置快照") + _safe_print_lines(out_text.splitlines()) + if err_text.strip(): + _print_header("stderr") + _safe_print_lines(err_text.splitlines()) + return code + + +def build_parser() -> argparse.ArgumentParser: + """构建命令行参数解析器。""" + parser = argparse.ArgumentParser(description="MaiBot 远端维护脚本") + parser.add_argument("--host", default="192.168.2.240", help="远端服务器地址") + parser.add_argument("--user", default="root", help="SSH 用户名") + parser.add_argument("--password", default="lw123456", help="SSH 密码") + parser.add_argument("--container", default="maibot-core-lite", help="MaiBot 容器名") + parser.add_argument("--webui-port", type=int, default=18001, help="WebUI 对外端口") + + subparsers = parser.add_subparsers(dest="action", required=True) + + status_parser = subparsers.add_parser("status", help="查看容器状态") + status_parser.set_defaults(func=cmd_status) + + logs_parser = subparsers.add_parser("logs", help="查看关键日志") + logs_parser.add_argument("--since", default="20m", help="日志时间窗口，例如 10m / 1h") + logs_parser.set_defaults(func=cmd_logs) + + restart_parser = subparsers.add_parser("restart", help="重启容器") + restart_parser.set_defaults(func=cmd_restart) + + health_parser = subparsers.add_parser("health", help="检查 WebUI 健康接口") + health_parser.set_defaults(func=cmd_health) + + config_parser = subparsers.add_parser("config", help="查看远端核心配置") + config_parser.set_defaults(func=cmd_config) + + return parser + + +def main() -> int: + """脚本入口。""" + parser = build_parser() + args = parser.parse_args() + return int(args.func(args)) + + +if __name__ == "__main__": + sys.exit(main())