long_articles_agentic_supply/supply/jobs/task_utils.py

"""向后兼容 shim —— 委托到 supply.domain.task.exceptions

新代码请直接使用:
    from supply.domain.task import TaskError, TaskValidationError, ...
    TaskUtils 工具方法后续迁移到 domain/task/utils.py
"""

import traceback
from typing import Any

from supply.domain.task.exceptions import (  # noqa: F401
    TaskError,
    TaskValidationError,
    TaskTimeoutError,
    TaskConcurrencyError,
    TaskLockError,
)


class TaskUtils:
    """任务工具方法（向后兼容）"""

    @staticmethod
    def validate_table_name(table_name: str) -> str:
        cleaned = table_name.strip().replace(" ", "_")
        if not cleaned or "--" in cleaned or ";" in cleaned.lower():
            raise TaskValidationError(f"非法的表名: {table_name}")
        return cleaned

    @staticmethod
    def validate_task_name(task_name: Any) -> str:
        task_name = str(task_name).strip()
        if not task_name:
            raise TaskValidationError("task_name 不能为空")
        return task_name

    @staticmethod
    def format_error_detail(e: Exception) -> str:
        return "".join(
            traceback.format_exception(type(e), e, e.__traceback__)
        )[-1000:]