import json import logging import httpx import requests from fastapi import APIRouter, BackgroundTasks, HTTPException from app.core.config import settings from app.schemas.generate_image import GenerateImageModel, GenerateProductImageModel, GenerateSingleLogoImageModel, GenerateRelightImageModel, GenerateMultiViewModel, BatchGenerateProductImageModel, BatchGenerateRelightImageModel, AgentTollGenerateImageModel, Flux2ToProductImgModel, GenerateSloganImageModel, GenerateImageFlux2KleinModel from app.schemas.pose_transform import BatchPoseTransformModel from app.schemas.response_template import ResponseModel from app.service.generate_batch_image.service import start_product_batch_generate, start_relight_batch_generate, start_pose_transform_batch_generate from app.service.generate_image.service_agent_tool_generate_image import AgentToolGenerateImage from app.service.generate_image.service_generate_image import GenerateImage, infer_cancel as generate_image_infer_cancel from app.service.generate_image.service_generate_multi_view import GenerateMultiView, infer_cancel as generate_multi_view_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_relight_image import GenerateRelightImage, infer_cancel as generate_relight_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''' # flux2 klein @router.post("/generate_image_flux2_klein") async def generate_image_flux2_klein(request_item: GenerateImageFlux2KleinModel): """ 创建一个具有以下参数的请求体: - **bucket_name**: OSS桶名 (必填) - **object_name**: OSS对象名（文件路径）(必填) - **width**: 图片宽度，默认1024像素 (非必填,1024) - **height**: 图片高度，默认1024像素 (非必填,默认1024) - **prompt**: 文本提示词，用于模型推理等场景 (非必填,默认"") - **steps**: 推理步数，控制模型生成过程的迭代次数 (非必填,默认4) - **guidance**: 引导系数，调节提示词对生成结果的影响程度 (非必填,默认 4.0 ) ### 示例参数： ``` { "bucket_name": "aida-users", "object_name": "89/moodboard/5fdc698c-cb9b-4b36-afa9ce4-1-89.png", "prompt": "a single item of sketch of dress, 4k, white background" } ``` ### 输出示例: ``` { "code": 200, "msg": "OK!", "data": { "output_path": "aida-users/89/moodboard/5fdc698c-cb9b-4b36-afa9ce4-1-89.png" } } ``` """ try: logger.info(f"generate_image_flux2_gen_img request: {json.dumps(request_item.model_dump(), indent=4)}") async with httpx.AsyncClient(timeout=120) as client: resp = await client.post( f"http://{settings.FLUX2_GEN_IMG_MODEL_URL}/predict", json=request_item.model_dump(), ) if resp.status_code == 200: result = resp.json() logger.info(f"flux2_gen_img response: {json.dumps(result, indent=4)}") return ResponseModel(data=result) else: error = resp.json() logger.info(f"flux2_gen_img response: {json.dumps(error, indent=4)}") return ResponseModel(data=error, msg="ERROR!", code=500) except Exception as e: logger.warning(f"generate_image_flux2_gen_img Run Exception @@@@@@:{e}") raise HTTPException(status_code=404, detail=str(e)) # sdxl @router.post("/generate_image") def generate_image(request_item: GenerateImageModel, background_tasks: BackgroundTasks): """ 创建一个具有以下参数的请求体: - **tasks_id**: 任务id 用于取消生成任务和获取生成结果 - **prompt**: 想要生成图片的描述词 - **image_url**: 图生图的输入，minio或S3 url 地址 - **mode**: 生成模式，img2img或者txt2img - **category**: 生成图片的类别，sketch print 等等 - **gender**: 生成sketch专用，服装类别 - **version**: 使用模型版本 fast 或者 high 示例参数： 1. txt 2 img { "tasks_id": "bd2cf809-24bc-49a6-91c9-193c6272a52e-2-89", "prompt": "a single item of sketch of dress, 4k, white background", "image_url": "", "mode": "txt2img", "category": "sketch", "gender": "Female", "version": "fast" } 2. img 2 img { "tasks_id": "b861d4fa-5ae3-4a30-9c7a-7ba6bb9aa37b-1-89", "prompt": "a single item of sketch of dress, 4k, white background", "image_url": "aida-collection-element/89/Sketchboard/548da3a2-834f-49a7-b52c-e729c5ab5062.png", "mode": "img2img", "category": "sketch", "gender": "Female", "version": "fast" } """ try: logger.info(f"generate_image request item is : @@@@@@:{json.dumps(request_item.dict(), indent=4)}") service = GenerateImage(request_item) background_tasks.add_task(service.get_result) except Exception as e: logger.warning(f"generate_image Run Exception @@@@@@:{e}") raise HTTPException(status_code=404, detail=str(e)) return ResponseModel() @router.get("/generate_cancel/{tasks_id}") def generate_image(tasks_id: str): try: logger.info(f"generate_cancel request item is : @@@@@@:{tasks_id}") data = generate_image_infer_cancel(tasks_id) logger.info(f"generate_cancel response @@@@@@:{data}") except Exception as e: logger.warning(f"generate_cancel Run Exception @@@@@@:{e}") raise HTTPException(status_code=404, detail=str(e)) return ResponseModel(data=data['data']) '''multi view 停用''' # @router.post("/generate_multi_view") # def generate_multi_view(request_item: GenerateMultiViewModel, background_tasks: BackgroundTasks): # """ # 创建一个具有以下参数的请求体: # - **tasks_id**: 任务id 用于取消生成任务和获取生成结果 # - **image_url**: 前视角图的输入，minio或S3 url 地址 # # 示例参数： # { # "tasks_id": "123-89", # "image_url": "aida-collection-element/87/Printboard/842c09cf-7297-42d9-9e6e-9c17d4a13cb5.jpg" # } # """ # try: # logger.info(f"generate_multi_view request item is : @@@@@@:{json.dumps(request_item.dict(),indent=4)}") # service = GenerateMultiView(request_item) # background_tasks.add_task(service.get_result) # except Exception as e: # logger.warning(f"generate_multi_view Run Exception @@@@@@:{e}") # raise HTTPException(status_code=404, detail=str(e)) # return ResponseModel() # @router.get("/generate_multi_view_cancel/{tasks_id}") # def generate_multi_view(tasks_id: str): # try: # logger.info(f"generate_cancel request item is : @@@@@@:{tasks_id}") # data = generate_multi_view_cancel(tasks_id) # logger.info(f"generate_cancel response @@@@@@:{data}") # except Exception as e: # logger.warning(f"generate_cancel Run Exception @@@@@@:{e}") # raise HTTPException(status_code=404, detail=str(e)) # return ResponseModel(data=data['data']) '''single logo''' @router.post("/generate_single_logo") def generate_single_logo(request_item: GenerateSingleLogoImageModel, background_tasks: BackgroundTasks): """ 创建一个具有以下参数的请求体: - **tasks_id**: 任务id 用于取消生成任务和获取生成结果 - **prompt**: 想要生成图片的描述词 - **seed**: 固定的prompt和固定的seed 每次的生成结果都是一样的 示例参数： { "tasks_id": "123-89", "prompt": "an apple", "seed": "2" } """ try: logger.info(f"generate_single_logo request item is : @@@@@@:{json.dumps(request_item.dict(), indent=4)}") service = GenerateSingleLogoImage(request_item) background_tasks.add_task(service.get_result) except Exception as e: logger.warning(f"generate_single_logo Run Exception @@@@@@:{e}") raise HTTPException(status_code=404, detail=str(e)) return ResponseModel() @router.get("/generate_single_logo_cancel/{tasks_id}") def generate_single_logo_image(tasks_id: str): try: logger.info(f"generate_single_logo_cancel request item is : @@@@@@:{tasks_id}") data = generate_single_logo_cancel(tasks_id) logger.info(f"generate_single_logo_cancel response @@@@@@:{data}") except Exception as e: logger.warning(f"generate_single_logo_cancel Run Exception @@@@@@:{e}") raise HTTPException(status_code=404, detail=str(e)) return ResponseModel(data=data['data']) """slogan """ @router.post("/generate_slogan") async def generate_slogan(request_data: GenerateSloganImageModel): """ ### 请求体示例: ```json { "num_point": 16, "image_url": "aida-slogan/6886785f-0aac-4052-b6fd-7ae20a841d8d.png", "prompt": "123", "tasks_id": "string-89" } ``` """ try: logger.info(f"generate_slogan request item is : @@@@@@:{json.dumps(request_data.dict(), indent=4)}") data = requests.post(f"http://{settings.A6000_SERVICE_HOST}:10020/api/slogan", json=request_data.dict()) logger.info(f"generate_slogan response @@@@@@:{json.dumps(json.loads(data.content), indent=4)}") return ResponseModel(data=json.loads(data.content)) except Exception as e: logger.warning(f"generate_slogan Run Exception @@@@@@:{e}") """product image flux2.0""" # @router.post("/img_to_product") # async def img_to_product(request_data: Flux2ToProductImgModel): # """ # 创建一个具有以下参数的请求体: # - **tasks_id**: 任务id 用于取消生成任务和获取生成结果 # - **prompt**: 想要生成图片的描述词 # - **image_path**: 被生成图片的S3或minio url地址 # - **infer_step**: 推理步数 # # ### 请求体示例: # ```json # point # { # "prompt": "Create realistic studio photo with real people model standing and wearing this garment, in white studio, Keep original model if present, or generate appropriate model, Standing pose, facing camera.", # "image_path":"aida-results/result_38151e0a-f83b-11f0-89f6-0242ac130002.png", # "infer_step":4, # "tasks_id":"123456-123" # } # ``` # """ # try: # logger.info(f"img_to_product request item is : @@@@@@:{json.dumps(request_data.dict(), indent=4)}") # data = requests.post(f"http://{settings.A6000_SERVICE_HOST}:10090/api/v1/to_product", json=request_data.dict()) # logger.info(f"img_to_product response @@@@@@:{json.dumps(json.loads(data.content), indent=4)}") # return ResponseModel(data=json.loads(data.content)) # except Exception as e: # logger.warning(f"img_to_product Run Exception @@@@@@:{e}") '''product image''' @router.post("/generate_product_image") def generate_product_image(request_item: GenerateProductImageModel, background_tasks: BackgroundTasks): """ 创建一个具有以下参数的请求体: - **tasks_id**: 任务id 用于取消生成任务和获取生成结果 - **prompt**: 想要生成图片的描述词 - **image_url**: 被生成图片的S3或minio url地址 - **image_strength**: 生成强度，越低越接近原图 - **product_type**: 输入single item 还是 overall item 示例参数： { "tasks_id": "123-89", "prompt": "the best quality, masterpiece. detailed, high-res, simple background, studio photography, extremely detailed, updo, detailed face, face, close-up, HDR, UHD, 8K realistic, Highly detailed, simple background, Studio lighting", "image_url": "aida-results/result_00097282-ebb2-11ee-a822-b48351119060.png", "image_strength": 0.8, "product_type": "overall" } """ try: logger.info(f"generate_product_image request item is : @@@@@@:{json.dumps(request_item.dict(), indent=4)}") service = GenerateProductImage(request_item) background_tasks.add_task(service.get_result) except Exception as e: logger.warning(f"generate_product_image Run Exception @@@@@@:{e}") raise HTTPException(status_code=404, detail=str(e)) return ResponseModel() @router.get("/generate_product_image_cancel_cancel/{tasks_id}") def generate_product_image(tasks_id: str): try: logger.info(f"generate_product_image_cancel_cancel request item is : @@@@@@:{tasks_id}") data = generate_product_image_cancel(tasks_id) logger.info(f"generate_product_image_cancel_cancel response @@@@@@:{data}") except Exception as e: logger.warning(f"generate_product_image_cancel_cancel Run Exception @@@@@@:{e}") raise HTTPException(status_code=404, detail=str(e)) return ResponseModel(data=data['data']) '''relight image 停用''' # @router.post("/generate_relight_image") # def generate_relight_image(request_item: GenerateRelightImageModel, background_tasks: BackgroundTasks): # """ # 创建一个具有以下参数的请求体: # - **tasks_id**: 任务id 用于取消生成任务和获取生成结果 # - **prompt**: 想要生成图片的描述词 # - **image_url**: 被生成图片的S3或minio url地址 # - **direction**: 光源方向 Right Light Left Light Top Light Bottom Light # - **product_type**: 输入single item 还是 overall item # # # 示例参数： # { # "tasks_id": "123-89", # "prompt": "beautiful woman, detailed face, sunshine, outdoor, warm atmosphere", # "image_url": "aida-results/result_0000b606-1902-11ef-9424-0242ac180002.png", # "direction": "Right Light", # "product_type": "overall" # } # """ # try: # logger.info(f"generate_relight_image request item is : @@@@@@:{json.dumps(request_item.dict(),indent=4)}") # service = GenerateRelightImage(request_item) # background_tasks.add_task(service.get_result) # except Exception as e: # logger.warning(f"generate_relight_image Run Exception @@@@@@:{e}") # raise HTTPException(status_code=404, detail=str(e)) # return ResponseModel() # # # @router.get("/generate_relight_image_cancel_cancel/{tasks_id}") # def generate_relight_image(tasks_id: str): # try: # logger.info(f"generate_relight_image_cancel_cancel request item is : @@@@@@:{tasks_id}") # data = generate_relight_image_cancel(tasks_id) # logger.info(f"generate_relight_image_cancel_cancel response @@@@@@:{data}") # except Exception as e: # logger.warning(f"generate_relight_image_cancel_cancel Run Exception @@@@@@:{e}") # raise HTTPException(status_code=404, detail=str(e)) # return ResponseModel(data=data['data']) """batch generate img 停用""" # @router.post("/batch_generate_product_image") # async def batch_generate_product(request_batch_item: BatchGenerateProductImageModel): # """ # 创建一个具有以下参数的请求体: # - **tasks_id**: 任务id 用于获取生成结果 # - **prompt**: 想要生成图片的描述词 # - **image_url**: 被生成图片的S3或minio url地址 # - **image_strength**: 生成强度，越低越接近原图 # - **product_type**: 输入single item 还是 overall item # - **batch_size**: 批生成数量 # # # 示例参数： # { # "tasks_id": "123-89", # "prompt": "the best quality, masterpiece. detailed, high-res, simple background, studio photography, extremely detailed, updo, detailed face, face, close-up, HDR, UHD, 8K realistic, Highly detailed, simple background, Studio lighting", # "image_url": "aida-results/result_00097282-ebb2-11ee-a822-b48351119060.png", # "image_strength": 0.8, # "product_type": "overall", # "batch_size": 1 # } # """ # return await start_product_batch_generate(request_batch_item) # # # @router.post("/batch_generate_relight_image") # async def batch_generate_relight(request_batch_item: BatchGenerateRelightImageModel): # """ # 创建一个具有以下参数的请求体: # - **tasks_id**: 任务id 用于获取生成结果 # - **prompt**: 想要生成图片的描述词 # - **image_url**: 被生成图片的S3或minio url地址 # - **direction**: 光源方向 Right Light Left Light Top Light Bottom Light # - **product_type**: 输入single item 还是 overall item # - **batch_size**: 批生成数量 # # # 示例参数： # { # "tasks_id": "123-89", # "prompt": "beautiful woman, detailed face, sunshine, outdoor, warm atmosphere", # "image_url": "aida-results/result_0000b606-1902-11ef-9424-0242ac180002.png", # "direction": "Right Light", # "product_type": "overall", # "batch_size": 1 # } # """ # return await start_relight_batch_generate(request_batch_item) # @router.post("/batch_generate_pose_transform_image") # async def batch_generate_pose_transform(request_batch_item: BatchPoseTransformModel): # """ # 创建一个具有以下参数的请求体: # - **tasks_id**: 任务id 用于取消生成任务和获取生成结果 # - **image_url**: 被生成图片的S3或minio url地址 # - **pose_id**: 1 # - **batch_size**: 批生成数量 # # # 示例参数： # { # "tasks_id": "123-89", # "image_url": "aida-results/result_0000b606-1902-11ef-9424-0242ac180002.png", # "pose_id": "1", # "batch_size": 1 # } # """ # return await start_pose_transform_batch_generate(request_batch_item) # # # """agent tool""" # # # @router.post("/agent_tool_generate_image") # def agent_tool_generate_image(request_item: AgentTollGenerateImageModel): # """ # 创建一个具有以下参数的请求体: # - **prompt**: 想要生成图片的描述词 # - **category**: 生成图片的类别，sketch print 等等 # - **gender**: 生成sketch专用，服装类别 # - **version**: 使用模型版本 fast 或者 high # - **size**: 生成数量 # - **version**: 使用模型版本 fast 或者 high # # # 示例参数： # { # "prompt": "a single item of sketch of Wabi-sabi, skirt, tiered, 4k, white background", # "category": "sketch", # "gender": "male", # "size":2, # "version":"high" # } # """ # try: # logger.info(f"agent_tool_generate_image request item is : @@@@@@:{request_item.dict()}") # request_data = request_item.dict() # service = AgentToolGenerateImage(request_data['version']) # image_url_list, clothing_category_list = service.get_result( # prompt=request_data['prompt'], # size=request_data['size'], # version=request_data['version'], # category=request_data['category'], # gender=request_data['gender'] # ) # data = { # "image_url_list": image_url_list, # "clothing_category_list": clothing_category_list # } # logger.info(f"agent_tool_generate_image response item is : @@@@@@:{data}") # except Exception as e: # logger.warning(f"agent_tool_generate_image Run Exception @@@@@@:{e}") # raise HTTPException(status_code=404, detail=str(e)) # return ResponseModel(data=data)