25.6 智能维护
约 2523 字大约 8 分钟
智能维护概述
Claude Code 可以自动化维护任务、监控系统健康状态、预测潜在问题，并提供维护优化建议。通过智能维护功能，开发者可以降低维护成本、提高系统可靠性。
自动化维护
1. 依赖管理
示例：自动化依赖管理
用户请求：
"为项目实现自动化依赖管理"
Claude Code 生成的代码：
## requirements.txt
```text
Flask-SQLAlchemy==3.0.5
Flask-JWT-Extended==4.5.2
Flask-Mail==0.9.1
Flask-Migrate==4.0.5
gunicorn==21.2.0
psycopg2-binary==2.9.7
redis==4.6.0
celery==5.3.4
python-dotenv==1.0.0
```
## requirements-dev.txt
```text
pytest==7.4.2
pytest-cov==4.1.0
pytest-mock==3.11.1
black==23.9.1
flake8==6.1.0
mypy==1.5.1
isort==5.12.0
pre-commit==3.4.0
## 依赖更新脚本
```python
#!/usr/bin/env python3
"""
依赖更新脚本
自动检查并更新过期的依赖包
"""
import subprocess
import json
from typing import List, Dict
import sys
def get_outdated_packages() -> List[Dict]:
"""获取过期的包"""
result = subprocess.run(
['pip', 'list', '--outdated', '--format=json'],
capture_output=True,
text=True
)
if result.returncode != 0:
print("Error: Failed to get outdated packages")
return []
packages = json.loads(result.stdout)
return packages
def check_compatibility(package: str, version: str) -> bool:
"""检查包的兼容性"""
try:
result = subprocess.run(
['pip', 'install', '--dry-run', f'{package}=={version}'],
capture_output=True,
text=True
)
return result.returncode == 0
except Exception as e:
print(f"Error checking compatibility for {package}: {e}")
return False
def update_package(package: str, version: str) -> bool:
"""更新包"""
try:
result = subprocess.run(
['pip', 'install', '--upgrade', f'{package}=={version}'],
capture_output=True,
text=True
)
return result.returncode == 0
except Exception as e:
print(f"Error updating {package}: {e}")
return False
def update_requirements():
"""更新 requirements.txt"""
try:
result = subprocess.run(
['pip', 'freeze'],
capture_output=True,
text=True
)
with open('requirements.txt', 'w') as f:
f.write(result.stdout)
return True
except Exception as e:
print(f"Error updating requirements.txt: {e}")
return False
def main():
"""主函数"""
print("检查过期依赖...")
outdated_packages = get_outdated_packages()
if not outdated_packages:
print("所有依赖都是最新的！")
return
print(f"\n发现 {len(outdated_packages)} 个过期包:")
for pkg in outdated_packages:
print(f" - {pkg['name']}: {pkg['version']} -> {pkg['latest_version']}")
print("\n检查兼容性...")
compatible_packages = []
for pkg in outdated_packages:
package_name = pkg['name']
latest_version = pkg['latest_version']
if check_compatibility(package_name, latest_version):
compatible_packages.append(pkg)
print(f" ✓ {package_name} {latest_version}")
else:
print(f" ✗ {package_name} {latest_version} (不兼容)")
if not compatible_packages:
print("\n没有可更新的包")
return
print(f"\n准备更新 {len(compatible_packages)} 个包...")
response = input("是否继续？(y/n): ")
if response.lower() != 'y':
print("取消更新")
return
print("\n开始更新...")
for pkg in compatible_packages:
package_name = pkg['name']
latest_version = pkg['latest_version']
print(f"更新 {package_name} 到 {latest_version}...")
if update_package(package_name, latest_version):
print(f" ✓ {package_name} 更新成功")
else:
print(f" ✗ {package_name} 更新失败")
print("\n更新 requirements.txt...")
if update_requirements():
print(" ✓ requirements.txt 更新成功")
else:
print(" ✗ requirements.txt 更新失败")
print("\n依赖更新完成！")
if __name__ == '__main__':
main()
```
## 依赖安全检查
```python
#!/usr/bin/env python3
"""
依赖安全检查脚本
检查依赖包的安全漏洞
"""
import subprocess
import json
from typing import List, Dict
def check_security() -> List[Dict]:
"""检查依赖安全性"""
try:
result = subprocess.run(
['pip', 'audit'],
capture_output=True,
text=True
)
if result.returncode != 0:
print("Error: Failed to check security")
return []
return json.loads(result.stdout)
except Exception as e:
print(f"Error checking security: {e}")
return []
def main():
"""主函数"""
print("检查依赖安全性...")
vulnerabilities = check_security()
if not vulnerabilities:
print("未发现安全漏洞！")
return
print(f"\n发现 {len(vulnerabilities)} 个安全漏洞:")
for vuln in vulnerabilities:
print(f"\n 包: {vuln['name']}")
print(f" 版本: {vuln['version']}")
print(f" 漏洞: {vuln['advisory']}")
print(f" 严重性: {vuln['severity']}")
print(f" 修复版本: {vuln['fix_versions']}")
if __name__ == '__main__':
main()
### 2. 数据库维护
```python
# 示例：自动化数据库维护
用户请求：
"为应用实现自动化数据库维护"
Claude Code 生成的代码：
```python
```python
from datetime import datetime, timedelta
from sqlalchemy import text
from app import db
import logging
logger = logging.getLogger(__name__)
class DatabaseMaintenance:
"""数据库维护类"""
def __init__(self, db_session):
self.db = db_session
def analyze_tables(self):
"""分析表"""
tables = ['users', 'products', 'orders', 'payments']
for table in tables:
try:
result = self.db.execute(text(f"ANALYZE {table}"))
logger.info(f"Table {table} analyzed successfully")
except Exception as e:
logger.error(f"Error analyzing table {table}: {e}")
def vacuum_tables(self):
"""清理表"""
tables = ['users', 'products', 'orders', 'payments']
for table in tables:
try:
result = self.db.execute(text(f"VACUUM ANALYZE {table}"))
logger.info(f"Table {table} vacuumed successfully")
except Exception as e:
logger.error(f"Error vacuuming table {table}: {e}")
def rebuild_indexes(self):
"""重建索引"""
indexes = [
'idx_users_username',
'idx_users_email',
'idx_orders_user_id',
'idx_orders_status',
'idx_products_name'
]
for index in indexes:
try:
result = self.db.execute(text(f"REINDEX INDEX {index}"))
logger.info(f"Index {index} rebuilt successfully")
except Exception as e:
logger.error(f"Error rebuilding index {index}: {e}")
def clean_old_logs(self, days=30):
"""清理旧日志"""
cutoff_date = datetime.utcnow() - timedelta(days=days)
try:
result = self.db.execute(
text("""
DELETE FROM logs
WHERE created_at < :cutoff_date
"""),
{'cutoff_date': cutoff_date}
)
deleted_count = result.rowcount
logger.info(f"Deleted {deleted_count} old log entries")
self.db.commit()
return deleted_count
except Exception as e:
logger.error(f"Error cleaning old logs: {e}")
self.db.rollback()
return 0
def clean_old_sessions(self, days=7):
"""清理旧会话"""
cutoff_date = datetime.utcnow() - timedelta(days=days)
try:
result = self.db.execute(
text("""
DELETE FROM sessions
WHERE expires_at < :cutoff_date
"""),
{'cutoff_date': cutoff_date}
)
deleted_count = result.rowcount
logger.info(f"Deleted {deleted_count} old sessions")
self.db.commit()
return deleted_count
except Exception as e:
logger.error(f"Error cleaning old sessions: {e}")
self.db.rollback()
return 0
def optimize_database(self):
"""优化数据库"""
try:
self.analyze_tables()
self.vacuum_tables()
self.rebuild_indexes()
logger.info("Database optimization completed successfully")
except Exception as e:
logger.error(f"Error optimizing database: {e}")
def get_database_stats(self):
"""获取数据库统计信息"""
stats = {}
try:
# 表大小
result = self.db.execute(text("""
SELECT
schemaname,
tablename,
pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) AS size
FROM pg_tables
WHERE schemaname = 'public'
ORDER BY pg_total_relation_size(schemaname||'.'||tablename) DESC
"""))
stats['table_sizes'] = [
{
'schema': row[0],
'table': row[1],
'size': row[2]
}
for row in result
]
# 索引使用情况
result = self.db.execute(text("""
SELECT
schemaname,
tablename,
indexname,
idx_scan,
idx_tup_read,
idx_tup_fetch
FROM pg_stat_user_indexes
ORDER BY idx_scan DESC
"""))
stats['index_usage'] = [
{
'schema': row[0],
'table': row[1],
'index': row[2],
'scans': row[3],
'tuples_read': row[4],
'tuples_fetched': row[5]
}
for row in result
]
# 慢查询
result = self.db.execute(text("""
SELECT
query,
calls,
total_time,
mean_time,
max_time
FROM pg_stat_statements
ORDER BY mean_time DESC
LIMIT 10
"""))
stats['slow_queries'] = [
{
'query': row[0],
'calls': row[1],
'total_time': row[2],
'mean_time': row[3],
'max_time': row[4]
}
for row in result
]
return stats
except Exception as e:
logger.error(f"Error getting database stats: {e}")
return stats
def run_maintenance():
"""运行维护任务"""
maintenance = DatabaseMaintenance(db.session)
print("开始数据库维护...")
# 优化数据库
print("优化数据库...")
maintenance.optimize_database()
# 清理旧数据
print("清理旧日志...")
maintenance.clean_old_logs(days=30)
print("清理旧会话...")
maintenance.clean_old_sessions(days=7)
# 获取统计信息
print("获取数据库统计信息...")
stats = maintenance.get_database_stats()
print("\n数据库统计信息:")
print(f" 表数量: {len(stats['table_sizes'])}")
print(f" 索引数量: {len(stats['index_usage'])}")
print(f" 慢查询数量: {len(stats['slow_queries'])}")
print("\n数据库维护完成！")
if __name__ == '__main__':
run_maintenance()
```系统监控
1. 性能监控
示例：实现性能监控
用户请求：
"为应用实现性能监控"
Claude Code 生成的代码：
```python
import time
import psutil
from functools import wraps
from flask import request, g
from prometheus_client import Counter, Histogram, Gauge, generate_latest
import logging
logger = logging.getLogger(__name__)
# Prometheus 指标
request_count = Counter(
'http_requests_total',
'Total HTTP requests',
['method', 'endpoint', 'status']
)
request_duration = Histogram(
'http_request_duration_seconds',
'HTTP request duration',
['method', 'endpoint']
)
active_connections = Gauge(
'active_connections',
'Number of active connections'
)
memory_usage = Gauge(
'memory_usage_bytes',
'Memory usage in bytes'
)
cpu_usage = Gauge(
'cpu_usage_percent',
'CPU usage percentage'
)
disk_usage = Gauge(
'disk_usage_percent',
'Disk usage percentage'
)
def track_request_metrics(f):
"""跟踪请求指标"""
@wraps(f)
def decorated_function(*args, **kwargs):
start_time = time.time()
# 记录请求开始
g.start_time = start_time
try:
response = f(*args, **kwargs)
status_code = response.status_code if hasattr(response, 'status_code') else 200
# 记录请求计数
request_count.labels(
method=request.method,
endpoint=request.endpoint or 'unknown',
status=status_code
).inc()
# 记录请求持续时间
duration = time.time() - start_time
request_duration.labels(
method=request.method,
endpoint=request.endpoint or 'unknown'
).observe(duration)
return response
except Exception as e:
# 记录错误
request_count.labels(
method=request.method,
endpoint=request.endpoint or 'unknown',
status=500
).inc()
raise
return decorated_function
def update_system_metrics():
"""更新系统指标"""
# 内存使用
memory = psutil.virtual_memory()
memory_usage.set(memory.used)
# CPU 使用
cpu_usage.set(psutil.cpu_percent())
# 磁盘使用
disk = psutil.disk_usage('/')
disk_usage.set(disk.percent)
# 活跃连接
active_connections.set(len(psutil.net_connections()))
class PerformanceMonitor:
"""性能监控类"""
def __init__(self, app):
self.app = app
self.metrics = {}
def track_function(self, name):
"""跟踪函数性能"""
def decorator(f):
@wraps(f)
def decorated_function(*args, **kwargs):
```
def setup_monitoring(app):
"""设置监控"""2. 错误监控
示例：实现错误监控
用户请求： "为应用实现错误监控"
Claude Code 生成的代码：
import logging
import traceback
from datetime import datetime
from typing import Dict, List
from flask import request, g
import sentry_sdk
from sentry_sdk.integrations.flask import FlaskIntegration
logger = logging.getLogger(__name__)
class ErrorMonitor:
"""错误监控类"""
def __init__(self, app):
self.app = app
self.errors = []
self.error_stats = {}
def capture_exception(self, exception, context=None):
"""捕获异常"""
error_data = {
'type': type(exception).__name__,
'message': str(exception),
'traceback': traceback.format_exc(),
'timestamp': datetime.utcnow().isoformat(),
'context': context or {}
}
# 添加请求信息
if request:
error_data['request'] = {
'method': request.method,
'path': request.path,
'url': request.url,
'ip': request.remote_addr,
'user_agent': request.user_agent.string
}
# 添加用户信息
if hasattr(g, 'user_id'):
error_data['user_id'] = g.user_id
self.errors.append(error_data)
# 更新统计
error_type = error_data['type']
if error_type not in self.error_stats:
self.error_stats[error_type] = {
'count': 0,
'last_occurrence': None
}
self.error_stats[error_type]['count'] += 1
self.error_stats[error_type]['last_occurrence'] = error_data['timestamp']
# 记录日志
logger.error(
f"Exception captured: {error_type}",
extra=error_data
)
# 发送到 Sentry
sentry_sdk.capture_exception(exception)
def get_errors(self, limit=100):
"""获取错误列表"""
return self.errors[-limit:]
def get_error_stats(self):
"""获取错误统计"""
return self.error_stats
def get_frequent_errors(self, threshold=10):
"""获取频繁错误"""
frequent_errors = []
for error_type, stats in self.error_stats.items():
if stats['count'] >= threshold:
frequent_errors.append({
'type': error_type,
'count': stats['count'],
'last_occurrence': stats['last_occurrence']
})
return sorted(frequent_errors, key=lambda x: x['count'], reverse=True)
def clear_errors(self):
"""清除错误"""
self.errors = []
self.error_stats = {}
def setup_error_monitoring(app, sentry_dsn):
"""设置错误监控"""
# 初始化 Sentry
sentry_sdk.init(
dsn=sentry_dsn,
integrations=[FlaskIntegration()],
traces_sample_rate=1.0,
profiles_sample_rate=1.0
)
error_monitor = ErrorMonitor(app)
@app.errorhandler(Exception)
def handle_exception(e):
"""处理异常"""
error_monitor.capture_exception(e)
if request.is_json:
return {'error': str(e)}, 500
else:
return str(e), 500
@app.errorhandler(404)
def handle_not_found(e):
"""处理 404"""
logger.warning(f"404 Not Found: {request.path}")
return {'error': 'Not found'}, 404
@app.errorhandler(500)
def handle_server_error(e):
"""处理 500"""
error_monitor.capture_exception(e)
return {'error': 'Internal server error'}, 500
@app.route('/admin/errors')
def get_errors():
"""获取错误列表"""
errors = error_monitor.get_errors()
return {'errors': errors}
@app.route('/admin/errors/stats')
def get_error_stats():
"""获取错误统计"""
stats = error_monitor.get_error_stats()
return {'stats': stats}
@app.route('/admin/errors/frequent')
def get_frequent_errors():
"""获取频繁错误"""
frequent_errors = error_monitor.get_frequent_errors()
return {'frequent_errors': frequent_errors}
return app预测性维护
1. 容量预测
示例：实现容量预测
用户请求：
"为应用实现容量预测"
Claude Code 生成的代码：
import numpy as np
from datetime import datetime, timedelta
from typing import List, Dict
import logging
logger = logging.getLogger(__name__)
class CapacityPredictor:
"""容量预测器"""
def __init__(self):
self.history = []
self.predictions = {}
def add_metric(self, timestamp: datetime, metric_name: str, value: float):
"""添加指标"""
self.history.append({
'timestamp': timestamp,
'metric': metric_name,
'value': value
})
def predict_capacity(self, metric_name: str, days: int = 7) -> List[Dict]:
"""预测容量"""
# 获取历史数据
data = [
entry for entry in self.history
if entry['metric'] == metric_name
]
if len(data) < 30:
logger.warning(f"Insufficient data for {metric_name}")
return []
# 提取值
values = [entry['value'] for entry in data]
# 计算趋势
trend = self._calculate_trend(values)
# 预测未来值
predictions = []
for i in range(days):
predicted_value = values[-1] + trend * (i + 1)
predicted_date = datetime.utcnow() + timedelta(days=i + 1)
predictions.append({
'date': predicted_date.isoformat(),
'value': predicted_value,
'metric': metric_name
})
self.predictions[metric_name] = predictions
return predictions
def _calculate_trend(self, values: List[float]) -> float:
"""计算趋势"""
if len(values) < 2:
return 0
# 使用线性回归
x = np.arange(len(values))
y = np.array(values)
# 计算斜率
slope = np.polyfit(x, y, 1)[0]
return slope
def check_capacity_alerts(self, threshold: float = 0.9) -> List[Dict]:
"""检查容量告警"""
alerts = []
for metric_name, predictions in self.predictions.items():
for prediction in predictions:
if prediction['value'] >= threshold:
alerts.append({
'metric': metric_name,
'date': prediction['date'],
'value': prediction['value'],
'threshold': threshold
})
return sorted(alerts, key=lambda x: x['value'], reverse=True)
def get_capacity_recommendations(self) -> List[Dict]:
"""获取容量建议"""
recommendations = []
alerts = self.check_capacity_alerts()
if alerts:
recommendations.append({
'type': 'scale_up',
'message': f"发现 {len(alerts)} 个容量告警，建议扩容",
'alerts': alerts
})
# 检查资源利用率
for metric_name, predictions in self.predictions.items():
avg_value = np.mean([p['value'] for p in predictions])
if avg_value < 0.3:
recommendations.append({
'type': 'scale_down',
'message': f"{metric_name} 平均利用率较低，建议缩容",
'metric': metric_name,
'avg_value': avg_value
})
return recommendations
def run_capacity_prediction():
"""运行容量预测"""
predictor = CapacityPredictor()
# 添加历史数据（示例）
now = datetime.utcnow()
for i in range(30):
timestamp = now - timedelta(days=30 - i)
value = 0.5 + (i / 100) + np.random.normal(0, 0.05)
predictor.add_metric(timestamp, 'cpu_usage', value)
# 预测容量
print("预测 CPU 使用率...")
predictions = predictor.predict_capacity('cpu_usage', days=7)
print("\n预测结果:")
for prediction in predictions:
print(f" {prediction['date']}: {prediction['value']:.2%}")
# 检查告警
print("\n检查容量告警...")
alerts = predictor.check_capacity_alerts(threshold=0.9)
if alerts:
print(f"发现 {len(alerts)} 个告警:")
for alert in alerts:
print(f" {alert['date']}: {alert['value']:.2%} (阈值: {alert['threshold']:.0%})")
else:
print("未发现容量告警")
# 获取建议
print("\n获取容量建议...")
recommendations = predictor.get_capacity_recommendations()
for rec in recommendations:
print(f" {rec['type']}: {rec['message']}")
if __name__ == '__main__':
run_capacity_prediction()