diff --git a/backend/app/api/v1/endpoints/data_process.py b/backend/app/api/v1/endpoints/data_process.py index abe91f5..9dd3b1f 100644 --- a/backend/app/api/v1/endpoints/data_process.py +++ b/backend/app/api/v1/endpoints/data_process.py @@ -715,6 +715,8 @@ def task_detail( store: DataProcessStore = Depends(get_data_process_store), ) -> dict[str, Any]: with api_errors(): + # 兼容旧版曾在第五步前清空结果的异常任务；严格特征匹配且幂等。 + store.recover_legacy_aborted_regeneration(task_id) task = store.get_task(task_id) source_files = store.list_source_files(task_id) task["source_files"] = source_files diff --git a/backend/app/modules/data_process/store.py b/backend/app/modules/data_process/store.py index fc49f60..520173f 100644 --- a/backend/app/modules/data_process/store.py +++ b/backend/app/modules/data_process/store.py @@ -408,6 +408,173 @@ class DataProcessStore: if task.get("output_dataset_id") and not _is_regeneration_prepared(task): raise InvalidStateError("published task cannot be edited") + def recover_legacy_aborted_regeneration(self, task_id: str) -> dict[str, Any]: + """恢复旧版在真正开始生成前误删的上一轮结果。 + + 旧实现会在 ``POST /regenerate`` 时立即把已发布任务置为 pending、 + 清空结果并解除输出指针。三个已发布数据集仍是独立完整产物，因此只在 + 这个特征完全匹配时，使用其记录恢复结果和任务状态。该操作幂等，不会 + 触碰正常的新建待生成任务或已经开始的新一轮生成。 + """ + + with self.connect() as conn: + task = self._task_in_connection(conn, task_id, for_update=True) + if ( + task.get("status") != "pending" + or task.get("generation_run_id") + or task.get("output_dataset_id") + or int(task.get("output_count") or 0) != 0 + ): + return {"recovered": False, "result_count": 0} + + result_count = int( + ( + conn.execute( + "SELECT COUNT(*) AS count FROM data_process_results WHERE task_id=%s", + (task_id,), + ).fetchone() + or {} + ).get("count") + or 0 + ) + if result_count: + return {"recovered": False, "result_count": result_count} + + datasets = conn.execute( + """ + SELECT id, type, count, created_at + FROM datasets + 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 CASE type + WHEN 'train' THEN 1 WHEN 'val' THEN 2 WHEN 'test' THEN 3 ELSE 4 + END, created_at, id + """, + (task_id, task_id), + ).fetchall() + train_dataset = next( + (dataset for dataset in datasets if dataset.get("type") == "train"), + None, + ) + if not train_dataset: + return {"recovered": False, "result_count": 0} + + dataset_ids = [str(dataset["id"]) for dataset in datasets] + records = conn.execute( + """ + SELECT id, dataset_id, line_no, split, instruction, input, output, + raw, status, source_result_id, preview_item_id, created_at + FROM dataset_records + WHERE dataset_id = ANY(%s) + ORDER BY created_at, dataset_id, line_no NULLS LAST, id + """, + (dataset_ids,), + ).fetchall() + if not records: + return {"recovered": False, "result_count": 0} + + preview_rows = conn.execute( + "SELECT id FROM data_process_preview_items WHERE task_id=%s", + (task_id,), + ).fetchall() + preview_ids = {str(row["id"]) for row in preview_rows} + used_result_ids: set[str] = set() + recovered_count = 0 + for record in records: + raw = _json_value(record.get("raw"), {}) + raw = raw if isinstance(raw, dict) else {} + candidate_id = str( + record.get("source_result_id") + or raw.get("source_result_id") + or "" + ) + result_id = ( + candidate_id + if candidate_id and candidate_id not in used_result_ids + else new_id("dpr") + ) + used_result_ids.add(result_id) + candidate_preview_id = str( + record.get("preview_item_id") + or raw.get("preview_item_id") + or "" + ) + preview_item_id = ( + candidate_preview_id if candidate_preview_id in preview_ids else None + ) + instruction = str(record.get("instruction") or raw.get("instruction") or "") + input_text = str(record.get("input") or raw.get("input") or "") + output = str(record.get("output") or raw.get("output") or "") + split = str(record.get("split") or raw.get("split") or "") or None + status = str(record.get("status") or "valid") + if status not in {"valid", "modified", "invalid"}: + status = "valid" + created_at = record.get("created_at") or utcnow() + conn.execute( + """ + INSERT INTO data_process_results + (id, task_id, preview_item_id, instruction, input, output, + original_instruction, original_input, original_output, status, + error, split, quality_score, created_at, updated_at) + VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, + NULL, %s, '{}', %s, %s) + """, + ( + result_id, + task_id, + preview_item_id, + instruction, + input_text, + output, + instruction, + input_text, + output, + status, + split, + created_at, + created_at, + ), + ) + conn.execute( + """ + UPDATE dataset_records + SET source_result_id=%s, preview_item_id=%s + WHERE id=%s + """, + (result_id, preview_item_id, record["id"]), + ) + recovered_count += 1 + + now = utcnow() + published_at = max( + (dataset.get("created_at") for dataset in datasets if dataset.get("created_at")), + default=None, + ) + conn.execute( + """ + UPDATE data_process_tasks + SET status='completed', progress=100, output_dataset_id=%s, + output_count=%s, filtered_count=0, duplicate_count=0, + error_count=(SELECT COUNT(*) FROM data_process_results + WHERE task_id=%s AND status='invalid'), + failure_reason=NULL, completed_at=COALESCE(completed_at, %s), + updated_at=%s + WHERE id=%s + """, + ( + train_dataset["id"], + recovered_count, + task_id, + published_at, + now, + task_id, + ), + ) + return {"recovered": True, "result_count": recovered_count} + def update_task(self, task_id: str, payload: dict[str, Any]) -> dict[str, Any]: allowed = { "name", @@ -504,6 +671,8 @@ class DataProcessStore: 生成时，才在同一事务内切换运行状态并清理上一轮结果。 """ + # 先修复曾被旧版 prepare 提前清空的任务，再建立新的非破坏性草稿标记。 + self.recover_legacy_aborted_regeneration(task_id) try: with self.connect() as conn: task = self._task_in_connection(conn, task_id, for_update=True) diff --git a/backend/tests/test_data_process_api.py b/backend/tests/test_data_process_api.py index f1b29c5..6dc36ef 100644 --- a/backend/tests/test_data_process_api.py +++ b/backend/tests/test_data_process_api.py @@ -95,6 +95,10 @@ class FakeDataProcessStore: ) 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)) diff --git a/backend/tests/test_data_process_store.py b/backend/tests/test_data_process_store.py index d316bef..2101688 100644 --- a/backend/tests/test_data_process_store.py +++ b/backend/tests/test_data_process_store.py @@ -357,6 +357,120 @@ class _TaskListStore(DataProcessStore): yield self._conn +class _LegacyRecoveryConnection: + def __init__(self) -> None: + self.task = { + **_regeneration_task( + status="pending", + progress=20, + output_dataset_id=None, + output_count=0, + generation_run_id=None, + started_at=None, + completed_at=None, + ), + } + self.datasets = [ + {"id": "dataset_train", "type": "train", "count": 1, "created_at": "2026-07-25T18:12:00Z"}, + {"id": "dataset_test", "type": "test", "count": 1, "created_at": "2026-07-25T18:12:00Z"}, + ] + self.records = [ + { + "id": "record_train", + "dataset_id": "dataset_train", + "line_no": 1, + "split": "train", + "instruction": "训练问题", + "input": "", + "output": "训练答案", + "raw": json.dumps( + { + "source_result_id": "result_train", + "preview_item_id": "preview_1", + } + ), + "status": "valid", + "source_result_id": None, + "preview_item_id": None, + "created_at": "2026-07-25T18:12:00Z", + }, + { + "id": "record_test", + "dataset_id": "dataset_test", + "line_no": 1, + "split": "test", + "instruction": "测试问题", + "input": "输入", + "output": "测试答案", + "raw": json.dumps({"source_result_id": "result_test"}), + "status": "modified", + "source_result_id": None, + "preview_item_id": None, + "created_at": "2026-07-25T18:12:00Z", + }, + ] + self.previews = [{"id": "preview_1"}] + self.results: list[dict[str, Any]] = [] + + def execute(self, sql: str, params: Any = None) -> _Result: + normalized = " ".join(sql.split()) + if normalized.startswith("SELECT COUNT(*) AS count FROM data_process_results"): + return _Result(row={"count": len(self.results)}) + if normalized.startswith("SELECT id, type, count, created_at FROM datasets"): + return _Result(rows=list(self.datasets)) + if normalized.startswith("SELECT id, dataset_id, line_no, split"): + return _Result(rows=list(self.records)) + if normalized.startswith("SELECT id FROM data_process_preview_items"): + return _Result(rows=list(self.previews)) + if normalized.startswith("INSERT INTO data_process_results"): + self.results.append( + { + "id": params[0], + "task_id": params[1], + "preview_item_id": params[2], + "instruction": params[3], + "input": params[4], + "output": params[5], + "status": params[9], + "split": params[10], + } + ) + return _Result() + if normalized.startswith("UPDATE dataset_records SET source_result_id="): + record = next(item for item in self.records if item["id"] == params[2]) + record["source_result_id"] = params[0] + record["preview_item_id"] = params[1] + return _Result() + if normalized.startswith("UPDATE data_process_tasks SET status='completed'"): + self.task.update( + { + "status": "completed", + "progress": 100, + "output_dataset_id": params[0], + "output_count": params[1], + "completed_at": params[3], + } + ) + return _Result() + raise AssertionError(f"unexpected SQL: {normalized}") + + +class _LegacyRecoveryStore(DataProcessStore): + def __init__(self, conn: _LegacyRecoveryConnection) -> None: + self._conn = conn + + @contextmanager + def connect(self) -> Iterator[_LegacyRecoveryConnection]: + 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) + + class _StartGenerationConnection: def __init__(self, *, published_prepared: bool = False, preview_count: int = 1) -> None: config = {"chunk_method": "fixed", "temperature": 0.7} @@ -616,6 +730,40 @@ def test_prepared_published_task_survives_generation_preflight_failure() -> None assert conn.results == [{"id": "old-result"}] +def test_legacy_aborted_regeneration_recovers_results_and_published_state() -> None: + conn = _LegacyRecoveryConnection() + store = _LegacyRecoveryStore(conn) + + recovered = store.recover_legacy_aborted_regeneration("task-1") + + assert recovered == {"recovered": True, "result_count": 2} + assert conn.task["status"] == "completed" + assert conn.task["progress"] == 100 + assert conn.task["output_dataset_id"] == "dataset_train" + assert conn.task["output_count"] == 2 + assert [item["id"] for item in conn.results] == ["result_train", "result_test"] + assert conn.results[0]["preview_item_id"] == "preview_1" + assert conn.results[1]["preview_item_id"] is None + assert conn.records[0]["source_result_id"] == "result_train" + assert conn.records[0]["preview_item_id"] == "preview_1" + + repeated = store.recover_legacy_aborted_regeneration("task-1") + assert repeated == {"recovered": False, "result_count": 0} + assert len(conn.results) == 2 + + +def test_normal_pending_task_is_not_mistaken_for_legacy_regeneration() -> None: + conn = _LegacyRecoveryConnection() + conn.datasets = [] + conn.records = [] + + result = _LegacyRecoveryStore(conn).recover_legacy_aborted_regeneration("task-1") + + assert result == {"recovered": False, "result_count": 0} + assert conn.task["status"] == "pending" + assert conn.results == [] + + def _legacy_published_datasets(task_id: str = "task-1") -> list[dict[str, Any]]: specs = ( ("dataset_train", "制度问答-训练集", "train", "train", 22),