import boto3 import logging import argparse from botocore.exceptions import ClientError from datetime import datetime import time # Configure logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) def parse_arguments(): """Parse command line arguments.""" parser = argparse.ArgumentParser( description='Create an S3 Batch Operations job to copy objects between S3 buckets. ' 'This script uploads a manifest file to S3 and creates a batch operation job to copy ' 'the objects specified in the manifest to a destination bucket with the specified prefix.' ) parser.add_argument( '--manifest-bucket', required=True, help='Name of the S3 bucket where the manifest CSV file will be uploaded' ) parser.add_argument( '--manifest-key', required=True, help='S3 key (path) where the manifest CSV file will be uploaded (e.g., "path/to/manifest.csv")' ) parser.add_argument( '--destination-bucket', required=True, help='Name of the S3 bucket where objects will be copied to' ) parser.add_argument( '--destination-prefix', required=True, help='Prefix to prepend to copied objects in the destination bucket (e.g., "backup/")' ) parser.add_argument( '--report-bucket', required=True, help='Name of the S3 bucket where the batch operations completion report will be stored' ) parser.add_argument( '--report-prefix', required=True, help='Prefix for the batch operations completion report (e.g., "reports/")' ) parser.add_argument( '--iam-role', required=True, help='ARN of the IAM role that S3 Batch Operations will assume to perform the copy operation' ) parser.add_argument( '--csv-file', required=True, help='Path to the local CSV manifest file containing the list of objects to copy. ' 'The CSV must have "Bucket" and "Key" columns' ) parser.add_argument( '--poll-results', action='store_true', help='Poll for batch job results and print status updates' ) parser.add_argument( '--download-result', action='store_true', help='Download the batch operations completion report after polling' ) return parser.parse_args() def create_batch_operations_job(args): """Create an S3 batch operations job for copying objects.""" try: s3control_client = boto3.client('s3control') account_id = boto3.client('sts').get_caller_identity()['Account'] # Generate a unique job description timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') job_description = f"Copy job created at {timestamp}" response = s3control_client.create_job( AccountId=account_id, ConfirmationRequired=False, Operation={ 'S3PutObjectCopy': { 'TargetResource': f'arn:aws:s3:::{args.destination_bucket}', 'CannedAccessControlList': 'private', 'MetadataDirective': 'COPY', 'StorageClass': 'STANDARD', 'TargetKeyPrefix': args.destination_prefix } }, Manifest={ 'Spec': { 'Format': 'S3BatchOperations_CSV_20180820', 'Fields': ['Bucket', 'Key', 'VersionId'] }, 'Location': { 'ObjectArn': f'arn:aws:s3:::{args.manifest_bucket}/{args.manifest_key}', 'ETag': boto3.client('s3').head_object( Bucket=args.manifest_bucket, Key=args.manifest_key )['ETag'] } }, Report={ 'Bucket': f'arn:aws:s3:::{args.report_bucket}', 'Prefix': args.report_prefix, 'Format': 'Report_CSV_20180820', 'Enabled': True, 'ReportScope': 'AllTasks' }, Priority=10, RoleArn=args.iam_role, Description=job_description ) job_id = response['JobId'] logger.info(f"Successfully created batch operations job with ID: {job_id}") logger.info(f"Job response: {response}") return job_id except ClientError as e: logger.error(f"Failed to create batch operations job: {str(e)}") raise except Exception as e: logger.error(f"Unexpected error occurred: {str(e)}") raise def upload_manifest_to_s3(csv_file_path, bucket, key): """Upload the manifest CSV file to S3.""" try: s3_client = boto3.client('s3') s3_client.upload_file(csv_file_path, bucket, key) logger.info(f"Successfully uploaded manifest file to s3://{bucket}/{key}") except ClientError as e: logger.error(f"Failed to upload manifest file: {str(e)}") raise except FileNotFoundError: logger.error(f"CSV file not found: {csv_file_path}") raise def poll_job_status(s3control_client, account_id, job_id): """Poll the batch operations job status and print updates.""" logger.info("Starting to poll job status...") while True: response = s3control_client.describe_job( AccountId=account_id, JobId=job_id ) status = response['Job']['Status'] progress = response.get('Job', {}).get('Progress') logger.info(f"Job Status: {status}") if progress: logger.info(f"Progress: Succeeded={progress.get('TimedOut', 0)}, " f"Failed={progress.get('Failed', 0)}, " f"Completed={progress.get('NumberOfTasksCompleted', 0)}") if status in ['Complete', 'Failed', 'Cancelled']: break time.sleep(30) # Poll every 30 seconds def download_results(s3_client, report_bucket, report_prefix, job_id): """Download the batch operations completion report.""" try: report_key = f"{report_prefix}{job_id}.csv" local_file_path = f"{job_id}_report.csv" s3_client.download_file(report_bucket, report_key, local_file_path) logger.info(f"Successfully downloaded report to {local_file_path}") except ClientError as e: logger.error(f"Failed to download report: {str(e)}") raise def main(): """Main function to run the script.""" try: args = parse_arguments() # Upload manifest file to S3 upload_manifest_to_s3(args.csv_file, args.manifest_bucket, args.manifest_key) # Validate input parameters if not args.destination_prefix.endswith('/'): args.destination_prefix += '/' logger.info("Added trailing slash to destination prefix") job_id = create_batch_operations_job(args) logger.info("Job creation completed successfully") if args.poll_results: s3control_client = boto3.client('s3control') account_id = boto3.client('sts').get_caller_identity()['Account'] poll_job_status(s3control_client, account_id, job_id) if args.download_result: s3_client = boto3.client('s3') download_results(s3_client, args.report_bucket, args.report_prefix, job_id) except Exception as e: logger.error(f"Script execution failed: {str(e)}") raise if __name__ == '__main__': main()