diff --git a/backend/app/modules/data_process/store.py b/backend/app/modules/data_process/store.py index 6765460..79bab2e 100644 --- a/backend/app/modules/data_process/store.py +++ b/backend/app/modules/data_process/store.py @@ -174,6 +174,7 @@ def _decode_row(row: dict[str, Any] | None) -> dict[str, Any] | None: "metadata": {}, "quality_score": {}, "versions": [], + "output_datasets": [], }.items(): if key in item: item[key] = _json_value(item[key], default) @@ -318,7 +319,12 @@ class DataProcessStore: ) ORDER BY CASE dataset.type WHEN 'train' THEN 1 WHEN 'val' THEN 2 WHEN 'test' THEN 3 ELSE 4 END), '[]'::json) FROM datasets dataset - WHERE dataset.source_task_id=task.id AND dataset.source='task') + WHERE dataset.source='task' + AND dataset.deleted_at IS NULL + AND ( + dataset.source_task_id=task.id + OR (dataset.source_task_id IS NULL AND dataset.task_id=task.id) + )) AS output_datasets, CASE WHEN task.started_at IS NOT NULL AND task.completed_at IS NOT NULL @@ -459,14 +465,30 @@ class DataProcessStore: current_config, next_config, ) + now = utcnow() + # 002 迁移前发布的数据集只有 task_id。先补齐新关联字段，保证 + # 解除任务输出指针后，详情和后续重新发布仍能定位原来的三份数据集。 + conn.execute( + """ + UPDATE datasets + SET source_task_id=%s, updated_at=%s + WHERE source='task' AND source_task_id IS NULL AND task_id=%s + AND deleted_at IS NULL + """, + (task_id, now, task_id), + ) published_row = conn.execute( """ SELECT EXISTS( SELECT 1 FROM datasets - WHERE source_task_id=%s AND source='task' + WHERE source='task' AND deleted_at IS NULL + AND ( + source_task_id=%s + OR (source_task_id IS NULL AND task_id=%s) + ) ) AS exists """, - (task_id,), + (task_id, task_id), ).fetchone() published_outputs_preserved = bool(task.get("output_dataset_id")) or bool( published_row and published_row.get("exists") @@ -493,7 +515,6 @@ class DataProcessStore: ).fetchone() preview_count = int((preview_row or {}).get("count") or 0) - now = utcnow() row = conn.execute( """ UPDATE data_process_tasks @@ -1460,10 +1481,14 @@ class DataProcessStore: existing_datasets = conn.execute( """ SELECT * FROM datasets - WHERE source_task_id=%s AND source='task' + WHERE source='task' AND deleted_at IS NULL + AND ( + source_task_id=%s + OR (source_task_id IS NULL AND task_id=%s) + ) ORDER BY created_at, id """, - (task_id,), + (task_id, task_id), ).fetchall() existing_by_split: dict[str, dict[str, Any]] = {} primary_existing = None diff --git a/backend/tests/test_data_process_store.py b/backend/tests/test_data_process_store.py index 761faec..a673f73 100644 --- a/backend/tests/test_data_process_store.py +++ b/backend/tests/test_data_process_store.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json from collections.abc import Iterator from contextlib import contextmanager from decimal import Decimal @@ -31,9 +32,13 @@ class _Result: class _PublishConnection: - def __init__(self, results: list[dict[str, Any]]): + def __init__( + self, + results: list[dict[str, Any]], + datasets: list[dict[str, Any]] | None = None, + ): self.results = results - self.datasets: list[dict[str, Any]] = [] + self.datasets: list[dict[str, Any]] = datasets or [] self.files: list[dict[str, Any]] = [] self.records: list[dict[str, Any]] = [] @@ -47,16 +52,38 @@ class _PublishConnection: ) if normalized.startswith("SELECT * FROM data_process_results"): return _Result(rows=self.results) - if normalized.startswith("SELECT * FROM datasets WHERE source_task_id"): - return _Result(rows=self.datasets) + if normalized.startswith("SELECT * FROM datasets WHERE source='task'"): + assert "source_task_id=%s" in normalized + assert "source_task_id IS NULL AND task_id=%s" in normalized + assert "deleted_at IS NULL" in normalized + source_task_id, legacy_task_id = params + return _Result( + rows=[ + item + for item in self.datasets + if item.get("source") == "task" + and item.get("deleted_at") is None + and ( + item.get("source_task_id") == source_task_id + or ( + item.get("source_task_id") is None + and item.get("task_id") == legacy_task_id + ) + ) + ] + ) if normalized.startswith("INSERT INTO datasets"): dataset = { "id": params[0], "name": params[1], "type": params[2], + "source": "task", + "task_id": params[4], + "source_task_id": params[5], "count": params[8], "record_count": params[9], "metadata": params[11], + "deleted_at": None, } self.datasets.append(dataset) return _Result(row=dataset) @@ -118,12 +145,34 @@ class _PublishStore(DataProcessStore): class _RegenerationConnection: - def __init__(self, task: dict[str, Any]) -> None: + def __init__( + self, + task: dict[str, Any], + datasets: list[dict[str, Any]] | None = None, + ) -> None: self.task = task - self.datasets = [ - {"id": "dataset_train"}, - {"id": "dataset_validation"}, - {"id": "dataset_test"}, + self.datasets = datasets or [ + { + "id": "dataset_train", + "source": "task", + "task_id": task["id"], + "source_task_id": task["id"], + "deleted_at": None, + }, + { + "id": "dataset_validation", + "source": "task", + "task_id": task["id"], + "source_task_id": task["id"], + "deleted_at": None, + }, + { + "id": "dataset_test", + "source": "task", + "task_id": task["id"], + "source_task_id": task["id"], + "deleted_at": None, + }, ] self.sources = [{"id": "source_1"}] self.previews = [{"id": "preview_1"}] @@ -133,8 +182,35 @@ class _RegenerationConnection: normalized = " ".join(sql.split()) if params is not None: assert normalized.count("%s") == len(params) + if normalized.startswith("UPDATE datasets SET source_task_id="): + source_task_id, _, legacy_task_id = params + for dataset in self.datasets: + if ( + dataset.get("source") == "task" + and dataset.get("source_task_id") is None + and dataset.get("task_id") == legacy_task_id + and dataset.get("deleted_at") is None + ): + dataset["source_task_id"] = source_task_id + return _Result() if normalized.startswith("SELECT EXISTS("): - return _Result(row={"exists": bool(self.datasets)}) + assert "source_task_id=%s" in normalized + assert "source_task_id IS NULL AND task_id=%s" in normalized + assert "deleted_at IS NULL" in normalized + source_task_id, legacy_task_id = params + exists = any( + dataset.get("source") == "task" + and dataset.get("deleted_at") is None + and ( + dataset.get("source_task_id") == source_task_id + or ( + dataset.get("source_task_id") is None + and dataset.get("task_id") == legacy_task_id + ) + ) + for dataset in self.datasets + ) + return _Result(row={"exists": exists}) if normalized.startswith("DELETE FROM data_process_results"): self.results.clear() return _Result() @@ -183,6 +259,63 @@ class _RegenerationStore(DataProcessStore): return dict(self._conn.task) +class _TaskDetailConnection: + def __init__( + self, + task: dict[str, Any], + datasets: list[dict[str, Any]], + ) -> None: + self.task = task + self.datasets = datasets + + def execute(self, sql: str, params: Any = None) -> _Result: + normalized = " ".join(sql.split()) + assert normalized.startswith("SELECT task.*") + 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 + task_id = params[0] + visible = [ + dataset + for dataset in self.datasets + if dataset.get("source") == "task" + and dataset.get("deleted_at") is None + and ( + dataset.get("source_task_id") == task_id + or ( + dataset.get("source_task_id") is None + and dataset.get("task_id") == task_id + ) + ) + ] + return _Result( + row={ + **self.task, + "output_datasets": json.dumps( + [ + { + "id": item["id"], + "name": item["name"], + "type": item["type"], + "count": item["count"], + "dataset_split": item["dataset_split"], + } + for item in visible + ] + ), + } + ) + + +class _TaskDetailStore(DataProcessStore): + def __init__(self, conn: _TaskDetailConnection) -> None: + self._conn = conn + + @contextmanager + def connect(self) -> Iterator[_TaskDetailConnection]: + yield self._conn + + def test_decode_row_serializes_postgres_numeric_values_as_json_numbers() -> None: decoded = _decode_row( { @@ -194,6 +327,20 @@ def test_decode_row_serializes_postgres_numeric_values_as_json_numbers() -> None assert decoded == {"progress": 100.0, "duration_seconds": 389.0} +def test_decode_row_decodes_aggregated_output_datasets_json() -> None: + decoded = _decode_row( + { + "id": "task-1", + "output_datasets": '[{"id":"dataset_train","type":"train"}]', + } + ) + + assert decoded == { + "id": "task-1", + "output_datasets": [{"id": "dataset_train", "type": "train"}], + } + + @pytest.mark.parametrize( ("process_type", "current", "next_config", "expected"), [ @@ -314,6 +461,37 @@ def _regeneration_task(**updates: Any) -> dict[str, Any]: return task +def _legacy_published_datasets(task_id: str = "task-1") -> list[dict[str, Any]]: + specs = ( + ("dataset_train", "制度问答-训练集", "train", "train", 22), + ("dataset_validation", "制度问答-验证集", "val", "validation", 3), + ("dataset_test", "制度问答-测试集", "test", "test", 3), + ) + dataset_ids = {split: dataset_id for dataset_id, _, _, split, _ in specs} + return [ + { + "id": dataset_id, + "name": name, + "type": dataset_type, + "source": "task", + "task_id": task_id, + "source_task_id": None, + "count": count, + "record_count": count, + "dataset_split": split, + "metadata": json.dumps( + { + "base_dataset_name": "制度问答", + "dataset_split": split, + "split_dataset_ids": dataset_ids, + } + ), + "deleted_at": None, + } + for dataset_id, name, dataset_type, split, count in specs + ] + + def test_prepare_regeneration_preserves_outputs_sources_and_generation_only_preview() -> None: conn = _RegenerationConnection(_regeneration_task()) original_datasets = list(conn.datasets) @@ -344,6 +522,64 @@ def test_prepare_regeneration_preserves_outputs_sources_and_generation_only_prev assert conn.sources == original_sources +def test_prepare_regeneration_backfills_and_keeps_legacy_task_datasets_visible() -> None: + legacy_datasets = _legacy_published_datasets() + deleted_dataset = { + **legacy_datasets[0], + "id": "dataset_deleted", + "name": "已删除训练集", + "deleted_at": "2026-07-25T20:00:00Z", + } + unrelated_dataset = { + **legacy_datasets[0], + "id": "dataset_unrelated", + "name": "其他任务训练集", + "task_id": "task-other", + } + conn = _RegenerationConnection( + _regeneration_task(), + [*legacy_datasets, deleted_dataset, unrelated_dataset], + ) + + before = _TaskDetailStore(_TaskDetailConnection(conn.task, conn.datasets)).get_task( + "task-1" + ) + assert [item["id"] for item in before["output_datasets"]] == [ + "dataset_train", + "dataset_validation", + "dataset_test", + ] + + result = _RegenerationStore(conn).prepare_regeneration( + "task-1", + { + "name": "原任务", + "description": "", + "process_type": "unstructured", + "config": {"chunk_method": "fixed", "temperature": 0.2}, + "expected_updated_at": "2026-07-25T18:05:00Z", + }, + ) + + assert result["published_outputs_preserved"] is True + assert result["task"]["output_dataset_id"] is None + assert len(conn.datasets) == 5 + assert all( + item["source_task_id"] == "task-1" for item in conn.datasets[:3] + ) + assert deleted_dataset["source_task_id"] is None + assert unrelated_dataset["source_task_id"] is None + + after = _TaskDetailStore(_TaskDetailConnection(conn.task, conn.datasets)).get_task( + "task-1" + ) + assert [item["id"] for item in after["output_datasets"]] == [ + "dataset_train", + "dataset_validation", + "dataset_test", + ] + + def test_prepare_regeneration_deletes_preview_when_chunk_configuration_changes() -> None: conn = _RegenerationConnection(_regeneration_task(output_dataset_id=None)) @@ -467,6 +703,37 @@ def test_publish_creates_three_independent_datasets_with_exact_counts() -> None: assert republished["created"] is False +def test_publish_reuses_legacy_task_id_only_split_datasets() -> None: + results = [ + { + "id": f"result-{index}", + "status": "valid", + "instruction": f"问题 {index}", + "input": "", + "output": f"答案 {index}", + "preview_item_id": f"preview-{index}", + } + for index in range(10) + ] + legacy_datasets = _legacy_published_datasets() + original_ids = [item["id"] for item in legacy_datasets] + conn = _PublishConnection(results, legacy_datasets) + + published = _PublishStore(conn).publish( + "task-1", + { + "dataset_name": "不会创建新数据集", + "storage_type": "local", + "format": "alpaca_jsonl", + "split": {"train": 80, "validation": 10, "test": 10}, + }, + ) + + assert published["created"] is False + assert [item["id"] for item in published["datasets"]] == original_ids + assert [item["id"] for item in conn.datasets] == original_ids + + def test_publish_keeps_all_three_datasets_when_a_small_split_is_empty() -> None: conn = _PublishConnection( [