Mutual_Fund/nl2sql/job_history.py

"""NL2SQL 运维任务执行历史适配器。"""
from __future__ import annotations

import json
from datetime import datetime

from sqlalchemy import text

from nl2sql.audit import _sanitize
from utils.pagination import normalize_pagination, pagination_result


async def record_job_history(db, result, *, elapsed_ms: float, parameter_summary: dict | None = None) -> None:
    """保存任务结果摘要，不保存原始参数和连接凭据。"""
    statement = text(
        """
        INSERT INTO nl2sql_job_history
            (job_name, status, attempts, detail, error_type, parameter_summary, elapsed_ms, create_time)
        VALUES
            (:job_name, :status, :attempts, :detail, :error_type, :parameter_summary, :elapsed_ms, :create_time)
        """
    )
    await db.execute(
        statement,
        {
            "job_name": result.name,
            "status": result.status,
            "attempts": result.attempts,
            "detail": json.dumps(_sanitize(result.detail or {}), ensure_ascii=False),
            "error_type": result.error_type,
            "parameter_summary": json.dumps(_sanitize(parameter_summary or {}), ensure_ascii=False),
            "elapsed_ms": elapsed_ms,
            "create_time": datetime.now(),
        },
    )
    await db.commit()


async def record_job_history_safely(db, result, *, elapsed_ms: float, parameter_summary: dict | None = None) -> bool:
    """任务历史写入失败时降级，不影响任务结果返回。"""
    try:
        await record_job_history(
            db,
            result,
            elapsed_ms=elapsed_ms,
            parameter_summary=parameter_summary,
        )
    except Exception:  # noqa: BLE001 历史故障不能阻断运维任务
        return False
    return True


async def list_job_history(db, *, page: int = 1, page_size: int = 10, status: str | None = None) -> dict:
    """分页查询任务历史，只返回执行摘要，不返回 detail 明细。"""
    page, page_size, offset = normalize_pagination(page, page_size)
    conditions = "WHERE (:status IS NULL OR status = :status)"
    count_result = await db.execute(
        text(f"SELECT COUNT(*) AS total FROM nl2sql_job_history {conditions}"),
        {"status": status},
    )
    total = int(count_result.scalar() or 0)
    statement = text(
        f"""
        SELECT id, job_name, status, attempts, error_type, elapsed_ms, create_time
        FROM nl2sql_job_history
        {conditions}
        ORDER BY id DESC
        LIMIT :limit OFFSET :offset
        """
    )
    result = await db.execute(
        statement,
        {
            "status": status,
            "limit": page_size,
            "offset": offset,
        },
    )
    return pagination_result(
        [dict(row) for row in result.mappings().all()],
        total,
        page=page,
        page_size=page_size,
    )