企业官网建设流程全解析

年底做资源盘点这件事，我估计不少运维和研发同行都干过：一台台登录控制台，点开实例详情，抄CPU、内存、IP、到期时间，几十上百台搞下来，半天时间搭进去，还特别容易抄错行。后来我换了个思路，直接用Python SDK批量提取实例信息，脚本一跑，CSV一导出，几秒钟全部搞定，还能顺便按标签筛出业务归属、统计各区域资源数。这篇文章不绕弯子，把整个脚本怎么写、参数怎么调、坑在哪里，一次性完整过一遍，需要的话可以直接把代码拿走改改就能用。

1. 先理清需求：批量提取到底要解决什么

1.1 高频场景：资源盘点、成本分析、巡检报告

批量化获取云实例信息听起来是一个很技术的需求，但落到实际场景里，其实大多时候是业务驱动的。

最常见的就是资源盘点。公司账面上有多少台服务器、分布在哪几个地域、分别跑什么业务、每个实例是什么规格，这些信息如果靠人工去控制台翻，不仅慢，而且容易漏。尤其到了年底预算复盘或者内部审计的时候，运维得交一份完整的资源清单，这时候批量提取脚本就是救命稻草。

其次是成本分析和资源治理。超卖、浪费、闲置资源是云上普遍存在的问题，但前提是你得先有一份完整准确的实例清单。脚本拉下来的数据里包含实例规格、CPU、内存、带宽、创建时间、到期时间，把这些字段跟计费数据一关联，哪些实例快到期了、哪些实例规格过大利用率低，一眼就能看出来。

还有一类是自动化巡检。比如每周定期跑一遍脚本，检查是不是有实例处于Running状态但超过多少天没有登录，或者有没有实例快到期了还没有设置自动续费。数据一旦能批量拿到，后续的自动化判断就顺理成章了。

我做这套脚本的时候，手头正好是跨3个地域、两百多台ECS实例的存量项目，前后对比非常明显：人工操作光汇总就要大半天，脚本跑下来不到1分钟。这里面的核心思路其实不复杂：云平台本身提供了标准API，不管你有几十台还是几千台实例，本质都是把分页接口拿出来的数据拼成一张表。

1.2 方案选型：为什么选择Python SDK而不是控制台导出

先别急着写代码，方案选型这一步很值得说几句。

其实阿里云控制台本身是提供导出功能的，在ECS实例列表页也能点按钮导出CSV。那为什么还要写脚本？原因有三个。

第一，控制台导出有数量上限和字段限制。实例数量一多，控制台导出往往只导当前页，或者导出的字段只有固定几个，关键信息比如Tags、VPC私网IP不一定都带。而SDK拿的是API原始数据，字段全、数量不做截断。

第二，控制台导出的数据不方便二次处理。导出来是扁平表格，实际上一个实例的私网IP、公网IP、Tags都是嵌套结构，需要自己拼装。脚本可以直接在拉取的过程中清洗格式化，导出一版就是你想要的样子。

第三，无法自动化。控制台得靠人点，脚本却可以挂在定时任务上。配合RAM子账号的最小授权，让脚本定期跑，数据落库，形成长期趋势记录，这才能真正体现批量提取的价值。

再对比一下CLI工具和Python SDK。云平台的CLI（比如阿里云CLI）确实也能拿到实例列表，而且写一条命令就能跑，适合临时查一下。但如果你要做复杂的字段解析、多地域遍历、把结果写进Excel、对接内部门户，CLI的二次处理能力就弱了，Python脚本可以灵活操作数据结构，配合pandas、openpyxl这类库做起来非常顺手。

所以我最终选了Python SDK，具体来说是aliyun-python-sdk-core加上aliyun-python-sdk-ecs这套组合。代码直白、资料多、排错容易，适合绝大多数团队成员接手维护。

2. 环境准备：装SDK、配权限、初始化客户端

2.1 安装Python SDK

开发环境建议用Python 3.6以上的版本，我用的是3.9，目前跑得很稳。安装命令很简单，直接pip装两个包就行：

pip install aliyun-python-sdk-core pip install aliyun-python-sdk-ecs

如果你想用新版SDK方式，阿里云现在主推的是alibabacloud_ecs20140526，底层走的是RPC风格API，代码结构更贴近OpenAPI规范：

pip install alibabacloud_ecs20140526

两套SDK都能用，我这个脚本基于老版SDK写，原因主要是存量代码多、网上排错案例丰富。新版SDK的API响应类型注解更完整，适合新项目。如果你只是想要一个能快速跑的脚本，老版完全够用，后面我会把关键的区别点也提一下。

