#!/usr/bin/env python3 """Read-only Alibaba Cloud asset and alert inventory exporter. The script uses the locally configured Alibaba Cloud CLI identity. It never creates, updates, restarts, or deletes cloud resources. Detailed results are intended for a private local folder because they can contain resource IDs and IP addresses. """ from __future__ import annotations import argparse import collections import datetime as dt import html import json import re import subprocess import sys from pathlib import Path from typing import Any from openpyxl import Workbook, load_workbook from openpyxl.styles import Alignment, Border, Font, PatternFill, Side from openpyxl.utils import get_column_letter TZ = dt.timezone(dt.timedelta(hours=8)) NOW = dt.datetime.now(TZ) NAMESPACE_ASSET_TYPES = { "acs_ecs_dashboard": ["ACS::ECS::Instance"], "acs_rds_dashboard": ["ACS::RDS::DBInstance"], "acs_oss_dashboard": ["ACS::OSS::Bucket"], "acs_slb_dashboard": ["ACS::SLB::LoadBalancer"], "acs_kvstore": ["ACS::Redis::DBInstance"], "acs_mqtt": ["ACS::OnsMqtt::Instance"], "acs_sls_dashboard": ["ACS::SLS::Project", "ACS::SLS::LogStore"], "acs_serverless": ["ACS::FC::Service", "ACS::FC::Function", "ACS::FCV3::Function"], "acs_hbr": ["ACS::HBR::Vault"], "acs_mongodb": ["ACS::MongoDB::DBInstance"], "acs_hbase": ["ACS::HBase::Cluster"], "acs_elasticsearch": ["ACS::Elasticsearch::Instance"], "acs_opensearch": ["ACS::OpenSearch::AppGroup"], "acs_iot": ["ACS::IOT::Instance"], "acs_hologres": ["ACS::Hologres::Instance"], "acs_drds": ["ACS::DRDS::DrdsInstance"], "acs_mns_new": ["ACS::MNS::Queue", "ACS::MNS::Topic"], } PRODUCT_CN = { "ECS": "云服务器 ECS", "RDS": "云数据库 RDS", "OSS": "对象存储 OSS", "Redis": "Tair / Redis", "SLB": "传统型负载均衡 CLB", "OnsMqtt": "微消息队列 MQTT", "SLS": "日志服务 SLS", "VPC": "专有网络 VPC", "RAM": "访问控制 RAM", "ARMS": "应用实时监控 ARMS", "Cms": "云监控 CMS", "FC": "函数计算 FC", "FCV3": "函数计算 FC", "SWAS": "轻量应用服务器", "NAS": "文件存储 NAS", "OTS": "表格存储 OTS", "CDN": "内容分发网络 CDN", "DCDN": "全站加速 DCDN", "Alidns": "云解析 DNS", "Domain": "域名", } class Collector: def __init__(self) -> None: self.errors: list[dict[str, str]] = [] def json(self, *args: str, label: str | None = None) -> dict[str, Any]: cmd = ["aliyun", *args] try: proc = subprocess.run(cmd, check=True, text=True, capture_output=True) return json.loads(proc.stdout) except (subprocess.CalledProcessError, json.JSONDecodeError) as exc: stderr = getattr(exc, "stderr", "") or str(exc) self.errors.append({"label": label or " ".join(args[:2]), "error": sanitize_error(stderr)}) return {} def text(self, *args: str, label: str | None = None) -> str: cmd = ["aliyun", *args] try: return subprocess.run(cmd, check=True, text=True, capture_output=True).stdout except subprocess.CalledProcessError as exc: self.errors.append({"label": label or " ".join(args[:2]), "error": sanitize_error(exc.stderr or str(exc))}) return "" def sanitize_error(value: str) -> str: """Remove signed-request credentials from CLI diagnostics before storage.""" text = value.strip() for key in ("AccessKeyId", "SecurityToken", "Signature"): text = re.sub(rf"({key}=)[^&\s\"]+", rf"\1[REDACTED]", text, flags=re.IGNORECASE) text = re.sub(r"STS\.[A-Za-z0-9]+", "STS.[REDACTED]", text) return text[:1200] def epoch_ms(value: Any) -> str: if value in (None, "", 0): return "" try: return dt.datetime.fromtimestamp(int(value) / 1000, TZ).strftime("%Y-%m-%d %H:%M:%S") except (TypeError, ValueError, OSError): return str(value) def flat_json(value: Any) -> str: if value in (None, "", [], {}): return "" return json.dumps(value, ensure_ascii=False, separators=(",", ":")) def values_from_json_text(value: str) -> set[str]: if not value: return set() try: parsed = json.loads(value) except (TypeError, json.JSONDecodeError): return set() found: set[str] = set() def walk(item: Any) -> None: if isinstance(item, dict): for v in item.values(): walk(v) elif isinstance(item, list): for v in item: walk(v) elif isinstance(item, (str, int, float)): found.add(str(item)) walk(parsed) return found def collect_resource_center(c: Collector) -> list[dict[str, Any]]: resources: list[dict[str, Any]] = [] token = "" while True: args = ["resourcecenter", "SearchResources", "--endpoint", "resourcecenter.aliyuncs.com", "--MaxResults", "500"] if token: args += ["--NextToken", token] page = c.json(*args, label="ResourceCenter.SearchResources") resources.extend(page.get("Resources") or []) token = page.get("NextToken") or "" if not token: break return resources def collect_page_number( c: Collector, base: list[str], list_path: tuple[str, ...], total_key: str, page_key: str = "--PageNumber", size_key: str = "--PageSize", size: int = 100, ) -> list[dict[str, Any]]: result: list[dict[str, Any]] = [] for page_num in range(1, 501): data = c.json(*base, page_key, str(page_num), size_key, str(size), label=".".join(base[:2])) current: Any = data for key in list_path: current = current.get(key, {}) if isinstance(current, dict) else {} rows = current if isinstance(current, list) else [] result.extend(rows) total = int(data.get(total_key) or len(result)) if not rows or len(result) >= total or len(rows) < size: break return result def collect_metric_rules(c: Collector) -> list[dict[str, Any]]: return collect_page_number( c, ["cms", "DescribeMetricRuleList"], ("Alarms", "Alarm"), "Total", page_key="--Page", size_key="--PageSize", ) def collect_alert_logs(c: Collector, last_minutes: int = 10080) -> tuple[list[dict[str, Any]], bool]: rows: list[dict[str, Any]] = [] truncated = False for page_num in range(1, 21): data = c.json( "cms", "DescribeAlertLogList", "--LastMin", str(last_minutes), "--PageNumber", str(page_num), "--PageSize", "100", label="Cms.DescribeAlertLogList", ) batch = data.get("AlertLogList") or [] rows.extend(batch) if len(batch) < 100: break else: truncated = True return rows, truncated def collect_cms(c: Collector) -> dict[str, Any]: rules = collect_metric_rules(c) events = collect_page_number(c, ["cms", "DescribeEventRuleList"], ("EventRules", "EventRule"), "Total") for event in events: target = c.json("cms", "DescribeEventRuleTargetList", "--RuleName", event.get("Name", ""), label="Cms.DescribeEventRuleTargetList") event["Targets"] = { key: value for key, value in target.items() if key not in {"Code", "RequestId", "Success"} } groups = collect_page_number(c, ["cms", "DescribeContactGroupList"], ("ContactGroupList", "ContactGroup"), "Total") logs, logs_truncated = collect_alert_logs(c) current_alerts = collect_current_alerts(c, rules) return { "metric_rules": rules, "event_rules": events, "contact_groups": groups, "alert_logs_7d": logs, "alert_logs_truncated": logs_truncated, "current_alerts": current_alerts, } def collect_current_alerts(c: Collector, rules: list[dict[str, Any]]) -> list[dict[str, Any]]: active: list[dict[str, Any]] = [] for rule in rules: if not rule.get("EnableState") or rule.get("AlertState") != "ALARM": continue data = c.json( "cms", "DescribeAlertLogList", "--RuleId", rule.get("RuleId", ""), "--LastMin", "10080", "--PageNumber", "1", "--PageSize", "100", label=f"Cms.CurrentAlert:{rule.get('RuleId','')}", ) newest: dict[tuple[tuple[str, str], ...], dict[str, Any]] = {} for log in sorted(data.get("AlertLogList") or [], key=lambda x: x.get("AlertTime", 0), reverse=True): dimensions = tuple(sorted((str(x.get("Key", "")), str(x.get("Value", ""))) for x in log.get("Dimensions") or [])) newest.setdefault(dimensions, log) for dimensions, log in newest.items(): if log.get("EventName") != "AlertAlarm": continue dim_dict = dict(dimensions) instance_id = dim_dict.get("instanceId") or log.get("InstanceId") or "" metric = c.json( "cms", "DescribeMetricLast", "--Namespace", rule.get("Namespace", ""), "--MetricName", rule.get("MetricName", ""), "--Dimensions", json.dumps({"instanceId": instance_id}), "--Period", str(rule.get("Period") or 60), "--Length", "100", label=f"Cms.DescribeMetricLast:{rule.get('RuleId','')}", ) if instance_id else {} try: points = json.loads(metric.get("Datapoints") or "[]") except (TypeError, json.JSONDecodeError): points = [] comparable_dims = {k: v for k, v in dim_dict.items() if k != "userId"} matching = [p for p in points if all(str(p.get(k, "")) == str(v) for k, v in comparable_dims.items())] latest = max(matching or points or [{}], key=lambda x: x.get("timestamp", 0)) escalation = (rule.get("Escalations") or {}).get("Warn") or (rule.get("Escalations") or {}).get("Critical") or {} statistics = escalation.get("Statistics") or "Average" active.append({ "RuleId": rule.get("RuleId", ""), "RuleName": rule.get("RuleName", ""), "Namespace": rule.get("Namespace", ""), "MetricName": rule.get("MetricName", ""), "InstanceId": instance_id, "Dimensions": dim_dict, "AlertTime": log.get("AlertTime"), "Level": log.get("Level", ""), "ContactGroups": rule.get("ContactGroups", ""), "Threshold": threshold_text(rule), "Statistics": statistics, "CurrentValue": latest.get(statistics, latest.get("Value", "")), "MetricTime": latest.get("timestamp", ""), }) return active def collect_product_details(c: Collector, resources: list[dict[str, Any]]) -> dict[str, Any]: regions_by_type: dict[str, set[str]] = collections.defaultdict(set) all_regions: set[str] = set() for row in resources: region = row.get("RegionId") or "" if region: regions_by_type[row.get("ResourceType", "")].add(region) all_regions.add(region) ecs: list[dict[str, Any]] = [] for region in sorted(regions_by_type["ACS::ECS::Instance"]): ecs.extend(collect_page_number(c, ["ecs", "DescribeInstances", "--RegionId", region], ("Instances", "Instance"), "TotalCount")) rds: list[dict[str, Any]] = [] for region in sorted(regions_by_type["ACS::RDS::DBInstance"]): rds.extend(collect_page_number(c, ["rds", "DescribeDBInstances", "--RegionId", region], ("Items", "DBInstance"), "TotalRecordCount")) slb: list[dict[str, Any]] = [] for region in sorted(regions_by_type["ACS::SLB::LoadBalancer"]): slb.extend(collect_page_number(c, ["slb", "DescribeLoadBalancers", "--RegionId", region], ("LoadBalancers", "LoadBalancer"), "TotalCount")) redis: list[dict[str, Any]] = [] for region in sorted(regions_by_type["ACS::Redis::DBInstance"]): redis.extend(collect_page_number(c, ["r-kvstore", "DescribeInstances", "--RegionId", region], ("Instances", "KVStoreInstance"), "TotalCount", size=50)) mqtt: list[dict[str, Any]] = [] mqtt_regions = {"cn-beijing", "cn-hangzhou", "cn-shanghai", "cn-shenzhen", "ap-southeast-1"} for region in sorted(mqtt_regions): data = c.json("onsmqtt", "ListInstances", "--region", region, label=f"OnsMqtt.ListInstances:{region}") mqtt.extend(data.get("Instances") or []) mqtt = list({x.get("InstanceId"): x for x in mqtt if x.get("InstanceId")}.values()) oss_rows: list[dict[str, Any]] = [] pattern = re.compile(r"^(\S+\s+\S+\s+\S+\s+\S+)\s+(\S+)\s+(\S+)\s+oss://(.+)$") for line in c.text("oss", "ls", label="OSS.ListBuckets").splitlines(): match = pattern.match(line.strip()) if match: oss_rows.append({"CreationTime": match.group(1), "Region": match.group(2), "StorageClass": match.group(3), "BucketName": match.group(4)}) arms_rules: list[dict[str, Any]] = [] arms_regions = {x.get("RegionId") for x in resources if x.get("ResourceType") in {"ACS::ARMS::Prometheus", "ACS::Cms::PrometheusInstance", "ACS::ARMS::RetcodeApp"}} arms_regions.discard(None) arms_regions.add("cn-beijing") for region in sorted(arms_regions): data = c.json("arms", "SearchAlertRules", "--RegionId", region, "--CurrentPage", "1", "--PageSize", "100", label=f"ARMS.SearchAlertRules:{region}") for row in (data.get("PageBean") or {}).get("AlertRules") or []: row = dict(row) row["RegionId"] = region arms_rules.append(row) prometheus_rules: list[dict[str, Any]] = [] prometheus_assets = [x for x in resources if x.get("ResourceType") in {"ACS::ARMS::Prometheus", "ACS::Cms::PrometheusInstance"}] for asset in prometheus_assets: region = asset.get("RegionId") or "" cluster_id = asset.get("ResourceId") or "" data = c.json("arms", "ListPrometheusAlertRules", "--RegionId", region, "--ClusterId", cluster_id, label=f"ARMS.ListPrometheusAlertRules:{cluster_id}") for row in data.get("PrometheusAlertRules") or []: row = dict(row) row["RegionId"] = region row["ClusterId"] = cluster_id prometheus_rules.append(row) return { "ecs": ecs, "rds": rds, "slb": slb, "redis": redis, "mqtt": mqtt, "oss": oss_rows, "arms_rules": arms_rules, "prometheus_rules": prometheus_rules, } def detail_index(details: dict[str, Any]) -> dict[str, dict[str, Any]]: index: dict[str, dict[str, Any]] = {} for row in details["ecs"]: index[row.get("InstanceId", "")] = { "status": row.get("Status", ""), "spec": row.get("InstanceType", ""), "billing": row.get("InstanceChargeType", ""), "expires": row.get("ExpiredTime", ""), "public_ip": ",".join((row.get("PublicIpAddress") or {}).get("IpAddress") or []), "private_ip": ",".join((row.get("VpcAttributes") or {}).get("PrivateIpAddress", {}).get("IpAddress") or []), "detail": row, } for row in details["rds"]: index[row.get("DBInstanceId", "")] = { "status": row.get("DBInstanceStatus", ""), "spec": row.get("DBInstanceClass", ""), "billing": row.get("PayType", ""), "expires": row.get("ExpireTime", ""), "public_ip": "", "private_ip": row.get("ConnectionString", ""), "detail": row, } for row in details["slb"]: index[row.get("LoadBalancerId", "")] = { "status": row.get("LoadBalancerStatus", ""), "spec": row.get("LoadBalancerSpec", ""), "billing": row.get("PayType", ""), "expires": "", "public_ip": row.get("Address", ""), "private_ip": "", "detail": row, } for row in details["redis"]: index[row.get("InstanceId", "")] = { "status": row.get("InstanceStatus", ""), "spec": row.get("InstanceClass", ""), "billing": row.get("ChargeType", ""), "expires": row.get("EndTime", ""), "public_ip": "", "private_ip": row.get("PrivateIp", ""), "detail": row, } for row in details["oss"]: index[row.get("BucketName", "")] = { "status": "Available", "spec": row.get("StorageClass", ""), "billing": "", "expires": "", "public_ip": "", "private_ip": "", "detail": row, } return index def all_assets(resources: list[dict[str, Any]], details: dict[str, Any]) -> list[dict[str, Any]]: index = detail_index(details) assets: list[dict[str, Any]] = [] seen: set[tuple[str, str, str]] = set() for row in resources: resource_id = row.get("ResourceId") or "" key = (row.get("ResourceType") or "", row.get("RegionId") or "", resource_id) if key in seen: continue seen.add(key) d = index.get(resource_id, {}) assets.append({ "resource_type": row.get("ResourceType") or "", "product": (row.get("ResourceType") or "").split("::")[1] if "::" in (row.get("ResourceType") or "") else "", "region": row.get("RegionId") or "global", "resource_id": resource_id, "resource_name": row.get("ResourceName") or "", "created": row.get("CreateTime") or "", "resource_group_id": row.get("ResourceGroupId") or "", "status": d.get("status", ""), "spec": d.get("spec", ""), "billing": d.get("billing", ""), "expires": d.get("expires", ""), "public_ip": d.get("public_ip", ""), "private_ip": d.get("private_ip", ""), "tags": flat_json(row.get("Tags") or []), "source": "Resource Center", }) for row in details["mqtt"]: key = ("ACS::OnsMqtt::Instance", row.get("RegionId") or "", row.get("InstanceId") or "") if key in seen: continue seen.add(key) assets.append({ "resource_type": "ACS::OnsMqtt::Instance", "product": "OnsMqtt", "region": row.get("RegionId") or "", "resource_id": row.get("InstanceId") or "", "resource_name": row.get("InstanceName") or "", "created": epoch_ms(row.get("CreateTime")), "resource_group_id": "", "status": str(row.get("InstanceStatus", "")), "spec": row.get("Specific") or str(row.get("InstanceType", "")), "billing": "", "expires": epoch_ms(row.get("ExpireTime")), "public_ip": "", "private_ip": "", "tags": flat_json(row.get("MqttTags") or []), "source": "OnsMqtt API", }) return sorted(assets, key=lambda x: (x["product"], x["resource_type"], x["region"], x["resource_name"], x["resource_id"])) def threshold_text(rule: dict[str, Any]) -> str: parts: list[str] = [] for level in ("Critical", "Warn", "Info"): item = (rule.get("Escalations") or {}).get(level) or {} if not item: continue op = item.get("ComparisonOperator", "") op_cn = { "GreaterThanThreshold": ">", "GreaterThanOrEqualToThreshold": ">=", "LessThanThreshold": "<", "LessThanOrEqualToThreshold": "<=", "EqualToThreshold": "=", "NotEqualToThreshold": "!=", }.get(op, op) parts.append(f"{level}:{item.get('Statistics','Value')} {op_cn} {item.get('Threshold','')} ×{item.get('Times','')}") return "; ".join(parts) def coverage_rows(assets: list[dict[str, Any]], rules: list[dict[str, Any]]) -> list[dict[str, Any]]: by_type: dict[str, list[dict[str, Any]]] = collections.defaultdict(list) for asset in assets: by_type[asset["resource_type"]].append(asset) rows: list[dict[str, Any]] = [] for namespace, types in NAMESPACE_ASSET_TYPES.items(): related_rules = [r for r in rules if r.get("Namespace") == namespace] enabled = [r for r in related_rules if r.get("EnableState")] all_scope = [r for r in enabled if "_ALL" in values_from_json_text(r.get("Resources", ""))] for resource_type in types: asset_list = by_type.get(resource_type, []) for asset in asset_list: matched = [r for r in enabled if asset["resource_id"] in values_from_json_text(r.get("Resources", "")) | values_from_json_text(r.get("Dimensions", ""))] if matched: level = "直接绑定" elif all_scope: level = "产品全量规则" elif enabled: level = "仅存在同产品规则，未直接匹配" else: level = "未发现启用的经典云监控规则" rows.append({ "resource_type": resource_type, "resource_id": asset["resource_id"], "resource_name": asset["resource_name"], "region": asset["region"], "coverage": level, "matched_rules": ", ".join(r.get("RuleName", "") for r in matched), "namespace": namespace, "enabled_rule_count": len(enabled), "all_scope_rule_count": len(all_scope), }) return rows def operational_asset(asset: dict[str, Any]) -> bool: return asset["resource_type"] in { "ACS::ECS::Instance", "ACS::RDS::DBInstance", "ACS::OSS::Bucket", "ACS::Redis::DBInstance", "ACS::SLB::LoadBalancer", "ACS::OnsMqtt::Instance", "ACS::SWAS::Instance", "ACS::NAS::FileSystem", "ACS::ARMS::Prometheus", "ACS::Cms::PrometheusInstance", "ACS::FC::Service", "ACS::FCV3::Function", "ACS::SLS::Project", "ACS::OTS::Instance", "ACS::CDN::Domain", "ACS::DCDN::Domain", } def build_summary(assets: list[dict[str, Any]], cms: dict[str, Any], details: dict[str, Any], errors: list[dict[str, str]]) -> dict[str, Any]: rules = cms["metric_rules"] operational = [x for x in assets if operational_asset(x)] stopped_ecs = [x for x in assets if x["resource_type"] == "ACS::ECS::Instance" and x["status"] != "Running"] now_utc = NOW.astimezone(dt.timezone.utc) expired: list[dict[str, Any]] = [] for asset in operational: raw = asset.get("expires") or "" if raw: try: expiry = dt.datetime.fromisoformat(raw.replace("Z", "+00:00")) if expiry.tzinfo is None: expiry = expiry.replace(tzinfo=dt.timezone.utc) if expiry < now_utc: expired.append(asset) except ValueError: pass ns_assets: dict[str, int] = {} for ns, types in NAMESPACE_ASSET_TYPES.items(): ns_assets[ns] = sum(1 for x in assets if x["resource_type"] in types) rule_ns = collections.Counter(r.get("Namespace", "") for r in rules) suspect_stale_namespaces = sorted(ns for ns, count in rule_ns.items() if count and ns_assets.get(ns, 0) == 0 and ns != "acs_networkmonitor") return { "asset_total": len(assets), "resource_types": len({x["resource_type"] for x in assets}), "regions": len({x["region"] for x in assets}), "operational_assets": len(operational), "metric_rule_total": len(rules), "metric_rule_enabled": sum(bool(r.get("EnableState")) for r in rules), "metric_rule_disabled": sum(not bool(r.get("EnableState")) for r in rules), "state_ok": sum(r.get("AlertState") == "OK" for r in rules), "state_alarm": sum(r.get("AlertState") == "ALARM" for r in rules), "active_alarm": sum(r.get("AlertState") == "ALARM" and bool(r.get("EnableState")) for r in rules), "insufficient_data": sum(r.get("AlertState") == "INSUFFICIENT_DATA" for r in rules), "enabled_insufficient_data": sum(r.get("AlertState") == "INSUFFICIENT_DATA" and bool(r.get("EnableState")) for r in rules), "event_rule_total": len(cms["event_rules"]), "event_rule_enabled": sum(r.get("State") == "ENABLED" for r in cms["event_rules"]), "contact_groups": len(cms["contact_groups"]), "alert_logs_7d": len(cms["alert_logs_7d"]), "arms_rules": len(details["arms_rules"]), "prometheus_rules": len(details["prometheus_rules"]), "stopped_ecs": len(stopped_ecs), "expired_operational_assets": len(expired), "suspect_stale_namespaces": suspect_stale_namespaces, "collector_errors": len(errors), "collected_at": NOW.strftime("%Y-%m-%d %H:%M:%S %z"), } def prepare_rows(assets: list[dict[str, Any]], cms: dict[str, Any], details: dict[str, Any]) -> dict[str, list[dict[str, Any]]]: asset_names = {x["resource_id"]: x["resource_name"] for x in assets} type_counts = collections.Counter(x["resource_type"] for x in assets) region_counts = collections.Counter(x["region"] for x in assets) product_counts = collections.Counter(x["product"] for x in assets) namespace_counts = collections.Counter(r.get("Namespace", "") for r in cms["metric_rules"]) rules = [] for r in cms["metric_rules"]: rules.append({ "rule_id": r.get("RuleId", ""), "rule_name": r.get("RuleName", ""), "namespace": r.get("Namespace", ""), "metric_name": r.get("MetricName", ""), "enabled": r.get("EnableState", False), "alert_state": r.get("AlertState", ""), "threshold": threshold_text(r), "effective_interval": r.get("EffectiveInterval", ""), "period": r.get("Period", ""), "interval": r.get("Interval", ""), "silence_time": r.get("SilenceTime", ""), "no_data_policy": r.get("NoDataPolicy", ""), "contact_groups": r.get("ContactGroups", ""), "resources": r.get("Resources", ""), "dimensions": r.get("Dimensions", ""), "send_ok": r.get("SendOK", ""), "webhook": r.get("Webhook", ""), "updated_at": epoch_ms(r.get("GmtUpdate")), }) logs = [] for row in cms["alert_logs_7d"]: message = row.get("Message") or "" rule_name = "" try: rule_name = json.loads(message).get("ruleName", "") except (json.JSONDecodeError, TypeError): pass logs.append({ "alert_time": epoch_ms(row.get("AlertTime")), "rule_name": rule_name, "rule_id": row.get("RuleId", ""), "namespace": row.get("Namespace", "") or row.get("MetricProject", ""), "metric_name": row.get("MetricName", ""), "level": row.get("Level", ""), "event_name": row.get("EventName", ""), "send_status": row.get("SendStatus", ""), "contact_groups": ",".join(row.get("ContactGroups") or []), "instance_id": row.get("InstanceId", ""), "message_summary": (row.get("Message") or "")[:500], }) events = [] for row in cms["event_rules"]: events.append({ "name": row.get("Name", ""), "state": row.get("State", ""), "event_type": row.get("EventType", ""), "group_id": row.get("GroupId", ""), "silence_time": row.get("SilenceTime", ""), "event_pattern": flat_json(row.get("EventPattern") or {}), "targets": flat_json(row.get("Targets") or {}), "description": row.get("Description", ""), }) groups = [] for row in cms["contact_groups"]: groups.append({ "name": row.get("Name", ""), "contacts": ", ".join((row.get("Contacts") or {}).get("Contact") or []), "description": row.get("Describe", ""), "weekly_report": row.get("EnabledWeeklyReport", ""), "updated_at": epoch_ms(row.get("UpdateTime")), }) current_alerts = [] for row in cms.get("current_alerts") or []: current_alerts.append({ "rule_name": row.get("RuleName", ""), "rule_id": row.get("RuleId", ""), "namespace": row.get("Namespace", ""), "metric_name": row.get("MetricName", ""), "instance_id": row.get("InstanceId", ""), "instance_name": asset_names.get(row.get("InstanceId", ""), ""), "dimensions": flat_json(row.get("Dimensions") or {}), "level": row.get("Level", ""), "current_value": row.get("CurrentValue", ""), "statistics": row.get("Statistics", ""), "threshold": row.get("Threshold", ""), "alert_time": epoch_ms(row.get("AlertTime")), "metric_time": epoch_ms(row.get("MetricTime")), "contact_groups": row.get("ContactGroups", ""), }) return { "assets": assets, "operational": [x for x in assets if operational_asset(x)], "type_summary": [{"resource_type": k, "product_cn": PRODUCT_CN.get(k.split("::")[1] if "::" in k else "", ""), "count": v} for k, v in type_counts.most_common()], "product_summary": [{"product": k, "product_cn": PRODUCT_CN.get(k, k), "count": v} for k, v in product_counts.most_common()], "region_summary": [{"region": k, "count": v} for k, v in region_counts.most_common()], "namespace_summary": [{"namespace": k, "rule_count": v} for k, v in namespace_counts.most_common()], "rules": rules, "current_alerts": current_alerts, "logs": logs, "events": events, "groups": groups, "coverage": coverage_rows(assets, cms["metric_rules"]), "arms_rules": details["arms_rules"], "prometheus_rules": details["prometheus_rules"], } HEADER_FILL = PatternFill("solid", fgColor="0F766E") HEADER_FONT = Font(name="微软雅黑", size=12, bold=True, color="FFFFFF") BODY_FONT = Font(name="微软雅黑", size=12, color="182221") THIN = Side(style="thin", color="DCE5E2") def add_table_sheet(wb: Workbook, title: str, rows: list[dict[str, Any]], headers: list[tuple[str, str]], freeze: str = "A2") -> None: ws = wb.create_sheet(title) ws.freeze_panes = freeze ws.sheet_view.showGridLines = False ws.append([label for _, label in headers]) for row in rows: ws.append([row.get(key, "") for key, _ in headers]) for cell in ws[1]: cell.fill = HEADER_FILL cell.font = HEADER_FONT cell.alignment = Alignment(horizontal="center", vertical="center") for row in ws.iter_rows(min_row=2): for cell in row: cell.font = BODY_FONT cell.alignment = Alignment(vertical="top", wrap_text=True) cell.border = Border(bottom=THIN) ws.auto_filter.ref = ws.dimensions for idx, (_, label) in enumerate(headers, 1): sample = [str(ws.cell(r, idx).value or "") for r in range(1, min(ws.max_row, 100) + 1)] width = min(max(len(label) * 2, *(len(x) for x in sample)) + 2, 48) ws.column_dimensions[get_column_letter(idx)].width = max(width, 12) ws.row_dimensions[1].height = 26 def build_workbook(path: Path, summary: dict[str, Any], rows: dict[str, list[dict[str, Any]]], errors: list[dict[str, str]]) -> None: wb = Workbook() wb.remove(wb.active) ws = wb.create_sheet("管理摘要") ws.sheet_view.showGridLines = False ws["A1"] = "阿里云资产与报警统一管理台账" ws["A1"].font = Font(name="微软雅黑", size=22, bold=True, color="FFFFFF") ws["A1"].fill = PatternFill("solid", fgColor="123D3A") ws.merge_cells("A1:D1") summary_rows = [ ("盘点时间", summary["collected_at"]), ("资源总数", summary["asset_total"]), ("资源类型数", summary["resource_types"]), ("地域数", summary["regions"]), ("重点运行资产数", summary["operational_assets"]), ("经典云监控指标规则", summary["metric_rule_total"]), ("启用规则", summary["metric_rule_enabled"]), ("停用规则", summary["metric_rule_disabled"]), ("启用且正在报警", summary["active_alarm"]), ("无数据规则", summary["insufficient_data"]), ("启用但无数据", summary["enabled_insufficient_data"]), ("事件规则", summary["event_rule_total"]), ("联系人组", summary["contact_groups"]), ("近7日告警日志", summary["alert_logs_7d"]), ("ARMS应用规则", summary["arms_rules"]), ("Prometheus规则", summary["prometheus_rules"]), ("非运行中ECS", summary["stopped_ecs"]), ("已过期重点资产", summary["expired_operational_assets"]), ("采集API异常数", summary["collector_errors"]), ] for i, (label, value) in enumerate(summary_rows, 3): ws.cell(i, 1, label) ws.cell(i, 2, value) ws.cell(i, 1).font = Font(name="微软雅黑", size=12, bold=True, color="365653") ws.cell(i, 2).font = BODY_FONT ws.cell(i, 1).fill = PatternFill("solid", fgColor="F4F8F7") ws.cell(i, 1).border = ws.cell(i, 2).border = Border(bottom=THIN) ws["D3"] = "重点治理结论" ws["D3"].font = Font(name="微软雅黑", size=14, bold=True, color="B42318") conclusions = [ f"1. 当前启用且正在报警的经典云监控规则：{summary['active_alarm']} 条。", f"2. 启用但无数据：{summary['enabled_insufficient_data']} 条，应优先核实资源是否已下线、Agent/指标是否失效。", f"3. 停用规则：{summary['metric_rule_disabled']} 条；停用不等于已清理，需保留/归档/删除决策。", f"4. 疑似历史产品规则命名空间：{', '.join(summary['suspect_stale_namespaces']) or '无'}。资源中心未发现对应现存资产，需产品侧二次确认。", "5. 事件规则中存在启用的测试临时规则，应由 owner 确认是否归档；本次只读盘点未做变更。", "6. 完整资源 ID、IP、规则资源维度仅保存在本机私有台账，不公开上传。", ] for idx, line in enumerate(conclusions, 4): ws.cell(idx, 4, line) ws.cell(idx, 4).font = BODY_FONT ws.cell(idx, 4).alignment = Alignment(wrap_text=True, vertical="top") ws.column_dimensions["A"].width = 24 ws.column_dimensions["B"].width = 28 ws.column_dimensions["C"].width = 4 ws.column_dimensions["D"].width = 88 asset_headers = [ ("product", "产品"), ("resource_type", "资源类型"), ("region", "地域"), ("resource_id", "资源ID"), ("resource_name", "资源名称"), ("status", "状态"), ("spec", "规格/存储类型"), ("billing", "计费方式"), ("expires", "到期时间"), ("public_ip", "公网地址"), ("private_ip", "私网地址/连接地址"), ("created", "创建时间"), ("resource_group_id", "资源组ID"), ("tags", "标签"), ("source", "数据来源"), ] add_table_sheet(wb, "重点运行资产", rows["operational"], asset_headers) add_table_sheet(wb, "全部资产明细", rows["assets"], asset_headers) add_table_sheet(wb, "产品汇总", rows["product_summary"], [("product", "产品代码"), ("product_cn", "产品名称"), ("count", "数量")]) add_table_sheet(wb, "资源类型汇总", rows["type_summary"], [("resource_type", "资源类型"), ("product_cn", "产品名称"), ("count", "数量")]) add_table_sheet(wb, "地域汇总", rows["region_summary"], [("region", "地域"), ("count", "数量")]) rule_headers = [ ("rule_id", "规则ID"), ("rule_name", "规则名称"), ("namespace", "命名空间"), ("metric_name", "监控指标"), ("enabled", "是否启用"), ("alert_state", "当前状态"), ("threshold", "阈值与连续次数"), ("effective_interval", "生效时段"), ("period", "统计周期"), ("interval", "检测周期"), ("silence_time", "沉默期"), ("no_data_policy", "无数据策略"), ("contact_groups", "联系人组"), ("resources", "资源范围"), ("dimensions", "维度"), ("send_ok", "恢复通知"), ("webhook", "Webhook"), ("updated_at", "更新时间"), ] add_table_sheet(wb, "指标报警规则", rows["rules"], rule_headers) add_table_sheet(wb, "当前有效报警", rows["current_alerts"], [ ("rule_name", "规则名称"), ("rule_id", "规则ID"), ("namespace", "命名空间"), ("metric_name", "指标"), ("instance_id", "实例ID"), ("instance_name", "实例名称"), ("dimensions", "报警维度"), ("level", "级别"), ("current_value", "当前值"), ("statistics", "统计方式"), ("threshold", "阈值"), ("alert_time", "最近触发时间"), ("metric_time", "最新指标时间"), ("contact_groups", "联系人组"), ]) add_table_sheet(wb, "规则产品汇总", rows["namespace_summary"], [("namespace", "命名空间"), ("rule_count", "规则数")]) add_table_sheet(wb, "近7日告警日志", rows["logs"], [ ("alert_time", "告警时间"), ("rule_name", "规则名称"), ("rule_id", "规则ID"), ("namespace", "命名空间"), ("metric_name", "指标"), ("level", "级别"), ("event_name", "事件"), ("send_status", "发送状态"), ("contact_groups", "联系人组"), ("instance_id", "实例ID"), ("message_summary", "消息摘要"), ]) add_table_sheet(wb, "事件报警规则", rows["events"], [ ("name", "规则名称"), ("state", "状态"), ("event_type", "事件类型"), ("group_id", "应用分组ID"), ("silence_time", "沉默期"), ("event_pattern", "事件匹配"), ("targets", "通知目标"), ("description", "描述"), ]) add_table_sheet(wb, "联系人组", rows["groups"], [ ("name", "联系人组"), ("contacts", "联系人名称/脱敏邮箱"), ("description", "描述"), ("weekly_report", "周报"), ("updated_at", "更新时间"), ]) add_table_sheet(wb, "重点资产监控覆盖", rows["coverage"], [ ("resource_type", "资源类型"), ("resource_id", "资源ID"), ("resource_name", "资源名称"), ("region", "地域"), ("coverage", "经典云监控覆盖判定"), ("matched_rules", "直接匹配规则"), ("namespace", "命名空间"), ("enabled_rule_count", "同产品启用规则数"), ("all_scope_rule_count", "产品全量规则数"), ]) add_table_sheet(wb, "ARMS与Prometheus规则", [ {"source": "ARMS", "payload": flat_json(x)} for x in rows["arms_rules"] ] + [ {"source": "Prometheus", "payload": flat_json(x)} for x in rows["prometheus_rules"] ], [("source", "规则来源"), ("payload", "规则详情")]) add_table_sheet(wb, "采集异常", errors, [("label", "采集项"), ("error", "异常摘要")]) path.parent.mkdir(parents=True, exist_ok=True) wb.save(path) load_workbook(path, read_only=True).close() def build_html(path: Path, summary: dict[str, Any], rows: dict[str, list[dict[str, Any]]]) -> None: type_rows = "".join( f"{html.escape(r['product_cn'] or r['resource_type'])}{html.escape(r['resource_type'])}{r['count']}" for r in rows["type_summary"][:20] ) ns_rows = "".join( f"{html.escape(r['namespace'])}{r['rule_count']}" for r in rows["namespace_summary"] ) active = rows["current_alerts"] active_cards = "".join( f"
{html.escape(r['rule_name'])}

