feat: 完成数据处理接口与前端接入

backend/tests/test_data_process_algorithms.py

@@ -0,0 +1,279 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from app.modules.data_process.algorithms import (
+    chunk_unstructured,
+    desensitize_pii,
+    detect_text_format,
+    extract_structured_records,
+    generate_standard_records,
+    normalize_text,
+    parse_text_content,
+    record_fingerprint,
+    score_quality,
+    stable_split,
+)
+
+
+def test_parse_utf8_json_jsonl_csv_markdown_and_txt() -> None:
+    parsed_json = parse_text_content(
+        b'\xef\xbb\xbf{"data":[{"name":"\xe5\xbc\xa0\xe4\xb8\x89"}]}',
+        filename="records.json",
+    )
+    assert parsed_json.format == "json"
+    assert parsed_json.records == ({"name": "张三"},)
+
+    parsed_jsonl = parse_text_content('{"id":1}\n\n{"id":2}\n', filename="records.jsonl")
+    assert parsed_jsonl.format == "jsonl"
+    assert parsed_jsonl.records == ({"id": 1}, {"id": 2})
+
+    parsed_csv = parse_text_content("name,answer\r\nAlice,yes\r\nBob,no", filename="records.csv")
+    assert parsed_csv.format == "csv"
+    assert parsed_csv.text == "name,answer\nAlice,yes\nBob,no"
+    assert parsed_csv.records[1] == {"name": "Bob", "answer": "no"}
+
+    parsed_markdown = parse_text_content("# 标题\n\n正文", filename="README.md")
+    assert parsed_markdown.format == "markdown"
+    assert parsed_markdown.records == ()
+
+    parsed_txt = parse_text_content("普通文本", filename="note.txt")
+    assert parsed_txt.format == "txt"
+    assert parsed_txt.text == "普通文本"
+
+
+def test_invalid_utf8_and_malformed_structured_content_fail_loudly() -> None:
+    with pytest.raises(ValueError, match="not valid UTF-8"):
+        parse_text_content(b"\xff\xfe", filename="broken.txt")
+    with pytest.raises(ValueError, match="invalid JSONL at line 2"):
+        extract_structured_records('{"id":1}\nnot-json', "jsonl")
+    with pytest.raises(ValueError, match="more fields"):
+        extract_structured_records("a,b\n1,2,3", "csv")
+
+
+def test_detect_format_from_content_and_normalize() -> None:
+    assert detect_text_format(text='{"id":1}\n{"id":2}') == "jsonl"
+    assert detect_text_format(text="# Heading\ntext") == "markdown"
+    assert detect_text_format(text="a,b\n1,2") == "csv"
+    assert normalize_text("\ufeffＡＢＣ  \r\n第二\x00行\u200b\t \r\n") == "ABC\n第二行"
+
+
+def test_extract_json_scalar_and_nested_values_are_stable() -> None:
+    assert extract_structured_records("[1, true, null]", "json") == [
+        {"value": 1},
+        {"value": True},
+        {"value": None},
+    ]
+    result = extract_structured_records(
+        json.dumps({"items": [{"text": "  内容  "}], "ignored": 1}, ensure_ascii=False),
+        "json",
+    )
+    assert result == [{"text": "内容"}]
+
+
+def test_desensitize_pii_returns_masked_text_and_counts() -> None:
+    source = "邮箱 a.user+tag@example.com，手机 +86 13800138000，身份证 11010519491231002X。"
+    masked, counts = desensitize_pii(source)
+    assert masked == "邮箱 [EMAIL]，手机 [PHONE]，身份证 [ID_CARD]。"
+    assert counts == {"email": 1, "phone": 1, "id_card": 1, "total": 3}
+
+
+@pytest.mark.parametrize("method", ["semantic", "heading", "fixed", "custom"])
+def test_chunk_methods_preserve_offsets_and_always_advance(method: str) -> None:
+    text = "# 第一章\n" + "甲。" * 18 + "\n# 第二章\n" + "乙。" * 18
+    kwargs = {"custom_delimiter": "\\n"} if method == "custom" else {}
+    chunks = chunk_unstructured(
+        text,
+        method=method,  # type: ignore[arg-type]
+        chunk_size=12,
+        chunk_overlap=2,
+        min_chunk_size=4,
+        **kwargs,
+    )
+    assert len(chunks) > 1
+    assert all(chunk.content == normalize_text(text)[chunk.start : chunk.end] for chunk in chunks)
+    assert all(chunk.end > chunk.start for chunk in chunks)
+    assert all(left.start < right.start for left, right in zip(chunks, chunks[1:]))
+    assert all(chunk.start_line <= chunk.end_line for chunk in chunks)
+
+
+def test_fixed_chunk_overlap_is_exact_when_chunks_are_large_enough() -> None:
+    text = " ".join(f"token{i}" for i in range(30))
+    chunks = chunk_unstructured(
+        text,
+        method="fixed",
+        chunk_size=10,
+        chunk_overlap=3,
+        min_chunk_size=4,
+    )
+    first_tokens = chunks[0].content.split()
+    second_tokens = chunks[1].content.split()
+    assert first_tokens[-3:] == second_tokens[:3]
+    assert chunks[0].token_count == 10
+
+
+def test_chunk_line_numbers_treat_newline_as_previous_line_boundary() -> None:
+    chunks = chunk_unstructured(
+        "第一行。\n第二行。\n第三行。",
+        method="custom",
+        chunk_size=8,
+        chunk_overlap=0,
+        min_chunk_size=2,
+        custom_delimiter="\\n",
+    )
+    assert chunks[0].content.endswith("\n")
+    assert chunks[0].start_line == 1
+    assert chunks[0].end_line == 1
+    assert chunks[1].start_line == 2
+
+
+def test_heading_and_custom_boundaries_are_respected() -> None:
+    heading_text = "前言 " * 8 + "\n# 第二章\n" + "正文 " * 12
+    heading_chunks = chunk_unstructured(
+        heading_text,
+        method="heading",
+        chunk_size=20,
+        chunk_overlap=0,
+        min_chunk_size=4,
+    )
+    assert "# 第二章" not in heading_chunks[0].content
+    assert heading_chunks[1].content.startswith("#")
+
+    custom_chunks = chunk_unstructured(
+        "a b c d <CUT> e f g h i j",
+        method="custom",
+        chunk_size=8,
+        chunk_overlap=0,
+        min_chunk_size=2,
+        custom_delimiter="<CUT>",
+    )
+    assert custom_chunks[0].content.endswith("<CUT>")
+
+
+@pytest.mark.parametrize(
+    ("field", "block"),
+    [
+        (
+            "preserve_code_blocks",
+            "```python\n" + "\n".join(f"value_{i} = {i}" for i in range(30)) + "\n```",
+        ),
+        (
+            "preserve_tables",
+            "| 字段 | 说明 |\n| --- | --- |\n"
+            + "\n".join(f"| field_{i} | value_{i} |" for i in range(30)),
+        ),
+        (
+            "preserve_lists",
+            "\n".join(f"- 第 {i} 项需要完整保留" for i in range(30)),
+        ),
+    ],
+)
+def test_markdown_protected_blocks_are_not_split(field: str, block: str) -> None:
+    text = "前言。" * 15 + "\n" + block + "\n" + "结尾。" * 40
+    chunks = chunk_unstructured(
+        text,
+        method="fixed",
+        chunk_size=40,
+        chunk_overlap=0,
+        min_chunk_size=10,
+        **{field: True},
+    )
+    assert any(block in chunk.content for chunk in chunks)
+
+
+@pytest.mark.parametrize(
+    ("kwargs", "message"),
+    [
+        ({"chunk_size": 0}, "chunk_size"),
+        ({"chunk_size": 10, "chunk_overlap": 10}, "chunk_overlap"),
+        ({"chunk_size": 10, "chunk_overlap": 0, "min_chunk_size": 11}, "min_chunk_size"),
+        (
+            {"chunk_size": 10, "chunk_overlap": 5, "min_chunk_size": 6},
+            "cannot exceed",
+        ),
+        ({"method": "custom", "custom_delimiter": ""}, "custom_delimiter"),
+    ],
+)
+def test_chunk_configuration_validation(kwargs: dict[str, object], message: str) -> None:
+    with pytest.raises(ValueError, match=message):
+        chunk_unstructured("some text", **kwargs)  # type: ignore[arg-type]
+
+
+def test_quality_scoring_covers_all_dimensions_and_duplicates() -> None:
+    valid = {
+        "instruction": "如何修改收货地址？",
+        "input": "订单尚未发货",
+        "output": "可以在订单详情页申请修改收货地址。",
+    }
+    source = "订单尚未发货时，可以在订单详情页申请修改收货地址。"
+    first_score = score_quality(valid, min_output_length=10, source_content=source)
+    assert first_score.is_valid
+    assert first_score.completeness == 100
+    assert first_score.length == 100
+    assert first_score.readability >= 90
+    assert first_score.relevance >= 70
+    assert first_score.duplicate == 100
+
+    duplicate_score = score_quality(valid, known_fingerprints={first_score.fingerprint})
+    assert duplicate_score.duplicate == 0
+    assert "duplicate_record" in duplicate_score.flags
+
+    unrelated_score = score_quality(
+        valid,
+        min_output_length=10,
+        source_content="量子计算使用量子比特处理信息。",
+    )
+    assert unrelated_score.relevance < first_score.relevance
+    assert "low_source_relevance" in unrelated_score.flags
+
+    invalid_score = score_quality({"instruction": "", "output": "短"}, min_output_length=10)
+    assert not invalid_score.is_valid
+    assert {"missing_instruction", "output_too_short"}.issubset(invalid_score.flags)
+    assert record_fingerprint(valid) == record_fingerprint(dict(reversed(list(valid.items()))))
+
+
+def test_stable_split_is_reproducible_and_validates_ratios() -> None:
+    first = stable_split("record-42", seed="task-1")
+    assert stable_split("record-42", seed="task-1") == first
+    assert first in {"train", "validation", "test"}
+    assert stable_split("record-42", {"train": 100, "validation": 0, "test": 0}) == "train"
+    with pytest.raises(ValueError, match="sum to 100"):
+        stable_split("record", {"train": 80, "validation": 10, "test": 9})
+
+
+def test_generate_standard_records_supports_json_qa_and_stable_variants() -> None:
+    previews = [
+        {
+            "id": "preview-json",
+            "edited_content": json.dumps(
+                {"instruction": "问题", "input": "上下文", "output": "答案"},
+                ensure_ascii=False,
+            ),
+        },
+        {"id": "preview-qa", "editedContent": "问：如何操作？\n答：按步骤操作。"},
+    ]
+    records = generate_standard_records(
+        previews,
+        qa_pairs_per_item=2,
+        semantic_enrichment=True,
+        split={"train": 100, "validation": 0, "test": 0},
+        split_seed="task-1",
+    )
+    assert len(records) == 4
+    assert records[0]["instruction"] == "问题"
+    assert records[0]["input"] == "上下文"
+    assert records[0]["output"] == "答案"
+    assert records[1]["instruction"].endswith("问题")
+    assert records[2]["instruction"] == "如何操作?"
+    assert records[2]["output"] == "按步骤操作。"
+    assert all(record["status"] == "valid" for record in records)
+    assert all(record["split"] == "train" for record in records)
+    assert records == generate_standard_records(
+        previews,
+        qa_pairs_per_item=2,
+        semantic_enrichment=True,
+        split={"train": 100, "validation": 0, "test": 0},
+        split_seed="task-1",
+    )