2.2 创建RAM子账号与最小权限授权

这一步很多人会跳过，直接拿主账号的AccessKey写在脚本里。我强烈不建议这么干，尤其是脚本要固化在服务器上长期跑的时候，主账号Key泄露等于整个云账号被脱管。正确做法是在RAM控制台建一个子账号，只授予它“只读”级别的ECS权限。

具体操作流程是这样的：

1. 在RAM访问控制台创建用户，比如叫cloudecs_reader，选择“编程访问”，勾选AccessKey生成方式。
2. 给这个子账号附加权限策略，系统策略里找AliyunECSReadOnlyAccess，这个策略刚好覆盖ECS实例相关的只读操作权限。
3. 如果脚本还要读取其他产品信息，比如磁盘、快照，再按需添加对应的只读策略，原则是最小够用。

有个点要提醒：RAM权限变更往往不是即时生效的，有时候刚授权就调用API会出现Forbidden，等一两分钟再重试就好了。别问我怎么知道的，第一次跑脚本遇到这个错，我还以为是密钥配错了。

AccessKey生成后，建议把ID和Secret放到环境变量里，不要硬编码在脚本文件中，尤其是脚本如果提交到Git仓库，密钥一旦被扫描工具发现，后果很严重。我自己的习惯是读~/.env或者直接用os.getenv()。

2.3 初始化AcsClient，配置Region

老版SDK的核心入口是AcsClient，初始化的时候需要三个参数：AccessKey ID、AccessKey Secret、默认地域ID。

from aliyunsdkcore.client import AcsClient client = AcsClient( os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'), os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'), 'cn-hangzhou' )

这里有一个容易忽略的细节：AcsClient初始化时指定的地域是“默认地域”，但实际调用ECS接口时，请求的地域由DescribeInstancesRequest中的RegionId参数决定。所以如果你的资源分布在多个地域，单独写死一个Region是不够的，需要在每个请求里显式指定RegionId，这一点在后面的多Region汇总部分还会用到。

另外，SDK底层默认使用HTTP协议，如果你对传输安全有更高要求，可以开启HTTPS（老版SDK在请求对象里有set_protocol_type('https')方法）。公网环境里我建议开启，传输过程中AccessKey不会被明文抓包。

3. 核心实现流程：从单页请求到批量导出

3.1 最小脚本：先拿到第一页数据

先把最小可用的脚本写出来，目标很简单：请求指定地域的ECS实例列表，打印返回的JSON。这一步跑通了，后面就顺了。

from aliyunsdkcore.client import AcsClient from aliyunsdkecs.request.v20140526.DescribeInstancesRequest import DescribeInstancesRequest import json client = AcsClient( os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID'), os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET'), 'cn-hangzhou' ) request = DescribeInstancesRequest() request.set_accept_format('json') request.set_RegionId('cn-hangzhou') request.set_PageSize(10) request.set_PageNumber(1) response = client.do_action_with_exception(request) data = json.loads(response) print(json.dumps(data, ensure_ascii=False, indent=2))

有几个点解释一下。set_accept_format('json')很关键，如果不设置，默认返回的是XML字符串，解析起来更麻烦。do_action_with_exception是同步调用方法，如果请求出错，SDK会直接抛出异常，我们可以捕获异常做重试或者输出错误日志。

响应里的TotalCount是实例总数，Instances字段下有一个Instance列表，每个元素就是一台实例的所有属性。这里我建议第一次跑的时候把完整的JSON打出来看一眼，你会发现字段比控制台上展示的多得多，包括HostName、SerialNumber、VpcAttributes、EipAddress、Tags等，这些都是控制台列表页不一定展示的数据。

3.2 翻页逻辑：PageNumber/PageSize 与 NextToken

实例数量超过一页之后就涉及到翻页逻辑。先说一个最直观的翻页方式——PageNumber/PageSize：

• PageSize：每页返回条数，范围是1~100，默认值是10。
• PageNumber：页码，从1开始计数。

翻页的终止条件有两种判断思路。一种是拿当前已拉取的数量跟TotalCount比较，小于TotalCount就继续拉下一页；另一种是判断当前返回的列表是否为空。我更推荐第一种，因为哪怕最后一页恰好是空列表也能退出循环，但用数量判断更加直白：

