71 lines
2.5 KiB
Python
71 lines
2.5 KiB
Python
import boto3
|
|
import os
|
|
import logging # Import logging
|
|
|
|
# Get a logger instance for this module
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def get_s3_client(aws_access_key_id, aws_secret_access_key, region_name):
|
|
"""
|
|
Initializes and returns an S3 client.
|
|
"""
|
|
try:
|
|
s3_client = boto3.client(
|
|
's3',
|
|
aws_access_key_id=aws_access_key_id,
|
|
aws_secret_access_key=aws_secret_access_key,
|
|
region_name=region_name
|
|
)
|
|
logger.debug("S3 client initialized successfully.")
|
|
return s3_client
|
|
except Exception as e:
|
|
logger.error(f"Failed to initialize S3 client: {e}")
|
|
return None
|
|
|
|
def download_image_from_s3(s3_client, bucket_name, object_key):
|
|
"""
|
|
Downloads an image from S3 and returns its binary data.
|
|
"""
|
|
try:
|
|
response = s3_client.get_object(Bucket=bucket_name, Key=object_key)
|
|
image_data = response['Body'].read()
|
|
logger.info(f"Downloaded '{object_key}' from S3 bucket '{bucket_name}'.")
|
|
return image_data
|
|
except s3_client.exceptions.NoSuchKey:
|
|
logger.warning(f"S3 object '{object_key}' not found in bucket '{bucket_name}'.")
|
|
return None
|
|
except Exception as e:
|
|
logger.error(f"Error downloading '{object_key}' from S3 bucket '{bucket_name}': {e}")
|
|
return None
|
|
|
|
def upload_object_to_s3(s3_client, bucket_name, object_key, data_bytes, content_type='application/octet-stream'):
|
|
"""
|
|
Uploads binary data to S3.
|
|
"""
|
|
try:
|
|
s3_client.put_object(
|
|
Bucket=bucket_name,
|
|
Key=object_key,
|
|
Body=data_bytes,
|
|
ContentType=content_type
|
|
)
|
|
logger.info(f"Uploaded object to s3://{bucket_name}/{object_key} with content type '{content_type}'.")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Error uploading '{object_key}' to S3 bucket '{bucket_name}': {e}")
|
|
return False
|
|
|
|
def generate_thumbnail_s3_key(original_s3_key, thumbnail_prefix):
|
|
"""
|
|
Generates a suitable S3 key for a thumbnail based on the original key.
|
|
E.g., original/path/image.jpg -> thumbnails/original/path/image.jpg
|
|
"""
|
|
thumbnail_name = os.path.basename(original_s3_key)
|
|
thumbnail_directory = os.path.dirname(original_s3_key)
|
|
|
|
if thumbnail_directory:
|
|
thumbnail_key = f"{thumbnail_prefix}{thumbnail_directory}/{thumbnail_name}"
|
|
else:
|
|
thumbnail_key = f"{thumbnail_prefix}{thumbnail_name}"
|
|
logger.debug(f"Generated thumbnail S3 key: '{thumbnail_key}' from original: '{original_s3_key}'")
|
|
return thumbnail_key |