# src/server/canvas_generate_3D/tasks.py import asyncio from celery import shared_task import httpx from src.core.config import settings from src.server.utils.mq_util import send_to_rabbitmq import logging logger = logging.getLogger(__name__) @shared_task(bind=True, queue="img_to_3d_queue", max_retries=3, name='src.server.canvas_generate_3D.tasks.img_to_3d_task') def img_to_3d_task(self, input_images: list, model: str = "single"): """img_to_3D 主任务""" job_id = self.request.id logger.info(f"开始处理 img_to_3D 任务 | job_id: {job_id}") try: input_data = { "image_paths": input_images, "model": model, } with httpx.Client(timeout=300.0) as client: resp = client.post( f"http://{settings.IMAGE_TO_3D_MODEL_URL}/canvas/img_to_3D", json=input_data ) resp.raise_for_status() result = resp.json() logger.info(f"img_to_3D 任务处理完成 | job_id: {job_id}") # 发送到对应的结果队列 asyncio.run(send_to_rabbitmq( result=result, job_id=job_id, status="completed", routing_key=f"img_to_3d_results-{settings.SERVE_ENV}" # ← 第一个任务的结果队列 )) return result except Exception as exc: logger.error(f"img_to_3D 任务失败 | job_id: {job_id}", exc_info=True) asyncio.run(send_to_rabbitmq( result={"error": str(exc)}, job_id=job_id, status="failed", routing_key=f"img_to_3d_results-{settings.SERVE_ENV}" )) raise self.retry(exc=exc, countdown=60, max_retries=3) @shared_task(bind=True, queue="three_d_to_3views_queue", max_retries=3, name='src.server.canvas_generate_3D.tasks.three_d_to_3views_task') def three_d_to_3views_task(self, minio_glb_path: str): """3D to 3views 主任务""" job_id = self.request.id logger.info(f"开始处理 three_d_to_3views_task | job_id: {job_id}") try: input_data = { "minio_glb_path": minio_glb_path, } with httpx.Client(timeout=300.0) as client: resp = client.post( f"http://{settings.IMAGE_TO_3D_MODEL_URL}/canvas/3d_to_3views", json=input_data ) resp.raise_for_status() result = resp.json() logger.info(f"three_d_to_3views_task 任务处理完成 | job_id: {job_id}") # 发送到对应的结果队列 asyncio.run(send_to_rabbitmq( result=result, job_id=job_id, status="completed", routing_key="three_d_to_3views_results" # ← 第二个任务的结果队列 )) return result except Exception as exc: logger.error(f"three_d_to_3views_task 任务失败 | job_id: {job_id}", exc_info=True) asyncio.run(send_to_rabbitmq( result={"error": str(exc)}, job_id=job_id, status="failed", routing_key="three_d_to_3views_results" )) raise self.retry(exc=exc, countdown=60, max_retries=3)