backend/tests/test_data_process_api.py

@@ -0,0 +1,704 @@
+from __future__ import annotations
+
+from copy import deepcopy
+from typing import Any
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+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.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.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 = list(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,
+        }
+        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")
+        return deepcopy(self.tasks[task_id])
+
+    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 delete_task(self, task_id: str, **_: Any) -> None:
+        self.get_task(task_id)
+        if self.tasks[task_id]["status"] == "running":
+            raise InvalidStateError("running task must be stopped before deletion")
+        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)
+        source = {
+            "id": self._id("dpsf"),
+            "task_id": task_id,
+            "version_no": 1,
+            **deepcopy(payload),
+        }
+        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)
+        return [self.add_source_file(task_id, **payload) for payload in files]
+
+    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] = []
+
+    def replace_preview_items(
+        self, task_id: str, items: list[dict[str, Any]]
+    ) -> list[dict[str, Any]]:
+        self.previews[task_id] = [
+            {"id": self._id("dpp"), "task_id": task_id, **deepcopy(item)} for item in items
+        ]
+        self.results[task_id] = []
+        self.tasks[task_id]["progress"] = 20
+        return deepcopy(self.previews[task_id])
+
+    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")
+        if replace_existing:
+            self.results[task_id] = []
+        self.tasks[task_id].update(
+            status="running",
+            progress=30,
+            generation_run_id=self._id("dprun"),
+        )
+        return self.get_task(task_id)
+
+    def generation_is_running(self, task_id: str, generation_run_id: str) -> bool:
+        return (
+            self.tasks[task_id]["status"] == "running"
+            and self.tasks[task_id].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),
+            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",
+        )}
+        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 publish(self, task_id: str, payload: dict[str, Any]) -> dict[str, Any]:
+        task = self.tasks[task_id]
+        if task.get("output_dataset_id"):
+            return {"dataset": deepcopy(self.datasets[task["output_dataset_id"]]), "created": False}
+        if task["status"] != "completed":
+            raise InvalidStateError("only a completed task can be published")
+        dataset_id = self._id("dataset")
+        dataset = {"id": dataset_id, "name": payload["dataset_name"], "source_task_id": task_id}
+        self.datasets[dataset_id] = dataset
+        task["output_dataset_id"] = dataset_id
+        return {"dataset": deepcopy(dataset), "created": True}
+
+
+def make_client() -> tuple[TestClient, FakeDataProcessStore]:
+    store = FakeDataProcessStore()
+    app = FastAPI()
+    app.include_router(router, prefix="/modelTF")
+    app.dependency_overrides[get_data_process_store] = lambda: store
+    return TestClient(app), store
+
+
+def test_data_process_full_contract_without_database() -> None:
+    client, store = make_client()
+    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
+    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
+    progress = client.get(f"/modelTF/data-process/{task_id}/progress")
+    assert progress.json()["data"]["status"] == "completed"
+    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
+
+    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_external_source_never_returns_fake_success() -> None:
+    client, _ = make_client()
+    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_config_validation_and_stop_state() -> None:
+    client, store = make_client()
+    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
+
+    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"
+
+
+def test_upload_batch_is_atomic_and_empty_files_are_rejected() -> None:
+    client, store = make_client()
+    task_id = client.post(
+        "/modelTF/data-process",
+        json={"name": "批量上传", "process_type": "structured", "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] == []
+
+    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] == []
+
+
+def test_preprocess_deduplicates_and_quality_filter_removes_short_results() -> None:
+    client, _ = make_client()
+    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_result_status_cannot_be_forged_by_client() -> None:
+    client, _ = make_client()
+    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() -> None:
+    client, _ = make_client()
+    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() -> None:
+    client, _ = make_client()
+    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": ("document.pdf", b"not a pdf", "application/pdf")},
+    )
+    assert response.status_code == 415

