# Create buckets with different storage classes
gsutil mb -c STANDARD -l US gs://my-hot-data
gsutil mb -c NEARLINE -l US gs://my-backup-data
gsutil mb -c COLDLINE -l US gs://my-archive-data
gsutil mb -c ARCHIVE -l US gs://my-long-term-archive

# Change storage class of existing bucket
gsutil defstorageclass set NEARLINE gs://my-bucket

# Set storage class for specific objects
gsutil -m setmeta -h "Cache-Control:private, max-age=0" \
    -h "Content-Type:application/pdf" \
    -h "x-goog-storage-class:NEARLINE" \
    gs://my-bucket/documents/*.pdf

# storage_optimizer.py
from google.cloud import storage
from datetime import datetime, timedelta
import json

class StorageOptimizer:
    """Optimize storage costs based on access patterns."""

    def __init__(self, project_id):
        self.client = storage.Client(project=project_id)
        self.storage_classes = {
            'STANDARD': {'min_duration': 0, 'retrieval_cost': 0},
            'NEARLINE': {'min_duration': 30, 'retrieval_cost': 0.01},
            'COLDLINE': {'min_duration': 90, 'retrieval_cost': 0.02},
            'ARCHIVE': {'min_duration': 365, 'retrieval_cost': 0.05}
        }

    def analyze_access_patterns(self, bucket_name):
        """Analyze object access patterns to recommend storage classes."""
        bucket = self.client.bucket(bucket_name)
        recommendations = []

        for blob in bucket.list_blobs():
            access_info = self.get_access_info(blob)
            recommended_class = self.recommend_storage_class(access_info)

            if blob.storage_class != recommended_class:
                recommendations.append({
                    'object': blob.name,
                    'current_class': blob.storage_class,
                    'recommended_class': recommended_class,
                    'last_access': access_info['last_access'],
                    'access_frequency': access_info['frequency']
                })

        return recommendations

    def get_access_info(self, blob):
        """Get access information for a blob."""
        # In production, this would query access logs
        last_modified = blob.updated
        days_since_modified = (datetime.now(last_modified.tzinfo) - last_modified).days

        return {
            'last_access': last_modified,
            'days_inactive': days_since_modified,
            'frequency': self.estimate_access_frequency(days_since_modified)
        }

    def recommend_storage_class(self, access_info):
        """Recommend storage class based on access patterns."""
        days_inactive = access_info['days_inactive']

        if days_inactive < 30:
            return 'STANDARD'
        elif days_inactive < 90:
            return 'NEARLINE'
        elif days_inactive < 365:
            return 'COLDLINE'
        else:
            return 'ARCHIVE'

    def apply_recommendations(self, bucket_name, recommendations):
        """Apply storage class recommendations."""
        bucket = self.client.bucket(bucket_name)

        for rec in recommendations:
            blob = bucket.blob(rec['object'])
            blob.update_storage_class(rec['recommended_class'])
            print(f"Updated {rec['object']} from {rec['current_class']} to {rec['recommended_class']}")

// lifecycle-policy.json
{
  "lifecycle": {
    "rule": [
      {
        "action": {
          "type": "SetStorageClass",
          "storageClass": "NEARLINE"
        },
        "condition": {
          "age": 30,
          "matchesStorageClass": ["STANDARD"]
        }
      },
      {
        "action": {
          "type": "SetStorageClass",
          "storageClass": "COLDLINE"
        },
        "condition": {
          "age": 90,
          "matchesStorageClass": ["NEARLINE"]
        }
      },
      {
        "action": {
          "type": "Delete"
        },
        "condition": {
          "age": 365,
          "matchesPrefix": ["temp/", "logs/"]
        }
      },
      {
        "action": {
          "type": "Delete"
        },
        "condition": {
          "numNewerVersions": 3,
          "isLive": false
        }
      },
      {
        "action": {
          "type": "AbortIncompleteMultipartUpload"
        },
        "condition": {
          "age": 7
        }
      }
    ]
  }
}

# Apply lifecycle configuration
gsutil lifecycle set lifecycle-policy.json gs://my-bucket

# View current lifecycle configuration
gsutil lifecycle get gs://my-bucket

# advanced_storage.py
from google.cloud import storage
from google.cloud.storage import Blob
import hashlib
import base64
from concurrent.futures import ThreadPoolExecutor
import os

