102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
#!/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 *
|
||
|
||
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, 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):
|
||
try:
|
||
_, img_byte_array = cv2.imencode('.jpg', image)
|
||
minio_req = minio_client.put_object(
|
||
GI_MINIO_BUCKET,
|
||
f'{user_id}/{category}/{object_name}',
|
||
io.BytesIO(img_byte_array),
|
||
len(img_byte_array),
|
||
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_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}")
|