Delete cold objects

import os
import boto3
import csv
import json
from botocore.client import Config
from datetime import datetime, timedelta, timezone

boto_config = Config(
    region_name='US-EAST-04A',
    s3={'addressing_style': 'virtual'}
)

s3 = boto3.client(
    's3',
    endpoint_url='https://cwobject.com',
    aws_access_key_id=os.environ['ACCESS_KEY_ID'],
    aws_secret_access_key=os.environ['SECRET_ACCESS_KEY'],
    config=boto_config
)

bucket_name = os.environ['BUCKET']

# Column order is set by your inventory configuration and listed in
# manifest.json (fileSchema). Adjust if your schema differs.
FIELDS = [
    'Bucket', 'Key', 'VersionId', 'IsLatest', 'IsDeleteMarker', 'Size',
    'LastModifiedDate', 'IsMultipartUploaded', 'EncryptionStatus',
    'StorageClass', 'ChecksumAlgorithm', 'LastAccessedDate', 'ETag',
]

threshold_days = 90
cutoff = datetime.now(timezone.utc) - timedelta(days=threshold_days)

candidates = []
with open('inventory.csv') as f:
    # Inventory CSVs are headerless; provide column names explicitly.
    reader = csv.DictReader(f, fieldnames=FIELDS)
    for row in reader:
        # Only consider current live objects.
        if row['IsLatest'] != 'true' or row['IsDeleteMarker'] == 'true':
            continue
        if not row['LastAccessedDate']:
            continue
        # The "Z" -> "+00:00" replacement makes fromisoformat() work on
        # Python versions before 3.11.
        last_access = datetime.fromisoformat(
            row['LastAccessedDate'].replace('Z', '+00:00')
        )
        if last_access < cutoff:
            candidates.append(row['Key'])

quarantine_date = datetime.now(timezone.utc).strftime('%Y-%m-%d')
quarantine_tag = {'Key': 'quarantine_set', 'Value': quarantine_date}

for key in candidates:
    # Fetch existing tags so you don't overwrite them.
    response = s3.get_object_tagging(Bucket=bucket_name, Key=key)
    existing_tags = response.get('TagSet', [])
    # Drop any prior quarantine_set tag, then append the current one.
    merged_tags = [t for t in existing_tags if t['Key'] != 'quarantine_set']
    merged_tags.append(quarantine_tag)
    s3.put_object_tagging(
        Bucket=bucket_name,
        Key=key,
        Tagging={'TagSet': merged_tags},
    )

with open('quarantine_state.json', 'w') as f:
    json.dump({'quarantine_date': quarantine_date, 'candidates': candidates}, f)

import os
import boto3
import csv
import json
from botocore.client import Config
from datetime import datetime, timezone

boto_config = Config(
    region_name='US-EAST-04A',
    s3={'addressing_style': 'virtual'}
)

s3 = boto3.client(
    's3',
    endpoint_url='https://cwobject.com',
    aws_access_key_id=os.environ['ACCESS_KEY_ID'],
    aws_secret_access_key=os.environ['SECRET_ACCESS_KEY'],
    config=boto_config
)

bucket_name = os.environ['BUCKET']

# Same FIELDS list used in Stage 1. It must match your inventory schema.
FIELDS = [
    'Bucket', 'Key', 'VersionId', 'IsLatest', 'IsDeleteMarker', 'Size',
    'LastModifiedDate', 'IsMultipartUploaded', 'EncryptionStatus',
    'StorageClass', 'ChecksumAlgorithm', 'LastAccessedDate', 'ETag',
]

s3.put_bucket_versioning(
    Bucket=bucket_name,
    VersioningConfiguration={'Status': 'Enabled'},
)

# Restore state persisted at the end of Stage 2.
with open('quarantine_state.json') as f:
    state = json.load(f)

quarantine_date = state['quarantine_date']  # the YYYY-MM-DD quarantine_set value
candidates = state['candidates']  # the key list from Stage 1

quarantine_start = datetime.strptime(quarantine_date, '%Y-%m-%d').replace(
    tzinfo=timezone.utc
)
quarantined_keys = set(candidates)

confirmed_candidates = []
with open('inventory_fresh.csv') as f:
    reader = csv.DictReader(f, fieldnames=FIELDS)
    for row in reader:
        if row['Key'] not in quarantined_keys:
            continue
        if row['IsLatest'] != 'true' or row['IsDeleteMarker'] == 'true':
            continue
        if not row['LastAccessedDate']:
            continue
        last_access = datetime.fromisoformat(
            row['LastAccessedDate'].replace('Z', '+00:00')
        )
        # If last_access advanced past quarantine_start, the object was
        # accessed during the buffer window, so skip it.
        if last_access < quarantine_start:
            confirmed_candidates.append(row['Key'])

# delete_objects accepts up to 1000 keys per call. It returns HTTP 200 even
# when individual keys fail, so inspect the Errors list in each response.
errors = []
for i in range(0, len(confirmed_candidates), 1000):
    batch = [{'Key': k} for k in confirmed_candidates[i:i + 1000]]
    resp = s3.delete_objects(Bucket=bucket_name, Delete={'Objects': batch})
    errors.extend(resp.get('Errors', []))

if errors:
    print(f'{len(errors)} delete(s) failed; first error: {errors[0]}')

with open('confirmed_candidates.json', 'w') as f:
    json.dump(confirmed_candidates, f)

# Restore a single mistakenly deleted key by removing its current delete marker.
key = 'path/to/object'