class AdvancedStorage:
    """Advanced GCS operations with optimization."""

    def __init__(self, project_id):
        self.client = storage.Client(project=project_id)
        self.executor = ThreadPoolExecutor(max_workers=10)

    def parallel_upload(self, bucket_name, file_paths, prefix=""):
        """Upload multiple files in parallel."""
        bucket = self.client.bucket(bucket_name)

        def upload_file(file_path):
            blob_name = os.path.join(prefix, os.path.basename(file_path))
            blob = bucket.blob(blob_name)

            # Calculate checksum
            with open(file_path, 'rb') as f:
                content = f.read()
                md5_hash = hashlib.md5(content).digest()
                blob.md5_hash = base64.b64encode(md5_hash).decode('utf-8')

            # Upload with retry
            blob.upload_from_filename(
                file_path,
                checksum="md5",
                if_generation_match=0  # Only upload if doesn't exist
            )

            return blob_name

        # Upload files in parallel
        futures = [self.executor.submit(upload_file, fp) for fp in file_paths]
        results = [f.result() for f in futures]

        return results

    def resumable_upload(self, bucket_name, file_path, blob_name):
        """Perform resumable upload for large files."""
        bucket = self.client.bucket(bucket_name)
        blob = bucket.blob(blob_name)

        # Set chunk size for resumable upload (must be multiple of 256KB)
        blob.chunk_size = 5 * 1024 * 1024  # 5MB chunks

        # Upload with progress tracking
        def progress_callback(bytes_uploaded, total_bytes):
            percent = (bytes_uploaded / total_bytes) * 100
            print(f"Upload progress: {percent:.2f}%")

        blob.upload_from_filename(
            file_path,
            resumable=True,
            progress_callback=progress_callback
        )

        return blob

    def composite_upload(self, bucket_name, source_files, destination_blob):
        """Upload large file using composite objects."""
        bucket = self.client.bucket(bucket_name)

        # Upload parts in parallel
        part_blobs = []
        for i, file_path in enumerate(source_files):
            part_name = f"{destination_blob}.part{i}"
            part_blob = bucket.blob(part_name)
            part_blob.upload_from_filename(file_path)
            part_blobs.append(part_blob)

        # Compose parts into final object
        destination = bucket.blob(destination_blob)
        destination.compose(part_blobs)

        # Delete part objects
        for part in part_blobs:
            part.delete()

        return destination

    def signed_url_with_conditions(self, bucket_name, blob_name, expiration_hours=1):
        """Generate signed URL with upload conditions."""
        bucket = self.client.bucket(bucket_name)
        blob = bucket.blob(blob_name)

        # Define upload conditions
        conditions = [
            {"content-length-range": [0, 10485760]},  # Max 10MB
            {"x-goog-meta-uploaded-by": "${uploader}"},
            ["starts-with", "$Content-Type", "image/"]
        ]

        # Generate signed URL for upload
        url = blob.generate_signed_url(
            version="v4",
            expiration=timedelta(hours=expiration_hours),
            method="PUT",
            conditions=conditions
        )

        return url

# security_manager.py
from google.cloud import storage
from google.cloud.storage import Bucket
import json

class SecurityManager:
    """Manage GCS security and access controls."""

    def __init__(self, project_id):
        self.client = storage.Client(project=project_id)

    def enable_uniform_access(self, bucket_name):
        """Enable uniform bucket-level access."""
        bucket = self.client.bucket(bucket_name)
        bucket.iam_configuration.uniform_bucket_level_access_enabled = True
        bucket.patch()

        print(f"Uniform bucket-level access enabled for {bucket_name}")

    def set_bucket_iam_policy(self, bucket_name, member, role):
        """Set IAM policy for bucket."""
        bucket = self.client.bucket(bucket_name)
        policy = bucket.get_iam_policy(requested_policy_version=3)

        # Add member to role
        policy.bindings.append({
            "role": role,
            "members": [member],
            "condition": {
                "title": "expires_after_2024",
                "description": "Expires at end of 2024",
                "expression": 'request.time < timestamp("2024-12-31T23:59:59.999Z")'
            }
        })

        bucket.set_iam_policy(policy)
        print(f"Added {member} with role {role} to {bucket_name}")

    def create_hmac_key(self, service_account_email):
        """Create HMAC key for service account."""
        hmac_key, secret = self.client.create_hmac_key(
            service_account_email=service_account_email
        )

        print(f"HMAC Key ID: {hmac_key.id}")
        print(f"HMAC Secret: {secret}")
        print(f"Access ID: {hmac_key.access_id}")

        return hmac_key, secret

    def configure_cors(self, bucket_name, origins, methods=['GET', 'POST']):
        """Configure CORS for bucket."""
        bucket = self.client.bucket(bucket_name)
        bucket.cors = [{
            "origin": origins,
            "method": methods,
            "responseHeader": ["Content-Type", "x-goog-meta-*"],
            "maxAgeSeconds": 3600
        }]
        bucket.patch()

        print(f"CORS configured for {bucket_name}")

