feat generate single logo 部署

app/service/generate_image/utils/upload_sd_image.py

@@ -10,6 +10,7 @@
 import io
 import logging
 
+import boto3
 import cv2
 from PIL import Image
 from minio import Minio
@@ -17,6 +18,39 @@ from minio import Minio
 from app.core.config import *
 
 minio_client = Minio(MINIO_URL, access_key=MINIO_ACCESS, secret_key=MINIO_SECRET, secure=MINIO_SECURE)
+s3 = boto3.client('s3', aws_access_key_id=S3_ACCESS_KEY, aws_secret_access_key=S3_AWS_SECRET_ACCESS_KEY, region_name=S3_REGION_NAME)
+
+
+# def upload_single_logo(image, user_id, category, object_name):
+#     with io.BytesIO() as output:
+#         image.save(output, format='PNG')
+#         data = output.getvalue()
+#     # 创建一个 S3 客户端
+#     try:
+#         key = f'{user_id}/{category}/{object_name}'
+#         image_url = f"{AIDA_CLOTHING}/{key}"
+#         s3.put_object(Bucket=GSL_MINIO_BUCKET, Key=key, Body=data, ContentType='image/png')
+#         return image_url
+#     except Exception as e:
+#         print(f'上传到 S3 失败: {e}')
+
+def upload_single_logo(image, user_id, category, object_name):
+    try:
+        image_data = io.BytesIO()
+        image.save(image_data, format='PNG')
+        image_data.seek(0)
+        image_bytes = image_data.read()
+        minio_req = minio_client.put_object(
+            GI_MINIO_BUCKET,
+            f'{user_id}/{category}/{object_name}',
+            io.BytesIO(image_bytes),
+            len(image_bytes),
+            content_type='image/jpeg'
+        )
+        image_url = f"aida-users/{minio_req.object_name}"
+        return image_url
+    except Exception as e:
+        logging.warning(f"upload_png_mask runtime exception : {e}")
 
 
 def upload_png_sd(image, user_id, category, object_name):