YG-Soft/YG_FT
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(data-process): 支持任务重新生成

Browse Source
This commit is contained in:
caoxiaozhu
2026-07-25 22:40:55 +08:00
parent 64d7414b04
commit 396d3f6f47
5 changed files with 667 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
backend/app/api/v1/endpoints/data_process.py
View File

@@ -68,6 +68,7 @@ from app.modules.data_process.store import (
new_id,
)
from app.schemas.data_process import (
DataProcessRegenerateRequest,
DataProcessStatus,
DataProcessTaskCreate,
DataProcessTaskUpdate,
@@ -732,6 +733,19 @@ def update_task(
)
@router.post("/{task_id}/regenerate")
def prepare_regeneration(
task_id: str,
payload: DataProcessRegenerateRequest,
store: DataProcessStore = Depends(get_data_process_store),
) -> dict[str, Any]:
with api_errors():
return ok(
store.prepare_regeneration(task_id, payload.model_dump(mode="json")),
"data process task prepared for regeneration",
)
@router.delete("/{task_id}")
def delete_task(
task_id: str,

176
backend/app/modules/data_process/store.py
View File

@@ -20,6 +20,28 @@ from app.modules.data_process.algorithms import estimate_token_count, stable_spl
TASK_STATUSES = {"pending", "running", "completed", "failed", "stopped"}
EDITABLE_STATUSES = {"pending", "failed", "stopped", "completed"}
_PREVIEW_CONFIG_ALIASES = {
"preprocess_options": "preprocessOptions",
"chunk_method": "chunkMethod",
"chunk_size": "chunkSize",
"chunk_overlap": "chunkOverlap",
"min_chunk_size": "minChunkSize",
"semantic_breakpoint_percentile": "semanticBreakpointPercentile",
"preserve_tables": "preserveTables",
"preserve_code_blocks": "preserveCodeBlocks",
"preserve_lists": "preserveLists",
}
_UNSTRUCTURED_PREVIEW_DEFAULTS: dict[str, Any] = {
"chunk_method": "layout_hybrid",
"chunk_size": 800,
"chunk_overlap": 100,
"min_chunk_size": 100,
"semantic_breakpoint_percentile": 95,
"preserve_tables": True,
"preserve_code_blocks": True,
"preserve_lists": True,
}
class DataProcessStoreError(RuntimeError):
pass
@@ -64,6 +86,49 @@ def _json_value(value: Any, default: Any) -> Any:
return default
def _preview_config_value(config: dict[str, Any], key: str, default: Any) -> Any:
if key in config:
return config[key]
return config.get(_PREVIEW_CONFIG_ALIASES[key], default)
def _normalized_preprocess_options(config: dict[str, Any]) -> Any:
value = _preview_config_value(config, "preprocess_options", [])
if isinstance(value, (list, tuple, set)):
return tuple(sorted({str(item) for item in value}))
return value
def _preview_config_projection(
process_type: str,
config: dict[str, Any],
) -> dict[str, Any]:
"""只投影会改变预览切片的配置。
生成模型、提示词、温度等参数不影响源文切片，因此不应该
破坏用户已经校对过的预览内容。
"""
projection: dict[str, Any] = {
"preprocess_options": _normalized_preprocess_options(config),
}
if process_type != "unstructured":
return projection
for key, default in _UNSTRUCTURED_PREVIEW_DEFAULTS.items():
projection[key] = _preview_config_value(config, key, default)
return projection
def _preview_config_changed(
process_type: str,
current_config: dict[str, Any],
next_config: dict[str, Any],
) -> bool:
return _preview_config_projection(process_type, current_config) != _preview_config_projection(
process_type, next_config
)
def _serialize_value(value: Any) -> Any:
if isinstance(value, (datetime, date)):
return value.isoformat().replace("+00:00", "Z")
@@ -361,6 +426,102 @@ class DataProcessStore:
raise ConflictError("data process task name already exists") from exc
return _decode_row(row) or {}
def prepare_regeneration(
self,
task_id: str,
payload: dict[str, Any],
) -> dict[str, Any]:
"""在一个事务中将已有任务恢复为可重新生成状态。
已发布数据集是可被其他任务使用的独立产物，此处只解除当前任务的
输出指针，不删除数据集实体。用户完成新一轮生成后，可再次发布
以原子替换三个分割数据集的内容。
"""
try:
with self.connect() as conn:
task = self._task_in_connection(conn, task_id, for_update=True)
if task["status"] == "running":
raise ConflictError("running task cannot be prepared for regeneration")
current_updated_at = _serialize_value(task.get("updated_at"))
if payload["expected_updated_at"] != current_updated_at:
raise ConflictError("data process task was modified by another request")
process_type = str(payload["process_type"])
if process_type != str(task["process_type"]):
raise InvalidStateError("process_type cannot be changed during regeneration")
current_config = dict(task.get("config") or {})
next_config = dict(payload.get("config") or {})
preview_invalidated = _preview_config_changed(
process_type,
current_config,
next_config,
)
published_row = conn.execute(
"""
SELECT EXISTS(
SELECT 1 FROM datasets
WHERE source_task_id=%s AND source='task'
) AS exists
""",
(task_id,),
).fetchone()
published_outputs_preserved = bool(task.get("output_dataset_id")) or bool(
published_row and published_row.get("exists")
)
# 删除顺序保证结果不再指向即将被替换的切片。
conn.execute(
"DELETE FROM data_process_results WHERE task_id=%s",
(task_id,),
)
if preview_invalidated:
conn.execute(
"DELETE FROM data_process_preview_items WHERE task_id=%s",
(task_id,),
)
preview_count = 0
else:
preview_row = conn.execute(
"""
SELECT COUNT(*) AS count FROM data_process_preview_items
WHERE task_id=%s
""",
(task_id,),
).fetchone()
preview_count = int((preview_row or {}).get("count") or 0)
now = utcnow()
row = conn.execute(
"""
UPDATE data_process_tasks
SET name=%s, description=%s, config=%s, status='pending', progress=%s,
output_dataset_id=NULL, output_count=0, filtered_count=0,
duplicate_count=0, error_count=0, failure_reason=NULL,
generation_run_id=NULL, started_at=NULL, completed_at=NULL,
updated_at=%s
WHERE id=%s
RETURNING *
""",
(
payload["name"],
payload.get("description") or "",
json_dumps(next_config),
20 if preview_count else 0,
now,
task_id,
),
).fetchone()
except psycopg.errors.UniqueViolation as exc:
raise ConflictError("data process task name already exists") from exc
return {
"task": _decode_row(row) or {},
"preview_invalidated": preview_invalidated,
"published_outputs_preserved": published_outputs_preserved,
}
def delete_task(self, task_id: str, *, deleted_by: str | None = None) -> None:
with self.connect() as conn:
task = self._task_in_connection(conn, task_id, for_update=True)
@@ -480,12 +641,21 @@ class DataProcessStore:
),
).fetchone()
created.append(_decode_row(row) or {})
# 前端会只对本次新增的源文件构建预览，因此必须保留旧
# 文件的切片。任何源文件增加都会使旧生成结果失效。
conn.execute("DELETE FROM data_process_results WHERE task_id=%s", (task_id,))
conn.execute("DELETE FROM data_process_preview_items WHERE task_id=%s", (task_id,))
preview_row = conn.execute(
"""
SELECT COUNT(*) AS count FROM data_process_preview_items
WHERE task_id=%s
""",
(task_id,),
).fetchone()
preview_count = int((preview_row or {}).get("count") or 0)
conn.execute(
"""
UPDATE data_process_tasks
SET status='pending', progress=0, output_count=0, filtered_count=0,
SET status='pending', progress=%s, output_count=0, filtered_count=0,
duplicate_count=0, error_count=0, failure_reason=NULL,
generation_run_id=NULL, started_at=NULL, completed_at=NULL, input_count=(
SELECT COALESCE(SUM(record_count), 0)
@@ -494,7 +664,7 @@ class DataProcessStore:
), updated_at=%s
WHERE id=%s
""",
(task_id, now, task_id),
(20 if preview_count else 0, task_id, now, task_id),
)
except psycopg.errors.UniqueViolation as exc:
raise ConflictError(

30
backend/app/schemas/data_process.py
View File

@@ -156,6 +156,36 @@ class DataProcessTaskUpdate(BaseModel):
return self
class DataProcessRegenerateRequest(BaseModel):
"""以一份完整配置准备任务重新生成。
``expected_updated_at`` 用于防止详情页的旧快照覆盖其他人刚刚
保存的配置。重新生成不允许改变处理类型，避免旧源文件在新解析
规则下被静默误用。
"""
model_config = ConfigDict(extra="forbid")
name: str = Field(min_length=1, max_length=150)
description: str
process_type: ProcessType
config: dict[str, Any]
expected_updated_at: str = Field(min_length=1)
@field_validator("name")
@classmethod
def normalize_name(cls, value: str) -> str:
value = value.strip()
if not value:
raise ValueError("task name cannot be empty")
return value
@model_validator(mode="after")
def validate_config(self) -> "DataProcessRegenerateRequest":
_validate_process_config(self.config)
return self
class PreviewBuildRequest(BaseModel):
model_config = ConfigDict(extra="forbid")

163
backend/tests/test_data_process_api.py
View File

@@ -82,6 +82,43 @@ class FakeDataProcessStore:
self.tasks[task_id].update(deepcopy(payload))
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"))
self.results[task_id] = []
task.update(
{
"name": payload["name"],
"description": payload["description"],
"config": deepcopy(payload["config"]),
"status": "pending",
"progress": 20,
"output_dataset_id": None,
"output_count": 0,
"filtered_count": 0,
"duplicate_count": 0,
"error_count": 0,
"failure_reason": None,
"generation_run_id": None,
"started_at": None,
"completed_at": None,
"updated_at": "2026-07-25T20:00:00Z",
}
)
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)
if self.tasks[task_id]["status"] == "running":
@@ -133,7 +170,23 @@ class FakeDataProcessStore:
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]
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,
"started_at": None,
"completed_at": None,
}
)
return created
def get_source_file(
self, task_id: str, file_id: str, *, include_content: bool = True
@@ -769,6 +822,81 @@ def test_external_source_never_returns_fake_success(tmp_path: Path) -> None:
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.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"] == "pending"
assert data["task"]["output_dataset_id"] is None
assert data["preview_invalidated"] is False
assert data["published_outputs_preserved"] is True
assert store.results[task_id] == []
assert store.previews[task_id][0]["id"] == "preview_1"
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(
@@ -855,6 +983,39 @@ def test_upload_batch_is_atomic_and_empty_files_are_rejected(tmp_path: Path) ->
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,

288
backend/tests/test_data_process_store.py
View File

@@ -8,9 +8,12 @@ from typing import Any
import pytest
from app.modules.data_process.store import (
ConflictError,
DataProcessStore,
DataProcessStoreError,
InvalidStateError,
_decode_row,
_preview_config_changed,
_source_storage_descriptor,
)
@@ -114,6 +117,72 @@ class _PublishStore(DataProcessStore):
return []
class _RegenerationConnection:
def __init__(self, task: dict[str, Any]) -> None:
self.task = task
self.datasets = [
{"id": "dataset_train"},
{"id": "dataset_validation"},
{"id": "dataset_test"},
]
self.sources = [{"id": "source_1"}]
self.previews = [{"id": "preview_1"}]
self.results = [{"id": "result_1"}]
def execute(self, sql: str, params: Any = None) -> _Result:
normalized = " ".join(sql.split())
if params is not None:
assert normalized.count("%s") == len(params)
if normalized.startswith("SELECT EXISTS("):
return _Result(row={"exists": bool(self.datasets)})
if normalized.startswith("DELETE FROM data_process_results"):
self.results.clear()
return _Result()
if normalized.startswith("DELETE FROM data_process_preview_items"):
self.previews.clear()
return _Result()
if normalized.startswith("SELECT COUNT(*) AS count FROM data_process_preview_items"):
return _Result(row={"count": len(self.previews)})
if normalized.startswith("UPDATE data_process_tasks SET name="):
self.task.update(
{
"name": params[0],
"description": params[1],
"config": params[2],
"status": "pending",
"progress": params[3],
"output_dataset_id": None,
"output_count": 0,
"filtered_count": 0,
"duplicate_count": 0,
"error_count": 0,
"failure_reason": None,
"generation_run_id": None,
"started_at": None,
"completed_at": None,
"updated_at": params[4],
}
)
return _Result(row=dict(self.task))
raise AssertionError(f"unexpected SQL: {normalized}")
class _RegenerationStore(DataProcessStore):
def __init__(self, conn: _RegenerationConnection):
self._conn = conn
@contextmanager
def connect(self) -> Iterator[_RegenerationConnection]:
yield self._conn
def _task_in_connection(
self, conn: Any, task_id: str, *, for_update: bool = False
) -> dict[str, Any]:
assert for_update is True
assert task_id == self._conn.task["id"]
return dict(self._conn.task)
def test_decode_row_serializes_postgres_numeric_values_as_json_numbers() -> None:
decoded = _decode_row(
{
@@ -125,6 +194,225 @@ def test_decode_row_serializes_postgres_numeric_values_as_json_numbers() -> None
assert decoded == {"progress": 100.0, "duration_seconds": 389.0}
@pytest.mark.parametrize(
("process_type", "current", "next_config", "expected"),
[
(
"structured",
{"preprocess_options": ["deduplicate"]},
{"preprocess_options": ["deduplicate"], "temperature": 0.2},
False,
),
(
"structured",
{"preprocess_options": ["deduplicate"]},
{"preprocess_options": ["clean_invalid"]},
True,
),
(
"structured",
{"preprocess_options": ["a", "b"]},
{"preprocessOptions": ["b", "a", "a"], "chunk_size": 2048},
False,
),
(
"unstructured",
{"chunk_method": "fixed"},
{"chunk_method": "fixed", "generation_prompt": "new"},
False,
),
(
"unstructured",
{"chunk_method": "fixed"},
{"chunk_method": "semantic"},
True,
),
(
"unstructured",
{"chunk_size": 800, "chunk_overlap": 100},
{"chunk_size": 900, "chunk_overlap": 100},
True,
),
(
"unstructured",
{"min_chunk_size": 100},
{"min_chunk_size": 120},
True,
),
(
"unstructured",
{"semantic_breakpoint_percentile": 95},
{"semantic_breakpoint_percentile": 90},
True,
),
(
"unstructured",
{},
{
"preserve_tables": True,
"preserve_code_blocks": True,
"preserve_lists": True,
},
False,
),
(
"unstructured",
{"preserve_tables": False},
{"preserve_tables": True},
True,
),
(
"unstructured",
{"preserve_code_blocks": False},
{"preserve_code_blocks": True},
True,
),
(
"unstructured",
{"preserve_lists": False},
{"preserve_lists": True},
True,
),
(
"unstructured",
{"preprocess_options": ["deduplicate"]},
{"preprocess_options": ["clean_invalid"]},
True,
),
],
)
def test_regeneration_preview_invalidation_matrix(
process_type: str,
current: dict[str, Any],
next_config: dict[str, Any],
expected: bool,
) -> None:
assert _preview_config_changed(process_type, current, next_config) is expected
def _regeneration_task(**updates: Any) -> dict[str, Any]:
task = {
"id": "task-1",
"name": "原任务",
"description": "",
"process_type": "unstructured",
"config": {"chunk_method": "fixed", "temperature": 0.7},
"status": "completed",
"progress": 100,
"output_dataset_id": "dataset_train",
"output_count": 28,
"filtered_count": 1,
"duplicate_count": 1,
"error_count": 0,
"failure_reason": None,
"generation_run_id": None,
"started_at": "2026-07-25T18:00:00Z",
"completed_at": "2026-07-25T18:05:00Z",
"updated_at": "2026-07-25T18:05:00Z",
}
task.update(updates)
return task
def test_prepare_regeneration_preserves_outputs_sources_and_generation_only_preview() -> None:
conn = _RegenerationConnection(_regeneration_task())
original_datasets = list(conn.datasets)
original_sources = list(conn.sources)
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["preview_invalidated"] is False
assert result["published_outputs_preserved"] is True
assert result["task"]["output_dataset_id"] is None
assert result["task"]["status"] == "pending"
assert result["task"]["progress"] == 20
assert result["task"]["output_count"] == 0
assert result["task"]["started_at"] is None
assert result["task"]["completed_at"] is None
assert conn.results == []
assert conn.previews == [{"id": "preview_1"}]
assert conn.datasets == original_datasets
assert conn.sources == original_sources
def test_prepare_regeneration_deletes_preview_when_chunk_configuration_changes() -> None:
conn = _RegenerationConnection(_regeneration_task(output_dataset_id=None))
result = _RegenerationStore(conn).prepare_regeneration(
"task-1",
{
"name": "原任务",
"description": "",
"process_type": "unstructured",
"config": {"chunk_method": "semantic", "temperature": 0.7},
"expected_updated_at": "2026-07-25T18:05:00Z",
},
)
assert result["preview_invalidated"] is True
assert result["published_outputs_preserved"] is True
assert result["task"]["progress"] == 0
assert conn.previews == []
assert conn.results == []
assert len(conn.datasets) == 3
@pytest.mark.parametrize(
("task_updates", "payload_updates", "error_type", "message"),
[
(
{"status": "running"},
{},
ConflictError,
"running task cannot be prepared",
),
(
{},
{"expected_updated_at": "2026-07-25T17:00:00Z"},
ConflictError,
"modified by another request",
),
(
{},
{"process_type": "structured"},
InvalidStateError,
"process_type cannot be changed",
),
],
)
def test_prepare_regeneration_rejects_running_stale_and_type_change_without_mutation(
task_updates: dict[str, Any],
payload_updates: dict[str, Any],
error_type: type[Exception],
message: str,
) -> None:
conn = _RegenerationConnection(_regeneration_task(**task_updates))
payload = {
"name": "原任务",
"description": "",
"process_type": "unstructured",
"config": {"chunk_method": "fixed"},
"expected_updated_at": "2026-07-25T18:05:00Z",
**payload_updates,
}
with pytest.raises(error_type, match=message):
_RegenerationStore(conn).prepare_regeneration("task-1", payload)
assert conn.results == [{"id": "result_1"}]
assert conn.previews == [{"id": "preview_1"}]
assert conn.task["output_dataset_id"] == "dataset_train"
def test_publish_creates_three_independent_datasets_with_exact_counts() -> None:
results = [
{