def fetch_all_instances(region_id, page_size=100): instances = [] page_number = 1 while True: request = DescribeInstancesRequest() request.set_accept_format('json') request.set_RegionId(region_id) request.set_PageSize(page_size) request.set_PageNumber(page_number) response = client.do_action_with_exception(request) data = json.loads(response) page_instances = data.get('Instances', {}).get('Instance', []) instances.extend(page_instances) total = int(data.get('TotalCount', 0)) if len(instances) >= total: break page_number += 1 return instances

这里有一个高频坑：PageNumber从1开始，而不是从0开始。如果你按照编程里数组下标从0开始的习惯写，第一页数据和第二页数据会重复，而且最后一页数据永远拉不到。我第一版脚本就栽在这里，当时排查了很久，后来打印请求参数才发现页码从一开始传错了。

另外，新版ECS API其实还支持NextToken配合MaxResults的分页方式。我简单说下区别：NextToken方式适合深度翻页场景，大数据量下性能更好，不用每次从头算偏移量。老版PageNumber方式代码更好懂，脚本体量不大的话完全足够。实际业务里，几百台实例用PageNumber完全没压力。

3.3 字段解析与实例信息清洗

拿到原始数据后，重点是把嵌套结构解析成扁平化的记录。我最常用的字段包括：

• InstanceId：实例ID，唯一标识
• InstanceName：实例名称
• Status：运行状态，比如Running、Stopped
• InstanceType：实例规格
• Cpu/Memory：CPU核数和内存大小，单位分别是核和MiB
• OSName：操作系统名称
• PublicIpAddress：公网IP数组
• VpcAttributes.PrivateIpAddress：私网IP数组
• CreationTime/ExpiredTime：创建时间和到期时间
• Tags.Tag：标签数组

这里面的嵌套结构是重点。PublicIpAddress在API响应里是一个对象，对象下是IpAddress数组；私网IP藏在VpcAttributes下面。解析的时候要格外小心，如果某个实例是纯VPC网络且没有绑定公网IP，这些字段可能是空对象，直接索引取值会报KeyError。我的处理方式是统一用.get()加默认值，并且把多个IP用逗号拼接成一个字符串：

def extract_instance_info(instance, region_id): public_ips = instance.get('PublicIpAddress', {}).get('IpAddress', []) private_ips = instance.get('VpcAttributes', {}).get('PrivateIpAddress', {}).get('IpAddress', []) tags = [] for tag in instance.get('Tags', {}).get('Tag', []): tags.append(f"{tag.get('TagKey')}:{tag.get('TagValue')}") return { 'region_id': region_id, 'instance_id': instance.get('InstanceId'), 'instance_name': instance.get('InstanceName'), 'status': instance.get('Status'), 'instance_type': instance.get('InstanceType'), 'cpu': instance.get('Cpu'), 'memory_mib': instance.get('Memory'), 'os_name': instance.get('OSName'), 'public_ip': ','.join(public_ips), 'private_ip': ','.join(private_ips), 'creation_time': instance.get('CreationTime'), 'expired_time': instance.get('ExpiredTime'), 'tags': ';'.join(tags) }

这个函数是整个脚本里改动最多的部分。第一次跑完我发现导出的CSV里IP列很多是空的，排查后才发现VpcAttributes里的私网IP字段路径不对。所以建议你写完解析函数后，先拿一台实例完整打印JSON，对照字段路径逐层确认，不要靠猜。

3.4 导出CSV报表，一份可直接用的脚本模板

数据解析完成后，导出CSV是性价比最高的方案。注意Windows下用Excel打开CSV经常会出现中文乱码，解决方法是写入时加encoding='utf-8-sig'，也就是在文件开头加BOM。这一个小细节能省很多不必要的麻烦。

import csv def export_to_csv(instances, filename): if not instances: print('没有拉取到任何实例，不生成文件') return fieldnames = list(instances[0].keys()) with open(filename, 'w', newline='', encoding='utf-8-sig') as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(instances)

把上面这些函数拼起来，就是一个完整的批量提取脚本：

