feat: 数据集下载鉴权、MinIO 发布与数据库兼容增强

- 数据集下载接口鉴权：单文件直接返回、多文件打包 ZIP，前端统一走请求客户端携带 Token
- 预检/同步支持 MinIO 对象补建与资源副本记录，兼容接入 MinIO 前的历史数据集
- create_dataset 增加名称重复校验，软删记录释放名称并保留墓碑
- 数据处理发布结果支持写入 MinIO storage_objects 并关联 dataset_file
- 数据存储根目录支持 YG_FT_DATA_ROOT 环境变量
- SQL 迁移兼容旧版 approval_steps / retention_policies / data_process_results
- 训练日志图表按曲线独立过滤缺失采样点

Co-Authored-By: Claude <noreply@anthropic.com>

backend/app/api/v1/endpoints/platform.py

@@ -5,12 +5,15 @@ import asyncio
 import hashlib
 import uuid
 import time
+from io import BytesIO
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Any
+from urllib.parse import quote
+from zipfile import ZIP_DEFLATED, ZipFile
 
 from fastapi import APIRouter, BackgroundTasks, Body, Depends, File, HTTPException, Query, Request, UploadFile
-from fastapi.responses import PlainTextResponse, StreamingResponse
+from fastapi.responses import PlainTextResponse, Response, StreamingResponse
 
 import httpx
 
@@ -363,9 +366,16 @@ async def _fine_tune_preflight_payload(
     store: Any,
     payload: dict[str, Any],
     validate: bool = True,
+    sync_resources: bool = True,
 ) -> dict[str, Any]:
     node, job_payload = store.prepare_compute_job_payload_from_payload(payload)
-    return await _fine_tune_preflight_with_job_payload(node, job_payload, validate=validate, sync_resources=False, store=store)
+    return await _fine_tune_preflight_with_job_payload(
+        node,
+        job_payload,
+        validate=validate,
+        sync_resources=sync_resources,
+        store=store,
+    )
 
 
 async def _fine_tune_preflight_with_job_payload(
@@ -1132,13 +1142,42 @@ async def _sync_training_dataset_to_compute_node(
 ) -> list[dict[str, Any]]:
     if get_settings().minio_enabled:
         files = store.training_dataset_files(dataset_id)
-        objects = store.storage_objects_for_resource("dataset", dataset_id)
-        object_by_name = {Path(str(item.get("file_name") or item.get("object_key") or "")).name: item for item in objects}
+        object_by_resource_name: dict[tuple[str, str], dict[str, Any]] = {}
+        resource_ids = {str(dataset_id)} | {
+            str(item.get("dataset_id"))
+            for item in files
+            if item.get("dataset_id")
+        }
+        for resource_id in resource_ids:
+            for obj in store.storage_objects_for_resource("dataset", resource_id):
+                file_name = Path(str(obj.get("file_name") or obj.get("object_key") or "")).name
+                object_by_resource_name[(resource_id, file_name)] = obj
         results: list[dict[str, Any]] = []
         client = ComputeNodeClient(node["api_base_url"])
         for item in files:
             target_name = Path(str(item.get("name") or f"{item['id']}.jsonl")).name
-            obj = object_by_name.get(target_name)
+            item_dataset_id = str(item.get("dataset_id") or dataset_id)
+            obj = object_by_resource_name.get((item_dataset_id, target_name))
+            if not obj and item.get("content"):
+                # 兼容 MinIO 接入前已经发布的数据处理数据集：
+                # 预检时用数据库正文补建对象，避免要求用户重新处理数据集。
+                raw = str(item.get("content") or "").encode("utf-8")
+                version_id = str(item.get("active_version_id") or item["id"])
+                object_key = f"datasets/{item_dataset_id}/versions/{version_id}/{target_name}"
+                uploaded = get_object_storage().put_bytes(object_key, raw, "application/jsonl")
+                obj = store.create_storage_object({
+                    "resource_type": "dataset",
+                    "resource_id": item_dataset_id,
+                    "version_id": version_id,
+                    "bucket": uploaded["bucket"],
+                    "object_key": object_key,
+                    "file_name": target_name,
+                    "content_type": "application/jsonl",
+                    "byte_size": len(raw),
+                    "checksum_sha256": hashlib.sha256(raw).hexdigest(),
+                    "status": "available",
+                })
+                store.link_dataset_file_storage_object(str(item["id"]), obj["id"])
             if not obj:
                 raise RuntimeError(f"dataset file is not available in MinIO: {target_name}")
             url = get_object_storage().presigned_get(obj["object_key"])
@@ -1150,6 +1189,12 @@ async def _sync_training_dataset_to_compute_node(
                 "byte_size": obj.get("byte_size") or 0,
                 "relative_path": f"datasets/{dataset_id}/{target_name}",
             })
