杰瑞科技汇

Python AlertSender如何实现告警发送?

Of course! The term "alertsender" isn't a standard library in Python, but it's a very common concept. It refers to a script or application whose job is to send notifications or alerts based on certain events or conditions.

Python AlertSender如何实现告警发送?-图1
(图片来源网络,侵删)

I'll guide you through creating a flexible and reusable Python alert sender. We'll build it in a modular way, so you can easily add new notification methods (like Slack, email, SMS) without changing the core logic.

Here's a complete guide, starting with a simple version and building up to a more advanced, configurable one.


Part 1: The Core Concept - A Simple Console Alert Sender

Let's start with the most basic version: printing an alert to the console. This is useful for testing and simple scripts.

# alertsender_simple.py
def send_console_alert(message: str, title: str = "Alert"):
    """Sends an alert message to the console."""
    print("="*20)
    print(f"ALERT: {title}")
    print(message)
    print("="*20)
# --- Example Usage ---
if __name__ == "__main__":
    # Simulate a condition that triggers an alert
    disk_usage = 98.5 # In a real script, you'd get this from a system call
    if disk_usage > 90:
        send_console_alert(
            message=f"Disk space is critically low at {disk_usage}%",
            title="Disk Space Warning"
        )

To run this:

Python AlertSender如何实现告警发送?-图2
(图片来源网络,侵删)
python alertsender_simple.py

Output:

====================
ALERT: Disk Space Warning
Disk space is critically low at 98.5%
====================

Part 2: Adding a Real Notification Channel - Email

Now, let's make it useful by sending an email. We'll use Python's built-in smtplib library.

Prerequisites: You need access to an SMTP server. You can use a free service like Gmail (with an "App Password"), or your own mail server.

# alertsender_email.py
import smtplib
from email.message import EmailMessage
def send_email_alert(
    message: str, str,
    recipient_email: str,
    sender_email: str,
    sender_password: str,
    smtp_server: str = "smtp.gmail.com",
    smtp_port: int = 465
):
    """Sends an alert email."""
    msg = EmailMessage()
    msg['Subject'] = title
    msg['From'] = sender_email
    msg['To'] = recipient_email
    msg.set_content(message)
    try:
        # Use SSL for secure connection
        with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
            server.login(sender_email, sender_password)
            server.send_message(msg)
        print("Email alert sent successfully!")
    except Exception as e:
        print(f"Failed to send email: {e}")
# --- Example Usage ---
if __name__ == "__main__":
    # IMPORTANT: Use environment variables for credentials in a real app!
    # For this example, we'll use placeholders.
    # NEVER hardcode passwords in your source code.
    # --- Configuration ---
    config = {
        "recipient_email": "admin@example.com",
        "sender_email": "your_sender@gmail.com",
        "sender_password": "YOUR_APP_PASSWORD", # Use an App Password for Gmail
        "smtp_server": "smtp.gmail.com",
        "smtp_port": 465
    }
    # Simulate a condition
    api_error_rate = 15.5 # 15.5% errors
    if api_error_rate > 10:
        send_email_alert(
            message=f"The API error rate has spiked to {api_error_rate}%. Please investigate immediately.",
            title="High API Error Rate Alert",
            **config
        )

Part 3: Building a Reusable, Pluggable Alert Sender

Hardcoding email logic isn't flexible. What if you also want to send a message to Slack? Let's create a system where you can "plug in" different sender types.

We'll use a dict to hold our configuration and a factory function to create the right sender.

# alertsender_factory.py
import smtplib
from email.message import EmailMessage
import requests # We'll need this for Slack
# --- Sender Implementations ---
def _send_console(message: str, title: str, **kwargs):
    """Internal function for sending to console."""
    print(f"[{title}] {message}")
def _send_email(message: str, title: str, config: dict):
    """Internal function for sending to email."""
    msg = EmailMessage()
    msg['Subject'] = title
    msg['From'] = config['email_sender']
    msg['To'] = config['email_recipient']
    msg.set_content(message)
    try:
        with smtplib.SMTP_SSL(config['email_smtp_server'], config['email_smtp_port']) as server:
            server.login(config['email_sender'], config['email_sender_password'])
            server.send_message(msg)
        print("Email alert sent successfully!")
    except Exception as e:
        print(f"Failed to send email: {e}")
