本文目录导读：
我来介绍几种Python监控数据同步异常并发送告警的方法：
基础告警框架
import smtplib
import requests
import logging
from email.mime.text import MIMEText
from datetime import datetime
from typing import Optional
class DataSyncMonitor:
def __init__(self, config: dict):
self.config = config
self.setup_logging()
def setup_logging(self):
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('sync_monitor.log'),
logging.StreamHandler()
]
)
self.logger = logging.getLogger(__name__)
def check_sync_status(self) -> dict:
"""检查同步状态，返回检查结果"""
raise NotImplementedError
告警方式实现
邮件告警
class EmailAlert:
def __init__(self, smtp_config: dict):
self.smtp_host = smtp_config['host']
self.smtp_port = smtp_config['port']
self.username = smtp_config['username']
self.password = smtp_config['password']
self.from_addr = smtp_config['from_addr']
self.to_addrs = smtp_config['to_addrs']
def send_alert(self, subject: str, content: str):
msg = MIMEText(content, 'html', 'utf-8')
msg['Subject'] = subject
msg['From'] = self.from_addr
msg['To'] = ','.join(self.to_addrs)
try:
with smtplib.SMTP(self.smtp_host, self.smtp_port) as server:
server.starttls()
server.login(self.username, self.password)
server.send_message(msg)
print(f"邮件告警发送成功: {subject}")
except Exception as e:
print(f"邮件告警发送失败: {e}")
企业微信/钉钉告警
class WebhookAlert:
def __init__(self, webhook_url: str):
self.webhook_url = webhook_url
def send_wechat_alert(self, content: str):
"""企业微信机器人告警"""
data = {
"msgtype": "text",
"text": {
"content": f"【数据同步异常】\n{content}"
}
}
try:
response = requests.post(self.webhook_url, json=data)
if response.status_code == 200:
print("企业微信告警发送成功")
except Exception as e:
print(f"企业微信告警发送失败: {e}")
def send_dingtalk_alert(self, content: str):
"""钉钉机器人告警"""
data = {
"msgtype": "text",
"text": {
"content": f"【数据同步异常】\n{content}"
}
}
try:
response = requests.post(self.webhook_url, json=data)
if response.status_code == 200:
print("钉钉告警发送成功")
except Exception as e:
print(f"钉钉告警发送失败: {e}")
具体同步场景监控
数据库同步监控
import pymysql
from datetime import datetime, timedelta
class DBSyncMonitor(DataSyncMonitor):
def __init__(self, source_db_config: dict, target_db_config: dict, config: dict):
super().__init__(config)
self.source_db = source_db_config
self.target_db = target_db_config
self.email_alert = EmailAlert(config['email_config'])
self.webhook_alert = WebhookAlert(config['webhook_url'])
def check_record_count(self, table: str) -> dict:
"""检查表记录数是否一致"""
try:
source_conn = pymysql.connect(**self.source_db)
target_conn = pymysql.connect(**self.target_db)
with source_conn.cursor() as cursor:
cursor.execute(f"SELECT COUNT(*) FROM {table}")
source_count = cursor.fetchone()[0]
with target_conn.cursor() as cursor:
cursor.execute(f"SELECT COUNT(*) FROM {table}")
target_count = cursor.fetchone()[0]
return {
'table': table,
'source_count': source_count,
'target_count': target_count,
'match': source_count == target_count
}
except Exception as e:
return {
'table': table,
'error': str(e),
'match': False
}
finally:
if 'source_conn' in locals():
source_conn.close()
if 'target_conn' in locals():
target_conn.close()
def check_sync_timestamp(self, table: str, time_field: str) -> dict:
"""检查同步时间是否在合理范围内"""
try:
target_conn = pymysql.connect(**self.target_db)
with target_conn.cursor() as cursor:
cursor.execute(f"SELECT MAX({time_field}) FROM {table}")
max_time = cursor.fetchone()[0]
if max_time:
time_diff = datetime.now() - max_time
is_delayed = time_diff > timedelta(minutes=self.config.get('max_delay_minutes', 30))
return {
'table': table,
'last_sync_time': max_time,
'time_diff_minutes': time_diff.total_seconds() / 60,
'is_delayed': is_delayed
}
except Exception as e:
return {'table': table, 'error': str(e), 'is_delayed': True}
finally:
if 'target_conn' in locals():
target_conn.close()
文件同步监控
import os
import hashlib
from pathlib import Path
class FileSyncMonitor(DataSyncMonitor):
def __init__(self, source_dir: str, target_dir: str, config: dict):
super().__init__(config)
self.source_dir = Path(source_dir)
self.target_dir = Path(target_dir)
def check_files_exist(self, pattern: str = "*") -> dict:
"""检查文件是否存在且完整"""
source_files = set(self.source_dir.glob(pattern))
target_files = set(self.target_dir.glob(pattern))
missing_files = source_files - target_files
extra_files = target_files - source_files
return {
'missing_files': [str(f) for f in missing_files],
'extra_files': [str(f) for f in extra_files],
'is_complete': len(missing_files) == 0
}
def check_file_integrity(self, file_path: str) -> bool:
"""检查文件完整性（MD5校验）"""
source_file = self.source_dir / file_path
target_file = self.target_dir / file_path
if not source_file.exists() or not target_file.exists():
return False
def get_md5(file_path):
hash_md5 = hashlib.md5()
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
hash_md5.update(chunk)
return hash_md5.hexdigest()
return get_md5(source_file) == get_md5(target_file)
完整的监控脚本示例
import time
import json
from datetime import datetime
class SyncMonitorMain:
def __init__(self, config_file: str):
with open(config_file, 'r') as f:
self.config = json.load(f)
self.monitors = []
self.setup_monitors()
self.alert_senders = self.setup_alert_senders()
def setup_monitors(self):
"""初始化各类监控器"""
if 'db_sync' in self.config:
self.monitors.append(
DBSyncMonitor(
self.config['db_sync']['source'],
self.config['db_sync']['target'],
self.config
)
)
def setup_alert_senders(self):
"""初始化告警发送器"""
senders = []
if 'email' in self.config:
senders.append(EmailAlert(self.config['email']))
if 'webhook' in self.config:
senders.append(WebhookAlert(self.config['webhook']))
return senders
def send_alerts(self, subject: str, content: str):
"""发送告警到所有渠道"""
for sender in self.alert_senders:
try:
if isinstance(sender, EmailAlert):
sender.send_alert(subject, content)
elif isinstance(sender, WebhookAlert):
sender.send_wechat_alert(content)
except Exception as e:
print(f"告警发送失败: {e}")
def monitor_forever(self, interval: int = 300):
"""持续监控"""
print(f"开始数据同步监控，检查间隔：{interval}秒")
while True:
try:
for monitor in self.monitors:
# 执行各种检查
for table in self.config.get('monitor_tables', []):
result = monitor.check_record_count(table)
if not result.get('match'):
alert_content = f"""
<h3>数据同步异常</h3>
<p>时间：{datetime.now()}</p>
<p>表名：{table}</p>
<p>源库记录数：{result.get('source_count')}</p>
<p>目标库记录数：{result.get('target_count')}</p>
<p>状态：不匹配</p>
"""
self.send_alerts("数据同步异常告警", alert_content)
time.sleep(interval)
except KeyboardInterrupt:
print("监控已停止")
break
except Exception as e:
print(f"监控出错: {e}")
time.sleep(60) # 出错后等待1分钟重试
# 使用示例
if __name__ == "__main__":
config = {
"db_sync": {
"source": {
"host": "source_host",
"port": 3306,
"user": "user",
"password": "password",
"database": "source_db"
},
"target": {
"host": "target_host",
"port": 3306,
"user": "user",
"password": "password",
"database": "target_db"
}
},
"monitor_tables": ["users", "orders", "products"],
"email": {
"host": "smtp.example.com",
"port": 587,
"username": "alert@example.com",
"password": "password",
"from_addr": "alert@example.com",
"to_addrs": ["admin@example.com"]
},
"webhook": "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=xxx",
"max_delay_minutes": 30
}
# 保存配置到文件
with open('sync_config.json', 'w') as f:
json.dump(config, f, indent=2)
# 启动监控
monitor = SyncMonitorMain('sync_config.json')
monitor.monitor_forever(interval=300) # 每5分钟检查一次
模板
def generate_alert_template(sync_type: str, details: dict) -> str:
"""生成告警信息模板"""
templates = {
'db_sync': f"""
<h2>数据库同步异常告警</h2>
<p><strong>时间：</strong>{datetime.now()}</p>
<p><strong>类型：</strong>数据库同步</p>
<p><strong>数据库：</strong>{details.get('database')}</p>
<p><strong>表名：</strong>{details.get('table')}</p>
<p><strong>异常描述：</strong>{details.get('error_message')}</p>
<p><strong>源库状态：</strong>{details.get('source_status')}</p>
<p><strong>目标库状态：</strong>{details.get('target_status')}</p>
<hr>
<p><small>此邮件由数据同步监控系统自动发送，请勿回复。</small></p>
""",
'file_sync': f"""
<h2>文件同步异常告警</h2>
<p><strong>时间：</strong>{datetime.now()}</p>
<p><strong>类型：</strong>文件同步</p>
<p><strong>文件路径：</strong>{details.get('file_path')}</p>
<p><strong>异常描述：</strong>{details.get('error_message')}</p>
<p><strong>缺失文件列表：</strong></p>
<ul>
{''.join(f'<li>{f}</li>' for f in details.get('missing_files', []))}
</ul>
"""
}
return templates.get(sync_type, "未知同步类型异常")
使用建议
- 分级告警：根据异常严重程度（警告、严重、紧急）设置不同的告警方式
- 告警抑制：避免短时间内重复发送相同告警
- 告警升级：如果长时间未处理，自动升级告警级别
- 可视化监控：结合Grafana等工具进行可视化
- 告警记录：记录所有告警历史，便于问题追踪
这个监控系统可以根据你的具体需求进行定制和扩展。
