本文目录导读：
我来提供一个全面的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()
使用建议
- 逐步部署：从基础监控开始，逐步添加更多监控维度
- 设置合理阈值：根据项目特点设置告警阈值
- 定期审查：定期检查监控日志和报告
- 持续优化：根据实际需求调整监控策略
- 备份配置：定期备份监控配置文件
这个方案提供了全方位的监控能力,可以根据实际需求选择使用其中的部分或全部功能。
版权声明：NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
NBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
CBA直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
足球直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
英超直播
