"""Feishu notification module""" import json import requests from datetime import datetime from typing import Dict, List, Optional, Any from ..common.utils import format_currency class FeishuNotifier: """飞书通知器""" def __init__(self, webhook_url: Optional[str] = None, timeout: int = 10, account_details: Optional[Dict] = None): """初始化飞书通知器 Args: webhook_url: 飞书webhook URL timeout: 请求超时时间（秒） account_details: 账户详情字典 """ self.webhook_url = webhook_url self.timeout = timeout self.session = requests.Session() self.account_details = account_details or {} def send_message(self, message: str, webhook_url: Optional[str] = None) -> bool: """发送文本消息 Args: message: 消息内容 webhook_url: webhook URL，如果为None则使用默认URL Returns: 发送是否成功 """ url = webhook_url or self.webhook_url if not url: print("⚠️ 未配置Feishu webhook URL") return False payload = { "msg_type": "text", "content": { "text": message } } return self._send_request(url, payload) def send_card(self, title: str, content: str, color: str = "blue", webhook_url: Optional[str] = None) -> bool: """发送卡片消息 Args: title: 卡片标题 content: 卡片内容 color: 卡片颜色 (blue, green, red, orange) webhook_url: webhook URL Returns: 发送是否成功 """ url = webhook_url or self.webhook_url if not url: print("⚠️ 未配置Feishu webhook URL") return False payload = { "msg_type": "interactive", "card": { "config": { "wide_screen_mode": True }, "header": { "title": { "tag": "plain_text", "content": title }, "template": color }, "elements": [ { "tag": "div", "text": { "tag": "lark_md", "content": content } } ] } } return self._send_request(url, payload) def _generate_top5_data(self, account_results: Dict) -> Dict: """生成Top5数据统计 Args: account_results: 账户分析结果 Returns: 包含Top5数据的字典 """ account_costs = {} service_costs = {} tag_costs = {} # 聚合数据 for account_id, result in account_results.items(): if 'current_data' not in result: continue account_total = 0 for record in result['current_data']: cost = float(record.get('cost', 0)) account_total += cost # 服务成本 service = record.get('product', 'Unknown') service_costs[service] = service_costs.get(service, 0) + cost # 标签成本 (tag_usedby) usedby = record.get('tag_usedby', '').strip() if usedby: tag_costs[usedby] = tag_costs.get(usedby, 0) + cost if account_total > 0: account_costs[account_id] = account_total # 生成Top5列表 top5_accounts = sorted(account_costs.items(), key=lambda x: x[1], reverse=True)[:5] top5_services = sorted(service_costs.items(), key=lambda x: x[1], reverse=True)[:5] top5_tags = sorted(tag_costs.items(), key=lambda x: x[1], reverse=True)[:5] return { 'accounts': top5_accounts, 'services': top5_services, 'tags': top5_tags, 'tag_total': sum(cost for cost in tag_costs.values() if cost > 0) } def send_environment_summary(self, env_name: str, anomalies: List[Dict], account_results: Dict, analysis_date: str, webhook_url: str) -> bool: """发送环境汇总通知（使用表格格式） Args: env_name: 环境名称 anomalies: 异常列表 account_results: 账户分析结果 analysis_date: 分析日期 webhook_url: webhook URL Returns: 发送是否成功 """ # 统计汇总信息 total_anomalies = len(anomalies) critical_count = len([a for a in anomalies if a.get('severity') == 'critical']) warning_count = len([a for a in anomalies if a.get('severity') == 'warning']) affected_accounts = len(set(a.get('account_id') for a in anomalies if a.get('account_id'))) # 计算总成本 total_cost = 0 total_accounts = len(account_results) for result in account_results.values(): if 'current_data' in result: total_cost += sum(float(record.get('cost', 0)) for record in result['current_data']) # 构建环境显示名称 env_display_map = { 'nonprod': 'AWS Dev/Test Accounts', 'prod': 'AWS Production Accounts', 'shared': 'AWS Shared Test Accounts' } env_display = env_display_map.get(env_name, f'AWS {env_name.title()} Accounts') # 构建状态文本 if total_anomalies == 0: status_text = "✅ Normal" header_color = "green" elif critical_count > 0: status_text = f"🚨 {critical_count} Critical, {warning_count} Warning" header_color = "red" else: status_text = f"⚠️ {warning_count} Warning" header_color = "orange" # 构建卡片 card = { "msg_type": "interactive", "card": { "config": {"wide_screen_mode": True}, "header": { "title": {"tag": "plain_text", "content": f"{env_display} - Daily Cost Report"}, "template": header_color }, "elements": [ { "tag": "div", "text": {"tag": "lark_md", "content": "📋 **AWS Daily Cost Report**"} }, { "tag": "div", "fields": [ {"is_short": True, "text": {"tag": "lark_md", "content": f"**Analysis Date**\n{analysis_date}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Status**\n{status_text}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Total Cost**\n{format_currency(total_cost)}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Accounts**\n{total_accounts}"}} ] } ] } } if total_anomalies > 0: card["card"]["elements"].append({"tag": "hr"}) # 分离账户异常和服务异常 account_anomalies = [a for a in anomalies if a.get('type') == 'account'] service_anomalies = [a for a in anomalies if a.get('type') == 'service'] # 账户异常表格 if account_anomalies: card["card"]["elements"].append({ "tag": "div", "text": {"tag": "lark_md", "content": "**🚨 Account Anomalies**"} }) sorted_account_anomalies = sorted( account_anomalies, key=lambda a: ( a.get('increase_amount') if a.get('increase_amount') is not None else a.get('current_cost', 0) ), reverse=True ) account_rows = [] for anomaly in sorted_account_anomalies: severity_icon = "🚨" if anomaly['severity'] == 'critical' else "⚠️" identity = self._get_account_identity(anomaly['account_id']) current_cost = anomaly.get('current_cost', 0) baseline_cost = anomaly.get('baseline_avg', 0) ratio = anomaly.get('ratio') increase_amount = anomaly.get('increase_amount') details = [format_currency(current_cost), f"was {format_currency(baseline_cost)}"] if ratio is not None: details.append(f"{ratio:.1f}x") if increase_amount is not None: delta_sign = '+' if increase_amount >= 0 else '-' details.append(f"{delta_sign}{format_currency(abs(increase_amount))}") alert_text = f"{severity_icon} Total Cost: {', '.join(details)}" account_rows.append({ "account_id": identity['account_id_display'], "account_name": identity['account_name'], "alert": alert_text }) card["card"]["elements"].append({ "tag": "table", "page_size": 10, "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "default", "text_color": "default", "bold": True }, "columns": [ {"name": "account_id", "display_name": "Account ID", "horizontal_align": "left", "data_type": "text"}, {"name": "account_name", "display_name": "Account Name", "horizontal_align": "left", "data_type": "text"}, {"name": "alert", "display_name": "Alert", "horizontal_align": "left", "data_type": "text"} ], "rows": account_rows }) # 服务异常表格 if service_anomalies: card["card"]["elements"].append({ "tag": "div", "text": {"tag": "lark_md", "content": "**⚙️ Service Anomalies**"} }) sorted_service_anomalies = sorted( service_anomalies, key=lambda a: ( a.get('increase_amount') if a.get('increase_amount') is not None else a.get('current_cost', 0) ), reverse=True ) service_rows = [] for anomaly in sorted_service_anomalies: severity_icon = "🚨" if anomaly['severity'] == 'critical' else "⚠️" identity = self._get_account_identity(anomaly['account_id']) current_cost = anomaly.get('current_cost', 0) baseline_cost = anomaly.get('baseline_avg', 0) ratio = anomaly.get('ratio') increase_amount = anomaly.get('increase_amount') details = [format_currency(current_cost), f"was {format_currency(baseline_cost)}"] if ratio is not None: details.append(f"{ratio:.1f}x") if increase_amount is not None: delta_sign = '+' if increase_amount >= 0 else '-' details.append(f"{delta_sign}{format_currency(abs(increase_amount))}") alert_text = f"{severity_icon} {anomaly.get('service', 'Unknown')}: {', '.join(details)}" service_rows.append({ "account_id": identity['account_id_display'], "account_name": identity['account_name'], "alert": alert_text }) card["card"]["elements"].append({ "tag": "table", "page_size": 10, "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "default", "text_color": "default", "bold": True }, "columns": [ {"name": "account_id", "display_name": "Account ID", "horizontal_align": "left", "data_type": "text"}, {"name": "account_name", "display_name": "Account Name", "horizontal_align": "left", "data_type": "text"}, {"name": "alert", "display_name": "Alert", "horizontal_align": "left", "data_type": "text"} ], "rows": service_rows }) # 添加Top5数据展示（始终显示） card["card"]["elements"].append({"tag": "hr"}) top5_data = self._generate_top5_data(account_results) # Top5账户成本 if top5_data['accounts']: card["card"]["elements"].append({ "tag": "div", "text": {"tag": "lark_md", "content": "**💰 Top 5 Account Costs (Daily)**"} }) account_rows = [] for _, (account_id, cost) in enumerate(top5_data['accounts'], 1): identity = self._get_account_identity(account_id) percent = ( f"{(cost / total_cost * 100):.1f}%" if total_cost > 0 else "0%" ) account_rows.append({ "account_id": identity['account_id_display'], "account_name": identity['account_name'], "cost": format_currency(cost), "percent": percent }) card["card"]["elements"].append({ "tag": "table", "page_size": 5, "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "default", "text_color": "default", "bold": True }, "columns": [ {"name": "account_id", "display_name": "Account ID", "horizontal_align": "left", "data_type": "text"}, {"name": "account_name", "display_name": "Account Name", "horizontal_align": "left", "data_type": "text"}, {"name": "cost", "display_name": "Daily Cost", "horizontal_align": "right", "data_type": "text"}, {"name": "percent", "display_name": "Percent", "horizontal_align": "right", "data_type": "text"} ], "rows": account_rows }) # Top5服务成本 if top5_data['services']: card["card"]["elements"].append({ "tag": "div", "text": {"tag": "lark_md", "content": "**⚙️ Top 5 Service Costs (Daily)**"} }) service_rows = [] for service, cost in top5_data['services']: percent = f"{(cost / total_cost * 100):.1f}%" if total_cost > 0 else "0%" service_rows.append({ "service": service or "Unknown", "cost": format_currency(cost), "percent": percent }) card["card"]["elements"].append({ "tag": "table", "page_size": 5, "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "default", "text_color": "default", "bold": True }, "columns": [ {"name": "service", "display_name": "Service", "horizontal_align": "left", "data_type": "text"}, {"name": "cost", "display_name": "Daily Cost", "horizontal_align": "right", "data_type": "text"}, {"name": "percent", "display_name": "Percent", "horizontal_align": "right", "data_type": "text"} ], "rows": service_rows }) # Top5标签成本 (tag_usedby) if top5_data['tags']: card["card"]["elements"].append({ "tag": "div", "text": {"tag": "lark_md", "content": "**🏷️ Top 5 Tag Costs (tag_usedby)**"} }) # 仅统计正向成本避免因未打标签的抵扣导致占比超过100% tagged_total_cost = top5_data.get('tag_total', 0) tag_rows = [] for tag, cost in top5_data['tags']: if tagged_total_cost > 0: percent_value = cost / tagged_total_cost elif total_cost > 0: percent_value = cost / total_cost else: percent_value = 0 percent = f"{percent_value * 100:.1f}%" tag_rows.append({ "tag": tag or "Unknown", "cost": format_currency(cost), "percent": percent }) card["card"]["elements"].append({ "tag": "table", "page_size": 5, "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "default", "text_color": "default", "bold": True }, "columns": [ {"name": "tag", "display_name": "Tag (tag_usedby)", "horizontal_align": "left", "data_type": "text"}, {"name": "cost", "display_name": "Daily Cost", "horizontal_align": "right", "data_type": "text"}, {"name": "percent", "display_name": "Percent of Tagged Spend", "horizontal_align": "right", "data_type": "text"} ], "rows": tag_rows }) return self._send_request(webhook_url, card) def send_business_type_summary(self, analysis_date: str, summary: Dict[str, Any], webhook_url: str) -> bool: """发送业务类型标签成本汇总通知""" if not summary or not summary.get('enabled'): return False card = { "msg_type": "interactive", "card": { "config": {"wide_screen_mode": True}, "header": { "title": { "tag": "plain_text", "content": "AWS Production Accounts - Tagged Business Type Daily Report" }, "template": "blue" }, "elements": [ { "tag": "div", "text": { "tag": "lark_md", "content": f"📌 **Tagged Business Type Spend**\nAnalysis Date: {analysis_date}" } } ] } } if not summary.get('has_tag_data', True): cutover_date = summary.get('cutover_date', '2025-10-01') card["card"]["elements"].append({ "tag": "div", "text": { "tag": "lark_md", "content": f"ℹ️ Tag data is incomplete before {cutover_date}; no tagged spend found for this day." } }) return self._send_request(webhook_url, card) type_rows = [] for entry in summary.get('types', []): account_count = len(entry.get('accounts', [])) type_rows.append({ "business_type": entry.get('name', 'Unknown'), "total_cost": format_currency(entry.get('total_cost', 0)), "accounts": str(account_count) }) if type_rows: card["card"]["elements"].append({ "tag": "hr" }) card["card"]["elements"].append({ "tag": "div", "text": {"tag": "lark_md", "content": "**Tagged Business Type Spend (Daily)**"} }) card["card"]["elements"].append({ "tag": "table", "page_size": min(len(type_rows), 10), "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "default", "text_color": "default", "bold": True }, "columns": [ {"name": "business_type", "display_name": "Business Type", "horizontal_align": "left", "data_type": "text"}, {"name": "total_cost", "display_name": "Daily Cost", "horizontal_align": "right", "data_type": "text"}, {"name": "accounts", "display_name": "Accounts", "horizontal_align": "right", "data_type": "text"} ], "rows": type_rows }) control_breakdown = summary.get('control_servicetype_breakdown', {}) or {} breakdown_lines = [] for control_type in ['Global Control', 'Regional Control']: items = control_breakdown.get(control_type) or {} if not items: continue sorted_items = sorted(items.items(), key=lambda item: item[1], reverse=True) detail_lines = ", ".join( f"{name}: {format_currency(cost)}" for name, cost in sorted_items ) breakdown_lines.append(f"- **{control_type} servicetype**: {detail_lines}") if breakdown_lines: card["card"]["elements"].append({ "tag": "div", "text": { "tag": "lark_md", "content": "**Control Servicetype Breakdown**\n" + "\n".join(breakdown_lines) } }) per_type_limit = 5 account_rows = [] for entry in summary.get('types', []): accounts = entry.get('accounts', []) or [] for account_id, cost in accounts[:per_type_limit]: identity = self._get_account_identity(account_id) account_rows.append({ "business_type": entry.get('name', 'Unknown'), "account_id": identity['account_id_display'], "account_name": identity['account_name'], "cost": format_currency(cost) }) if account_rows: card["card"]["elements"].append({"tag": "hr"}) card["card"]["elements"].append({ "tag": "div", "text": { "tag": "lark_md", "content": f"**Top {per_type_limit} Accounts per Business Type**" } }) card["card"]["elements"].append({ "tag": "table", "page_size": min(len(account_rows), 30), "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "default", "text_color": "default", "bold": True }, "columns": [ {"name": "business_type", "display_name": "Business Type", "horizontal_align": "left", "data_type": "text"}, {"name": "account_id", "display_name": "Account ID", "horizontal_align": "left", "data_type": "text"}, {"name": "account_name", "display_name": "Account Name", "horizontal_align": "left", "data_type": "text"}, {"name": "cost", "display_name": "Daily Cost", "horizontal_align": "right", "data_type": "text"} ], "rows": account_rows }) return self._send_request(webhook_url, card) def _get_account_identity(self, account_id: str) -> Dict[str, str]: """构建账户展示信息""" default_name = f'Account-{account_id}' default_email = 'unknown@pingcap.com' account_name = default_name owner_email = default_email shared_tag = '' account_info = self.account_details.get(account_id) if account_info: if hasattr(account_info, 'account_name'): account_name = account_info.account_name or default_name metadata = getattr(account_info, 'metadata', {}) or {} owner_email = getattr(account_info, 'owner_email', None) \ or metadata.get('owner_email') \ or metadata.get('email') \ or default_email if getattr(account_info, 'is_shared', False): shared_tag = ' [Public Test Account]' else: account_name = account_info.get('account_name', default_name) owner_email = account_info.get('owner_email') \ or account_info.get('email') \ or default_email shared_value = str(account_info.get('is_shared_dev_account', '')).strip().lower() if shared_value in {'yes', 'true', '1', 'y'}: shared_tag = ' [Public Test Account]' return { 'account_id': account_id, 'account_id_display': f'{account_id}', 'account_name': f"{account_name}{shared_tag}", 'owner_email': owner_email } def _get_account_display_name(self, account_id: str) -> str: """获取账户显示名称""" identity = self._get_account_identity(account_id) return ( f"{identity['account_id_display']} **{identity['account_name']}**" f" ({identity['owner_email']})" ) def send_anomaly_alert(self, anomalies: List[Dict], account_id: str, environment: str, analysis_date: str, webhook_url: Optional[str] = None) -> bool: """发送异常告警 Args: anomalies: 异常列表 account_id: 账户ID environment: 环境名称 analysis_date: 分析日期 webhook_url: webhook URL Returns: 发送是否成功 """ if not anomalies: return True # 按严重程度分组 critical_anomalies = [a for a in anomalies if a.get('severity') == 'critical'] warning_anomalies = [a for a in anomalies if a.get('severity') == 'warning'] # 确定告警级别和颜色 if critical_anomalies: alert_level = "🚨 严重告警" color = "red" elif warning_anomalies: alert_level = "⚠️ 警告告警" color = "orange" else: alert_level = "ℹ️ 信息通知" color = "blue" # 构建标题 title = f"{alert_level} - AWS 成本异常检测" # 构建内容 identity = self._get_account_identity(account_id) content_lines = [ f"**账户**: {identity['account_name']} ({identity['account_id_display']})", f"**环境**: {environment}", f"**分析日期**: {analysis_date}", f"**异常数量**: {len(anomalies)} 个", "" ] # 添加异常详情 if critical_anomalies: content_lines.append("**🔴 严重异常:**") for anomaly in critical_anomalies[:5]: # 最多显示5个 content_lines.append(self._format_anomaly(anomaly)) content_lines.append("") if warning_anomalies: content_lines.append("**🟡 警告异常:**") for anomaly in warning_anomalies[:5]: # 最多显示5个 content_lines.append(self._format_anomaly(anomaly)) content_lines.append("") # 添加汇总信息 total_current = sum(a.get('current_cost', 0) for a in anomalies) total_baseline = sum(a.get('baseline_avg', 0) for a in anomalies) content_lines.extend([ "**汇总信息:**", f"- 当前总成本: {format_currency(total_current)}", f"- 基线总成本: {format_currency(total_baseline)}", f"- 增加金额: {format_currency(total_current - total_baseline)}" ]) content = "\n".join(content_lines) return self.send_card(title, content, color, webhook_url) def send_summary_report(self, report_data: Dict, webhook_url: Optional[str] = None) -> bool: """发送汇总报告 Args: report_data: 报告数据 webhook_url: webhook URL Returns: 发送是否成功 """ title = f"📊 AWS 成本分析报告 - {report_data.get('analysis_date', '')}" content_lines = [ f"**分析范围**: {report_data.get('environments', [])}", f"**账户数量**: {report_data.get('account_count', 0)}", "" ] # 成本汇总 if 'cost_summary' in report_data: cost_summary = report_data['cost_summary'] content_lines.extend([ "**成本汇总:**", f"- 总成本: {format_currency(cost_summary.get('total_cost', 0))}", f"- 平均成本: {format_currency(cost_summary.get('avg_cost', 0))}", f"- 最高成本账户: {cost_summary.get('top_account', 'N/A')}", "" ]) # 异常汇总 if 'anomaly_summary' in report_data: anomaly_summary = report_data['anomaly_summary'] content_lines.extend([ "**异常汇总:**", f"- 严重异常: {anomaly_summary.get('critical', 0)} 个", f"- 警告异常: {anomaly_summary.get('warning', 0)} 个", f"- 受影响账户: {anomaly_summary.get('affected_accounts', 0)} 个", "" ]) # TOP异常 if 'top_anomalies' in report_data: content_lines.append("**TOP 异常:**") for anomaly in report_data['top_anomalies'][:3]: content_lines.append(self._format_anomaly(anomaly)) content = "\n".join(content_lines) return self.send_card(title, content, "blue", webhook_url) def _format_anomaly(self, anomaly: Dict) -> str: """格式化异常信息 Args: anomaly: 异常字典 Returns: 格式化的异常描述 """ anomaly_type = anomaly.get('type', 'unknown') if anomaly_type == 'account': identity = self._get_account_identity(anomaly.get('account_id', 'Unknown')) return ( f"- {identity['account_name']} ({identity['account_id_display']}): " f"{format_currency(anomaly.get('current_cost', 0))} " f"({anomaly.get('increase_percentage', 0):.1f}% ↑)" ) elif anomaly_type == 'service': identity = self._get_account_identity(anomaly.get('account_id', 'Unknown')) return ( f"- {identity['account_name']} ({identity['account_id_display']}) / {anomaly.get('service', 'Unknown')}: " f"{format_currency(anomaly.get('current_cost', 0))} " f"({anomaly.get('increase_percentage', 0):.1f}% ↑)" ) elif anomaly_type == 'tag': return (f"- 标签 **{anomaly['tag_value']}**: " f"{format_currency(anomaly['current_cost'])} " f"({anomaly['increase_percentage']:.1f}% ↑)") else: return f"- 未知异常类型: {anomaly}" def send_monthly_cost_summary(self, env_name: str, monthly_costs: Dict, analysis_date: str, webhook_url: str) -> bool: """发送月度成本汇总通知（独立通知） Args: env_name: 环境名称 monthly_costs: 月度成本汇总数据 analysis_date: 分析日期 webhook_url: webhook URL Returns: 发送是否成功 """ if not monthly_costs: return False # 构建环境显示名称 env_display_map = { 'nonprod': 'Dev/Test', 'prod': 'Production', 'shared': 'Shared Test' } env_display = env_display_map.get(env_name, env_name.title()) # 构建卡片标题 title = f"📊 AWS {env_display} Monthly Cost Report" # 构建卡片 card = { "msg_type": "interactive", "card": { "schema": "2.0", "header": { "title": {"tag": "plain_text", "content": title}, "template": "blue" }, "body": { "elements": [ { "tag": "div", "text": {"tag": "lark_md", "content": f"**📅 Month-to-Date Cost Summary**\n{monthly_costs['start_date']} to {monthly_costs['end_date']}"} }, { "tag": "div", "fields": [ {"is_short": True, "text": {"tag": "lark_md", "content": f"**Total Cost**\n{format_currency(monthly_costs['total_cost'])}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Total Accounts**\n{len(monthly_costs.get('account_costs', {}))}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Analysis Date**\n{analysis_date}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Environment**\n{env_display}"}} ] } ] } } } # 月度Top账户成本表格 sorted_accounts = monthly_costs.get('sorted_accounts', []) if sorted_accounts: card["card"]["body"]["elements"].append({"tag": "hr"}) card["card"]["body"]["elements"].append({ "tag": "div", "text": {"tag": "lark_md", "content": "**💰 Top Accounts by MTD Cost**"} }) account_rows = [] total_monthly_cost = monthly_costs.get('total_cost', 0) for i, (account_id, cost) in enumerate(sorted_accounts[:15], 1): # 显示Top 15 identity = self._get_account_identity(account_id) account_rows.append({ "account_id": identity['account_id_display'], "account_name": identity['account_name'], "cost": format_currency(cost), "percentage": f"{(cost / total_monthly_cost * 100):.1f}%" if total_monthly_cost > 0 else "0%" }) # 确保表格有有效数据 if account_rows: card["card"]["body"]["elements"].append({ "tag": "table", "page_size": 15, "row_height": "auto", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "default", "text_color": "default", "bold": True, "lines": 1 }, "columns": [ { "name": "account_id", "display_name": "Account ID", "data_type": "text", "horizontal_align": "left", "vertical_align": "top" }, { "name": "account_name", "display_name": "Account Name", "data_type": "text", "horizontal_align": "left", "vertical_align": "top" }, { "name": "cost", "display_name": "MTD Cost", "data_type": "text", "horizontal_align": "right", "vertical_align": "top" }, { "name": "percentage", "display_name": "Percent", "data_type": "text", "horizontal_align": "right", "vertical_align": "top" } ], "rows": account_rows }) else: card["card"]["body"]["elements"].append({ "tag": "div", "text": {"tag": "lark_md", "content": "暂无账户成本数据"} }) # 月度Top服务成本 sorted_services = monthly_costs.get('sorted_services', []) if sorted_services: card["card"]["body"]["elements"].append({"tag": "hr"}) card["card"]["body"]["elements"].append({ "tag": "div", "text": {"tag": "lark_md", "content": "**⚙️ Top Services by MTD Cost**"} }) service_rows = [] for i, (service, cost) in enumerate(sorted_services[:10], 1): service_rows.append({ "service": str(service), "cost": str(format_currency(cost)), "percentage": str(f"{(cost / monthly_costs['total_cost'] * 100):.1f}%" if monthly_costs['total_cost'] > 0 else "0%") }) if service_rows: card["card"]["body"]["elements"].append({ "tag": "table", "page_size": 10, "row_height": "auto", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "default", "text_color": "default", "bold": True, "lines": 1 }, "columns": [ { "name": "service", "display_name": "Service", "data_type": "text", "horizontal_align": "left", "vertical_align": "top" }, { "name": "cost", "display_name": "MTD Cost", "data_type": "text", "horizontal_align": "right", "vertical_align": "top" }, { "name": "percentage", "display_name": "Percent", "data_type": "text", "horizontal_align": "right", "vertical_align": "top" } ], "rows": service_rows }) else: card["card"]["body"]["elements"].append({ "tag": "div", "text": {"tag": "lark_md", "content": "暂无服务成本数据"} }) # 添加成本趋势说明 card["card"]["body"]["elements"].append({"tag": "hr"}) card["card"]["body"]["elements"].append({ "tag": "div", "text": { "tag": "lark_md", "content": f"📈 **Daily Average**: {format_currency(monthly_costs['total_cost'] / int(monthly_costs['end_date'].split('-')[2]))}" } }) return self._send_request(webhook_url, card) def send_dev_test_region_monitor(self, env_name: str, monitor_data: Dict[str, Any], analysis_date: str, webhook_url: str) -> bool: """发送开发/测试账户区域费用监控通知""" env_display_map = { 'nonprod': 'AWS Dev/Test Accounts', 'shared': 'AWS Shared Test Accounts', 'prod': 'AWS Production Accounts' } env_display = env_display_map.get(env_name, f'AWS {env_name.title()} Accounts') total_accounts = monitor_data.get('total_accounts', 0) abnormal_accounts = monitor_data.get('abnormal_accounts', 0) involved_regions = monitor_data.get('regions', []) or [] allowed_regions = monitor_data.get('common_regions_display') or monitor_data.get('common_regions') or [] details = monitor_data.get('details', []) or [] top_regions = monitor_data.get('top_regions', []) or [] top_regions_mtd = monitor_data.get('top_regions_mtd', []) or [] has_region_data = monitor_data.get('has_region_data', False) threshold = monitor_data.get('threshold', 0) if not allowed_regions: allowed_regions = ['us-east-1', 'us-west-2', 'global'] allowed_regions_str = ', '.join(allowed_regions) involved_regions_str = ', '.join(involved_regions) if involved_regions else '-' if not has_region_data: status_text = "ℹ️ No region data available" header_color = "blue" elif abnormal_accounts == 0: status_text = "✅ Within allowed regions" header_color = "green" else: status_text = f"⚠️ {abnormal_accounts} accounts outside allowed regions" header_color = "orange" card = { "msg_type": "interactive", "card": { "config": {"wide_screen_mode": True}, "header": { "title": {"tag": "plain_text", "content": f"{env_display} - Region Spend Monitor"}, "template": header_color }, "elements": [ { "tag": "div", "text": {"tag": "lark_md", "content": "🛰️ **Dev/Test Region Cost Watch**"} }, { "tag": "div", "fields": [ {"is_short": True, "text": {"tag": "lark_md", "content": f"**Analysis Date**\n{analysis_date}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Status**\n{status_text}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Accounts**\n{total_accounts}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Abnormal Accounts**\n{abnormal_accounts}"}}, {"is_short": False, "text": {"tag": "lark_md", "content": f"**Allowed Regions**\n{allowed_regions_str}"}}, {"is_short": False, "text": {"tag": "lark_md", "content": f"**Involved Regions**\n{involved_regions_str}"}} ] } ] } } if details and has_region_data: card["card"]["elements"].append({"tag": "hr"}) card["card"]["elements"].append({ "tag": "div", "text": {"tag": "lark_md", "content": "**🚨 Non-standard Region Spend**"} }) rows = [] for entry in details: identity = self._get_account_identity(entry['account_id']) rows.append({ "account_id": identity['account_id_display'], "account_name": identity['account_name'], "region": entry['region'], "cost": format_currency(entry['cost']) }) card["card"]["elements"].append({ "tag": "table", "page_size": 10, "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "default", "text_color": "default", "bold": True }, "columns": [ {"name": "account_id", "display_name": "Account ID", "horizontal_align": "left", "data_type": "text"}, {"name": "account_name", "display_name": "Account Name", "horizontal_align": "left", "data_type": "text"}, {"name": "region", "display_name": "Region", "horizontal_align": "left", "data_type": "text"}, {"name": "cost", "display_name": "Daily Cost", "horizontal_align": "right", "data_type": "text"} ], "rows": rows }) elif has_region_data: card["card"]["elements"].append({"tag": "hr"}) card["card"]["elements"].append({ "tag": "div", "text": {"tag": "lark_md", "content": "✅ All dev/test accounts stayed within common regions (us-east-1 / us-west-2 / global)"} }) else: card["card"]["elements"].append({"tag": "hr"}) card["card"]["elements"].append({ "tag": "div", "text": {"tag": "lark_md", "content": "ℹ️ Current export is missing the `region` field, so distribution cannot be calculated"} }) if top_regions: card["card"]["elements"].append({"tag": "hr"}) card["card"]["elements"].append({ "tag": "div", "text": {"tag": "lark_md", "content": "**📊 Region Daily Cost Top 5 (Non-standard)**"} }) region_rows = [ {"region": region or 'Unknown', "cost": format_currency(cost)} for region, cost in top_regions ] card["card"]["elements"].append({ "tag": "table", "page_size": 5, "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "default", "text_color": "default", "bold": True }, "columns": [ {"name": "region", "display_name": "Region", "horizontal_align": "left", "data_type": "text"}, {"name": "cost", "display_name": "Daily Cost", "horizontal_align": "right", "data_type": "text"} ], "rows": region_rows }) if top_regions_mtd: card["card"]["elements"].append({"tag": "hr"}) card["card"]["elements"].append({ "tag": "div", "text": {"tag": "lark_md", "content": "**🗓️ Month-to-Date Non-standard Spend Top 5**"} }) mtd_rows = [] for entry in top_regions_mtd: identity = self._get_account_identity(entry['account_id']) mtd_rows.append({ "account_id": identity['account_id_display'], "account_name": identity['account_name'], "region": entry['region'] or 'Unknown', "cost": format_currency(entry['cost']) }) card["card"]["elements"].append({ "tag": "table", "page_size": 5, "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "default", "text_color": "default", "bold": True }, "columns": [ {"name": "account_id", "display_name": "Account ID", "horizontal_align": "left", "data_type": "text"}, {"name": "account_name", "display_name": "Account Name", "horizontal_align": "left", "data_type": "text"}, {"name": "region", "display_name": "Region", "horizontal_align": "left", "data_type": "text"}, {"name": "cost", "display_name": "MTD Cost", "horizontal_align": "right", "data_type": "text"} ], "rows": mtd_rows }) card["card"]["elements"].append({"tag": "hr"}) card["card"]["elements"].append({ "tag": "div", "text": { "tag": "lark_md", "content": ( f"> Monitoring note: tracks dev/test spend outside us-east-1, us-west-2, and global, " f"and alerts when a single account-region daily cost ≥ {format_currency(threshold)}." ) } }) return self._send_request(webhook_url, card) def send_untagged_service_monitor(self, analysis_date: str, environments: List[str], services: List[str], rows: List[Dict[str, Any]], webhook_url: str) -> bool: """发送不可打标签服务监控通知""" if not rows or not services: return False def display_service_name(service_name: str) -> str: if service_name.startswith("AWS "): return service_name[4:] if service_name.startswith("Amazon "): return service_name[7:] return service_name env_display = ", ".join(env.upper() for env in environments if env) or "PROD" title = f"AWS Account-Scoped Services Daily Cost Monitor - {env_display}" critical_accounts = 0 warning_accounts = 0 for row in rows: has_critical = False has_warning = False for service in services: cell = row.get('service_cells', {}).get(service, '') if "🚨" in cell: has_critical = True elif "⚠️" in cell: has_warning = True if has_critical: critical_accounts += 1 elif has_warning: warning_accounts += 1 if critical_accounts > 0: status_text = f"🚨 {critical_accounts} Critical, {warning_accounts} Warning" header_color = "red" elif warning_accounts > 0: status_text = f"⚠️ {warning_accounts} Warning" header_color = "orange" else: status_text = "✅ Normal" header_color = "green" card = { "msg_type": "interactive", "card": { "config": {"wide_screen_mode": True}, "header": { "title": {"tag": "plain_text", "content": title}, "template": header_color }, "elements": [ { "tag": "div", "text": {"tag": "lark_md", "content": "🧩 **Account-scoped Service Cost Watch**"} }, { "tag": "div", "fields": [ {"is_short": True, "text": {"tag": "lark_md", "content": f"**Analysis Date**\n{analysis_date}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Status**\n{status_text}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Accounts**\n{len(rows)}"}}, {"is_short": False, "text": {"tag": "lark_md", "content": f"**Services**\n{', '.join(display_service_name(service) for service in services)}"}} ] } ] } } table_columns = [ {"name": "account_id", "display_name": "Account ID", "horizontal_align": "left", "data_type": "text"}, {"name": "account_name", "display_name": "Account Name", "horizontal_align": "left", "data_type": "text"}, {"name": "total_cost", "display_name": "Total Cost", "horizontal_align": "right", "data_type": "text"} ] for service in services: table_columns.append({ "name": service, "display_name": display_service_name(service), "horizontal_align": "left", "data_type": "text" }) table_rows = [] for row in rows: identity = self._get_account_identity(row['account_id']) row_data = { "account_id": identity['account_id_display'], "account_name": identity['account_name'], "total_cost": format_currency(row.get('total_cost', 0)) } service_cells = row.get('service_cells', {}) for service in services: row_data[service] = service_cells.get(service, "-") table_rows.append(row_data) card["card"]["elements"].append({"tag": "hr"}) card["card"]["elements"].append({ "tag": "table", "page_size": min(len(table_rows), 20), "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "default", "text_color": "default", "bold": True }, "columns": table_columns, "rows": table_rows }) card["card"]["elements"].append({"tag": "hr"}) card["card"]["elements"].append({ "tag": "div", "text": { "tag": "lark_md", "content": ( "Legend: ✅ within baseline, ⚠️ warning, 🚨 critical, " "ℹ️ below threshold or no baseline." ) } }) return self._send_request(webhook_url, card) def send_untagged_tag_monitor(self, analysis_date: str, environments: List[str], rows: List[Dict[str, Any]], total_untagged_cost: float, webhook_url: str) -> bool: """发送漏打标签资源监控通知""" if not rows: return False env_display = ", ".join(env.upper() for env in environments if env) or "PROD" title = f"AWS Untagged Resource Monitor - {env_display}" card = { "msg_type": "interactive", "card": { "config": {"wide_screen_mode": True}, "header": { "title": {"tag": "plain_text", "content": title}, "template": "red" }, "elements": [ { "tag": "div", "text": {"tag": "lark_md", "content": "🏷️ **Untagged Resource Cost Alert**"} }, { "tag": "div", "fields": [ {"is_short": True, "text": {"tag": "lark_md", "content": f"**Analysis Date**\n{analysis_date}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": "**Status**\n🚨 Untagged resources detected"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Accounts**\n{len({row['account_id'] for row in rows})}"}}, {"is_short": True, "text": {"tag": "lark_md", "content": f"**Total Untagged Cost**\n{format_currency(total_untagged_cost)}"}} ] }, { "tag": "div", "text": { "tag": "lark_md", "content": "Missing `tag_usedby` and `tag_servicetype` on tagged resource records." } } ] } } grouped_rows: Dict[str, List[Dict[str, Any]]] = {} for row in rows: grouped_rows.setdefault(row['account_id'], []).append(row) for account_id, account_rows in grouped_rows.items(): identity = self._get_account_identity(account_id) account_untagged_cost = sum( row.get('untagged_cost', 0.0) for row in account_rows ) card["card"]["elements"].append({"tag": "hr"}) card["card"]["elements"].append({ "tag": "div", "text": { "tag": "lark_md", "content": ( f"**{identity['account_name']}** " f"({identity['account_id_display']})\n" f"Untagged: {format_currency(account_untagged_cost)}" ) } }) table_rows = [] for row in account_rows: untagged_cost = row.get('untagged_cost', 0.0) table_rows.append({ "service": row.get('product', 'Unknown'), "untagged_cost": format_currency(untagged_cost) }) card["card"]["elements"].append({ "tag": "table", "page_size": min(len(table_rows), 20), "row_height": "low", "header_style": { "text_align": "left", "text_size": "normal", "background_style": "default", "text_color": "default", "bold": True }, "columns": [ {"name": "service", "display_name": "Service", "horizontal_align": "left", "data_type": "text"}, {"name": "untagged_cost", "display_name": "Untagged Cost", "horizontal_align": "right", "data_type": "text"} ], "rows": table_rows }) return self._send_request(webhook_url, card) def _send_request(self, url: str, payload: Dict) -> bool: """发送HTTP请求 Args: url: 请求URL payload: 请求负载 Returns: 请求是否成功 """ try: response = self.session.post( url, json=payload, timeout=self.timeout, headers={'Content-Type': 'application/json'} ) if response.status_code == 200: result = response.json() if result.get('StatusCode') == 0: print("✓ 飞书通知发送成功") return True else: print(f"✗ 飞书通知发送失败: {result.get('msg', 'Unknown error')}") return False else: print(f"✗ 飞书通知请求失败: HTTP {response.status_code}") return False except requests.exceptions.Timeout: print("✗ 飞书通知发送超时") return False except requests.exceptions.RequestException as e: print(f"✗ 飞书通知请求异常: {e}") return False except Exception as e: print(f"✗ 飞书通知发送失败: {e}") return False def test_connection(self, webhook_url: Optional[str] = None) -> bool: """测试连接 Args: webhook_url: webhook URL Returns: 连接是否正常 """ test_message = f"测试消息 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}" return self.send_message(test_message, webhook_url)