# encryption_manager.py
from google.cloud import storage
from google.cloud import kms
import base64

class EncryptionManager:
    """Manage encryption for GCS objects."""

    def __init__(self, project_id):
        self.storage_client = storage.Client(project=project_id)
        self.kms_client = kms.KeyManagementServiceClient()
        self.project_id = project_id

    def create_cmek_bucket(self, bucket_name, kms_key_name):
        """Create bucket with CMEK encryption."""
        bucket = self.storage_client.bucket(bucket_name)
        bucket.default_kms_key_name = kms_key_name
        bucket.create(location="US")

        print(f"Created bucket {bucket_name} with CMEK: {kms_key_name}")
        return bucket

    def upload_with_customer_encryption(self, bucket_name, source_file, blob_name, encryption_key):
        """Upload file with customer-supplied encryption key."""
        bucket = self.storage_client.bucket(bucket_name)
        blob = bucket.blob(blob_name, encryption_key=encryption_key)

        blob.upload_from_filename(source_file)
        print(f"Uploaded {blob_name} with customer encryption")

        return blob

    def rotate_encryption_key(self, bucket_name, blob_name, old_key, new_key):
        """Rotate customer-supplied encryption key."""
        bucket = self.storage_client.bucket(bucket_name)

        # Download with old key
        blob = bucket.blob(blob_name, encryption_key=old_key)
        content = blob.download_as_bytes()

        # Re-upload with new key
        new_blob = bucket.blob(blob_name, encryption_key=new_key)
        new_blob.upload_from_string(content)

        print(f"Rotated encryption key for {blob_name}")

# performance_optimizer.py
from google.cloud import storage
from concurrent.futures import ThreadPoolExecutor, as_completed
import multiprocessing
import time

class PerformanceOptimizer:
    """Optimize GCS operations for performance."""

    def __init__(self, project_id):
        self.client = storage.Client(project=project_id)
        self.cpu_count = multiprocessing.cpu_count()

    def parallel_download(self, bucket_name, blob_names, local_dir):
        """Download multiple files in parallel."""
        bucket = self.client.bucket(bucket_name)

        def download_blob(blob_name):
            blob = bucket.blob(blob_name)
            local_path = os.path.join(local_dir, blob_name)
            os.makedirs(os.path.dirname(local_path), exist_ok=True)

            start_time = time.time()
            blob.download_to_filename(local_path)
            duration = time.time() - start_time

            return {
                'blob': blob_name,
                'size': blob.size,
                'duration': duration,
                'speed': blob.size / duration / 1024 / 1024  # MB/s
            }

        with ThreadPoolExecutor(max_workers=self.cpu_count * 2) as executor:
            futures = {executor.submit(download_blob, bn): bn for bn in blob_names}
            results = []

            for future in as_completed(futures):
                result = future.result()
                results.append(result)
                print(f"Downloaded {result['blob']} at {result['speed']:.2f} MB/s")

        return results

    def stream_large_file(self, bucket_name, blob_name, chunk_size=1024*1024):
        """Stream download large file in chunks."""
        bucket = self.client.bucket(bucket_name)
        blob = bucket.blob(blob_name)

        def stream_generator():
            start = 0
            while start < blob.size:
                end = min(start + chunk_size - 1, blob.size - 1)
                chunk = blob.download_as_bytes(start=start, end=end)
                yield chunk
                start = end + 1

        return stream_generator()

    def optimized_list(self, bucket_name, prefix="", delimiter="/"):
        """List objects with optimized pagination."""
        bucket = self.client.bucket(bucket_name)

        # Use fields parameter to reduce response size
        iterator = bucket.list_blobs(
            prefix=prefix,
            delimiter=delimiter,
            max_results=1000,
            fields="items(name,size,updated),prefixes,nextPageToken"
        )

        objects = []
        prefixes = []

        for page in iterator.pages:
            objects.extend(page)
            if delimiter:
                prefixes.extend(page.prefixes)

        return objects, prefixes

