IT技术社区_专业开发者平台_最新编程教程与解决方案 -沐辰社区

Python脚本如何告警数据同步异常问题

wen python案例 282

本文目录导读：

Python脚本如何告警数据同步异常问题

  1. 基础告警框架
  2. 告警方式实现
  3. 具体同步场景监控
  4. 完整的监控脚本示例
  5. 告警内容模板
  6. 使用建议

我来介绍几种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, "未知同步类型异常")

使用建议

  1. 分级告警：根据异常严重程度（警告、严重、紧急）设置不同的告警方式
  2. 告警抑制：避免短时间内重复发送相同告警
  3. 告警升级：如果长时间未处理，自动升级告警级别
  4. 可视化监控：结合Grafana等工具进行可视化
  5. 告警记录：记录所有告警历史，便于问题追踪

这个监控系统可以根据你的具体需求进行定制和扩展。

版权声明：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直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 足球直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播 英超直播

上一篇Python脚本如何减少缓存数据库交互次数

下一篇Python脚本如何监控缓存数据库同步状态

相关文章

抱歉，评论功能暂时关闭!

友情链接攻略窝 · 礼包窝 · 配置窝 · 换算窝 · 算法窝 · 查文窝 · 手册窝 · 问答窝 · 全文搜 · 手册库
相关站点礼包窝 · 查文窝