+            store.upsert_resource_replica(
+                node["id"],
+                "dataset",
+                dataset_id,
+                str(result.get("local_path") or ""),
+            )
             results.append({**result, "file_id": item.get("id"), "name": target_name, "node_id": node["id"]})
         return results
     if not dataset_id:
@@ -1218,7 +1263,7 @@ async def upload_dataset_files(
             if get_settings().minio_enabled:
                 object_key = f"datasets/{dataset_id}/versions/{created_file.get('active_version_id') or created_file['id']}/{Path(created_file['name']).name}"
                 uploaded = get_object_storage().put_bytes(object_key, raw, file.content_type or "application/octet-stream")
-                get_platform_store().create_storage_object({
+                storage_object = get_platform_store().create_storage_object({
                     "resource_type": "dataset", "resource_id": dataset_id,
                     "version_id": created_file.get("active_version_id") or created_file["id"],
                     "bucket": uploaded["bucket"], "object_key": object_key,
@@ -1226,6 +1271,7 @@ async def upload_dataset_files(
                     "byte_size": len(raw), "checksum_sha256": hashlib.sha256(raw).hexdigest(),
                     "status": "available",
                 })
+                store.link_dataset_file_storage_object(created_file["id"], storage_object["id"])
     if sync_to_compute:
         for file_id, file_name, raw in pending_sync:
             compute_sync.extend(
@@ -1241,10 +1287,60 @@ async def upload_dataset_files(
 
 
 @router.get("/dataset-manage/download/{dataset_id}")
-async def download_dataset(dataset_id: str) -> PlainTextResponse:
-    dataset = get_platform_store().dataset(dataset_id)
-    content = "\n".join([f"{file['name']}" for file in dataset.get("files", [])])
-    return PlainTextResponse(content, media_type="text/plain")
+async def download_dataset(dataset_id: str, current_user: dict = Depends(get_current_user)) -> Response:
+    store = get_platform_store()
+    try:
+        dataset = store.dataset(dataset_id)
+    except KeyError:
+        raise fail(404, "dataset not found")
+    if not has_resource_access("dataset", dataset_id, current_user, "read"):
+        raise fail(403, "no permission to access this dataset")
+
+    files = []
+    for item in dataset.get("files", []):
+        if item.get("deleted_at"):
+            continue
+        try:
+            full_file = store.dataset_file(str(item["id"]))
+        except KeyError:
+            continue
+        files.append({**item, "content": full_file.get("content") or ""})
+    if not files:
+        raise fail(404, "dataset has no downloadable files")
+
+    if len(files) == 1:
+        item = files[0]
+        filename = Path(str(item.get("name") or f"{dataset_id}.jsonl")).name
+        encoded_name = quote(filename, safe="")
+        return Response(
+            content=str(item.get("content") or "").encode("utf-8"),
+            media_type="application/octet-stream",
+            headers={
+                "Content-Disposition": f"attachment; filename=dataset-file; filename*=UTF-8''{encoded_name}",
+            },
+        )
+
+    archive = BytesIO()
+    with ZipFile(archive, "w", compression=ZIP_DEFLATED) as bundle:
+        used_names: set[str] = set()
+        for index, item in enumerate(files, start=1):
+            filename = Path(str(item.get("name") or f"file-{index}.jsonl")).name
+            unique_name = filename
+            if unique_name in used_names:
+                stem = Path(filename).stem
+                suffix = Path(filename).suffix
+                unique_name = f"{stem}-{index}{suffix}"
+            used_names.add(unique_name)
+            bundle.writestr(unique_name, str(item.get("content") or ""))
+    archive.seek(0)
+    archive_name = quote(f"{dataset.get('name') or dataset_id}.zip", safe="")
+    return StreamingResponse(
+        archive,
+        media_type="application/zip",
+        headers={
+            "Content-Disposition": f"attachment; filename=dataset.zip; filename*=UTF-8''{archive_name}",
+        },
+    )
 
 
 @router.get("/dataset-manage/download/{dataset_id}/{file_id}")
@@ -1268,8 +1364,11 @@ async def dataset_list(current_user: dict = Depends(get_current_user)) -> dict[s
 @router.post("/dataset-manage")
 async def create_dataset(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
     payload.setdefault("created_by", current_user.get("id"))
-    dataset = get_platform_store().create_dataset(payload)
-    return ok({"id": dataset["id"]})
+    try:
+        dataset = get_platform_store().create_dataset(payload)
+        return ok({"id": dataset["id"]})
+    except ValueError as exc:
+        raise fail(409, str(exc))
 
 
 @router.get("/dataset-manage/{dataset_id}")
@@ -1386,7 +1485,7 @@ async def start_fine_tune(
 @router.post("/fine-tune/preflight")
 async def fine_tune_create_preflight(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
     try:
-        return ok(await _fine_tune_preflight_payload(get_platform_store(), payload, validate=True))
+        return ok(await _fine_tune_preflight_payload(get_platform_store(), payload, validate=True, sync_resources=True))
     except RuntimeError as exc:
         return ok({"valid": False, "errors": [str(exc)], "warnings": [], "diagnostics": _training_diagnostics([str(exc)])})
     except Exception as exc:  # noqa: BLE001 - expose compute validation errors to training create page
@@ -1396,7 +1495,7 @@ async def fine_tune_create_preflight(payload: dict[str, Any] = Body(...)) -> dic
 @router.post("/fine-tune/command-preview")
 async def fine_tune_create_command_preview(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
     try:
-        return ok(await _fine_tune_preflight_payload(get_platform_store(), payload, validate=False))
+        return ok(await _fine_tune_preflight_payload(get_platform_store(), payload, validate=False, sync_resources=False))
     except RuntimeError as exc:
         return ok({"valid": False, "errors": [str(exc)], "warnings": [], "diagnostics": _training_diagnostics([str(exc)])})
     except Exception as exc:  # noqa: BLE001

backend/app/db/platform_store.py

@@ -1607,7 +1607,24 @@ class PlatformStore:
 
     def create_dataset(self, payload: dict[str, Any]) -> dict[str, Any]:
         dataset_id = payload.get("id") or new_id("ds")
+        name = str(payload.get("name") or "").strip()
+        if not name:
+            raise ValueError("dataset name is required")
         with self.connect() as conn:
+            existing = conn.execute(
+                "SELECT id, deleted_at FROM datasets WHERE name=?",
+                (name,),
+            ).fetchone()
+            if existing and not existing.get("deleted_at"):
+                raise ValueError(f"dataset name already exists: {name}")
+            # Soft-deleted records remain in the database for audit/history and
+            # still participate in the legacy unique constraint. Free the name
+            # while retaining a traceable tombstone before creating the new row.
+            if existing:
+                conn.execute(
+                    "UPDATE datasets SET name=? WHERE id=?",
+                    (f"{name}__deleted__{existing['id']}", existing["id"]),
+                )
             conn.execute(
                 """
                 INSERT INTO datasets
@@ -1616,7 +1633,7 @@ class PlatformStore:
                 """,
                 (
                     dataset_id,
-                    payload["name"],
+                    name,
                     payload.get("type", "train"),
                     payload.get("storage_type", "local"),
                     payload.get("source", "upload"),
@@ -2965,6 +2982,27 @@ class PlatformStore:
             ).fetchone()
             return dict(row)
 
+    def link_dataset_file_storage_object(self, file_id: str, storage_object_id: str) -> None:
+        """Link an uploaded dataset file to its canonical MinIO object."""
+        with self.connect() as conn:
+            conn.execute(
+                """
+                UPDATE dataset_files
+                SET storage_object_id=?
+                WHERE id=?
+                """,
+                (storage_object_id, file_id),
+            )
+            row = conn.execute(
+                "SELECT dataset_id FROM dataset_files WHERE id=?",
+                (file_id,),
+            ).fetchone()
+            if row:
+                conn.execute(
+                    "UPDATE datasets SET storage_type='minio' WHERE id=?",
+                    (row["dataset_id"],),
+                )
+
     def storage_objects_for_resource(self, resource_type: str, resource_id: str) -> list[dict[str, Any]]:
         with self.connect() as conn:
             rows = conn.execute(

backend/app/db/sql/000_full_init.sql

@@ -506,6 +506,13 @@ CREATE INDEX IF NOT EXISTS idx_approval_instances_applicant ON approval_instance
 CREATE INDEX IF NOT EXISTS idx_approval_instances_resource ON approval_instances(resource_type, resource_id, status);
 CREATE INDEX IF NOT EXISTS idx_approval_steps_approver ON approval_steps(approver_id, status);
 
+-- Compatibility for databases created from an older approval_steps definition.
+ALTER TABLE approval_steps ADD COLUMN IF NOT EXISTS id TEXT;
+UPDATE approval_steps
+SET id = 'astep_' || md5(concat_ws(':', instance_id, step_index, coalesce(approver_id, ''), coalesce(time, '')))
+WHERE id IS NULL;
+CREATE UNIQUE INDEX IF NOT EXISTS idx_approval_steps_id ON approval_steps(id);
+
 CREATE TABLE IF NOT EXISTS audit_logs (
     id TEXT PRIMARY KEY,
     tenant_id TEXT,
@@ -535,6 +542,38 @@ CREATE TABLE IF NOT EXISTS retention_policies (
     updated_at TEXT
 );
 
+-- Compatibility for the earlier retention policy schema
+-- (resource_type/retention_days). Keep legacy columns if they exist.
+ALTER TABLE retention_policies ADD COLUMN IF NOT EXISTS scope TEXT;
+ALTER TABLE retention_policies ADD COLUMN IF NOT EXISTS rule TEXT;
+ALTER TABLE retention_policies ADD COLUMN IF NOT EXISTS status TEXT DEFAULT 'active';
+ALTER TABLE retention_policies ADD COLUMN IF NOT EXISTS create_by TEXT;
+ALTER TABLE retention_policies ADD COLUMN IF NOT EXISTS updated_at TEXT;
+DO $$
+BEGIN
+    IF EXISTS (
+        SELECT 1 FROM information_schema.columns
+        WHERE table_schema='public' AND table_name='retention_policies' AND column_name='resource_type'
+    ) AND EXISTS (
+        SELECT 1 FROM information_schema.columns
+        WHERE table_schema='public' AND table_name='retention_policies' AND column_name='retention_days'
+    ) THEN
+        EXECUTE $migration$
+            UPDATE retention_policies
+            SET scope = COALESCE(scope, resource_type),
+                rule = COALESCE(rule, retention_days::text),
+                status = COALESCE(status, 'active'),
+                updated_at = COALESCE(updated_at, create_time)
+            WHERE scope IS NULL OR rule IS NULL OR status IS NULL OR updated_at IS NULL
+        $migration$;
+    ELSE
+        UPDATE retention_policies
+        SET status = COALESCE(status, 'active'),
+            updated_at = COALESCE(updated_at, create_time)
+        WHERE status IS NULL OR updated_at IS NULL;
+    END IF;
+END $$;
+
 -- ============================================================================
 -- 三、租户配额扩展（来源：003_tenant_quota.sql）
 -- ============================================================================
@@ -723,9 +762,13 @@ CREATE TABLE IF NOT EXISTS data_process_results (
     instruction TEXT NOT NULL,
     input TEXT NOT NULL DEFAULT '',
     output TEXT NOT NULL,
+    chosen TEXT NOT NULL DEFAULT '',
+    rejected TEXT NOT NULL DEFAULT '',
     original_instruction TEXT,
     original_input TEXT,
     original_output TEXT,
+    original_chosen TEXT,
+    original_rejected TEXT,
     status VARCHAR(20) NOT NULL DEFAULT 'valid'
         CHECK (status IN ('valid', 'modified', 'invalid')),
     error TEXT,
@@ -735,6 +778,12 @@ CREATE TABLE IF NOT EXISTS data_process_results (
     updated_at TIMESTAMPTZ NOT NULL DEFAULT now()
 );
 
+-- Keep existing databases compatible with the current generation result model.
+ALTER TABLE data_process_results ADD COLUMN IF NOT EXISTS chosen TEXT NOT NULL DEFAULT '';
+ALTER TABLE data_process_results ADD COLUMN IF NOT EXISTS rejected TEXT NOT NULL DEFAULT '';
+ALTER TABLE data_process_results ADD COLUMN IF NOT EXISTS original_chosen TEXT;
+ALTER TABLE data_process_results ADD COLUMN IF NOT EXISTS original_rejected TEXT;
+
 CREATE INDEX IF NOT EXISTS idx_data_process_results_task_status
     ON data_process_results(task_id, status, id);
 CREATE INDEX IF NOT EXISTS idx_data_process_results_task_split

backend/app/modules/data_process/storage.py

@@ -28,6 +28,9 @@ class StagedSourceObject:
 
 
 def _default_storage_root() -> Path:
+    data_root = os.getenv("YG_FT_DATA_ROOT", "").strip()
+    if data_root:
+        return Path(data_root).expanduser() / "data-process"
     return Path(__file__).resolve().parents[3] / "storage" / "data-process"
 
 

backend/app/modules/data_process/store/datasets.py

@@ -9,6 +9,9 @@ import hashlib
 
 import psycopg
 
+from app.core.config import get_settings
+from app.modules.storage.minio_store import get_object_storage
+
 from .base import (
     StoreBase,
     utcnow,
@@ -165,6 +168,7 @@ class DatasetsMixin:
                 split_name: assignments.count(split_name) for split_name in split_order
             }
             split_specs: list[dict[str, Any]] = []
+            use_minio = bool(get_settings().minio_enabled)
             for split_name in split_order:
                 split_records = [
                     (source_row, record)
@@ -194,7 +198,7 @@ class DatasetsMixin:
             source_result_ids = [row["id"] for row in rows]
             common_metadata = {
                 "source": "data_process",
-                "storage_backend": "database",
+                "storage_backend": "minio" if use_minio else "database",
                 "source_task_id": task_id,
                 "output_type": _task_output_type(task),
                 "reasoning_detail": _task_reasoning_detail(task),
@@ -268,6 +272,53 @@ class DatasetsMixin:
                 for spec in split_specs:
                     split_name = str(spec["split"])
                     dataset_id = dataset_ids[split_name]
+                    storage_object_id = str(spec["storage_object_id"])
+                    if use_minio:
+                        file_name = f"{base_dataset_name}.{split_name}.jsonl"
+                        object_key = f"datasets/{dataset_id}/versions/{spec['version_id']}/{file_name}"
+                        uploaded = get_object_storage().put_bytes(
+                            object_key,
+                            spec["raw"],
+                            "application/jsonl",
+                        )
+                        conn.execute(
+                            """
+                            INSERT INTO storage_objects
+                            (id, resource_type, resource_id, version_id, bucket, object_key,
+                             file_name, content_type, checksum_sha256, byte_size, status,
+                             created_by, create_time)
+                            VALUES (%s, 'dataset', %s, %s, %s, %s, %s, %s, %s, %s,
+                                    'available', %s, %s)
+                            ON CONFLICT (resource_type, resource_id, version_id, object_key)
+                            DO UPDATE SET file_name=EXCLUDED.file_name,
+                                content_type=EXCLUDED.content_type,
+                                checksum_sha256=EXCLUDED.checksum_sha256,
+                                byte_size=EXCLUDED.byte_size,
+                                status='available',
+                                created_by=EXCLUDED.created_by
+                            """,
+                            (
+                                new_id("object"),
+                                dataset_id,
+                                spec["version_id"],
+                                uploaded["bucket"],
+                                object_key,
+                                file_name,
+                                "application/jsonl",
+                                spec["checksum"],
+                                len(spec["raw"]),
+                                payload.get("created_by") or task.get("created_by"),
+                                now,
+                            ),
+                        )
+                        storage_object_id = conn.execute(
+                            """
+                            SELECT id FROM storage_objects
+                            WHERE resource_type='dataset' AND resource_id=%s
+                              AND version_id=%s AND object_key=%s
+                            """,
+                            (dataset_id, spec["version_id"], object_key),
+                        ).fetchone()["id"]
                     existing_dataset = existing_by_split.get(split_name)
                     dataset_metadata = {
                         **common_metadata,
@@ -305,7 +356,7 @@ class DatasetsMixin:
                             (
                                 dataset_name,
                                 dataset_types[split_name],
-                                payload.get("storage_type") or "local",
+                                "minio" if use_minio else (payload.get("storage_type") or "local"),
                                 f"{len(spec['raw'])} B",
                                 len(spec["raw"]),
                                 len(spec["records"]),
@@ -335,7 +386,7 @@ class DatasetsMixin:
                                 dataset_id,
                                 dataset_name,
                                 dataset_types[split_name],
-                                payload.get("storage_type") or "local",
+                                "minio" if use_minio else (payload.get("storage_type") or "local"),
                                 task_id,
                                 task_id,
                                 f"{len(spec['raw'])} B",
@@ -366,7 +417,7 @@ class DatasetsMixin:
                         "created_at": now,
                         "create_time": now,
                         "source_task_id": task_id,
-                        "storage_object_id": spec["storage_object_id"],
+                        "storage_object_id": storage_object_id,
                     }
                     conn.execute(
                         """
@@ -383,7 +434,7 @@ class DatasetsMixin:
                             spec["file_id"],
                             dataset_id,
                             f"{base_dataset_name}.{split_name}.jsonl",
-                            spec["storage_object_id"],
+                            storage_object_id,
                             f"{len(spec['raw'])} B",
                             spec["content"],
                             spec["version_id"],
@@ -414,7 +465,7 @@ class DatasetsMixin:
                         (
                             spec["version_id"],
                             spec["file_id"],
-                            spec["storage_object_id"],
+                            storage_object_id,
                             spec["content"][:2000],
                             f"data process {split_name} publish",
                             len(spec["raw"]),