{html.escape(r['instance_name'] or r['instance_id'])}

{html.escape(r['dimensions'])}

当前值
{html.escape(str(r['current_value']))}
阈值
{html.escape(r['threshold'])}
联系人组
{html.escape(r['contact_groups'])}
" for r in active ) or "

无启用且正在报警的经典云监控规则

" stale = ", ".join(summary["suspect_stale_namespaces"]) or "无" doc = f""" 阿里云资产与报警统一盘点

ALIYUN ASSET & ALERT INVENTORY · {NOW.strftime('%Y-%m-%d')}

阿里云资产与报警统一盘点

当前账号的资源中心、核心产品 API 与云监控规则已做只读汇总。完整资源 ID、IP 与规则维度保存在同目录私有 Excel，不公开上传。

Executive Summary

{summary['asset_total']}资源总数
{summary['metric_rule_total']}指标报警规则
{summary['active_alarm']}启用且正在报警
{summary['enabled_insufficient_data']}启用但无数据

现在需要先处理什么

P0：处理当前报警。 当前启用且处于 ALARM 的规则为 {summary['active_alarm']} 条，详见下方报警卡片与 Excel 的“当前有效报警”Sheet。
{active_cards}

P1：清理无数据与历史规则。 启用但无数据 {summary['enabled_insufficient_data']} 条；资源中心未发现对应现存资产、疑似历史遗留的命名空间包括：{html.escape(stale)}。这只是“待复核”，不能直接删除。

