YG_FT/backend/tests/test_data_process_api.py

from __future__ import annotations

from copy import deepcopy
from io import BytesIO
from pathlib import Path
from threading import Barrier, Lock
from typing import Any

import psycopg
import pytest
from docx import Document as WordDocument
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient
from openpyxl import Workbook

from app.api.v1.endpoints import data_process as data_process_endpoint
from app.api.v1.endpoints.data_process import router
from app.modules.data_process.algorithms import DocumentNoiseSpan, normalize_text
from app.modules.data_process.storage import (
    DataProcessStorageError,
    LocalDataProcessStorage,
    get_data_process_storage,
)
from app.modules.data_process.store import InvalidStateError, NotFoundError, get_data_process_store


class FakeDataProcessStore:
    """接口测试专用内存实现，确保测试不会连接或迁移真实数据库。"""

    def __init__(self) -> None:
        self.tasks: dict[str, dict[str, Any]] = {}
        self.sources: dict[str, list[dict[str, Any]]] = {}
        self.previews: dict[str, list[dict[str, Any]]] = {}
        self.results: dict[str, list[dict[str, Any]]] = {}
        self.datasets: dict[str, dict[str, Any]] = {}
        self.models: dict[str, dict[str, Any]] = {}
        self.regeneration_prepared: set[str] = set()
        self.sequence = 0

    def _id(self, prefix: str) -> str:
        self.sequence += 1
        return f"{prefix}_{self.sequence}"

    def list_tasks(self, *, page: int, page_size: int, **filters: Any) -> dict[str, Any]:
        items = [
            {
                **item,
                "source_file_count": len(self.sources.get(str(item["id"]), [])),
            }
            for item in self.tasks.values()
        ]
        for field in ("status", "process_type", "tenant_id", "project_id"):
            if filters.get(field):
                items = [item for item in items if item.get(field) == filters[field]]
        keyword = filters.get("keyword")
        if keyword:
            items = [item for item in items if keyword in item["name"]]
        return {
            "items": deepcopy(items[(page - 1) * page_size : page * page_size]),
            "total": len(items),
            "page": page,
            "page_size": page_size,
        }

    def create_task(self, payload: dict[str, Any]) -> dict[str, Any]:
        task_id = self._id("dpt")
        task = {
            "id": task_id,
            **deepcopy(payload),
            "status": "pending",
            "progress": 0,
            "input_count": 0,
            "output_count": 0,
            "filtered_count": 0,
            "duplicate_count": 0,
            "error_count": 0,
            "failure_reason": None,
            "output_dataset_id": None,
            "results_confirmed": False,
            "workflow_step": "create",
            "preview_status": "idle",
            "preview_progress": 0,
            "preview_run_id": None,
            "preview_failure_reason": None,
            "preview_total_files": 0,
            "preview_completed_files": 0,
        }
        self.tasks[task_id] = task
        self.sources[task_id] = []
        self.previews[task_id] = []
        self.results[task_id] = []
        return deepcopy(task)

    def get_task(self, task_id: str) -> dict[str, Any]:
        if task_id not in self.tasks:
            raise NotFoundError("data process task not found")
        task = deepcopy(self.tasks[task_id])
        split_order = {"train": 0, "val": 1, "test": 2}
        task["output_datasets"] = sorted(
            (
                deepcopy(dataset)
                for dataset in self.datasets.values()
                if dataset.get("source_task_id") == task_id
                and dataset.get("deleted_at") is None
            ),
            key=lambda dataset: split_order.get(str(dataset.get("type")), 3),
        )
        return task

    def recover_legacy_aborted_regeneration(self, task_id: str) -> dict[str, Any]:
        self.get_task(task_id)
        return {"recovered": False, "result_count": 0}

    def update_task(self, task_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        self.get_task(task_id)
        self.tasks[task_id].update(deepcopy(payload))
        return self.get_task(task_id)

    def update_workflow_step(self, task_id: str, workflow_step: str) -> dict[str, Any]:
        self.get_task(task_id)
        self.tasks[task_id]["workflow_step"] = workflow_step
        return self.get_task(task_id)

    def prepare_regeneration(self, task_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        task = self.tasks.get(task_id)
        if task is None:
            raise NotFoundError("data process task not found")
        if task["status"] == "running":
            raise InvalidStateError("running task cannot be prepared for regeneration")
        if payload["expected_updated_at"] != task.get("updated_at"):
            raise InvalidStateError("data process task was modified by another request")
        if payload["process_type"] != task["process_type"]:
            raise InvalidStateError("process_type cannot be changed during regeneration")
        published_outputs_preserved = bool(task.get("output_dataset_id")) or any(
            dataset.get("source_task_id") == task_id
            for dataset in self.datasets.values()
        )
        task.update(
            {
                "name": payload["name"],
                "description": payload["description"],
                "config": deepcopy(payload["config"]),
                "updated_at": "2026-07-25T20:00:00Z",
            }
        )
        self.regeneration_prepared.add(task_id)
        return {
            "task": deepcopy(task),
            "preview_invalidated": False,
            "published_outputs_preserved": published_outputs_preserved,
        }

    def delete_task(self, task_id: str, **_: Any) -> None:
        self.get_task(task_id)
        del self.tasks[task_id]

    def list_source_files(self, task_id: str) -> list[dict[str, Any]]:
        self.get_task(task_id)
        return [
            {key: value for key, value in item.items() if key != "content"}
            for item in self.sources[task_id]
        ]

    def add_source_file(self, task_id: str, **payload: Any) -> dict[str, Any]:
        self.get_task(task_id)
        values = deepcopy(payload)
        source_id = str(values.pop("id", None) or self._id("dpsf"))
        storage_object_id = str(
            values.pop("storage_object_id", None)
            or f"db://data-process/{task_id}/{source_id}/v1"
        )
        raw_size = int(values.pop("raw_size"))
        metadata = deepcopy(values.pop("metadata", {}))
        metadata.setdefault(
            "storage_backend",
            "local" if storage_object_id.startswith("local://data-process/") else "database",
        )
        source = {
            "id": source_id,
            "task_id": task_id,
            "version_no": 1,
            "storage_object_id": storage_object_id,
            "size_bytes": raw_size,
            "metadata": metadata,
            **values,
        }
        self.sources[task_id].append(source)
        self.tasks[task_id]["input_count"] += payload["record_count"]
        return {key: value for key, value in deepcopy(source).items() if key != "content"}

    def add_source_files(
        self, task_id: str, files: list[dict[str, Any]]
    ) -> list[dict[str, Any]]:
        # 先验证整个批次，模拟数据库事务的 all-or-nothing 语义。
        checksums = {item["checksum_sha256"] for item in self.sources.get(task_id, [])}
        incoming: set[str] = set()
        for payload in files:
            checksum = payload["checksum_sha256"]
            if checksum in checksums or checksum in incoming:
                raise ValueError("the same source file content is already attached to this task")
            incoming.add(checksum)
        created = [self.add_source_file(task_id, **payload) for payload in files]
        self.results[task_id] = []
        self.tasks[task_id].update(
            {
                "status": "pending",
                "progress": 20 if self.previews[task_id] else 0,
                "output_count": 0,
                "filtered_count": 0,
                "duplicate_count": 0,
                "error_count": 0,
                "failure_reason": None,
                "generation_run_id": None,
                "workflow_step": "upload",
                "preview_status": "idle",
                "preview_progress": 0,
                "preview_run_id": None,
                "preview_failure_reason": None,
                "preview_total_files": 0,
                "preview_completed_files": 0,
                "started_at": None,
                "completed_at": None,
            }
        )
        return created

    def get_source_file(
        self, task_id: str, file_id: str, *, include_content: bool = True
    ) -> dict[str, Any]:
        source = next(
            (item for item in self.sources.get(task_id, []) if item["id"] == file_id),
            None,
        )
        if not source:
            raise NotFoundError("source file not found")
        result = deepcopy(source)
        if not include_content:
            result.pop("content", None)
        return result

    def source_content_window(
        self, task_id: str, file_id: str, offset: int, limit: int
    ) -> dict[str, Any]:
        source = self.get_source_file(task_id, file_id)
        content = source.pop("content")
        return {
            "file": source,
            "content": content[offset : offset + limit],
            "offset": offset,
            "limit": limit,
            "total_chars": len(content),
            "has_more": offset + limit < len(content),
        }

    def source_content_lines(
        self, task_id: str, file_id: str, start_line: int, line_count: int
    ) -> dict[str, Any]:
        source = self.get_source_file(task_id, file_id)
        lines = source.pop("content").splitlines(keepends=True)
        selected = lines[start_line - 1 : start_line - 1 + line_count]
        return {
            "file": source,
            "content": "".join(selected),
            "start_line": start_line,
            "end_line": start_line - 1 + len(selected),
            "line_count": len(selected),
            "total_lines": len(lines),
            "has_more": start_line - 1 + len(selected) < len(lines),
        }

    def delete_source_file(self, task_id: str, file_id: str) -> None:
        self.get_source_file(task_id, file_id)
        self.sources[task_id] = [item for item in self.sources[task_id] if item["id"] != file_id]
        self.previews[task_id] = [
            item for item in self.previews[task_id] if item["source_file_id"] != file_id
        ]
        self.results[task_id] = []
        self.tasks[task_id].update(
            workflow_step="upload",
            preview_status="idle",
            preview_progress=0,
            preview_run_id=None,
            preview_failure_reason=None,
            preview_total_files=0,
            preview_completed_files=0,
        )

    def replace_preview_items(
        self,
        task_id: str,
        items: list[dict[str, Any]],
        *,
        source_file_ids: list[str] | None = None,
        preview_run_id: str | None = None,
    ) -> list[dict[str, Any]]:
        if preview_run_id is not None and not self.preview_is_running(task_id, preview_run_id):
            raise InvalidStateError("preview run is no longer active")
        created = [
            {"id": self._id("dpp"), "task_id": task_id, **deepcopy(item)} for item in items
        ]
        if source_file_ids is None:
            self.previews[task_id] = created
        else:
            selected = set(source_file_ids)
            self.previews[task_id] = [
                item
                for item in self.previews[task_id]
                if item["source_file_id"] not in selected
            ] + created
        self.results[task_id] = []
        self.tasks[task_id]["progress"] = 20
        if preview_run_id is None:
            file_count = len(source_file_ids or {item["source_file_id"] for item in created})
            self.tasks[task_id].update(
                workflow_step="preview",
                preview_status="completed",
                preview_progress=100,
                preview_run_id=None,
                preview_failure_reason=None,
                preview_total_files=file_count,
                preview_completed_files=file_count,
            )
        return deepcopy(created)

    def start_preview(
        self,
        task_id: str,
        *,
        source_file_ids: list[str] | None = None,
    ) -> tuple[dict[str, Any], list[str]]:
        task = self.tasks[task_id]
        if task.get("preview_status") in {"queued", "running"}:
            raise InvalidStateError("task cannot be edited while preview is running")
        selected_ids = source_file_ids or [str(item["id"]) for item in self.sources[task_id]]
        found = {str(item["id"]) for item in self.sources[task_id]}
        missing = set(selected_ids) - found
        if missing:
            raise NotFoundError(f"source files not found: {', '.join(sorted(missing))}")
        if not selected_ids:
            raise InvalidStateError("at least one source file is required")
        run_id = self._id("dpprun")
        self.results[task_id] = []
        task.update(
            status="pending",
            workflow_step="upload",
            preview_status="queued",
            preview_progress=0,
            preview_run_id=run_id,
            preview_failure_reason=None,
            preview_total_files=len(selected_ids),
            preview_completed_files=0,
            results_confirmed=False,
        )
        return self.get_task(task_id), list(selected_ids)

    def mark_preview_running(self, task_id: str, preview_run_id: str) -> bool:
        task = self.tasks.get(task_id)
        if not task or task.get("preview_status") != "queued" or task.get("preview_run_id") != preview_run_id:
            return False
        task["preview_status"] = "running"
        return True

    def preview_is_running(self, task_id: str, preview_run_id: str) -> bool:
        task = self.tasks.get(task_id)
        return bool(
            task
            and task.get("preview_status") in {"queued", "running"}
            and task.get("preview_run_id") == preview_run_id
        )

    def update_preview_progress(
        self,
        task_id: str,
        preview_run_id: str,
        completed_files: int,
        total_files: int,
    ) -> bool:
        if not self.preview_is_running(task_id, preview_run_id):
            return False
        self.tasks[task_id]["preview_completed_files"] = completed_files
        self.tasks[task_id]["preview_progress"] = completed_files / max(1, total_files) * 100
        return True

    def complete_preview(self, task_id: str, preview_run_id: str) -> bool:
        if not self.preview_is_running(task_id, preview_run_id):
            return False
        task = self.tasks[task_id]
        task.update(
            workflow_step="preview",
            preview_status="completed",
            preview_progress=100,
            preview_run_id=None,
            preview_failure_reason=None,
            preview_completed_files=task["preview_total_files"],
        )
        return True

    def mark_preview_failed(
        self,
        task_id: str,
        reason: str,
        *,
        preview_run_id: str,
    ) -> bool:
        if not self.preview_is_running(task_id, preview_run_id):
            return False
        self.tasks[task_id].update(
            preview_status="failed",
            preview_run_id=None,
            preview_failure_reason=reason,
        )
        return True

    def preview_progress(self, task_id: str) -> dict[str, Any]:
        task = self.get_task(task_id)
        return {
            "task_id": task_id,
            "workflow_step": task["workflow_step"],
            "preview_status": task["preview_status"],
            "preview_progress": task["preview_progress"],
            "preview_run_id": task["preview_run_id"],
            "preview_failure_reason": task["preview_failure_reason"],
            "preview_total_files": task["preview_total_files"],
            "preview_completed_files": task["preview_completed_files"],
        }

    def list_preview_items(
        self,
        task_id: str,
        *,
        page: int,
        page_size: int,
        source_file_id: str | None = None,
        keyword: str | None = None,
    ) -> dict[str, Any]:
        items = self.previews[task_id]
        if source_file_id:
            items = [item for item in items if item["source_file_id"] == source_file_id]
        if keyword:
            items = [item for item in items if keyword in item["edited_content"]]
        return {
            "items": deepcopy(items[(page - 1) * page_size : page * page_size]),
            "total": len(items),
            "page": page,
            "page_size": page_size,
        }

    def get_preview_item(self, task_id: str, preview_id: str) -> dict[str, Any]:
        item = next(
            (item for item in self.previews.get(task_id, []) if item["id"] == preview_id),
            None,
        )
        if not item:
            raise NotFoundError("preview item not found")
        return deepcopy(item)

    def create_preview_item(self, task_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        item = {"id": self._id("dpp"), "task_id": task_id, **deepcopy(payload)}
        self.previews[task_id].append(item)
        self.results[task_id] = []
        return deepcopy(item)

    def update_preview_item(
        self, task_id: str, preview_id: str, payload: dict[str, Any]
    ) -> dict[str, Any]:
        item = next(
            (item for item in self.previews[task_id] if item["id"] == preview_id),
            None,
        )
        if not item:
            raise NotFoundError("preview item not found")
        item.update(deepcopy(payload))
        self.results[task_id] = []
        return deepcopy(item)

    def delete_preview_item(self, task_id: str, preview_id: str) -> None:
        before = len(self.previews[task_id])
        self.previews[task_id] = [
            item for item in self.previews[task_id] if item["id"] != preview_id
        ]
        if len(self.previews[task_id]) == before:
            raise NotFoundError("preview item not found")

    def start_generation(self, task_id: str, *, replace_existing: bool) -> dict[str, Any]:
        if not self.previews[task_id]:
            raise InvalidStateError("preview must be built before generation")
        task = self.tasks[task_id]
        if task.get("output_dataset_id") and task_id not in self.regeneration_prepared:
            raise InvalidStateError("published task cannot be regenerated")
        if replace_existing:
            self.results[task_id] = []
        task.update(
            status="running",
            progress=30,
            output_dataset_id=None,
            output_count=0,
            results_confirmed=False,
            workflow_step="generate",
            generation_run_id=self._id("dprun"),
        )
        self.regeneration_prepared.discard(task_id)
        return self.get_task(task_id)

    def generation_is_running(self, task_id: str, generation_run_id: str) -> bool:
        task = self.tasks.get(task_id)
        return (
            bool(task)
            and task["status"] == "running"
            and task.get("generation_run_id") == generation_run_id
        )

    def update_generation_progress(
        self,
        task_id: str,
        generation_run_id: str,
        processed_count: int,
        total_count: int,
    ) -> bool:
        if not self.generation_is_running(task_id, generation_run_id):
            return False
        self.tasks[task_id]["progress"] = min(
            95,
            30 + processed_count / max(1, total_count) * 65,
        )
        return True

    def complete_generation(
        self,
        task_id: str,
        results: list[dict[str, Any]],
        *,
        generation_run_id: str,
        **counts: Any,
    ) -> dict[str, Any]:
        if not self.generation_is_running(task_id, generation_run_id):
            return self.get_task(task_id)
        self.results[task_id] = deepcopy(results)
        self.tasks[task_id].update(
            status="completed",
            progress=100,
            output_count=len(results),
            results_confirmed=False,
            generation_run_id=None,
            **counts,
        )
        return self.get_task(task_id)

    def mark_failed(
        self, task_id: str, reason: str, *, generation_run_id: str
    ) -> dict[str, Any]:
        if self.generation_is_running(task_id, generation_run_id):
            self.tasks[task_id].update(
                status="failed",
                failure_reason=reason,
                generation_run_id=None,
            )
        return self.get_task(task_id)

    def stop_task(self, task_id: str) -> dict[str, Any]:
        if self.tasks[task_id]["status"] != "running":
            raise InvalidStateError("only a running task can be stopped")
        self.tasks[task_id].update(status="stopped", generation_run_id=None)
        return self.get_task(task_id)

    def progress(self, task_id: str) -> dict[str, Any]:
        task = self.get_task(task_id)
        result = {key: task.get(key) for key in (
            "status", "progress", "input_count", "output_count",
            "filtered_count", "duplicate_count", "error_count", "failure_reason",
            "results_confirmed",
        )}
        result["task_id"] = task["id"]
        return result

    def list_results(
        self,
        task_id: str,
        *,
        page: int,
        page_size: int,
        status: str | None = None,
        split: str | None = None,
        keyword: str | None = None,
    ) -> dict[str, Any]:
        items = self.results[task_id]
        if status:
            items = [item for item in items if item["status"] == status]
        if split:
            items = [item for item in items if item["split"] == split]
        if keyword:
            items = [
                item
                for item in items
                if any(keyword in item[field] for field in ("instruction", "input", "output"))
            ]
        return {
            "items": deepcopy(items),
            "total": len(items),
            "page": page,
            "page_size": page_size,
        }

    def update_result(
        self, task_id: str, result_id: str, payload: dict[str, Any]
    ) -> dict[str, Any]:
        item = next((item for item in self.results[task_id] if item["id"] == result_id), None)
        if not item:
            raise NotFoundError("data process result not found")
        for field in ("instruction", "input", "output", "quality_score"):
            if field in payload:
                item[field] = deepcopy(payload[field])
        hard_valid = bool(item["instruction"].strip() and item["output"].strip())
        quality_valid = bool((item.get("quality_score") or {}).get("is_valid", hard_valid))
        changed = any(
            item[field] != item[f"original_{field}"]
            for field in ("instruction", "input", "output")
        )
        item["status"] = (
            "invalid"
            if not hard_valid or not quality_valid
            else "modified" if changed else "valid"
        )
        self.tasks[task_id]["error_count"] = sum(
            result["status"] == "invalid" for result in self.results[task_id]
        )
        return deepcopy(item)

    def get_result(self, task_id: str, result_id: str) -> dict[str, Any]:
        item = next((item for item in self.results[task_id] if item["id"] == result_id), None)
        if not item:
            raise NotFoundError("data process result not found")
        return deepcopy(item)

    def restore_result(self, task_id: str, result_id: str) -> dict[str, Any]:
        item = next((item for item in self.results[task_id] if item["id"] == result_id), None)
        if not item:
            raise NotFoundError("data process result not found")
        for field in ("instruction", "input", "output"):
            item[field] = item[f"original_{field}"]
        item["status"] = "valid"
        return deepcopy(item)

    def get_generation_model(self, model_id: str) -> dict[str, Any]:
        model = self.models.get(model_id)
        if not model:
            raise NotFoundError("generation model not found")
        return deepcopy(model)

    def replace_generated_result(
        self,
        task_id: str,
        result_id: str,
        replacement: dict[str, Any],
        *,
        expected_updated_at: str,
    ) -> dict[str, Any]:
        task = self.tasks[task_id]
        if task["status"] != "completed" or task.get("workflow_step") != "results":
            raise InvalidStateError("task is not editing generation results")
        if task.get("results_confirmed") or task.get("output_dataset_id"):
            raise InvalidStateError("confirmed or published results cannot be regenerated")
        item = next((item for item in self.results[task_id] if item["id"] == result_id), None)
        if not item:
            raise NotFoundError("data process result not found")
        if item["status"] != "invalid":
            raise InvalidStateError("only an invalid result can be regenerated")
        if expected_updated_at != item.get("updated_at"):
            raise InvalidStateError("data process result was modified by another request")
        for field in ("instruction", "input", "output"):
            item[field] = replacement[field]
            item[f"original_{field}"] = replacement[field]
        item.update(
            status=replacement["status"],
            error=replacement.get("error"),
            quality_score=deepcopy(replacement["quality_score"]),
            updated_at="2026-07-27T22:00:00Z",
        )
        task["error_count"] = sum(
            result["status"] == "invalid" for result in self.results[task_id]
        )
        return deepcopy(item)

    def confirm_results(self, task_id: str) -> dict[str, Any]:
        task = self.tasks[task_id]
        if task["status"] != "completed":
            raise InvalidStateError("only a completed task can confirm results")
        if task.get("workflow_step") != "results":
            raise InvalidStateError("workflow must be on results before confirmation")
        invalid_count = sum(
            1
            for item in self.results[task_id]
            if item["status"] == "invalid"
            or not item["instruction"].strip()
            or not item["output"].strip()
        )
        if invalid_count:
            raise InvalidStateError(f"task contains {invalid_count} invalid results")
        if not self.results[task_id]:
            raise InvalidStateError("task has no results to confirm")
        task["results_confirmed"] = True
        return self.get_task(task_id)

    def publish(self, task_id: str, payload: dict[str, Any]) -> dict[str, Any]:
        task = self.tasks[task_id]
        if task_id in self.regeneration_prepared:
            raise InvalidStateError("regeneration must start and complete before publishing")
        published = [
            dataset
            for dataset in self.datasets.values()
            if dataset.get("source_task_id") == task_id
            and dataset.get("deleted_at") is None
        ]
        if task.get("output_dataset_id"):
            train_dataset = self.datasets[task["output_dataset_id"]]
            return {
                "dataset": deepcopy(train_dataset),
                "datasets": deepcopy(published),
                "output_datasets": deepcopy(published),
                "created": False,
            }
        if task["status"] != "completed":
            raise InvalidStateError("only a completed task can be published")
        if not task.get("results_confirmed"):
            raise InvalidStateError("results must be confirmed before publishing")
        split_specs = (
            ("train", "训练集"),
            ("val", "验证集"),
            ("test", "测试集"),
        )
        for dataset_type, label in split_specs:
            dataset_id = self._id("dataset")
            self.datasets[dataset_id] = {
                "id": dataset_id,
                "name": f"{payload['dataset_name']}-{label}",
                "type": dataset_type,
                "count": len(self.results[task_id]),
                "source": "task",
                "task_id": task_id,
                "source_task_id": task_id,
                "deleted_at": None,
            }
        published = [
            dataset
            for dataset in self.datasets.values()
            if dataset.get("source_task_id") == task_id
            and dataset.get("deleted_at") is None
        ]
        train_dataset = next(dataset for dataset in published if dataset["type"] == "train")
        task["output_dataset_id"] = train_dataset["id"]
        return {
            "dataset": deepcopy(train_dataset),
            "datasets": deepcopy(published),
            "output_datasets": deepcopy(published),
            "created": True,
        }


def make_client(
    tmp_path: Path,
) -> tuple[TestClient, FakeDataProcessStore, LocalDataProcessStorage]:
    store = FakeDataProcessStore()
    storage = LocalDataProcessStorage(tmp_path / "data-process")
    app = FastAPI()
    app.include_router(router, prefix="/modelTF")
    app.dependency_overrides[get_data_process_store] = lambda: store
    app.dependency_overrides[get_data_process_storage] = lambda: storage
    return TestClient(app), store, storage


def _stored_files(storage: LocalDataProcessStorage) -> list[Path]:
    return [path for path in storage.root.rglob("*") if path.is_file() or path.is_symlink()]


def _minimal_pdf_pages(*page_texts: str) -> bytes:
    if not page_texts:
        raise ValueError("at least one PDF page is required")
    font_object_number = 3 + len(page_texts) * 2
    page_object_numbers = [3 + index * 2 for index in range(len(page_texts))]
    objects = [
        b"<< /Type /Catalog /Pages 2 0 R >>",
        (
            b"<< /Type /Pages /Kids ["
            + b" ".join(f"{number} 0 R".encode() for number in page_object_numbers)
            + b"] /Count "
            + str(len(page_texts)).encode()
            + b" >>"
        ),
    ]
    for index, text in enumerate(page_texts):
        content_object_number = page_object_numbers[index] + 1
        commands = [b"BT /F1 12 Tf 72 720 Td"]
        for line_index, line in enumerate(text.splitlines()):
            escaped = line.replace("\\", "\\\\").replace("(", "\\(").replace(")", "\\)")
            if line_index:
                commands.append(b"0 -16 Td")
            commands.append(f"({escaped}) Tj".encode("ascii"))
        commands.append(b"ET")
        stream = b" ".join(commands)
        objects.extend(
            [
                (
                    b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 612 792] "
                    b"/Resources << /Font << /F1 "
                    + str(font_object_number).encode()
                    + b" 0 R >> >> /Contents "
                    + str(content_object_number).encode()
                    + b" 0 R >>"
                ),
                b"<< /Length "
                + str(len(stream)).encode()
                + b" >>\nstream\n"
                + stream
                + b"\nendstream",
            ]
        )
    objects.append(b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>")
    result = bytearray(b"%PDF-1.4\n%\xe2\xe3\xcf\xd3\n")
    offsets = [0]
    for object_number, value in enumerate(objects, start=1):
        offsets.append(len(result))
        result.extend(f"{object_number} 0 obj\n".encode())
        result.extend(value)
        result.extend(b"\nendobj\n")
    xref_offset = len(result)
    result.extend(f"xref\n0 {len(objects) + 1}\n".encode())
    result.extend(b"0000000000 65535 f \n")
    for offset in offsets[1:]:
        result.extend(f"{offset:010d} 00000 n \n".encode())
    result.extend(
        (
            f"trailer\n<< /Size {len(objects) + 1} /Root 1 0 R >>\n"
            f"startxref\n{xref_offset}\n%%EOF\n"
        ).encode()
    )
    return bytes(result)


def _minimal_pdf(text: str = "Hello PDF") -> bytes:
    return _minimal_pdf_pages(text)


def test_outdated_data_process_schema_returns_actionable_503() -> None:
    with pytest.raises(HTTPException) as captured, data_process_endpoint.api_errors():
        raise psycopg.errors.UndefinedColumn("missing runtime column")

    assert captured.value.status_code == 503
    assert "schema is missing or out of date" in captured.value.detail["message"]


def test_data_process_full_contract_without_database(tmp_path: Path) -> None:
    client, store, _ = make_client(tmp_path)
    created = client.post(
        "/modelTF/data-process",
        json={
            "name": "客服问答处理",
            "process_type": "structured",
            "config": {"dataset_split": {"train": 80, "validation": 10, "test": 10}},
        },
    )
    assert created.status_code == 200
    assert created.json()["data"]["results_confirmed"] is False
    task_id = created.json()["data"]["id"]

    source_content = (
        '{"question":"如何修改地址？",'
        '"answer":"订单发货前可在订单详情申请修改收货地址。"}\n'
        '{"question":"如何申请退款？",'
        '"answer":"请在订单详情提交退款申请并等待审核处理。"}\n'
    )
    uploaded = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={"files": ("customer.jsonl", source_content.encode(), "application/jsonl")},
    )
    assert uploaded.status_code == 200
    source = uploaded.json()["data"]["files"][0]
    assert len(source["checksum_sha256"]) == 64
    assert source["version_no"] == 1

    window = client.get(
        f"/modelTF/data-process/{task_id}/source-files/{source['id']}/content",
        params={"offset": 0, "limit": 20},
    )
    assert window.status_code == 200
    assert window.json()["data"]["has_more"] is True
    line_window = client.get(
        f"/modelTF/data-process/{task_id}/source-files/{source['id']}/content",
        params={"start_line": 2, "line_count": 1},
    )
    assert line_window.json()["data"]["start_line"] == 2
    assert line_window.json()["data"]["end_line"] == 2
    assert line_window.json()["data"]["total_lines"] == 2

    preview = client.post(
        f"/modelTF/data-process/{task_id}/preview/build",
        json={"source_file_ids": [source["id"]]},
    )
    assert preview.status_code == 200
    assert preview.json()["data"]["total"] == 2
    listed_preview = client.get(f"/modelTF/data-process/{task_id}/preview")
    assert listed_preview.json()["data"]["total"] == 2
    preview_item = listed_preview.json()["data"]["items"][0]
    updated_preview = client.put(
        f"/modelTF/data-process/{task_id}/preview/{preview_item['id']}",
        json={
            "edited_content": preview_item["edited_content"],
            "expected_updated_at": "2026-07-23T00:00:00Z",
        },
    )
    assert "quality_score" in updated_preview.json()["data"]

    generated = client.post(f"/modelTF/data-process/{task_id}/generate")
    assert generated.status_code == 200
    assert generated.json()["data"]["results_confirmed"] is False
    progress = client.get(f"/modelTF/data-process/{task_id}/progress")
    assert progress.json()["data"]["status"] == "completed"
    assert progress.json()["data"]["results_confirmed"] is False
    result_page = client.get(f"/modelTF/data-process/{task_id}/results").json()["data"]
    assert result_page["total"] == 2
    keyword_page = client.get(
        f"/modelTF/data-process/{task_id}/results", params={"keyword": "地址"}
    ).json()["data"]
    assert keyword_page["total"] == 1

    result = result_page["items"][0]
    edited = client.put(
        f"/modelTF/data-process/{task_id}/results/{result['id']}",
        json={
            "output": "人工修改后的完整答案。",
            "expected_updated_at": "2026-07-23T00:00:00Z",
        },
    )
    assert edited.json()["data"]["status"] == "modified"
    assert "quality_score" in edited.json()["data"]
    invalid_edit = client.put(
        f"/modelTF/data-process/{task_id}/results/{result['id']}",
        json={"output": ""},
    )
    assert invalid_edit.json()["data"]["status"] == "invalid"
    assert store.tasks[task_id]["error_count"] == 1
    restored = client.post(
        f"/modelTF/data-process/{task_id}/results/{result['id']}/restore"
    )
    assert restored.json()["data"]["output"] == result["original_output"]
    assert restored.json()["data"]["status"] == "valid"
    assert store.tasks[task_id]["error_count"] == 0

    workflow = client.put(
        f"/modelTF/data-process/{task_id}/workflow-step",
        json={"workflow_step": "results"},
    )
    assert workflow.status_code == 200
    confirmed = client.post(
        f"/modelTF/data-process/{task_id}/confirm-results"
    )
    assert confirmed.status_code == 200
    assert confirmed.json()["data"]["results_confirmed"] is True

    publish_payload = {"dataset_name": "客服问答清洗集"}
    first_publish = client.post(
        f"/modelTF/data-process/{task_id}/publish", json=publish_payload
    )
    second_publish = client.post(
        f"/modelTF/data-process/{task_id}/publish", json=publish_payload
    )
    assert first_publish.json()["data"]["created"] is True
    assert second_publish.json()["data"]["created"] is False
    assert (
        first_publish.json()["data"]["dataset"]["id"]
        == second_publish.json()["data"]["dataset"]["id"]
    )


def test_task_list_exposes_document_and_generation_counts(
    tmp_path: Path,
) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "列表契约", "process_type": "unstructured", "config": {}},
    ).json()["data"]["id"]
    store.sources[task_id] = [{"id": "source-1"}, {"id": "source-2"}]
    store.tasks[task_id].update(
        {
            "status": "pending",
            "output_count": 17,
            "output_dataset_id": None,
        }
    )

    response = client.get("/modelTF/data-process")

    assert response.status_code == 200
    item = response.json()["data"]["items"][0]
    assert item["status"] == "pending"
    assert item["source_file_count"] == 2
    assert item["output_count"] == 17
    assert item["output_dataset_id"] is None