import os import json import csv import time from aliyunsdkcore.client import AcsClient from aliyunsdkecs.request.v20140526.DescribeInstancesRequest import DescribeInstancesRequest ACCESS_KEY_ID = os.getenv('ALIBABA_CLOUD_ACCESS_KEY_ID') ACCESS_KEY_SECRET = os.getenv('ALIBABA_CLOUD_ACCESS_KEY_SECRET') client = AcsClient(ACCESS_KEY_ID, ACCESS_KEY_SECRET, 'cn-hangzhou') def fetch_all_instances(region_id, page_size=100): instances = [] page_number = 1 while True: request = DescribeInstancesRequest() request.set_accept_format('json') request.set_RegionId(region_id) request.set_PageSize(page_size) request.set_PageNumber(page_number) response = client.do_action_with_exception(request) data = json.loads(response) page_instances = data.get('Instances', {}).get('Instance', []) instances.extend(page_instances) total = int(data.get('TotalCount', 0)) if len(instances) >= total: break page_number += 1 return instances def extract_instance_info(instance, region_id): public_ips = instance.get('PublicIpAddress', {}).get('IpAddress', []) private_ips = instance.get('VpcAttributes', {}).get('PrivateIpAddress', {}).get('IpAddress', []) tags = [] for tag in instance.get('Tags', {}).get('Tag', []): tags.append(f"{tag.get('TagKey')}:{tag.get('TagValue')}") return { 'region_id': region_id, 'instance_id': instance.get('InstanceId'), 'instance_name': instance.get('InstanceName'), 'status': instance.get('Status'), 'instance_type': instance.get('InstanceType'), 'cpu': instance.get('Cpu'), 'memory_mib': instance.get('Memory'), 'os_name': instance.get('OSName'), 'public_ip': ','.join(public_ips), 'private_ip': ','.join(private_ips), 'creation_time': instance.get('CreationTime'), 'expired_time': instance.get('ExpiredTime'), 'tags': ';'.join(tags) } def export_to_csv(instances, filename): if not instances: print('没有拉取到任何实例，不生成文件') return fieldnames = list(instances[0].keys()) with open(filename, 'w', newline='', encoding='utf-8-sig') as f: writer = csv.DictWriter(f, fieldnames=fieldnames) writer.writeheader() writer.writerows(instances) if __name__ == '__main__': region = 'cn-hangzhou' raw_instances = fetch_all_instances(region) rows = [extract_instance_info(ins, region) for ins in raw_instances] export_to_csv(rows, f'ecs_instances_{region}.csv') print(f'共拉取 {len(rows)} 台实例，已导出到 ecs_instances_{region}.csv')

这段代码放到服务器上，配好环境变量，直接python fetch_ecs.py就能跑。核心逻辑不复杂，但每一步都验证过，实用优先。

4. 实战中的坑：分页遗漏、权限异常、限流重试

4.1 常见问题速查表

我在实际跑这个脚本的过程中，遇到过不少典型问题，整理成表格方便你直接对照排查。

这个表里的问题我基本都踩过。其中最有迷惑性的是分页遗漏：表面上看脚本没有报错，CSV也生成了，数量却对不上。如果不拿TotalCount跟实际拉取数做核对，很难发现。所以我建议脚本运行完以后，把这个数字打印出来，跟控制台实例列表页的数量做一次人工比对，确认无误后再长期使用。

4.2 三个实操避坑心得

再说三个比较隐蔽的坑，这些都是实际项目里反反复复验证过的。

第一个坑是不同地域的默认VPC与交换机信息差异。有些老实例的网络类型是经典网络，VpcAttributes字段的结构跟VPC实例不完全一样，直接取PrivateIpAddress可能得到空值。比较稳的做法是解析的时候同时兼容经典网络和VPC网络。不过现在新购实例基本都是VPC网络，这个坑主要存在于存量资源很老的环境里。

第二个坑是Tags解析。有些实例没打任何标签，响应里的Tags可能是空对象，也可能是缺失字段，直接遍历会报TypeError。写解析函数时一定要先判空再遍历，不要偷懒直接取。

第三个坑是RAM策略同步延迟。新创建的子账号即使立刻绑定了授权策略，第一次调用API也可能报Forbidden。这不是代码问题，等一两分钟再跑往往就正常了。如果等到五分钟后还报错，再去排查策略有没有绑定到正确的主体。

4.3 加上异常重试，脚本更皮实

云平台API偶尔会出现瞬时网络抖动或者后端超时，尤其是拉几百上千台实例、连续发很多请求的时候。所以我在实际项目里给请求封装了重试逻辑。

def do_request_with_retry(request, retry_times=3, sleep_seconds=1): for attempt in range(retry_times): try: response = client.do_action_with_exception(request) return json.loads(response) except ServerException as e: if attempt < retry_times - 1: print(f'第 {attempt+1} 次请求失败: {e}，{sleep_seconds} 秒后重试...') time.sleep(sleep_seconds) else: raise

这里是基于异常类型分类的：ServerException表示服务端错误，可以重试；ClientException表示客户端参数错误，比如AccessKey不对、RegionId不存在，这类问题重试多少次都没用，直接抛出去反而方便定位。如果是限流错误，比如Throttling，建议重试间隔从1秒逐步拉长，避免雪上加霜。