backend/tests/test_data_process_generation.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+import json
+
+import httpx
+
+from app.modules.data_process.generation import chat_completions_url, generate_model_records
+
+
+def test_chat_completions_url_accepts_host_base_and_complete_url() -> None:
+    assert chat_completions_url("www.caoxiaozhu.com") == (
+        "https://www.caoxiaozhu.com/v1/chat/completions"
+    )
+    assert chat_completions_url("https://model.example/v1") == (
+        "https://model.example/v1/chat/completions"
+    )
+    complete = "https://model.example/openai/v1/chat/completions"
+    assert chat_completions_url(complete) == complete
+
+
+def test_generate_model_records_uses_prompt_auth_and_stable_split() -> None:
+    requests: list[httpx.Request] = []
+    progress_updates: list[tuple[int, int]] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(request)
+        payload = json.loads(request.content)
+        assert payload["model"] == "qwen-plus"
+        assert payload["response_format"] == {"type": "json_object"}
+        assert "客户反馈页面加载慢" in payload["messages"][1]["content"]
+        return httpx.Response(
+            200,
+            json={
+                "choices": [
+                    {
+                        "message": {
+                            "content": json.dumps(
+                                {
+                                    "items": [
+                                        {
+                                            "instruction": "请生成简洁客服回复",
+                                            "input": "客户反馈页面加载慢",
+                                            "output": "已收到反馈，我们正在排查。",
+                                        }
+                                    ]
+                                },
+                                ensure_ascii=False,
+                            )
+                        }
+                    }
+                ]
+            },
+        )
+
+    client = httpx.Client(transport=httpx.MockTransport(handler))
+    records = generate_model_records(
+        [{"id": "preview-1", "edited_content": "客户反馈页面加载慢"}],
+        model={
+            "name": "Qwen",
+            "online_model_name": "qwen-plus",
+            "api_url": "model.example",
+            "api_key": "test-secret",
+        },
+        config={
+            "generation_prompt": "请处理：{{ content }}",
+            "json_mode": True,
+            "temperature": 0.2,
+            "max_tokens": 512,
+        },
+        task_id="task-1",
+        split={"train": 100, "validation": 0, "test": 0},
+        qa_pairs_per_item=1,
+        client=client,
+        on_progress=lambda processed, total: progress_updates.append((processed, total)),
+    )
+
+    assert len(records) == 1
+    assert records[0]["status"] == "valid"
+    assert records[0]["split"] == "train"
+    assert requests[0].headers["Authorization"] == "Bearer test-secret"
+    assert progress_updates == [(1, 1)]
+
+
+def test_generate_model_records_keeps_partial_failure_for_manual_repair() -> None:
+    client = httpx.Client(
+        transport=httpx.MockTransport(
+            lambda _: httpx.Response(200, json={"choices": [{"message": {"content": "not-json"}}]})
+        )
+    )
+    records = generate_model_records(
+        [{"id": "preview-1", "edited_content": "来源正文"}],
+        model={"name": "model", "api_url": "https://model.example/v1"},
+        config={"generation_retries": 1},
+        task_id="task-1",
+        split={"train": 80, "validation": 10, "test": 10},
+        qa_pairs_per_item=1,
+        client=client,
+    )
+
+    assert len(records) == 1
+    assert records[0]["status"] == "invalid"
+    assert records[0]["error"]

backend/tests/test_data_process_migration.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from app.modules.data_process.schema_cli import _target_label
+
+
+def test_runtime_migration_fails_fast_on_incompatible_schema() -> None:
+    sql_path = (
+        Path(__file__).resolve().parents[1]
+        / "app"
+        / "db"
+        / "sql"
+        / "002_data_process.sql"
+    )
+    sql = sql_path.read_text(encoding="utf-8")
+
+    assert "requires 001_platform_runtime.sql first" in sql
+    assert "supports only the current TEXT runtime schema" in sql
+    assert "generation_run_id" in sql
+    assert "CREATE TABLE IF NOT EXISTS data_process_results" in sql
+    assert sql.count("BEGIN;") == 1
+    assert sql.rstrip().endswith("COMMIT;")
+
+
+def test_schema_cli_target_label_never_contains_credentials() -> None:
+    label = _target_label("postgresql://secret-user:secret-password@db.example:5433/yg_ft")
+    assert label == "db.example:5433/yg_ft"
+    assert "secret" not in label