# regional_optimizer.py
from google.cloud import storage
import requests

class RegionalOptimizer:
    """Optimize GCS operations based on regions."""

    def __init__(self, project_id):
        self.client = storage.Client(project=project_id)
        self.region_endpoints = {
            'us-central1': 'https://storage.googleapis.com',
            'europe-west1': 'https://storage-europe-west1.googleapis.com',
            'asia-northeast1': 'https://storage-asia-northeast1.googleapis.com'
        }

    def create_multi_region_buckets(self, base_name, data_locations):
        """Create buckets in multiple regions for global distribution."""
        buckets = {}

        for location in data_locations:
            bucket_name = f"{base_name}-{location.lower()}"
            bucket = self.client.bucket(bucket_name)
            bucket.location = location
            bucket.storage_class = "STANDARD"

            # Enable CDN for public content
            bucket.iam_configuration.public_access_prevention = 'inherited'

            bucket.create()
            buckets[location] = bucket

            print(f"Created bucket {bucket_name} in {location}")

        return buckets

    def setup_cross_region_replication(self, source_bucket, dest_bucket):
        """Setup cross-region replication using Cloud Storage Transfer."""
        # This would typically use Cloud Storage Transfer Service API
        transfer_config = {
            "description": f"Replicate {source_bucket} to {dest_bucket}",
            "status": "ENABLED",
            "projectId": self.client.project,
            "schedule": {
                "scheduleStartDate": {"day": 1, "month": 1, "year": 2024},
                "startTimeOfDay": {"hours": 0, "minutes": 0, "seconds": 0}
            },
            "transferSpec": {
                "gcsDataSource": {"bucketName": source_bucket},
                "gcsDataSink": {"bucketName": dest_bucket},
                "transferOptions": {
                    "overwriteObjectsAlreadyExistingInSink": True,
                    "deleteObjectsFromSourceAfterTransfer": False
                }
            }
        }

        print(f"Replication configured from {source_bucket} to {dest_bucket}")
        return transfer_config

# event_processor.py
from google.cloud import storage
from google.cloud import pubsub_v1
import json

class EventProcessor:
    """Process GCS events with Pub/Sub."""

    def __init__(self, project_id):
        self.storage_client = storage.Client(project=project_id)
        self.publisher = pubsub_v1.PublisherClient()
        self.project_id = project_id

    def setup_bucket_notification(self, bucket_name, topic_name, event_types=None):
        """Setup Pub/Sub notifications for bucket."""
        bucket = self.storage_client.bucket(bucket_name)
        topic_path = self.publisher.topic_path(self.project_id, topic_name)

        if event_types is None:
            event_types = [
                'OBJECT_FINALIZE',
                'OBJECT_DELETE',
                'OBJECT_METADATA_UPDATE'
            ]

        notification = bucket.notification(
            topic_name=topic_path,
            event_types=event_types,
            custom_attributes={
                'bucket': bucket_name,
                'project': self.project_id
            }
        )

        notification.create()
        print(f"Created notification for {bucket_name} -> {topic_name}")

        return notification

    def process_storage_event(self, event_data):
        """Process storage event from Pub/Sub."""
        # Parse event
        event = json.loads(event_data)

        bucket_name = event['bucket']
        object_name = event['name']
        event_type = event['eventType']

        print(f"Processing {event_type} for {bucket_name}/{object_name}")

        # Route based on event type
        if event_type == 'OBJECT_FINALIZE':
            self.handle_new_object(bucket_name, object_name, event)
        elif event_type == 'OBJECT_DELETE':
            self.handle_deleted_object(bucket_name, object_name, event)
        elif event_type == 'OBJECT_METADATA_UPDATE':
            self.handle_metadata_update(bucket_name, object_name, event)

    def handle_new_object(self, bucket_name, object_name, event):
        """Handle new object creation."""
        bucket = self.storage_client.bucket(bucket_name)
        blob = bucket.blob(object_name)

        # Example: Process image uploads
        if blob.content_type and blob.content_type.startswith('image/'):
            print(f"New image uploaded: {object_name}")
            # Trigger image processing workflow

        # Example: Process data files
        elif object_name.endswith('.csv'):
            print(f"New CSV file: {object_name}")
            # Trigger data processing pipeline