def test_workflow_step_update_is_independent_and_validated(tmp_path: Path) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "步骤持久化", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]
    store.previews[task_id] = [{"id": "preview-1"}]
    store.results[task_id] = [{"id": "result-1"}]
    store.tasks[task_id].update(status="running", generation_run_id="run-1")

    updated = client.put(
        f"/modelTF/data-process/{task_id}/workflow-step",
        json={"workflow_step": "generate"},
    )

    assert updated.status_code == 200
    assert updated.json()["data"]["workflow_step"] == "generate"
    assert store.previews[task_id] == [{"id": "preview-1"}]
    assert store.results[task_id] == [{"id": "result-1"}]
    assert store.tasks[task_id]["status"] == "running"
    assert client.put(
        f"/modelTF/data-process/{task_id}/workflow-step",
        json={"workflow_step": "unknown"},
    ).status_code == 422


def test_background_preview_persists_progress_and_items(tmp_path: Path) -> None:
    client, _, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "后台切分", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]
    source = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={
            "files": (
                "one.jsonl",
                b'{"question":"What is one?","answer":"One."}\n',
                "application/jsonl",
            )
        },
    ).json()["data"]["files"][0]

    started = client.post(
        f"/modelTF/data-process/{task_id}/preview/start",
        json={"source_file_ids": [source["id"]]},
    )

    assert started.status_code == 202
    assert started.json()["data"]["preview_status"] == "queued"
    assert started.json()["data"]["preview_run_id"]
    progress = client.get(
        f"/modelTF/data-process/{task_id}/preview/progress"
    ).json()["data"]
    assert progress == {
        "task_id": task_id,
        "workflow_step": "preview",
        "preview_status": "completed",
        "preview_progress": 100.0,
        "preview_run_id": None,
        "preview_failure_reason": None,
        "preview_total_files": 1,
        "preview_completed_files": 1,
    }
    assert client.get(
        f"/modelTF/data-process/{task_id}/preview"
    ).json()["data"]["total"] == 1


