feat generate slogan | to product image | slogan 接口部署

app/api/api_generate_image.py

@@ -1,11 +1,15 @@
 import logging
 from fastapi import APIRouter, BackgroundTasks
-from app.schemas.generate_image import GenerateImageModel
+from app.schemas.generate_image import GenerateImageModel, GenerateProductImageModel, GenerateSingleLogoImageModel
-from app.service.generate_image.service import GenerateImage, infer_cancel
+from app.service.generate_image.service_generate_image import GenerateImage, infer_cancel as generate_image_infer_cancel
+from app.service.generate_image.service_generate_product_image import GenerateProductImage, infer_cancel as generate_product_image_cancel
+from app.service.generate_image.service_generate_single_logo import GenerateSingleLogoImage, infer_cancel as generate_single_logo_cancel
 
 router = APIRouter()
 logger = logging.getLogger()
 
+'''generate image'''
+
 
 @router.post("/generate_image")
 def generate_image(request_item: GenerateImageModel, background_tasks: BackgroundTasks):
@@ -24,5 +28,53 @@ def generate_image(request_item: GenerateImageModel, background_tasks: Backgroun
 
 @router.get("/generate_cancel/{tasks_id}>")
 def generate_image(tasks_id):
-    result = infer_cancel(tasks_id)
+    result = generate_image_infer_cancel(tasks_id)
+    return {"code": 200, "message": result['message'], "data": result['data']}
+
+
+'''single logo'''
+
+
+@router.post("/generate_single_logo")
+def generate_single_logo(request_item: GenerateSingleLogoImageModel, background_tasks: BackgroundTasks):
+    try:
+        logger.info(f"request data ### : {request_item}")
+        service = GenerateSingleLogoImage(request_item)
+        background_tasks.add_task(service.get_result)
+        code = 200
+        message = "access"
+    except Exception as e:
+        code = 400
+        message = e
+        logger.warning(e)
+    return {"code": code, "message": message}
+
+
+@router.get("/generate_single_logo_cancel/{tasks_id}>")
+def generate_single_logo_image(tasks_id):
+    result = generate_single_logo_cancel(tasks_id)
+    return {"code": 200, "message": result['message'], "data": result['data']}
+
+
+'''product image'''
+
+
+@router.post("/generate_product_image")
+def generate_product_image(request_item: GenerateProductImageModel, background_tasks: BackgroundTasks):
+    try:
+        logger.info(f"request data ### : {request_item}")
+        service = GenerateProductImage(request_item)
+        background_tasks.add_task(service.get_result)
+        code = 200
+        message = "access"
+    except Exception as e:
+        code = 400
+        message = e
+        logger.warning(e)
+    return {"code": code, "message": message}
+
+
+@router.get("/generate_product_image_cancel_cancel/{tasks_id}>")
+def generate_single_logo_image(tasks_id):
+    result = generate_product_image_cancel(tasks_id)
     return {"code": 200, "message": result['message'], "data": result['data']}

app/api/api_route.py

@@ -8,6 +8,7 @@ from app.api import api_design
 from app.api import api_chat_robot
 from app.api import api_prompt_generation
 from app.api import api_design_pre_processing
+from app.api import api_slogan
 
 
 router = APIRouter()
@@ -20,3 +21,4 @@ router.include_router(api_design.router, tags=['design'], prefix="/api")
 router.include_router(api_chat_robot.router, tags=['chat_robot'], prefix="/api")
 router.include_router(api_prompt_generation.router, tags=['prompt_generation'], prefix="/api")
 router.include_router(api_design_pre_processing.router, tags=['design_pre_processing'], prefix="/api")
+router.include_router(api_slogan.router, tags=['slogan'], prefix="/api")

app/api/api_slogan.py

@@ -0,0 +1,24 @@
+import logging
+import time
+from fastapi import APIRouter, BackgroundTasks
+
+from app.schemas.slogan import SloganModel
+from app.service.slogan.service import Slogan
+
+router = APIRouter()
+logger = logging.getLogger()
+
+
+@router.post("/slogan")
+def slogan(request_item: SloganModel, background_tasks: BackgroundTasks):
+    try:
+        logger.info(f"request data ### : {request_item}")
+        service = Slogan(request_item)
+        background_tasks.add_task(service.get_result)
+        code = 200
+        message = "access"
+    except Exception as e:
+        code = 400
+        message = e
+        logger.warning(e)
+    return {"code": code, "message": message}

app/core/config.py

@@ -103,6 +103,15 @@ GI_MINIO_BUCKET = "aida-users"
 GI_RABBITMQ_QUEUES = os.getenv("GI_RABBITMQ_QUEUES", f"GenerateImage{RABBITMQ_ENV}")
 GI_SYS_IMAGE_URL = "aida-sys-image/generate_image/white_image.jpg"
 
+# SLOGAN service config
+SLOGAN_RABBITMQ_QUEUES = os.getenv("SLOGAN_RABBITMQ_QUEUES", f"Slogan{RABBITMQ_ENV}")
+
+# Generate Single Logo service config
+GEN_SINGLE_LOGO_RABBITMQ_QUEUES = os.getenv("GEN_SINGLE_LOGO_RABBITMQ_QUEUES", f"GenSingleLogo{RABBITMQ_ENV}")
+
+# Generate Single Logo service config
+GEN_PRODUCT_IMAGE_RABBITMQ_QUEUES = os.getenv("GEN_PRODUCT_IMAGE_RABBITMQ_QUEUES", f"GenProductImage{RABBITMQ_ENV}")
+
 # SEG service config
 SEG_MODEL_URL = '10.1.1.240:10000'
 SEGMENTATION = {

app/schemas/generate_image.py

@@ -8,3 +8,15 @@ class GenerateImageModel(BaseModel):
     mode: str
     category: str
     gender: str
+
+
+class GenerateSingleLogoImageModel(BaseModel):
+    tasks_id: str
+    prompt: str
+    image_url: str
+
+
+class GenerateProductImageModel(BaseModel):
+    tasks_id: str
+    prompt: str
+    image_url: str

app/schemas/slogan.py

@@ -0,0 +1,7 @@
+from pydantic import BaseModel
+
+
+class SloganModel(BaseModel):
+    prompt: str
+    svg: str
+    tasks_id: str

app/service/design/items/pipelines/split.py

@@ -2,6 +2,7 @@ import logging
 import cv2
 import numpy as np
 from cv2 import cvtColor, COLOR_BGR2RGBA
+
 from app.service.utils.generate_uuid import generate_uuid
 from ..builder import PIPELINES
 from PIL import Image
@@ -45,8 +46,11 @@ class Split(object):
                 result_front_image[front_mask != 0] = rgba_image[front_mask != 0]
 
                 result_front_image_pil = Image.fromarray(cvtColor(result_front_image, COLOR_BGR2RGBA))
-                front_new_size = (int(result_front_image_pil.width * result["scale"] * result["resize_scale"]), int(result_front_image_pil.height * result["scale"] * result["resize_scale"]))
+                front_new_size = (int(result_front_image_pil.width * result["scale"] * result["resize_scale"][0]), int(result_front_image_pil.height * result["scale"] * result["resize_scale"][1]))
                 result_front_image_pil = result_front_image_pil.resize(front_new_size, Image.LANCZOS)
+                # TODO 多线程外部上传图片到minio
+                # result['front_mask_image'] = cv2.resize(front_mask, front_new_size)
+                # result['front_image'] = result_front_image_pil
                 front_mask = cv2.resize(front_mask, front_new_size)
                 result['front_image'], result["front_image_url"], result["front_mask_url"] = upload_png_mask(result_front_image_pil, f'{generate_uuid()}', mask=front_mask)
 
@@ -55,61 +59,19 @@ class Split(object):
                     result_back_image[back_mask != 0] = rgba_image[back_mask != 0]
 
                     result_back_image_pil = Image.fromarray(cvtColor(result_back_image, COLOR_BGR2RGBA))
-                    back_new_size = (int(result_back_image_pil.width * result["scale"] * result["resize_scale"]), int(result_back_image_pil.height * result["scale"] * result["resize_scale"]))
+                    back_new_size = (int(result_back_image_pil.width * result["scale"] * result["resize_scale"][0]), int(result_back_image_pil.height * result["scale"] * result["resize_scale"][1]))
                     result_back_image_pil = result_back_image_pil.resize(back_new_size, Image.LANCZOS)
+                    # TODO 多线程外部上传图片到minio
+                    # result['back_mask_image'] = cv2.resize(back_mask, back_new_size)
+                    # result['back_image'] = result_back_image_pil
+
                     back_mask = cv2.resize(back_mask, back_new_size)
                     result['back_image'], result["back_image_url"], result["back_mask_url"] = upload_png_mask(result_back_image_pil, f'{generate_uuid()}', mask=back_mask)
                 else:
                     result['back_image'] = None
                     result["back_image_url"] = None
                     result["back_mask_url"] = None
+                    result['back_mask_image'] = None
             return result
         except Exception as e:
             logging.warning(f"split runtime exception : {e} image_id : {result['image_id']}")
-
-    # @ RunTime
-    # def __call__(self, result):
-    #     try:
-    #         if 'mask' not in result.keys():
-    #             raise KeyError(f'Cannot find mask in result dict, please check ContourDetection is included in process pipelines.')
-    #         if 'seg_result' not in result.keys():  # 没过seg模型
-    #             result['front_mask'] = result['mask'].copy()
-    #             result['back_mask'] = np.zeros_like(result['mask'])
-    #         else:
-    #             temp_front = result['seg_result'] == 1
-    #             result['front_mask'] = (result['mask'] * (temp_front + 0).astype(np.uint8))
-    #             temp_back = result['seg_result'] == 2
-    #             result['back_mask'] = (result['mask'] * (temp_back + 0).astype(np.uint8))
-    #
-    #         if result['name'] in ('outwear', 'dress', 'blouse', 'skirt', 'trousers', 'tops', 'bottoms'):
-    #             if len(result['front_mask'].shape) > 2:
-    #                 front_mask = result['front_mask'][0]
-    #             else:
-    #                 front_mask = result['front_mask']
-    #
-    #             rgba_image = rgb_to_rgba((result['final_image'].shape[0], result['final_image'].shape[1]), result['final_image'], result['mask'])
-    #             result_front_image = np.zeros_like(rgba_image)
-    #             result_front_image[front_mask != 0] = rgba_image[front_mask != 0]
-    #
-    #             result_front_image_pil = Image.fromarray(cvtColor(result_front_image, COLOR_BGR2RGBA))
-    #             front_new_size = (int(result_front_image_pil.width * result["scale"] * result["resize_scale"]), int(result_front_image_pil.height * result["scale"] * result["resize_scale"]))
-    #             result_front_image_pil = result_front_image_pil.resize(front_new_size, Image.LANCZOS)
-    #             front_mask = cv2.resize(front_mask, front_new_size)
-    #             result['front_image'], result["front_image_url"], result["front_mask_url"] = upload_png_mask(result_front_image_pil, f'{generate_uuid()}', mask=front_mask)
-    #
-    #             if result["name"] in ('blouse', 'dress', 'outwear', 'tops'):
-    #                 result_back_image = np.zeros_like(rgba_image)
-    #                 result_back_image[result['back_mask'] != 0] = rgba_image[result['back_mask'] != 0]
-    #
-    #                 result_back_image_pil = Image.fromarray(cvtColor(result_back_image, COLOR_BGR2RGBA))
-    #                 back_new_size = (int(result_back_image_pil.width * result["scale"] * result["resize_scale"]), int(result_back_image_pil.height * result["scale"] * result["resize_scale"]))
-    #                 result_back_image_pil = result_back_image_pil.resize(back_new_size, Image.LANCZOS)
-    #                 back_mask = cv2.resize(result['back_mask'], back_new_size)
-    #                 result['back_image'], result["back_image_url"], result["back_mask_url"] = upload_png_mask(result_back_image_pil, f'{generate_uuid()}', mask=back_mask)
-    #             else:
-    #                 result['back_image'] = None
-    #                 result["back_image_url"] = None
-    #                 result["back_mask_url"] = None
-    #         return result
-    #     except Exception as e:
-    #         logging.warning(f"split runtime exception : {e} image_id : {result['image_id']}")

app/service/design/utils/synthesis_item.py

@@ -17,14 +17,15 @@ import numpy as np
 from PIL import Image
 from minio import Minio
 
+from app.core.config import *
 from app.service.utils.decorator import RunTime
 from app.service.utils.generate_uuid import generate_uuid
 
-# minio_client = Minio(
+minio_client = Minio(
-#     f"{MINIO_IP}:{MINIO_PORT}",
+    MINIO_URL,
-#     access_key=MINIO_ACCESS,
+    access_key=MINIO_ACCESS,
-#     secret_key=MINIO_SECRET,
+    secret_key=MINIO_SECRET,
-#     secure=MINIO_SECURE)
+    secure=MINIO_SECURE)
 
 s3 = boto3.client(
     's3',
@@ -130,19 +131,19 @@ def synthesis(data, size):
             result_image.save(output, format='PNG')
             data = output.getvalue()
 
-        # image_data = io.BytesIO()
+        image_data = io.BytesIO()
-        # result_image.save(image_data, format='PNG')
+        result_image.save(image_data, format='PNG')
-        # image_data.seek(0)
+        image_data.seek(0)
-        # image_bytes = image_data.read()
+        image_bytes = image_data.read()
-        # return f"aida-results/{minio_client.put_object('aida-results', f'result_{generate_uuid()}.png', io.BytesIO(image_bytes), len(image_bytes), content_type='image/png').object_name}"
+        return f"aida-results/{minio_client.put_object('aida-results', f'result_{generate_uuid()}.png', io.BytesIO(image_bytes), len(image_bytes), content_type='image/png').object_name}"
 
-        object_name = f'result_{generate_uuid()}.png'
+        # object_name = f'result_{generate_uuid()}.png'
-        response = s3.put_object(Bucket="aida-results", Key=object_name, Body=data, ContentType='image/png')
+        # response = s3.put_object(Bucket="aida-results", Key=object_name, Body=data, ContentType='image/png')
-        object_url = f"aida-results/{object_name}"
+        # object_url = f"aida-results/{object_name}"
-        if response['ResponseMetadata']['HTTPStatusCode'] == 200:
+        # if response['ResponseMetadata']['HTTPStatusCode'] == 200:
-            return object_url
+        #     return object_url
-        else:
+        # else:
-            return ""
+        #     return ""
 
     except Exception as e:
         logging.warning(f"synthesis runtime exception : {e}")

app/service/design/utils/upload_image.py

@@ -33,128 +33,129 @@ s3 = boto3.client(
 )
 
 
-@RunTime
-def upload_png_mask(front_image, object_name, mask=None):
-    start_time = time.time()
-    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 格式
 #         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)
-#             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}"
+#         try:
-#
+#             key = f"mask/mask_{object_name}.png"
-#         image_data = io.BytesIO()
+#             mask_url = f"{AIDA_CLOTHING}/{key}"
-#         front_image.save(image_data, format='PNG')
+#             s3.put_object(Bucket=AIDA_CLOTHING, Key=key, Body=image_bytes, ContentType='image/png')
-#         image_data.seek(0)
+#         except Exception as e:
-#         image_bytes = image_data.read()
+#             print(f'上传到 S3 失败: {e}')
-#         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}"
+#     with io.BytesIO() as output:
-#         # print(f"upload_png_mask {object_name} = {time.time() - start_time}")
+#         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:
-#         logging.warning(f"upload_png_mask runtime exception : {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:
-#         image_data = io.BytesIO()
+#         key = f"image/image_{object_name}.png"
-#         image.save(image_data, format='PNG')
+#         image_url = f"{AIDA_CLOTHING}/{key}"
-#         image_data.seek(0)
+#         s3.put_object(Bucket=AIDA_CLOTHING, Key=key, Body=data, ContentType='image/png')
-#         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}")
+#         print(f'上传到 S3 失败: {e}')
 #
 #
 # @RunTime
 # def upload_mask_image(mask, object_name):
-#     try:
+#     # 反转掩模
 #     mask_inverted = cv2.bitwise_not(mask)
-#         # 将掩模的3通道转换为4通道，白色部分不透明，黑色部分透明
+#     # 将掩模转换为 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)
-#         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}"
+#     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:
-#         logging.warning(f"upload_png_mask runtime exception : {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}")

app/service/slogan/service.py

@@ -0,0 +1,36 @@
+import json
+import logging
+
+import redis
+
+from app.core.config import *
+
+logger = logging.getLogger()
+
+
+class Slogan:
+    def __init__(self, request_data):
+        self.tasks_id = request_data.tasks_id
+        self.prompt = request_data.prompt
+        self.svg = request_data.svg
+        self.redis_client = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, db=REDIS_DB, decode_responses=True)
+        self.slogan_data = {'tasks_id': self.tasks_id, 'status': 'PENDING', 'message': "pending", 'image_url': ''}
+        self.redis_client.set(self.tasks_id, json.dumps(self.slogan_data))
+        self.redis_client.expire(self.tasks_id, 600)
+
+        # if DEBUG is False:
+        self.connection = pika.BlockingConnection(pika.ConnectionParameters(**RABBITMQ_PARAMS))
+        self.channel = self.connection.channel()
+        self.result_image_url = "test/slogan/init_img.png"
+
+    def read_tasks_status(self):
+        status_data = self.redis_client.get(self.tasks_id)
+        return json.loads(status_data), status_data
+
+    def get_result(self):
+        self.slogan_data['status'] = "SUCCESS"
+        self.slogan_data['message'] = "success"
+        self.slogan_data['image_url'] = "test/slogan/init_img.png"
+        dict_slogan_data, str_slogan_data = self.read_tasks_status()
+        self.channel.basic_publish(exchange='', routing_key=SLOGAN_RABBITMQ_QUEUES, body=str_slogan_data)
+        logger.info(f" [x] Sent {json.dumps(dict_slogan_data, indent=4)}")