#!/usr/bin/env python3 """Build a sanitized WeCom smart-sheet payload from read-only Aliyun inventory. The script never exports credentials, environment-variable values, RAM identities, or raw cloud configuration blobs. It emits only operational asset metadata. """ from __future__ import annotations import argparse import json import re from datetime import datetime from pathlib import Path from typing import Any from zoneinfo import ZoneInfo TZ = ZoneInfo("Asia/Shanghai") SINGLE_SELECT_FIELDS = {"资源类型", "运行状态", "付费方式", "风险等级"} def read_json(path: Path, default: Any = None) -> Any: if not path.exists() or path.stat().st_size == 0: return default return json.loads(path.read_text(encoding="utf-8")) def local_time(value: Any) -> str | None: if value in (None, "", 0): return None try: if isinstance(value, (int, float)): dt = datetime.fromtimestamp(value / 1000, tz=ZoneInfo("UTC")) else: text = str(value).replace("CST", "+08:00") if text.endswith("Z"): text = text[:-1] + "+00:00" dt = datetime.fromisoformat(text) if dt.tzinfo is None: dt = dt.replace(tzinfo=TZ) return dt.astimezone(TZ).strftime("%Y-%m-%d %H:%M:%S") except (ValueError, TypeError, OSError): return None def days_left(expiry: str | None, snapshot_dt: datetime) -> int | None: if not expiry: return None try: return (datetime.strptime(expiry, "%Y-%m-%d %H:%M:%S").replace(tzinfo=TZ).date() - snapshot_dt.date()).days except ValueError: return None def first(*values: Any, default: Any = "") -> Any: for value in values: if value not in (None, "", [], {}): return value return default def metric_map(items: list[dict[str, Any]]) -> dict[tuple[str, str], dict[str, Any]]: return {(item.get("instance_id", ""), item.get("metric", "")): item.get("points", {}) for item in items} def public_rules(security_groups: list[dict[str, Any]]) -> dict[str, list[str]]: result: dict[str, list[str]] = {} for group in security_groups: rules: list[str] = [] for rule in group.get("rules", []): if rule.get("policy") != "Accept" or rule.get("source") not in {"0.0.0.0/0", "::/0"}: continue protocol = str(rule.get("protocol", "")).upper() port = str(rule.get("port", "")) rules.append(f"{protocol}:{port}") result[group.get("security_group", "")] = sorted(set(rules)) return result def risk_level(reasons: list[str], warnings: list[str]) -> str: if reasons: return "高" if warnings: return "中" return "低" def put(rows: list[dict[str, Any]], values: dict[str, Any]) -> None: clean = {key: value for key, value in values.items() if value not in (None, "", [], {})} rows.append({"values": clean}) def build_rows(base: Path, resource_center_path: Path, snapshot_dt: datetime) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] snapshot_text = snapshot_dt.strftime("%Y-%m-%d %H:%M:%S") metrics = metric_map(read_json(base / "ecs-metrics.json", [])) security = public_rules(read_json(base / "ecs-security.json", [])) disks: list[dict[str, Any]] = [] for region in ("cn-beijing", "cn-qingdao"): disks.extend(read_json(base / f"disks-{region}.json", {}).get("Disks", {}).get("Disk", [])) disks_by_instance: dict[str, list[dict[str, Any]]] = {} for disk in disks: disks_by_instance.setdefault(disk.get("InstanceId", ""), []).append(disk) # ECS: every server, plus seven-day CloudMonitor and security-group evidence. for region in ("cn-beijing", "cn-qingdao"): instances = read_json(Path(f"/tmp/ecs-{region}.json"), {}).get("Instances", {}).get("Instance", []) for item in instances: instance_id = item.get("InstanceId", "") expiry = local_time(item.get("ExpiredTime")) remaining = days_left(expiry, snapshot_dt) cpu = metrics.get((instance_id, "CPUUtilization"), {}) memory = metrics.get((instance_id, "memory_usedutilization"), {}) disk_metric = metrics.get((instance_id, "diskusage_utilization"), {}) instance_disks = disks_by_instance.get(instance_id, []) disk_parts = [] snapshot_bound = 0 for disk in instance_disks: disk_type = "系统盘" if disk.get("Type") == "system" else "数据盘" category = first(disk.get("Category"), default="disk") level = first(disk.get("PerformanceLevel"), default="") disk_parts.append(f"{disk_type}{disk.get('Size', 0)}GiB {category}{(' ' + level) if level else ''}") if disk.get("AutoSnapshotPolicyId"): snapshot_bound += 1 public_ports: list[str] = [] for group_id in item.get("SecurityGroupIds", {}).get("SecurityGroupId", []): public_ports.extend(security.get(group_id, [])) public_ports = sorted(set(public_ports)) risky_ports = [ rule for rule in public_ports if rule.startswith("ALL:") or re.search(r"(?:^|/)(20|21|23|3389|3306|5432|6379|1433)(?:/|$)", rule) ] public_ssh = any(re.search(r"(?:^|/)22(?:/|$)", rule) for rule in public_ports) reasons: list[str] = [] warnings: list[str] = [] actions: list[str] = [] os_name = str(item.get("OSName", "")) if remaining is not None and remaining <= 30: reasons.append(f"{remaining}天内到期") actions.append("立即确认续费或迁移") elif remaining is not None and remaining <= 60: warnings.append(f"{remaining}天内到期") actions.append("纳入本月续费计划") if item.get("Status") != "Running": reasons.append(f"实例状态{item.get('Status', '未知')}") actions.append("确认停机原因和保留策略") if re.search(r"CentOS\s*[67](?:\.|_|\b)|Windows Server\s+2012", os_name, re.I): reasons.append("操作系统已过或接近支持周期边界") actions.append("制定系统升级或迁移计划") disk_max = disk_metric.get("max") if isinstance(disk_max, (int, float)) and disk_max >= 90: reasons.append(f"近7日磁盘峰值{disk_max:.1f}%") actions.append("核对具体分区并清理或扩容") if risky_ports: reasons.append("数据库/缓存等高危端口对公网放行") actions.append("收敛安全组至白名单或私网") if public_ssh: warnings.append("SSH对公网放行") actions.append("SSH改为固定出口、VPN或Workbench") if instance_disks and snapshot_bound < len(instance_disks): warnings.append(f"自动快照仅覆盖{snapshot_bound}/{len(instance_disks)}块云盘") actions.append("补齐自动快照并做恢复演练") memory_max = memory.get("max") if isinstance(memory_max, (int, float)) and memory_max >= 85: warnings.append(f"近7日内存峰值{memory_max:.1f}%") actions.append("检查常驻进程和容量余量") public_ips = item.get("PublicIpAddress", {}).get("IpAddress", []) if item.get("EipAddress", {}).get("IpAddress"): public_ips.append(item["EipAddress"]["IpAddress"]) private_ips = item.get("VpcAttributes", {}).get("PrivateIpAddress", {}).get("IpAddress", []) tags = [f"{tag.get('TagKey')}={tag.get('TagValue')}" for tag in item.get("Tags", {}).get("Tag", []) if tag.get("TagKey")] purpose = "；".join(tags) if tags else "按实例名称判断用途，待责任人复核" public_port_text = "、".join(public_ports[:20]) if len(public_ports) > 20: public_port_text += f"……共{len(public_ports)}条公网规则" put(rows, { "快照时间": snapshot_text, "资源类型": "ECS", "资源名称": first(item.get("InstanceName"), instance_id), "资源实例ID": instance_id, "地域": region, "可用区": item.get("ZoneId"), "运行状态": "正常" if item.get("Status") == "Running" else "停止", "业务用途/环境": purpose, "规格": f"{item.get('InstanceType', '')}｜{item.get('Cpu', 0)}C/{round(item.get('Memory', 0)/1024)}G", "软件/引擎版本": os_name, "存储/容量": "；".join(disk_parts), "公网地址/访问端点": "；".join(sorted(set(public_ips))), "私网地址": "；".join(private_ips), "付费方式": "包年包月" if item.get("InstanceChargeType") == "PrePaid" else "按量付费", "到期时间": expiry, "剩余天数": remaining, "近7日CPU均值%": cpu.get("avg"), "近7日CPU峰值%": cpu.get("max"), "近7日内存均值%": memory.get("avg"), "近7日内存峰值%": memory.get("max"), "近7日磁盘峰值%": disk_max, "备份/快照状态": f"自动快照覆盖{snapshot_bound}/{len(instance_disks)}块云盘" if instance_disks else "未取得云盘信息", "网络与权限状态": "公网放行：" + (public_port_text if public_ports else "未发现"), "风险等级": risk_level(reasons, warnings), "风险说明": "；".join(reasons + warnings) or "当前只读指标未发现高风险项", "运维建议": "；".join(dict.fromkeys(actions)) or "保持监控，按月复核", "数据来源": "ECS实例/云盘/安全组API + CloudMonitor近7日", }) # RDS: instance metadata plus backup policy, without accounts or database contents. for attr_path in sorted((base / "rds").glob("*-attr.json")): dbid = attr_path.name.removesuffix("-attr.json") item = read_json(attr_path, {}).get("Items", {}).get("DBInstanceAttribute", [{}])[0] backup = read_json(base / "rds" / f"{dbid}-backup.json", {}) expiry = local_time(item.get("ExpireTime")) remaining = days_left(expiry, snapshot_dt) engine = f"{item.get('Engine', '')} {item.get('EngineVersion', '')}".strip() reasons: list[str] = [] warnings: list[str] = [] actions: list[str] = [] if re.search(r"MySQL (5\.5|5\.6)|SQLServer 2012", engine, re.I): reasons.append(f"数据库版本老旧：{engine}") actions.append("评估兼容性并制定数据库升级计划") if remaining is not None and remaining <= 30: reasons.append(f"{remaining}天内到期") actions.append("立即确认续费") elif remaining is not None and remaining <= 60: warnings.append(f"{remaining}天内到期") actions.append("纳入本月续费计划") if str(backup.get("EnableBackupLog", "")) in {"0", "False", "false"}: warnings.append("日志备份未启用") actions.append("确认恢复点目标并评估开启日志备份") put(rows, { "快照时间": snapshot_text, "资源类型": "RDS", "资源名称": first(item.get("DBInstanceDescription"), dbid), "资源实例ID": dbid, "地域": "cn-beijing", "可用区": item.get("ZoneId"), "运行状态": "正常" if item.get("DBInstanceStatus") == "Running" else "异常", "业务用途/环境": first(item.get("DBInstanceDescription"), default="待确认"), "规格": f"{item.get('DBInstanceClass', '')}｜{item.get('DBInstanceCPU', '')}C/{round((item.get('DBInstanceMemory') or 0)/1024, 1)}G｜{item.get('Category', '')}", "软件/引擎版本": engine, "存储/容量": f"{item.get('DBInstanceStorage', '')}GiB {item.get('DBInstanceStorageType', '')}", "私网地址": f"{item.get('ConnectionString', '')}:{item.get('Port', '')}", "付费方式": "包年包月" if item.get("PayType") == "Prepaid" else "按量付费", "到期时间": expiry, "剩余天数": remaining, "备份/快照状态": f"备份周期：{backup.get('PreferredBackupPeriod', '待确认')}；保留{backup.get('BackupRetentionPeriod', '待确认')}天；日志备份{backup.get('EnableBackupLog', '待确认')}", "网络与权限状态": f"{item.get('InstanceNetworkType', '')}；白名单/公网连接需控制台专项复核", "风险等级": risk_level(reasons, warnings), "风险说明": "；".join(reasons + warnings) or "实例运行正常，未读取业务库内容", "运维建议": "；".join(dict.fromkeys(actions)) or "保持备份和可用性监控", "数据来源": "RDS实例属性 + 备份策略API（未读取数据库业务数据）", }) # Redis. redis = read_json(base / "redis-instances.json", {}).get("Instances", {}).get("KVStoreInstance", []) redis_backup = read_json(base / "redis-backup.json", {}) for item in redis: expiry = local_time(item.get("EndTime")) remaining = days_left(expiry, snapshot_dt) warnings = [] actions = [] if remaining is not None and remaining <= 60: warnings.append(f"{remaining}天内到期") actions.append("纳入续费计划") if str(item.get("EngineVersion", "")) == "5.0": warnings.append("Redis 5.0需评估升级窗口") actions.append("核对客户端兼容性后规划升级") put(rows, { "快照时间": snapshot_text, "资源类型": "Redis", "资源名称": first(item.get("InstanceName"), item.get("InstanceId")), "资源实例ID": item.get("InstanceId"), "地域": "cn-beijing", "可用区": item.get("ZoneId"), "运行状态": "正常" if item.get("InstanceStatus") == "Normal" else "异常", "业务用途/环境": first(item.get("InstanceName"), default="待确认"), "规格": f"{item.get('InstanceClass', '')}｜{item.get('Capacity', 0)}MiB", "软件/引擎版本": f"Redis {item.get('EngineVersion', '')}", "存储/容量": f"{item.get('Capacity', 0)}MiB", "私网地址": f"{item.get('ConnectionDomain', '')}:{item.get('Port', '')}", "付费方式": "包年包月" if item.get("ChargeType") == "PrePaid" else "按量付费", "到期时间": expiry, "剩余天数": remaining, "备份/快照状态": f"每日备份；保留{redis_backup.get('BackupRetentionPeriod', '待确认')}天", "网络与权限状态": f"{item.get('NetworkType', '')}；密码/白名单未导出", "风险等级": risk_level([], warnings), "风险说明": "；".join(warnings) or "运行正常", "运维建议": "；".join(actions) or "保持备份和容量监控", "数据来源": "Tair/Redis实例 + 备份策略API", }) # MQTT. for item in read_json(base / "mqtt-instances.json", []): expiry = local_time(item.get("ExpireTime")) remaining = days_left(expiry, snapshot_dt) warnings = [] actions = [] if remaining is not None and remaining <= 60: warnings.append(f"{remaining}天内到期") actions.append("纳入续费计划") mqtt_type = {0: "基础版", 1: "铂金版", 2: "按量付费", 3: "预付费轻量", 4: "专业版", 5: "Serverless"}.get(item.get("InstanceType"), f"类型码{item.get('InstanceType')}") put(rows, { "快照时间": snapshot_text, "资源类型": "MQTT", "资源名称": first(item.get("InstanceName"), item.get("InstanceId")), "资源实例ID": item.get("InstanceId"), "地域": item.get("RegionId"), "运行状态": "服务中" if item.get("InstanceStatus") == 5 else "待确认", "业务用途/环境": first(item.get("InstanceName"), default="待确认"), "规格": f"{mqtt_type}{('｜' + item.get('Specific')) if item.get('Specific') else ''}", "软件/引擎版本": item.get("KernelVersion"), "付费方式": "包年包月" if item.get("InstanceType") != 2 else "按量付费", "到期时间": expiry, "剩余天数": remaining, "备份/快照状态": "MQTT托管服务；消息保留/容灾策略需实例专项复核", "网络与权限状态": "接入端点、Topic、Group和客户端数未导出", "风险等级": risk_level([], warnings), "风险说明": "；".join(warnings) or "服务中", "运维建议": "；".join(actions) or "补充连接数、消息堆积和告警视图", "数据来源": "云消息队列MQTT实例API", }) # Classic load balancers. for item in read_json(base / "slb-instances.json", {}).get("LoadBalancers", {}).get("LoadBalancer", []): warnings = [] actions = [] if str(item.get("NetworkType", "")).lower() == "classic": warnings.append("经典网络负载均衡") actions.append("评估迁移到VPC网络与新一代负载均衡") put(rows, { "快照时间": snapshot_text, "资源类型": "CLB", "资源名称": first(item.get("LoadBalancerName"), item.get("LoadBalancerId")), "资源实例ID": item.get("LoadBalancerId"), "地域": item.get("RegionId"), "可用区": item.get("MasterZoneId"), "运行状态": "正常" if item.get("LoadBalancerStatus") == "active" else "异常", "业务用途/环境": first(item.get("LoadBalancerName"), default="待确认"), "规格": item.get("LoadBalancerSpec"), "公网地址/访问端点": item.get("Address"), "付费方式": "包年包月" if item.get("PayType") == "PrePay" else "按量付费", "网络与权限状态": f"{item.get('AddressType', '')}/{item.get('NetworkType', '')}；主备可用区{item.get('MasterZoneId', '')}/{item.get('SlaveZoneId', '')}", "风险等级": risk_level([], warnings), "风险说明": "；".join(warnings) or "服务状态正常", "运维建议": "；".join(actions) or "持续监控后端健康检查", "数据来源": "CLB实例API", }) # OSS buckets, including ACL and versioning, without object listings. bucket_meta = {item.get("name"): item for item in read_json(base / "oss-buckets.json", [])} for item in read_json(base / "oss-summary.json", []): name = item.get("name", "") meta = bucket_meta.get(name, {}) reasons = [] warnings = [] actions = [] if item.get("acl") == "public-read-write": reasons.append("Bucket为公共读写") actions.append("立即复核用途并收敛为私有或最小权限") elif item.get("acl") == "public-read": warnings.append("Bucket为公共读") actions.append("确认是否为刻意公开的静态资源桶") if item.get("versioning") != "Enabled": warnings.append("版本控制未启用") actions.append("评估误删恢复需求并配置版本控制/生命周期") region = item.get("region", "") put(rows, { "快照时间": snapshot_text, "资源类型": "OSS", "资源名称": name, "资源实例ID": first(meta.get("id"), name), "地域": region, "运行状态": "正常", "业务用途/环境": "按Bucket名称判断用途，待责任人复核", "公网地址/访问端点": f"https://{name}.oss-{region}.aliyuncs.com", "付费方式": "按量付费", "备份/快照状态": f"版本控制：{item.get('versioning', 'unknown')}", "网络与权限状态": f"Bucket ACL：{item.get('acl', 'unknown')}", "风险等级": risk_level(reasons, warnings), "风险说明": "；".join(reasons + warnings) or "私有Bucket且版本控制已开启", "运维建议": "；".join(dict.fromkeys(actions)) or "保持访问日志和生命周期策略", "数据来源": "资源中心 + OSS ACL/Versioning API（未读取对象）", }) # NAS. for item in read_json(base / "nas-instances.json", {}).get("FileSystems", {}).get("FileSystem", []): put(rows, { "快照时间": snapshot_text, "资源类型": "NAS", "资源名称": first(item.get("Description"), item.get("FileSystemId")), "资源实例ID": item.get("FileSystemId"), "地域": "cn-beijing", "运行状态": "正常" if item.get("Status") == "Running" else "异常", "业务用途/环境": first(item.get("Description"), default="待确认"), "规格": f"{item.get('FileSystemType', '')}/{item.get('StorageType', '')}/{item.get('ProtocolType', '')}", "存储/容量": f"计量{item.get('MeteredSize', 0)}字节；容量上限{item.get('Capacity', 0)}字节", "付费方式": "按量付费" if item.get("ChargeType") == "PayAsYouGo" else "包年包月", "备份/快照状态": "NAS备份策略需专项复核", "网络与权限状态": "挂载点和权限组未导出", "风险等级": "待确认", "风险说明": "运行正常，但备份和挂载权限未纳入本快照", "运维建议": "补充挂载点、权限组、容量趋势和备份策略", "数据来源": "NAS实例API", }) # Table Store. for item in read_json(base / "ots-summary.json", []): name = first(item.get("alias"), item.get("name"), item.get("id")) put(rows, { "快照时间": snapshot_text, "资源类型": "表格存储", "资源名称": name, "资源实例ID": item.get("id"), "地域": item.get("region"), "运行状态": "正常" if item.get("status") == "normal" else "异常", "业务用途/环境": first(item.get("description"), default="待确认"), "规格": item.get("spec"), "软件/引擎版本": "Tablestore", "付费方式": "按量付费" if item.get("payment") == "PayAsYouGo" else "包年包月", "备份/快照状态": "数据表备份策略未纳入本快照", "网络与权限状态": item.get("network"), "风险等级": "待确认", "风险说明": "实例正常；表、容量、备份和访问策略需专项盘点", "运维建议": "补充表数量、容量、热点和备份策略", "数据来源": "资源中心资源配置API", }) resources = read_json(resource_center_path, {}).get("Resources", []) config_index: dict[tuple[str, str], dict[str, Any]] = {} config_root = base / "resource-config" for config_path in config_root.glob("*/*.json"): config = read_json(config_path, {}) config_index[(config.get("ResourceType", ""), config.get("ResourceId", ""))] = config # Remaining operational resources from Resource Center. type_map = { "ACS::SWAS::Instance": "轻量应用服务器", "ACS::FC::Function": "函数计算", "ACS::FCV3::Function": "函数计算", "ACS::WAFV3::DefenseResource": "WAF", "ACS::CloudFirewall::Instance": "云防火墙", "ACS::CDN::Domain": "CDN", "ACS::DCDN::Domain": "DCDN", } for meta in resources: resource_type = meta.get("ResourceType", "") if resource_type not in type_map: continue config_wrap = config_index.get((resource_type, meta.get("ResourceId", "")), {}) cfg = config_wrap.get("Configuration", {}) or {} name = first(meta.get("ResourceName"), cfg.get("InstanceName"), cfg.get("FunctionName"), cfg.get("DomainName"), meta.get("ResourceId")) status = first(cfg.get("Status"), cfg.get("InstanceStatus"), cfg.get("BusinessStatus"), cfg.get("DomainStatus"), default="待确认") expiry = local_time(first(cfg.get("ExpiredTime"), cfg.get("EndTime"), meta.get("ExpireTime"), default=None)) remaining = days_left(expiry, snapshot_dt) reasons: list[str] = [] warnings: list[str] = [] actions: list[str] = [] software = "" spec = "" public_address = "" private_address = "" payment = "按量付费" backup = "待专项复核" network = "资源中心已登记" if resource_type == "ACS::SWAS::Instance": software = f"{cfg.get('ImageName', '')} {cfg.get('ImageVersion', '')}".strip() spec = cfg.get("PlanId", "") public_address = cfg.get("PublicIpAddress", "") private_address = cfg.get("InnerIpAddress", "") payment = "包年包月" if cfg.get("PaymentType") == "PrePaid" else "按量付费" if remaining is not None and remaining <= 30: reasons.append(f"{remaining}天内到期") actions.append("立即确认用途和续费") elif resource_type in {"ACS::FC::Function", "ACS::FCV3::Function"}: software = cfg.get("Runtime", "配置读取失败") spec = f"内存{cfg.get('MemorySize', '待确认')}MiB" env = cfg.get("EnvironmentVariables", {}) or {} secret_keys = [key for key in env if re.search(r"PASSWORD|SECRET|TOKEN|ACCESS_KEY|PRIVATE_KEY", key, re.I)] if secret_keys: reasons.append("函数环境变量包含明文敏感凭据字段") actions.append("迁移至KMS/Secrets并轮换相关凭据") if re.search(r"nodejs14|debian10", software, re.I): warnings.append(f"运行时老旧：{software}") actions.append("升级函数运行时并回归验证") status = "待确认" network = "公网访问" if cfg.get("InternetAccess") else "网络配置待确认" elif resource_type == "ACS::CloudFirewall::Instance": spec = f"规格{cfg.get('Spec', '待确认')}" payment = "按量付费" if cfg.get("PaymentType") == "PayAsYouGo" else "包年包月" if cfg.get("CfwLog") is False: warnings.append("云防火墙日志未启用") actions.append("评估开启日志并设置留存周期") status = "正常" if str(status).lower() == "normal" else "待确认" network = "云防火墙实例" elif resource_type in {"ACS::CDN::Domain", "ACS::DCDN::Domain"}: public_address = name software = first(cfg.get("CdnType"), "DCDN" if resource_type.endswith("DCDN::Domain") else "CDN") network = f"CNAME：{cfg.get('Cname', '待确认')}" ssl_value = first(cfg.get("SslProtocol"), (cfg.get("CertInfos") or [{}])[0].get("SslProtocol"), default="unknown") if ssl_value == "off": warnings.append("CDN/DCDN HTTPS未启用") actions.append("确认域名用途并评估启用HTTPS") status = "正常" if str(status).lower() == "online" else "待确认" elif resource_type == "ACS::WAFV3::DefenseResource": detail = cfg.get("Detail", {}) or {} name = first(meta.get("ResourceName"), cfg.get("Resource"), meta.get("ResourceId")) spec = f"{cfg.get('Product', '')} {detail.get('protocol', '')}:{detail.get('port', '')}".strip() network = "WAF防护资源已接入；策略和告警需专项复核" status = "待确认" if remaining is not None and remaining <= 60 and not any("到期" in text for text in reasons): warnings.append(f"{remaining}天内到期") actions.append("纳入续费计划") put(rows, { "快照时间": snapshot_text, "资源类型": type_map[resource_type], "资源名称": name, "资源实例ID": meta.get("ResourceId"), "地域": meta.get("RegionId"), "运行状态": status if status in {"正常", "停止", "服务中", "异常", "待确认"} else "待确认", "业务用途/环境": first(cfg.get("Description"), meta.get("ResourceName"), default="待责任人补充"), "规格": spec, "软件/引擎版本": software, "公网地址/访问端点": public_address, "私网地址": private_address, "付费方式": payment, "到期时间": expiry, "剩余天数": remaining, "备份/快照状态": backup, "网络与权限状态": network, "风险等级": risk_level(reasons, warnings) if (reasons or warnings) else "待确认", "风险说明": "；".join(reasons + warnings) or "资源已登记，运行指标和策略待专项复核", "运维建议": "；".join(dict.fromkeys(actions)) or "补充负责人、告警和恢复验证", "数据来源": "阿里云资源中心资源配置API", }) # One network summary row keeps the view useful without expanding every VPC/vSwitch into low-signal rows. counts: dict[str, int] = {} for meta in resources: counts[meta.get("ResourceType", "")] = counts.get(meta.get("ResourceType", ""), 0) + 1 put(rows, { "快照时间": snapshot_text, "资源类型": "网络汇总", "资源名称": "VPC与交换网络汇总", "资源实例ID": "aggregate:vpc-network", "地域": "多地域", "运行状态": "待确认", "业务用途/环境": "ECS、RDS、Redis、CLB等资源的网络承载", "规格": f"VPC {counts.get('ACS::VPC::VPC', 0)}个；vSwitch {counts.get('ACS::VPC::VSwitch', 0)}个；路由表 {counts.get('ACS::VPC::RouteTable', 0)}个", "网络与权限状态": "本工作表按资源实例展示网络归属；完整路由和ACL需专项拓扑盘点", "风险等级": "待确认", "风险说明": "网络资源已计数，尚未逐条验证路由、ACL和跨域关系", "运维建议": "后续形成VPC-交换机-安全组-实例拓扑视图", "数据来源": "阿里云资源中心计数", }) return rows def apply_options(rows: list[dict[str, Any]], options: dict[str, dict[str, str]]) -> None: for record in rows: values = record["values"] for field in SINGLE_SELECT_FIELDS: text = values.get(field) if not text: continue option_id = options.get(field, {}).get(str(text)) if not option_id: raise ValueError(f"Missing option id for {field}={text}") values[field] = [{"id": option_id, "text": text}] def main() -> None: parser = argparse.ArgumentParser() parser.add_argument("--input-dir", type=Path, required=True) parser.add_argument("--resource-center", type=Path, required=True) parser.add_argument("--snapshot-time", required=True, help="YYYY-MM-DD HH:MM:SS Asia/Shanghai") parser.add_argument("--options-json", type=Path) args = parser.parse_args() snapshot_dt = datetime.strptime(args.snapshot_time, "%Y-%m-%d %H:%M:%S").replace(tzinfo=TZ) rows = build_rows(args.input_dir, args.resource_center, snapshot_dt) if args.options_json: apply_options(rows, read_json(args.options_json, {})) print(json.dumps({"records": rows, "count": len(rows)}, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()