5. 再进一步：多Region汇总、标签筛选与定时巡检

5.1 多Region遍历汇总

如果你的资源分布在多个地域，单地域脚本就不够用了。思路很简单：维护一个地域列表，循环调用同一个拉取函数，最后把结果合并进同一个CSV。

regions = ['cn-hangzhou', 'cn-shanghai', 'cn-beijing', 'cn-shenzhen', 'us-west-1'] all_rows = [] for region in regions: try: raw_instances = fetch_all_instances(region) rows = [extract_instance_info(ins, region) for ins in raw_instances] all_rows.extend(rows) print(f'{region}: {len(rows)} 台实例') time.sleep(0.3) except Exception as e: print(f'{region} 拉取失败: {e}') continue

注意循环里每次请求后加了time.sleep(0.3)，控制请求频率，避免触发限流。另外不同地域的实例可能数量差异非常大，我见过有些地域只有两三台，有些地域上百台，循环里做好逐地域日志输出很有必要。万一某个地域的Key权限没配置好，也能快速定位是哪一步挂掉了。

5.2 按标签筛选目标实例

批量提取脚本跑熟了之后，下一步基本就是做精细化筛选。最常用的筛选条件是标签，比如只提取Department:Marketing这个标签的实例，或者只提取Environment:Production的实例。

在DescribeInstancesRequest里可以直接传标签条件过滤，不用在本地二次过滤：

request.set_TagKey('Environment') request.set_TagValue('Production')

这个方式的好处是服务端直接把结果过滤掉，拉取数据量和传输开销都小很多。但要注意的是，SDK里一次最多只能传一组TagKey/TagValue条件，如果你需要“多标签AND过滤”，从服务端拉全量数据后在本地用Python过滤可能更灵活。我通常的做法是：全量拉取到本地，然后用列表推导式筛选，这样标签多了也不怕。

还有一个比较常用的本地筛选是按到期时间。把ExpiredTime字符串转成日期，跟当前日期做对比，定位90天内到期的实例，这个信息对续费决策非常有用：

from datetime import datetime threshold = datetime.now().replace(year=datetime.now().year + 1) expiring_soon = [] for row in all_rows: try: expired_at = datetime.strptime(row['expired_time'], '%Y-%m-%dT%H:%M:%SZ') if expired_at < threshold: expiring_soon.append(row) except (ValueError, TypeError): continue

ExpiredTime的格式是UTC时间字符串，转换的时候记得加时区偏移，不要直接把UTC时间当本地时间用，否则到期判断可能偏几个小时。

5.3 接入定时任务与到期提醒

脚本稳定后，可以挂到定时任务上，实现真正意义上的自动化巡检。

Linux服务器上用crontab最简单，比如每周一早上9点跑一次：

0 9 * * 1 cd /opt/ecs-inventory && /usr/bin/python3 fetch_ecs.py >> logs/ecs_inventory.log 2>&1

跑完以后，如果你希望把结果推送到内部协作工具，可以在脚本里加一个发送消息的逻辑。比如统计出即将到期的实例清单后，往企业微信群里发一条机器人消息，或者把生成的CSV作为附件发送。这一步纯粹是业务定制，但脚本的基础能力已经提供了。

除了定时任务，还有一个更优雅的载体是云上的定时触发函数，比如函数计算配合定时触发器，每个月自动跑一次资源盘点，结果直接写回到对象存储，连维护服务器都省了。不过这个方案涉及的产品会多一些，我这里就不展开了，感兴趣的可以从定时触发这个思路入手，把脚本迁移到函数计算的代码包里。

最后说一点个人心得

这套脚本我从最早的单地域简单版，一路改到现在的多地域汇总、标签筛选、到期预警版本，最大的体会是：云平台的API本身不难，真正花时间的是理解接口的返回结构和边界条件。一次把字段路径摸透、把分页逻辑写对，后面所有基于这份数据的分析都会省力很多。如果你也是第一次上手，建议先拿一个小规模账号跑通单地域版本，确认字段没问题，再逐步扩展，不要一上来就写多地域加上各种筛选，调优的时候容易找不到问题出在哪个环节。另外，AccessKey的保密始终是底线，脚本里的密钥一律走环境变量，能不用主账号Key就不用。这套东西跑起来之后，你会发现资源盘点从一件苦力活变成了一件自动化的事，后面再接成本分析、闲置资源识别，都顺理成章。