# transfer_manager.py
from google.cloud import storage_transfer
from datetime import datetime, timedelta
import pytz

class TransferManager:
    """Manage data transfers to/from GCS."""

    def __init__(self, project_id):
        self.client = storage_transfer.StorageTransferServiceClient()
        self.project_id = project_id

    def create_s3_to_gcs_transfer(self, aws_access_key, aws_secret_key, 
                                  s3_bucket, gcs_bucket, prefix=""):
        """Create transfer job from S3 to GCS."""
        transfer_job = {
            "description": f"Transfer from s3://{s3_bucket} to gs://{gcs_bucket}",
            "status": "ENABLED",
            "projectId": self.project_id,
            "schedule": {
                "scheduleStartDate": {
                    "day": datetime.now().day,
                    "month": datetime.now().month,
                    "year": datetime.now().year
                },
                "startTimeOfDay": {"hours": 0, "minutes": 0, "seconds": 0},
                "repeatInterval": "86400s"  # Daily
            },
            "transferSpec": {
                "awsS3DataSource": {
                    "bucketName": s3_bucket,
                    "awsAccessKey": {
                        "accessKeyId": aws_access_key,
                        "secretAccessKey": aws_secret_key
                    },
                    "path": prefix
                },
                "gcsDataSink": {
                    "bucketName": gcs_bucket,
                    "path": prefix
                },
                "objectConditions": {
                    "minTimeElapsedSinceLastModification": "300s",
                    "includePrefixes": [prefix] if prefix else []
                },
                "transferOptions": {
                    "overwriteObjectsAlreadyExistingInSink": True,
                    "deleteObjectsFromSourceAfterTransfer": False,
                    "deleteObjectsUniqueInSink": False
                }
            }
        }

        result = self.client.create_transfer_job(
            request={"transfer_job": transfer_job}
        )

        print(f"Created transfer job: {result.name}")
        return result

    def create_url_list_transfer(self, url_list_file, gcs_bucket):
        """Create transfer job from URL list."""
        # Upload URL list to GCS first
        storage_client = storage.Client(project=self.project_id)
        bucket = storage_client.bucket(gcs_bucket)
        blob = bucket.blob("transfer-lists/urls.txt")
        blob.upload_from_filename(url_list_file)

        transfer_job = {
            "description": "Transfer from URL list",
            "status": "ENABLED",
            "projectId": self.project_id,
            "transferSpec": {
                "httpDataSource": {
                    "listUrl": f"gs://{gcs_bucket}/transfer-lists/urls.txt"
                },
                "gcsDataSink": {
                    "bucketName": gcs_bucket,
                    "path": "downloads/"
                }
            }
        }

        result = self.client.create_transfer_job(
            request={"transfer_job": transfer_job}
        )

        return result

# storage_analytics.py
from google.cloud import storage
from google.cloud import bigquery
from google.cloud import monitoring_v3
import pandas as pd
from datetime import datetime, timedelta

