运行之前,提供AWS密钥
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 分配")