使用Python脚本快速批量删除OSS Bucket

importoss2importtimeimportsignalfromtqdmimporttqdmclassTimeoutException(Exception):
passdeftimeout_handler(signum, frame):
raiseTimeoutException()
signal.signal(signal.SIGALRM, timeout_handler)
defwrite_results_to_file(result_filename, bucket_count, buckets_detail, delete_success, delete_failed, delete_timings, include_buckets, exclude_buckets):
withopen(result_filename, 'w') asf:
f.write(f"Total buckets listed: {bucket_count}\n")
f.write("Bucket Details (Name, Region, Status):\n")
forbucketinbuckets_detail:
status='Included'ifbucket['bucket_name'] ininclude_bucketselse'Excluded'ifbucket['bucket_name'] inexclude_bucketselse'Not Specified'f.write(f"{bucket['bucket_name']}: {bucket['region']} (status: {status})\n")
f.write(f"\nTotal buckets successfully deleted: {len(delete_success)}\n")
f.write("Successfully Deleted Buckets:\n")
forbucket_nameindelete_success:
f.write(f"{bucket_name} (time taken: {delete_timings.get(bucket_name, 'N/A'):.2f}s)\n")
total_failed=len(delete_failed)
f.write(f"\nTotal buckets failed to delete: {total_failed}\n")
f.write("Failed to Delete Buckets:\n")
forfailed_bucketindelete_failed:
bucket_name=failed_bucket['bucket_name']
error_details=failed_bucket['error']
time_taken=delete_timings.get(bucket_name, 'N/A')
# Customize the error message based on the error code (EC)if'0024-00000001'inerror_details:
reason="Data lake storage is disabled. Please delete using the console."elif'0055-00000011'inerror_details:
reason="Bucket still binding access points. Please delete access points, then delete the bucket."elif"WORM Locked state"inerror_details:
reason="WORM enabled; not allowed to delete before expiration."else:
reason=error_details# Default reason if none of the specific EC codes are present# Write the failure reason to the filef.write(f"{bucket_name} (time taken: {time_taken}s, reason: {reason})\n")
defcheck_bucket_worm_status(bucket):
try:
worm_info=bucket.get_bucket_worm()
ifworm_info.state=='Locked':
print(f"Bucket '{bucket.bucket_name}' is in WORM state 'Locked', skipping deletion.")
returnFalseexceptoss2.exceptions.NoSuchWORMConfiguration:
print(f"Bucket '{bucket.bucket_name}' does not have a WORM configuration.")
exceptoss2.exceptions.RequestErrorase:
# 处理网络相关的异常print(f"Failed to check WORM status for bucket '{bucket.bucket_name}' due to network error: {e}")
returnFalseexceptoss2.exceptions.OssErrorase:
# 处理其他 OSS API 错误print(f"Failed to check WORM status for bucket '{bucket.bucket_name}' due to OSS error: {e}")
returnFalsereturnTruedefdelete_all_objects(bucket):
objects_to_delete= []
object_count=0# 初始化对象计数器forobjinoss2.ObjectIterator(bucket):
objects_to_delete.append(obj.key)
object_count+=1# 如果积累了足够多的对象，进行批量删除iflen(objects_to_delete) >=1000:
print(f"Deleting batch of 1000 objects...")  # 添加此行以打印正在执行的批量删除操作bucket.batch_delete_objects(objects_to_delete)
objects_to_delete= []
print(f"Deleted 1000 objects, continuing...")  # 添加此行以显示当前进度# 执行剩余的删除操作（如果有）ifobject_count>0andobjects_to_delete:
print(f"Deleting final batch of {len(objects_to_delete)} objects...")  # 添加此行以打印最后一批对象的删除操作bucket.batch_delete_objects(objects_to_delete)
elifobject_count==0:
print(f"No objects to delete in bucket '{bucket.bucket_name}'.")
defdelete_all_live_channels(bucket):
live_channel_count=0forlive_channel_infoinoss2.LiveChannelIterator(bucket):
name=live_channel_info.namebucket.delete_live_channel(name)
live_channel_count+=1iflive_channel_count>0:
print(f"All live channels deleted in bucket '{bucket.bucket_name}'.")
else:
print(f"No live channels to delete in bucket '{bucket.bucket_name}'.")
defdelete_all_multipart_uploads(bucket):
multipart_upload_count=0forupload_infoinoss2.MultipartUploadIterator(bucket):
key=upload_info.keyupload_id=upload_info.upload_idbucket.abort_multipart_upload(key, upload_id)
multipart_upload_count+=1ifmultipart_upload_count>0:
print(f"All multipart uploads aborted in bucket '{bucket.bucket_name}'.")
else:
print(f"No multipart uploads to abort in bucket '{bucket.bucket_name}'.")
defdelete_all_object_versions(bucket):
next_key_marker=Nonenext_versionid_marker=NonewhileTrue:
result=bucket.list_object_versions(key_marker=next_key_marker, versionid_marker=next_versionid_marker)
# 判断是否存在任何版本或删除标记ifnotresult.versionsandnotresult.delete_marker:
print(f"No object versions or delete markers to delete in bucket '{bucket.bucket_name}'.")
breakversions_to_delete=oss2.models.BatchDeleteObjectVersionList()
# 追加待删除的对象版本和删除标记forversion_infoinresult.versions:
versions_to_delete.append(oss2.models.BatchDeleteObjectVersion(version_info.key, version_info.versionid))
fordel_maker_infoinresult.delete_marker:
versions_to_delete.append(oss2.models.BatchDeleteObjectVersion(del_maker_info.key, del_maker_info.versionid))
# 执行批量删除操作versions_count=len(result.versions) +len(result.delete_marker)  # 计算将要删除的数量print(f"Deleting {versions_count} object versions and/or delete markers...")  # 使用计算出的数量bucket.delete_object_versions(versions_to_delete)
# 更新下一轮迭代的标记next_key_marker=result.next_key_markernext_versionid_marker=result.next_versionid_marker# 如果没有更多的版本或删除标记，则退出循环ifnotresult.is_truncated:
print(f"All object versions and delete markers deleted.")
breakcurrent_timestamp=time.strftime("%Y%m%d%H%M%S")
# 添加 include_buckets 集合，列出不删除的 bucket。exclude_buckets=set(['<your-bucket>'])
# 添加 include_buckets 集合，列出要删除的 bucket。如果为空，则应用 exclude。如果不为空，则只应用 include。include_buckets=set([])
auth=oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())
service=oss2.Service(auth, 'oss-cn-hangzhou.aliyuncs.com')
bucket_list=list(oss2.BucketIterator(service))
ifinclude_buckets:
buckets_to_process= [bforbinbucket_listifb.nameininclude_buckets]
else:
buckets_to_process= [bforbinbucket_listifb.namenotinexclude_buckets]
bucket_count=len(buckets_to_process)
buckets_detail= [{'bucket_name': b.name, 'region': b.location} forbinbuckets_to_process]
delete_success= []
delete_failed= []
delete_timings= {b.name: 'Not Attempted'forbinbuckets_to_process}
pbar=tqdm(total=bucket_count, desc="Deleting Buckets", unit="bucket", leave=False)
try:
forbucket_infoinbuckets_to_process:
bucket_name=bucket_info.nameregion=bucket_info.locationendpoint=f'https://{region}.aliyuncs.com'bucket=oss2.Bucket(auth, endpoint, bucket_name)
# 调用 check_bucket_worm_status 函数来检查 WORM 状态can_delete=check_bucket_worm_status(bucket)
ifnotcan_delete:
delete_failed.append({
'bucket_name': bucket_name,
'error': "Skipped due to WORM Locked state or unable to check status."            })
delete_timings[bucket_name] ='Skipped'pbar.update(1)
continue# 如果 WORM 状态允许，继续执行删除操作print(f"Processing bucket '{bucket_name}'...")
try:
signal.alarm(3600)
start_time=time.time()
# 删除所有直播频道delete_all_live_channels(bucket)
# 删除所有分片上传delete_all_multipart_uploads(bucket)
# 删除所有对象delete_all_objects(bucket)
# 删除所有对象版本delete_all_object_versions(bucket)
# 尝试删除bucketprint(f"Deleting bucket '{bucket_name}'...")
bucket.delete_bucket()
signal.alarm(0)
end_time=time.time()
time_taken=end_time-start_timedelete_timings[bucket_name] =time_takendelete_success.append(bucket_name)
print(f"Bucket '{bucket_name}' deleted successfully in {time_taken:.2f}s")
pbar.update(1)
exceptTimeoutException:
print(f"Deleting bucket '{bucket_name}' timed out after 300 seconds.")
delete_failed.append({'bucket_name': bucket_name, 'error': 'Timeout after 300 seconds'})
delete_timings[bucket_name] ='Timeout'pbar.update(1)
exceptoss2.exceptions.OssErrorase:
signal.alarm(0)
end_time=time.time()
time_taken=end_time-start_timedelete_timings[bucket_name] =time_takenerror_message=str(e)
print(f"Failed to delete bucket '{bucket_name}' in {time_taken:.2f}s: {error_message}")
delete_failed.append({'bucket_name': bucket_name, 'error': error_message})
pbar.update(1)
exceptKeyboardInterrupt:
print("\nOperation interrupted by user.")
finally:
pbar.close()
result_filename=f"{current_timestamp}_bucket_delete_result.txt"write_results_to_file(result_filename, bucket_count, buckets_detail, delete_success, delete_failed, delete_timings, include_buckets, exclude_buckets)
print(f"Script execution completed. Check {result_filename} for details.")