AiDA_Python/app/service/design/utils/upload_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 time

# import boto3
import cv2
from minio import Minio

from app.core.config import *
from app.service.utils.decorator import RunTime

minio_client = Minio(
    f"{MINIO_URL}",
    access_key=MINIO_ACCESS,
    secret_key=MINIO_SECRET,
    secure=MINIO_SECURE)

"""S3 上传"""
# 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)



@RunTime
def upload_png_mask(front_image, object_name, mask=None):
    mask_url = None
    if mask is not None:
        # 反转掩模
        mask_inverted = cv2.bitwise_not(mask)
        # 将掩模转换为 RGBA 格式
        rgba_image = cv2.cvtColor(mask_inverted, cv2.COLOR_BGR2BGRA)
        rgba_image[rgba_image[:, :, 0] == 0] = [0, 0, 0, 0]
        # 将图像数据保存到内存中的 BytesIO 对象中
        image_bytes = io.BytesIO()
        image_bytes.write(cv2.imencode('.png', rgba_image)[1].tobytes())
        image_bytes.seek(0)
        try:
            key = f"mask/mask_{object_name}.png"
            mask_url = f"{AIDA_CLOTHING}/{key}"
            s3.put_object(Bucket=AIDA_CLOTHING, Key=key, Body=image_bytes, ContentType='image/png')
        except Exception as e:
            print(f'上传到 S3 失败: {e}')
    with io.BytesIO() as output:
        front_image.save(output, format='PNG')
        data = output.getvalue()
    # 创建一个 S3 客户端
    try:
        key = f"image/image_{object_name}.png"
        image_url = f"{AIDA_CLOTHING}/{key}"
        s3.put_object(Bucket=AIDA_CLOTHING, Key=key, Body=data, ContentType='image/png')
        return front_image, image_url, mask_url
    except Exception as e:
        print(f'上传到 S3 失败: {e}')


@RunTime
def upload_layer_image(image, object_name):
    with io.BytesIO() as output:
        image.save(output, format='PNG')
        data = output.getvalue()
    # 创建一个 S3 客户端
    try:
        key = f"image/image_{object_name}.png"
        image_url = f"{AIDA_CLOTHING}/{key}"
        s3.put_object(Bucket=AIDA_CLOTHING, Key=key, Body=data, ContentType='image/png')
        return image_url
    except Exception as e:
        print(f'上传到 S3 失败: {e}')


@RunTime
def upload_mask_image(mask, object_name):
    # 反转掩模
    mask_inverted = cv2.bitwise_not(mask)
    # 将掩模转换为 RGBA 格式
    rgba_image = cv2.cvtColor(mask_inverted, cv2.COLOR_BGR2BGRA)
    rgba_image[rgba_image[:, :, 0] == 0] = [0, 0, 0, 0]
    # 将图像数据保存到内存中的 BytesIO 对象中
    image_bytes = io.BytesIO()
    image_bytes.write(cv2.imencode('.png', rgba_image)[1].tobytes())
    image_bytes.seek(0)
    try:
        key = f"mask/mask_{object_name}.png"
        mask_url = f"{AIDA_CLOTHING}/{key}"
        s3.put_object(Bucket=AIDA_CLOTHING, Key=key, Body=image_bytes, ContentType='image/png')
        return mask_url
    except Exception as e:
        print(f'上传到 S3 失败: {e}')


"""minio 上传"""

# @RunTime
# def upload_png_mask(front_image, object_name, mask=None):
#     start_time = time.time()
#     try:
#         mask_url = None
#         if mask is not None:
#             mask_inverted = cv2.bitwise_not(mask)
#             # 将掩模的3通道转换为4通道，白色部分不透明，黑色部分透明
#             rgba_image = cv2.cvtColor(mask_inverted, cv2.COLOR_BGR2BGRA)
#             rgba_image[rgba_image[:, :, 0] == 0] = [0, 0, 0, 0]
#             image_bytes = io.BytesIO()
#             image_bytes.write(cv2.imencode('.png', rgba_image)[1].tobytes())
#
#             image_bytes.seek(0)
#             mask_url = f"{AIDA_CLOTHING}/{minio_client.put_object('aida-clothing', f'mask/mask_{object_name}.png', image_bytes, len(image_bytes.getvalue()), content_type='image/png').object_name}"
#
#         image_data = io.BytesIO()
#         front_image.save(image_data, format='PNG')
#         image_data.seek(0)
#         image_bytes = image_data.read()
#         image_url = f"{AIDA_CLOTHING}/{minio_client.put_object('aida-clothing', f'image/image_{object_name}.png', io.BytesIO(image_bytes), len(image_bytes), content_type='image/png').object_name}"
#         # print(f"upload_png_mask {object_name} = {time.time() - start_time}")
#         return front_image, image_url, mask_url
#     except Exception as e:
#         logging.warning(f"upload_png_mask runtime exception : {e}")
#
#
# @RunTime
# def upload_layer_image(image, object_name):
#     try:
#         image_data = io.BytesIO()
#         image.save(image_data, format='PNG')
#         image_data.seek(0)
#         image_bytes = image_data.read()
#         image_url = f"{AIDA_CLOTHING}/{minio_client.put_object('aida-clothing', f'image/image_{object_name}.png', io.BytesIO(image_bytes), len(image_bytes), content_type='image/png').object_name}"
#         return image_url
#     except Exception as e:
#         logging.warning(f"upload_png_mask runtime exception : {e}")
#
#
# @RunTime
# def upload_mask_image(mask, object_name):
#     try:
#         mask_inverted = cv2.bitwise_not(mask)
#         # 将掩模的3通道转换为4通道，白色部分不透明，黑色部分透明
#         rgba_image = cv2.cvtColor(mask_inverted, cv2.COLOR_BGR2BGRA)
#         rgba_image[rgba_image[:, :, 0] == 0] = [0, 0, 0, 0]
#         image_bytes = io.BytesIO()
#         image_bytes.write(cv2.imencode('.png', rgba_image)[1].tobytes())
#
#         image_bytes.seek(0)
#         mask_url = f"{AIDA_CLOTHING}/{minio_client.put_object('aida-clothing', f'mask/mask_{object_name}.png', image_bytes, len(image_bytes.getvalue()), content_type='image/png').object_name}"
#         return mask_url
#     except Exception as e:
#         logging.warning(f"upload_png_mask runtime exception : {e}")