"""
GCS (Google Cloud Storage) operations for forecast pictures.
Handles upload, download, and deletion of forecast pictures in GCS buckets.
"""
import os
import base64
import logging
from typing import Optional
from google.cloud import storage
from datetime import datetime
# GCS bucket name from environment
GCS_BUCKET_NAME = os.getenv("GCS_BUCKET_NAME", "")
def upload_picture_to_gcs(
city: str,
picture_data_base64: str,
forecast_at: str,
file_extension: str = "png"
) -> Optional[str]:
"""
Upload forecast picture to GCS bucket.
Args:
city: City name (e.g., 'chicago')
picture_data_base64: Base64-encoded picture bytes
forecast_at: Timestamp string (e.g., '2026-01-09_120000')
file_extension: File extension (e.g., 'png', 'jpg')
Returns:
GCS URL (gs://bucket/city/timestamp.ext) or None if upload failed
"""
if not GCS_BUCKET_NAME:
logging.error("GCS_BUCKET_NAME environment variable not set")
return None
try:
# Decode base64 picture data
picture_bytes = base64.b64decode(picture_data_base64)
# Generate GCS path: city/timestamp.ext
# Replace : with - for filesystem compatibility
safe_timestamp = forecast_at.replace(':', '-')
blob_name = f"{city.lower()}/{safe_timestamp}.{file_extension}"
# Initialize GCS client
client = storage.Client()
bucket = client.bucket(GCS_BUCKET_NAME)
blob = bucket.blob(blob_name)
# Set content type based on extension
content_types = {
'png': 'image/png',
'jpg': 'image/jpeg',
'jpeg': 'image/jpeg',
'webp': 'image/webp'
}
content_type = content_types.get(file_extension.lower(), 'application/octet-stream')
# Upload to GCS
blob.upload_from_string(picture_bytes, content_type=content_type)
# Return public HTTPS URL (for browser/website access)
https_url = f"https://storage.googleapis.com/{GCS_BUCKET_NAME}/{blob_name}"
logging.info(f"Picture uploaded successfully: {https_url}")
return https_url
except Exception as e:
logging.error(f"Failed to upload picture to GCS: {e}")
return None
def download_picture_from_gcs(gcs_url: str) -> Optional[str]:
"""
Download picture from GCS and return as base64.
Args:
gcs_url: GCS URL (gs://bucket/path/file.png)
Returns:
Base64-encoded picture bytes or None if download failed
"""
try:
# Parse GCS URL: gs://bucket/path/file.png
if not gcs_url.startswith("gs://"):
logging.error(f"Invalid GCS URL format: {gcs_url}")
return None
parts = gcs_url[5:].split("/", 1)
bucket_name = parts[0]
blob_name = parts[1]
# Download from GCS
client = storage.Client()
bucket = client.bucket(bucket_name)
blob = bucket.blob(blob_name)
picture_bytes = blob.download_as_bytes()
# Return as base64
return base64.b64encode(picture_bytes).decode('utf-8')
except Exception as e:
logging.error(f"Failed to download picture from GCS: {e}")
return None
