import boto3 def find_cloudfront_distribution_id(domain_name, aws_access_key_id, aws_secret_access_key): # 创建 CloudFront 资源 cloudfront = boto3.client('cloudfront', aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key) # 获取所有 CloudFront 分配 distributions = cloudfront.list_distributions()['DistributionList']['Items'] for distribution in distributions: # 检查分配是否有别名 if 'Aliases' in distribution and 'Items' in distribution['Aliases']: # 检查分配的域名是否包含所提供的域名 for alias in distribution['Aliases']['Items']: if alias == domain_name: return distribution['Id'] return None def invalidate_cloudfront_cache(distribution_id, paths, aws_access_key_id, aws_secret_access_key): # 创建 CloudFront 客户端 cloudfront = boto3.client('cloudfront', aws_access_key_id=aws_access_key_id, aws_secret_access_key=aws_secret_access_key) # 创建失效批次 invalidation = { 'DistributionId': distribution_id, 'InvalidationBatch': { 'Paths': { 'Quantity': len(paths), 'Items': paths }, 'CallerReference': 'unique-identifier' # 为失效批次提供唯一标识符 } } # 提交失效请求 response = cloudfront.create_invalidation(**invalidation) # 检查失效请求是否成功 if response['ResponseMetadata']['HTTPStatusCode'] == 201: print("缓存失效请求成功。") else: print("缓存失效请求失败。") # 输入要清理的域名 domain_name = input('请输入需要清理缓存的域名:\n') # AWS 访问密钥和秘密访问密钥 aws_access_key_id = '' aws_secret_access_key = '' # 查找 CloudFront 分配的 ID distribution_id = find_cloudfront_distribution_id(domain_name, aws_access_key_id, aws_secret_access_key) if distribution_id: # 要失效的路径 paths_to_invalidate = ['/*'] # 失效缓存 invalidate_cloudfront_cache(distribution_id, paths_to_invalidate, aws_access_key_id, aws_secret_access_key) else: print(f"未找到与域名 '{domain_name}' 相关联的 CloudFront 分配")