def test_stale_preview_worker_cannot_replace_new_run(tmp_path: Path) -> None:
    _, store, storage = make_client(tmp_path)
    task = store.create_task(
        {"name": "切分代次", "process_type": "structured", "config": {}}
    )
    task_id = str(task["id"])
    store.sources[task_id] = [{"id": "source-1"}]
    first, source_ids = store.start_preview(task_id, source_file_ids=["source-1"])
    first_run_id = str(first["preview_run_id"])
    store.tasks[task_id].update(preview_status="cancelled", preview_run_id=None)
    second, _ = store.start_preview(task_id, source_file_ids=["source-1"])

    data_process_endpoint._run_preview(
        store,
        storage,
        task_id,
        first_run_id,
        source_ids,
    )

    assert store.tasks[task_id]["preview_status"] == "queued"
    assert store.tasks[task_id]["preview_run_id"] == second["preview_run_id"]
    assert store.previews[task_id] == []


def test_delete_invalidates_active_generation_and_preview(tmp_path: Path) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "删除运行任务", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]
    store.tasks[task_id].update(
        status="running",
        generation_run_id="generation-1",
        preview_status="running",
        preview_run_id="preview-1",
    )

    deleted = client.delete(f"/modelTF/data-process/{task_id}")

    assert deleted.status_code == 200
    assert store.generation_is_running(task_id, "generation-1") is False
    assert store.preview_is_running(task_id, "preview-1") is False