paginator = s3.get_paginator('list_object_versions')
for page in paginator.paginate(Bucket=bucket_name, Prefix=key):
    for marker in page.get('DeleteMarkers', []):
        if marker['Key'] == key and marker['IsLatest']:
            s3.delete_object(
                Bucket=bucket_name, Key=key, VersionId=marker['VersionId']
            )

# Also drop the recovered key from the persisted cleanup list so a later
# re-deletion can't make cleanup purge it again. Cleanup already skips keys
# whose current version is a live object, but removing it here makes the
# recovery permanent regardless of what happens to the key before cleanup runs.
with open('confirmed_candidates.json') as f:
    confirmed_candidates = json.load(f)

confirmed_candidates = [k for k in confirmed_candidates if k != key]

with open('confirmed_candidates.json', 'w') as f:
    json.dump(confirmed_candidates, f)

# Batch recovery: restore several mistakenly deleted keys at once.
keys_to_recover = {'path/to/object', 'another/object'}

paginator = s3.get_paginator('list_object_versions')
markers_to_remove = []
for page in paginator.paginate(Bucket=bucket_name):
    for marker in page.get('DeleteMarkers', []):
        if marker['Key'] in keys_to_recover and marker['IsLatest']:
            markers_to_remove.append(
                {'Key': marker['Key'], 'VersionId': marker['VersionId']}
            )

# delete_objects accepts up to 1000 versioned entries per call. Removing each
# current delete marker promotes that key's recovery copy back to current.
# A 200 response can still report per-entry failures, so inspect Errors.
errors = []
for i in range(0, len(markers_to_remove), 1000):
    resp = s3.delete_objects(
        Bucket=bucket_name, Delete={'Objects': markers_to_remove[i:i + 1000]}
    )
    errors.extend(resp.get('Errors', []))

failed_keys = {e['Key'] for e in errors}
if failed_keys:
    print(f'{len(failed_keys)} key(s) could not be recovered; first error: {errors[0]}')

# Drop only the successfully recovered keys from the persisted cleanup list, so
# a key whose marker removal failed stays in the list and is retried or cleaned up.
recovered = keys_to_recover - failed_keys
with open('confirmed_candidates.json') as f:
    confirmed_candidates = json.load(f)

confirmed_candidates = [k for k in confirmed_candidates if k not in recovered]

with open('confirmed_candidates.json', 'w') as f:
    json.dump(confirmed_candidates, f)

import os
import boto3
import json
from collections import defaultdict
from botocore.client import Config

boto_config = Config(
    region_name='US-EAST-04A',
    s3={'addressing_style': 'virtual'}
)

s3 = boto3.client(
    's3',
    endpoint_url='https://cwobject.com',
    aws_access_key_id=os.environ['ACCESS_KEY_ID'],
    aws_secret_access_key=os.environ['SECRET_ACCESS_KEY'],
    config=boto_config
)

bucket_name = os.environ['BUCKET']

# Reload the keys this workflow deleted (persisted at the end of Stage 3).
with open('confirmed_candidates.json') as f:
    confirmed_candidates = set(json.load(f))

paginator = s3.get_paginator('list_object_versions')
versions_by_key = defaultdict(list)  # key -> [{"Key", "VersionId"}, ...]
current_is_delete_marker = {}  # key -> True if the current version is a marker

# A full-bucket listing is simplest. If confirmed_candidates is small relative
# to the bucket, listing per key with Prefix=key is cheaper. Loop the keys and
# paginate with Prefix set to each one instead of scanning the whole bucket.
for page in paginator.paginate(Bucket=bucket_name):
    for v in page.get('Versions', []):
        if v['Key'] in confirmed_candidates:
            versions_by_key[v['Key']].append(
                {'Key': v['Key'], 'VersionId': v['VersionId']}
            )
            if v['IsLatest']:
                current_is_delete_marker[v['Key']] = False
    for m in page.get('DeleteMarkers', []):
        if m['Key'] in confirmed_candidates:
            versions_by_key[m['Key']].append(
                {'Key': m['Key'], 'VersionId': m['VersionId']}
            )
            if m['IsLatest']:
                current_is_delete_marker[m['Key']] = True

# Purge only keys whose current version is still a delete marker.
to_delete = []
skipped = []
for key in confirmed_candidates:
    if current_is_delete_marker.get(key) is True:
        to_delete.extend(versions_by_key[key])  # noncurrent versions + the marker
    else:
        skipped.append(key)  # restored or re-created since Stage 3, leave it alone

# delete_objects accepts up to 1000 versioned entries per call. It returns
# HTTP 200 even when individual versions fail. Inspect Errors and re-run
# cleanup to retry, since it only purges keys still in a deleted state.
errors = []
for i in range(0, len(to_delete), 1000):
    resp = s3.delete_objects(Bucket=bucket_name, Delete={'Objects': to_delete[i:i + 1000]})
    errors.extend(resp.get('Errors', []))

if errors:
    print(f'{len(errors)} version(s) failed to delete; first error: {errors[0]}')
if skipped:
    print(f'Skipped {len(skipped)} key(s) no longer in a deleted state.')

s3.put_bucket_lifecycle_configuration(
    Bucket=bucket_name,
    LifecycleConfiguration={
        'Rules': [
            {
                'ID': 'CleanupNoncurrentVersions',
                'Status': 'Enabled',
                'Filter': {},
                'NoncurrentVersionExpiration': {'NoncurrentDays': 30},
            },
            {
                'ID': 'CleanupDeleteMarkers',
                'Status': 'Enabled',
                'Filter': {},
                'Expiration': {'ExpiredObjectDeleteMarker': True},
            },
        ]
    }
)