From b14b2ecf2250fb5d56fcca97c6bfaab6ea9d0691 Mon Sep 17 00:00:00 2001 From: caoxiaozhu Date: Mon, 27 Jul 2026 10:03:46 +0800 Subject: [PATCH] =?UTF-8?q?fix(data-process):=20=E4=BF=AE=E6=AD=A3?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E8=AF=A6=E6=83=85=E6=95=B0=E6=8D=AE=E5=A5=91?= =?UTF-8?q?=E7=BA=A6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/v1/endpoints/data_process.py | 4 +- backend/app/modules/data_process/store.py | 11 +- backend/tests/test_data_process_api.py | 172 ++++++++++++++++++- backend/tests/test_data_process_store.py | 71 ++++++++ 4 files changed, 247 insertions(+), 11 deletions(-) diff --git a/backend/app/api/v1/endpoints/data_process.py b/backend/app/api/v1/endpoints/data_process.py index afd8d05..abe91f5 100644 --- a/backend/app/api/v1/endpoints/data_process.py +++ b/backend/app/api/v1/endpoints/data_process.py @@ -716,7 +716,9 @@ def task_detail( ) -> dict[str, Any]: with api_errors(): task = store.get_task(task_id) - task["source_files"] = store.list_source_files(task_id) + source_files = store.list_source_files(task_id) + task["source_files"] = source_files + task["source_file_count"] = len(source_files) return ok(task) diff --git a/backend/app/modules/data_process/store.py b/backend/app/modules/data_process/store.py index cd547b1..b347dda 100644 --- a/backend/app/modules/data_process/store.py +++ b/backend/app/modules/data_process/store.py @@ -320,7 +320,12 @@ class DataProcessStore: 'name', dataset.name, 'type', dataset.type, 'count', dataset.count, - 'dataset_split', dataset.metadata::jsonb->>'dataset_split' + 'dataset_split', CASE dataset.type + WHEN 'train' THEN 'train' + WHEN 'val' THEN 'validation' + WHEN 'test' THEN 'test' + ELSE NULL + END ) ORDER BY CASE dataset.type WHEN 'train' THEN 1 WHEN 'val' THEN 2 WHEN 'test' THEN 3 ELSE 4 END), '[]'::json) FROM datasets dataset @@ -1077,8 +1082,8 @@ class DataProcessStore: """ UPDATE data_process_tasks SET status='running', progress=30, failure_reason=NULL, started_at=%s, - completed_at=NULL, filtered_count=0, duplicate_count=0, error_count=0, - generation_run_id=%s, updated_at=%s + completed_at=NULL, output_count=0, filtered_count=0, + duplicate_count=0, error_count=0, generation_run_id=%s, updated_at=%s WHERE id=%s RETURNING * """, diff --git a/backend/tests/test_data_process_api.py b/backend/tests/test_data_process_api.py index 4e98305..c0d9ad7 100644 --- a/backend/tests/test_data_process_api.py +++ b/backend/tests/test_data_process_api.py @@ -81,7 +81,18 @@ class FakeDataProcessStore: 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]) + 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 update_task(self, task_id: str, payload: dict[str, Any]) -> dict[str, Any]: self.get_task(task_id) @@ -334,6 +345,7 @@ class FakeDataProcessStore: self.tasks[task_id].update( status="running", progress=30, + output_count=0, generation_run_id=self._id("dprun"), ) return self.get_task(task_id) @@ -475,15 +487,53 @@ class FakeDataProcessStore: def publish(self, task_id: str, payload: dict[str, Any]) -> dict[str, Any]: task = self.tasks[task_id] + 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"): - return {"dataset": deepcopy(self.datasets[task["output_dataset_id"]]), "created": False} + 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") - 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} + 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( @@ -706,6 +756,56 @@ def test_task_list_exposes_document_and_generation_counts( assert item["output_dataset_id"] is None +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_preview_build_replaces_only_selected_files_and_reports_file_counts( tmp_path: Path, ) -> None: @@ -893,6 +993,64 @@ def test_regenerate_endpoint_prepares_an_existing_published_task(tmp_path: Path) assert store.previews[task_id][0]["id"] == "preview_1" +def test_published_split_datasets_remain_in_detail_after_regeneration( + 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-27T09:00:00Z", + "output_count": 1, + } + ) + store.results[task_id] = [ + { + "id": "result_1", + "status": "valid", + "instruction": "问题", + "input": "", + "output": "答案", + } + ] + + 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 + assert regenerated.json()["data"]["task"]["output_dataset_id"] is None + + detail = client.get(f"/modelTF/data-process/{task_id}") + assert detail.status_code == 200 + detail_data = detail.json()["data"] + assert detail_data["status"] == "pending" + assert detail_data["output_dataset_id"] is None + 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 + + def test_regenerate_endpoint_validates_snapshot_and_locked_process_type( tmp_path: Path, ) -> None: diff --git a/backend/tests/test_data_process_store.py b/backend/tests/test_data_process_store.py index 991c0d8..a2612ff 100644 --- a/backend/tests/test_data_process_store.py +++ b/backend/tests/test_data_process_store.py @@ -274,6 +274,10 @@ class _TaskDetailConnection: assert "dataset.source_task_id=task.id" in normalized assert "dataset.source_task_id IS NULL AND dataset.task_id=task.id" in normalized assert "dataset.deleted_at IS NULL" in normalized + assert "dataset.metadata::jsonb" not in normalized + assert "WHEN 'train' THEN 'train'" in normalized + assert "WHEN 'val' THEN 'validation'" in normalized + assert "WHEN 'test' THEN 'test'" in normalized task_id = params[0] visible = [ dataset @@ -364,6 +368,61 @@ class _TaskListStore(DataProcessStore): yield self._conn +class _StartGenerationConnection: + def __init__(self) -> None: + self.task = { + **_regeneration_task( + status="pending", + output_dataset_id=None, + output_count=28, + ), + "generation_run_id": None, + } + self.results = [{"id": "old-result"}] + + def execute(self, sql: str, params: Any = None) -> _Result: + normalized = " ".join(sql.split()) + if normalized.startswith("SELECT COUNT(*) AS count FROM data_process_preview_items"): + return _Result(row={"count": 1}) + if normalized.startswith("DELETE FROM data_process_results"): + self.results.clear() + return _Result() + assert normalized.startswith("UPDATE data_process_tasks SET status='running'") + assert "output_count=0" in normalized + self.task.update( + { + "status": "running", + "progress": 30, + "output_count": 0, + "failure_reason": None, + "started_at": params[0], + "completed_at": None, + "filtered_count": 0, + "duplicate_count": 0, + "error_count": 0, + "generation_run_id": params[1], + "updated_at": params[2], + } + ) + return _Result(row=dict(self.task)) + + +class _StartGenerationStore(DataProcessStore): + def __init__(self, conn: _StartGenerationConnection) -> None: + self._conn = conn + + @contextmanager + def connect(self) -> Iterator[_StartGenerationConnection]: + yield self._conn + + def _task_in_connection( + self, conn: Any, task_id: str, *, for_update: bool = False + ) -> dict[str, Any]: + assert task_id == "task-1" + assert for_update is True + return dict(self._conn.task) + + def test_decode_row_serializes_postgres_numeric_values_as_json_numbers() -> None: decoded = _decode_row( { @@ -522,6 +581,16 @@ def test_list_tasks_exposes_source_and_generation_counts() -> None: assert item["output_dataset_id"] is None +def test_start_generation_clears_previous_output_count() -> None: + conn = _StartGenerationConnection() + + task = _StartGenerationStore(conn).start_generation("task-1") + + assert task["status"] == "running" + assert task["output_count"] == 0 + assert conn.results == [] + + def _legacy_published_datasets(task_id: str = "task-1") -> list[dict[str, Any]]: specs = ( ("dataset_train", "制度问答-训练集", "train", "train", 22), @@ -585,6 +654,8 @@ def test_prepare_regeneration_preserves_outputs_sources_and_generation_only_prev def test_prepare_regeneration_backfills_and_keeps_legacy_task_datasets_visible() -> None: legacy_datasets = _legacy_published_datasets() + # 历史 metadata 可能不是合法 JSON，详情查询不能再依赖 metadata::jsonb。 + legacy_datasets[0]["metadata"] = "{legacy-invalid-json" deleted_dataset = { **legacy_datasets[0], "id": "dataset_deleted",