This Lambda implementation scans selected gid-* prefixes, validates required files, builds index.7z, and uploads it back to the same S3 folder. It is designed for manual execution with explicit input prefixes and clear skip conditions.
Prerequisites
- Lambda Layer containing
/opt/bin/7z - IAM permissions for S3 list/get/put
- Sufficient Lambda memory and timeout for archive size
Lambda Handler (Python)
import os
import boto3
import subprocess
import shutil
import logging
import re
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger()
s3_client = boto3.client('s3')
def lambda_handler(event, context):
# Get bucket name from event, default to "my-game"
bucket = event.get('bucket', 'my-game')
# Get list of prefixes, default to [''] if not provided (processes entire bucket)
prefixes = event.get('prefixes', [''])
created_archives = []
# Loop through each prefix
for prefix in prefixes:
logger.info(f"Processing prefix: {prefix}")
# List all objects ending with "index.zip" within the prefix
continuation_token = None
index_zip_keys = []
while True:
list_kwargs = {'Bucket': bucket, 'Prefix': prefix}
if continuation_token:
list_kwargs['ContinuationToken'] = continuation_token
response = s3_client.list_objects_v2(**list_kwargs)
if 'Contents' not in response:
break
for obj in response['Contents']:
key = obj['Key']
if key.endswith("index.zip"):
index_zip_keys.append(key)
if response.get('IsTruncated'):
continuation_token = response.get('NextContinuationToken')
else:
break
logger.info(f"Found {len(index_zip_keys)} index.zip files within prefix '{prefix}'.")
# Process each folder containing an index.zip
for zip_key in index_zip_keys:
folder_prefix = zip_key.rsplit('/', 1)[0] + '/'
logger.info(f"Checking folder: {folder_prefix}")
# Skip if not a gid-* folder
if not folder_prefix.startswith('gid-'):
logger.info(f"Skipping {folder_prefix}: not a gid-* folder")
continue
# List objects in the folder to check conditions
response = s3_client.list_objects_v2(Bucket=bucket, Prefix=folder_prefix, Delimiter='/')
if 'Contents' not in response:
logger.info(f"No objects in {folder_prefix}")
continue
direct_keys = [obj['Key'] for obj in response.get('Contents', [])]
subfolders = [cp['Prefix'] for cp in response.get('CommonPrefixes', [])]
# Define expected keys
index_zip_key = folder_prefix + "index.zip"
index_html_key = folder_prefix + "index.html"
index_7z_key = folder_prefix + "index.7z"
# Check conditions
has_index_zip = index_zip_key in direct_keys
has_index_html = index_html_key in direct_keys
has_index_7z = index_7z_key in direct_keys
# Proceed only if index.zip and index.html exist, and index.7z does not
if not has_index_zip or not has_index_html or has_index_7z:
logger.info(f"Skipping {folder_prefix}: conditions not met (zip={has_index_zip}, html={has_index_html}, 7z={has_index_7z})")
continue
# List all objects to archive, excluding numbered subfolders
all_objects_response = s3_client.list_objects_v2(Bucket=bucket, Prefix=folder_prefix)
objects_to_archive = [
obj['Key'] for obj in all_objects_response.get('Contents', [])
if not obj['Key'].endswith("index.zip")
and not obj['Key'].endswith("index.7z")
and not any(re.match(r'^\d+$', part) for part in obj['Key'][len(folder_prefix):].split('/')) # Exclude numbered segments
]
if not objects_to_archive:
logger.info(f"No objects to archive in {folder_prefix}")
continue
# Create a working directory
safe_prefix = folder_prefix.replace('/', '_')
work_dir = f"/tmp/game_folder_{context.aws_request_id}_{safe_prefix}"
if os.path.exists(work_dir):
shutil.rmtree(work_dir)
os.makedirs(work_dir)
# Download objects
for obj_key in objects_to_archive:
relative_path = obj_key[len(folder_prefix):]
local_file_path = os.path.join(work_dir, relative_path)
local_dir = os.path.dirname(local_file_path)
if not os.path.exists(local_dir):
os.makedirs(local_dir)
logger.info(f"Downloading s3://{bucket}/{obj_key} to {local_file_path}")
try:
s3_client.download_file(bucket, obj_key, local_file_path)
except Exception as e:
logger.error(f"Error downloading {obj_key}: {e}")
continue
# Create a 7z archive
local_archive = f"/tmp/index_{context.aws_request_id}_{safe_prefix}.7z"
try:
subprocess.check_call(['/opt/bin/7z', 'a', local_archive, '.'], cwd=work_dir)
except subprocess.CalledProcessError as e:
logger.error(f"Error creating archive for {folder_prefix}: {e}")
shutil.rmtree(work_dir)
continue
# Upload the archive to S3
archive_key = folder_prefix + "index.7z"
logger.info(f"Uploading archive to s3://{bucket}/{archive_key}")
try:
s3_client.upload_file(local_archive, bucket, archive_key)
created_archives.append(archive_key)
except Exception as e:
logger.error(f"Error uploading archive {archive_key}: {e}")
# Clean up
shutil.rmtree(work_dir)
if os.path.exists(local_archive):
os.remove(local_archive)
return {"statusCode": 200, "body": f"Created archives: {created_archives}"}Manual Test Input Example
{
"bucket": "my-game",
"prefixes": [
"gid-1/",
"gid-2/",
"gid-3/"
]
}Operational Notes
- The function skips folders where
index.7zalready exists. - It requires both
index.zipandindex.htmlbefore archive creation. - Numbered subfolder segments are excluded by regex rule.
Official Documentation
- AWS Lambda Developer Guide: https://docs.aws.amazon.com/lambda/latest/dg/welcome.html
- Amazon S3 API (ListObjectsV2): https://docs.aws.amazon.com/AmazonS3/latest/API/API_ListObjectsV2.html
- Boto3 S3 Client Reference: https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3.html
