OpenClaw 实用工具集
🛠️ 开发辅助工具
1. 配置文件生成器
#!/bin/bash
# generate-config.sh
echo "OpenClaw 配置文件生成器"
echo "========================"
# 基础配置
read -p "请输入 App ID: " app_id
read -p "请输入 App Secret: " app_secret
read -p "请选择域名 (feishu/lark): " domain
read -p "请输入监听端口 (默认3000): " port
port=${port:-3000}
domain=${domain:-feishu}
# 生成配置文件
cat > config.yaml << EOF
server:
port: ${port}
host: 0.0.0.0
channels:
feishu:
enabled: true
appId: "${app_id}"
appSecret: "${app_secret}"
domain: "${domain}"
connectionMode: "websocket"
dmPolicy: "pairing"
groupPolicy: "allowlist"
requireMention: true
mediaMaxMb: 30
renderMode: "auto"
logging:
level: info
file: "./logs/openclaw.log"
maxSize: 100MB
maxFiles: 5
plugins:
autoUpdate: true
registry: "https://registry.npmjs.org"
EOF
echo "✅ 配置文件已生成: config.yaml"
2. 环境健康检查脚本
#!/bin/bash
# health-check.sh
echo "🔍 OpenClaw 环境健康检查"
echo "========================"
# 检查 Node.js
echo "📋 检查 Node.js..."
if command -v node &> /dev/null; then
echo "✅ Node.js 版本: $(node --version)"
else
echo "❌ Node.js 未安装"
exit 1
fi
# 检查 npm
echo "📋 检查 npm..."
if command -v npm &> /dev/null; then
echo "✅ npm 版本: $(npm --version)"
else
echo "❌ npm 未安装"
fi
# 检查 OpenClaw CLI
echo "📋 检查 OpenClaw CLI..."
if command -v openclaw &> /dev/null; then
echo "✅ OpenClaw 版本: $(openclaw --version)"
else
echo "❌ OpenClaw CLI 未安装"
fi
# 检查配置文件
echo "📋 检查配置文件..."
if [ -f "config.yaml" ]; then
echo "✅ 配置文件存在"
else
echo "❌ 配置文件不存在"
fi
# 检查端口占用
echo "📋 检查端口占用..."
port=${1:-3000}
if lsof -Pi :${port} -sTCP:LISTEN -t >/dev/null ; then
echo "⚠️ 端口 ${port} 已被占用"
else
echo "✅ 端口 ${port} 可用"
fi
echo "✅ 健康检查完成"
3. 日志分析工具
#!/usr/bin/env python3
# log-analyzer.py
import re
import json
from collections import defaultdict, Counter
from datetime import datetime, timedelta
class LogAnalyzer:
def __init__(self, log_file):
self.log_file = log_file
self.patterns = {
'error': r'"level":"error"',
'warn': r'"level":"warn"',
'info': r'"level":"info"',
'request': r'"method":"(\w+)"',
'response_time': r'"duration":(\d+)'
}
def analyze(self):
stats = {
'total_lines': 0,
'levels': Counter(),
'methods': Counter(),
'slow_requests': 0,
'errors': 0
}
with open(self.log_file, 'r') as f:
for line in f:
stats['total_lines'] += 1
# 分析日志级别
for level, pattern in self.patterns.items():
if re.search(pattern, line):
stats['levels'][level] += 1
# 分析响应时间
time_match = re.search(r'"duration":(\d+)', line)
if time_match:
duration = int(time_match.group(1))
if duration > 1000: # 超过1秒的慢请求
stats['slow_requests'] += 1
return stats
def generate_report(self):
stats = self.analyze()
print("📊 OpenClaw 日志分析报告")
print("=" * 40)
print(f"总日志行数: {stats['total_lines']}")
print(f"错误数量: {stats['levels']['error']}")
print(f"警告数量: {stats['levels']['warn']}")
print(f"慢请求 (>1s): {stats['slow_requests']}")
print("\n日志级别分布:")
for level, count in stats['levels'].most_common():
percentage = (count / stats['total_lines']) * 100
print(f" {level}: {count} ({percentage:.1f}%)")
if __name__ == "__main__":
import sys
if len(sys.argv) != 2:
print("使用方法: python log-analyzer.py <log_file>")
sys.exit(1)
analyzer = LogAnalyzer(sys.argv[1])
analyzer.generate_report()
🎯 测试工具
1. 消息发送测试工具
// message-tester.js
const axios = require('axios');
class MessageTester {
constructor(config) {
this.baseUrl = config.baseUrl || 'http://localhost:3000';
this.token = config.token;
}
async sendMessage(message, userId = 'test_user') {
try {
const response = await axios.post(`${this.baseUrl}/api/messages`, {
user_id: userId,
content: message,
timestamp: new Date().toISOString()
}, {
headers: {
'Authorization': `Bearer ${this.token}`,
'Content-Type': 'application/json'
}
});
console.log('✅ 消息发送成功');
console.log('响应:', response.data);
return response.data;
} catch (error) {
console.error('❌ 消息发送失败:', error.response?.data || error.message);
throw error;
}
}
async runTestSuite() {
const testCases = [
'你好，测试一下',
'今天的天气怎么样？',
'/help',
'帮我查询订单状态'
];
console.log('🚀 开始测试套件...');
for (let i = 0; i < testCases.length; i++) {
console.log(`\n测试用例 ${i + 1}: "${testCases[i]}"`);
try {
await this.sendMessage(testCases[i]);
await new Promise(resolve => setTimeout(resolve, 1000)); // 间隔1秒
} catch (error) {
console.log('测试失败，继续下一个...');
}
}
console.log('\n✅ 测试套件执行完成');
}
}
// 使用示例
const tester = new MessageTester({
baseUrl: 'http://localhost:3000',
token: 'your_test_token'
});
tester.runTestSuite();
2. 性能压力测试
#!/bin/bash
# stress-test.sh
CONCURRENT_USERS=${1:-10}
DURATION=${2:-60}
TARGET_URL=${3:-"http://localhost:3000/api/health"}
echo "🏋️ OpenClaw 性能压力测试"
echo "并发用户数: ${CONCURRENT_USERS}"
echo "测试时长: ${DURATION}秒"
echo "目标地址: ${TARGET_URL}"
echo "========================"
# 使用 wrk 进行压力测试
wrk -t4 -c${CONCURRENT_USERS} -d${DURATION}s ${TARGET_URL}
# 或使用 ab (Apache Bench)
# ab -n 1000 -c ${CONCURRENT_USERS} ${TARGET_URL}
📊 监控告警工具
1. 系统监控脚本
#!/bin/bash
# system-monitor.sh
THRESHOLD_CPU=80
THRESHOLD_MEM=80
THRESHOLD_DISK=90
check_system_health() {
# CPU 使用率
cpu_usage=$(top -bn1 | grep "Cpu(s)" | awk '{print $2}' | cut -d'%' -f1)
# 内存使用率
mem_usage=$(free | grep Mem | awk '{printf("%.0f", $3/$2 * 100.0)}')
# 磁盘使用率
disk_usage=$(df -h / | awk 'NR==2 {print $5}' | cut -d'%' -f1)
echo "📊 系统健康状态检查"
echo "CPU 使用率: ${cpu_usage}%"
echo "内存使用率: ${mem_usage}%"
echo "磁盘使用率: ${disk_usage}%"
# 告警检查
if (( $(echo "$cpu_usage > $THRESHOLD_CPU" | bc -l) )); then
echo "⚠️ CPU 使用率过高: ${cpu_usage}%"
fi
if [ "$mem_usage" -gt "$THRESHOLD_MEM" ]; then
echo "⚠️ 内存使用率过高: ${mem_usage}%"
fi
if [ "$disk_usage" -gt "$THRESHOLD_DISK" ]; then
echo "⚠️ 磁盘使用率过高: ${disk_usage}%"
fi
}
# 持续监控
while true; do
check_system_health
echo "--- $(date) ---"
sleep 60
done
2. 服务健康检查
# health-check.yaml
checks:
- name: "OpenClaw API 健康检查"
type: "http"
url: "http://localhost:3000/health"
interval: "30s"
timeout: "5s"
expected_status: 200
- name: "数据库连接检查"
type: "tcp"
host: "localhost"
port: 5432
interval: "60s"
timeout: "3s"
- name: "Redis 缓存检查"
type: "redis"
host: "localhost"
port: 6379
interval: "30s"
timeout: "2s"
alerts:
- condition: "consecutive_failures > 3"
notification:
type: "feishu_webhook"
url: "https://open.feishu.cn/open-apis/bot/v2/hook/your-webhook-url"
message: "🚨 OpenClaw 服务出现异常，请及时处理！"
🔧 部署辅助工具
1. 自动化部署脚本
#!/bin/bash
# deploy.sh
set -e
echo "🚀 OpenClaw 自动化部署脚本"
echo "=========================="
# 配置变量
APP_NAME="openclaw-bot"
DEPLOY_DIR="/opt/${APP_NAME}"
BACKUP_DIR="/backup/${APP_NAME}"
VERSION=$(date +%Y%m%d_%H%M%S)
# 备份当前版本
echo "📦 创建备份..."
mkdir -p ${BACKUP_DIR}/${VERSION}
cp -r ${DEPLOY_DIR}/* ${BACKUP_DIR}/${VERSION}/ 2>/dev/null || true
# 拉取最新代码
echo "📥 拉取最新代码..."
cd ${DEPLOY_DIR}
git pull origin main
# 安装依赖
echo "🔧 安装依赖..."
npm ci --production
# 运行数据库迁移
echo "🔄 运行数据库迁移..."
npm run migrate
# 重启服务
echo "🔄 重启服务..."
sudo systemctl restart ${APP_NAME}
# 等待服务启动
echo "⏳ 等待服务启动..."
sleep 10
# 健康检查
echo "🔍 健康检查..."
for i in {1..30}; do
if curl -f http://localhost:3000/health > /dev/null 2>&1; then
echo "✅ 服务启动成功"
exit 0
fi
sleep 2
done
echo "❌ 服务启动失败"
exit 1
2. Docker 环境清理工具
#!/bin/bash
# docker-cleanup.sh
echo "🧹 Docker 环境清理工具"
echo "======================"
# 清理停止的容器
echo "📦 清理停止的容器..."
docker container prune -f
# 清理未使用的镜像
echo "🖼️ 清理未使用的镜像..."
docker image prune -a -f
# 清理未使用的卷
echo "💾 清理未使用的卷..."
docker volume prune -f
# 清理未使用的网络
echo "🌐 清理未使用的网络..."
docker network prune -f
# 显示清理结果
echo "📊 清理结果:"
docker system df
echo "✅ 清理完成"
📈 数据分析工具
1. 用户行为分析
#!/usr/bin/env python3
# user-analytics.py
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta
class UserAnalytics:
def __init__(self, data_file):
self.df = pd.read_csv(data_file)
self.df['timestamp'] = pd.to_datetime(self.df['timestamp'])
def analyze_active_users(self):
"""分析活跃用户"""
# 按日统计活跃用户数
daily_active = self.df.groupby(self.df['timestamp'].dt.date)['user_id'].nunique()
plt.figure(figsize=(12, 6))
daily_active.plot(kind='line')
plt.title('日活跃用户数')
plt.xlabel('日期')
plt.ylabel('用户数')
plt.xticks(rotation=45)
plt.tight_layout()
plt.savefig('daily_active_users.png')
return daily_active
def analyze_peak_hours(self):
"""分析高峰时段"""
hourly_usage = self.df.groupby(self.df['timestamp'].dt.hour).size()
plt.figure(figsize=(10, 6))
hourly_usage.plot(kind='bar')
plt.title('小时使用量分布')
plt.xlabel('小时')
plt.ylabel('消息数')
plt.savefig('hourly_usage.png')
return hourly_usage.idxmax()
def generate_report(self):
"""生成分析报告"""
print("📊 用户行为分析报告")
print("=" * 30)
total_users = self.df['user_id'].nunique()
total_messages = len(self.df)
avg_messages_per_user = total_messages / total_users
print(f"总用户数: {total_users}")
print(f"总消息数: {total_messages}")
print(f"人均消息数: {avg_messages_per_user:.1f}")
peak_hour = self.analyze_peak_hours()
print(f"高峰时段: {peak_hour}点")
self.analyze_active_users()
if __name__ == "__main__":
import sys
if len(sys.argv) != 2:
print("使用方法: python user-analytics.py <data_file.csv>")
sys.exit(1)
analyzer = UserAnalytics(sys.argv[1])
analyzer.generate_report()
2. 性能指标仪表板
<!DOCTYPE html>
<html>
<head>
<title>OpenClaw 性能监控仪表板</title>
<script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
</head>
<body>
<div style="width: 800px; margin: 0 auto;">
<h1>OpenClaw 性能监控</h1>
<div>
<canvas id="responseTimeChart"></canvas>
</div>
<div>
<canvas id="errorRateChart"></canvas>
</div>
</div>
<script>
// 模拟数据
const responseTimes = [120, 150, 180, 200, 170, 160, 140];
const timestamps = ['09:00', '10:00', '11:00', '12:00', '13:00', '14:00', '15:00'];
// 响应时间图表
new Chart(document.getElementById('responseTimeChart'), {
type: 'line',
data: {
labels: timestamps,
datasets: [{
label: '平均响应时间 (ms)',
data: responseTimes,
borderColor: 'rgb(75, 192, 192)',
tension: 0.1
}]
},
options: {
responsive: true,
scales: {
y: {
beginAtZero: true
}
}
}
});
</script>
</body>
</html>
📚 使用说明
工具分类说明
- 开发辅助工具: 日常开发中提高效率的脚本
- 测试工具: 功能测试和性能测试工具
- 监控告警工具: 系统监控和异常告警
- 部署工具: 自动化部署和环境管理
- 数据分析工具: 用户行为和性能数据分析
使用建议
- 根据实际需求选择合适的工具
- 定期更新工具以适配最新版本
- 结合 CI/CD 流程自动化使用
- 建立标准化的操作流程
这些工具旨在提高 OpenClaw 开发和运维效率，可根据具体需求进行定制化修改
