18.3 异常检测与告警（邮件/钉钉/飞书）

# 阈值检测示例
def check_threshold_anomaly(current_value, metric_config):
    """
    检查指标是否超过阈值
    """
    threshold = metric_config['threshold']
    direction = metric_config.get('direction', 'both')  # 'up', 'down', 'both'
    
    if direction == 'up' and current_value > threshold:
        return True, f"指标超过上限: {current_value} > {threshold}"
    elif direction == 'down' and current_value < threshold:
        return True, f"指标低于下限: {current_value} < {threshold}"
    elif direction == 'both' and (current_value > threshold['high'] or current_value < threshold['low']):
        return True, f"指标超出范围: {current_value}"
    return False, None

import numpy as np
from collections import deque

class MovingAverageDetector:
    def __init__(self, window_size=7, std_multiplier=3):
        self.window_size = window_size
        self.std_multiplier = std_multiplier
        self.history = deque(maxlen=window_size)
    
    def add_value(self, value):
        self.history.append(value)
    
    def is_anomaly(self, current_value):
        if len(self.history) < self.window_size:
            return False
        
        mean = np.mean(self.history)
        std = np.std(self.history)
        
        if std == 0:
            return abs(current_value - mean) > 0.1 * abs(mean)
        
        z_score = (current_value - mean) / std
        return abs(z_score) > self.std_multiplier

from sklearn.ensemble import IsolationForest
import pandas as pd

class MLAnomalyDetector:
    def __init__(self, contamination=0.05):
        self.model = IsolationForest(contamination=contamination, random_state=42)
        self.is_trained = False
    
    def train(self, historical_data):
        """
        历史数据格式: DataFrame with columns ['value', 'hour', 'day_of_week', 'is_weekend']
        """
        features = self._extract_features(historical_data)
        self.model.fit(features)
        self.is_trained = True
    
    def predict(self, current_data_point):
        if not self.is_trained:
            return False
        
        features = self._extract_features(pd.DataFrame([current_data_point]))
        prediction = self.model.predict(features)
        return prediction[0] == -1  # -1表示异常
    
    def _extract_features(self, df):
        features = df.copy()
        if 'timestamp' in features.columns:
            features['hour'] = pd.to_datetime(features['timestamp']).dt.hour
            features['day_of_week'] = pd.to_datetime(features['timestamp']).dt.dayofweek
            features['is_weekend'] = features['day_of_week'].isin([5, 6]).astype(int)
            features.drop('timestamp', axis=1, inplace=True)
        return features[['value', 'hour', 'day_of_week', 'is_weekend']]

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

class EmailAlert:
    def __init__(self, smtp_server, smtp_port, username, password):
        self.smtp_server = smtp_server
        self.smtp_port = smtp_port
        self.username = username
        self.password = password
    
    def send_alert(self, to_emails, subject, body):
        msg = MIMEMultipart()
        msg['From'] = self.username
        msg['To'] = ', '.join(to_emails)
        msg['Subject'] = subject
        
        msg.attach(MIMEText(body, 'html'))
        
        with smtplib.SMTP(self.smtp_server, self.smtp_port) as server:
            server.starttls()
            server.login(self.username, self.password)
            server.send_message(msg)

import requests
import json
import hmac
import hashlib
import base64
import time
import urllib.parse

class DingTalkAlert:
    def __init__(self, webhook_url, secret=None):
        self.webhook_url = webhook_url
        self.secret = secret
    
    def _sign(self, timestamp):
        if not self.secret:
            return ''
        string_to_sign = f'{timestamp}\n{self.secret}'
        hmac_code = hmac.new(
            self.secret.encode('utf-8'),
            string_to_sign.encode('utf-8'),
            digestmod=hashlib.sha256
        ).digest()
        sign = base64.b64encode(hmac_code).decode('utf-8')
        return urllib.parse.quote_plus(sign)
    
    def send_markdown(self, title, text):
        timestamp = str(round(time.time() * 1000))
        sign = self._sign(timestamp)
        
        url = f"{self.webhook_url}&timestamp={timestamp}&sign={sign}" if self.secret else self.webhook_url
        
        payload = {
            "msgtype": "markdown",
            "markdown": {
                "title": title,
                "text": text
            }
        }
        
        response = requests.post(url, json=payload)
        return response.json()
    
    def send_action_card(self, title, text, btn_orientation="1", btns=None):
        """发送行动卡片，包含跳转链接"""
        timestamp = str(round(time.time() * 1000))
        sign = self._sign(timestamp)
        
        url = f"{self.webhook_url}&timestamp={timestamp}&sign={sign}" if self.secret else self.webhook_url
        
        payload = {
            "msgtype": "actionCard",
            "actionCard": {
                "title": title,
                "text": text,
                "btnOrientation": btn_orientation,
                "btns": btns or []
            }
        }
        
        response = requests.post(url, json=payload)
        return response.json()

import requests
import json

