| 1 | import logging |
| 2 | import traceback |
| 3 | |
| 4 | from . import action_transfer, digital_human, standard |
| 5 | from .storage import mark_completed, mark_failed, mark_running, update_task |
| 6 | |
| 7 | logger = logging.getLogger(__name__) |
| 8 | |
| 9 | PIPELINE_REGISTRY = { |
| 10 | "standard": standard.run, |
| 11 | "quick_create": standard.run, |
| 12 | "action_transfer": action_transfer.run, |
| 13 | "digital_human": digital_human.run, |
| 14 | } |
| 15 | |
| 16 | |
| 17 | async def run_pipeline_task(task_id: str, pipeline: str, params: dict) -> None: |
| 18 | runner = PIPELINE_REGISTRY[pipeline] |
| 19 | try: |
| 20 | logger.info("Pipeline task started: task_id=%s pipeline=%s", task_id, pipeline) |
| 21 | mark_running(task_id) |
| 22 | output, artifacts = await runner(task_id, params) |
| 23 | mark_completed(task_id, output=output, artifacts=artifacts) |
| 24 | logger.info( |
| 25 | "Pipeline task completed: task_id=%s pipeline=%s artifacts=%d", |
| 26 | task_id, |
| 27 | pipeline, |
| 28 | len(artifacts or []), |
| 29 | ) |
| 30 | except Exception as exc: |
| 31 | logger.exception("Pipeline task failed: task_id=%s pipeline=%s", task_id, pipeline) |
| 32 | update_task(task_id, progress=0) |
| 33 | mark_failed(task_id, f"{exc}\n{traceback.format_exc()}") |
| 34 |