Underground-Digital/Workflow-Engine

1import logging2import uuid3from typing import IO, Any4 5from core.model_runtime.model_providers.__base.model_provider import ModelProvider6 7logger = logging.getLogger(__name__)8 9 10class SageMakerProvider(ModelProvider):11    def validate_provider_credentials(self, credentials: dict) -> None:12        """13        Validate provider credentials14 15        if validate failed, raise exception16 17        :param credentials: provider credentials, credentials form defined in `provider_credential_schema`.18        """19        pass20 21 22def buffer_to_s3(s3_client: Any, file: IO[bytes], bucket: str, s3_prefix: str) -> str:23    """24    return s3_uri of this file25    """26    s3_key = f"{s3_prefix}{uuid.uuid4()}.mp3"27    s3_client.put_object(Body=file.read(), Bucket=bucket, Key=s3_key, ContentType="audio/mp3")28    return s3_key29 30 31def generate_presigned_url(s3_client: Any, file: IO[bytes], bucket_name: str, s3_prefix: str, expiration=600) -> str:32    object_key = buffer_to_s3(s3_client, file, bucket_name, s3_prefix)33    try:34        response = s3_client.generate_presigned_url(35            "get_object", Params={"Bucket": bucket_name, "Key": object_key}, ExpiresIn=expiration36        )37    except Exception as e:38        print(f"Error generating presigned URL: {e}")39        return None40 41    return response42