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

fix(data-process): 延迟重新生成破坏性变更

Browse Source
This commit is contained in:
caoxiaozhu
2026-07-27 10:43:42 +08:00
parent 03bd0b6d03
commit 53014bb381
3 changed files with 299 additions and 143 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -41,6 +41,7 @@ _UNSTRUCTURED_PREVIEW_DEFAULTS: dict[str, Any] = {
"preserve_code_blocks": True,
"preserve_lists": True,
}
_REGENERATION_MARKER_KEY = "_regeneration_prepared"
class DataProcessStoreError(RuntimeError):
@@ -129,6 +130,34 @@ def _preview_config_changed(
)
def _regeneration_marker(task: dict[str, Any]) -> dict[str, Any] | None:
config = task.get("config")
if not isinstance(config, dict):
return None
marker = config.get(_REGENERATION_MARKER_KEY)
if not isinstance(marker, dict) or marker.get("prepared") is not True:
return None
return marker
def _is_regeneration_prepared(task: dict[str, Any]) -> bool:
return _regeneration_marker(task) is not None
def _public_task(item: dict[str, Any] | None) -> dict[str, Any] | None:
"""从 API 任务快照中移除服务端内部重新生成标记。"""
if item is None:
return None
public = dict(item)
config = public.get("config")
if isinstance(config, dict) and _REGENERATION_MARKER_KEY in config:
public["config"] = {
key: value for key, value in config.items() if key != _REGENERATION_MARKER_KEY
}
return public
def _serialize_value(value: Any) -> Any:
if isinstance(value, (datetime, date)):
return value.isoformat().replace("+00:00", "Z")
@@ -258,7 +287,7 @@ class DataProcessStore:
[*params, page_size, (page - 1) * page_size],
).fetchall()
return {
"items": [_decode_row(row) for row in rows],
"items": [_public_task(_decode_row(row)) for row in rows],
"total": int(total),
"page": page,
"page_size": page_size,
@@ -284,7 +313,13 @@ class DataProcessStore:
payload.get("description") or "",
payload["process_type"],
payload.get("source_dataset_id"),
json_dumps(payload.get("config") or {}),
json_dumps(
{
key: value
for key, value in (payload.get("config") or {}).items()
if key != _REGENERATION_MARKER_KEY
}
),
payload.get("tenant_id"),
payload.get("project_id"),
payload.get("owner_id"),
@@ -296,7 +331,7 @@ class DataProcessStore:
).fetchone()
except psycopg.errors.UniqueViolation as exc:
raise ConflictError("data process task name already exists") from exc
return _decode_row(row) or {}
return _public_task(_decode_row(row)) or {}
def get_task(self, task_id: str, *, for_update: bool = False) -> dict[str, Any]:
lock = " FOR UPDATE" if for_update else ""
@@ -348,7 +383,7 @@ class DataProcessStore:
).fetchone()
if not row:
raise NotFoundError("data process task not found")
return _decode_row(row) or {}
return _public_task(_decode_row(row)) or {}
def _task_in_connection(
self,
@@ -370,7 +405,7 @@ class DataProcessStore:
def _ensure_editable(task: dict[str, Any]) -> None:
if task["status"] not in EDITABLE_STATUSES:
raise InvalidStateError(f"task cannot be edited while status is {task['status']}")
if task.get("output_dataset_id"):
if task.get("output_dataset_id") and not _is_regeneration_prepared(task):
raise InvalidStateError("published task cannot be edited")
def update_task(self, task_id: str, payload: dict[str, Any]) -> dict[str, Any]:
@@ -381,14 +416,30 @@ class DataProcessStore:
"source_dataset_id",
}
values: dict[str, Any] = {key: value for key, value in payload.items() if key in allowed}
if payload.get("config") is not None:
values["config"] = json_dumps(payload["config"])
if not values:
if not values and payload.get("config") is None:
return self.get_task(task_id)
try:
with self.connect() as conn:
task = self._task_in_connection(conn, task_id, for_update=True)
self._ensure_editable(task)
regeneration_prepared = _is_regeneration_prepared(task)
if regeneration_prepared and any(
key in payload and payload.get(key) != task.get(key)
for key in ("process_type", "source_dataset_id")
):
raise InvalidStateError(
"process type and source dataset cannot change during regeneration"
)
if payload.get("config") is not None:
next_config = {
key: value
for key, value in payload["config"].items()
if key != _REGENERATION_MARKER_KEY
}
current_marker = _regeneration_marker(task)
if current_marker:
next_config[_REGENERATION_MARKER_KEY] = current_marker
values["config"] = json_dumps(next_config)
invalidates_results = (
("config" in payload and payload.get("config") != task.get("config"))
or (
@@ -400,7 +451,7 @@ class DataProcessStore:
and payload.get("source_dataset_id") != task.get("source_dataset_id")
)
)
if invalidates_results:
if invalidates_results and not regeneration_prepared:
values.update(
{
"status": "pending",
@@ -440,18 +491,17 @@ class DataProcessStore:
).fetchone()
except psycopg.errors.UniqueViolation as exc:
raise ConflictError("data process task name already exists") from exc
return _decode_row(row) or {}
return _public_task(_decode_row(row)) or {}
def prepare_regeneration(
self,
task_id: str,
payload: dict[str, Any],
) -> dict[str, Any]:
"""在一个事务中将已有任务恢复为可重新生成状态。
"""非破坏性地保存重新生成配置。
已发布数据集是可被其他任务使用的独立产物，此处只解除当前任务的
输出指针，不删除数据集实体。用户完成新一轮生成后，可再次发布
以原子替换三个分割数据集的内容。
准备阶段保留任务当前状态、结果、切片及已发布数据集。真正开始
生成时，才在同一事务内切换运行状态并清理上一轮结果。
"""
try:
@@ -470,12 +520,18 @@ class DataProcessStore:
current_config = dict(task.get("config") or {})
next_config = dict(payload.get("config") or {})
next_config.pop(_REGENERATION_MARKER_KEY, None)
preview_invalidated = _preview_config_changed(
process_type,
current_config,
next_config,
)
now = utcnow()
next_config[_REGENERATION_MARKER_KEY] = {
"prepared": True,
"preview_invalidated": preview_invalidated,
"prepared_at": now,
}
# 002 迁移前发布的数据集只有 task_id。先补齐新关联字段，保证
# 解除任务输出指针后，详情和后续重新发布仍能定位原来的三份数据集。
conn.execute(
@@ -504,35 +560,10 @@ class DataProcessStore:
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)
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
SET name=%s, description=%s, config=%s, updated_at=%s
WHERE id=%s
RETURNING *
""",
@@ -540,7 +571,6 @@ class DataProcessStore:
payload["name"],
payload.get("description") or "",
json_dumps(next_config),
20 if preview_count else 0,
now,
task_id,
),
@@ -548,7 +578,7 @@ class DataProcessStore:
except psycopg.errors.UniqueViolation as exc:
raise ConflictError("data process task name already exists") from exc
return {
"task": _decode_row(row) or {},
"task": _public_task(_decode_row(row)) or {},
"preview_invalidated": preview_invalidated,
"published_outputs_preserved": published_outputs_preserved,
}
@@ -672,9 +702,6 @@ class DataProcessStore:
),
).fetchone()
created.append(_decode_row(row) or {})
# 前端会只对本次新增的源文件构建预览，因此必须保留旧
# 文件的切片。任何源文件增加都会使旧生成结果失效。
conn.execute("DELETE FROM data_process_results WHERE task_id=%s", (task_id,))
preview_row = conn.execute(
"""
SELECT COUNT(*) AS count FROM data_process_preview_items
@@ -683,20 +710,40 @@ class DataProcessStore:
(task_id,),
).fetchone()
preview_count = int((preview_row or {}).get("count") or 0)
conn.execute(
"""
UPDATE data_process_tasks
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)
FROM data_process_source_files
WHERE task_id=%s AND deleted_at IS NULL
), updated_at=%s
WHERE id=%s
""",
(20 if preview_count else 0, task_id, now, task_id),
)
if _is_regeneration_prepared(task):
conn.execute(
"""
UPDATE data_process_tasks
SET input_count=(
SELECT COALESCE(SUM(record_count), 0)
FROM data_process_source_files
WHERE task_id=%s AND deleted_at IS NULL
), updated_at=%s
WHERE id=%s
""",
(task_id, now, task_id),
)
else:
# 前端会只对本次新增的源文件构建预览，因此保留旧文件切片，
# 但普通未发布任务的旧生成结果已经不再有效。
conn.execute(
"DELETE FROM data_process_results WHERE task_id=%s", (task_id,)
)
conn.execute(
"""
UPDATE data_process_tasks
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)
FROM data_process_source_files
WHERE task_id=%s AND deleted_at IS NULL
), updated_at=%s
WHERE id=%s
""",
(20 if preview_count else 0, task_id, now, task_id),
)
except psycopg.errors.UniqueViolation as exc:
raise ConflictError(
"the same source file content is already attached to this task"
@@ -777,23 +824,39 @@ class DataProcessStore:
).fetchone()
if not row:
raise NotFoundError("source file not found")
conn.execute("DELETE FROM data_process_results WHERE task_id=%s", (task_id,))
conn.execute(
"DELETE FROM data_process_preview_items WHERE source_file_id=%s", (file_id,)
)
conn.execute(
"""
UPDATE data_process_tasks
SET status='pending', progress=0, output_count=0, filtered_count=0,
duplicate_count=0, error_count=0, failure_reason=NULL,
input_count=(SELECT COALESCE(SUM(record_count), 0)
FROM data_process_source_files
WHERE task_id=%s AND deleted_at IS NULL),
updated_at=%s
WHERE id=%s
""",
(task_id, utcnow(), task_id),
)
now = utcnow()
if _is_regeneration_prepared(task):
conn.execute(
"""
UPDATE data_process_tasks
SET input_count=(SELECT COALESCE(SUM(record_count), 0)
FROM data_process_source_files
WHERE task_id=%s AND deleted_at IS NULL),
updated_at=%s
WHERE id=%s
""",
(task_id, now, task_id),
)
else:
conn.execute(
"DELETE FROM data_process_results WHERE task_id=%s", (task_id,)
)
conn.execute(
"""
UPDATE data_process_tasks
SET status='pending', progress=0, output_count=0, filtered_count=0,
duplicate_count=0, error_count=0, failure_reason=NULL,
input_count=(SELECT COALESCE(SUM(record_count), 0)
FROM data_process_source_files
WHERE task_id=%s AND deleted_at IS NULL),
updated_at=%s
WHERE id=%s
""",
(task_id, now, task_id),
)
def replace_preview_items(
self,
@@ -823,7 +886,9 @@ class DataProcessStore:
with self.connect() as conn:
task = self._task_in_connection(conn, task_id, for_update=True)
self._ensure_editable(task)
conn.execute("DELETE FROM data_process_results WHERE task_id=%s", (task_id,))
regeneration_prepared = _is_regeneration_prepared(task)
if not regeneration_prepared:
conn.execute("DELETE FROM data_process_results WHERE task_id=%s", (task_id,))
if selected_ids is None:
conn.execute(
"DELETE FROM data_process_preview_items WHERE task_id=%s", (task_id,)
@@ -878,15 +943,21 @@ class DataProcessStore:
),
).fetchone()
created.append(_decode_row(row) or {})
conn.execute(
"""
UPDATE data_process_tasks
SET status='pending', progress=20, output_count=0, filtered_count=0,
duplicate_count=0, error_count=0, failure_reason=NULL, updated_at=%s
WHERE id=%s
""",
(now, task_id),
)
if regeneration_prepared:
conn.execute(
"UPDATE data_process_tasks SET updated_at=%s WHERE id=%s",
(now, task_id),
)
else:
conn.execute(
"""
UPDATE data_process_tasks
SET status='pending', progress=20, output_count=0, filtered_count=0,
duplicate_count=0, error_count=0, failure_reason=NULL, updated_at=%s
WHERE id=%s
""",
(now, task_id),
)
return created
def list_preview_items(
@@ -980,7 +1051,7 @@ class DataProcessStore:
now,
),
).fetchone()
self._invalidate_results(conn, task_id, now)
self._invalidate_results(conn, task, task_id, now)
return _decode_row(row) or {}
def update_preview_item(
@@ -1027,7 +1098,7 @@ class DataProcessStore:
task_id,
),
).fetchone()
self._invalidate_results(conn, task_id, now)
self._invalidate_results(conn, task, task_id, now)
return _decode_row(row) or {}
def delete_preview_item(self, task_id: str, preview_id: str) -> None:
@@ -1040,14 +1111,21 @@ class DataProcessStore:
).fetchone()
if not row:
raise NotFoundError("preview item not found")
self._invalidate_results(conn, task_id, utcnow())
self._invalidate_results(conn, task, task_id, utcnow())
def _invalidate_results(
self,
conn: psycopg.Connection[dict[str, Any]],
task: dict[str, Any],
task_id: str,
now: str,
) -> None:
if _is_regeneration_prepared(task):
conn.execute(
"UPDATE data_process_tasks SET updated_at=%s WHERE id=%s",
(now, task_id),
)
return
conn.execute("DELETE FROM data_process_results WHERE task_id=%s", (task_id,))
conn.execute(
"""
@@ -1065,7 +1143,8 @@ class DataProcessStore:
raise DataProcessStoreError("incremental generation is not supported")
with self.connect() as conn:
task = self._task_in_connection(conn, task_id, for_update=True)
if task.get("output_dataset_id"):
regeneration_prepared = _is_regeneration_prepared(task)
if task.get("output_dataset_id") and not regeneration_prepared:
raise InvalidStateError("published task cannot be regenerated")
if task["status"] == "running":
raise ConflictError("data process task is already running")
@@ -1078,16 +1157,19 @@ class DataProcessStore:
conn.execute("DELETE FROM data_process_results WHERE task_id=%s", (task_id,))
now = utcnow()
generation_run_id = new_id("dprun")
next_config = dict(task.get("config") or {})
next_config.pop(_REGENERATION_MARKER_KEY, None)
row = conn.execute(
"""
UPDATE data_process_tasks
SET status='running', progress=30, failure_reason=NULL, started_at=%s,
completed_at=NULL, output_count=0, filtered_count=0,
duplicate_count=0, error_count=0, generation_run_id=%s, updated_at=%s
SET config=%s, status='running', progress=30, failure_reason=NULL,
started_at=%s, completed_at=NULL, output_dataset_id=NULL,
output_count=0, filtered_count=0, duplicate_count=0, error_count=0,
generation_run_id=%s, updated_at=%s
WHERE id=%s
RETURNING *
""",
(now, generation_run_id, now, task_id),
(json_dumps(next_config), now, generation_run_id, now, task_id),
).fetchone()
return _decode_row(row) or {}
@@ -1405,6 +1487,10 @@ class DataProcessStore:
"""按精确配额发布训练、验证、测试三个独立数据集。"""
with self.connect() as conn:
task = self._task_in_connection(conn, task_id, for_update=True)
if _is_regeneration_prepared(task):
raise InvalidStateError(
"regeneration must start and complete before publishing"
)
if task["status"] != "completed":
raise InvalidStateError("only a completed task can be published")
rows = conn.execute(