def test_task_detail_uses_returned_source_files_as_document_count(tmp_path: Path) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "详情文档数", "process_type": "unstructured", "config": {}},
    ).json()["data"]["id"]
    store.tasks[task_id]["source_file_count"] = 99
    store.sources[task_id] = [
        {"id": "source-1", "name": "一.pdf", "content": "正文一"},
        {"id": "source-2", "name": "二.pdf", "content": "正文二"},
    ]

    response = client.get(f"/modelTF/data-process/{task_id}")

    assert response.status_code == 200
    detail = response.json()["data"]
    assert len(detail["source_files"]) == 2
    assert detail["source_file_count"] == 2


def test_generation_start_response_clears_previous_output_count(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "新一轮生成", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]
    store.tasks[task_id]["output_count"] = 28
    store.previews[task_id] = [
        {
            "id": "preview-1",
            "source_file_id": None,
            "original_content": '{"question":"新问题","answer":"新答案"}',
            "edited_content": '{"question":"新问题","answer":"新答案"}',
            "status": "original",
        }
    ]
    monkeypatch.setattr(data_process_endpoint, "_run_generation", lambda *args: None)

    response = client.post(f"/modelTF/data-process/{task_id}/generate")

    assert response.status_code == 200
    progress = response.json()["data"]
    assert progress["status"] == "running"
    assert progress["output_count"] == 0
    assert store.tasks[task_id]["output_count"] == 0


def test_results_must_be_generated_and_valid_before_confirmation(
    tmp_path: Path,
) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "结果确认门禁", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]

    pending = client.post(f"/modelTF/data-process/{task_id}/confirm-results")
    assert pending.status_code == 409

    store.tasks[task_id].update(status="completed", progress=100, workflow_step="generate")
    store.results[task_id] = [
        {
            "id": "result_valid",
            "status": "valid",
            "instruction": "问题",
            "input": "",
            "output": "答案",
        }
    ]
    wrong_step = client.post(f"/modelTF/data-process/{task_id}/confirm-results")
    assert wrong_step.status_code == 409

    store.tasks[task_id]["workflow_step"] = "results"
    store.results[task_id] = [
        {
            "id": "result_invalid",
            "status": "invalid",
            "instruction": "问题",
            "input": "",
            "output": "",
        }
    ]
    invalid = client.post(f"/modelTF/data-process/{task_id}/confirm-results")
    assert invalid.status_code == 409

    store.results[task_id][0].update(status="valid", output="答案")
    confirmed = client.post(f"/modelTF/data-process/{task_id}/confirm-results")
    assert confirmed.status_code == 200
    assert confirmed.json()["data"]["results_confirmed"] is True
    assert client.post(
        f"/modelTF/data-process/{task_id}/confirm-results"
    ).status_code == 200


def test_invalid_result_can_be_regenerated_in_place(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={
            "name": "单条重生成",
            "process_type": "structured",
            "config": {
                "generation_model_id": "model-1",
                "output_type": "standard",
                "min_output_length": 5,
                "generation_retries": 5,
                "request_timeout_seconds": 120,
            },
        },
    ).json()["data"]["id"]
    store.tasks[task_id].update(
        status="completed",
        progress=100,
        workflow_step="results",
        results_confirmed=False,
        error_count=1,
    )
    store.models["model-1"] = {
        "id": "model-1",
        "name": "测试模型",
        "online_model_name": "test-model",
        "api_url": "https://model.example/v1",
        "api_key": "secret",
    }
    store.previews[task_id] = [{
        "id": "preview-1",
        "status": "original",
        "original_content": "申请编号字段用于标识报销申请。",
        "edited_content": "申请编号字段用于标识报销申请。",
    }]
    store.results[task_id] = [{
        "id": "result-1",
        "preview_item_id": "preview-1",
        "instruction": "模型生成失败，请人工补充",
        "input": "申请编号字段用于标识报销申请。",
        "output": "",
        "original_instruction": "模型生成失败，请人工补充",
        "original_input": "申请编号字段用于标识报销申请。",
        "original_output": "",
        "status": "invalid",
        "error": "model response is not valid JSON",
        "split": "train",
        "quality_score": {},
        "updated_at": "2026-07-27T21:00:00Z",
    }]

    captured: dict[str, Any] = {}

    def fake_generate(preview_items: Any, **kwargs: Any) -> list[dict[str, Any]]:
        captured["preview_items"] = list(preview_items)
        captured["qa_pairs_per_item"] = kwargs["qa_pairs_per_item"]
        captured["config"] = kwargs["config"]
        return [{
            "id": "temporary-result",
            "preview_item_id": "preview-1",
            "instruction": "申请编号字段有什么作用？",
            "input": "申请编号字段用于标识报销申请。",
            "output": "申请编号字段用于唯一标识一笔报销申请。",
            "original_instruction": "申请编号字段有什么作用？",
            "original_input": "申请编号字段用于标识报销申请。",
            "original_output": "申请编号字段用于唯一标识一笔报销申请。",
            "status": "valid",
            "error": None,
            "split": "train",
        }]

    monkeypatch.setattr(data_process_endpoint, "generate_model_records", fake_generate)
    response = client.post(
        f"/modelTF/data-process/{task_id}/results/result-1/regenerate",
        json={"expected_updated_at": "2026-07-27T21:00:00Z"},
    )

    assert response.status_code == 200
    regenerated = response.json()["data"]
    assert regenerated["id"] == "result-1"
    assert regenerated["status"] == "valid"
    assert regenerated["instruction"] == regenerated["original_instruction"]
    assert regenerated["output"] == regenerated["original_output"]
    assert store.tasks[task_id]["error_count"] == 0
    assert captured["qa_pairs_per_item"] == 1
    assert [item["id"] for item in captured["preview_items"]] == ["preview-1"]
    assert captured["config"]["generation_retries"] == 0
    assert captured["config"]["request_timeout_seconds"] == 60


def test_failed_result_regeneration_keeps_the_original_error(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={
            "name": "失败结果保留",
            "process_type": "structured",
            "config": {
                "generation_model_id": "model-1",
                "output_type": "standard",
            },
        },
    ).json()["data"]["id"]
    store.tasks[task_id].update(
        status="completed",
        progress=100,
        workflow_step="results",
        results_confirmed=False,
        error_count=1,
    )
    store.models["model-1"] = {
        "id": "model-1",
        "online_model_name": "test-model",
        "api_url": "https://model.example/v1",
        "api_key": "secret",
    }
    store.previews[task_id] = [{
        "id": "preview-1",
        "status": "original",
        "original_content": "原始内容",
        "edited_content": "原始内容",
    }]
    original_result = {
        "id": "result-1",
        "preview_item_id": "preview-1",
        "instruction": "模型生成失败，请人工补充",
        "input": "原始内容",
        "output": "",
        "status": "invalid",
        "error": "first failure",
        "updated_at": "2026-07-27T21:30:00Z",
    }
    store.results[task_id] = [original_result.copy()]

    monkeypatch.setattr(
        data_process_endpoint,
        "generate_model_records",
        lambda *args, **kwargs: [{
            "status": "invalid",
            "error": "second failure",
        }],
    )
    response = client.post(
        f"/modelTF/data-process/{task_id}/results/result-1/regenerate",
        json={"expected_updated_at": "2026-07-27T21:30:00Z"},
    )

    assert response.status_code == 409
    assert store.results[task_id] == [original_result]
    assert store.tasks[task_id]["error_count"] == 1


def test_failed_results_can_be_regenerated_in_parallel_with_partial_success(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={
            "name": "批量重生成",
            "process_type": "structured",
            "config": {
                "generation_model_id": "model-1",
                "output_type": "standard",
                "min_output_length": 5,
            },
        },
    ).json()["data"]["id"]
    store.tasks[task_id].update(
        status="completed",
        progress=100,
        workflow_step="results",
        results_confirmed=False,
        error_count=2,
    )
    store.models["model-1"] = {
        "id": "model-1",
        "online_model_name": "test-model",
        "api_url": "https://model.example/v1",
        "api_key": "secret",
    }
    store.previews[task_id] = [
        {
            "id": "preview-1",
            "status": "original",
            "original_content": "申请编号用于唯一标识一笔报销申请。",
            "edited_content": "申请编号用于唯一标识一笔报销申请。",
        },
        {
            "id": "preview-2",
            "status": "original",
            "original_content": "联系电话用于联系申请人。",
            "edited_content": "联系电话用于联系申请人。",
        },
    ]
    original_results = [
        {
            "id": "result-1",
            "preview_item_id": "preview-1",
            "instruction": "模型生成失败，请人工补充",
            "input": "申请编号用于唯一标识一笔报销申请。",
            "output": "",
            "original_instruction": "模型生成失败，请人工补充",
            "original_input": "申请编号用于唯一标识一笔报销申请。",
            "original_output": "",
            "status": "invalid",
            "error": "first failure",
            "split": "train",
            "quality_score": {},
            "updated_at": "2026-07-28T09:00:00Z",
        },
        {
            "id": "result-2",
            "preview_item_id": "preview-2",
            "instruction": "模型生成失败，请人工补充",
            "input": "联系电话用于联系申请人。",
            "output": "",
            "original_instruction": "模型生成失败，请人工补充",
            "original_input": "联系电话用于联系申请人。",
            "original_output": "",
            "status": "invalid",
            "error": "first failure",
            "split": "train",
            "quality_score": {},
            "updated_at": "2026-07-28T09:00:01Z",
        },
    ]
    store.results[task_id] = deepcopy(original_results)
    barrier = Barrier(2, timeout=2)
    activity_lock = Lock()
    active_calls = 0
    max_active_calls = 0
    model_clients: list[Any] = []

    def fake_generate(preview_items: Any, **kwargs: Any) -> list[dict[str, Any]]:
        nonlocal active_calls, max_active_calls
        preview = next(iter(preview_items))
        with activity_lock:
            active_calls += 1
            max_active_calls = max(max_active_calls, active_calls)
            model_clients.append(kwargs.get("client"))
        try:
            barrier.wait()
            if preview["id"] == "preview-2":
                return [{"status": "invalid", "error": "second failure"}]
            return [{
                "id": "temporary-result",
                "preview_item_id": preview["id"],
                "instruction": "申请编号有什么作用？",
                "input": preview["edited_content"],
                "output": "申请编号用于唯一标识一笔报销申请。",
                "status": "valid",
                "error": None,
                "split": "train",
            }]
        finally:
            with activity_lock:
                active_calls -= 1

    monkeypatch.setattr(data_process_endpoint, "generate_model_records", fake_generate)
    response = client.post(
        f"/modelTF/data-process/{task_id}/results/regenerate-batch",
        json={
            "items": [
                {
                    "result_id": "result-1",
                    "expected_updated_at": "2026-07-28T09:00:00Z",
                },
                {
                    "result_id": "result-2",
                    "expected_updated_at": "2026-07-28T09:00:01Z",
                },
            ],
        },
    )

    assert response.status_code == 200
    data = response.json()["data"]
    assert data["total"] == 2
    assert data["succeeded"] == 1
    assert data["failed"] == 1
    assert data["remaining_invalid_count"] == 1
    assert [item["id"] for item in data["items"]] == ["result-1"]
    assert data["failures"][0]["result_id"] == "result-2"
    assert store.results[task_id][0]["status"] == "valid"
    assert store.results[task_id][0]["id"] == "result-1"
    assert store.results[task_id][1] == original_results[1]
    assert store.tasks[task_id]["error_count"] == 1
    assert max_active_calls == 2
    assert all(client is not None for client in model_clients)
    assert len({id(client) for client in model_clients}) == 1