class FeishuAlert:
    def __init__(self, webhook_url):
        self.webhook_url = webhook_url
    
    def send_text(self, content):
        payload = {
            "msg_type": "text",
            "content": {
                "text": content
            }
        }
        response = requests.post(self.webhook_url, json=payload)
        return response.json()
    
    def send_interactive(self, title, content, buttons=None):
        """发送富文本消息，支持按钮交互"""
        elements = []
        
        # 添加内容
        elements.append({
            "tag": "markdown",
            "content": content
        })
        
        # 添加按钮
        if buttons:
            elements.append({
                "tag": "action",
                "actions": [
                    {
                        "tag": "button",
                        "text": {
                            "tag": "plain_text",
                            "content": btn['text']
                        },
                        "url": btn.get('url', ''),
                        "type": btn.get('type', 'default')
                    }
                    for btn in buttons
                ]
            })
        
        payload = {
            "msg_type": "interactive",
            "card": {
                "header": {
                    "title": {
                        "tag": "plain_text",
                        "content": title
                    }
                },
                "elements": elements
            }
        }
        
        response = requests.post(self.webhook_url, json=payload)
        return response.json()

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
SEO/GEO异常检测与告警系统
"""

import os
import sys
import time
import json
import logging
from datetime import datetime, timedelta
from typing import Dict, List, Optional

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('/var/log/seo_alert.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger('SEOAlert')

class AlertManager:
    def __init__(self, config_path='alert_config.json'):
        self.config = self._load_config(config_path)
        self.detectors = self._init_detectors()
        self.alerts = self._init_alert_channels()
        self.metrics_history = {}
        
    def _load_config(self, path):
        with open(path, 'r') as f:
            return json.load(f)
    
    def _init_detectors(self):
        detectors = {}
        for metric_name, metric_config in self.config.get('metrics', {}).items():
            if metric_config.get('detector') == 'threshold':
                detectors[metric_name] = {
                    'type': 'threshold',
                    'config': metric_config
                }
            elif metric_config.get('detector') == 'moving_average':
                detectors[metric_name] = {
                    'type': 'moving_average',
                    'instance': MovingAverageDetector(
                        window_size=metric_config.get('window', 7),
                        std_multiplier=metric_config.get('std_multiplier', 3)
                    ),
                    'config': metric_config
                }
        return detectors
    
    def _init_alert_channels(self):
        channels = {}
        channel_config = self.config.get('channels', {})
        
        if 'email' in channel_config:
            channels['email'] = EmailAlert(
                smtp_server=channel_config['email']['smtp_server'],
                smtp_port=channel_config['email']['smtp_port'],
                username=channel_config['email']['username'],
                password=channel_config['email']['password']
            )
        
        if 'dingtalk' in channel_config:
            channels['dingtalk'] = DingTalkAlert(
                webhook_url=channel_config['dingtalk']['webhook_url'],
                secret=channel_config['dingtalk'].get('secret')
            )
        
        if 'feishu' in channel_config:
            channels['feishu'] = FeishuAlert(
                webhook_url=channel_config['feishu']['webhook_url']
            )
        
        return channels
    
    def check_metric(self, metric_name, current_value, metadata=None):
        """检查单个指标是否异常"""
        if metric_name not in self.detectors:
            logger.warning(f"未知指标: {metric_name}")
            return False, None
        
        detector = self.detectors[metric_name]
        
        if detector['type'] == 'threshold':
            is_anomaly, message = check_threshold_anomaly(current_value, detector['config'])
        elif detector['type'] == 'moving_average':
            detector['instance'].add_value(current_value)
            is_anomaly = detector['instance'].is_anomaly(current_value)
            message = f"移动平均检测异常: {current_value}"
        
        if is_anomaly:
            logger.warning(f"检测到异常 - {metric_name}: {message}")
            self._send_alert(metric_name, message, current_value, metadata)
        
        return is_anomaly, message
    
    def _send_alert(self, metric_name, message, value, metadata=None):
        """发送告警到所有配置的渠道"""
        alert_level = self.config.get('metrics', {}).get(metric_name, {}).get('alert_level', 'warning')
        timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
        
        # 构建告警内容
        title = f"[{alert_level.upper()}] SEO/GEO异常告警 - {metric_name}"
        
        text = f"""
## {title}

**时间**: {timestamp}
**指标**: {metric_name}
**当前值**: {value}
**异常信息**: {message}

**元数据**:
- 来源: {metadata.get('source', 'unknown') if metadata else 'unknown'}
- 页面: {metadata.get('url', 'N/A') if metadata else 'N/A'}
- 引擎: {metadata.get('engine', 'N/A') if metadata else 'N/A'}

**建议操作**:
1. 检查该指标的历史趋势
2. 确认是否为临时波动
3. 如持续异常，立即排查原因
"""
        
        # 发送到各渠道
        for channel_name, channel in self.alerts.items():
            try:
                if channel_name == 'email':
                    channel.send_alert(
                        to_emails=self.config['channels']['email']['to_emails'],
                        subject=title,
                        body=text
                    )
                elif channel_name == 'dingtalk':
                    channel.send_markdown(title, text)
                elif channel_name == 'feishu':
                    channel.send_interactive(title, text)
                
                logger.info(f"告警已发送到 {channel_name}")
            except Exception as e:
                logger.error(f"发送告警到 {channel_name} 失败: {e}")
    
    def run_check_cycle(self):
        """执行一次完整的检查周期"""
        logger.info("开始异常检测周期...")
        
        # 这里集成实际的数据采集逻辑
        # 示例：从不同来源获取指标数据
        metrics_to_check = [
            # (metric_name, current_value, metadata)
            ('search_console_clicks', 1500, {'source': 'google', 'url': 'https://example.com'}),
            ('generative_reference_count', 25, {'source': 'perplexity', 'engine': 'perplexity'}),
            ('page_load_time', 2.5, {'source': 'lighthouse', 'url': 'https://example.com/page'}),
        ]
        
        for metric_name, value, metadata in metrics_to_check:
            self.check_metric(metric_name, value, metadata)
        
        logger.info("异常检测周期完成")

def main():
    manager = AlertManager('alert_config.json')
    
    # 持续运行，每5分钟检查一次
    while True:
        try:
            manager.run_check_cycle()
            time.sleep(300)  # 5分钟
        except KeyboardInterrupt:
            logger.info("收到中断信号，退出")
            break
        except Exception as e:
            logger.error(f"运行异常: {e}")
            time.sleep(60)

if __name__ == '__main__':
    main()