def _send_slack(message: str, title: str, config: dict):
    """Internal function for sending to Slack."""
    # Slack messages are typically JSON
    slack_data = {"text": f"*{title}*\n{message}"}
    try:
        response = requests.post(
            config['slack_webhook_url'], 
            json=slack_data, 
            headers={'Content-Type': 'application/json'}
        )
        response.raise_for_status()  # Raise an exception for bad status codes
        print("Slack alert sent successfully!")
    except requests.exceptions.RequestException as e:
        print(f"Failed to send Slack alert: {e}")
# --- The Main Alert Sender Class ---
class AlertSender:
    def __init__(self, config: dict):
        self.config = config
        self.sender_methods = {
            'console': _send_console,
            'email': _send_email,
            'slack': _send_slack,
        }
    def send(self, message: str, title: str = "Alert", channels: list = None):
        """
        Sends an alert through one or more configured channels.
        Args:
            message (str): The body of the alert.
            title (str): The title of the alert.
            channels (list, optional): A list of channels to send to (e.g., ['console', 'email']).
                                       If None, sends to all configured channels.
        """
        if channels is None:
            # If no channels are specified, send to all that are configured
            channels = [ch for ch in self.sender_methods.keys() if ch in self.config]
        for channel in channels:
            if channel in self.sender_methods:
                print(f"Attempting to send alert via {channel}...")
                self.sender_methods[channel](message, title, self.config)
            else:
                print(f"Warning: Channel '{channel}' is not configured or not a valid option.")
# --- Example Usage ---
if __name__ == "__main__":
    # --- Configuration ---
    # In a real application, load this from a file (e.g., config.json or .env)
    app_config = {
        # This will be sent to both console and email
        'default_channels': ['console', 'email'],
        # Email configuration
        'email_sender': 'your_sender@gmail.com',
        'email_sender_password': 'YOUR_APP_PASSWORD',
        'email_recipient': 'admin@example.com',
        'email_smtp_server': 'smtp.gmail.com',
        'email_smtp_port': 465,
        # Slack configuration
        'slack_webhook_url': 'https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK',
    }
    # Create the alert sender instance
    sender = AlertSender(config=app_config)
    # --- Simulate different scenarios ---
    # Scenario 1: High CPU usage - send to console and email
    cpu_usage = 95.2
    if cpu_usage > 90:
        sender.send(
            message=f"Server CPU usage is at {cpu_usage}%",
            title="High CPU Usage",
            channels=['console', 'email'] # Override default channels
        )
    print("\n" + "-"*40 + "\n")
    # Scenario 2: New deployment - send only to Slack
    if True: # Simulating a deployment event
        sender.send(
            message="Application version 2.1.0 has been deployed successfully to production.",
            title="Deployment Successful",
            channels=['slack'] # Send only to Slack
        )

How to Improve and Extend This

  1. Use a Configuration File: Instead of a hardcoded dictionary, load your configuration from a file like config.json or .env files. This is much better for security and manageability.

    • config.json:
      {
        "default_channels": ["console", "email"],
        "email": {
          "sender": "your_sender@gmail.com",
          "sender_password": "YOUR_APP_PASSWORD",
          "recipient": "admin@example.com",
          "smtp_server": "smtp.gmail.com",
          "smtp_port": 465
        },
        "slack": {
          "webhook_url": "https://hooks.slack.com/services/YOUR/SLACK/WEBHOOK"
        }
      }
    • You can then load it with import json and json.load(open('config.json')).
  2. Add More Channels: The sender_methods dictionary makes it easy to add new services.

    • SMS: Use a service like Twilio. You'd add a _send_twilio function and its required config keys (twilio_account_sid, twilio_auth_token, twilio_phone_number, recipient_phone_number).
    • PagerDuty / Opsgenie: These are popular for on-call rotation. They have APIs that you can call with a requests library, similar to Slack.
  3. Use a Logging Library: For a production system, replace print() with the logging module. This allows you to control log levels (INFO, WARNING, ERROR) and output to files, which is essential for debugging.

  4. Asynchronous Sending: If you send many alerts or use slow services (like email), sending them synchronously can block your main application. You can make the senders asynchronous using asyncio and aiohttp for non-blocking network calls.

  5. Create a Command-Line Interface (CLI): Use a library like click or argparse to turn your script into a command-line tool, making it easy to trigger alerts from anywhere.

分享:
扫描分享到社交APP
上一篇
下一篇