@pytest.mark.parametrize(
    "locked_state",
    [
        {"results_confirmed": True},
        {"output_dataset_id": "dataset-published"},
    ],
    ids=["confirmed", "published"],
)
def test_batch_result_regeneration_rejects_locked_tasks_before_model_call(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
    locked_state: dict[str, Any],
) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={
            "name": "批量重生成门禁",
            "process_type": "structured",
            "config": {"generation_model_id": "model-1"},
        },
    ).json()["data"]["id"]
    store.tasks[task_id].update(
        status="completed",
        workflow_step="results",
        results_confirmed=False,
    )
    store.tasks[task_id].update(locked_state)
    model_calls = 0

    def fake_generate(*args: Any, **kwargs: Any) -> list[dict[str, Any]]:
        nonlocal model_calls
        model_calls += 1
        return []

    monkeypatch.setattr(data_process_endpoint, "generate_model_records", fake_generate)
    response = client.post(
        f"/modelTF/data-process/{task_id}/results/regenerate-batch",
        json={
            "items": [{
                "result_id": "result-1",
                "expected_updated_at": "2026-07-28T09:00:00Z",
            }],
        },
    )

    assert response.status_code == 409
    assert model_calls == 0


def test_preview_build_replaces_only_selected_files_and_reports_file_counts(
    tmp_path: Path,
) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "逐文件预览", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]
    uploaded = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files=[
            ("files", ("first.jsonl", b'{"id":1}\n', "application/jsonl")),
            (
                "files",
                ("second.jsonl", b'{"id":2}\n{"id":3}\n', "application/jsonl"),
            ),
        ],
    )
    assert uploaded.status_code == 200
    first_source, second_source = uploaded.json()["data"]["files"]

    first_build = client.post(
        f"/modelTF/data-process/{task_id}/preview/build",
        json={"source_file_ids": [first_source["id"]]},
    )
    assert first_build.status_code == 200
    first_data = first_build.json()["data"]
    assert first_data["file_counts"] == {first_source["id"]: 1}
    assert first_data["files"] == [
        {
            "source_file_id": first_source["id"],
            "preview_count": 1,
            "status": "completed",
        }
    ]
    first_item = first_data["items"][0]
    edited = client.put(
        f"/modelTF/data-process/{task_id}/preview/{first_item['id']}",
        json={"edited_content": "人工确认后的第一文件预览"},
    )
    assert edited.status_code == 200

    second_build = client.post(
        f"/modelTF/data-process/{task_id}/preview/build",
        json={"source_file_id": second_source["id"]},
    )
    assert second_build.status_code == 200
    second_data = second_build.json()["data"]
    assert second_data["file_counts"] == {second_source["id"]: 2}
    assert second_data["files"] == [
        {
            "source_file_id": second_source["id"],
            "preview_count": 2,
            "status": "completed",
        }
    ]
    assert {item["source_file_id"] for item in store.previews[task_id]} == {
        first_source["id"],
        second_source["id"],
    }
    preserved_first = next(
        item
        for item in store.previews[task_id]
        if item["source_file_id"] == first_source["id"]
    )
    assert preserved_first["id"] == first_item["id"]
    assert preserved_first["edited_content"] == "人工确认后的第一文件预览"

    previous_second_ids = {
        item["id"]
        for item in store.previews[task_id]
        if item["source_file_id"] == second_source["id"]
    }
    next(
        source
        for source in store.sources[task_id]
        if source["id"] == second_source["id"]
    )["content"] = '{"id":4}\n'
    rebuilt = client.post(
        f"/modelTF/data-process/{task_id}/preview/build",
        json={"source_file_ids": [second_source["id"]]},
    )
    assert rebuilt.status_code == 200
    assert rebuilt.json()["data"]["file_counts"] == {second_source["id"]: 1}
    current_second_ids = {
        item["id"]
        for item in store.previews[task_id]
        if item["source_file_id"] == second_source["id"]
    }
    assert current_second_ids.isdisjoint(previous_second_ids)
    assert len(current_second_ids) == 1
    assert next(
        item
        for item in store.previews[task_id]
        if item["source_file_id"] == first_source["id"]
    )["id"] == first_item["id"]


def test_preview_build_rejects_unknown_and_cross_task_source_file_ids(
    tmp_path: Path,
) -> None:
    client, _, _ = make_client(tmp_path)
    first_task_id = client.post(
        "/modelTF/data-process",
        json={"name": "归属任务一", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]
    second_task_id = client.post(
        "/modelTF/data-process",
        json={"name": "归属任务二", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]
    foreign_source = client.post(
        f"/modelTF/data-process/{second_task_id}/source-files",
        files={"files": ("foreign.jsonl", b'{"id":2}\n', "application/jsonl")},
    ).json()["data"]["files"][0]

    unknown = client.post(
        f"/modelTF/data-process/{first_task_id}/preview/build",
        json={"source_file_ids": ["dpsf_not_found"]},
    )
    assert unknown.status_code == 404
    foreign = client.post(
        f"/modelTF/data-process/{first_task_id}/preview/build",
        json={"source_file_id": foreign_source["id"]},
    )
    assert foreign.status_code == 404
    ambiguous = client.post(
        f"/modelTF/data-process/{first_task_id}/preview/build",
        json={
            "source_file_id": foreign_source["id"],
            "source_file_ids": [foreign_source["id"]],
        },
    )
    assert ambiguous.status_code == 422


def test_external_source_never_returns_fake_success(tmp_path: Path) -> None:
    client, _, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "外部数据", "process_type": "external", "config": {}},
    ).json()["data"]["id"]
    response = client.post(
        f"/modelTF/data-process/{task_id}/external/test",
        json={"type": "mysql", "url": "mysql://db.example/test"},
    )
    assert response.status_code == 501
    assert response.json()["detail"]["code"] == 501


def test_regenerate_endpoint_prepares_an_existing_published_task(tmp_path: Path) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "待重新生成", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]
    store.tasks[task_id].update(
        {
            "status": "completed",
            "updated_at": "2026-07-25T19:00:00Z",
            "output_dataset_id": "dataset_train",
            "output_count": 2,
        }
    )
    store.datasets["dataset_train"] = {
        "id": "dataset_train",
        "name": "原训练集",
        "type": "train",
        "source_task_id": task_id,
        "deleted_at": None,
    }
    store.previews[task_id] = [{"id": "preview_1", "edited_content": "原切片"}]
    store.results[task_id] = [{"id": "result_1"}]

    response = client.post(
        f"/modelTF/data-process/{task_id}/regenerate",
        json={
            "name": "重新生成后名称",
            "description": "更换生成模型",
            "process_type": "structured",
            "config": {"generation_model_id": "model_2"},
            "expected_updated_at": "2026-07-25T19:00:00Z",
        },
    )

    assert response.status_code == 200
    data = response.json()["data"]
    assert data["task"]["status"] == "completed"
    assert data["task"]["output_dataset_id"] == "dataset_train"
    assert data["task"]["output_count"] == 2
    assert data["preview_invalidated"] is False
    assert data["published_outputs_preserved"] is True
    assert store.results[task_id] == [{"id": "result_1"}]
    assert store.previews[task_id][0]["id"] == "preview_1"

    detail = client.get(f"/modelTF/data-process/{task_id}").json()["data"]
    assert detail["status"] == "completed"
    assert detail["output_dataset_id"] == "dataset_train"
    assert detail["output_count"] == 2
    assert [item["id"] for item in detail["output_datasets"]] == ["dataset_train"]


def test_published_split_datasets_remain_in_detail_after_regeneration(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "保留旧发布数据", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]
    store.tasks[task_id].update(
        {
            "status": "completed",
            "updated_at": "2026-07-27T09:00:00Z",
            "output_count": 1,
            "results_confirmed": True,
        }
    )
    store.results[task_id] = [
        {
            "id": "result_1",
            "status": "valid",
            "instruction": "问题",
            "input": "",
            "output": "答案",
        }
    ]
    store.previews[task_id] = [{"id": "preview_1", "edited_content": "原切片"}]

    published = client.post(
        f"/modelTF/data-process/{task_id}/publish",
        json={"dataset_name": "保留旧发布数据集"},
    )
    assert published.status_code == 200
    published_datasets = published.json()["data"]["datasets"]
    assert len(published_datasets) == 3
    published_ids = {item["id"] for item in published_datasets}
    assert store.tasks[task_id]["output_dataset_id"] in published_ids

    regenerated = client.post(
        f"/modelTF/data-process/{task_id}/regenerate",
        json={
            "name": "保留旧发布数据",
            "description": "更换生成配置后退出",
            "process_type": "structured",
            "config": {"generation_model_id": "model_2"},
            "expected_updated_at": "2026-07-27T09:00:00Z",
        },
    )
    assert regenerated.status_code == 200
    original_output_dataset_id = store.tasks[task_id]["output_dataset_id"]
    prepared_task = regenerated.json()["data"]["task"]
    assert prepared_task["status"] == "completed"
    assert prepared_task["output_dataset_id"] == original_output_dataset_id
    assert prepared_task["output_count"] == 1
    assert store.results[task_id][0]["id"] == "result_1"

    detail = client.get(f"/modelTF/data-process/{task_id}")
    assert detail.status_code == 200
    detail_data = detail.json()["data"]
    assert detail_data["status"] == "completed"
    assert detail_data["output_dataset_id"] == original_output_dataset_id
    assert detail_data["output_count"] == 1
    assert len(detail_data["output_datasets"]) == 3
    assert {item["id"] for item in detail_data["output_datasets"]} == published_ids
    assert set(store.datasets) == published_ids
    retained_results = client.get(f"/modelTF/data-process/{task_id}/results").json()["data"]
    assert retained_results["total"] == 1
    assert retained_results["items"][0]["id"] == "result_1"

    monkeypatch.setattr(data_process_endpoint, "_run_generation", lambda *args: None)
    started = client.post(f"/modelTF/data-process/{task_id}/generate")
    assert started.status_code == 200
    running = started.json()["data"]
    assert running["status"] == "running"
    assert running["output_count"] == 0
    assert store.results[task_id] == []
    assert set(store.datasets) == published_ids

    running_detail = client.get(f"/modelTF/data-process/{task_id}").json()["data"]
    assert running_detail["status"] == "running"
    assert running_detail["output_dataset_id"] is None
    assert running_detail["output_count"] == 0
    assert {item["id"] for item in running_detail["output_datasets"]} == published_ids


