AiDA_Python/app/service/generate_image/utils/upload_sd_image.py

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
"""
@Project ：trinity_client
@File    ：upload_image.py
@Author  ：周成融
@Date    ：2023/8/28 13:49:20
@detail  ：
"""
import io
import logging

# import boto3
import cv2
from PIL import Image
from minio import Minio

from app.core.config import *
from app.service.utils.oss_client import oss_upload_image

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_SDXL_image(image, user_id, category, file_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}/{file_name}',
        #     io.BytesIO(image_bytes),
        #     len(image_bytes),
        #     content_type='image/jpeg'
        # )
        object_name = f'{user_id}/{category}/{file_name}'
        req = oss_upload_image(bucket=GI_MINIO_BUCKET, object_name=object_name, image_bytes=image_bytes)
        image_url = f"aida-users/{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, file_name):
    try:
        _, img_byte_array = cv2.imencode('.jpg', image)
        object_name = f'{user_id}/{category}/{file_name}'
        req = oss_upload_image(bucket=GI_MINIO_BUCKET, object_name=object_name, image_bytes=img_byte_array)
        image_url = f"aida-users/{object_name}"
        return image_url
    except Exception as e:
        logging.warning(f"upload_png_mask runtime exception : {e}")


def upload_stain_png_sd(image, user_id, category, object_name):
    try:
        _, img_byte_array = cv2.imencode('.jpg', image)
        minio_req = minio_client.put_object(
            "test",
            f'generate_result/stain/{user_id}_{category}_{object_name}',
            io.BytesIO(img_byte_array),
            len(img_byte_array),
            content_type='image/jpeg'
        )
        image_url = f"test/{minio_req.object_name}"
        return image_url
    except Exception as e:
        logging.warning(f"upload_png_mask runtime exception : {e}")


def upload_face_png_sd(image, user_id, category, object_name):
    try:
        _, img_byte_array = cv2.imencode('.jpg', image)
        minio_req = minio_client.put_object(
            "test",
            f'generate_result/face/{user_id}_{category}_{object_name}',
            io.BytesIO(img_byte_array),
            len(img_byte_array),
            content_type='image/jpeg'
        )
        image_url = f"test/{minio_req.object_name}"
        return image_url
    except Exception as e:
        logging.warning(f"upload_png_mask runtime exception : {e}")