基于Python和Boto3批量采集多AWS账户资源并导出至Excel/CSV求助

Hey there! Let's break down what's going wrong with your initial Python script first, then build out a complete solution that handles all your resource collection, multi-account support, and Excel/CSV export needs.

The error you're seeing happens because you're misunderstanding how boto3 returns data. When you iterate over resp (the response from describe_instances), you're looping through the keys of a dictionary (like 'Reservations'), not instance objects. You can't use dot notation (.instance-id) on a string key—instead, you need to traverse the nested dictionary structure boto3 returns.

Here's the corrected EC2 instance code that matches the data you were collecting in your bash script:

import boto3

session = boto3.Session(profile_name='default', region_name='me-south-1')
ec2_client = session.client('ec2')
response = ec2_client.describe_instances(
    Filters=[
        {
            'Name': 'instance-state-name',
            'Values': ['running']
        }
    ]
)

# Traverse the nested reservation -> instances structure
for reservation in response['Reservations']:
    for instance in reservation['Instances']:
        # Get the Name tag (handle cases where no Name tag exists)
        instance_name = next((tag['Value'] for tag in instance.get('Tags', []) if tag['Key'] == 'Name'), 'No Name')
        
        print(f"Name: {instance_name}")
        print(f"Instance ID: {instance['InstanceId']}")
        print(f"Instance Type: {instance['InstanceType']}")
        print(f"State: {instance['State']['Name']}\n")

To handle all your resources (EC2, ELBv2, ASG, EIP, RDS) and export to Excel, we'll use pandas for easy data formatting and Excel writing. First, install the required dependencies:

pip install boto3 pandas openpyxl

Here's the complete script that supports 80+ AWS accounts, collects all your desired resource data, and exports each account's data to a separate Excel file with dedicated sheets for each resource type:

import boto3
import pandas as pd
from typing import List, Dict

def get_ec2_data(session: boto3.Session) -> List[Dict]:
    """Collect EC2 instance details matching your bash query"""
    ec2_client = session.client('ec2')
    response = ec2_client.describe_instances()
    ec2_records = []
    
    for reservation in response['Reservations']:
        for instance in reservation['Instances']:
            name = next((tag['Value'] for tag in instance.get('Tags', []) if tag['Key'] == 'Name'), '')
            ec2_records.append({
                'Name': name,
                'InstanceId': instance['InstanceId'],
                'InstanceType': instance['InstanceType'],
                'Platform': instance.get('Platform', ''),
                'State': instance['State']['Name'],
                'PrivateIpAddress': instance.get('PrivateIpAddress', ''),
                'PublicIpAddress': instance.get('PublicIpAddress', ''),
                'AvailabilityZone': instance['Placement']['AvailabilityZone']
            })
    return ec2_records

def get_elbv2_data(session: boto3.Session) -> List[Dict]:
    """Collect ALB/NLB details"""
    elbv2_client = session.client('elbv2')
    response = elbv2_client.describe_load_balancers()
    elb_records = []
    
    for lb in response['LoadBalancers']:
        elb_records.append({
            'LoadBalancerArn': lb['LoadBalancerArn'],
            'DNSName': lb['DNSName'],
            'LoadBalancerName': lb['LoadBalancerName'],
            'Type': lb['Type'],
            'Scheme': lb['Scheme'],
            'State': lb['State']['Code']
        })
    return elb_records

def get_asg_data(session: boto3.Session) -> List[Dict]:
    """Collect Auto Scaling Group details"""
    asg_client = session.client('autoscaling')
    response = asg_client.describe_auto_scaling_groups()
    asg_records = []
    
    for asg in response['AutoScalingGroups']:
        asg_records.append({
            'AutoScalingGroupName': asg['AutoScalingGroupName'],
            'AutoScalingGroupARN': asg['AutoScalingGroupARN'],
            'MinSize': asg['MinSize'],
            'MaxSize': asg['MaxSize'],
            'DesiredCapacity': asg['DesiredCapacity'],
            'DefaultCooldown': asg['DefaultCooldown']
        })
    return asg_records