def test_regenerate_endpoint_validates_snapshot_and_locked_process_type(
    tmp_path: Path,
) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "并发校验", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]
    store.tasks[task_id]["updated_at"] = "2026-07-25T19:00:00Z"
    payload = {
        "name": "并发校验",
        "description": "",
        "process_type": "structured",
        "config": {},
        "expected_updated_at": "stale",
    }

    stale = client.post(f"/modelTF/data-process/{task_id}/regenerate", json=payload)
    assert stale.status_code == 409

    payload.update(
        {
            "process_type": "unstructured",
            "expected_updated_at": "2026-07-25T19:00:00Z",
        }
    )
    locked_type = client.post(f"/modelTF/data-process/{task_id}/regenerate", json=payload)
    assert locked_type.status_code == 409

    for missing_field in ("expected_updated_at", "description", "config"):
        missing_required_field = client.post(
            f"/modelTF/data-process/{task_id}/regenerate",
            json={key: value for key, value in payload.items() if key != missing_field},
        )
        assert missing_required_field.status_code == 422


def test_config_validation_and_stop_state(tmp_path: Path) -> None:
    client, store, _ = make_client(tmp_path)
    invalid = client.post(
        "/modelTF/data-process",
        json={
            "name": "错误切片配置",
            "process_type": "unstructured",
            "config": {
                "dataset_split": {"train": 80, "validation": 30, "test": 0},
                "chunk_size": 100,
                "chunk_overlap": 90,
                "min_chunk_size": 20,
            },
        },
    )
    assert invalid.status_code == 422

    semantic = client.post(
        "/modelTF/data-process",
        json={
            "name": "语义切分策略",
            "process_type": "unstructured",
            "config": {"chunk_method": "semantic"},
        },
    )
    assert semantic.status_code == 200

    removed_custom_method = client.post(
        "/modelTF/data-process",
        json={
            "name": "已移除的自定义分隔符",
            "process_type": "unstructured",
            "config": {"chunk_method": "custom"},
        },
    )
    assert removed_custom_method.status_code == 422
    assert "chunk_method" in removed_custom_method.text

    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "可停止任务", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]
    store.tasks[task_id]["status"] = "running"
    stopped = client.post(f"/modelTF/data-process/{task_id}/stop")
    assert stopped.status_code == 200
    assert stopped.json()["data"]["status"] == "stopped"


@pytest.mark.parametrize("config_key", ["qa_pairs_per_row", "qa_pairs_per_chunk"])
@pytest.mark.parametrize("count", [1, 50])
def test_qa_pair_config_accepts_supported_boundaries(
    tmp_path: Path,
    config_key: str,
    count: int,
) -> None:
    client, _, _ = make_client(tmp_path)
    response = client.post(
        "/modelTF/data-process",
        json={
            "name": "问答数量边界",
            "process_type": "unstructured",
            "config": {config_key: count},
        },
    )

    assert response.status_code == 200


@pytest.mark.parametrize("config_key", ["qa_pairs_per_row", "qa_pairs_per_chunk"])
@pytest.mark.parametrize("count", [0, 51])
def test_qa_pair_config_rejects_out_of_range_boundaries(
    tmp_path: Path,
    config_key: str,
    count: int,
) -> None:
    client, _, _ = make_client(tmp_path)
    response = client.post(
        "/modelTF/data-process",
        json={
            "name": "问答数量越界",
            "process_type": "unstructured",
            "config": {config_key: count},
        },
    )

    assert response.status_code == 422
    assert "[1, 50]" in response.text


def test_upload_batch_is_atomic_and_empty_files_are_rejected(tmp_path: Path) -> None:
    client, store, storage = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "批量上传", "process_type": "unstructured", "config": {}},
    ).json()["data"]["id"]

    duplicate_batch = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files=[
            ("files", ("first.txt", b"same content", "text/plain")),
            ("files", ("second.txt", b"same content", "text/plain")),
        ],
    )
    assert duplicate_batch.status_code == 400
    assert store.sources[task_id] == []
    assert _stored_files(storage) == []

    parse_failure = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files=[
            ("files", ("valid.txt", "先暂存的内容".encode(), "text/plain")),
            ("files", ("broken.txt", b"\xff", "text/plain")),
        ],
    )
    assert parse_failure.status_code == 400
    assert store.sources[task_id] == []
    assert _stored_files(storage) == []

    empty = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={"files": ("empty.txt", b"", "text/plain")},
    )
    assert empty.status_code == 400
    assert store.sources[task_id] == []
    assert _stored_files(storage) == []


def test_incremental_upload_keeps_existing_file_previews(tmp_path: Path) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "增量上传预览", "process_type": "unstructured", "config": {}},
    ).json()["data"]["id"]
    first_batch = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files=[
            ("files", ("first.txt", "第一个文件内容".encode(), "text/plain")),
            ("files", ("second.txt", "第二个文件内容".encode(), "text/plain")),
        ],
    )
    assert first_batch.status_code == 200
    first_source_ids = {item["id"] for item in first_batch.json()["data"]["files"]}
    built = client.post(f"/modelTF/data-process/{task_id}/preview/build", json={})
    assert built.status_code == 200
    assert {item["source_file_id"] for item in built.json()["data"]["items"]} == first_source_ids
    store.results[task_id] = [{"id": "old_result"}]

    third = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={"files": ("third.txt", "第三个文件内容".encode(), "text/plain")},
    )

    assert third.status_code == 200
    previews = client.get(f"/modelTF/data-process/{task_id}/preview").json()["data"]
    assert previews["total"] == 2
    assert {item["source_file_id"] for item in previews["items"]} == first_source_ids
    assert store.results[task_id] == []
    assert store.tasks[task_id]["progress"] == 20


def test_upload_preserves_store_error_when_storage_rollback_fails(
    tmp_path: Path,
    monkeypatch: Any,
) -> None:
    client, store, storage = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "回滚异常", "process_type": "unstructured", "config": {}},
    ).json()["data"]["id"]
    cleanup_attempts: list[str] = []

    def fail_store(*_: Any, **__: Any) -> list[dict[str, Any]]:
        raise ValueError("simulated database transaction failure")

    def fail_cleanup(reference: str, **_: Any) -> bool:
        cleanup_attempts.append(reference)
        raise OSError("simulated storage cleanup failure")

    monkeypatch.setattr(store, "add_source_files", fail_store)
    monkeypatch.setattr(storage, "delete", fail_cleanup)

    response = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={"files": ("rollback.txt", b"rollback payload", "text/plain")},
    )

    assert response.status_code == 400
    assert response.json()["detail"]["message"] == "simulated database transaction failure"
    assert len(cleanup_attempts) == 1
    assert store.sources[task_id] == []


def test_preprocess_deduplicates_and_quality_filter_removes_short_results(
    tmp_path: Path,
) -> None:
    client, _, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={
            "name": "去重与质量筛选",
            "process_type": "structured",
            "config": {
                "preprocess_options": ["clean_invalid", "deduplicate"],
                "quality_filter_enabled": True,
                "filter_low_quality": False,
                "filter_short_content": True,
                "min_output_length": 100,
            },
        },
    ).json()["data"]["id"]
    content = (
        '{"question":"问题","answer":"短答案"}\n'
        '{"question":"问题","answer":"短答案"}\n'
    ).encode()
    uploaded = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={"files": ("duplicates.jsonl", content, "application/jsonl")},
    )
    assert uploaded.status_code == 200
    preview = client.post(f"/modelTF/data-process/{task_id}/preview/build")
    assert preview.json()["data"]["total"] == 1

    generated = client.post(f"/modelTF/data-process/{task_id}/generate")
    assert generated.status_code == 200
    progress = client.get(f"/modelTF/data-process/{task_id}/progress").json()["data"]
    assert progress["status"] == "completed"
    assert progress["filtered_count"] == 1
    assert client.get(f"/modelTF/data-process/{task_id}/results").json()["data"]["total"] == 0


def test_stale_generation_worker_cannot_overwrite_new_run(monkeypatch: Any) -> None:
    store = FakeDataProcessStore()
    task = store.create_task(
        {"name": "并发代次", "process_type": "structured", "config": {}}
    )
    task_id = task["id"]
    store.replace_preview_items(
        task_id,
        [
            {
                "source_file_id": None,
                "original_content": "来源内容",
                "edited_content": "来源内容",
                "status": "manual",
            }
        ],
    )
    first = store.start_generation(task_id, replace_existing=True)
    first_run_id = first["generation_run_id"]
    second_run_id = ""

    def restart_while_old_worker_runs(*_: Any, **__: Any) -> list[dict[str, Any]]:
        nonlocal second_run_id
        store.stop_task(task_id)
        second = store.start_generation(task_id, replace_existing=True)
        second_run_id = second["generation_run_id"]
        return []

    monkeypatch.setattr(
        data_process_endpoint,
        "generate_standard_records",
        restart_while_old_worker_runs,
    )
    data_process_endpoint._run_generation(store, task_id, first_run_id)

    assert second_run_id and second_run_id != first_run_id
    assert store.tasks[task_id]["status"] == "running"
    assert store.tasks[task_id]["generation_run_id"] == second_run_id
    assert store.results[task_id] == []
    store.mark_failed(task_id, "old failure", generation_run_id=first_run_id)
    assert store.tasks[task_id]["status"] == "running"


def test_reasoning_output_requires_generation_model() -> None:
    store = FakeDataProcessStore()
    task = store.create_task(
        {
            "name": "思维链模型校验",
            "process_type": "structured",
            "config": {"output_type": "reasoning"},
        }
    )
    task_id = task["id"]
    store.replace_preview_items(
        task_id,
        [
            {
                "source_file_id": None,
                "original_content": "需要推理的来源内容",
                "edited_content": "需要推理的来源内容",
                "status": "manual",
            }
        ],
    )
    started = store.start_generation(task_id, replace_existing=True)

    data_process_endpoint._run_generation(
        store,
        task_id,
        started["generation_run_id"],
    )

    failed = store.get_task(task_id)
    assert failed["status"] == "failed"
    assert failed["failure_reason"] == "思维链输出必须配置可用的数据生成模型"


def test_generation_failure_is_written_to_structured_log(caplog: pytest.LogCaptureFixture) -> None:
    store = FakeDataProcessStore()
    task = store.create_task(
        {
            "name": "生成失败日志",
            "process_type": "structured",
            "config": {"output_type": "reasoning"},
        }
    )
    task_id = task["id"]
    store.replace_preview_items(
        task_id,
        [
            {
                "source_file_id": None,
                "original_content": "需要推理的来源内容",
                "edited_content": "需要推理的来源内容",
                "status": "manual",
            }
        ],
    )
    started = store.start_generation(task_id, replace_existing=True)

    with caplog.at_level("INFO", logger=data_process_endpoint.__name__):
        data_process_endpoint._run_generation(
            store,
            task_id,
            started["generation_run_id"],
        )

    messages = [record.getMessage() for record in caplog.records]
    assert any("generation worker started" in message for message in messages)
    assert any("generation failed" in message for message in messages)


def test_result_status_cannot_be_forged_by_client(tmp_path: Path) -> None:
    client, _, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "状态保护", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]
    response = client.put(
        f"/modelTF/data-process/{task_id}/results/not-created",
        json={"instruction": "", "output": "", "status": "valid"},
    )
    assert response.status_code == 422


