"""LeAudit 任务入口。"""

from __future__ import annotations

import asyncio
import os
from pathlib import Path
import tempfile
import time
from typing import Any, Dict, Optional

import fitz
from fastapi_common.fastapi_common_logger import logger
from leaudit.converters import doc2pdf
from leaudit.ocr.chandra_client import ChandraOCRError
from sqlalchemy import select

from fastapi_admin.celery_app import celery_app
from fastapi_admin.config import (
    LEAUDIT_RULES_DIR,
    LEAUDIT_STUCK_TIMEOUT_MINUTES,
    LEAUDIT_WORKER_QUEUE_NORMAL,
    LEAUDIT_WORKER_QUEUE_URGENT,
)
from fastapi_modules.fastapi_leaudit.leaudit_bridge.nativeRunner import (
    NativeRunRequest,
    NativeRunner,
)
from fastapi_modules.fastapi_leaudit.leaudit_bridge.auditCtxBuilder import (
    NativeAuditMetadata,
)
from fastapi_modules.fastapi_leaudit.leaudit_bridge.fileSourceResolver import (
    FileSourceResolver,
)
from fastapi_modules.fastapi_leaudit.leaudit_bridge.ruleVersionResolver import (
    RuleVersionResolver,
)
from fastapi_modules.fastapi_leaudit.leaudit_bridge.rules_loader import RulesLoader
from fastapi_modules.fastapi_leaudit.leaudit_bridge.storage_adapter import StorageAdapter
from fastapi_modules.fastapi_leaudit.models import (
    LeauditAuditRun,
    LeauditDocument,
    LeauditDocumentFile,
)

log = logger


def _classify_run_failure(exc: Exception) -> tuple[str, str, dict[str, Any]]:
    """将底层异常归一化为可落库、可展示的失败信息。"""
    raw_message = str(exc).strip() or exc.__class__.__name__
    detail: dict[str, Any] = {
        "rawMessage": raw_message,
        "errorType": type(exc).__name__,
    }

    if isinstance(exc, ChandraOCRError):
        lower_message = raw_message.lower()
        detail["service"] = "ocr"
        if "all connection attempts failed" in lower_message or "connect" in lower_message:
            return (
                "OCR_SERVICE_UNAVAILABLE",
                "OCR服务暂时不可用，文档未完成识别，请稍后重试。",
                {**detail, "reason": "connection_failed"},
            )
        if "timeout" in lower_message:
            return (
                "OCR_SERVICE_TIMEOUT",
                "OCR服务处理超时，文档未完成识别，请稍后重试。",
                {**detail, "reason": "timeout"},
            )
        if "ocr api returned" in lower_message:
            return (
                "OCR_SERVICE_BAD_RESPONSE",
                "OCR服务响应异常，文档未完成识别，请稍后重试。",
                {**detail, "reason": "bad_status"},
            )
        return (
            "OCR_PROCESSING_FAILED",
            "OCR处理失败，文档未完成识别，请稍后重试。",
            {**detail, "reason": "ocr_failed"},
        )

    return (
        "AUDIT_RUN_FAILED",
        raw_message,
        detail,
    )


def leaudit_process_document(
    document_id: int,
    file_content: bytes,
    filename: str,
    upload_info: Optional[Dict[str, Any]] = None,
    source_port: Optional[int] = None,
    rules_path: Optional[str] = None,
):
    """处理单个文档的 LeAudit 任务。"""
    task_id = os.urandom(8).hex()
    log.info(f"[任务ID: {task_id}] leaudit管线开始处理: {filename}")

    # 新平台：region 通过参数传递，不再依赖 os.environ 切换
    if source_port:
        log.info(f"[任务ID: {task_id}] 来源端口: {source_port}")

    if upload_info is None:
        upload_info = {}

    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    temp_paths: list[str] = []
    storage = StorageAdapter()

    try:
        run_id = _resolve_run_id(document_id, upload_info, loop)
        rules_resolution = _resolve_rules_runtime(document_id, run_id, rules_path, loop)
        loop.run_until_complete(_update_run_status_safe(run_id, "running"))
        rules_path_resolved = rules_resolution["rules_path"]
        attachment_inputs = list((upload_info or {}).get("attachments") or [])

        rules_file = None
        if rules_path_resolved:
            loader = RulesLoader()
            rules_file = loader.load(rules_path_resolved)
            temp_rule_path = rules_resolution.get("temp_rule_path")
            if isinstance(temp_rule_path, str):
                temp_paths.append(temp_rule_path)
            log.info(
                f"[任务ID: {task_id}] RulesFile pre-loaded: {rules_path_resolved} "
                f"({len(rules_file.flat_rules)} rules, {len(rules_file.flat_extract)} fields)"
            )
        else:
            log.info(
                f"[任务ID: {task_id}] No fixed rules_path — "
                "will classify from document content after OCR"
            )

        attachment_metadata: list[dict[str, Any]] = []
        for attachment in attachment_inputs:
            attachment_metadata.append(
                {
                    "fileId": attachment.get("file_id"),
                    "fileName": attachment.get("file_name") or attachment.get("filename"),
                    "sourceType": attachment.get("source_type"),
                    "sourcePath": attachment.get("source_path"),
                }
            )

        temp_path, merged_attachment_metadata = _prepare_execution_input(
            filename=filename,
            file_content=file_content,
            attachments=attachment_inputs,
            temp_paths=temp_paths,
        )
        if merged_attachment_metadata:
            for item in attachment_metadata:
                for merged in merged_attachment_metadata:
                    if item["fileId"] == merged["fileId"]:
                        item["localPath"] = merged["localPath"]
                        item["mergedPdfPath"] = merged["mergedPdfPath"]
                        break
            log.info("[任务ID: %s] 已装配并合并附件 %s 个到单输入文件", task_id, len(merged_attachment_metadata))

        runner = NativeRunner()

        t0 = time.time()
        native_result = loop.run_until_complete(
            runner.run(
                NativeRunRequest(
                    metadata=NativeAuditMetadata(
                        run_id=run_id,
                        document_id=document_id,
                        rule_version_id=_optional_int(upload_info, "rule_version_id", "ruleVersionId"),
                        extras={
                            "taskId": task_id,
                            "attachments": attachment_metadata,
                            "mergedInput": bool(merged_attachment_metadata),
                        },
                    ),
                    local_file_path=temp_path,
                    rules_file=rules_file,
                    rule_source_path=rules_path_resolved,
                )
            )
        )
        loop.run_until_complete(runner.persist_result(native_result))
        elapsed = round(time.time() - t0, 2)

        ctx = native_result.ctx
        loop.run_until_complete(_update_run_phase_safe(run_id, ctx.phase))
        loop.run_until_complete(_update_run_status_safe(run_id, "completed"))
        loop.run_until_complete(_update_status_safe(document_id, "completed"))
        log.info(
            f"[任务ID: {task_id}] leaudit管线完成: phase={ctx.phase}, "
            f"timing={dict(ctx.timing)}, 总耗时={elapsed:.1f}s"
        )

        return {
            "status": "success",
            "document_id": document_id,
            "run_id": run_id,
            "phase": ctx.phase,
            "timing": dict(ctx.timing),
            "errors": list(ctx.extraction.all_errors) if ctx.extraction is not None else list(ctx.extraction_errors),
        }

    except Exception as e:
        log.error(f"[任务ID: {task_id}] leaudit管线失败: {e}", exc_info=True)
        error_code, user_message, error_detail = _classify_run_failure(e)
        try:
            loop.run_until_complete(_update_status_safe(document_id, "failed"))
            if 'run_id' in locals():
                failed_phase = "persist"
                if "native_result" in locals():
                    failed_phase = native_result.ctx.phase or failed_phase
                loop.run_until_complete(
                    storage.fail_run(
                        document_id,
                        run_id=run_id,
                        phase=failed_phase,
                        message=user_message,
                        error_code=error_code,
                        detail_json={
                            "taskId": task_id,
                            "filename": filename,
                            **error_detail,
                        },
                    )
                )
        except Exception:
            pass
        raise

    finally:
        for temp_path in temp_paths:
            try:
                if Path(temp_path).exists():
                    os.remove(temp_path)
            except OSError:
                pass
        loop.close()


def leaudit_process_document_by_run(
    run_id: int,
    *,
    task_id: str | None = None,
    rules_path: str | None = None,
    queue_name: str | None = None,
) -> dict[str, Any]:
    """按 runId 加载执行上下文并执行原生 leaudit。"""
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        claimed = loop.run_until_complete(_claim_run_safe(run_id, task_id))
        if not claimed:
            log.warning("run_id=%s 未抢占成功，跳过重复消费", run_id)
            return {"status": "skipped", "run_id": run_id, "reason": "already_claimed"}

        context = loop.run_until_complete(_load_run_context(run_id))
        log.info(
            "run_id=%s worker开始执行: queue=%s, speed=%s, filename=%s",
            run_id,
            queue_name or resolve_worker_queue(context.get("trigger_source")),
            _queue_label(queue_name or resolve_worker_queue(context.get("trigger_source"))),
            context["filename"],
        )
        return leaudit_process_document(
            document_id=context["document_id"],
            file_content=context["file_content"],
            filename=context["filename"],
            upload_info={
                "run_id": run_id,
                "rule_version_id": context["rule_version_id"],
                "rule_source_oss_url": context["rule_source_oss_url"],
                "source_type": context["source_type"],
                "source_path": context["source_path"],
                "trigger_source": context["trigger_source"],
                "attachments": context.get("attachments") or [],
            },
            rules_path=rules_path,
        )
    finally:
        loop.close()


@celery_app.task(
    bind=True,
    name="leaudit.process_document",
    acks_late=True,
)
def leaudit_process_document_task(self, run_id: int, rules_path: str | None = None) -> dict[str, Any]:
    """Celery worker 入口 —— 按 runId 执行评查。"""
    delivery_info = getattr(self.request, "delivery_info", {}) or {}
    queue_name = delivery_info.get("routing_key") or delivery_info.get("queue")
    return leaudit_process_document_by_run(
        run_id=run_id,
        task_id=self.request.id,
        rules_path=rules_path,
        queue_name=queue_name,
    )


@celery_app.task(
    bind=True,
    name="leaudit.scan_stuck_documents",
)
def leaudit_scan_stuck_documents_task(self) -> dict[str, Any]:
    """周期扫描长时间无进展的评查文档，并自动标记失败。"""
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        return loop.run_until_complete(
            _scan_and_fail_stuck_documents(timeout_minutes=max(1, int(LEAUDIT_STUCK_TIMEOUT_MINUTES)))
        )
    finally:
        loop.close()


# type_id → rules directory mapping (only fixed-mapping types)
# 行政许可 (type_id=2) has 9 sub-types, NOT mapped here —
# must come from document metadata (rules_file_path) or content classification.
_TYPE_ID_RULES_MAP: dict[int, str] = {
    3: "行政处罚",
}


def _resolve_rules_path(document_id: int, loop: asyncio.AbstractEventLoop) -> str | None:
    """Resolve rules_path: config override → document metadata → type_id mapping."""
    from fastapi_common.fastapi_common_sqlalchemy.database import GetAsyncSession
    from sqlalchemy import text as sa_text

    # 1. Config override (when explicitly set in app.toml)
    if LEAUDIT_RULES_DIR:
        return LEAUDIT_RULES_DIR

    try:
        async def _fetch():
            async with GetAsyncSession() as session:
                result = await session.execute(
                    sa_text("SELECT type_id FROM leaudit_documents WHERE id = :did"),
                    {"did": document_id},
                )
                row = result.fetchone()
                if row and row[0] and row[0] in _TYPE_ID_RULES_MAP:
                    return f"{_TYPE_ID_RULES_MAP[row[0]]}/rules.yaml"
                return None

        return loop.run_until_complete(_fetch())
    except Exception as e:
        log.warning(f"Failed to resolve rules_path from document: {e}")

    return None


def _resolve_rules_runtime(
    document_id: int,
    run_id: int,
    explicit_rules_path: str | None,
    loop: asyncio.AbstractEventLoop,
) -> dict[str, str | None]:
    """解析本次执行使用的规则来源。"""
    if explicit_rules_path:
        return {
            "rules_path": explicit_rules_path,
            "temp_rule_path": None,
            "source_type": "explicit",
            "source_path": explicit_rules_path,
        }

    resolver = RuleVersionResolver()
    try:
        payload = loop.run_until_complete(resolver.ResolveForRun(run_id))
        if payload:
            log.info(
                f"run_id={run_id} 规则来源已解析: sourceType={payload.sourceType}, "
                f"sourcePath={payload.sourcePath}, localPath={payload.localPath}"
            )
            return {
                "rules_path": payload.localPath,
                "temp_rule_path": payload.localPath if payload.sourceType == "oss" else None,
                "source_type": payload.sourceType,
                "source_path": payload.sourcePath,
            }
    except Exception as e:
        log.warning(f"Failed to resolve rule version from run: run_id={run_id}, error={e}")

    fallback_rules_path = _resolve_rules_path(document_id, loop)
    return {
        "rules_path": fallback_rules_path,
        "temp_rule_path": None,
        "source_type": "legacy_fallback",
        "source_path": fallback_rules_path,
    }


def _resolve_run_id(
    document_id: int,
    upload_info: Dict[str, Any] | None,
    loop: asyncio.AbstractEventLoop,
) -> int:
    """解析本次任务对应的运行 ID。"""
    if upload_info:
        for key in ("run_id", "runId"):
            value = upload_info.get(key)
            if isinstance(value, int):
                return value

    from fastapi_common.fastapi_common_sqlalchemy.database import GetAsyncSession
    from sqlalchemy import text as sa_text

    async def _fetch() -> int:
        async with GetAsyncSession() as session:
            result = await session.execute(
                sa_text("SELECT id FROM leaudit_audit_runs WHERE document_id = :did ORDER BY id DESC LIMIT 1"),
                {"did": document_id},
            )
            row = result.fetchone()
            if not row:
                raise ValueError(f"未找到 document_id={document_id} 对应的 run 记录")
            return int(row[0])

    return loop.run_until_complete(_fetch())


def _optional_int(payload: Dict[str, Any] | None, *keys: str) -> int | None:
    """从字典中按顺序取可用整数。"""
    if not payload:
        return None

    for key in keys:
        value = payload.get(key)
        if isinstance(value, int):
            return value
    return None


async def _update_status_safe(document_id: int, status: str) -> None:
    """Safely update document status via SQLAlchemy, ignoring errors."""
    try:
        from fastapi_common.fastapi_common_sqlalchemy.database import GetAsyncSession
        from sqlalchemy import text as sa_text

        async with GetAsyncSession() as session:
            await session.execute(
                sa_text("UPDATE leaudit_documents SET processing_status = :s, updated_at = now() WHERE id = :did"),
                {"s": status, "did": document_id},
            )
            await session.commit()
    except Exception:
        pass


async def _update_run_status_safe(run_id: int, status: str) -> None:
    """安全更新运行状态。"""
    try:
        from fastapi_common.fastapi_common_sqlalchemy.database import GetAsyncSession
        from sqlalchemy import text as sa_text

        async with GetAsyncSession() as session:
            await session.execute(
                sa_text("UPDATE leaudit_audit_runs SET status = :s, updated_at = now() WHERE id = :rid"),
                {"s": status, "rid": run_id},
            )
            await session.commit()
    except Exception:
        pass


async def _update_run_phase_safe(run_id: int, phase: str | None) -> None:
    """安全更新运行阶段。"""
    try:
        from fastapi_common.fastapi_common_sqlalchemy.database import GetAsyncSession
        from sqlalchemy import text as sa_text

        async with GetAsyncSession() as session:
            await session.execute(
                sa_text("UPDATE leaudit_audit_runs SET phase = :p, updated_at = now() WHERE id = :rid"),
                {"p": phase, "rid": run_id},
            )
            await session.commit()
    except Exception:
        pass


async def _claim_run_safe(run_id: int, task_id: str | None) -> bool:
    """原子抢占 queued/pending 运行，避免重复消费。"""
    try:
        from fastapi_common.fastapi_common_sqlalchemy.database import GetAsyncSession
        from sqlalchemy import text as sa_text

        async with GetAsyncSession() as session:
            result = await session.execute(
                sa_text(
                    """
                    UPDATE leaudit_audit_runs
                    SET status = 'running',
                        phase = 'prepare',
                        task_id = COALESCE(:task_id, task_id),
                        started_at = COALESCE(started_at, now()),
                        updated_at = now()
                    WHERE id = :rid
                      AND status IN ('queued', 'pending', 'retrying')
                    RETURNING id
                    """
                ),
                {"rid": run_id, "task_id": task_id},
            )
            row = result.fetchone()
            await session.commit()
            return row is not None
    except Exception:
        log.exception("run_id=%s 抢占执行权失败", run_id)
        return False


async def _load_run_context(run_id: int) -> dict[str, Any]:
    """按 runId 加载执行所需文档文件上下文。"""
    from fastapi_common.fastapi_common_sqlalchemy.database import GetAsyncSession

    async with GetAsyncSession() as session:
        run = await session.get(LeauditAuditRun, run_id)
        if not run:
            raise ValueError(f"未找到 run_id={run_id} 对应的运行记录")

        document = await session.get(LeauditDocument, run.documentId)
        if not document:
            raise ValueError(f"未找到 document_id={run.documentId} 对应的文档记录")

        document_file = await session.get(LeauditDocumentFile, run.documentFileId)
        if not document_file:
            raise ValueError(f"未找到 document_file_id={run.documentFileId} 对应的文件记录")

        resolver = FileSourceResolver()
        payload = await resolver.ResolvePayload(document_file)
        attachmentResult = await session.execute(
            select(LeauditDocumentFile)
            .where(
                LeauditDocumentFile.documentId == document.Id,
                LeauditDocumentFile.isActive.is_(True),
                LeauditDocumentFile.fileRole == "attachment",
            )
            .order_by(LeauditDocumentFile.Id.asc())
        )
        attachmentFiles = list(attachmentResult.scalars().all())
        attachmentPayloads = await resolver.ResolvePayloads(attachmentFiles) if attachmentFiles else []

        return {
            "document_id": document.Id,
            "filename": payload.fileName,
            "file_content": payload.fileContent,
            "source_type": payload.sourceType,
            "source_path": payload.sourcePath,
            "rule_version_id": run.ruleVersionId,
            "rule_source_oss_url": run.ruleSourceOssUrl,
            "trigger_source": run.triggerSource,
            "attachments": [
                {
                    "file_id": attachmentFile.Id,
                    "file_name": attachmentPayload.fileName,
                    "file_content": attachmentPayload.fileContent,
                    "source_type": attachmentPayload.sourceType,
                    "source_path": attachmentPayload.sourcePath,
                }
                for attachmentFile, attachmentPayload in zip(attachmentFiles, attachmentPayloads)
            ],
        }


def _get_suffix(filename: str) -> str:
    """Extract file suffix from filename."""
    _, ext = os.path.splitext(filename)
    return ext if ext else ".pdf"


def _write_temp_file(*, filename: str, content: bytes, temp_paths: list[str]) -> str:
    """Write bytes to a temp file preserving original suffix."""
    suffix = _get_suffix(filename)
    with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as temp:
        temp.write(content)
        temp_path = temp.name
    temp_paths.append(temp_path)
    return temp_path


def _convert_to_pdf_path(*, source_path: str, temp_paths: list[str]) -> str:
    """Convert a local source file to PDF temp path when needed."""
    source = Path(source_path)
    if source.suffix.lower() == ".pdf":
        return source_path
    pdf_temp = tempfile.NamedTemporaryFile(suffix=".pdf", delete=False)
    pdf_temp.close()
    temp_paths.append(pdf_temp.name)
    doc2pdf.convert(source, pdf_temp.name, soffice="auto", pdfa=False, force=True, verify=False)
    return pdf_temp.name


def _merge_pdf_paths(*, pdf_paths: list[str], temp_paths: list[str]) -> str:
    """Merge many pdf files into one temp pdf."""
    merged_temp = tempfile.NamedTemporaryFile(suffix=".pdf", delete=False)
    merged_temp.close()
    temp_paths.append(merged_temp.name)
    output = fitz.open()
    try:
        for pdf_path in pdf_paths:
            src = fitz.open(pdf_path)
            try:
                output.insert_pdf(src)
            finally:
                src.close()
        output.save(merged_temp.name)
    finally:
        output.close()
    return merged_temp.name


def _prepare_execution_input(
    *,
    filename: str,
    file_content: bytes,
    attachments: list[dict[str, Any]],
    temp_paths: list[str],
) -> tuple[str, list[dict[str, Any]]]:
    """Prepare the actual execution input file.

    Without attachments: use the main file directly.
    With attachments: convert main file and attachments to PDFs, merge into one PDF.
    """
    main_local_path = _write_temp_file(filename=filename, content=file_content, temp_paths=temp_paths)
    if not attachments:
        return main_local_path, []

    main_pdf_path = _convert_to_pdf_path(source_path=main_local_path, temp_paths=temp_paths)
    merged_attachment_metadata: list[dict[str, Any]] = []
    attachment_pdf_paths: list[str] = []
    for attachment in attachments:
        attachment_name = str(attachment.get("file_name") or attachment.get("filename") or "attachment.bin")
        attachment_content = attachment["file_content"]
        attachment_local_path = _write_temp_file(
            filename=attachment_name,
            content=attachment_content,
            temp_paths=temp_paths,
        )
        attachment_pdf_path = _convert_to_pdf_path(source_path=attachment_local_path, temp_paths=temp_paths)
        attachment_pdf_paths.append(attachment_pdf_path)
        merged_attachment_metadata.append(
            {
                "fileId": attachment.get("file_id"),
                "fileName": attachment_name,
                "localPath": attachment_local_path,
                "mergedPdfPath": attachment_pdf_path,
            }
        )

    merged_pdf_path = _merge_pdf_paths(
        pdf_paths=[main_pdf_path, *attachment_pdf_paths],
        temp_paths=temp_paths,
    )
    return merged_pdf_path, merged_attachment_metadata


def dispatch_leaudit_task(
    run_id: int,
    *,
    queue_name: str | None = None,
    rules_path: Optional[str] = None,
) -> str:
    """投递 runId 到 Celery worker 队列。"""
    target_queue = queue_name or LEAUDIT_WORKER_QUEUE_NORMAL
    task = leaudit_process_document_task.apply_async(
        kwargs={"run_id": run_id, "rules_path": rules_path},
        queue=target_queue,
    )
    log.info(
        "run_id=%s 已投递到 worker 队列: queue=%s, speed=%s, task_id=%s",
        run_id,
        target_queue,
        _queue_label(target_queue),
        task.id,
    )
    return task.id


def resolve_worker_queue(trigger_source: str | None) -> str:
    """按触发来源选择 worker 队列。"""
    normalized = (trigger_source or "").strip().lower()
    if "urgent" in normalized or "high" in normalized:
        return LEAUDIT_WORKER_QUEUE_URGENT
    return LEAUDIT_WORKER_QUEUE_NORMAL


def _queue_label(queue_name: str | None) -> str:
    """Map queue name to a user-facing speed label for logs."""
    if queue_name == LEAUDIT_WORKER_QUEUE_URGENT:
        return "urgent"
    return "normal"


async def _scan_and_fail_stuck_documents(*, timeout_minutes: int) -> dict[str, Any]:
    """扫描当前 run 长时间无更新时间的文档，并将其标记为失败。"""
    from fastapi_common.fastapi_common_sqlalchemy.database import GetAsyncSession
    from sqlalchemy import text as sa_text

    lock_key = 20260512
    timed_out_rows: list[dict[str, Any]] = []

    async with GetAsyncSession() as session:
        lock_row = (
            await session.execute(
                sa_text("SELECT pg_try_advisory_lock(:lock_key)"),
                {"lock_key": lock_key},
            )
        ).fetchone()
        has_lock = bool(lock_row[0]) if lock_row else False
        if not has_lock:
            log.info("stuck-scan skipped: advisory lock already held")
            return {
                "status": "skipped",
                "reason": "lock_not_acquired",
                "timeout_minutes": timeout_minutes,
                "matched": 0,
                "failed": 0,
            }

        try:
            rows = (
                await session.execute(
                    sa_text(
                        """
                        SELECT
                            d.id AS document_id,
                            d.processing_status,
                            d.current_run_id,
                            d.updated_at AS document_updated_at,
                            d.region,
                            d.normalized_name,
                            ar.id AS run_id,
                            ar.status AS run_status,
                            ar.phase,
                            ar.task_id,
                            ar.updated_at AS run_updated_at,
                            EXTRACT(EPOCH FROM (
                                NOW() - CASE
                                    WHEN LOWER(COALESCE(ar.status, '')) IN ('pending', 'queued', 'running', 'retrying')
                                        THEN COALESCE(ar.updated_at, d.updated_at)
                                    ELSE COALESCE(d.updated_at, ar.updated_at)
                                END
                            ))::bigint AS idle_seconds
                        FROM leaudit_documents d
                        LEFT JOIN leaudit_audit_runs ar
                          ON ar.id = d.current_run_id
                        WHERE d.deleted_at IS NULL
                          AND (
                              LOWER(COALESCE(ar.status, '')) IN ('pending', 'queued', 'running', 'retrying')
                              OR LOWER(COALESCE(d.processing_status, '')) IN ('waiting', 'queued', 'running', 'processing')
                          )
                          AND CASE
                              WHEN LOWER(COALESCE(ar.status, '')) IN ('pending', 'queued', 'running', 'retrying')
                                  THEN COALESCE(ar.updated_at, d.updated_at)
                              ELSE COALESCE(d.updated_at, ar.updated_at)
                          END < NOW() - make_interval(mins => :timeout_minutes)
                        ORDER BY CASE
                            WHEN LOWER(COALESCE(ar.status, '')) IN ('pending', 'queued', 'running', 'retrying')
                                THEN COALESCE(ar.updated_at, d.updated_at)
                            ELSE COALESCE(d.updated_at, ar.updated_at)
                        END ASC,
                        d.id ASC
                        """
                    ),
                    {"timeout_minutes": timeout_minutes},
                )
            ).mappings().all()

            timed_out_rows = [dict(row) for row in rows]
        finally:
            await session.execute(
                sa_text("SELECT pg_advisory_unlock(:lock_key)"),
                {"lock_key": lock_key},
            )
            await session.commit()

    storage = StorageAdapter()
    failed_items: list[dict[str, Any]] = []

    for row in timed_out_rows:
        document_id = int(row["document_id"])
        run_id = int(row["run_id"]) if row.get("run_id") is not None else None
        idle_seconds = int(row.get("idle_seconds") or 0)
        run_phase = row.get("phase")
        run_status = row.get("run_status")
        processing_status = row.get("processing_status")
        file_name = row.get("normalized_name") or f"document-{document_id}"

        message = (
            f"文档处理长时间无进展，已自动终止："
            f"document_id={document_id}, run_id={run_id}, "
            f"processing_status={processing_status}, run_status={run_status}, "
            f"phase={run_phase}, idle_seconds={idle_seconds}"
        )

        try:
            await _update_status_safe(document_id, "failed")
            if run_id is not None:
                await storage.fail_run(
                    document_id,
                    run_id=run_id,
                    phase=run_phase or "dispatch",
                    message=message,
                    detail_json={
                        "reason": "stuck_timeout",
                        "timeoutMinutes": timeout_minutes,
                        "idleSeconds": idle_seconds,
                        "taskId": row.get("task_id"),
                        "runStatus": run_status,
                        "processingStatus": processing_status,
                        "phase": run_phase,
                        "fileName": file_name,
                        "region": row.get("region"),
                    },
                )
            failed_items.append(
                {
                    "document_id": document_id,
                    "run_id": run_id,
                    "phase": run_phase,
                    "idle_seconds": idle_seconds,
                }
            )
            log.warning("stuck document auto-failed: %s", message)
        except Exception:
            log.exception("stuck document auto-fail failed: document_id=%s run_id=%s", document_id, run_id)

    return {
        "status": "success",
        "timeout_minutes": timeout_minutes,
        "matched": len(timed_out_rows),
        "failed": len(failed_items),
        "items": failed_items,
    }