class StorageAnalytics:
    """Analyze GCS usage and performance."""

    def __init__(self, project_id):
        self.storage_client = storage.Client(project=project_id)
        self.bq_client = bigquery.Client(project=project_id)
        self.monitoring_client = monitoring_v3.MetricServiceClient()
        self.project_id = project_id

    def analyze_bucket_usage(self, bucket_name):
        """Analyze storage usage for a bucket."""
        bucket = self.storage_client.bucket(bucket_name)

        stats = {
            'total_objects': 0,
            'total_size': 0,
            'storage_classes': {},
            'content_types': {},
            'age_distribution': {
                '0-30_days': 0,
                '31-90_days': 0,
                '91-365_days': 0,
                'over_365_days': 0
            }
        }

        now = datetime.now(pytz.UTC)

        for blob in bucket.list_blobs():
            stats['total_objects'] += 1
            stats['total_size'] += blob.size or 0

            # Storage class distribution
            storage_class = blob.storage_class or 'STANDARD'
            stats['storage_classes'][storage_class] = \
                stats['storage_classes'].get(storage_class, 0) + 1

            # Content type distribution
            content_type = blob.content_type or 'unknown'
            stats['content_types'][content_type] = \
                stats['content_types'].get(content_type, 0) + 1

            # Age distribution
            if blob.time_created:
                age_days = (now - blob.time_created).days
                if age_days <= 30:
                    stats['age_distribution']['0-30_days'] += 1
                elif age_days <= 90:
                    stats['age_distribution']['31-90_days'] += 1
                elif age_days <= 365:
                    stats['age_distribution']['91-365_days'] += 1
                else:
                    stats['age_distribution']['over_365_days'] += 1

        return stats

    def query_access_logs(self, bucket_name, days_back=7):
        """Query Cloud Storage access logs."""
        query = f"""
        SELECT
            time_micros,
            c_ip as client_ip,
            cs_method as method,
            cs_uri as uri,
            sc_status as status,
            cs_bytes as request_bytes,
            sc_bytes as response_bytes,
            time_taken_micros,
            cs_host as host,
            cs_user_agent as user_agent
        FROM
            `{self.project_id}.cloud_storage_logs.access_logs_*`
        WHERE
            _TABLE_SUFFIX >= FORMAT_DATE('%Y%m%d', 
                DATE_SUB(CURRENT_DATE(), INTERVAL {days_back} DAY))
            AND cs_bucket = '{bucket_name}'
        ORDER BY
            time_micros DESC
        LIMIT 1000
        """

        query_job = self.bq_client.query(query)
        results = query_job.result()

        return pd.DataFrame([dict(row) for row in results])

    def get_bucket_metrics(self, bucket_name):
        """Get Cloud Monitoring metrics for bucket."""
        project_name = f"projects/{self.project_id}"

        # Define time range
        interval = monitoring_v3.TimeInterval({
            "end_time": {"seconds": int(datetime.now().timestamp())},
            "start_time": {"seconds": int((datetime.now() - timedelta(hours=24)).timestamp())}
        })

        # Query total bytes metric
        results = self.monitoring_client.list_time_series(
            request={
                "name": project_name,
                "filter": f'metric.type="storage.googleapis.com/storage/total_bytes" '
                         f'AND resource.label.bucket_name="{bucket_name}"',
                "interval": interval,
                "view": monitoring_v3.ListTimeSeriesRequest.TimeSeriesView.FULL
            }
        )

        metrics = []
        for result in results:
            for point in result.points:
                metrics.append({
                    'timestamp': point.interval.end_time,
                    'value': point.value.int64_value,
                    'metric_type': result.metric.type
                })

        return metrics

# best_practices.py
from google.cloud import storage
import hashlib
import json

class StorageBestPractices:
    """Implement GCS best practices."""

    def __init__(self, project_id):
        self.client = storage.Client(project=project_id)

    def create_optimized_bucket(self, bucket_name, location="US"):
        """Create bucket with best practice configurations."""
        bucket = self.client.bucket(bucket_name)

        # Set location
        bucket.location = location

        # Enable versioning
        bucket.versioning_enabled = True

        # Set uniform bucket-level access
        bucket.iam_configuration.uniform_bucket_level_access_enabled = True

        # Set default storage class
        bucket.storage_class = "STANDARD"

        # Enable default event-based hold for compliance
        bucket.default_event_based_hold = False

        # Set lifecycle rules
        bucket.lifecycle_rules = [
            {
                "action": {"type": "Delete"},
                "condition": {
                    "age": 7,
                    "matchesPrefix": ["tmp/", "temp/"]
                }
            },
            {
                "action": {"type": "SetStorageClass", "storageClass": "NEARLINE"},
                "condition": {
                    "age": 30,
                    "matchesStorageClass": ["STANDARD"]
                }
            }
        ]

        # Create bucket
        bucket.create()

        # Set retention policy if needed
        bucket.retention_period = 7 * 24 * 60 * 60  # 7 days in seconds
        bucket.patch()

        print(f"Created optimized bucket: {bucket_name}")
        return bucket

    def upload_with_metadata(self, bucket_name, source_file, blob_name, 
                           custom_metadata=None):
        """Upload file with comprehensive metadata."""
        bucket = self.client.bucket(bucket_name)
        blob = bucket.blob(blob_name)

        # Calculate checksums
        with open(source_file, 'rb') as f:
            content = f.read()
            md5_hash = hashlib.md5(content).hexdigest()
            crc32c = google_crc32c.Checksum()
            crc32c.update(content)

        # Set metadata
        blob.metadata = custom_metadata or {}
        blob.metadata.update({
            'uploaded-by': 'storage-best-practices',
            'upload-timestamp': datetime.now().isoformat(),
            'original-filename': os.path.basename(source_file),
            'file-size': str(len(content))
        })

        # Set content type
        content_type = mimetypes.guess_type(source_file)[0]
        if content_type:
            blob.content_type = content_type

        # Set cache control
        if content_type and content_type.startswith('image/'):
            blob.cache_control = 'public, max-age=86400'  # 1 day
        else:
            blob.cache_control = 'private, max-age=0'

        # Upload with verification
        blob.upload_from_filename(
            source_file,
            checksum='crc32c'
        )

        print(f"Uploaded {blob_name} with metadata and checksums")
        return blob

    def implement_backup_strategy(self, source_bucket, backup_bucket):
        """Implement backup strategy between buckets."""
        source = self.client.bucket(source_bucket)
        backup = self.client.bucket(backup_bucket)

        # Configure backup bucket
        backup.storage_class = "NEARLINE"  # Cost-effective for backups
        backup.versioning_enabled = True
        backup.lifecycle_rules = [{
            "action": {"type": "Delete"},
            "condition": {"age": 365}  # Keep backups for 1 year
        }]
        backup.patch()

        # Copy objects
        for blob in source.list_blobs():
            if not blob.name.startswith('tmp/'):  # Skip temporary files
                copy_blob = backup.copy_blob(
                    blob, 
                    backup,
                    new_name=f"backup/{datetime.now().strftime('%Y%m%d')}/{blob.name}"
                )
                print(f"Backed up: {blob.name}")

        return backup