def test_start_rebuilds_preview_and_generates_in_one_request(tmp_path: Path) -> None:
    client, _, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "一键处理", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]
    uploaded = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={
            "files": (
                "one.jsonl",
                b'{"question":"What is one?","answer":"One."}\n',
                "application/jsonl",
            )
        },
    )
    assert uploaded.status_code == 200

    started = client.post(f"/modelTF/data-process/{task_id}/start")
    assert started.status_code == 200
    assert started.json()["data"]["task_id"] == task_id
    assert started.json()["data"]["status"] == "running"
    assert client.get(f"/modelTF/data-process/{task_id}/progress").json()["data"]["status"] == "completed"
    assert client.get(f"/modelTF/data-process/{task_id}/preview").json()["data"]["total"] == 1
    assert client.get(f"/modelTF/data-process/{task_id}/results").json()["data"]["total"] == 1


def test_unsupported_upload_format_returns_415(tmp_path: Path) -> None:
    client, _, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "格式限制", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]
    response = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={"files": ("payload.exe", b"not supported", "application/octet-stream")},
    )
    assert response.status_code == 415

    legacy = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={"files": ("document.doc", b"legacy", "application/msword")},
    )
    assert legacy.status_code == 415
    assert "convert the file to .docx" in legacy.json()["detail"]["message"]


def test_xlsx_upload_is_accepted_as_structured_records(tmp_path: Path) -> None:
    client, store, storage = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "XLSX 上传", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]
    workbook = Workbook()
    worksheet = workbook.active
    worksheet.append(["question", "answer"])
    worksheet.append(["问题一", "答案一"])
    worksheet.append(["问题二", "答案二"])
    output = BytesIO()
    workbook.save(output)
    workbook.close()
    original_bytes = output.getvalue()

    response = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={
            "files": (
                "records.xlsx",
                original_bytes,
                "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
            )
        },
    )
    assert response.status_code == 200
    source = response.json()["data"]["files"][0]
    assert source["file_format"] == "xlsx"
    assert source["record_count"] == 2
    assert source["size_bytes"] == len(original_bytes)
    assert source["storage_object_id"].startswith("local://data-process/")
    assert str(storage.root) not in response.text
    assert storage.read(source["storage_object_id"]) == original_bytes

    stored_source = store.get_source_file(task_id, source["id"])
    assert stored_source["id"] == source["id"]
    assert stored_source["storage_object_id"] == source["storage_object_id"]
    assert stored_source["metadata"]["storage_backend"] == "local"
    assert stored_source["metadata"]["original_size_bytes"] == len(original_bytes)
    assert '"question":"问题一"' in stored_source["content"]

    content = client.get(
        f"/modelTF/data-process/{task_id}/source-files/{source['id']}/content"
    )
    assert content.status_code == 200
    assert '"answer":"答案二"' in content.json()["data"]["content"]
    office_preview_url = (
        f"/modelTF/data-process/{task_id}/source-files/{source['id']}/office-preview"
    )
    first_page = client.get(office_preview_url, params={"offset": 0, "limit": 1})
    assert first_page.status_code == 200
    preview_data = first_page.json()["data"]
    assert preview_data["format"] == "xlsx"
    assert preview_data["sheets"] == [{"index": 0, "name": "Sheet", "state": "visible"}]
    assert preview_data["active_sheet"]["columns"] == ["question", "answer"]
    assert preview_data["active_sheet"]["rows"][0]["record"] == {
        "question": "问题一",
        "answer": "答案一",
    }
    assert preview_data["active_sheet"]["has_more"] is True

    second_page = client.get(office_preview_url, params={"offset": 1, "limit": 1})
    assert second_page.status_code == 200
    assert second_page.json()["data"]["active_sheet"]["rows"][0]["record"] == {
        "question": "问题二",
        "answer": "答案二",
    }
    assert second_page.json()["data"]["active_sheet"]["has_more"] is False

    raw_preview = client.get(
        f"/modelTF/data-process/{task_id}/source-files/{source['id']}/raw"
    )
    assert raw_preview.status_code == 200
    assert raw_preview.content == original_bytes
    assert raw_preview.headers["content-type"].startswith(
        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"
    )
    preview = client.post(f"/modelTF/data-process/{task_id}/preview/build")
    assert preview.status_code == 200
    assert preview.json()["data"]["total"] == 2


def test_docx_preview_preserves_document_block_order_and_source_offsets(
    tmp_path: Path,
) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "Word 原件预览", "process_type": "unstructured", "config": {}},
    ).json()["data"]["id"]
    document = WordDocument()
    document.add_heading("费用管理办法", level=1)
    document.add_paragraph("第一条 本办法用于规范费用报销。")
    table = document.add_table(rows=2, cols=2)
    table.cell(0, 0).text = "费用类型"
    table.cell(0, 1).text = "审批人"
    table.cell(1, 0).text = "差旅费"
    table.cell(1, 1).text = "部门负责人"
    output = BytesIO()
    document.save(output)
    original_bytes = output.getvalue()

    uploaded = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={
            "files": (
                "费用 管理.docx",
                original_bytes,
                "application/vnd.openxmlformats-officedocument.wordprocessingml.document",
            )
        },
    ).json()["data"]["files"][0]
    response = client.get(
        f"/modelTF/data-process/{task_id}/source-files/{uploaded['id']}/office-preview"
    )

    assert response.status_code == 200
    data = response.json()["data"]
    assert data["format"] == "docx"
    assert data["file_name"] == "费用 管理.docx"
    assert data["truncated"] is False
    assert [block["type"] for block in data["blocks"]] == [
        "paragraph",
        "paragraph",
        "table",
    ]
    assert data["blocks"][0]["heading_level"] == 1
    assert data["blocks"][0]["text"] == "费用管理办法"
    assert data["blocks"][2]["rows"][1]["cells"] == ["差旅费", "部门负责人"]

    source_text = store.get_source_file(task_id, uploaded["id"])["content"]
    first_paragraph = data["blocks"][0]
    assert (
        source_text[first_paragraph["source_start"] : first_paragraph["source_end"]]
        == first_paragraph["text"]
    )
    table_row = data["blocks"][2]["rows"][1]
    assert (
        source_text[table_row["source_start"] : table_row["source_end"]]
        == "\t".join(table_row["cells"])
    )

    raw_preview = client.get(
        f"/modelTF/data-process/{task_id}/source-files/{uploaded['id']}/raw"
    )
    assert raw_preview.status_code == 200
    assert raw_preview.content == original_bytes
    assert raw_preview.headers["content-type"].startswith(
        "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
    )


def test_pdf_raw_preview_streams_original_file_and_supports_ranges(tmp_path: Path) -> None:
    client, store, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "PDF 原件预览", "process_type": "unstructured", "config": {}},
    ).json()["data"]["id"]
    original_pdf = _minimal_pdf()
    uploaded = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={"files": ("说明 文档.pdf", original_pdf, "application/pdf")},
    ).json()["data"]["files"][0]
    raw_url = f"/modelTF/data-process/{task_id}/source-files/{uploaded['id']}/raw"

    full = client.get(raw_url)
    assert full.status_code == 200
    assert full.content == original_pdf
    assert full.headers["content-type"] == "application/pdf"
    assert full.headers["accept-ranges"] == "bytes"
    assert full.headers["cache-control"] == "private, no-store"
    assert full.headers["content-length"] == str(len(original_pdf))
    assert full.headers["content-disposition"].startswith("inline;")
    assert "%E8%AF%B4%E6%98%8E%20%E6%96%87%E6%A1%A3.pdf" in full.headers[
        "content-disposition"
    ]
    assert full.headers["etag"] == f'"{uploaded["checksum_sha256"]}"'

    partial = client.get(raw_url, headers={"Range": "bytes=5-14"})
    assert partial.status_code == 206
    assert partial.content == original_pdf[5:15]
    assert partial.headers["content-range"] == f"bytes 5-14/{len(original_pdf)}"
    assert partial.headers["content-length"] == "10"

    suffix = client.get(raw_url, headers={"Range": "bytes=-8"})
    assert suffix.status_code == 206
    assert suffix.content == original_pdf[-8:]

    invalid = client.get(raw_url, headers={"Range": "bytes=0-1,4-5"})
    assert invalid.status_code == 416
    assert invalid.headers["content-range"] == f"bytes */{len(original_pdf)}"

    pages_url = (
        f"/modelTF/data-process/{task_id}/source-files/{uploaded['id']}/pdf-pages"
    )
    pages = client.get(pages_url)
    assert pages.status_code == 200
    assert pages.json()["data"] == {
        "page_count": 1,
        "pages": [
            {
                "page_number": 1,
                "source_start": 0,
                "source_end": len("Hello PDF"),
            }
        ],
    }

    legacy_id = "dpsf_legacy_pdf"
    store.add_source_file(
        task_id,
        id=legacy_id,
        storage_object_id=f"db://data-process/{task_id}/{legacy_id}/v1",
        name="legacy.pdf",
        content="legacy extracted PDF text",
        raw_size=len(original_pdf),
        checksum_sha256="a" * 64,
        file_format="pdf",
        record_count=1,
        metadata={"legacy": True},
    )
    legacy = client.get(
        f"/modelTF/data-process/{task_id}/source-files/{legacy_id}/raw"
    )
    assert legacy.status_code == 410
    legacy_pages = client.get(
        f"/modelTF/data-process/{task_id}/source-files/{legacy_id}/pdf-pages"
    )
    assert legacy_pages.status_code == 410


def test_pdf_preview_build_cleans_stored_document_noise_without_offset_drift(
    tmp_path: Path,
) -> None:
    client, _, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={
            "name": "PDF 文档噪声清理",
            "process_type": "unstructured",
            "config": {
                "chunk_method": "fixed",
                "chunk_size": 200,
                "chunk_overlap": 0,
                "min_chunk_size": 20,
                "preprocess_options": ["clean_invalid_content"],
            },
        },
    ).json()["data"]["id"]
    raw = _minimal_pdf_pages(
        "ACME Internal Manual\nBody page one keeps this guidance and explanation.",
        "ACME Internal Manual\nContents\n"
        "Chapter One........3\nChapter Two........4\nAppendix........5",
        "ACME Internal Manual\n1.1 Policy........6\n1.2 Approval........7\n1.3 Archive........8",
        "ACME Internal Manual\nBody page four keeps operational details and examples.",
        "ACME Internal Manual\nBody page five keeps the final effective-date clause.",
    )
    uploaded = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={"files": ("manual.pdf", raw, "application/pdf")},
    ).json()["data"]["files"][0]
    source_content = client.get(
        f"/modelTF/data-process/{task_id}/source-files/{uploaded['id']}/content"
    ).json()["data"]["content"]

    built = client.post(f"/modelTF/data-process/{task_id}/preview/build")

    assert built.status_code == 200
    items = built.json()["data"]["items"]
    assert items
    edited = "\n".join(item["edited_content"] for item in items)
    assert "ACME Internal Manual" not in edited
    assert "Contents" not in edited
    assert "Chapter One" not in edited
    assert "1.2 Approval" not in edited
    assert "Body page one" in edited
    assert "Body page five" in edited
    assert all(
        item["original_content"]
        == source_content[item["source_start"] : item["source_end"]]
        for item in items
    )


def test_raw_inline_preview_rejects_non_pdf_source(tmp_path: Path) -> None:
    client, _, _ = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "非 PDF 原件", "process_type": "unstructured", "config": {}},
    ).json()["data"]["id"]
    uploaded = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={"files": ("notes.txt", b"plain source text", "text/plain")},
    ).json()["data"]["files"][0]

    response = client.get(
        f"/modelTF/data-process/{task_id}/source-files/{uploaded['id']}/raw"
    )
    assert response.status_code == 415
    pages_response = client.get(
        f"/modelTF/data-process/{task_id}/source-files/{uploaded['id']}/pdf-pages"
    )
    assert pages_response.status_code == 415