def get_eip_data(session: boto3.Session) -> List[Dict]:
    """Collect Elastic IP details"""
    ec2_client = session.client('ec2')
    response = ec2_client.describe_addresses()
    eip_records = []
    
    for addr in response['Addresses']:
        name = next((tag['Value'] for tag in addr.get('Tags', []) if tag['Key'] == 'Name'), '')
        eip_records.append({
            'Name': name,
            'PublicIp': addr['PublicIp'],
            'PrivateIpAddress': addr.get('PrivateIpAddress', '')
        })
    return eip_records

def get_rds_data(session: boto3.Session) -> List[Dict]:
    """Collect RDS instance details"""
    rds_client = session.client('rds')
    response = rds_client.describe_db_instances()
    rds_records = []
    
    for db in response['DBInstances']:
        rds_records.append({
            'DBInstanceIdentifier': db['DBInstanceIdentifier'],
            'DBInstanceClass': db['DBInstanceClass'],
            'Engine': db['Engine'],
            'EngineVersion': db['EngineVersion'],
            'DBInstanceStatus': db['DBInstanceStatus'],
            'MasterUsername': db['MasterUsername'],
            'EndpointAddress': db['Endpoint']['Address'] if 'Endpoint' in db else '',
            'MultiAZ': db['MultiAZ'],
            'AllocatedStorage': db['AllocatedStorage'],
            'PreferredBackupWindow': db['PreferredBackupWindow'],
            'BackupRetentionPeriod': db['BackupRetentionPeriod'],
            'StorageEncrypted': db['StorageEncrypted']
        })
    return rds_records

def process_account(profile_name: str, region: str = 'me-south-1'):
    """Process a single AWS account and export data to Excel"""
    try:
        session = boto3.Session(profile_name=profile_name, region_name=region)
        print(f"Processing account: {profile_name}")
        
        # Collect all resource data
        ec2_df = pd.DataFrame(get_ec2_data(session))
        elb_df = pd.DataFrame(get_elbv2_data(session))
        asg_df = pd.DataFrame(get_asg_data(session))
        eip_df = pd.DataFrame(get_eip_data(session))
        rds_df = pd.DataFrame(get_rds_data(session))
        
        # Export to Excel with separate sheets
        output_file = f"aws_resources_{profile_name}_{region}.xlsx"
        with pd.ExcelWriter(output_file, engine='openpyxl') as writer:
            ec2_df.to_excel(writer, sheet_name='EC2', index=False)
            elb_df.to_excel(writer, sheet_name='ELBv2', index=False)
            asg_df.to_excel(writer, sheet_name='AutoScaling', index=False)
            eip_df.to_excel(writer, sheet_name='EIP', index=False)
            rds_df.to_excel(writer, sheet_name='RDS', index=False)
        
        print(f"Successfully exported data to {output_file}\n")
    except Exception as e:
        print(f"Failed to process account {profile_name}: {str(e)}\n")

if __name__ == "__main__":
    # Replace this list with all your 80+ AWS profile names (from ~/.aws/credentials/config)
    AWS_PROFILES = ['account1', 'account2', 'default', 'prod-account', 'staging-account']
    
    # Process each account sequentially
    for profile in AWS_PROFILES:
        process_account(profile)

• Profile Management: Ensure all your AWS accounts are set up as profiles in your ~/.aws/credentials or ~/.aws/config files. If you prefer using IAM roles instead of static credentials, modify the session creation to use sts.assume_role.
• Rate Limiting: For 80+ accounts, you might hit AWS API rate limits. To fix this, add retry logic (use the tenacity library) or process accounts in parallel with concurrent.futures.ThreadPoolExecutor (be cautious not to overwhelm AWS APIs).
• Error Handling: The script includes basic error handling to ensure one failed account doesn't stop the entire batch. You can expand this to log errors to a file for later review.

内容的提问来源于stack exchange，提问作者jawad846