# cost_optimizer.py
from google.cloud import storage
from google.cloud import bigquery
import pandas as pd

class CostOptimizer:
    """Optimize GCS costs."""

    def __init__(self, project_id):
        self.storage_client = storage.Client(project=project_id)
        self.bq_client = bigquery.Client(project=project_id)
        self.project_id = project_id

    def analyze_storage_costs(self):
        """Analyze storage costs across all buckets."""
        query = f"""
        WITH storage_costs AS (
            SELECT
                resource.labels.bucket_name as bucket,
                resource.labels.location as location,
                metric.labels.storage_class as storage_class,
                value.int64_value as bytes_stored,
                timestamp
            FROM
                `{self.project_id}.cloud_monitoring.storage_total_bytes`
            WHERE
                timestamp >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 30 DAY)
        ),
        cost_per_gb AS (
            SELECT
                'STANDARD' as storage_class, 0.020 as cost_per_gb_month
            UNION ALL SELECT 'NEARLINE', 0.010
            UNION ALL SELECT 'COLDLINE', 0.004
            UNION ALL SELECT 'ARCHIVE', 0.0012
        )
        SELECT
            s.bucket,
            s.storage_class,
            AVG(s.bytes_stored) / POW(1024, 3) as avg_gb_stored,
            c.cost_per_gb_month,
            (AVG(s.bytes_stored) / POW(1024, 3)) * c.cost_per_gb_month as estimated_monthly_cost
        FROM
            storage_costs s
        JOIN
            cost_per_gb c ON s.storage_class = c.storage_class
        GROUP BY
            s.bucket, s.storage_class, c.cost_per_gb_month
        ORDER BY
            estimated_monthly_cost DESC
        """

        results = self.bq_client.query(query).to_dataframe()
        return results

    def recommend_cost_savings(self):
        """Recommend cost-saving opportunities."""
        recommendations = []

        for bucket in self.storage_client.list_buckets():
            # Check for lifecycle policies
            if not bucket.lifecycle_rules:
                recommendations.append({
                    'bucket': bucket.name,
                    'recommendation': 'Add lifecycle rules',
                    'potential_savings': 'Up to 50% on aged data',
                    'action': 'Implement age-based storage class transitions'
                })

            # Check for versioning without lifecycle
            if bucket.versioning_enabled and not any(
                rule.get('action', {}).get('type') == 'Delete' 
                for rule in bucket.lifecycle_rules or []
            ):
                recommendations.append({
                    'bucket': bucket.name,
                    'recommendation': 'Clean up old versions',
                    'potential_savings': '10-30% reduction',
                    'action': 'Add lifecycle rule to delete old versions'
                })

            # Check storage class distribution
            stats = self.analyze_bucket_usage(bucket.name)
            if stats['storage_classes'].get('STANDARD', 0) > 0.7 * stats['total_objects']:
                recommendations.append({
                    'bucket': bucket.name,
                    'recommendation': 'Review storage classes',
                    'potential_savings': 'Up to 80% on infrequently accessed data',
                    'action': 'Move cold data to NEARLINE or COLDLINE'
                })

        return recommendations