def test_delete_source_removes_owned_local_object_and_accepts_legacy_db_reference(
    tmp_path: Path,
) -> None:
    client, store, storage = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "删除原件", "process_type": "unstructured", "config": {}},
    ).json()["data"]["id"]
    uploaded = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={"files": ("原件.txt", "本地原始内容".encode(), "text/plain")},
    ).json()["data"]["files"][0]
    reference = uploaded["storage_object_id"]
    assert storage.read(reference) == "本地原始内容".encode()

    deleted = client.delete(
        f"/modelTF/data-process/{task_id}/source-files/{uploaded['id']}"
    )
    assert deleted.status_code == 200
    assert deleted.json()["data"]["storage_cleanup_pending"] is False
    with pytest.raises(DataProcessStorageError, match="does not exist"):
        storage.read(reference)

    legacy_id = "dpsf_legacy"
    store.add_source_file(
        task_id,
        id=legacy_id,
        storage_object_id=f"db://data-process/{task_id}/{legacy_id}/v1",
        name="legacy.txt",
        content="旧记录正文",
        raw_size=len("旧记录正文".encode()),
        checksum_sha256="a" * 64,
        file_format="txt",
        record_count=1,
        metadata={"legacy": True},
    )
    legacy_deleted = client.delete(
        f"/modelTF/data-process/{task_id}/source-files/{legacy_id}"
    )
    assert legacy_deleted.status_code == 200
    assert legacy_deleted.json()["data"]["storage_cleanup_pending"] is False


def test_delete_reports_pending_cleanup_after_database_soft_delete(
    tmp_path: Path,
    monkeypatch: Any,
) -> None:
    client, store, storage = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "待清理原件", "process_type": "unstructured", "config": {}},
    ).json()["data"]["id"]
    uploaded = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={"files": ("pending.txt", b"pending cleanup", "text/plain")},
    ).json()["data"]["files"][0]

    def fail_cleanup(*_: Any, **__: Any) -> bool:
        raise OSError("simulated storage failure")

    monkeypatch.setattr(storage, "delete", fail_cleanup)
    response = client.delete(
        f"/modelTF/data-process/{task_id}/source-files/{uploaded['id']}"
    )

    assert response.status_code == 200
    assert response.json()["data"]["storage_cleanup_pending"] is True
    with pytest.raises(NotFoundError):
        store.get_source_file(task_id, uploaded["id"])
    assert storage.read(uploaded["storage_object_id"]) == b"pending cleanup"


def test_delete_rejects_polluted_reference_owned_by_another_source(tmp_path: Path) -> None:
    client, store, storage = make_client(tmp_path)
    task_id = client.post(
        "/modelTF/data-process",
        json={"name": "归属校验", "process_type": "unstructured", "config": {}},
    ).json()["data"]["id"]
    uploaded = client.post(
        f"/modelTF/data-process/{task_id}/source-files",
        files={"files": ("safe.txt", b"owned content", "text/plain")},
    ).json()["data"]["files"][0]
    target_reference = uploaded["storage_object_id"]

    polluted_id = "dpsf_polluted"
    store.add_source_file(
        task_id,
        id=polluted_id,
        storage_object_id=target_reference,
        name="polluted.txt",
        content="polluted",
        raw_size=8,
        checksum_sha256="b" * 64,
        file_format="txt",
        record_count=1,
        metadata={},
    )
    rejected = client.delete(
        f"/modelTF/data-process/{task_id}/source-files/{polluted_id}"
    )
    assert rejected.status_code == 400
    assert storage.read(target_reference) == b"owned content"
    assert store.get_source_file(task_id, polluted_id)["id"] == polluted_id


def test_upload_format_must_match_process_type(tmp_path: Path) -> None:
    client, _, _ = make_client(tmp_path)
    structured_id = client.post(
        "/modelTF/data-process",
        json={"name": "结构化格式约束", "process_type": "structured", "config": {}},
    ).json()["data"]["id"]
    structured_pdf = client.post(
        f"/modelTF/data-process/{structured_id}/source-files",
        files={"files": ("manual.pdf", b"not parsed", "application/pdf")},
    )
    assert structured_pdf.status_code == 415

    unstructured_id = client.post(
        "/modelTF/data-process",
        json={"name": "非结构化格式约束", "process_type": "unstructured", "config": {}},
    ).json()["data"]["id"]
    unstructured_xlsx = client.post(
        f"/modelTF/data-process/{unstructured_id}/source-files",
        files={"files": ("records.xlsx", b"not parsed", "application/octet-stream")},
    )
    assert unstructured_xlsx.status_code == 415

    external_id = client.post(
        "/modelTF/data-process",
        json={"name": "外部数据格式约束", "process_type": "external", "config": {}},
    ).json()["data"]["id"]
    external_upload = client.post(
        f"/modelTF/data-process/{external_id}/source-files",
        files={"files": ("records.jsonl", b'{"id":1}', "application/jsonl")},
    )
    assert external_upload.status_code == 409


def _preview_task(
    content: str,
    *,
    options: list[str],
    config: dict[str, Any] | None = None,
    source_id: str = "source-1",
    file_format: str = "txt",
) -> list[dict[str, Any]]:
    task_config = {
        "preprocess_options": options,
        "chunk_method": "fixed",
        "chunk_size": 200,
        "chunk_overlap": 20,
        "min_chunk_size": 20,
        **(config or {}),
    }
    return data_process_endpoint._build_preview_items(
        {"process_type": "unstructured", "config": task_config},
        [
            {
                "id": source_id,
                "name": f"{source_id}.{file_format}",
                "file_format": file_format,
                "content": content,
            }
        ],
    )


def test_fixed_preview_preserves_source_offsets() -> None:
    content = (
        "# 第一章\n"
        + " ".join(f"alpha{index}" for index in range(18))
        + "\n# 第二章\n"
        + " ".join(f"beta{index}" for index in range(18))
    )
    normalized = normalize_text(content)
    second_chapter_start = normalized.index("# 第二章")
    common_config = {"chunk_size": 10, "chunk_overlap": 3, "min_chunk_size": 4}

    default_items = _preview_task(
        content,
        options=["preserve_context"],
        config=common_config,
    )
    fixed_items = _preview_task(
        content,
        options=["preserve_context"],
        config={**common_config, "chunk_method": "fixed"},
    )

    def snapshot(items: list[dict[str, Any]]) -> list[tuple[Any, ...]]:
        return [
            (
                item["original_content"],
                item["source_start"],
                item["source_end"],
                item["source_start_line"],
                item["source_end_line"],
            )
            for item in items
        ]

    assert snapshot(default_items) == snapshot(fixed_items)
    assert all(
        item["original_content"]
        == normalized[item["source_start"] : item["source_end"]]
        for item in fixed_items
    )
    second_chapter_items = [
        item for item in fixed_items if item["source_start"] >= second_chapter_start
    ]
    assert second_chapter_items


def test_every_unstructured_preprocess_option_changes_preview_behavior() -> None:
    repeated = "@" * 120
    assert len(_preview_task(repeated, options=[])) == 1
    assert _preview_task(repeated, options=["clean_invalid_content"]) == []

    mojibake = "这是无法可靠读取的内容，锟斤拷锟斤拷锟斤拷，需要预先过滤。"
    assert len(_preview_task(mojibake, options=[])) == 1
    assert _preview_task(mojibake, options=["filter_low_quality"]) == []

    first = "alpha beta gamma delta epsilon zeta eta theta iota kappa lambda mu nu xi omicron pi rho sigma tau upsilon phi chi psi omega"
    second = "alpha beta gamma, delta epsilon zeta eta theta iota kappa lambda mu nu xi omicron pi rho sigma tau upsilon phi chi psi omega"
    sources = [
        {"id": "near-1", "name": "one.txt", "file_format": "txt", "content": first},
        {"id": "near-2", "name": "two.txt", "file_format": "txt", "content": second},
    ]
    base_task = {
        "process_type": "unstructured",
        "config": {
            "chunk_method": "fixed",
            "chunk_size": 200,
            "chunk_overlap": 0,
            "min_chunk_size": 1,
            "preprocess_options": [],
        },
    }
    assert len(data_process_endpoint._build_preview_items(base_task, sources)) == 2
    deduplicated_task = deepcopy(base_task)
    deduplicated_task["config"]["preprocess_options"] = ["deduplicate_content"]
    assert len(data_process_endpoint._build_preview_items(deduplicated_task, sources)) == 1

    context_text = " ".join(f"token{index}" for index in range(45))
    no_context = _preview_task(
        context_text,
        options=[],
        config={"chunk_method": "fixed", "chunk_size": 20, "chunk_overlap": 5},
    )
    with_context = _preview_task(
        context_text,
        options=["preserve_context"],
        config={"chunk_method": "fixed", "chunk_size": 20, "chunk_overlap": 5},
    )
    assert no_context[1]["source_start"] >= no_context[0]["source_end"]
    assert with_context[1]["source_start"] < with_context[0]["source_end"]

    sensitive = "联系人：张三，手机 13800138000，邮箱 user@example.com。"
    plain = _preview_task(sensitive, options=[])[0]
    masked = _preview_task(sensitive, options=["desensitize"])[0]
    assert "张三" in plain["edited_content"]
    assert "联系人:[NAME]" in masked["edited_content"]
    assert "[PHONE]" in masked["edited_content"]
    assert "[EMAIL]" in masked["edited_content"]


def test_document_noise_cleaning_preserves_original_offsets_and_can_be_disabled() -> None:
    source_text = normalize_text("重复页眉\n这是应保留的 PDF 正文内容，用于生成训练数据。")
    source = {
        "id": "pdf-source",
        "name": "manual.pdf",
        "file_format": "pdf",
        "content": source_text,
        "document_noise_spans": (
            DocumentNoiseSpan(0, len("重复页眉"), "repeated_margin"),
        ),
    }
    config = {
        "chunk_method": "fixed",
        "chunk_size": 200,
        "chunk_overlap": 0,
        "min_chunk_size": 1,
    }

    cleaned_items = data_process_endpoint._build_preview_items(
        {
            "process_type": "unstructured",
            "config": {**config, "preprocess_options": ["clean_invalid_content"]},
        },
        [source],
    )
    original_items = data_process_endpoint._build_preview_items(
        {
            "process_type": "unstructured",
            "config": {**config, "preprocess_options": []},
        },
        [source],
    )

    assert len(cleaned_items) == 1
    cleaned = cleaned_items[0]
    assert cleaned["original_content"] == source_text[
        cleaned["source_start"] : cleaned["source_end"]
    ]
    assert "重复页眉" not in cleaned["edited_content"]
    assert "PDF 正文内容" in cleaned["edited_content"]
    assert cleaned["status"] == "modified"
    assert "document_noise_removed" in cleaned["quality_score"]["preprocess_flags"]
    assert "重复页眉" in original_items[0]["edited_content"]


def test_merge_short_content_applies_across_adjacent_fixed_chunks() -> None:
    content = "\n".join(f"{index}. 小节{index}\n内容{index}。" for index in range(1, 9))
    items = _preview_task(
        content,
        options=["merge_short_content"],
        config={
            "chunk_method": "fixed",
            "chunk_size": 40,
            "chunk_overlap": 0,
            "min_chunk_size": 20,
        },
    )

    assert len(items) == 3
    assert all(item["token_count"] <= 40 for item in items)
    assert items[0]["source_start_line"] == 1
    assert items[-1]["source_end_line"] == 16


def test_stored_binary_document_text_is_not_reparsed_as_binary() -> None:
    for file_format in ("pdf", "docx", "pptx"):
        items = _preview_task(
            f"{file_format.upper()} 已抽取正文，可直接进入切片处理。",
            options=[],
            file_format=file_format,
        )
        assert len(items) == 1
        assert "已抽取正文" in items[0]["edited_content"]