本文目录导读：
我来提供一个全面的Python项目数据监控方案：
基础监控框架
核心监控类
import psutil
import json
import time
import logging
from datetime import datetime
from typing import Dict, Any, Optional
from pathlib import Path
import threading
class ProjectMonitor:
"""项目全方位监控器"""
def __init__(self, config_path: str = "monitor_config.json"):
self.config = self._load_config(config_path)
self.logger = self._setup_logger()
self.metrics = {}
self.alerts = []
self._running = False
def _load_config(self, path: str) -> Dict:
"""加载监控配置"""
with open(path, 'r') as f:
return json.load(f)
def _setup_logger(self) -> logging.Logger:
"""设置日志系统"""
logger = logging.getLogger('ProjectMonitor')
handler = logging.FileHandler('monitor.log')
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
系统资源监控
class SystemMonitor:
"""系统资源监控"""
@staticmethod
def get_cpu_info() -> Dict:
"""获取CPU信息"""
return {
'cpu_percent': psutil.cpu_percent(interval=1),
'cpu_count': psutil.cpu_count(),
'cpu_freq': psutil.cpu_freq()._asdict() if psutil.cpu_freq() else None,
'load_avg': psutil.getloadavg()
}
@staticmethod
def get_memory_info() -> Dict:
"""获取内存信息"""
memory = psutil.virtual_memory()
swap = psutil.swap_memory()
return {
'memory_total': memory.total,
'memory_available': memory.available,
'memory_percent': memory.percent,
'swap_total': swap.total,
'swap_percent': swap.percent
}
@staticmethod
def get_disk_info() -> Dict:
"""获取磁盘信息"""
disk_info = {}
for partition in psutil.disk_partitions():
try:
usage = psutil.disk_usage(partition.mountpoint)
disk_info[partition.mountpoint] = {
'total': usage.total,
'used': usage.used,
'free': usage.free,
'percent': usage.percent
}
except PermissionError:
continue
return disk_info
@staticmethod
def get_network_info() -> Dict:
"""获取网络信息"""
net_io = psutil.net_io_counters()
connections = psutil.net_connections()
return {
'bytes_sent': net_io.bytes_sent,
'bytes_recv': net_io.bytes_recv,
'packets_sent': net_io.packets_sent,
'packets_recv': net_io.packets_recv,
'active_connections': len(connections)
}
应用性能监控
import time
from functools import wraps
from contextlib import contextmanager
class AppPerformanceMonitor:
"""应用性能监控"""
def __init__(self):
self.metrics = {
'api_calls': {},
'function_calls': {},
'db_queries': []
}
@contextmanager
def monitor_function(self, func_name: str):
"""监控函数执行时间"""
start_time = time.time()
try:
yield
finally:
execution_time = time.time() - start_time
if func_name not in self.metrics['function_calls']:
self.metrics['function_calls'][func_name] = {
'count': 0,
'total_time': 0,
'avg_time': 0,
'max_time': 0,
'min_time': float('inf')
}
stats = self.metrics['function_calls'][func_name]
stats['count'] += 1
stats['total_time'] += execution_time
stats['avg_time'] = stats['total_time'] / stats['count']
stats['max_time'] = max(stats['max_time'], execution_time)
stats['min_time'] = min(stats['min_time'], execution_time)
# 记录慢查询
if execution_time > 1.0: # 超过1秒
self._log_slow_query(func_name, execution_time)
def _log_slow_query(self, func_name: str, execution_time: float):
"""记录慢查询"""
self.metrics['db_queries'].append({
'function': func_name,
'execution_time': execution_time,
'timestamp': time.time()
})
# 装饰器版本
def monitor_performance(func):
@wraps(func)
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs)
elapsed = time.time() - start
# 记录到监控系统
print(f"Function {func.__name__} took {elapsed:.2f}s")
return result
return wrapper
数据库监控
import sqlite3
from datetime import datetime, timedelta
import pandas as pd
class DatabaseMonitor:
"""数据库监控"""
def __init__(self, db_path: str):
self.db_path = db_path
self.connection = sqlite3.connect(db_path)
def monitor_query_performance(self, query: str) -> Dict:
"""监控查询性能"""
cursor = self.connection.cursor()
# 执行前
start_time = time.time()
cursor.execute(f"EXPLAIN QUERY PLAN {query}")
query_plan = cursor.fetchall()
# 执行查询
cursor.execute(query)
result = cursor.fetchall()
# 执行后
execution_time = time.time() - start_time
return {
'query': query,
'execution_time': execution_time,
'rows_affected': len(result),
'query_plan': query_plan
}
def get_table_stats(self) -> Dict:
"""获取表统计信息"""
cursor = self.connection.cursor()
# 获取所有表
cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
tables = cursor.fetchall()
stats = {}
for table in tables:
table_name = table[0]
# 获取行数
cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
row_count = cursor.fetchone()[0]
# 获取表大小
cursor.execute(f"PRAGMA table_info({table_name})")
columns = cursor.fetchall()
stats[table_name] = {
'row_count': row_count,
'column_count': len(columns),
'columns': [col[1] for col in columns]
}
return stats
def monitor_connections(self) -> Dict:
"""监控数据库连接"""
cursor = self.connection.cursor()
cursor.execute("PRAGMA database_list")
databases = cursor.fetchall()
return {
'active_connections': 1, # SQLite单连接
'database_backends': databases,
'cache_size': self._get_cache_size(),
'page_count': self._get_page_count()
}
def _get_cache_size(self) -> int:
"""获取缓存大小"""
cursor = self.connection.cursor()
cursor.execute("PRAGMA cache_size")
return cursor.fetchone()[0]
def _get_page_count(self) -> int:
"""获取页数"""
cursor = self.connection.cursor()
cursor.execute("PRAGMA page_count")
return cursor.fetchone()[0]
文件和数据监控
import hashlib
import os
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class FileMonitor:
"""文件系统监控"""
def __init__(self, watch_directory: str):
self.watch_directory = watch_directory
self.observer = Observer()
self.file_stats = {}
def start_monitoring(self):
"""开始监控文件变化"""
event_handler = FileChangeHandler(self)
self.observer.schedule(event_handler, self.watch_directory, recursive=True)
self.observer.start()
def stop_monitoring(self):
"""停止监控"""
self.observer.stop()
self.observer.join()
def get_directory_stats(self) -> Dict:
"""获取目录统计信息"""
stats = {
'total_files': 0,
'total_dirs': 0,
'total_size': 0,
'file_types': {}
}
for root, dirs, files in os.walk(self.watch_directory):
stats['total_dirs'] += len(dirs)
stats['total_files'] += len(files)
for file in files:
file_path = os.path.join(root, file)
try:
file_size = os.path.getsize(file_path)
stats['total_size'] += file_size
# 统计文件类型
ext = os.path.splitext(file)[1]
if ext not in stats['file_types']:
stats['file_types'][ext] = {
'count': 0,
'total_size': 0
}
stats['file_types'][ext]['count'] += 1
stats['file_types'][ext]['total_size'] += file_size
except OSError:
continue
return stats
class FileChangeHandler(FileSystemEventHandler):
"""文件变化处理器"""
def __init__(self, monitor):
self.monitor = monitor
def on_modified(self, event):
if not event.is_directory:
print(f"File modified: {event.src_path}")
# 计算文件哈希值
file_hash = self._calculate_hash(event.src_path)
self.monitor.file_stats[event.src_path] = {
'last_modified': datetime.now(),
'hash': file_hash
}
def _calculate_hash(self, file_path: str) -> str:
"""计算文件哈希值"""
hasher = hashlib.md5()
with open(file_path, 'rb') as f:
buf = f.read()
hasher.update(buf)
return hasher.hexdigest()
数据完整性监控
class DataIntegrityMonitor:
"""数据完整性监控"""
def __init__(self, data_source):
self.data_source = data_source
def check_data_quality(self) -> Dict:
"""检查数据质量"""
issues = []
# 检查空值
null_values = self._check_null_values()
if null_values:
issues.append({
'type': 'null_values',
'details': null_values
})
# 检查重复值
duplicates = self._check_duplicates()
if duplicates:
issues.append({
'type': 'duplicates',
'details': duplicates
})
# 检查数据范围
range_issues = self._check_data_range()
if range_issues:
issues.append({
'type': 'range_issues',
'details': range_issues
})
return {
'has_issues': len(issues) > 0,
'issues': issues,
'timestamp': datetime.now()
}
def _check_null_values(self) -> List:
"""检查空值"""
# 实现具体的空值检查逻辑
pass
def _check_duplicates(self) -> List:
"""检查重复数据"""
# 实现具体的重复数据检查逻辑
pass
def _check_data_range(self) -> List:
"""检查数据范围"""
# 实现具体的数据范围检查逻辑
pass
告警系统
class AlertSystem:
"""告警系统"""
def __init__(self, config: Dict):
self.config = config
self.alert_handlers = []
self._setup_alert_handlers()
def _setup_alert_handlers(self):
"""设置告警处理器"""
if self.config.get('email'):
self.alert_handlers.append(EmailAlertHandler(self.config['email']))
if self.config.get('slack'):
self.alert_handlers.append(SlackAlertHandler(self.config['slack']))
if self.config.get('webhook'):
self.alert_handlers.append(WebhookAlertHandler(self.config['webhook']))
def send_alert(self, message: str, severity: str = 'info'):
"""发送告警"""
alert = {
'message': message,
'severity': severity,
'timestamp': datetime.now()
}
for handler in self.alert_handlers:
try:
handler.send(alert)
except Exception as e:
print(f"Failed to send alert: {e}")
class EmailAlertHandler:
"""邮件告警处理器"""
def __init__(self, config: Dict):
self.smtp_server = config['smtp_server']
self.smtp_port = config['smtp_port']
self.username = config['username']
self.password = config['password']
self.recipients = config['recipients']
def send(self, alert: Dict):
"""发送邮件告警"""
# 实现邮件发送逻辑
pass
class SlackAlertHandler:
"""Slack告警处理器"""
def __init__(self, config: Dict):
self.webhook_url = config['webhook_url']
self.channel = config.get('channel', '#monitoring')
def send(self, alert: Dict):
"""发送Slack告警"""
# 实现Slack发送逻辑
pass
class WebhookAlertHandler:
"""Webhook告警处理器"""
def __init__(self, config: Dict):
self.url = config['url']
self.headers = config.get('headers', {})
def send(self, alert: Dict):
"""发送Webhook告警"""
# 实现Webhook发送逻辑
pass
可视化仪表盘
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta
class MonitorDashboard:
"""监控仪表盘"""
def __init__(self, monitor: ProjectMonitor):
self.monitor = monitor
self.history = []
def generate_report(self, duration: timedelta = timedelta(hours=1)):
"""生成监控报告"""
report = {
'timestamp': datetime.now(),
'duration': duration,
'system': self._get_system_stats(),
'application': self._get_app_stats(),
'database': self._get_db_stats(),
'alerts': self.monitor.alerts[-10:] # 最近10条告警
}
return report
def _get_system_stats(self) -> Dict:
"""获取系统统计信息"""
return {
'cpu': SystemMonitor.get_cpu_info(),
'memory': SystemMonitor.get_memory_info(),
'disk': SystemMonitor.get_disk_info(),
'network': SystemMonitor.get_network_info()
}
def _get_app_stats(self) -> Dict:
"""获取应用统计信息"""
if hasattr(self.monitor, 'performance_monitor'):
return self.monitor.performance_monitor.metrics
return {}
def _get_db_stats(self) -> Dict:
"""获取数据库统计信息"""
if hasattr(self.monitor, 'db_monitor'):
return self.monitor.db_monitor.get_table_stats()
return {}
def plot_cpu_usage(self, history: List):
"""绘制CPU使用率图表"""
timestamps = [h['timestamp'] for h in history]
cpu_values = [h['system']['cpu']['cpu_percent'] for h in history]
plt.figure(figsize=(10, 6))
plt.plot(timestamps, cpu_values, label='CPU Usage')
plt.xlabel('Time')
plt.ylabel('CPU Usage (%)')
plt.title('CPU Usage Over Time')
plt.legend()
plt.grid(True)
plt.show()
def plot_memory_usage(self, history: List):
"""绘制内存使用率图表"""
timestamps = [h['timestamp'] for h in history]
memory_values = [h['system']['memory']['memory_percent'] for h in history]
plt.figure(figsize=(10, 6))
plt.plot(timestamps, memory_values, label='Memory Usage', color='red')
plt.xlabel('Time')
plt.ylabel('Memory Usage (%)')
plt.title('Memory Usage Over Time')
plt.legend()
plt.grid(True)
plt.show()
配置模板
{
"monitoring": {
"interval": 60,
"system_monitor": true,
"app_monitor": true,
"database_monitor": true,
"file_monitor": true
},
"alerts": {
"email": {
"smtp_server": "smtp.gmail.com",
"smtp_port": 587,
"username": "your-email@gmail.com",
"password": "your-password",
"recipients": ["admin@example.com"]
},
"slack": {
"webhook_url": "https://hooks.slack.com/services/xxx",
"channel": "#monitoring"
},
"webhook": {
"url": "http://your-webhook-url",
"headers": {
"Authorization": "Bearer your-token"
}
}
},
"thresholds": {
"cpu_warning": 80,
"cpu_critical": 90,
"memory_warning": 85,
"memory_critical": 95,
"disk_warning": 85,
"disk_critical": 95
},
"exclude_patterns": ["*.log", "*.tmp", "__pycache__"]
}
主程序入口
def main():
"""主程序入口"""
# 加载配置
config = load_config('monitor_config.json')
# 创建监控器
monitor = ProjectMonitor()
# 创建告警系统
alert_system = AlertSystem(config.get('alerts', {}))
# 创建仪表盘
dashboard = MonitorDashboard(monitor)
# 启动监控循环
while True:
try:
# 收集指标
metrics = monitor.collect_metrics()
# 检查阈值
check_thresholds(metrics, config['thresholds'], alert_system)
# 生成报告
if time.time() % 3600 == 0: # 每小时生成报告
report = dashboard.generate_report()
save_report(report)
# 等待下一次采集
time.sleep(config['monitoring']['interval'])
except KeyboardInterrupt:
print("Monitoring stopped by user")
break
except Exception as e:
print(f"Error in monitoring loop: {e}")
time.sleep(10)
if __name__ == "__main__":
main()
使用建议
- 逐步部署：从基础监控开始，逐步添加更多监控维度
- 设置合理阈值：根据项目特点设置告警阈值
- 定期审查：定期检查监控日志和报告
- 持续优化：根据实际需求调整监控策略
- 备份配置：定期备份监控配置文件
这个方案提供了全方位的监控能力,可以根据实际需求选择使用其中的部分或全部功能。
