diff --git a/fam-core/src/fam_core/db_layer.py b/fam-core/src/fam_core/db_layer.py index a19671c..1c5f9d0 100644 --- a/fam-core/src/fam_core/db_layer.py +++ b/fam-core/src/fam_core/db_layer.py @@ -92,6 +92,9 @@ def get_tasks_by_status(status: str, limit=10) -> List[Dict]: def update_task_status(task_id: int, status: str, error_message: str = None, failure_stage: str = None): """更新任务状态""" + valid_stages = {'download', 'extract', 'vlm_visual', 'vlm_fusion', 'callback'} + if failure_stage and failure_stage not in valid_stages: + failure_stage = 'callback' conn = get_conn() try: cursor = conn.cursor() diff --git a/fam-core/src/fam_core/dispatcher/dispatcher.py b/fam-core/src/fam_core/dispatcher/dispatcher.py index 34ea3b6..143dd59 100644 --- a/fam-core/src/fam_core/dispatcher/dispatcher.py +++ b/fam-core/src/fam_core/dispatcher/dispatcher.py @@ -165,7 +165,10 @@ class Dispatcher: tasks = db_layer.get_pending_tasks(limit=10) for task in tasks: if self._should_retry(task): - self._dispatch_one(task) + try: + self._dispatch_one(task) + except Exception as e: + logger.error(f"[task_id={task['task_id']}] dispatch 异常: {e}", exc_info=True) def _run(self): """线程主循环"""