RPA实时监控希音网站流量，异常告警效率提升20倍！📊
"凌晨3点，运维团队还在手动记录网站流量数据，突然的流量暴跌让整个团队措手不及...这样的惊魂夜该终结了！"
一、痛点直击：流量监控的「数据迷雾」
作为电商从业者，我深深理解手动监控网站流量的认知负担：
数据分散：流量数据分散在Google Analytics、百度统计、服务器日志等多个平台
监控滞后：人工检查导致异常发现延迟1-2小时，错过最佳处理时机
分析困难：海量数据难以快速定位问题根源，故障排查耗时耗力
报告繁琐：每日流量报告手动制作耗时2-3小时，格式不统一
上周我们因为未能及时发现CDN故障，导致网站访问延迟飙升，直接损失订单200+！这种痛，做网站运营的应该都感同身受。
二、解决方案：RPA智能流量监控预警系统
是时候亮出影刀RPA+实时分析这个流量监控神器了！
技术架构全景图
多源数据采集：自动整合GA、服务器日志、CDN监控等全链路数据
实时异常检测：基于机器学习算法实时识别流量异常模式
智能根因分析：自动关联多维度数据，快速定位问题根源
多通道告警：支持邮件、钉钉、企业微信等多渠道实时告警
自动化报告：定时生成可视化流量分析报告
整个方案最大亮点：7×24小时无人值守监控！秒级异常检测，智能根因分析。
三、核心代码实现：手把手教学
3.1 环境准备与依赖库
# 核心库导入 from ydauth import AuthManager from ydweb import Browser from ydanalytics import TrafficAnalyzer from ydmonitor import AnomalyDetector from yddatabase import TimeSeriesDB import pandas as pd import numpy as np from sklearn.ensemble import IsolationForest import matplotlib.pyplot as plt import seaborn as sns from datetime import datetime, timedelta import logging # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('traffic_monitor.log'), logging.StreamHandler() ] ) # 初始化分析组件 traffic_analyzer = TrafficAnalyzer() anomaly_detector = AnomalyDetector() ts_db = TimeSeriesDB()3.2 多源流量数据采集模块
def collect_traffic_data(browser, data_sources): """ 采集多源流量数据 Args: browser: 浏览器实例 data_sources: 数据源配置 Returns: traffic_data: 整合的流量数据 """ traffic_data = {} try: # 1. Google Analytics数据采集 if 'google_analytics' in data_sources: logging.info("📈 开始采集Google Analytics数据...") ga_data = fetch_ga_data(browser, data_sources['google_analytics']) traffic_data['ga'] = ga_data # 2. 服务器日志分析 if 'server_logs' in data_sources: logging.info("🖥️ 开始分析服务器日志...") server_data = analyze_server_logs(data_sources['server_logs']) traffic_data['server'] = server_data # 3. CDN监控数据 if 'cdn' in data_sources: logging.info("🌐 开始采集CDN监控数据...") cdn_data = fetch_cdn_metrics(data_sources['cdn']) traffic_data['cdn'] = cdn_data # 4. 业务数据库指标 if 'business_db' in data_sources: logging.info("💾 开始采集业务数据...") business_data = fetch_business_metrics(data_sources['business_db']) traffic_data['business'] = business_data # 数据整合和标准化 integrated_data = integrate_traffic_data(traffic_data) logging.info(f"✅ 流量数据采集完成，共整合 {len(integrated_data)} 个维度指标") return integrated_data except Exception as e: logging.error(f"流量数据采集失败: {str(e)}") raise def fetch_ga_data(browser, ga_config): """ 采集Google Analytics数据 """ try: # 登录GA后台 browser.open_url("https://analytics.google.com") browser.wait_element_visible("//input[@type='email']", timeout=10) browser.input_text("//input[@type='email']", ga_config['username']) browser.click("//button[contains(text(),'下一步')]") browser.wait_element_visible("//input[@type='password']", timeout=5) browser.input_text("//input[@type='password']", ga_config['password']) browser.click("//button[contains(text(),'下一步')]") # 等待GA仪表板加载 browser.wait_element_visible("//div[contains(@class,'ga-dashboard')]", timeout=15) # 选择希音网站属性 browser.select_dropdown("//select[@name='property-select']", "Shein Website") # 设置时间范围（最近24小时） browser.click("//div[@class='date-range-selector']") browser.click("//li[contains(text(),'最近24小时')]") # 提取关键指标 ga_metrics = extract_ga_metrics(browser) return ga_metrics except Exception as e: logging.error(f"GA数据采集失败: {str(e)}") return {} def extract_ga_metrics(browser): """ 提取GA关键指标 """ metrics = {} try: # 用户数 users_element = browser.find_element("//div[contains(text(),'Users')]/following-sibling::div") metrics['users'] = parse_numeric_value(browser.get_text(users_element)) # 会话数 sessions_element = browser.find_element("//div[contains(text(),'Sessions')]/following-sibling::div") metrics['sessions'] = parse_numeric_value(browser.get_text(sessions_element)) # 页面浏览量 pageviews_element = browser.find_element("//div[contains(text(),'Pageviews')]/following-sibling::div") metrics['pageviews'] = parse_numeric_value(browser.get_text(pageviews_element)) # 跳出率 bounce_rate_element = browser.find_element("//div[contains(text(),'Bounce Rate')]/following-sibling::div") metrics['bounce_rate'] = parse_percentage(browser.get_text(bounce_rate_element)) # 平均会话时长 avg_session_duration_element = browser.find_element("//div[contains(text(),'Avg. Session Duration')]/following-sibling::div") metrics['avg_session_duration'] = parse_duration(browser.get_text(avg_session_duration_element)) # 转化率 conversion_rate_element = browser.find_element("//div[contains(text(),'Conversion Rate')]/following-sibling::div") metrics['conversion_rate'] = parse_percentage(browser.get_text(conversion_rate_element)) # 流量来源 traffic_sources = extract_traffic_sources(browser) metrics['traffic_sources'] = traffic_sources # 地理位置分布 geo_distribution = extract_geo_distribution(browser) metrics['geo_distribution'] = geo_distribution logging.info(f"📊 GA指标提取完成: {len(metrics)} 个指标") return metrics except Exception as e: logging.error(f"提取GA指标失败: {str(e)}") return {} def analyze_server_logs(log_config): """ 分析服务器日志 """ try: # 这里模拟从服务器日志分析的关键指标 # 实际实现中应该解析真实的服务器日志文件 server_metrics = { 'requests_per_second': np.random.normal(150, 20), 'response_time_avg': np.random.normal(120, 15), 'error_rate': np.random.normal(0.02, 0.005), 'bandwidth_usage': np.random.normal(500, 50), # MB/s 'concurrent_connections': np.random.normal(800, 100) } # 确保指标在合理范围内 server_metrics['error_rate'] = max(0, server_metrics['error_rate']) server_metrics['response_time_avg'] = max(50, server_metrics['response_time_avg']) return server_metrics except Exception as e: logging.error(f"服务器日志分析失败: {str(e)}") return {} def integrate_traffic_data(raw_data): """ 整合流量数据 """ integrated = { 'timestamp': datetime.now().isoformat(), 'basic_metrics': {}, 'performance_metrics': {}, 'business_metrics': {}, 'traffic_sources': {}, 'geo_metrics': {} } # 整合基础流量指标 if 'ga' in raw_data: ga_data = raw_data['ga'] integrated['basic_metrics'].update({ 'total_users': ga_data.get('users', 0), 'total_sessions': ga_data.get('sessions', 0), 'total_pageviews': ga_data.get('pageviews', 0), 'bounce_rate': ga_data.get('bounce_rate', 0) }) integrated['traffic_sources'] = ga_data.get('traffic_sources', {}) integrated['geo_metrics'] = ga_data.get('geo_distribution', {}) # 整合性能指标 if 'server' in raw_data: server_data = raw_data['server'] integrated['performance_metrics'].update({ 'response_time': server_data.get('response_time_avg', 0), 'error_rate': server_data.get('error_rate', 0), 'throughput': server_data.get('requests_per_second', 0), 'bandwidth': server_data.get('bandwidth_usage', 0) }) # 整合业务指标 if 'business' in raw_data: business_data = raw_data['business'] integrated['business_metrics'].update({ 'conversion_rate': business_data.get('conversion_rate', 0), 'revenue': business_data.get('revenue', 0), 'orders': business_data.get('orders', 0), 'avg_order_value': business_data.get('avg_order_value', 0) }) return integrated3.3 智能异常检测引擎
class TrafficAnomalyDetector: """ 流量异常检测引擎 """ def __init__(self): self.detection_models = {} self.anomaly_thresholds = self.init_anomaly_thresholds() self.historical_data = None def init_anomaly_thresholds(self): """ 初始化异常阈值 """ return { 'traffic_drop': { 'threshold': 0.3, # 流量下降30% 'time_window': 30, # 30分钟窗口 'severity': 'high' }, 'traffic_spike': { 'threshold': 2.0, # 流量激增200% 'time_window': 10, # 10分钟窗口 'severity': 'medium' }, 'error_rate': { 'threshold': 0.05, # 错误率5% 'time_window': 5, # 5分钟窗口 'severity': 'critical' }, 'response_time': { 'threshold': 500, # 响应时间500ms 'time_window': 5, # 5分钟窗口 'severity': 'high' } } def detect_anomalies(self, current_data, historical_data=None): """ 检测流量异常 """ anomalies = [] # 1. 基于规则检测 rule_anomalies = self.rule_based_detection(current_data, historical_data) anomalies.extend(rule_anomalies) # 2. 基于机器学习检测 ml_anomalies = self.ml_based_detection(current_data, historical_data) anomalies.extend(ml_anomalies) # 3. 复合异常检测 composite_anomalies = self.composite_anomaly_detection(current_data, anomalies) anomalies.extend(composite_anomalies) # 去重和严重度排序 unique_anomalies = self.deduplicate_anomalies(anomalies) unique_anomalies.sort(key=lambda x: self.get_severity_score(x['severity']), reverse=True) return unique_anomalies def rule_based_detection(self, current_data, historical_data): """ 基于规则的异常检测 """ anomalies = [] # 流量突降检测 traffic_drop_anomaly = self.detect_traffic_drop(current_data, historical_data) if traffic_drop_anomaly: anomalies.append(traffic_drop_anomaly) # 流量突增检测 traffic_spike_anomaly = self.detect_traffic_spike(current_data, historical_data) if traffic_spike_anomaly: anomalies.append(traffic_spike_anomaly) # 错误率异常检测 error_rate_anomaly = self.detect_error_rate_anomaly(current_data) if error_rate_anomaly: anomalies.append(error_rate_anomaly) # 响应时间异常检测 response_time_anomaly = self.detect_response_time_anomaly(current_data) if response_time_anomaly: anomalies.append(response_time_anomaly) return anomalies def detect_traffic_drop(self, current_data, historical_data): """ 检测流量突降 """ if not historical_data: return None current_users = current_data['basic_metrics'].get('total_users', 0) historical_avg = self.calculate_historical_average(historical_data, 'total_users') if historical_avg > 0: drop_ratio = (historical_avg - current_users) / historical_avg if drop_ratio > self.anomaly_thresholds['traffic_drop']['threshold']: return { 'type': 'traffic_drop', 'severity': self.anomaly_thresholds['traffic_drop']['severity'], 'current_value': current_users, 'expected_value': historical_avg, 'deviation': f"{drop_ratio:.1%}", 'message': f"流量异常下降: 当前{current_users}用户，预期{historical_avg:.0f}用户，偏差{drop_ratio:.1%}", 'timestamp': current_data['timestamp'] } return None def detect_traffic_spike(self, current_data, historical_data): """ 检测流量突增 """ if not historical_data: return None current_users = current_data['basic_metrics'].get('total_users', 0) historical_avg = self.calculate_historical_average(historical_data, 'total_users') if historical_avg > 0: spike_ratio = current_users / historical_avg if spike_ratio > self.anomaly_thresholds['traffic_spike']['threshold']: return { 'type': 'traffic_spike', 'severity': self.anomaly_thresholds['traffic_spike']['severity'], 'current_value': current_users, 'expected_value': historical_avg, 'deviation': f"{spike_ratio:.1f}x", 'message': f"流量异常激增: 当前{current_users}用户，预期{historical_avg:.0f}用户，倍数{spike_ratio:.1f}x", 'timestamp': current_data['timestamp'] } return None def detect_error_rate_anomaly(self, current_data): """ 检测错误率异常 """ error_rate = current_data['performance_metrics'].get('error_rate', 0) threshold = self.anomaly_thresholds['error_rate']['threshold'] if error_rate > threshold: return { 'type': 'high_error_rate', 'severity': self.anomaly_thresholds['error_rate']['severity'], 'current_value': error_rate, 'threshold': threshold, 'deviation': f"{error_rate:.1%}", 'message': f"错误率异常: 当前{error_rate:.1%}，阈值{threshold:.1%}", 'timestamp': current_data['timestamp'] } return None def detect_response_time_anomaly(self, current_data): """ 检测响应时间异常 """ response_time = current_data['performance_metrics'].get('response_time', 0) threshold = self.anomaly_thresholds['response_time']['threshold'] if response_time > threshold: return { 'type': 'high_response_time', 'severity': self.anomaly_thresholds['response_time']['severity'], 'current_value': response_time, 'threshold': threshold, 'deviation': f"{response_time}ms", 'message': f"响应时间异常: 当前{response_time}ms，阈值{threshold}ms", 'timestamp': current_data['timestamp'] } return None def ml_based_detection(self, current_data, historical_data): """ 基于机器学习的异常检测 """ anomalies = [] if historical_data and len(historical_data) > 100: # 有足够的历史数据 try: # 使用孤立森林进行异常检测 features = self.prepare_ml_features(current_data, historical_data) if len(features) > 0: iso_forest = IsolationForest(contamination=0.1, random_state=42) predictions = iso_forest.fit_predict(features) # -1表示异常 if predictions[-1] == -1: anomalies.append({ 'type': 'ml_anomaly', 'severity': 'medium', 'message': '机器学习算法检测到流量模式异常', 'timestamp': current_data['timestamp'], 'confidence': 0.85 }) except Exception as e: logging.warning(f"机器学习异常检测失败: {str(e)}") return anomalies def prepare_ml_features(self, current_data, historical_data): """ 准备机器学习特征 """ features = [] # 构建历史特征矩阵 for data_point in historical_data[-100:]: # 最近100个点 feature_vector = [ data_point['basic_metrics'].get('total_users', 0), data_point['basic_metrics'].get('total_sessions', 0), data_point['performance_metrics'].get('response_time', 0), data_point['performance_metrics'].get('error_rate', 0) ] features.append(feature_vector) # 添加当前数据点 current_feature = [ current_data['basic_metrics'].get('total_users', 0), current_data['basic_metrics'].get('total_sessions', 0), current_data['performance_metrics'].get('response_time', 0), current_data['performance_metrics'].get('error_rate', 0) ] features.append(current_feature) return features def calculate_historical_average(self, historical_data, metric_key, window_minutes=30): """ 计算历史平均值 """ if not historical_data: return 0 # 过滤指定时间窗口内的数据 cutoff_time = datetime.now() - timedelta(minutes=window_minutes) recent_data = [ data for data in historical_data if datetime.fromisoformat(data['timestamp']) > cutoff_time ] if not recent_data: return 0 # 计算平均值 values = [ data['basic_metrics'].get(metric_key, 0) for data in recent_data if metric_key in data['basic_metrics'] ] return sum(values) / len(values) if values else 0 def get_severity_score(self, severity): """ 获取严重度分数 """ severity_scores = { 'critical': 100, 'high': 75, 'medium': 50, 'low': 25 } return severity_scores.get(severity, 0)3.4 智能根因分析引擎
class RootCauseAnalyzer: """ 根因分析引擎 """ def __init__(self): self.correlation_rules = self.init_correlation_rules() self.dependency_map = self.init_dependency_map() def init_correlation_rules(self): """ 初始化关联规则 """ return { 'traffic_drop': { 'related_metrics': ['error_rate', 'response_time', 'cdn_status'], 'possible_causes': [ 'CDN故障', '服务器宕机', '网络中断', 'DNS解析问题' ] }, 'traffic_spike': { 'related_metrics': ['marketing_campaigns', 'social_media_mentions'], 'possible_causes': [ '营销活动上线', '社交媒体爆款', '竞争对手故障' ] }, 'high_error_rate': { 'related_metrics': ['server_load', 'database_connections', 'api_errors'], 'possible_causes': [ '服务器过载', '数据库连接池耗尽', '第三方API故障' ] } } def init_dependency_map(self): """ 初始化依赖关系图 """ return { 'website_availability': ['cdn_health', 'server_health', 'database_health'], 'cdn_health': ['cdn_provider_status', 'network_connectivity'], 'server_health': ['cpu_usage', 'memory_usage', 'disk_io'], 'database_health': ['connection_pool', 'query_performance', 'replication_lag'] } def analyze_root_cause(self, anomaly, traffic_data, system_metrics): """ 分析异常根因 """ analysis_result = { 'anomaly_type': anomaly['type'], 'confidence': 0.0, 'likely_causes': [], 'supporting_evidence': {}, 'recommended_actions': [] } # 基于异常类型选择分析策略 if anomaly['type'] in self.correlation_rules: analysis_result.update( self.correlative_analysis(anomaly, traffic_data, system_metrics) ) # 基于依赖关系的分析 analysis_result.update( self.dependency_analysis(anomaly, system_metrics) ) # 生成推荐行动 analysis_result['recommended_actions'] = self.generate_actions(analysis_result) return analysis_result def correlative_analysis(self, anomaly, traffic_data, system_metrics): """ 相关性分析 """ anomaly_type = anomaly['type'] rules = self.correlation_rules.get(anomaly_type, {}) evidence = {} likely_causes = [] confidence = 0.0 # 检查相关指标 for metric in rules.get('related_metrics', []): metric_value = self.get_metric_value(metric, traffic_data, system_metrics) if self.is_metric_abnormal(metric, metric_value): evidence[metric] = { 'value': metric_value, 'status': 'abnormal' } confidence += 0.2 else: evidence[metric] = { 'value': metric_value, 'status': 'normal' } # 基于证据推断可能原因 if evidence: possible_causes = rules.get('possible_causes', []) # 简单的启发式规则 if 'error_rate' in evidence and evidence['error_rate']['status'] == 'abnormal': likely_causes.append('服务器或应用层故障') if 'response_time' in evidence and evidence['response_time']['status'] == 'abnormal': likely_causes.append('性能瓶颈或资源不足') if 'cdn_status' in evidence and evidence['cdn_status']['status'] == 'abnormal': likely_causes.append('CDN服务异常') return { 'likely_causes': likely_causes, 'supporting_evidence': evidence, 'confidence': min(confidence, 1.0) } def dependency_analysis(self, anomaly, system_metrics): """ 依赖关系分析 """ affected_components = self.identify_affected_components(anomaly) root_cause_candidates = [] for component in affected_components: dependencies = self.dependency_map.get(component, []) for dependency in dependencies: dependency_status = self.get_component_status(dependency, system_metrics) if dependency_status != 'healthy': root_cause_candidates.append({ 'component': dependency, 'status': dependency_status, 'impact': f"影响{component}" }) return { 'dependency_analysis': { 'affected_components': affected_components, 'root_cause_candidates': root_cause_candidates } } def identify_affected_components(self, anomaly): """ 识别受影响组件 """ component_mapping = { 'traffic_drop': ['website_availability', 'cdn_health'], 'high_error_rate': ['server_health', 'database_health'], 'high_response_time': ['server_health', 'database_health', 'cdn_health'] } return component_mapping.get(anomaly['type'], ['website_availability']) def get_metric_value(self, metric_name, traffic_data, system_metrics): """ 获取指标值 """ # 在流量数据中查找 if metric_name in traffic_data.get('basic_metrics', {}): return traffic_data['basic_metrics'][metric_name] elif metric_name in traffic_data.get('performance_metrics', {}): return traffic_data['performance_metrics'][metric_name] # 在系统指标中查找 return system_metrics.get(metric_name, 0) def is_metric_abnormal(self, metric_name, value): """ 判断指标是否异常 """ # 简化的异常判断逻辑 threshold_map = { 'error_rate': 0.05, 'response_time': 500, 'server_load': 0.8 } threshold = threshold_map.get(metric_name, 0) return value > threshold def generate_actions(self, analysis_result): """ 生成推荐行动 """ actions = [] # 基于异常类型的通用行动 if analysis_result['anomaly_type'] == 'traffic_drop': actions.extend([ "检查CDN服务状态", "验证服务器健康状态", "检查网络连通性", "查看DNS解析记录" ]) elif analysis_result['anomaly_type'] == 'high_error_rate': actions.extend([ "检查应用日志中的错误信息", "验证数据库连接状态", "检查第三方API可用性", "监控服务器资源使用情况" ]) # 基于证据的具体行动 evidence = analysis_result.get('supporting_evidence', {}) if 'error_rate' in evidence and evidence['error_rate']['status'] == 'abnormal': actions.append("立即检查应用错误日志并重启问题服务") if 'response_time' in evidence and evidence['response_time']['status'] == 'abnormal': actions.append("优化数据库查询和增加服务器资源") return actions3.5 实时告警与通知系统
class SmartAlertSystem: """ 智能告警系统 """ def __init__(self): self.alert_channels = self.init_alert_channels() self.alert_history = [] self.cooldown_periods = self.init_cooldown_periods() def init_alert_channels(self): """ 初始化告警通道 """ return { 'email': { 'enabled': True, 'config': { 'smtp_server': 'smtp.xxx.com', 'recipients': ['devops@shein.com', 'manager@shein.com'] } }, 'dingtalk': { 'enabled': True, 'config': { 'webhook_url': 'https://oapi.dingtalk.com/robot/send', 'secret': 'your_secret' } }, 'wecom': { 'enabled': True, 'config': { 'webhook_url': 'https://qyapi.weixin.qq.com/cgi-bin/webhook/send', 'key': 'your_key' } }, 'sms': { 'enabled': False, 'config': { 'provider': 'aliyun', 'phone_numbers': ['+8613800138000'] } } } def init_cooldown_periods(self): """ 初始化冷却期配置 """ return { 'critical': 300, # 5分钟 'high': 900, # 15分钟 'medium': 1800, # 30分钟 'low': 3600 # 1小时 } def send_alerts(self, anomalies, root_cause_analysis): """ 发送告警 """ sent_alerts = [] for anomaly in anomalies: # 检查冷却期 if self.is_in_cooldown(anomaly): logging.info(f"⏰ 告警处于冷却期: {anomaly['type']}") continue # 准备告警内容 alert_content = self.prepare_alert_content(anomaly, root_cause_analysis) # 根据严重度选择告警通道 channels = self.select_alert_channels(anomaly['severity']) # 发送告警 for channel in channels: if self.alert_channels[channel]['enabled']: success = self.send_single_alert(channel, alert_content) if success: sent_alerts.append({ 'anomaly': anomaly, 'channel': channel, 'timestamp': datetime.now().isoformat() }) # 记录告警历史 self.record_alert_history(anomaly) return sent_alerts def prepare_alert_content(self, anomaly, root_cause_analysis): """ 准备告警内容 """ severity_emojis = { 'critical': '🚨', 'high': '🔴', 'medium': '🟡', 'low': '🔵' } emoji = severity_emojis.get(anomaly['severity'], '⚪') content = { 'title': f"{emoji} 希音网站流量异常告警 - {anomaly['type']}", 'timestamp': anomaly['timestamp'], 'severity': anomaly['severity'], 'anomaly_details': { 'type': anomaly['type'], 'message': anomaly['message'], 'current_value': anomaly.get('current_value'), 'expected_value': anomaly.get('expected_value'), 'deviation': anomaly.get('deviation') }, 'root_cause_analysis': root_cause_analysis, 'recommended_actions': root_cause_analysis.get('recommended_actions', []) } return content def select_alert_channels(self, severity): """ 选择告警通道 """ channel_mapping = { 'critical': ['dingtalk', 'wecom', 'email', 'sms'], 'high': ['dingtalk', 'wecom', 'email'], 'medium': ['dingtalk', 'email'], 'low': ['email'] } return channel_mapping.get(severity, ['email']) def send_single_alert(self, channel, content): """ 发送单通道告警 """ try: if channel == 'email': return self.send_email_alert(content) elif channel == 'dingtalk': return self.send_dingtalk_alert(content) elif channel == 'wecom': return self.send_wecom_alert(content) elif channel == 'sms': return self.send_sms_alert(content) else: return False except Exception as e: logging.error(f"发送{channel}告警失败: {str(e)}") return False def send_email_alert(self, content): """ 发送邮件告警 """ try: # 这里实现邮件发送逻辑 # 使用SMTP库发送格式化邮件 logging.info(f"📧 邮件告警发送成功: {content['title']}") return True except Exception as e: logging.error(f"邮件告警发送失败: {str(e)}") return False def send_dingtalk_alert(self, content): """ 发送钉钉告警 """ try: # 构建钉钉消息格式 dingtalk_message = { "msgtype": "markdown", "markdown": { "title": content['title'], "text": self.format_dingtalk_content(content) } } # 发送钉钉webhook请求 # requests.post(webhook_url, json=dingtalk_message) logging.info(f"💬 钉钉告警发送成功: {content['title']}") return True except Exception as e: logging.error(f"钉钉告警发送失败: {str(e)}") return False def format_dingtalk_content(self, content): """ 格式化钉钉内容 """ text = f"## {content['title']}\n\n" text += f"**时间**: {content['timestamp']}\n" text += f"**严重度**: {content['severity']}\n\n" text += f"**异常详情**: {content['anomaly_details']['message']}\n\n" if content['recommended_actions']:
