from __future__ import annotations

import hashlib
import ipaddress
import json
import logging
import os
import re
import socket
import time
from collections.abc import Iterator, Mapping
from concurrent.futures import ThreadPoolExecutor, as_completed
from contextlib import contextmanager
from copy import deepcopy
from dataclasses import asdict
from pathlib import Path
from threading import BoundedSemaphore, Lock
from typing import Any, Literal
from urllib.parse import parse_qs, quote, urlsplit

import httpx
import psycopg
from fastapi import (
    APIRouter,
    BackgroundTasks,
    Body,
    Depends,
    File,
    Header,
    HTTPException,
    Query,
    Request,
    UploadFile,
)
from fastapi.responses import StreamingResponse
from psycopg.rows import dict_row

from app.core.auth import (
    filter_accessible_resource_ids,
    get_current_user,
    has_resource_access,
    is_admin,
)
from app.core.config import get_settings
from app.db.platform_store import get_platform_store
from app.modules.data_process.algorithms import (
    ParsedText,
    canonical_record_json,
    content_quality_flags,
    desensitize_pii,
    desensitize_structured_record,
    detect_pdf_document_noise,
    estimate_token_count,
    extract_pdf_page_texts,
    generate_standard_records,
    is_near_duplicate,
    near_duplicate_fingerprint,
    parse_text_content,
    preprocess_structured_records_with_lineage,
    remove_document_noise,
    score_quality,
    structured_json_dumps,
)
from app.modules.data_process.document_chunking import (
    DocumentChunk,
    chunk_fixed_text,
    chunk_layout_document,
    chunk_semantic_text,
    merge_short_chunks,
)
from app.modules.data_process.evaluation import (
    evaluate_result_record,
    reevaluate_edited_record,
)
from app.modules.data_process.generation import generate_model_records
from app.modules.data_process.office_preview import (
    MAX_XLSX_PREVIEW_ROWS,
    build_docx_preview,
    build_xlsx_preview,
)
from app.modules.data_process.storage import (
    LocalDataProcessStorage,
    StagedSourceObject,
    get_data_process_storage,
)
from app.modules.data_process.store import (
    ConflictError,
    DataProcessStore,
    DataProcessStoreError,
    InvalidStateError,
    NotFoundError,
    get_data_process_store,
    new_id,
    repeat_task_id,
)
from app.modules.storage.minio_store import get_object_storage
from app.modules.storage.policy import should_store_in_minio
from app.schemas.data_process import (
    DataProcessRegenerateRequest,
    DataProcessRepeatRequest,
    DataProcessStatus,
    DataProcessTaskCreate,
    DataProcessTaskUpdate,
    DataProcessWorkflowStepUpdate,
    ExternalPullRequest,
    ExternalSourceRequest,
    GenerateRequest,
    PreviewBuildRequest,
    PreviewItemCreate,
    PreviewItemUpdate,
    ProcessType,
    PublishRequest,
    ResultBatchEvaluateRequest,
    ResultBatchRegenerateRequest,
    ResultRegenerateRequest,
    ResultUpdate,
)

logger = logging.getLogger(__name__)
MAX_SOURCE_FILE_BYTES = 200 * 1024 * 1024
MAX_SOURCE_FILE_COUNT = 20
MAX_SOURCE_BATCH_BYTES = 500 * 1024 * 1024
MAX_EXTERNAL_PULL_BYTES = 50 * 1024 * 1024
STRUCTURED_SOURCE_SUFFIXES = frozenset(
    {".json", ".jsonl", ".ndjson", ".csv", ".tsv", ".xlsx"}
)
UNSTRUCTURED_SOURCE_SUFFIXES = frozenset(
    {
        ".txt",
        ".md",
        ".markdown",
        ".pdf",
        ".docx",
        ".pptx",
        ".json",
        ".jsonl",
        ".ndjson",
    }
)
SUPPORTED_SOURCE_SUFFIXES = STRUCTURED_SOURCE_SUFFIXES | UNSTRUCTURED_SOURCE_SUFFIXES
LEGACY_OFFICE_CONVERSIONS = {
    ".doc": ".docx",
    ".xls": ".xlsx",
    ".ppt": ".pptx",
}
RAW_INLINE_PREVIEW_MEDIA_TYPES = {
    "pdf": "application/pdf",
    "docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
    "xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
    "pptx": "application/vnd.openxmlformats-officedocument.presentationml.presentation",
}
RESULT_REGENERATION_CONCURRENCY = 4
RESULT_REGENERATION_RETRIES = 0
RESULT_REGENERATION_TIMEOUT_SECONDS = 60.0
_result_regeneration_slots = BoundedSemaphore(RESULT_REGENERATION_CONCURRENCY)
_result_regeneration_claims_lock = Lock()
_active_result_regenerations: set[tuple[str, str]] = set()


def ok(data: Any = None, message: str = "ok") -> dict[str, Any]:
    return {"code": 0, "message": message, "data": data}


def fail(status_code: int, message: str) -> HTTPException:
    return HTTPException(
        status_code=status_code,
        detail={"code": status_code, "message": message, "data": None},
    )


def _authorize_data_process_request(
    request: Request,
    task_id: str | None = None,
    current_user: dict[str, Any] = Depends(get_current_user),
) -> None:
    """Apply resource ACL and tenant checks to every task-scoped endpoint."""
    if not task_id or is_admin(current_user):
        return
    permission = "read" if request.method in {"GET", "HEAD"} else "write"
    if any(marker in request.url.path for marker in ("/start", "/generate", "/regenerate", "/repeat", "/publish", "/external/")):
        permission = "execute"
    if not has_resource_access("data_process", task_id, current_user, permission):
        raise fail(403, "no permission to access this data process task")


router = APIRouter(
    prefix="/data-process",
    dependencies=[Depends(_authorize_data_process_request)],
)


@contextmanager
def api_errors() -> Iterator[None]:
    try:
        yield
    except NotFoundError as exc:
        raise fail(404, str(exc)) from exc
    except ConflictError as exc:
        raise fail(409, str(exc)) from exc
    except InvalidStateError as exc:
        raise fail(409, str(exc)) from exc
    except (DataProcessStoreError, ValueError) as exc:
        raise fail(400, str(exc)) from exc
    except (psycopg.errors.UndefinedTable, psycopg.errors.UndefinedColumn) as exc:
        raise fail(
            503,
            "data process schema is missing or out of date; run schema_cli --check",
        ) from exc
    except psycopg.OperationalError as exc:
        raise fail(503, "data process database is unavailable") from exc


def _safe_file_name(value: str | None, fallback: str) -> str:
    name = Path((value or "").replace("\\", "/")).name.replace("\x00", "").strip()
    return name if name not in {"", ".", ".."} else fallback


def _range_not_satisfiable(size: int) -> HTTPException:
    return HTTPException(
        status_code=416,
        detail={"code": 416, "message": "invalid source byte range", "data": None},
        headers={"Content-Range": f"bytes */{size}"},
    )


def _source_byte_range(value: str | None, size: int) -> tuple[int, int] | None:
    if value is None:
        return None
    match = re.fullmatch(r"bytes=(\d*)-(\d*)", value.strip())
    if match is None or size <= 0:
        raise _range_not_satisfiable(size)
    start_text, end_text = match.groups()
    if not start_text and not end_text:
        raise _range_not_satisfiable(size)
    if start_text:
        start = int(start_text)
        end = int(end_text) if end_text else size - 1
        if start >= size or end < start:
            raise _range_not_satisfiable(size)
    else:
        suffix_length = int(end_text)
        if suffix_length <= 0:
            raise _range_not_satisfiable(size)
        start = max(0, size - suffix_length)
        end = size - 1
    return start, min(end, size - 1)


def _commit_source_batch(
    store: DataProcessStore,
    storage: LocalDataProcessStorage,
    task_id: str,
    prepared: list[dict[str, Any]],
    staged: list[StagedSourceObject],
) -> list[dict[str, Any]]:
    storage.publish(staged)
    storage_object_ids: list[str] = []
    try:
        # The source reference remains in the task schema for compatibility,
        # while storage_objects provides the authoritative MinIO index.
        if get_settings().minio_enabled:
            for item in prepared:
                reference = str(item.get("storage_object_id") or "")
                if not reference.startswith("minio://"):
                    continue
                object_key = storage.object_key(reference)
                metadata = get_object_storage().stat(object_key)
                object_row = get_platform_store().create_storage_object({
                    "resource_type": "data_process_source",
                    "resource_id": task_id,
                    "version_id": str(item.get("id") or new_id("dpsf")),
                    "bucket": get_object_storage().bucket,
                    "object_key": object_key,
                    "file_name": item.get("name"),
                    "content_type": (item.get("metadata") or {}).get("content_type", "application/octet-stream"),
                    "byte_size": metadata.get("byte_size") or item.get("raw_size") or 0,
                    "checksum_sha256": item.get("checksum_sha256"),
                    "status": "available",
                    "created_by": item.get("created_by"),
                })
                storage_object_ids.append(str(object_row["id"]))
        return store.add_source_files(task_id, prepared)
    except Exception:
        for object_id in storage_object_ids:
            try:
                get_platform_store().update_storage_object(object_id, {"status": "deleted"})
            except Exception:
                pass
        for item in staged:
            try:
                storage.delete(item.reference)
            except Exception:
                # 文件系统回滚失败不能覆盖数据库抛出的根因，并继续清理其余对象。
                logger.exception(
                    "failed to roll back data process source object task_id=%s",
                    task_id,
                )
        raise


def _value(config: dict[str, Any], snake_name: str, camel_name: str, default: Any) -> Any:
    if snake_name in config:
        return config[snake_name]
    return config.get(camel_name, default)


def _preprocess_options(config: dict[str, Any]) -> set[str]:
    values = _value(config, "preprocess_options", "preprocessOptions", [])
    return {str(item) for item in values} if isinstance(values, list) else set()


def _preview_quality(content: str, config: dict[str, Any]) -> dict[str, Any]:
    records = generate_standard_records(
        [{"id": "quality-preview", "edited_content": content}],
        split={"train": 100, "validation": 0, "test": 0},
    )
    record = records[0] if records else {"instruction": "", "input": "", "output": ""}
    minimum = int(_value(config, "min_output_length", "minOutputLength", 20) or 20)
    return asdict(
        score_quality(
            record,
            min_output_length=max(1, minimum),
            source_content=content,
        )
    )


def _parse_stored_source(source: dict[str, Any]) -> ParsedText:
    """重新读取已入库的规范化正文，避免把二进制格式当二进制重复解析。"""

    content = str(source.get("content") or "")
    file_format = str(source.get("file_format") or "").lower()
    if file_format == "xlsx":
        raw_content = source.get("raw_content")
        if isinstance(raw_content, bytes):
            return parse_text_content(
                raw_content,
                filename=str(source.get("name") or "source.xlsx"),
                file_format="xlsx",
            )
        # 兼容原始对象已缺失的历史文件：退化为上传阶段生成的 JSONL。
        return parse_text_content(content, file_format="jsonl")
    if file_format in {"pdf", "docx", "pptx"}:
        # 文档上传阶段已抽取文本，预览阶段只需要对正文切片。
        return ParsedText(format=file_format, text=content, records=())
    return parse_text_content(
        content,
        filename=source.get("name"),
        file_format=file_format or None,
    )


def _chunk_source_text(
    source: dict[str, Any],
    config: dict[str, Any],
    preprocess_options: set[str],
) -> list[DocumentChunk]:
    """根据任务配置调用真实的 Docling/LlamaIndex 切分器。"""

    method = str(_value(config, "chunk_method", "chunkMethod", "layout_hybrid"))
    preserve_context = "preserve_context" in preprocess_options
    chunk_size = int(_value(config, "chunk_size", "chunkSize", 800))
    overlap = (
        int(_value(config, "chunk_overlap", "chunkOverlap", 100))
        if preserve_context
        else 0
    )
    text = str(source.get("content") or "")
    if method == "layout_hybrid":
        raw = source.get("raw_content")
        if not isinstance(raw, bytes):
            raise InvalidStateError("版面结构混合切分需要原始文件，请重新上传后再处理")
        chunks = chunk_layout_document(
            raw,
            filename=str(source.get("name") or "document.pdf"),
            source_text=text,
            chunk_size=chunk_size,
        )
    elif method == "semantic":
        chunks = chunk_semantic_text(
            text,
            chunk_size=chunk_size,
            chunk_overlap=overlap,
            breakpoint_percentile_threshold=int(
                _value(
                    config,
                    "semantic_breakpoint_percentile",
                    "semanticBreakpointPercentile",
                    95,
                )
            ),
        )
    elif method == "fixed":
        chunks = chunk_fixed_text(text, chunk_size=chunk_size, chunk_overlap=overlap)
    else:
        raise ValueError(f"unsupported chunk method: {method}")
    if "merge_short_content" in preprocess_options:
        chunks = merge_short_chunks(
            chunks,
            source_text=text,
            min_token_count=int(_value(config, "min_chunk_size", "minChunkSize", 100)),
            max_token_count=chunk_size,
        )
    return chunks


_NEGATION_MARKERS = frozenset({"不", "无", "未", "否", "没有", "并非", "not", "no", "never"})


def _safe_near_duplicate(left: str, right: str) -> bool:
    """保守判断近重复，数字或否定含义变化时始终保留。"""

    if min(estimate_token_count(left), estimate_token_count(right)) < 20:
        return False
    if re.findall(r"\d+(?:\.\d+)?", left) != re.findall(r"\d+(?:\.\d+)?", right):
        return False
    left_lower = left.casefold()
    right_lower = right.casefold()
    left_negations = {marker for marker in _NEGATION_MARKERS if marker in left_lower}
    right_negations = {marker for marker in _NEGATION_MARKERS if marker in right_lower}
    if left_negations != right_negations:
        return False
    return is_near_duplicate(
        left,
        right,
        similarity_threshold=0.92,
        max_hamming_distance=2,
    )


def _near_duplicate_band_keys(content: str) -> tuple[tuple[int, int], ...]:
    """将 64 位 SimHash 分为三段，汉明距离不超过 2 时至少命中一段。"""

    fingerprint = int(near_duplicate_fingerprint(content), 16)
    widths = (22, 21, 21)
    shift = 0
    keys: list[tuple[int, int]] = []
    for index, width in enumerate(widths):
        keys.append((index, (fingerprint >> shift) & ((1 << width) - 1)))
        shift += width
    return tuple(keys)


def _build_preview_items(
    task: dict[str, Any], source_files: list[dict[str, Any]]
) -> list[dict[str, Any]]:
    config = task.get("config") or {}
    process_type = task["process_type"]
    preprocess_options = _preprocess_options(config)
    should_desensitize = "desensitize" in preprocess_options
    should_clean_invalid = bool(
        preprocess_options & {"clean_invalid", "clean_invalid_content"}
    )
    should_deduplicate = bool(
        preprocess_options & {"deduplicate", "deduplicate_content"}
    )
    seen_content_hashes: set[str] = set()
    seen_near_duplicate_bands: dict[tuple[int, int], list[str]] = {}
    items: list[dict[str, Any]] = []

    def append_item(item: dict[str, Any], *, dedup_content: str) -> None:
        content = str(item.get("edited_content") or "").strip()
        if should_clean_invalid and not content:
            return
        # 去重必须基于脱敏前内容，否则不同原文可能在替换 PII 后被错误合并。
        content_hash = hashlib.sha256(dedup_content.strip().encode("utf-8")).hexdigest()
        if should_deduplicate and content_hash in seen_content_hashes:
            return
        seen_content_hashes.add(content_hash)
        if not content:
            item["status"] = "invalid"
        items.append(item)

    for source in source_files:
        parsed = _parse_stored_source(source)
        if process_type == "unstructured":
            document_noise_spans = tuple(source.get("document_noise_spans") or ())
            chunks = _chunk_source_text(source, config, preprocess_options)
            for chunk in chunks:
                content = (
                    remove_document_noise(
                        chunk.contextualized_content,
                        document_noise_spans,
                        source_offset=chunk.source_start or 0,
                    )
                    if (
                        should_clean_invalid
                        and document_noise_spans
                        and chunk.source_start is not None
                    )
                    else chunk.contextualized_content
                )
                preprocess_flags = content_quality_flags(
                    content,
                    min_chars=0,
                    min_tokens=0,
                )
                if content != chunk.contextualized_content:
                    preprocess_flags = (*preprocess_flags, "document_noise_removed")
                flag_set = set(preprocess_flags)
                if "clean_invalid_content" in preprocess_options and flag_set & {
                    "empty_content",
                    "low_printable_ratio",
                    "repetitive_content",
                }:
                    continue
                if "filter_low_quality" in preprocess_options and flag_set & {
                    "content_too_long",
                    "mojibake",
                    "low_printable_ratio",
                    "repetitive_content",
                }:
                    continue
                if "deduplicate_content" in preprocess_options:
                    band_keys = _near_duplicate_band_keys(content)
                    candidates = {
                        previous
                        for key in band_keys
                        for previous in seen_near_duplicate_bands.get(key, ())
                    }
                    if any(
                        _safe_near_duplicate(content, previous)
                        for previous in candidates
                    ):
                        continue
                    for key in band_keys:
                        seen_near_duplicate_bands.setdefault(key, []).append(content)
                dedup_content = content
                pii_counts: dict[str, int] = {}
                if should_desensitize:
                    content, pii_counts = desensitize_pii(content)
                quality = _preview_quality(content, config)
                quality["pii_replacements"] = pii_counts
                quality["preprocess_flags"] = list(preprocess_flags)
                quality["chunk_method"] = str(
                    _value(config, "chunk_method", "chunkMethod", "layout_hybrid")
                )
                quality["heading_path"] = list(chunk.heading_path)
                quality["source_pages"] = list(chunk.source_pages)
                quality["doc_item_refs"] = list(chunk.doc_item_refs)
                quality["source_bboxes"] = list(chunk.source_bboxes)
                append_item(
                    {
                        "source_file_id": source["id"],
                        "original_content": chunk.original_content,
                        "edited_content": content,
                        "source_start": chunk.source_start,
                        "source_end": chunk.source_end,
                        "source_start_line": chunk.source_start_line,
                        "source_end_line": chunk.source_end_line,
                        "token_count": estimate_token_count(content),
                        "status": (
                            "modified"
                            if content != chunk.original_content
                            else "original"
                        ),
                        "quality_score": quality,
                    },
                    dedup_content=dedup_content,
                )
            continue

        structured_options = preprocess_options & {
            "clean_invalid",
            "detect_structure",
            "deduplicate",
            "normalize_format",
            "filter_anomaly",
        }
        source_records = list(parsed.records)
        processed_records = preprocess_structured_records_with_lineage(
            source_records,
            structured_options,
        )
        for processed in processed_records:
            source_index = processed.source_index
            record = processed.record
            original_record = (
                source_records[source_index]
                if source_index < len(source_records)
                else record
            )
            source_locator = (
                deepcopy(parsed.record_locators[source_index])
                if source_index < len(parsed.record_locators)
                else None
            )
            original_content = structured_json_dumps(original_record)
            pii_counts: dict[str, int] = {}
            edited_record = record
            dedup_content = (
                canonical_record_json(record)
                if "normalize_format" in preprocess_options
                else structured_json_dumps(record)
            )
            if should_desensitize:
                edited_record, pii_counts = desensitize_structured_record(record)
            content = (
                canonical_record_json(edited_record)
                if "normalize_format" in preprocess_options
                else structured_json_dumps(edited_record)
            )
            quality = _preview_quality(content, config)
            quality["pii_replacements"] = pii_counts
            if source_locator is not None:
                quality["source_locator"] = source_locator
            source_start = (
                source_locator.get("source_start")
                if source_locator is not None
                else None
            )
            source_end = (
                source_locator.get("source_end")
                if source_locator is not None
                else None
            )
            source_start_line = (
                source_locator.get("start_line")
                if source_locator is not None
                else None
            )
            source_end_line = (
                source_locator.get("end_line")
                if source_locator is not None
                else None
            )
            append_item(
                {
                    "source_file_id": source["id"],
                    "original_content": original_content,
                    "edited_content": content,
                    "source_start": source_start,
                    "source_end": source_end,
                    "source_start_line": source_start_line,
                    "source_end_line": source_end_line,
                    "token_count": estimate_token_count(content),
                    "status": "modified" if content != original_content else "original",
                    "quality_score": quality,
                },
                dedup_content=dedup_content,
            )
    return items


def _all_preview_items(store: DataProcessStore, task_id: str) -> list[dict[str, Any]]:
    """分页读取全部预览项，避免固定上限静默截断任务。"""

    items: list[dict[str, Any]] = []
    page = 1
    page_size = 5_000
    while True:
        result = store.list_preview_items(task_id, page=page, page_size=page_size)
        batch = result["items"]
        items.extend(batch)
        if len(items) >= int(result["total"]) or not batch:
            return items
        page += 1


def _run_generation(
    store: DataProcessStore, task_id: str, generation_run_id: str
) -> None:
    started_at = time.perf_counter()
    logger.info(
        "data process generation worker started task_id=%s generation_run_id=%s",
        task_id,
        generation_run_id,
    )
    try:
        task = store.get_task(task_id)
        if not store.generation_is_running(task_id, generation_run_id):
            logger.info(
                "data process generation worker skipped inactive run task_id=%s "
                "generation_run_id=%s",
                task_id,
                generation_run_id,
            )
            return
        all_preview_items = _all_preview_items(store, task_id)
        preview_items = [
            item
            for item in all_preview_items
            if item.get("status") != "invalid"
            and str(item.get("edited_content") or item.get("original_content") or "").strip()
        ]
        pre_filtered_count = len(all_preview_items) - len(preview_items)
        config = task.get("config") or {}
        model_id = _value(config, "generation_model_id", "generationModelId", None)
        generation_model: dict[str, Any] | None = None
        if model_id:
            generation_model = store.get_generation_model(str(model_id))
            task = store.save_generation_model_snapshot(
                task_id,
                generation_model,
                generation_run_id=generation_run_id,
            )
            config = task.get("config") or config
        split = _value(
            config,
            "dataset_split",
            "datasetSplit",
            {"train": 80, "validation": 10, "test": 10},
        )
        pairs = (
            _value(config, "qa_pairs_per_chunk", "qaPairsPerChunk", 1)
            if task["process_type"] == "unstructured"
            else _value(config, "qa_pairs_per_row", "qaPairsPerRow", 1)
        )
        output_type = str(
            _value(config, "output_type", "outputType", "standard")
        ).strip().lower()
        if generation_model:
            runtime_config = {
                **config,
                "generation_prompt": _value(
                    config, "generation_prompt", "generationPrompt", ""
                ),
                "output_type": output_type,
                "reasoning_detail": _value(
                    config, "reasoning_detail", "reasoningDetail", "normal"
                ),
                "max_tokens": _value(config, "max_tokens", "maxTokens", 1024),
                "json_mode": _value(config, "json_mode", "jsonMode", False),
            }
            def report_progress(processed_count: int, total_count: int) -> None:
                if not store.update_generation_progress(
                    task_id,
                    generation_run_id,
                    processed_count,
                    total_count,
                ):
                    raise InvalidStateError("generation run is no longer active")

            generated = generate_model_records(
                preview_items,
                model=generation_model,
                config=runtime_config,
                task_id=task_id,
                split=split,
                qa_pairs_per_item=int(pairs or 1),
                on_progress=report_progress,
            )
        elif output_type in {"reasoning", "dpo"}:
            if output_type == "reasoning":
                raise InvalidStateError("思维链输出必须配置可用的数据生成模型")
            raise InvalidStateError("DPO 输出必须配置可用的数据生成模型")
        else:
            generated = generate_standard_records(
                preview_items,
                qa_pairs_per_item=int(pairs or 1),
                semantic_enrichment=bool(
                    _value(config, "semantic_enrichment", "semanticEnrichment", False)
                ),
                split=split,
                split_seed=task_id,
            )
            if not store.update_generation_progress(
                task_id,
                generation_run_id,
                len(preview_items),
                len(preview_items),
            ):
                logger.info(
                    "data process generation stopped before completion task_id=%s "
                    "generation_run_id=%s",
                    task_id,
                    generation_run_id,
                )
                return

        known_fingerprints: set[str] = set()
        accepted: list[dict[str, Any]] = []
        filtered_count = pre_filtered_count
        duplicate_count = 0
        error_count = 0
        quality_filter = bool(
            _value(config, "quality_filter_enabled", "qualityFilterEnabled", False)
        )
        filter_low = bool(_value(config, "filter_low_quality", "filterLowQuality", True))
        filter_short = bool(
            _value(config, "filter_short_content", "filterShortContent", True)
        )
        deduplicate = bool(
            _preprocess_options(config)
            & {"deduplicate", "deduplicate_content"}
        )
        minimum = max(1, int(_value(config, "min_output_length", "minOutputLength", 20) or 20))
        preview_sources = {
            str(item["id"]): str(
                item.get("edited_content") or item.get("original_content") or ""
            )
            for item in preview_items
        }

        for record in generated:
            quality = score_quality(
                record,
                min_output_length=minimum,
                source_content=preview_sources.get(str(record.get("preview_item_id") or ""), ""),
                known_fingerprints=known_fingerprints,
            )
            if "duplicate_record" in quality.flags:
                duplicate_count += 1
            else:
                # 即使首条随后因短文本/低质量被过滤，也要阻止同批后续重复结果。
                known_fingerprints.add(quality.fingerprint)
            should_filter = (
                (deduplicate and "duplicate_record" in quality.flags)
                or (
                    quality_filter
                    and filter_short
                    and "output_too_short" in quality.flags
                )
                or (quality_filter and filter_low and not quality.is_valid)
            )
            if not quality.is_valid:
                error_count += 1
                record["status"] = "invalid"
                record["error"] = ", ".join(quality.flags) or "quality validation failed"
            if should_filter:
                filtered_count += 1
                continue
            record["quality_score"] = asdict(quality)
            accepted.append(record)

        # stop 请求可能在纯函数计算期间到达，最终写入前再次检查状态。
        if store.generation_is_running(task_id, generation_run_id):
            completed = store.complete_generation(
                task_id,
                accepted,
                generation_run_id=generation_run_id,
                filtered_count=filtered_count,
                duplicate_count=duplicate_count,
                error_count=error_count,
            )
            result_bytes = "".join(
                structured_json_dumps(item) + "\n" for item in accepted
            ).encode("utf-8")
            if should_store_in_minio(len(result_bytes)):
                result_key = f"data-process/{task_id}/results/{generation_run_id}.jsonl"
                uploaded = get_object_storage().put_bytes(result_key, result_bytes, "application/jsonl")
                get_platform_store().create_storage_object({
                    "resource_type": "data_process_result",
                    "resource_id": task_id,
                    "version_id": generation_run_id,
                    "bucket": uploaded["bucket"],
                    "object_key": result_key,
                    "file_name": f"{generation_run_id}.jsonl",
                    "content_type": "application/jsonl",
                    "byte_size": len(result_bytes),
                    "checksum_sha256": hashlib.sha256(result_bytes).hexdigest(),
                    "status": "available",
                    "created_by": (store.get_task(task_id) or {}).get("created_by"),
                })
            logger.info(
                "data process generation completed task_id=%s generation_run_id=%s "
                "output_count=%s filtered_count=%s duplicate_count=%s error_count=%s "
                "duration_ms=%.2f",
                task_id,
                generation_run_id,
                completed.get("output_count", len(accepted)),
                completed.get("filtered_count", filtered_count),
                completed.get("duplicate_count", duplicate_count),
                completed.get("error_count", error_count),
                (time.perf_counter() - started_at) * 1000,
            )
        else:
            logger.info(
                "data process generation stopped before result persistence task_id=%s "
                "generation_run_id=%s",
                task_id,
                generation_run_id,
            )
    except Exception as exc:
        logger.exception(
            "data process generation failed task_id=%s generation_run_id=%s duration_ms=%.2f",
            task_id,
            generation_run_id,
            (time.perf_counter() - started_at) * 1000,
        )
        try:
            if store.generation_is_running(task_id, generation_run_id):
                store.mark_failed(
                    task_id,
                    str(exc),
                    generation_run_id=generation_run_id,
                )
        except Exception:
            logger.exception(
                "failed to persist data process generation failure task_id=%s "
                "generation_run_id=%s",
                task_id,
                generation_run_id,
            )


@router.get("")
def list_tasks(
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=20, ge=1, le=200),
    keyword: str | None = Query(default=None),
    status: DataProcessStatus | None = Query(default=None),
    process_type: ProcessType | None = Query(default=None),
    tenant_id: str | None = Query(default=None),
    project_id: str | None = Query(default=None),
    store: DataProcessStore = Depends(get_data_process_store),
    current_user: dict = Depends(get_current_user),
) -> dict[str, Any]:
    with api_errors():
        tasks = store.list_tasks(
            page=page,
            page_size=page_size,
            keyword=keyword,
            status=status,
            process_type=process_type,
            tenant_id=tenant_id if is_admin(current_user) else (current_user.get("tenant_id") or "default"),
            project_id=project_id,
        )
        # #4 资源 ACL 过滤：admin 放行，普通用户只看到自己被授权的数据处理任务
        items = tasks.get("items", [])
        if not is_admin(current_user) and items:
            accessible_ids = set(
                filter_accessible_resource_ids(
                    "data-process", [t["id"] for t in items], current_user
                )
            )
            items = [t for t in items if t["id"] in accessible_ids]
            tasks["items"] = items
            tasks["total"] = len(items)
        return ok(tasks)


@router.post("")
def create_task(
    payload: DataProcessTaskCreate,
    store: DataProcessStore = Depends(get_data_process_store),
    current_user: dict = Depends(get_current_user),
) -> dict[str, Any]:
    with api_errors():
        values = payload.model_dump(mode="json")
        values["created_by"] = current_user.get("id")
        values["owner_id"] = current_user.get("id")
        values["tenant_id"] = current_user.get("tenant_id") or "default"
        get_platform_store().assert_active_tenant(values["tenant_id"])
        task = store.create_task(values)
        return ok(task, "data process task created")


@router.get("/{task_id}")
def task_detail(
    task_id: str,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        # 兼容旧版曾在第五步前清空结果的异常任务；严格特征匹配且幂等。
        store.recover_legacy_aborted_regeneration(task_id)
        task = store.get_task(task_id)
        source_files = store.list_source_files(task_id)
        task["source_files"] = source_files
        task["source_file_count"] = len(source_files)
        return ok(task)


@router.put("/{task_id}")
def update_task(
    task_id: str,
    payload: DataProcessTaskUpdate,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        return ok(
            store.update_task(task_id, payload.model_dump(exclude_unset=True, mode="json")),
            "data process task updated",
        )


@router.put("/{task_id}/workflow-step")
def update_workflow_step(
    task_id: str,
    payload: DataProcessWorkflowStepUpdate,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        return ok(
            store.update_workflow_step(task_id, payload.workflow_step.value),
            "data process workflow step updated",
        )


@router.post("/{task_id}/regenerate")
def prepare_regeneration(
    task_id: str,
    payload: DataProcessRegenerateRequest,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        return ok(
            store.prepare_regeneration(task_id, payload.model_dump(mode="json")),
            "data process task prepared for regeneration",
        )


def _repeat_file_copies(
    store: DataProcessStore,
    storage: LocalDataProcessStorage,
    source_task_id: str,
    request_id: str,
) -> tuple[dict[str, dict[str, str]], list[StagedSourceObject]]:
    """为新任务创建独立的源文件引用，避免删除任一任务时互相影响。"""

    target_task_id = repeat_task_id(source_task_id, request_id)
    copies: dict[str, dict[str, str]] = {}
    staged: list[StagedSourceObject] = []
    batch_id = storage.new_batch_id()
    for summary in store.list_source_files(source_task_id):
        old_file_id = str(summary["id"])
        source = store.get_source_file(source_task_id, old_file_id, include_content=True)
        new_file_id = new_id("dpsf")
        old_reference = str(source.get("storage_object_id") or "")
        if old_reference.startswith(("local://data-process/", "minio://data-process/")):
            staged_object = storage.stage_copy(
                batch_id=batch_id,
                source_reference=old_reference,
                expected_source_task_id=source_task_id,
                expected_source_file_id=old_file_id,
                task_id=target_task_id,
                source_file_id=new_file_id,
                version=1,
                name=str(source["name"]),
            )
            staged.append(staged_object)
            new_reference = staged_object.reference
        elif old_reference.startswith("db://data-process/") or not old_reference:
            new_reference = f"db://data-process/{target_task_id}/{new_file_id}/v1"
        else:
            raise ValueError("源任务包含不受支持的文件存储引用")
        copies[old_file_id] = {
            "id": new_file_id,
            "storage_object_id": new_reference,
        }
    return copies, staged


def _remove_repeated_storage_objects(
    storage: LocalDataProcessStorage,
    task_id: str,
    staged: list[StagedSourceObject],
    copies: dict[str, dict[str, str]],
) -> None:
    source_file_ids = {
        str(copy["storage_object_id"]): str(copy["id"])
        for copy in copies.values()
    }
    for item in staged:
        try:
            storage.delete(
                item.reference,
                expected_task_id=task_id,
                expected_source_file_id=source_file_ids[item.reference],
            )
        except Exception:
            logger.exception(
                "failed to roll back repeated data process source object task_id=%s",
                task_id,
            )


@router.post("/{task_id}/repeat", status_code=202)
def repeat_generation(
    task_id: str,
    payload: DataProcessRepeatRequest,
    background_tasks: BackgroundTasks,
    store: DataProcessStore = Depends(get_data_process_store),
    storage: LocalDataProcessStorage = Depends(get_data_process_storage),
) -> dict[str, Any]:
    """按原任务快照创建独立任务，并立即在后台开始新一批生成。"""

    with api_errors():
        repeated = store.find_repeated_task(task_id, payload.request_id)
        staged: list[StagedSourceObject] = []
        target_task_id = repeat_task_id(task_id, payload.request_id)
        if repeated is None:
            copies, staged = _repeat_file_copies(
                store,
                storage,
                task_id,
                payload.request_id,
            )
            storage.publish(staged)
            try:
                repeated = store.repeat_task(
                    task_id,
                    expected_updated_at=payload.expected_updated_at,
                    request_id=payload.request_id,
                    file_copies=copies,
                )
            except Exception:
                _remove_repeated_storage_objects(
                    storage,
                    target_task_id,
                    staged,
                    copies,
                )
                raise
            if not repeated["created"]:
                _remove_repeated_storage_objects(
                    storage,
                    target_task_id,
                    staged,
                    copies,
                )

        repeated_task = repeated["task"]
        if repeated_task.get("status") == "pending":
            try:
                started = store.start_generation(target_task_id, replace_existing=True)
                background_tasks.add_task(
                    _run_generation,
                    store,
                    target_task_id,
                    str(started["generation_run_id"]),
                )
            except ConflictError:
                latest = store.get_task(target_task_id)
                if latest.get("status") != "running":
                    raise
        repeated["task"] = store.get_task(target_task_id)
        repeated["progress"] = store.progress(target_task_id)
        return ok(repeated, "已按原配置创建新任务并开始后台生成")


@router.delete("/{task_id}")
def delete_task(
    task_id: str,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        store.delete_task(task_id)
        return ok({"deleted": task_id}, "data process task deleted")


@router.get("/{task_id}/source-files")
def source_files(
    task_id: str,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        return ok({"files": store.list_source_files(task_id)})


@router.post("/{task_id}/source-files")
async def upload_source_files(
    task_id: str,
    files: list[UploadFile] = File(...),
    store: DataProcessStore = Depends(get_data_process_store),
    storage: LocalDataProcessStorage = Depends(get_data_process_storage),
) -> dict[str, Any]:
    if not files:
        raise fail(400, "at least one source file is required")
    if len(files) > MAX_SOURCE_FILE_COUNT:
        raise fail(413, f"a source batch may contain at most {MAX_SOURCE_FILE_COUNT} files")
    prepared: list[dict[str, Any]] = []
    staged: list[StagedSourceObject] = []
    batch_id = storage.new_batch_id()
    batch_size = 0
    commit_attempted = False
    try:
        with api_errors():
            task = store.get_task(task_id)
            process_type = str(task["process_type"])
            source_mode = str(
                _value(task.get("config") or {}, "source_mode", "sourceMode", "local")
            )
            if process_type == "external" or source_mode == "external":
                raise InvalidStateError(
                    "external tasks must import data through the external source endpoint"
                )
            allowed_suffixes = (
                UNSTRUCTURED_SOURCE_SUFFIXES
                if process_type == "unstructured"
                else STRUCTURED_SOURCE_SUFFIXES
            )
            for upload in files:
                raw = await upload.read(MAX_SOURCE_FILE_BYTES + 1)
                if len(raw) > MAX_SOURCE_FILE_BYTES:
                    raise fail(413, f"source file exceeds {MAX_SOURCE_FILE_BYTES} bytes")
                name = _safe_file_name(upload.filename, "source.txt")
                suffix = Path(name).suffix.lower()
                if suffix in LEGACY_OFFICE_CONVERSIONS:
                    replacement = LEGACY_OFFICE_CONVERSIONS[suffix]
                    raise fail(
                        415,
                        f"legacy {suffix} format is not supported; "
                        f"convert the file to {replacement} and upload again",
                    )
                if suffix not in SUPPORTED_SOURCE_SUFFIXES:
                    raise fail(415, f"unsupported source file format: {suffix or 'none'}")
                if suffix not in allowed_suffixes:
                    raise fail(
                        415,
                        f"{suffix} is not supported for {process_type} data processing",
                    )
                parsed = parse_text_content(raw, filename=name)
                if not parsed.text.strip():
                    raise fail(400, f"source file is empty: {name}")
                batch_size += len(raw)
                if batch_size > MAX_SOURCE_BATCH_BYTES:
                    raise fail(413, f"source batch exceeds {MAX_SOURCE_BATCH_BYTES} bytes")
                source_file_id = new_id("dpsf")
                staged_object = storage.stage_bytes(
                    batch_id=batch_id,
                    task_id=task_id,
                    source_file_id=source_file_id,
                    version=1,
                    name=name,
                    content=raw,
                )
                staged.append(staged_object)
                record_count = (
                    len(parsed.records)
                    if process_type == "structured"
                    else (1 if parsed.text else 0)
                )
                prepared.append(
                    {
                        "id": source_file_id,
                        "storage_object_id": staged_object.reference,
                        "name": name,
                        "content": parsed.text,
                        "raw_size": len(raw),
                        "checksum_sha256": hashlib.sha256(raw).hexdigest(),
                        "file_format": parsed.format,
                        "record_count": record_count,
                        "metadata": {
                            "content_type": upload.content_type or "text/plain",
                            "original_size_bytes": len(raw),
                            "original_checksum_sha256": hashlib.sha256(raw).hexdigest(),
                        },
                        "created_by": None,
                    }
                )
            # publish 无论成功或失败都会消费并清理暂存对象，外层不能再次 discard。
            commit_attempted = True
            created = _commit_source_batch(store, storage, task_id, prepared, staged)
    finally:
        if not commit_attempted:
            storage.discard(staged)
    return ok({"files": created}, "source files uploaded")


@router.get("/{task_id}/source-files/{file_id}/content")
def source_file_content(
    task_id: str,
    file_id: str,
    start_line: int | None = Query(default=None, ge=1),
    line_count: int = Query(default=200, ge=1, le=10_000),
    offset: int = Query(default=0, ge=0),
    limit: int = Query(default=100_000, ge=1, le=1_000_000),
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        if start_line is not None:
            return ok(store.source_content_lines(task_id, file_id, start_line, line_count))
        return ok(store.source_content_window(task_id, file_id, offset, limit))


@router.get("/{task_id}/source-files/{file_id}/raw")
def source_file_raw(
    task_id: str,
    file_id: str,
    range_header: str | None = Header(default=None, alias="Range"),
    store: DataProcessStore = Depends(get_data_process_store),
    storage: LocalDataProcessStorage = Depends(get_data_process_storage),
) -> StreamingResponse:
    with api_errors():
        source = store.get_source_file(task_id, file_id, include_content=False)
        file_format = str(source.get("file_format") or "").lower()
        media_type = RAW_INLINE_PREVIEW_MEDIA_TYPES.get(file_format)
        if media_type is None:
            raise fail(
                415,
                "raw inline preview is only available for PDF and modern Office source files",
            )
        storage_object_id = str(source.get("storage_object_id") or "")
        actual_size = storage.file_size(
            storage_object_id,
            expected_task_id=task_id,
            expected_source_file_id=file_id,
        )
        if actual_size is None:
            raise fail(410, "the original file is unavailable for this legacy source file")
        expected_size = int(source.get("size_bytes") or 0)
        if actual_size != expected_size:
            raise ValueError("source object size does not match metadata")
        selected_range = _source_byte_range(range_header, actual_size)
        start, end = selected_range or (0, actual_size - 1)
        length = end - start + 1
        default_name = f"source.{file_format}"
        name = _safe_file_name(str(source.get("name") or default_name), default_name)
        headers = {
            "Accept-Ranges": "bytes",
            "Cache-Control": "private, no-store",
            "Content-Disposition": f"inline; filename*=UTF-8''{quote(name, safe='')}",
            "Content-Length": str(length),
            "X-Accel-Buffering": "no",
            "X-Content-Type-Options": "nosniff",
        }
        checksum = str(source.get("checksum_sha256") or "")
        if checksum:
            headers["ETag"] = f'"{checksum}"'
        if selected_range is not None:
            headers["Content-Range"] = f"bytes {start}-{end}/{actual_size}"
        body = storage.iter_bytes(
            storage_object_id,
            expected_task_id=task_id,
            expected_source_file_id=file_id,
            expected_size=actual_size,
            start=start,
            length=length,
        )
        return StreamingResponse(
            body,
            status_code=206 if selected_range is not None else 200,
            media_type=media_type,
            headers=headers,
        )


@router.get("/{task_id}/source-files/{file_id}/office-preview")
def source_file_office_preview(
    task_id: str,
    file_id: str,
    sheet_index: int = Query(default=0, ge=0),
    offset: int = Query(default=0, ge=0),
    limit: int = Query(default=100, ge=1, le=MAX_XLSX_PREVIEW_ROWS),
    store: DataProcessStore = Depends(get_data_process_store),
    storage: LocalDataProcessStorage = Depends(get_data_process_storage),
) -> dict[str, Any]:
    """返回 Word 版式块或 Excel 工作表网格，不把二进制内容下发给组件解析。"""

    with api_errors():
        source = store.get_source_file(task_id, file_id, include_content=False)
        file_format = str(source.get("file_format") or "").lower()
        if file_format not in {"docx", "xlsx"}:
            raise fail(415, "Office preview is only available for DOCX and XLSX source files")
        storage_object_id = str(source.get("storage_object_id") or "")
        actual_size = storage.file_size(
            storage_object_id,
            expected_task_id=task_id,
            expected_source_file_id=file_id,
        )
        if actual_size is None:
            raise fail(410, "the original Office file is unavailable for this legacy source file")
        expected_size = int(source.get("size_bytes") or 0)
        if actual_size != expected_size:
            raise ValueError("source object size does not match metadata")
        raw = b"".join(
            storage.iter_bytes(
                storage_object_id,
                expected_task_id=task_id,
                expected_source_file_id=file_id,
                expected_size=actual_size,
            )
        )
        preview = (
            build_docx_preview(raw)
            if file_format == "docx"
            else build_xlsx_preview(
                raw,
                sheet_index=sheet_index,
                offset=offset,
                limit=limit,
            )
        )
        preview["file_name"] = str(source.get("name") or f"source.{file_format}")
        return ok(preview)


@router.get("/{task_id}/source-files/{file_id}/pdf-pages")
def source_file_pdf_pages(
    task_id: str,
    file_id: str,
    store: DataProcessStore = Depends(get_data_process_store),
    storage: LocalDataProcessStorage = Depends(get_data_process_storage),
) -> dict[str, Any]:
    """返回切片全文偏移对应的 PDF 物理页范围。"""

    with api_errors():
        source = store.get_source_file(task_id, file_id, include_content=False)
        if str(source.get("file_format") or "").lower() != "pdf":
            raise fail(415, "PDF page mapping is only available for PDF source files")
        storage_object_id = str(source.get("storage_object_id") or "")
        actual_size = storage.file_size(
            storage_object_id,
            expected_task_id=task_id,
            expected_source_file_id=file_id,
        )
        if actual_size is None:
            raise fail(410, "the original PDF is unavailable for this legacy source file")
        expected_size = int(source.get("size_bytes") or 0)
        if actual_size != expected_size:
            raise ValueError("source object size does not match metadata")
        raw = b"".join(
            storage.iter_bytes(
                storage_object_id,
                expected_task_id=task_id,
                expected_source_file_id=file_id,
                expected_size=actual_size,
            )
        )
        pages = extract_pdf_page_texts(raw)
        return ok(
            {
                "page_count": len(pages),
                "pages": [
                    {
                        "page_number": page.page_number,
                        "source_start": page.source_start,
                        "source_end": page.source_end,
                    }
                    for page in pages
                ],
            }
        )


@router.delete("/{task_id}/source-files/{file_id}")
def delete_source_file(
    task_id: str,
    file_id: str,
    store: DataProcessStore = Depends(get_data_process_store),
    storage: LocalDataProcessStorage = Depends(get_data_process_storage),
) -> dict[str, Any]:
    with api_errors():
        source = store.get_source_file(task_id, file_id, include_content=False)
        storage_object_id = str(source.get("storage_object_id") or "")
        storage.validate_owner(
            storage_object_id,
            expected_task_id=task_id,
            expected_source_file_id=file_id,
        )
        store.delete_source_file(task_id, file_id)
        cleanup_pending = False
        try:
            storage.delete(
                storage_object_id,
                expected_task_id=task_id,
                expected_source_file_id=file_id,
            )
        except Exception:
            # 数据库软删除已经提交，不能再向客户端返回可重试的失败；保留逻辑引用，
            # 由后续存储清理任务重试物理删除。
            cleanup_pending = True
            logger.exception(
                "failed to remove data process source object after soft deletion",
                extra={"task_id": task_id, "source_file_id": file_id},
            )
        return ok(
            {"deleted": file_id, "storage_cleanup_pending": cleanup_pending},
            "source file removed",
        )


def _external_postgres_connection(payload: ExternalSourceRequest) -> psycopg.Connection[Any]:
    kind = payload.type.strip().lower()
    parsed_url = urlsplit(payload.url)
    scheme = parsed_url.scheme.lower()
    if kind not in {"postgres", "postgresql"} or scheme not in {"postgres", "postgresql"}:
        raise fail(501, f"external data source type is not supported: {payload.type}")
    if parsed_url.username or parsed_url.password:
        raise fail(400, "database credentials must use the account and password fields")
    if set(parse_qs(parsed_url.query)) & {
        "password", "secret", "token", "api_key", "user", "username"
    }:
        raise fail(400, "database URL query must not contain credentials")
    if payload.auth_mode not in {"none", "basic"}:
        raise fail(400, "PostgreSQL supports only none or basic authentication")
    if payload.auth_mode == "basic" and not payload.username:
        raise fail(400, "database username is required for basic authentication")
    hostname = parsed_url.hostname
    if not hostname:
        raise fail(400, "external PostgreSQL URL must include a hostname")
    allow_private = os.getenv("DATA_PROCESS_ALLOW_PRIVATE_EXTERNAL_DB", "").lower() in {
        "1",
        "true",
        "yes",
    }
    if not allow_private:
        try:
            addresses = {
                item[4][0]
                for item in socket.getaddrinfo(
                    hostname,
                    parsed_url.port or 5432,
                    type=socket.SOCK_STREAM,
                )
            }
        except socket.gaierror as exc:
            raise fail(400, "external PostgreSQL hostname cannot be resolved") from exc
        if any(
            (address := ipaddress.ip_address(value)).is_private
            or address.is_loopback
            or address.is_link_local
            or address.is_reserved
            or address.is_unspecified
            for value in addresses
        ):
            raise fail(
                403,
                "private or local database addresses are disabled; "
                "set DATA_PROCESS_ALLOW_PRIVATE_EXTERNAL_DB=true only in a trusted deployment",
            )
    kwargs: dict[str, Any] = {
        "connect_timeout": payload.connect_timeout_seconds,
        "row_factory": dict_row,
        "application_name": "yg-ft-data-process-readonly",
        "options": (
            "-c default_transaction_read_only=on "
            f"-c statement_timeout={payload.statement_timeout_seconds * 1000}"
        ),
        "sslmode": payload.ssl_mode,
    }
    if payload.auth_mode == "basic" and payload.username:
        kwargs["user"] = payload.username
    if payload.auth_mode == "basic" and payload.password:
        kwargs["password"] = payload.password
    return psycopg.connect(payload.url, **kwargs)


def _assert_external_source_task(task: dict[str, Any]) -> None:
    if str(task.get("process_type")) == "external":
        return
    config = task.get("config") or {}
    source_mode = str(_value(config, "source_mode", "sourceMode", "local"))
    if str(task.get("process_type")) != "structured" or source_mode != "external":
        raise InvalidStateError(
            "external source access requires a structured task with source_mode=external"
        )


@router.post("/{task_id}/external/test")
def test_external_source(
    task_id: str,
    payload: ExternalSourceRequest,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        task = store.get_task(task_id)
        _assert_external_source_task(task)
        try:
            with _external_postgres_connection(payload) as conn:
                conn.execute("SELECT 1 AS ok").fetchone()
        except psycopg.Error as exc:
            raise fail(502, "external PostgreSQL connection test failed") from exc
        return ok({"connected": True, "type": payload.type})


@router.post("/{task_id}/external/pull")
def pull_external_source(
    task_id: str,
    payload: ExternalPullRequest,
    store: DataProcessStore = Depends(get_data_process_store),
    storage: LocalDataProcessStorage = Depends(get_data_process_storage),
) -> dict[str, Any]:
    query = (payload.query or "").strip()
    if query.endswith(";"):
        query = query[:-1].rstrip()
    if ";" in query:
        raise fail(400, "external pull accepts exactly one read-only query")
    first_token = query.split(maxsplit=1)[0].lower() if query else ""
    if first_token not in {"select", "with"}:
        raise fail(400, "a read-only SELECT or WITH query is required for external pull")
    with api_errors():
        task = store.get_task(task_id)
        _assert_external_source_task(task)
        try:
            with _external_postgres_connection(payload) as conn:
                conn.execute("SET TRANSACTION READ ONLY")
                conn.execute(
                    "SELECT set_config('statement_timeout', %s, true)",
                    (f"{payload.statement_timeout_seconds}s",),
                )
                cursor = conn.execute(query)
                rows: list[dict[str, Any]] = []
                content_parts: list[str] = []
                content_size = 0
                while len(rows) < payload.limit:
                    batch = cursor.fetchmany(min(1_000, payload.limit - len(rows)))
                    if not batch:
                        break
                    for row in batch:
                        line = json.dumps(row, ensure_ascii=False, default=str) + "\n"
                        content_size += len(line.encode("utf-8"))
                        if content_size > MAX_EXTERNAL_PULL_BYTES:
                            raise fail(413, "external pull result exceeds the 50 MiB safety limit")
                        rows.append(row)
                        content_parts.append(line)
                conn.rollback()
        except psycopg.Error as exc:
            raise fail(502, "external PostgreSQL query failed") from exc
        if not rows:
            raise fail(400, "external query returned no rows")
        content = "".join(content_parts)
        raw = content.encode("utf-8")
        name = _safe_file_name(payload.file_name, "external-data.jsonl")
        source_file_id = new_id("dpsf")
        staged = storage.stage_bytes(
            batch_id=storage.new_batch_id(),
            task_id=task_id,
            source_file_id=source_file_id,
            version=1,
            name=name,
            content=raw,
        )
        sources = _commit_source_batch(
            store,
            storage,
            task_id,
            [
                {
                    "id": source_file_id,
                    "storage_object_id": staged.reference,
                    "name": name,
                    "content": content,
                    "raw_size": len(raw),
                    "checksum_sha256": hashlib.sha256(raw).hexdigest(),
                    "file_format": "jsonl",
                    "record_count": len(rows),
                    "metadata": {
                        "external_type": payload.type,
                        "external_host": urlsplit(payload.url).hostname,
                        "external_limit": payload.limit,
                        "external_ssl_mode": payload.ssl_mode,
                        "external_connect_timeout_seconds": payload.connect_timeout_seconds,
                        "external_statement_timeout_seconds": payload.statement_timeout_seconds,
                    },
                }
            ],
            [staged],
        )
        source = sources[0]
        return ok({"files": [source]}, "external source pulled")


def _prepare_preview_items(
    task_id: str,
    store: DataProcessStore,
    storage: LocalDataProcessStorage,
    source_file_ids: list[str] | None = None,
) -> list[dict[str, Any]]:
    task = store.get_task(task_id)
    source_summaries = store.list_source_files(task_id)
    if source_file_ids is not None:
        requested = set(source_file_ids)
        source_summaries = [item for item in source_summaries if item["id"] in requested]
        found = {item["id"] for item in source_summaries}
        missing = requested - found
        if missing:
            raise NotFoundError(f"source files not found: {', '.join(sorted(missing))}")
    sources = [
        store.get_source_file(task_id, item["id"], include_content=True)
        for item in source_summaries
    ]
    if not sources:
        raise InvalidStateError("at least one source file is required")
    config = task.get("config") or {}
    preprocess_options = _preprocess_options(config)
    chunk_method = str(
        _value(config, "chunk_method", "chunkMethod", "layout_hybrid")
    )
    is_unstructured = task.get("process_type") == "unstructured"
    needs_unstructured_raw = is_unstructured and (
        chunk_method == "layout_hybrid"
        or preprocess_options & {"clean_invalid", "clean_invalid_content"}
    )
    has_structured_xlsx = not is_unstructured and any(
        str(source.get("file_format") or "").lower() == "xlsx"
        for source in sources
    )
    if needs_unstructured_raw or has_structured_xlsx:
        for index, source in enumerate(sources):
            source_format = str(source.get("file_format") or "").lower()
            needs_structured_xlsx = not is_unstructured and source_format == "xlsx"
            needs_layout_raw = (
                is_unstructured
                and chunk_method == "layout_hybrid"
            )
            needs_pdf_noise = (
                is_unstructured
                and not needs_layout_raw
                and source_format == "pdf"
                and bool(
                    preprocess_options & {"clean_invalid", "clean_invalid_content"}
                )
            )
            if not (needs_structured_xlsx or needs_layout_raw or needs_pdf_noise):
                continue
            storage_object_id = str(source.get("storage_object_id") or "")
            actual_size = storage.file_size(
                storage_object_id,
                expected_task_id=task_id,
                expected_source_file_id=str(source["id"]),
            )
            if actual_size is None:
                if needs_layout_raw:
                    raise InvalidStateError(
                        "版面结构混合切分无法读取原始文件，请重新上传后再处理"
                    )
                continue
            expected_size = int(source.get("size_bytes") or 0)
            if expected_size and actual_size != expected_size:
                raise ValueError("source object size does not match metadata")
            raw = b"".join(
                storage.iter_bytes(
                    storage_object_id,
                    expected_task_id=task_id,
                    expected_source_file_id=str(source["id"]),
                    expected_size=actual_size,
                )
            )
            enriched = dict(source)
            if needs_structured_xlsx or needs_layout_raw:
                enriched["raw_content"] = raw
                sources[index] = enriched
                continue
            pages = extract_pdf_page_texts(raw)
            extracted_text = "\n\n".join(page.text for page in pages if page.text)
            if extracted_text != str(source.get("content") or ""):
                logger.warning(
                    "skip PDF document noise detection because stored offsets differ for %s",
                    source["id"],
                )
                continue
            enriched["document_noise_spans"] = detect_pdf_document_noise(pages)
            sources[index] = enriched
    items = _build_preview_items(task, sources)
    if not items and source_file_ids is None and is_unstructured:
        raise InvalidStateError("source files did not produce preview items")
    return items


def _run_preview(
    store: DataProcessStore,
    storage: LocalDataProcessStorage,
    task_id: str,
    preview_run_id: str,
    source_file_ids: list[str],
) -> None:
    """后台逐文件切分；所有写入均由 preview_run_id 保护。"""

    started_at = time.perf_counter()
    logger.info(
        "data process preview started task_id=%s preview_run_id=%s total_files=%s",
        task_id,
        preview_run_id,
        len(source_file_ids),
    )
    try:
        is_unstructured = store.get_task(task_id).get("process_type") == "unstructured"
        if not store.mark_preview_running(task_id, preview_run_id):
            logger.info(
                "data process preview skipped inactive run task_id=%s preview_run_id=%s",
                task_id,
                preview_run_id,
            )
            return
        total_files = len(source_file_ids)
        total_items = 0
        for completed_files, source_file_id in enumerate(source_file_ids, start=1):
            if not store.preview_is_running(task_id, preview_run_id):
                logger.info(
                    "data process preview cancelled task_id=%s preview_run_id=%s "
                    "completed_files=%s total_files=%s",
                    task_id,
                    preview_run_id,
                    completed_files - 1,
                    total_files,
                )
                return
            items = _prepare_preview_items(
                task_id,
                store,
                storage,
                [source_file_id],
            )
            if not items and is_unstructured:
                raise InvalidStateError(
                    f"source file did not produce preview items: {source_file_id}"
                )
            created = store.replace_preview_items(
                task_id,
                items,
                source_file_ids=[source_file_id],
                preview_run_id=preview_run_id,
            )
            total_items += len(created)
            if not store.update_preview_progress(
                task_id,
                preview_run_id,
                completed_files,
                total_files,
            ):
                logger.info(
                    "data process preview stopped before progress update task_id=%s "
                    "preview_run_id=%s completed_files=%s total_files=%s",
                    task_id,
                    preview_run_id,
                    completed_files,
                    total_files,
                )
                return
        if store.complete_preview(task_id, preview_run_id):
            logger.info(
                "data process preview completed task_id=%s preview_run_id=%s "
                "total_files=%s total_items=%s duration_ms=%.2f",
                task_id,
                preview_run_id,
                total_files,
                total_items,
                (time.perf_counter() - started_at) * 1000,
            )
        else:
            logger.info(
                "data process preview completion ignored for inactive run task_id=%s "
                "preview_run_id=%s",
                task_id,
                preview_run_id,
            )
    except Exception as exc:
        logger.exception(
            "data process preview failed task_id=%s preview_run_id=%s duration_ms=%.2f",
            task_id,
            preview_run_id,
            (time.perf_counter() - started_at) * 1000,
        )
        try:
            if store.preview_is_running(task_id, preview_run_id):
                store.mark_preview_failed(
                    task_id,
                    str(exc),
                    preview_run_id=preview_run_id,
                )
        except Exception:
            logger.exception(
                "failed to persist data process preview failure task_id=%s "
                "preview_run_id=%s",
                task_id,
                preview_run_id,
            )


@router.post("/{task_id}/preview/start", status_code=202)
def start_preview(
    task_id: str,
    background_tasks: BackgroundTasks,
    payload: PreviewBuildRequest = Body(default_factory=PreviewBuildRequest),
    store: DataProcessStore = Depends(get_data_process_store),
    storage: LocalDataProcessStorage = Depends(get_data_process_storage),
) -> dict[str, Any]:
    with api_errors():
        task, selected_ids = store.start_preview(
            task_id,
            source_file_ids=payload.source_file_ids,
        )
        preview_run_id = str(task["preview_run_id"])
        progress = store.preview_progress(task_id)
        background_tasks.add_task(
            _run_preview,
            store,
            storage,
            task_id,
            preview_run_id,
            selected_ids,
        )
        return ok(progress, "data process preview started")


@router.get("/{task_id}/preview/progress")
def preview_progress(
    task_id: str,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        return ok(store.preview_progress(task_id))


@router.post("/{task_id}/preview/build")
def build_preview(
    task_id: str,
    payload: PreviewBuildRequest = Body(default_factory=PreviewBuildRequest),
    store: DataProcessStore = Depends(get_data_process_store),
    storage: LocalDataProcessStorage = Depends(get_data_process_storage),
) -> dict[str, Any]:
    with api_errors():
        selected_ids = payload.source_file_ids
        items = _prepare_preview_items(task_id, store, storage, selected_ids)
        created = store.replace_preview_items(
            task_id,
            items,
            source_file_ids=selected_ids,
        )
        target_ids = selected_ids or [
            str(source["id"]) for source in store.list_source_files(task_id)
        ]
        file_counts = dict.fromkeys(target_ids, 0)
        for item in created:
            source_file_id = str(item.get("source_file_id") or "")
            if source_file_id in file_counts:
                file_counts[source_file_id] += 1
        files = [
            {
                "source_file_id": source_file_id,
                "preview_count": count,
                "status": "completed" if count else "empty",
            }
            for source_file_id, count in file_counts.items()
        ]
        return ok(
            {
                "items": created,
                "total": len(created),
                "page": 1,
                "page_size": len(created),
                "file_counts": file_counts,
                "files": files,
            },
            "preview built",
        )


@router.get("/{task_id}/preview")
def preview_items(
    task_id: str,
    source_file_id: str | None = Query(default=None),
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=200, ge=1, le=1000),
    keyword: str | None = Query(default=None),
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        return ok(
            store.list_preview_items(
                task_id,
                source_file_id=source_file_id,
                page=page,
                page_size=page_size,
                keyword=keyword,
            )
        )


@router.post("/{task_id}/preview")
def create_preview_item(
    task_id: str,
    payload: PreviewItemCreate,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        task = store.get_task(task_id)
        item_payload = payload.model_dump(mode="json")
        content = payload.edited_content
        item_payload["token_count"] = estimate_token_count(content)
        item_payload["quality_score"] = _preview_quality(
            content,
            task.get("config") or {},
        )
        if not content.strip():
            item_payload["status"] = "invalid"
        item = store.create_preview_item(task_id, item_payload)
        return ok(item, "preview item created")


@router.put("/{task_id}/preview/{preview_id}")
def update_preview_item(
    task_id: str,
    preview_id: str,
    payload: PreviewItemUpdate,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        task = store.get_task(task_id)
        existing = store.get_preview_item(task_id, preview_id)
        update = payload.model_dump(exclude_unset=True, mode="json")
        quality = _preview_quality(payload.edited_content, task.get("config") or {})
        source_locator = (existing.get("quality_score") or {}).get("source_locator")
        if isinstance(source_locator, Mapping):
            quality["source_locator"] = deepcopy(dict(source_locator))
        update["quality_score"] = quality
        item = store.update_preview_item(
            task_id,
            preview_id,
            update,
        )
        return ok(item, "preview item updated")


@router.delete("/{task_id}/preview/{preview_id}")
def delete_preview_item(
    task_id: str,
    preview_id: str,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        store.delete_preview_item(task_id, preview_id)
        return ok({"deleted": preview_id}, "preview item deleted")


def _start_generation(
    task_id: str,
    payload: GenerateRequest,
    background_tasks: BackgroundTasks,
    store: DataProcessStore,
) -> dict[str, Any]:
    with api_errors():
        task = store.start_generation(task_id, replace_existing=payload.replace_existing)
        background_tasks.add_task(
            _run_generation,
            store,
            task_id,
            str(task["generation_run_id"]),
        )
        return ok(store.progress(task_id), "data process generation started")


@router.post("/{task_id}/generate")
def generate(
    task_id: str,
    background_tasks: BackgroundTasks,
    payload: GenerateRequest = Body(default_factory=GenerateRequest),
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    return _start_generation(task_id, payload, background_tasks, store)


@router.post("/{task_id}/start")
def start(
    task_id: str,
    background_tasks: BackgroundTasks,
    payload: GenerateRequest = Body(default_factory=GenerateRequest),
    store: DataProcessStore = Depends(get_data_process_store),
    storage: LocalDataProcessStorage = Depends(get_data_process_storage),
) -> dict[str, Any]:
    with api_errors():
        items = _prepare_preview_items(task_id, store, storage)
        store.replace_preview_items(task_id, items)
    return _start_generation(task_id, payload, background_tasks, store)


@router.post("/{task_id}/stop")
def stop(
    task_id: str,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        store.stop_task(task_id)
        return ok(store.progress(task_id), "data process task stopped")


@router.get("/{task_id}/progress")
def progress(
    task_id: str,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        return ok(store.progress(task_id))


@router.get("/{task_id}/results")
def results(
    task_id: str,
    page: int = Query(default=1, ge=1),
    page_size: int = Query(default=100, ge=1, le=1000),
    status: Literal["valid", "modified", "invalid"] | None = Query(default=None),
    split: Literal["train", "validation", "test"] | None = Query(default=None),
    keyword: str | None = Query(default=None),
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        return ok(
            store.list_results(
                task_id,
                page=page,
                page_size=page_size,
                status=status,
                split=split,
                keyword=keyword,
            )
        )


@router.post("/{task_id}/confirm-results")
def confirm_results(
    task_id: str,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        return ok(store.confirm_results(task_id), "data process results confirmed")


@router.put("/{task_id}/results/{result_id}")
def update_result(
    task_id: str,
    result_id: str,
    payload: ResultUpdate,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        task = store.get_task(task_id)
        current = store.get_result(task_id, result_id)
        update = payload.model_dump(exclude_unset=True, mode="json")
        merged = {**current, **update}
        preview_id = current.get("preview_item_id")
        source_content = ""
        if preview_id:
            preview = store.get_preview_item(task_id, str(preview_id))
            source_content = str(
                preview.get("edited_content") or preview.get("original_content") or ""
            )
        minimum = max(
            1,
            int(
                _value(
                    task.get("config") or {},
                    "min_output_length",
                    "minOutputLength",
                    20,
                )
                or 20
            ),
        )
        # 编辑后内容已变化：重算规则与语义层，旧的评审分不再可信直接丢弃。
        update["quality_score"] = reevaluate_edited_record(
            merged,
            source_content=source_content,
            previous_quality=current.get("quality_score"),
            min_output_length=minimum,
        )
        result = store.update_result(
            task_id,
            result_id,
            update,
        )
        return ok(result, "data process result updated")


@router.post("/{task_id}/results/{result_id}/restore")
def restore_result(
    task_id: str,
    result_id: str,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        task = store.get_task(task_id)
        current = store.get_result(task_id, result_id)
        restored = {
            **current,
            "instruction": current.get("original_instruction") or current.get("instruction") or "",
            "input": current.get("original_input") or current.get("input") or "",
            "output": current.get("original_output") or current.get("output") or "",
            "chosen": current.get("original_chosen") or current.get("chosen") or "",
            "rejected": current.get("original_rejected") or current.get("rejected") or "",
        }
        preview_id = current.get("preview_item_id")
        source_content = ""
        if preview_id:
            preview = store.get_preview_item(task_id, str(preview_id))
            source_content = str(
                preview.get("edited_content") or preview.get("original_content") or ""
            )
        minimum = max(
            1,
            int(
                _value(
                    task.get("config") or {},
                    "min_output_length",
                    "minOutputLength",
                    20,
                )
                or 20
            ),
        )
        restored = store.update_result(
            task_id,
            result_id,
            {
                "instruction": restored["instruction"],
                "input": restored["input"],
                "output": restored["output"],
                "chosen": restored["chosen"],
                "rejected": restored["rejected"],
                "quality_score": reevaluate_edited_record(
                    restored,
                    source_content=source_content,
                    previous_quality=current.get("quality_score"),
                    min_output_length=minimum,
                ),
                "expected_updated_at": current.get("updated_at"),
            },
        )
        return ok(restored, "data process result restored")


class _ResultRegenerationFailed(InvalidStateError):
    """模型返回或质量校验失败，原失败结果必须保持不变。"""


def _assert_result_regeneration_allowed(task: dict[str, Any]) -> None:
    if task.get("status") != "completed" or task.get("workflow_step") != "results":
        raise InvalidStateError("task is not editing generation results")
    if task.get("results_confirmed"):
        raise InvalidStateError("confirmed results cannot be regenerated")
    if task.get("output_dataset_id"):
        raise InvalidStateError("published results cannot be regenerated")


def _result_regeneration_model(
    task: dict[str, Any],
    store: DataProcessStore,
) -> tuple[dict[str, Any], dict[str, Any]]:
    config = task.get("config") or {}
    model_id = _value(config, "generation_model_id", "generationModelId", None)
    if not model_id:
        raise InvalidStateError("task does not have a generation model")
    return config, store.get_generation_model(str(model_id))


def _result_regeneration_timeout(config: dict[str, Any]) -> float:
    configured = float(
        _value(config, "request_timeout_seconds", "requestTimeoutSeconds", 60)
    )
    return max(1.0, min(RESULT_REGENERATION_TIMEOUT_SECONDS, configured))


@contextmanager
def _claim_result_regeneration(task_id: str, result_id: str) -> Iterator[None]:
    key = (task_id, result_id)
    with _result_regeneration_claims_lock:
        if key in _active_result_regenerations:
            raise ConflictError("data process result regeneration is already running")
        _active_result_regenerations.add(key)
    try:
        yield
    finally:
        with _result_regeneration_claims_lock:
            _active_result_regenerations.discard(key)


def _generate_result_replacement(
    task_id: str,
    current: dict[str, Any],
    preview: dict[str, Any],
    config: dict[str, Any],
    generation_model: dict[str, Any],
    model_client: httpx.Client | None = None,
) -> dict[str, Any]:
    source_content = str(
        preview.get("edited_content") or preview.get("original_content") or ""
    ).strip()
    if not source_content or preview.get("status") == "invalid":
        raise InvalidStateError("result source preview item is invalid or empty")

    output_type = str(
        _value(config, "output_type", "outputType", "standard")
    ).strip().lower()
    previous_instruction = str(current.get("instruction") or "")[:1000]
    previous_output = str(current.get("output") or "")[:1000]
    previous_rejected = str(current.get("rejected") or "")[:1000]
    base_prompt = str(
        _value(config, "generation_prompt", "generationPrompt", "") or ""
    )
    regeneration_instruction = (
        "这是一次失败结果的重新生成。请使用新的提问角度和表达，"
        "不要复述旧结果。旧问题："
        f"{previous_instruction or '无'}；旧优选答案：{previous_output or '无'}；"
        f"旧拒选答案：{previous_rejected or '无'}。"
    )
    runtime_config = {
        **config,
        "generation_prompt": f"{base_prompt}\n{regeneration_instruction}".strip(),
        "output_type": output_type,
        "reasoning_detail": _value(
            config, "reasoning_detail", "reasoningDetail", "normal"
        ),
        "max_tokens": _value(config, "max_tokens", "maxTokens", 1024),
        "json_mode": _value(config, "json_mode", "jsonMode", False),
        # 交互式重新生成只做一次新尝试，避免继承整任务的重试配置后长时间等待。
        "generation_retries": RESULT_REGENERATION_RETRIES,
        "request_timeout_seconds": _result_regeneration_timeout(config),
    }
    with _result_regeneration_slots:
        generated = generate_model_records(
            [preview],
            model=generation_model,
            config=runtime_config,
            task_id=task_id,
            split={"train": 100, "validation": 0, "test": 0},
            qa_pairs_per_item=1,
            client=model_client,
        )
    if not generated or generated[0].get("status") == "invalid":
        reason = str(
            (generated[0] if generated else {}).get("error")
            or "model generation failed"
        )
        raise _ResultRegenerationFailed(f"重新生成结果仍无效：{reason}")

    minimum = max(
        1,
        int(_value(config, "min_output_length", "minOutputLength", 20) or 20),
    )
    replacement = generated[0]
    quality = score_quality(
        replacement,
        min_output_length=minimum,
        source_content=source_content,
    )
    if not quality.is_valid:
        reason = ", ".join(quality.flags) or "quality validation failed"
        raise _ResultRegenerationFailed(f"重新生成结果未通过质量校验：{reason}")
    replacement["quality_score"] = asdict(quality)
    replacement["status"] = "valid"
    replacement["error"] = None
    return replacement


def _regenerate_result_in_place(
    task_id: str,
    current: dict[str, Any],
    preview: dict[str, Any],
    config: dict[str, Any],
    generation_model: dict[str, Any],
    store: DataProcessStore,
    *,
    expected_updated_at: str,
    model_client: httpx.Client | None = None,
) -> dict[str, Any]:
    result_id = str(current["id"])
    with _claim_result_regeneration(task_id, result_id):
        replacement = _generate_result_replacement(
            task_id,
            current,
            preview,
            config,
            generation_model,
            model_client,
        )
        return store.replace_generated_result(
            task_id,
            result_id,
            replacement,
            expected_updated_at=expected_updated_at,
        )


def _safe_regeneration_error(exc: Exception) -> str:
    return re.sub(r"\s+", " ", str(exc)).strip()[:500] or "result regeneration failed"


def _evaluate_result_in_place(
    task_id: str,
    current: dict[str, Any],
    source_content: str,
    config: dict[str, Any],
    evaluation_model: dict[str, Any] | None,
    store: DataProcessStore,
    *,
    expected_updated_at: str,
    model_client: httpx.Client | None = None,
    minimum: int = 20,
) -> dict[str, Any]:
    """评测单条结果并落库；复用逐结果互斥锁避免与重生成并发写冲突。"""

    result_id = str(current["id"])
    with _claim_result_regeneration(task_id, result_id):
        quality = evaluate_result_record(
            {
                "instruction": current.get("instruction"),
                "input": current.get("input"),
                "output": current.get("output"),
                "chosen": current.get("chosen"),
                "rejected": current.get("rejected"),
            },
            source_content=source_content,
            model=evaluation_model,
            config=config,
            client=model_client,
            min_output_length=minimum,
        )
        return store.update_result(
            task_id,
            result_id,
            {
                "quality_score": quality,
                "expected_updated_at": expected_updated_at,
            },
        )


@router.post("/{task_id}/results/regenerate-batch")
def regenerate_results_batch(
    task_id: str,
    payload: ResultBatchRegenerateRequest,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    """并发重新生成一批失败结果；每条独立提交并允许部分成功。"""

    started_at = time.perf_counter()
    batch_id = new_id("dprb")
    with api_errors():
        task = store.get_task(task_id)
        _assert_result_regeneration_allowed(task)
        config, generation_model = _result_regeneration_model(task, store)
        prepared: list[tuple[int, dict[str, Any], dict[str, Any], str]] = []
        failures: list[tuple[int, dict[str, str]]] = []

        for index, requested in enumerate(payload.items):
            try:
                current = store.get_result(task_id, requested.result_id)
                if current.get("status") != "invalid":
                    raise InvalidStateError("only an invalid result can be regenerated")
                if requested.expected_updated_at != str(current.get("updated_at") or ""):
                    raise ConflictError("data process result was modified by another request")
                preview_id = current.get("preview_item_id")
                if not preview_id:
                    raise InvalidStateError(
                        "result is not associated with a source preview item"
                    )
                preview = store.get_preview_item(task_id, str(preview_id))
                prepared.append(
                    (index, current, preview, requested.expected_updated_at)
                )
            except ConflictError as exc:
                failures.append((index, {
                    "result_id": requested.result_id,
                    "code": "conflict",
                    "message": _safe_regeneration_error(exc),
                }))
            except (NotFoundError, InvalidStateError) as exc:
                failures.append((index, {
                    "result_id": requested.result_id,
                    "code": "skipped",
                    "message": _safe_regeneration_error(exc),
                }))

        logger.info(
            "data process result batch regeneration started batch_id=%s task_id=%s "
            "requested=%s prepared=%s concurrency=%s",
            batch_id,
            task_id,
            len(payload.items),
            len(prepared),
            min(RESULT_REGENERATION_CONCURRENCY, len(prepared)),
        )
        successes: list[tuple[int, dict[str, Any]]] = []
        if prepared:
            request_timeout = _result_regeneration_timeout(config)
            model_timeout = httpx.Timeout(
                request_timeout,
                connect=min(10.0, request_timeout),
            )
            model_limits = httpx.Limits(
                max_connections=RESULT_REGENERATION_CONCURRENCY,
                max_keepalive_connections=RESULT_REGENERATION_CONCURRENCY,
            )
            # httpx.Client 支持跨线程复用，批次内共享连接池可减少重复建连开销。
            with httpx.Client(timeout=model_timeout, limits=model_limits) as model_client, \
                    ThreadPoolExecutor(
                        max_workers=min(RESULT_REGENERATION_CONCURRENCY, len(prepared)),
                        thread_name_prefix="data-result-regeneration",
                    ) as executor:
                futures = {
                    executor.submit(
                        _regenerate_result_in_place,
                        task_id,
                        current,
                        preview,
                        config,
                        generation_model,
                        store,
                        expected_updated_at=expected_updated_at,
                        model_client=model_client,
                    ): (index, str(current["id"]), time.perf_counter())
                    for index, current, preview, expected_updated_at in prepared
                }
                for future in as_completed(futures):
                    index, result_id, item_started_at = futures[future]
                    try:
                        regenerated = future.result()
                        successes.append((index, regenerated))
                        outcome = "succeeded"
                    except ConflictError as exc:
                        outcome = "conflict"
                        failures.append((index, {
                            "result_id": result_id,
                            "code": outcome,
                            "message": _safe_regeneration_error(exc),
                        }))
                    except _ResultRegenerationFailed as exc:
                        outcome = "generation_failed"
                        failures.append((index, {
                            "result_id": result_id,
                            "code": outcome,
                            "message": _safe_regeneration_error(exc),
                        }))
                    except (NotFoundError, InvalidStateError) as exc:
                        outcome = "skipped"
                        failures.append((index, {
                            "result_id": result_id,
                            "code": outcome,
                            "message": _safe_regeneration_error(exc),
                        }))
                    except Exception as exc:  # pragma: no cover - defensive boundary
                        outcome = "internal_error"
                        logger.exception(
                            "data process result batch regeneration crashed "
                            "batch_id=%s task_id=%s result_id=%s",
                            batch_id,
                            task_id,
                            result_id,
                        )
                        failures.append((index, {
                            "result_id": result_id,
                            "code": outcome,
                            "message": _safe_regeneration_error(exc),
                        }))
                    logger.info(
                        "data process result batch item finished batch_id=%s task_id=%s "
                        "result_id=%s outcome=%s duration_ms=%.2f",
                        batch_id,
                        task_id,
                        result_id,
                        outcome,
                        (time.perf_counter() - item_started_at) * 1000,
                    )

        success_items = [item for _, item in sorted(successes, key=lambda pair: pair[0])]
        failure_items = [item for _, item in sorted(failures, key=lambda pair: pair[0])]
        remaining_invalid_count = int(
            store.list_results(
                task_id,
                page=1,
                page_size=1,
                status="invalid",
            )["total"]
        )
        duration_ms = (time.perf_counter() - started_at) * 1000
        logger.info(
            "data process result batch regeneration completed batch_id=%s task_id=%s "
            "succeeded=%s failed=%s remaining_invalid=%s duration_ms=%.2f",
            batch_id,
            task_id,
            len(success_items),
            len(failure_items),
            remaining_invalid_count,
            duration_ms,
        )
        return ok(
            {
                "batch_id": batch_id,
                "total": len(payload.items),
                "succeeded": len(success_items),
                "failed": len(failure_items),
                "remaining_invalid_count": remaining_invalid_count,
                "duration_ms": round(duration_ms, 2),
                "items": success_items,
                "failures": failure_items,
            },
            "data process results regenerated",
        )


@router.post("/{task_id}/results/evaluate-batch")
def evaluate_results_batch(
    task_id: str,
    payload: ResultBatchEvaluateRequest,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    """对一批结果执行三层质量评测（规则+语义+评审），允许部分成功。"""

    started_at = time.perf_counter()
    batch_id = new_id("dpeb")
    with api_errors():
        task = store.get_task(task_id)
        if task.get("status") == "running":
            raise ConflictError("data process task is running")
        if task.get("output_dataset_id"):
            raise InvalidStateError("published results cannot be evaluated")
        config = task.get("config") or {}
        evaluation_model: dict[str, Any] | None = None
        model_id = _value(config, "generation_model_id", "generationModelId", None)
        if model_id:
            try:
                evaluation_model = store.get_generation_model(str(model_id))
            except NotFoundError:
                logger.warning(
                    "data process evaluation model unavailable, judge layer "
                    "skipped task_id=%s model_id=%s",
                    task_id,
                    model_id,
                )
        evaluation_config = {
            **config,
            "output_type": str(
                _value(config, "output_type", "outputType", "standard")
            ).strip().lower(),
        }
        minimum = max(
            1,
            int(_value(config, "min_output_length", "minOutputLength", 20) or 20),
        )

        prepared: list[tuple[int, dict[str, Any], str, str]] = []
        failures: list[tuple[int, dict[str, str]]] = []
        for index, requested in enumerate(payload.items):
            try:
                current = store.get_result(task_id, requested.result_id)
                if requested.expected_updated_at != str(current.get("updated_at") or ""):
                    raise ConflictError("data process result was modified by another request")
                source_content = ""
                preview_id = current.get("preview_item_id")
                if preview_id:
                    preview = store.get_preview_item(task_id, str(preview_id))
                    source_content = str(
                        preview.get("edited_content")
                        or preview.get("original_content")
                        or ""
                    )
                prepared.append(
                    (index, current, source_content, requested.expected_updated_at)
                )
            except ConflictError as exc:
                failures.append((index, {
                    "result_id": requested.result_id,
                    "code": "conflict",
                    "message": _safe_regeneration_error(exc),
                }))
            except (NotFoundError, InvalidStateError) as exc:
                failures.append((index, {
                    "result_id": requested.result_id,
                    "code": "skipped",
                    "message": _safe_regeneration_error(exc),
                }))

        logger.info(
            "data process result batch evaluation started batch_id=%s task_id=%s "
            "requested=%s prepared=%s judge_enabled=%s",
            batch_id,
            task_id,
            len(payload.items),
            len(prepared),
            evaluation_model is not None,
        )
        successes: list[tuple[int, dict[str, Any]]] = []
        if prepared:
            try:
                from app.modules.data_process.algorithms.embedding import (
                    semantic_embedding_model,
                )

                semantic_embedding_model()
            except Exception:
                logger.warning(
                    "data process semantic embedding unavailable, semantic layer "
                    "will be skipped batch_id=%s",
                    batch_id,
                )
            request_timeout = _result_regeneration_timeout(config)
            model_timeout = httpx.Timeout(
                request_timeout,
                connect=min(10.0, request_timeout),
            )
            model_limits = httpx.Limits(
                max_connections=RESULT_REGENERATION_CONCURRENCY,
                max_keepalive_connections=RESULT_REGENERATION_CONCURRENCY,
            )
            with httpx.Client(timeout=model_timeout, limits=model_limits) as model_client, \
                    ThreadPoolExecutor(
                        max_workers=min(RESULT_REGENERATION_CONCURRENCY, len(prepared)),
                        thread_name_prefix="data-result-evaluation",
                    ) as executor:
                futures = {
                    executor.submit(
                        _evaluate_result_in_place,
                        task_id,
                        current,
                        source_content,
                        evaluation_config,
                        evaluation_model,
                        store,
                        expected_updated_at=expected_updated_at,
                        model_client=model_client if evaluation_model else None,
                        minimum=minimum,
                    ): (index, str(current["id"]), time.perf_counter())
                    for index, current, source_content, expected_updated_at in prepared
                }
                for future in as_completed(futures):
                    index, result_id, item_started_at = futures[future]
                    try:
                        evaluated = future.result()
                        successes.append((index, evaluated))
                        outcome = "succeeded"
                    except ConflictError as exc:
                        outcome = "conflict"
                        failures.append((index, {
                            "result_id": result_id,
                            "code": outcome,
                            "message": _safe_regeneration_error(exc),
                        }))
                    except Exception as exc:
                        outcome = "evaluation_failed"
                        failures.append((index, {
                            "result_id": result_id,
                            "code": outcome,
                            "message": _safe_regeneration_error(exc),
                        }))
                    logger.info(
                        "data process result batch evaluation item finished "
                        "batch_id=%s task_id=%s result_id=%s outcome=%s duration_ms=%.2f",
                        batch_id,
                        task_id,
                        result_id,
                        outcome,
                        (time.perf_counter() - item_started_at) * 1000,
                    )

        success_items = [item for _, item in sorted(successes, key=lambda pair: pair[0])]
        failure_items = [item for _, item in sorted(failures, key=lambda pair: pair[0])]
        duration_ms = (time.perf_counter() - started_at) * 1000
        logger.info(
            "data process result batch evaluation completed batch_id=%s task_id=%s "
            "succeeded=%s failed=%s duration_ms=%.2f",
            batch_id,
            task_id,
            len(success_items),
            len(failure_items),
            duration_ms,
        )
        return ok(
            {
                "batch_id": batch_id,
                "total": len(payload.items),
                "succeeded": len(success_items),
                "failed": len(failure_items),
                "duration_ms": round(duration_ms, 2),
                "items": success_items,
                "failures": failure_items,
            },
            "data process results evaluated",
        )


@router.post("/{task_id}/results/{result_id}/regenerate")
def regenerate_result(
    task_id: str,
    result_id: str,
    payload: ResultRegenerateRequest,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    """只重新生成一个失败结果，成功后原位替换且不影响其他结果。"""

    started_at = time.perf_counter()
    with api_errors():
        task = store.get_task(task_id)
        _assert_result_regeneration_allowed(task)
        current = store.get_result(task_id, result_id)
        if current.get("status") != "invalid":
            raise InvalidStateError("only an invalid result can be regenerated")
        if payload.expected_updated_at != str(current.get("updated_at") or ""):
            raise ConflictError("data process result was modified by another request")
        preview_id = current.get("preview_item_id")
        if not preview_id:
            raise InvalidStateError("result is not associated with a source preview item")
        preview = store.get_preview_item(task_id, str(preview_id))
        config, generation_model = _result_regeneration_model(task, store)
        try:
            result = _regenerate_result_in_place(
                task_id,
                current,
                preview,
                config,
                generation_model,
                store,
                expected_updated_at=payload.expected_updated_at,
            )
        except _ResultRegenerationFailed as exc:
            logger.warning(
                "data process result regeneration failed task_id=%s result_id=%s "
                "duration_ms=%.2f reason=%s",
                task_id,
                result_id,
                (time.perf_counter() - started_at) * 1000,
                _safe_regeneration_error(exc),
            )
            raise
        logger.info(
            "data process result regenerated task_id=%s result_id=%s duration_ms=%.2f",
            task_id,
            result_id,
            (time.perf_counter() - started_at) * 1000,
        )
        return ok(result, "data process result regenerated")


@router.post("/{task_id}/publish")
def publish(
    task_id: str,
    payload: PublishRequest,
    store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
    with api_errors():
        result = store.publish(task_id, payload.model_dump(mode="json"))
        message = "dataset published" if result["created"] else "dataset already published"
        return ok(result, message)