AWS Healthイベントを説明や影響を受けるリソースも含めて一覧で取得してみた（Boto3）

...
## 15.AWS_CLOUDSHELL_PERSISTENCE_EXPIRING

- arn:aws:sts::012345678910:assumed-role/cm-chiba.yukihiro/cm-chiba.yukihiro

## 16.AWS_LAMBDA_PLANNED_LIFECYCLE_EVENT

- arn:aws:lambda:ap-northeast-1:012345678910:function:sns-message-python
- arn:aws:lambda:ap-northeast-1:012345678910:function:NotifyBillingToSlack-HelloWorldFunction-16PVBRM43HUK9
- arn:aws:lambda:ap-northeast-1:012345678910:function:SecurityHubAccepter-LambdaFunction-pNvyVtCJmjl8

## 17.AWS_ACCOUNT_SECURITY_NOTIFICATION

- 012345678910

## 18.AWS_IAM_OPERATIONAL_ISSUE

- UNKNOWN
...

$ python aws_health_event_reporter.py 
AWS Health イベント情報の取得を開始します...
イベント一覧を取得中...

イベント取得成功: 21 件のイベント
ステータス別件数:
- closed: 13件
- open: 8件

closedステータスのイベントも出力しますか？ (y/n): y

イベントの詳細情報を5つの並列処理で取得中...
処理中: 5/21 件完了
処理中: 10/21 件完了
処理中: 15/21 件完了
処理中: 20/21 件完了
処理中: 21/21 件完了

処理完了:
- イベント情報: aws_health_events_20250317_084424.csv
- リソース詳細: aws_health_affected_resources_20250317_084424.txt
合計 21 件のイベント情報を出力しました

# 必要であればヘッダーを表示
echo -e "ARN\tSERVICE\tEVENT_TYPE_CODE\tEVENT_TYPE_CATEGORY\tREGION\tSTART_TIME\tEND_TIME\tLAST_UPDATED_TIME\tSTATUS_CODE\tEVENT_SCOPE_CODE"

# AWS Health イベントを取得
aws health describe-events \
  --region us-east-1 \
  --query "events[*].[
    arn,
    service,
    eventTypeCode,
    eventTypeCategory,
    region,
    startTime,
    endTime,
    lastUpdatedTime,
    statusCode,
    eventScopeCode
  ]" \
  --output text

import boto3
import csv
from datetime import datetime
import sys
from collections import Counter
import concurrent.futures

def main():
    """AWS Health イベント情報を取得してCSV・テキストファイルに出力する"""
    print("AWS Health イベント情報の取得を開始します...")

    try:
        # イベント一覧の取得
        print("イベント一覧を取得中...")
        health_client = boto3.client('health', region_name='us-east-1')
        events = get_all_events(health_client)

        # ステータス別の件数を表示
        status_counts = Counter(event.get('statusCode', 'N/A') for event in events)
        print(f"\nイベント取得成功: {len(events)} 件のイベント")
        print("ステータス別件数:")
        for status, count in status_counts.items():
            print(f"- {status}: {count}件")

        # closedステータスを含めるか確認
        include_closed = input("\nclosedステータスのイベントも出力しますか？ (y/n): ").lower().strip() in ['y', 'yes']
        if not include_closed:
            events = [e for e in events if e.get('statusCode') != 'closed']
            print(f"closedステータスを除外: {len(events)} 件のイベントを処理します")

        # イベントの詳細情報を並列取得
        print("\nイベントの詳細情報を5つの並列処理で取得中...")
        detailed_events = get_detailed_events_parallel(events)

        # ファイル出力
        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
        export_to_files(detailed_events, timestamp)

    except Exception as e:
        print(f"エラーが発生しました: {str(e)}")
        import traceback
        print(traceback.format_exc())
        sys.exit(1)

def get_all_events(client):
    """AWS Health APIから全イベントを取得"""
    events = []
    next_token = None

    while True:
        params = {'nextToken': next_token} if next_token else {}
        response = client.describe_events(**params)
        events.extend(response.get('events', []))
        next_token = response.get('nextToken')
        if not next_token:
            break

    return events

