Spaces:
Running
Running
| from google.cloud import storage | |
| from PIL import Image | |
| import os | |
| import uuid | |
| import tempfile | |
| from dotenv import load_dotenv | |
| load_dotenv() | |
| def get_credentials(): | |
| credentials_json_string = os.getenv("GOOGLE_APPLICATION_CREDENTIALS_JSON") | |
| # create a temp file with the credentials | |
| with tempfile.NamedTemporaryFile( | |
| mode="w+", delete=False, suffix=".json" | |
| ) as temp_file: | |
| temp_file.write(credentials_json_string) | |
| temp_file_path = temp_file.name | |
| return temp_file_path | |
| class GoogleCloudImageUploadService: | |
| BUCKET_NAME = "picchat-assets" | |
| MAX_DIMENSION = 1024 | |
| def __init__(self): | |
| # Using API key here as per your original code. Note that for production, | |
| # service account credentials are generally recommended. | |
| # get credentials from env | |
| credentials_json = get_credentials() | |
| self.storage_client = storage.Client.from_service_account_json(credentials_json) | |
| def upload_image_to_gcs(self, source_file_name): | |
| """ | |
| Uploads an image to the specified Google Cloud Storage bucket. | |
| Supports both JPEG and PNG formats. | |
| """ | |
| try: | |
| bucket = self.storage_client.bucket(self.BUCKET_NAME) | |
| blob_name = str(uuid.uuid4()) | |
| blob = bucket.blob(blob_name) | |
| # Open and optionally resize the image, then save to a temporary file. | |
| with Image.open(source_file_name) as image: | |
| # Determine the original format. If it's not JPEG or PNG, default to JPEG. | |
| original_format = ( | |
| image.format.upper() if image.format in ["JPEG", "PNG"] else "JPEG" | |
| ) | |
| # Resize if needed. | |
| if ( | |
| image.width > self.MAX_DIMENSION | |
| or image.height > self.MAX_DIMENSION | |
| ): | |
| image.thumbnail((self.MAX_DIMENSION, self.MAX_DIMENSION)) | |
| # Choose the file extension based on the image format. | |
| suffix = ".jpg" if original_format == "JPEG" else ".png" | |
| # Create a temporary file with the appropriate suffix. | |
| with tempfile.NamedTemporaryFile( | |
| delete=False, suffix=suffix | |
| ) as temp_file: | |
| temp_filename = temp_file.name | |
| image.save(temp_filename, format=original_format) | |
| try: | |
| # Set content type based on the image format. | |
| content_type = ( | |
| "image/jpeg" if original_format == "JPEG" else "image/png" | |
| ) | |
| blob.upload_from_filename(temp_filename, content_type=content_type) | |
| blob.make_public() | |
| finally: | |
| # Remove the temporary file. | |
| os.remove(temp_filename) | |
| print( | |
| f"File {source_file_name} uploaded to {blob_name} in bucket {self.BUCKET_NAME}." | |
| ) | |
| return blob.public_url | |
| except Exception as e: | |
| print(f"An error occurred: {e}") | |
| return None | |
| if __name__ == "__main__": | |
| image = "./assets/lakeview.jpg" # Replace with your JPEG or PNG image path. | |
| upload_service = GoogleCloudImageUploadService() | |
| url = upload_service.upload_image_to_gcs(image) | |
| print(url) | |