P1：处理临时规则。 事件报警中存在启用的“测试临时带宽升级触发通知”；本次按只读边界未修改，建议 owner 确认后归档或删除。

资源类型 Top 20

{type_rows}
产品资源类型数量

报警规则按产品命名空间

{ns_rows}
命名空间规则数

统一管理标准

  1. 每项重点资产必须登记业务 owner、技术 owner、环境、重要级、到期时间和恢复目标。
  2. 每条规则必须能关联现存资源或明确标注产品全量范围；连续 7 天无数据的规则进入复核队列。
  3. P0/P1 规则使用统一联系人组，Critical 立即通知，恢复通知必须开启；沉默期不能代替 5/15 分钟升级。
  4. ECS、RDS、OSS、Redis、SLB、MQTT 至少覆盖可用性、容量、性能、到期/续费和安全事件；业务链路再叠加项目级规则。
  5. 每季度做一次短信/电话/机器人演练，保存实收证据；测试规则必须带到期时间并在演练后清理。

口径与边界

""" path.parent.mkdir(parents=True, exist_ok=True) path.write_text(doc, encoding="utf-8") def validate(assets: list[dict[str, Any]], cms: dict[str, Any], workbook: Path, report: Path) -> dict[str, Any]: asset_keys = [(x["resource_type"], x["region"], x["resource_id"]) for x in assets] rule_ids = [x.get("RuleId") for x in cms["metric_rules"]] checks = { "asset_unique": len(asset_keys) == len(set(asset_keys)), "rule_unique": len(rule_ids) == len(set(rule_ids)), "workbook_exists": workbook.exists() and workbook.stat().st_size > 0, "html_exists": report.exists() and report.stat().st_size > 0, "test_metric_rules_remaining": sum("TEST-" in (x.get("RuleName") or "").upper() for x in cms["metric_rules"]), "mqtt_formal_rules": sum(x.get("Namespace") == "acs_mqtt" for x in cms["metric_rules"]), "asset_count": len(assets), "metric_rule_count": len(cms["metric_rules"]), } checks["passed"] = all(v is True or isinstance(v, int) for k, v in checks.items() if k not in {"test_metric_rules_remaining"}) and checks["asset_unique"] and checks["rule_unique"] and checks["test_metric_rules_remaining"] == 0 return checks def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--output-root", type=Path, required=True) args = parser.parse_args() output_root = args.output_root.expanduser().resolve() final_dir = output_root / "01-最终交付" raw_dir = output_root / "03-运行输出" final_dir.mkdir(parents=True, exist_ok=True) raw_dir.mkdir(parents=True, exist_ok=True) c = Collector() identity = c.json("sts", "GetCallerIdentity", label="STS.GetCallerIdentity") resources = collect_resource_center(c) cms = collect_cms(c) details = collect_product_details(c, resources) assets = all_assets(resources, details) summary = build_summary(assets, cms, details, c.errors) rows = prepare_rows(assets, cms, details) raw = { "metadata": {"collected_at": summary["collected_at"], "account_id": identity.get("AccountId", ""), "identity_type": identity.get("IdentityType", "")}, "summary": summary, "assets": assets, "resource_center": resources, "cms": cms, "product_details": details, "collector_errors": c.errors, } raw_path = raw_dir / "aliyun-asset-alert-inventory.json" raw_path.write_text(json.dumps(raw, ensure_ascii=False, indent=2), encoding="utf-8") workbook = final_dir / f"阿里云资产与报警统一管理台账_{NOW.strftime('%Y%m%d')}.xlsx" report = final_dir / "aliyun-asset-alert-inventory.html" build_workbook(workbook, summary, rows, c.errors) build_html(report, summary, rows) checks = validate(assets, cms, workbook, report) validation_path = raw_dir / "validation.json" validation_path.write_text(json.dumps(checks, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps({"summary": summary, "workbook": str(workbook), "report": str(report), "raw": str(raw_path), "validation": checks, "collector_errors": c.errors}, ensure_ascii=False, indent=2)) return 0 if checks["passed"] else 2 if __name__ == "__main__": sys.exit(main())