def get_event_details(event_with_index):
    """イベントの詳細情報を取得（並列処理用）"""
    index, event = event_with_index
    client = boto3.client('health', region_name='us-east-1')

    # 日時フォーマット用ヘルパー関数
    def format_time(dt):
        return dt.strftime('%Y-%m-%d %H:%M:%S') if isinstance(dt, datetime) else ''

    # イベント詳細取得
    try:
        details = client.describe_event_details(eventArns=[event['arn']])['successfulSet'][0]
        description = details['eventDescription']['latestDescription']
    except Exception as e:
        print(f"イベント詳細取得エラー ({event.get('eventTypeCode', 'unknown')}): {str(e)}")
        description = "詳細情報を取得できませんでした"

    # 影響リソース取得
    try:
        entities = client.describe_affected_entities(filter={'eventArns': [event['arn']]})['entities']
        resources = [entity.get('entityValue', 'N/A') for entity in entities]
    except Exception:
        resources = []

    # マッピング定義
    event_type_code = event.get('eventTypeCode', '')

    return {
        '連番': index,
        'サービス': event.get('service', ''),
        'イベントタイプ': event_type_code,
        '状態': event.get('statusCode', ''),
        'カテゴリ': event.get('eventTypeCategory', ''),
        'リージョン': event.get('region', ''),
        'アベイラビリティゾーン': event.get('availabilityZone', ''),
        '開始時刻': format_time(event.get('startTime')),
        '終了時刻': format_time(event.get('endTime')),
        '最終更新時刻': format_time(event.get('lastUpdatedTime')),
        'スコープ': event.get('eventScopeCode', ''),
        '影響を受けるリソースの数': len(resources),
        '説明': description,
        'リソース': resources,
        'EVENT_TYPE_CODE': event_type_code  # テキストファイル用
    }

def get_detailed_events_parallel(events, max_workers=5):
    """イベント詳細を並列取得"""
    indexed_events = list(enumerate(events, 1))
    detailed_events = []

    with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor:
        futures = {executor.submit(get_event_details, event_idx): i 
                  for i, event_idx in enumerate(indexed_events)}

        for i, future in enumerate(concurrent.futures.as_completed(futures)):
            if (i + 1) % 5 == 0 or i + 1 == len(indexed_events):
                print(f"処理中: {i + 1}/{len(indexed_events)} 件完了")

            try:
                detailed_events.append(future.result())
            except Exception as e:
                idx = futures[future] + 1
                print(f"イベント {idx} の処理中にエラー: {str(e)}")

    # 連番順にソート
    return sorted(detailed_events, key=lambda x: x['連番'])

def export_to_files(events, timestamp):
    """イベント情報をCSVとテキストファイルに出力"""
    # CSVファイル出力
    csv_filename = f"aws_health_events_{timestamp}.csv"
    fieldnames = ['連番', 'サービス', 'イベントタイプ', '状態', 'カテゴリ', 
                 'リージョン', 'アベイラビリティゾーン', '開始時刻', '終了時刻', '最終更新時刻', 
                 'スコープ', '影響を受けるリソースの数', '説明']

    with open(csv_filename, 'w', newline='', encoding='utf-8-sig') as csvfile:
        writer = csv.DictWriter(csvfile, fieldnames=fieldnames)
        writer.writeheader()

        for event in events:
            # 必要なフィールドのみ抽出し、説明の改行を削除
            row = {k: event[k] for k in fieldnames}
            row['説明'] = row['説明'].replace('\n', ' ').replace('\r', ' ')
            writer.writerow(row)

    # テキストファイル出力
    txt_filename = f"aws_health_affected_resources_{timestamp}.txt"

    with open(txt_filename, 'w', encoding='utf-8') as textfile:
        for event in events:
            textfile.write(f"## {event['連番']}.{event['EVENT_TYPE_CODE']}\n\n")

            if not event['リソース']:
                textfile.write("- 影響を受けるリソースはありません\n")
            else:
                for resource in event['リソース']:
                    textfile.write(f"- {resource}\n")

            textfile.write("\n")

    print(f"\n処理完了:")
    print(f"- イベント情報: {csv_filename}")
    print(f"- リソース詳細: {txt_filename}")
    print(f"合計 {len(events)} 件のイベント情報を出力しました")

if __name__ == "__main__":
    main()