diff --git a/backend/app/api/v1/endpoints/data_process.py b/backend/app/api/v1/endpoints/data_process.py index 2a23cde..8c68518 100644 --- a/backend/app/api/v1/endpoints/data_process.py +++ b/backend/app/api/v1/endpoints/data_process.py @@ -35,6 +35,8 @@ from fastapi.responses import StreamingResponse from psycopg.rows import dict_row from app.core.auth import filter_accessible_resource_ids, get_current_user, is_admin +from app.core.config import get_settings +from app.db.platform_store import get_platform_store from app.modules.data_process.algorithms import ( ParsedText, canonical_record_json, @@ -85,6 +87,8 @@ from app.modules.data_process.store import ( new_id, repeat_task_id, ) +from app.modules.storage.minio_store import get_object_storage +from app.modules.storage.policy import should_store_in_minio from app.schemas.data_process import ( DataProcessRegenerateRequest, DataProcessRepeatRequest, @@ -224,9 +228,38 @@ def _commit_source_batch( staged: list[StagedSourceObject], ) -> list[dict[str, Any]]: storage.publish(staged) + storage_object_ids: list[str] = [] try: + # The source reference remains in the task schema for compatibility, + # while storage_objects provides the authoritative MinIO index. + if get_settings().minio_enabled: + for item in prepared: + reference = str(item.get("storage_object_id") or "") + if not reference.startswith("minio://"): + continue + object_key = storage.object_key(reference) + metadata = get_object_storage().stat(object_key) + object_row = get_platform_store().create_storage_object({ + "resource_type": "data_process_source", + "resource_id": task_id, + "version_id": str(item.get("id") or new_id("dpsf")), + "bucket": get_object_storage().bucket, + "object_key": object_key, + "file_name": item.get("name"), + "content_type": (item.get("metadata") or {}).get("content_type", "application/octet-stream"), + "byte_size": metadata.get("byte_size") or item.get("raw_size") or 0, + "checksum_sha256": item.get("checksum_sha256"), + "status": "available", + "created_by": item.get("created_by"), + }) + storage_object_ids.append(str(object_row["id"])) return store.add_source_files(task_id, prepared) except Exception: + for object_id in storage_object_ids: + try: + get_platform_store().update_storage_object(object_id, {"status": "deleted"}) + except Exception: + pass for item in staged: try: storage.delete(item.reference) @@ -774,6 +807,25 @@ def _run_generation( duplicate_count=duplicate_count, error_count=error_count, ) + result_bytes = "".join( + structured_json_dumps(item) + "\n" for item in accepted + ).encode("utf-8") + if should_store_in_minio(len(result_bytes)): + result_key = f"data-process/{task_id}/results/{generation_run_id}.jsonl" + uploaded = get_object_storage().put_bytes(result_key, result_bytes, "application/jsonl") + get_platform_store().create_storage_object({ + "resource_type": "data_process_result", + "resource_id": task_id, + "version_id": generation_run_id, + "bucket": uploaded["bucket"], + "object_key": result_key, + "file_name": f"{generation_run_id}.jsonl", + "content_type": "application/jsonl", + "byte_size": len(result_bytes), + "checksum_sha256": hashlib.sha256(result_bytes).hexdigest(), + "status": "available", + "created_by": (store.get_task(task_id) or {}).get("created_by"), + }) logger.info( "data process generation completed task_id=%s generation_run_id=%s " "output_count=%s filtered_count=%s duplicate_count=%s error_count=%s " @@ -933,7 +985,7 @@ def _repeat_file_copies( source = store.get_source_file(source_task_id, old_file_id, include_content=True) new_file_id = new_id("dpsf") old_reference = str(source.get("storage_object_id") or "") - if old_reference.startswith("local://data-process/"): + if old_reference.startswith(("local://data-process/", "minio://data-process/")): staged_object = storage.stage_copy( batch_id=batch_id, source_reference=old_reference, diff --git a/backend/app/api/v1/endpoints/platform.py b/backend/app/api/v1/endpoints/platform.py index 7b9ac29..f3387b3 100644 --- a/backend/app/api/v1/endpoints/platform.py +++ b/backend/app/api/v1/endpoints/platform.py @@ -23,8 +23,9 @@ from app.core.audit import audit_log, AuditActions from app.core.op_log import op_log, OpModule, OpAction from app.db.platform_store import get_platform_store from app.modules.compute_gateway.client import ComputeNodeClient -from app.modules.compute_gateway.sync import fetch_eval_result_content, poll_compute_jobs_once +from app.modules.compute_gateway.sync import _archive_node_directory, fetch_eval_result_content, poll_compute_jobs_once from app.modules.storage.minio_store import ObjectStorageError, get_object_storage +from app.modules.storage.policy import should_store_in_minio router = APIRouter() _LOGIN_FAILURES: dict[str, list[float]] = {} @@ -100,6 +101,106 @@ async def _wait_for_object_storage() -> None: await asyncio.sleep(max(1, settings.storage_check_interval_seconds)) +def _store_json_snapshot( + store: Any, + resource_type: str, + resource_id: str, + version_id: str, + object_key: str, + payload: dict[str, Any], + created_by: str | None = None, +) -> dict[str, Any]: + """Persist non-secret task/model parameters as an auditable MinIO snapshot.""" + sanitized = { + key: value + for key, value in payload.items() + if key not in {"api_key", "secret_key", "password", "token", "access_token"} + } + raw = json.dumps(sanitized, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8") + # Task/evaluation payloads are already persisted in PostgreSQL. Avoid an + # extra MinIO round trip for small non-secret parameter snapshots. + if not should_store_in_minio(len(raw), content_type="application/json", file_format="json"): + return {} + uploaded = get_object_storage().put_bytes(object_key, raw, "application/json") + return store.create_storage_object({ + "resource_type": resource_type, + "resource_id": resource_id, + "version_id": version_id, + "bucket": uploaded["bucket"], + "object_key": object_key, + "file_name": Path(object_key).name, + "content_type": "application/json", + "byte_size": len(raw), + "checksum_sha256": hashlib.sha256(raw).hexdigest(), + "status": "available", + "created_by": created_by, + }) + + +def _dataset_file_bytes(store: Any, file_id: str) -> bytes: + """Return the canonical dataset bytes, lazily indexing legacy DB content.""" + row = store.dataset_file(file_id) + if not get_settings().minio_enabled: + return str(row.get("content") or "").encode("utf-8") + + storage_object = None + object_id = str(row.get("storage_object_id") or "") + if object_id: + try: + storage_object = store.storage_object(object_id) + except KeyError: + storage_object = None + if storage_object and storage_object.get("status") == "available": + try: + return get_object_storage().get_bytes(storage_object["object_key"]) + except Exception as exc: # noqa: BLE001 - expose storage outage to callers + raise RuntimeError(f"dataset object is unavailable in MinIO: {exc}") from exc + + # Compatibility migration for files created before MinIO was enabled. + raw = str(row.get("content") or "").encode("utf-8") + if not raw: + raise RuntimeError(f"dataset file has no MinIO object or legacy content: {file_id}") + # Small files intentionally remain database-backed. They can still be + # copied to a compute node directly when a task needs them. + if not should_store_in_minio(len(raw), file_format=row.get("file_format")): + return raw + object_key = ( + f"datasets/{row['dataset_id']}/versions/" + f"{row.get('active_version_id') or row['id']}/{Path(str(row.get('name') or row['id'])).name}" + ) + uploaded = get_object_storage().put_bytes(object_key, raw, "application/octet-stream") + created = store.create_storage_object({ + "resource_type": "dataset", + "resource_id": str(row["dataset_id"]), + "version_id": str(row.get("active_version_id") or row["id"]), + "bucket": uploaded["bucket"], + "object_key": object_key, + "file_name": row.get("name"), + "content_type": "application/octet-stream", + "byte_size": len(raw), + "checksum_sha256": hashlib.sha256(raw).hexdigest(), + "status": "available", + }) + store.link_dataset_file_storage_object(str(row["id"]), created["id"]) + return raw + + +def _dataset_version_bytes(store: Any, file_id: str, version_id: str) -> bytes: + row = store.dataset_file(file_id) + try: + version = next(item for item in store.file_versions(file_id)["versions"] if item["id"] == version_id) + except StopIteration as exc: + raise KeyError(version_id) from exc + object_id = str(version.get("storage_object_id") or "") + if get_settings().minio_enabled and object_id: + try: + obj = store.storage_object(object_id) + return get_object_storage().get_bytes(obj["object_key"]) + except KeyError: + pass + return _dataset_file_bytes(store, file_id) + + def _select_eval_node(store: Any, preferred_node_id: str | None = None) -> dict[str, Any] | None: """Select the compute node for an eval job. @@ -130,18 +231,63 @@ async def _prepare_resource_on_node(store: Any, resource_type: str, resource_id: if not get_settings().minio_enabled or not resource_id: return None objects = store.storage_objects_for_resource(resource_type, resource_id) + if not objects and resource_type in {"model", "trained_model"}: + resource = None + if resource_type == "model": + try: + resource = store.model(resource_id) + except KeyError: + try: + resource = store.model_by_name(resource_id) + except KeyError: + resource = next((item for item in store.models() if item.get("path") == resource_id), None) + else: + resource = next( + (item for item in store.trained_models() if item.get("id") == resource_id or item.get("name") == resource_id), + None, + ) + source_path = str( + (resource or {}).get("path") + or (resource or {}).get("merged_path") + or (resource or {}).get("artifact_dir") + or "" + ) + resolved_id = str((resource or {}).get("id") or resource_id) + if source_path and resolved_id: + client = ComputeNodeClient(node["api_base_url"], timeout=900) + await _archive_node_directory( + store, + client, + node, + source_path, + resource_type, + resolved_id, + "legacy-import", + f"models/{resolved_id}" if resource_type == "model" else f"trained_models/{resolved_id}", + ) + resource_id = resolved_id + objects = store.storage_objects_for_resource(resource_type, resource_id) if not objects: return None client = ComputeNodeClient(node["api_base_url"], timeout=900) root_name = "trained_models" if resource_type in {"trained_model", "model_artifact"} else f"{resource_type}s" for obj in objects: + object_key = str(obj["object_key"]) + marker = f"{root_name}/{resource_id}/versions/" + relative_name = Path(str(obj.get("file_name") or object_key)).name + if marker in object_key: + suffix = object_key.split(marker, 1)[1] + if "/" in suffix: + suffix = suffix.split("/", 1)[1] + if suffix: + relative_name = suffix await client.prepare_cache({ "resource_id": resource_id, "version_id": obj["version_id"], - "download_url": get_object_storage().presigned_get(obj["object_key"]), + "download_url": get_object_storage().presigned_get(object_key), "checksum_sha256": obj.get("checksum_sha256") or "", "byte_size": obj.get("byte_size") or 0, - "relative_path": f"{root_name}/{resource_id}/{Path(str(obj.get('file_name') or obj['object_key'])).name}", + "relative_path": f"{root_name}/{resource_id}/{relative_name}", }) return f"/data/yg-ft/{root_name}/{resource_id}" @@ -397,8 +543,23 @@ async def _submit_fine_tune_task(store: Any, payload: dict[str, Any]) -> dict[st if not preflight["valid"]: errors = "; ".join(preflight.get("errors") or ["preflight failed"]) raise RuntimeError(f"preflight failed: {errors}") - payload = {**payload, "compute_node_id": preflight["node"]["id"]} + prepared_job_payload = preflight.get("job_payload") or {} + payload = { + **payload, + "compute_node_id": preflight["node"]["id"], + "prepared_base_model_path": prepared_job_payload.get("model_name_or_path") or payload.get("prepared_base_model_path"), + } task = store.start_task(payload) + if get_settings().minio_enabled: + _store_json_snapshot( + store, + "fine_tune", + str(task["id"]), + str(task["id"]), + f"training/{task['id']}/versions/{task['id']}/training-config.json", + task, + task.get("created_by"), + ) if get_settings().compute_mode == "simulator": return task node, job_payload = store.build_compute_job_payload(task["id"]) @@ -454,6 +615,20 @@ async def _fine_tune_preflight_with_job_payload( if get_settings().minio_enabled and get_settings().compute_mode != "simulator": try: await _wait_for_object_storage() + base_model_path = str(job_payload.get("model_name_or_path") or job_payload.get("base_model") or "") + base_model_id = str(job_payload.get("base_model_id") or job_payload.get("model_id") or "") + base_model = None + if base_model_id: + try: + base_model = store.model(base_model_id) + except KeyError: + base_model = None + if base_model is None: + base_model = next((item for item in store.models() if item.get("path") == base_model_path), None) + if base_model: + prepared_model = await _prepare_resource_on_node(store, "model", str(base_model["id"]), node) + if prepared_model: + job_payload = {**job_payload, "base_model": prepared_model, "model_name_or_path": prepared_model} except Exception as exc: # noqa: BLE001 - preflight exposes node storage failure sync_errors.append(f"shared storage health check failed: {exc}") if get_settings().compute_mode == "simulator": @@ -1091,8 +1266,9 @@ async def merge_model(payload: dict[str, Any] = Body(...), current_user: dict = @router.get("/dataset-manage/preview/{file_id}") async def dataset_preview(file_id: str) -> dict[str, Any]: try: - row = get_platform_store().dataset_file(file_id) - return ok({"content": row["content"]}) + store = get_platform_store() + content = _dataset_file_bytes(store, file_id).decode("utf-8", errors="replace") + return ok({"content": content}) except KeyError: raise fail(404, "dataset file not found") @@ -1121,7 +1297,8 @@ async def dataset_version_content(file_id: str, version_id: str) -> dict[str, An version = next((item for item in versions if item["id"] == version_id), None) if not version: raise KeyError(version_id) - return ok({"version": version, "content": row["content"]}) + content = _dataset_version_bytes(get_platform_store(), file_id, version_id) + return ok({"version": version, "content": content.decode("utf-8", errors="replace")}) except KeyError: raise fail(404, "dataset version not found") @@ -1210,7 +1387,7 @@ async def _sync_training_dataset_to_compute_node( ) -> list[dict[str, Any]]: if get_settings().minio_enabled: files = store.training_dataset_files(dataset_id) - object_by_resource_name: dict[tuple[str, str], dict[str, Any]] = {} + object_by_resource_name: dict[tuple[str, str, str], dict[str, Any]] = {} resource_ids = {str(dataset_id)} | { str(item.get("dataset_id")) for item in files @@ -1219,35 +1396,58 @@ async def _sync_training_dataset_to_compute_node( 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 + object_by_resource_name[(resource_id, file_name, str(obj.get("version_id") or ""))] = 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 item_dataset_id = str(item.get("dataset_id") or dataset_id) - obj = object_by_resource_name.get((item_dataset_id, target_name)) + version_id = str(item.get("active_version_id") or item["id"]) + obj = object_by_resource_name.get((item_dataset_id, target_name, version_id)) if not obj and item.get("content"): - # 兼容 MinIO 接入前已经发布的数据处理数据集： - # 预检时用数据库正文补建对象，避免要求用户重新处理数据集。 + # 兼容 MinIO 接入前已经发布的数据处理数据集。大文件补建 + # 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 should_store_in_minio(len(raw)): + 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"]) + else: + result = await client.upload_file( + target_name, + raw, + f"datasets/{dataset_id}/{target_name}", + resource_type="dataset", + resource_id=dataset_id, + ) + store.upsert_resource_replica( + node["id"], "dataset", dataset_id, str(result.get("local_path") or "") + ) + results.append({ + "node_id": node["id"], + "node_code": node.get("code"), + "file_id": item.get("id"), + "name": target_name, + "local_path": result.get("local_path"), + "byte_size": result.get("byte_size"), + "checksum_sha256": result.get("checksum_sha256"), + "storage_backend": "database", + }) + continue if not obj: - raise RuntimeError(f"dataset file is not available in MinIO: {target_name}") + raise RuntimeError(f"dataset file is not available: {target_name}") url = get_object_storage().presigned_get(obj["object_key"]) result = await client.prepare_cache({ "resource_id": dataset_id, @@ -1263,7 +1463,13 @@ async def _sync_training_dataset_to_compute_node( dataset_id, str(result.get("local_path") or ""), ) - results.append({**result, "file_id": item.get("id"), "name": target_name, "node_id": node["id"]}) + results.append({ + **result, + "file_id": item.get("id"), + "name": target_name, + "node_id": node["id"], + "storage_backend": "minio", + }) return results if not dataset_id: raise RuntimeError("train_dataset_id is required") @@ -1328,7 +1534,11 @@ async def upload_dataset_files( created_file = store.add_dataset_file(conn, dataset_id, file.filename or "upload.jsonl", content) created.append(created_file) pending_sync.append((created_file["id"], created_file["name"], raw)) - if get_settings().minio_enabled: + if should_store_in_minio( + len(raw), + content_type=file.content_type, + file_format=Path(created_file["name"]).suffix, + ): 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") storage_object = get_platform_store().create_storage_object({ @@ -1372,7 +1582,10 @@ async def download_dataset(dataset_id: str, current_user: dict = Depends(get_cur full_file = store.dataset_file(str(item["id"])) except KeyError: continue - files.append({**item, "content": full_file.get("content") or ""}) + files.append({ + **item, + "content": _dataset_file_bytes(store, str(item["id"])).decode("utf-8", errors="replace"), + }) if not files: raise fail(404, "dataset has no downloadable files") @@ -1413,8 +1626,12 @@ async def download_dataset(dataset_id: str, current_user: dict = Depends(get_cur @router.get("/dataset-manage/download/{dataset_id}/{file_id}") async def download_dataset_file(dataset_id: str, file_id: str, version_id: str | None = Query(default=None)) -> PlainTextResponse: - row = get_platform_store().dataset_file(file_id) - return PlainTextResponse(row["content"], media_type="text/plain") + store = get_platform_store() + row = store.dataset_file(file_id) + if str(row.get("dataset_id")) != str(dataset_id): + raise fail(404, "dataset file not found") + content = _dataset_version_bytes(store, file_id, version_id) if version_id else _dataset_file_bytes(store, file_id) + return PlainTextResponse(content.decode("utf-8", errors="replace"), media_type="text/plain") @router.get("/dataset-manage") @@ -1880,12 +2097,26 @@ async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: di # 1. Create eval task record payload.setdefault("created_by", current_user.get("id")) task = store.create_eval_task({**payload, "status": "pending"}) + if get_settings().minio_enabled: + _store_json_snapshot( + store, + "eval", + str(task["id"]), + str(task["id"]), + f"evaluations/{task['id']}/versions/{task['id']}/evaluation-config.json", + payload, + current_user.get("id"), + ) # 2. Resolve model path (supports both regular models and trained models) model_id = str(payload.get("model_id", "")) model_path = "" adapter_path = payload.get("adapter_path", "") model_node_id = "" + ds_files: list[dict[str, Any]] = [] + model_resource_type = "model" + model_resource_id = model_id + adapter_resource_id = "" try: db_model = store.model(model_id) model_path = db_model.get("path", "") @@ -1894,6 +2125,9 @@ async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: di # Try trained_models table (IDs prefixed with tm_) trained = next((m for m in store.trained_models() if m["id"] == model_id), None) if trained: + model_resource_type = "trained_model" if trained.get("merged") else "model" + model_resource_id = trained.get("id") or model_id + adapter_resource_id = trained.get("id") or "" model_node_id = trained.get("compute_node_id") or "" merged_path = trained.get("merged_path", "") base_path = trained.get("base_model_path", "") @@ -1907,6 +2141,9 @@ async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: di adapter_path = merged_path else: model_path = merged_path or base_path + if model_resource_type == "model": + base_model = next((item for item in store.models() if item.get("path") == base_path), None) + model_resource_id = str((base_model or {}).get("id") or base_path) if not model_path: store.update_eval_task(task["id"], {"status": "failed", "error": "model not found or no path"}) return ok({"task_id": task["id"], "status": "failed", "error": "model not found or no path"}) @@ -1979,6 +2216,22 @@ async def model_eval_start(payload: dict[str, Any] = Body(...), current_user: di store.update_eval_task(task["id"], {"status": "failed", "error": message}) return ok({"task_id": task["id"], "status": "failed", "error": message}) + if get_settings().minio_enabled and get_settings().compute_mode != "simulator": + try: + prepared_model = await _prepare_resource_on_node(store, model_resource_type, model_resource_id, node) + if prepared_model: + model_path = prepared_model + if adapter_resource_id and model_resource_type == "model": + prepared_adapter = await _prepare_resource_on_node(store, "trained_model", adapter_resource_id, node) + if prepared_adapter: + adapter_path = prepared_adapter + dataset_sync = await _sync_training_dataset_to_compute_node(store, node, dataset_id) + if dataset_sync: + dataset_path = str(dataset_sync[0].get("local_path") or dataset_path) + except Exception as exc: + store.update_eval_task(task["id"], {"status": "failed", "error": f"MinIO resource preparation failed: {exc}"}) + return ok({"task_id": task["id"], "status": "failed", "error": str(exc)}) + node_gpus = { int(item.get("id", item.get("gpu_index", -1))): item for item in store.gpus() diff --git a/backend/app/core/audit.py b/backend/app/core/audit.py index 66544f2..6601c52 100644 --- a/backend/app/core/audit.py +++ b/backend/app/core/audit.py @@ -61,6 +61,7 @@ def audit_log( detail = _build_detail(detail_template, kwargs) _record_audit( action=action, + actor_id=_extract_actor_id(kwargs), target_type=target_type, target_id=target_id, detail=detail, @@ -87,6 +88,7 @@ def audit_log( detail = _build_detail(detail_template, kwargs) _record_audit( action=action, + actor_id=_extract_actor_id(kwargs), target_type=target_type, target_id=target_id, detail=detail, @@ -136,6 +138,7 @@ def _build_detail(template: str, kwargs: dict) -> str: def _record_audit( action: str, + actor_id: Optional[str], target_type: str, target_id: Optional[str], detail: str, @@ -149,6 +152,7 @@ def _record_audit( store = get_platform_store() store.record_audit( action=action, + actor_id=actor_id, target_type=target_type or None, target_id=target_id, detail=f"{detail} trace_id={trace_id} duration_ms={duration_ms:.1f}" if detail else f"trace_id={trace_id} duration_ms={duration_ms:.1f}", @@ -157,6 +161,15 @@ def _record_audit( logger.error("写入审计日志失败 action=%s", action, exc_info=True) +def _extract_actor_id(kwargs: dict) -> Optional[str]: + """从 FastAPI 注入的当前用户中提取操作人 ID。""" + for key in ("current_user", "user"): + value = kwargs.get(key) + if isinstance(value, dict) and value.get("id"): + return str(value["id"]) + return None + + # ==================== 预定义的审计操作常量 ==================== class AuditActions: diff --git a/backend/app/core/config.py b/backend/app/core/config.py index dd68d01..99d60e7 100644 --- a/backend/app/core/config.py +++ b/backend/app/core/config.py @@ -68,6 +68,9 @@ class Settings: minio_secret_key: str = os.getenv("MINIO_SECRET_KEY", "minioadmin") minio_bucket: str = os.getenv("MINIO_BUCKET", "yg-ft-resources") minio_secure: bool = _bool_env("MINIO_SECURE", False) + # Small text/data files stay inline in PostgreSQL to avoid unnecessary + # MinIO round trips. Larger files remain the shared canonical objects. + minio_inline_max_bytes: int = _int_env("MINIO_INLINE_MAX_BYTES", 256 * 1024) storage_wait_seconds: int = _int_env("STORAGE_WAIT_SECONDS", 300) storage_check_interval_seconds: int = _int_env("STORAGE_CHECK_INTERVAL_SECONDS", 10) compute_service_token: str = os.getenv("COMPUTE_SERVICE_TOKEN", "") diff --git a/backend/app/db/platform_store.py b/backend/app/db/platform_store.py index cea1469..2474354 100644 --- a/backend/app/db/platform_store.py +++ b/backend/app/db/platform_store.py @@ -551,6 +551,16 @@ class PlatformStore: "last_error": "TEXT", }, ) + self._ensure_columns( + conn, + "storage_objects", + {"metadata": "TEXT NOT NULL DEFAULT '{}'"}, + ) + self._ensure_columns( + conn, + "model_artifacts", + {"storage_object_id": "TEXT", "storage_backend": "TEXT NOT NULL DEFAULT 'minio'"}, + ) schema_dir = Path(__file__).with_name("sql") for extra in ( "002_governance.sql", @@ -562,7 +572,17 @@ class PlatformStore: if extra_path.exists(): conn.executescript(extra_path.read_text(encoding="utf-8")) # data_convert_tasks 表补充 created_by 字段（用于数据隔离） - self._ensure_columns(conn, "data_convert_tasks", {"created_by": "TEXT"}) + self._ensure_columns( + conn, + "data_convert_tasks", + { + "created_by": "TEXT", + "storage_backend": "TEXT NOT NULL DEFAULT 'minio'", + "output_storage_object_id": "TEXT", + "output_content": "TEXT", + }, + ) + self._ensure_columns(conn, "eval_tasks", {"report_storage_object_id": "TEXT"}) # 修复历史数据：将 data_convert_tasks.created_by 回填到关联的 datasets 记录 try: conn.execute(""" @@ -1260,6 +1280,13 @@ class PlatformStore: raise KeyError(artifact_id) return {**dict(row), "metadata": json_loads(row["metadata"], {})} + def link_model_artifact_storage_object(self, artifact_id: str, storage_object_id: str) -> None: + with self.connect() as conn: + conn.execute( + "UPDATE model_artifacts SET storage_object_id=?, storage_backend='minio' WHERE id=?", + (storage_object_id, artifact_id), + ) + def model_lineage(self, model_id: str) -> dict[str, Any]: with self.connect() as conn: parents = conn.execute( @@ -2227,7 +2254,7 @@ class PlatformStore: (str(validation_dataset_id),), ).fetchall(), ] - model_path = (model and model.get("path")) or task.get("model_name_or_path") or base_model_id + model_path = task.get("prepared_base_model_path") or (model and model.get("path")) or task.get("model_name_or_path") or base_model_id dataset_metadata = json_loads(dataset.get("metadata"), {}) if dataset else {} if not dataset or dataset.get("type") != "train" or dataset_metadata.get( "dataset_split" @@ -3034,17 +3061,18 @@ class PlatformStore: """ 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 (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + content_type, checksum_sha256, byte_size, status, metadata, created_by, create_time) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) 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=EXCLUDED.status, created_by=EXCLUDED.created_by + status=EXCLUDED.status, metadata=EXCLUDED.metadata, created_by=EXCLUDED.created_by """, ( object_id, payload["resource_type"], payload["resource_id"], payload["version_id"], payload["bucket"], payload["object_key"], payload.get("file_name"), payload.get("content_type"), payload.get("checksum_sha256"), int(payload.get("byte_size") or 0), payload.get("status", "pending"), + json_dumps(payload.get("metadata") or {}), payload.get("created_by"), payload.get("create_time") or utcnow(), ), ) @@ -3083,6 +3111,13 @@ class PlatformStore: ).fetchall() return [dict(row) for row in rows] + def storage_object(self, object_id: str) -> dict[str, Any]: + with self.connect() as conn: + row = conn.execute("SELECT * FROM storage_objects WHERE id=?", (object_id,)).fetchone() + if not row: + raise KeyError(object_id) + return dict(row) + def update_storage_object(self, object_id: str, payload: dict[str, Any]) -> dict[str, Any]: allowed = {"status", "checksum_sha256", "byte_size", "content_type"} fields = {key: value for key, value in payload.items() if key in allowed} @@ -3758,6 +3793,8 @@ class PlatformStore: actor_id: str | None = None, action: str | None = None, target_type: str | None = None, + target_id: str | None = None, + keyword: str | None = None, start_time: str | None = None, end_time: str | None = None, limit: int = 50, @@ -3780,6 +3817,13 @@ class PlatformStore: if target_type: clauses.append("target_type=?") params.append(target_type) + if target_id: + clauses.append("target_id=?") + params.append(target_id) + if keyword: + clauses.append("(target_id LIKE ? OR detail LIKE ?)") + pattern = f"%{keyword}%" + params.extend([pattern, pattern]) if start_time: clauses.append("time>=?") params.append(start_time) diff --git a/backend/app/db/sql/000_full_init.sql b/backend/app/db/sql/000_full_init.sql index 222e7db..7ec5813 100644 --- a/backend/app/db/sql/000_full_init.sql +++ b/backend/app/db/sql/000_full_init.sql @@ -111,6 +111,8 @@ CREATE TABLE IF NOT EXISTS model_artifacts ( path TEXT NOT NULL, size_bytes BIGINT NOT NULL DEFAULT 0, checksum_sha256 TEXT, + storage_object_id TEXT, + storage_backend TEXT NOT NULL DEFAULT 'minio', metadata TEXT NOT NULL, compute_job_id TEXT, create_time TEXT NOT NULL @@ -303,6 +305,7 @@ CREATE TABLE IF NOT EXISTS storage_objects ( checksum_sha256 TEXT, byte_size BIGINT NOT NULL DEFAULT 0, status TEXT NOT NULL DEFAULT 'pending', + metadata TEXT NOT NULL DEFAULT '{}', created_by TEXT, create_time TEXT NOT NULL, UNIQUE (resource_type, resource_id, version_id, object_key) @@ -630,6 +633,9 @@ ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS metadata TEXT NOT NULL DEFAUL ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS created_at TIMESTAMPTZ NOT NULL DEFAULT now(); ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS updated_at TIMESTAMPTZ NOT NULL DEFAULT now(); ALTER TABLE dataset_files ADD COLUMN IF NOT EXISTS deleted_at TIMESTAMPTZ; +ALTER TABLE model_artifacts ADD COLUMN IF NOT EXISTS storage_object_id TEXT; +ALTER TABLE model_artifacts ADD COLUMN IF NOT EXISTS storage_backend TEXT NOT NULL DEFAULT 'minio'; +ALTER TABLE storage_objects ADD COLUMN IF NOT EXISTS metadata TEXT NOT NULL DEFAULT '{}'; -- ---- 数据处理任务 / 源文件 / 预览 / 结果 ---- @@ -860,12 +866,17 @@ CREATE TABLE IF NOT EXISTS data_convert_tasks ( input_count INTEGER NOT NULL DEFAULT 0, output_count INTEGER NOT NULL DEFAULT 0, error_message TEXT, + output_content TEXT, create_time TEXT NOT NULL DEFAULT (to_char(now(), 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"')), update_time TEXT NOT NULL DEFAULT (to_char(now(), 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"')), deleted_at TIMESTAMPTZ ); CREATE INDEX IF NOT EXISTS idx_data_convert_tasks_status ON data_convert_tasks(status); CREATE INDEX IF NOT EXISTS idx_data_convert_tasks_create_time ON data_convert_tasks(create_time DESC); +ALTER TABLE data_convert_tasks ADD COLUMN IF NOT EXISTS storage_backend TEXT NOT NULL DEFAULT 'minio'; +ALTER TABLE data_convert_tasks ADD COLUMN IF NOT EXISTS output_storage_object_id TEXT; +ALTER TABLE data_convert_tasks ADD COLUMN IF NOT EXISTS output_content TEXT; +ALTER TABLE eval_tasks ADD COLUMN IF NOT EXISTS report_storage_object_id TEXT; -- ============================================================================ -- 七、种子数据：初始管理员 / 操作员 diff --git a/backend/app/modules/compute_gateway/sync.py b/backend/app/modules/compute_gateway/sync.py index e57b775..ce6183d 100644 --- a/backend/app/modules/compute_gateway/sync.py +++ b/backend/app/modules/compute_gateway/sync.py @@ -2,15 +2,75 @@ from __future__ import annotations import json import time +from pathlib import Path from typing import Any from app.db.platform_store import get_platform_store +from app.core.config import get_settings from app.modules.compute_gateway.client import ComputeNodeClient +from app.modules.storage.minio_store import get_object_storage # starting 状态允许的最大轮询次数（约 40 * 3s ≈ 2 分钟），超过即判定节点不可达 MAX_STARTING_ATTEMPTS = 40 +async def _archive_node_directory( + store: Any, + client: ComputeNodeClient, + node: dict[str, Any], + source_path: str, + resource_type: str, + resource_id: str, + version_id: str, + object_prefix: str, +) -> list[dict[str, Any]]: + """Archive a completed node directory to MinIO, preserving subdirectories.""" + data_root = Path(str(node.get("data_root") or "/data/yg-ft")).resolve() + source = Path(source_path).resolve() + try: + relative_root = source.relative_to(data_root).as_posix() + except ValueError as exc: + raise RuntimeError(f"artifact path is outside compute data root: {source_path}") from exc + queue = [relative_root] + archived: list[dict[str, Any]] = [] + while queue: + relative = queue.pop(0) + listing = await client.list_files(root="data", relative_path=relative) + for item in listing.get("items") or []: + item_relative = str(item.get("relative_path") or "") + if item.get("type") == "directory": + queue.append(item_relative) + continue + path = str(item.get("path") or "") + if not path: + continue + try: + relative_file = Path(item_relative).relative_to(Path(relative_root)).as_posix() + except ValueError: + relative_file = Path(str(item.get("name") or Path(path).name)).name + object_key = f"{object_prefix}/{version_id}/{relative_file}" + upload_url = get_object_storage().presigned_put(object_key) + result = await client.upload_file_to_url(path, upload_url, object_key) + metadata = get_object_storage().stat(object_key) + archived.append( + store.create_storage_object( + { + "resource_type": resource_type, + "resource_id": resource_id, + "version_id": version_id, + "bucket": get_object_storage().bucket, + "object_key": object_key, + "file_name": relative_file, + "content_type": "application/octet-stream", + "byte_size": metadata.get("byte_size") or result.get("byte_size") or 0, + "checksum_sha256": result.get("checksum_sha256") or "", + "status": "available", + } + ) + ) + return archived + + def _node_for_task(task: dict[str, Any]) -> dict[str, Any] | None: return next((node for node in get_platform_store().compute_nodes() if node["id"] == task.get("compute_node_id")), None) @@ -139,7 +199,38 @@ async def poll_compute_jobs_once() -> dict[str, Any]: job["log_snippet"] = str(last_logs.get("content") or "")[:8192] except Exception: pass - synced.append(store.apply_compute_job(task["id"], job)) + updated_task = store.apply_compute_job(task["id"], job) + if ( + get_settings().minio_enabled + and job.get("status") == "completed" + and job.get("output_dir") + ): + trained_model = next( + ( + item + for item in store.trained_models() + if item.get("name") + == (task.get("output_model_name") or f"{task.get('name')}-lora") + ), + None, + ) + if trained_model: + archived = await _archive_node_directory( + store, + client, + node, + str(job["output_dir"]), + "trained_model", + str(trained_model["id"]), + str(job.get("id") or task.get("compute_job_id") or task["id"]), + f"trained_models/{trained_model['id']}", + ) + artifacts = store.model_artifacts(str(trained_model["id"])) + if archived and artifacts: + store.link_model_artifact_storage_object( + str(artifacts[0]["id"]), str(archived[0]["id"]) + ) + synced.append(updated_task) except Exception as exc: # noqa: BLE001 - keep polling other jobs failed.append({"task_id": task["id"], "error": str(exc)}) standalone_synced: list[dict[str, Any]] = [] @@ -151,6 +242,30 @@ async def poll_compute_jobs_once() -> dict[str, Any]: try: job = await ComputeNodeClient(node["api_base_url"]).get_job(record["id"]) standalone_synced.append(store.sync_model_merge_job(record["id"], job)) + if get_settings().minio_enabled and job.get("status") == "completed" and job.get("output_dir"): + payload = (store.compute_job(record["id"]).get("payload") or {}) + trained_model_id = str(payload.get("trained_model_id") or payload.get("model_name") or "") + if trained_model_id: + trained_model = next( + (item for item in store.trained_models() if item.get("id") == trained_model_id or item.get("name") == trained_model_id), + None, + ) + if trained_model: + archived = await _archive_node_directory( + store, + ComputeNodeClient(node["api_base_url"], timeout=900), + node, + str(job["output_dir"]), + "trained_model", + str(trained_model["id"]), + str(job.get("id") or record["id"]), + f"trained_models/{trained_model['id']}", + ) + artifacts = store.model_artifacts(str(trained_model["id"])) + if archived and artifacts: + store.link_model_artifact_storage_object( + str(artifacts[0]["id"]), str(archived[0]["id"]) + ) except Exception as exc: # noqa: BLE001 - keep polling other jobs failed.append({"job_id": record["id"], "error": str(exc)}) @@ -175,6 +290,17 @@ async def poll_compute_jobs_once() -> dict[str, Any]: except Exception: pass store.apply_eval_job_result(eval_task["id"], job, result_content) + if get_settings().minio_enabled and job.get("status") == "completed" and job.get("output_dir"): + await _archive_node_directory( + store, + client, + node, + str(job["output_dir"]), + "eval", + str(eval_task["id"]), + str(job.get("id") or eval_task.get("compute_job_id") or eval_task["id"]), + f"evaluations/{eval_task['id']}", + ) # 评测 GPU 占用由 eval_tasks 状态派生，无需维护推理内存标记 eval_synced += 1 except Exception as exc: # noqa: BLE001 diff --git a/backend/app/modules/data_convert/router.py b/backend/app/modules/data_convert/router.py index 4ed11e0..2a364b6 100644 --- a/backend/app/modules/data_convert/router.py +++ b/backend/app/modules/data_convert/router.py @@ -1,17 +1,21 @@ from __future__ import annotations import json +import hashlib import os from pathlib import Path from typing import Any from fastapi import APIRouter, Body, Depends, File, UploadFile -from fastapi.responses import FileResponse +from fastapi.responses import FileResponse, Response from app.api.v1.endpoints.platform import ok, fail from app.core.auth import get_current_user, is_admin +from app.core.config import get_settings from app.core.op_log import op_log, OpModule, OpAction from app.db.platform_store import get_platform_store, new_id +from app.modules.storage.minio_store import get_object_storage +from app.modules.storage.policy import should_store_in_minio router = APIRouter(prefix="/data-convert", tags=["data-convert"]) @@ -56,6 +60,142 @@ def _output_dir(task_id: str) -> Path: return _task_dir(task_id) / "output" +def _minio_enabled() -> bool: + return bool(get_settings().minio_enabled) + + +def _input_object_key(task_id: str, name: str) -> str: + return f"data-convert/{task_id}/input/{Path(name).name}" + + +def _output_object_key(task_id: str, name: str) -> str: + return f"data-convert/{task_id}/output/{Path(name).name}" + + +def _task_objects(task_id: str) -> list[dict[str, Any]]: + return get_platform_store().storage_objects_for_resource("data_convert", task_id) + + +def _register_object( + task_id: str, + *, + version_id: str, + object_key: str, + file_name: str, + content_type: str, + content: bytes, + created_by: str | None, +) -> dict[str, Any]: + storage = get_object_storage() + uploaded = storage.put_bytes(object_key, content, content_type) + return get_platform_store().create_storage_object( + { + "resource_type": "data_convert", + "resource_id": task_id, + "version_id": version_id, + "bucket": uploaded["bucket"], + "object_key": object_key, + "file_name": file_name, + "content_type": content_type, + "byte_size": len(content), + "checksum_sha256": hashlib.sha256(content).hexdigest(), + "status": "available", + "created_by": created_by, + } + ) + + +def _input_objects(task_id: str) -> list[dict[str, Any]]: + prefix = f"data-convert/{task_id}/input/" + return sorted( + [item for item in _task_objects(task_id) if str(item.get("object_key") or "").startswith(prefix)], + key=lambda item: str(item.get("file_name") or item.get("object_key") or ""), + ) + + +def _output_object(task: dict[str, Any]) -> dict[str, Any] | None: + key = _output_object_key(task["id"], _safe_output_filename(task.get("output_filename"))) + return next((item for item in _task_objects(task["id"]) if item.get("object_key") == key), None) + + +def _read_output(task: dict[str, Any]) -> bytes | None: + if _minio_enabled(): + item = _output_object(task) + if item: + return get_object_storage().get_bytes(item["object_key"]) + inline = task.get("output_content") + return str(inline).encode("utf-8") if inline is not None else None + path = _task_output_path(task) + return path.read_bytes() if path.exists() else None + + +def _convert_from_minio(task: dict[str, Any], created_by: str | None) -> tuple[int, int, bytes]: + output_name = _safe_output_filename(task.get("output_filename")) + output_lines: list[str] = [] + input_count = 0 + output_count = 0 + for item in _input_objects(task["id"]): + input_count += 1 + raw = get_object_storage().get_bytes(item["object_key"]) + data = json.loads(raw.decode("utf-8")) + if isinstance(data, list): + records = data + elif isinstance(data, dict): + records = [data] + else: + raise ValueError(f"JSON must be object or array: {item.get('file_name')}") + for record in records: + output_lines.append(json.dumps(record, ensure_ascii=False) + "\n") + output_count += 1 + output = "".join(output_lines).encode("utf-8") + store = get_platform_store() + with store.connect() as conn: + if should_store_in_minio(len(output)): + output_object = _register_object( + task["id"], + version_id="output", + object_key=_output_object_key(task["id"], output_name), + file_name=output_name, + content_type="application/jsonl", + content=output, + created_by=created_by, + ) + conn.execute( + "UPDATE data_convert_tasks SET output_storage_object_id=%s, output_content=NULL, storage_backend='minio' WHERE id=%s", + (output_object["id"], task["id"]), + ) + else: + conn.execute( + "UPDATE data_convert_tasks SET output_storage_object_id=NULL, output_content=%s, storage_backend='database' WHERE id=%s", + (output.decode("utf-8"), task["id"]), + ) + return input_count, output_count, output + + +def _convert_from_local(task: dict[str, Any]) -> tuple[int, int, bytes]: + input_dir = _input_dir(task["id"]) + output_dir = _output_dir(task["id"]) + input_dir.mkdir(parents=True, exist_ok=True) + output_dir.mkdir(parents=True, exist_ok=True) + output_path = _task_output_path(task) + output_path.unlink(missing_ok=True) + input_count = 0 + output_count = 0 + with output_path.open("w", encoding="utf-8") as output_file: + for json_file in sorted(input_dir.iterdir()): + if not json_file.is_file() or not json_file.name.lower().endswith(".json"): + continue + input_count += 1 + data = json.loads(json_file.read_text(encoding="utf-8")) + records = data if isinstance(data, list) else [data] if isinstance(data, dict) else None + if records is None: + raise ValueError(f"JSON must be object or array: {json_file.name}") + for record in records: + output_file.write(json.dumps(record, ensure_ascii=False) + "\n") + output_count += 1 + return input_count, output_count, output_path.read_bytes() + + @router.get("") def list_tasks( page: int = 1, @@ -109,9 +249,10 @@ def create_task( "VALUES (%s, %s, %s, %s, %s)", (task_id, name, description, output_filename, user_id), ) - # 创建目录 - _input_dir(task_id).mkdir(parents=True, exist_ok=True) - _output_dir(task_id).mkdir(parents=True, exist_ok=True) + # MinIO 是正式存储；本地目录只在关闭 MinIO 的旧兼容模式下创建。 + if not _minio_enabled(): + _input_dir(task_id).mkdir(parents=True, exist_ok=True) + _output_dir(task_id).mkdir(parents=True, exist_ok=True) return ok(_get_task(task_id)) @@ -123,13 +264,19 @@ def get_task( task = _get_task(task_id) if not task: raise fail(404, "task not found") - # 附加输入文件列表 - input_dir = _input_dir(task_id) + # 附加输入文件列表；旧任务没有对象记录时继续读取本地兼容目录。 files = [] - if input_dir.exists(): - for f in sorted(input_dir.iterdir()): - if f.is_file(): - files.append({"name": f.name, "size": f.stat().st_size}) + if _minio_enabled(): + files = [ + {"name": item.get("file_name") or Path(item["object_key"]).name, "size": item.get("byte_size") or 0} + for item in _input_objects(task_id) + ] + else: + input_dir = _input_dir(task_id) + if input_dir.exists(): + for f in sorted(input_dir.iterdir()): + if f.is_file(): + files.append({"name": f.name, "size": f.stat().st_size}) task["input_files"] = files return ok(task) @@ -146,16 +293,26 @@ async def upload_source_files( raise fail(404, "task not found") if task["status"] not in ("pending", "uploaded"): raise fail(400, "task is not editable") - input_dir = _input_dir(task_id) - input_dir.mkdir(parents=True, exist_ok=True) staged = [] for upload in files: name = Path(upload.filename or "input.json").name if not name.lower().endswith(".json"): raise fail(415, f"only JSON files are supported: {name}") - target = input_dir / name content = await upload.read() - target.write_bytes(content) + if _minio_enabled(): + _register_object( + task_id, + version_id=f"input-{hashlib.sha256(name.encode('utf-8')).hexdigest()[:16]}", + object_key=_input_object_key(task_id, name), + file_name=name, + content_type=upload.content_type or "application/json", + content=content, + created_by=task.get("created_by") or current_user.get("id"), + ) + else: + input_dir = _input_dir(task_id) + input_dir.mkdir(parents=True, exist_ok=True) + (input_dir / name).write_bytes(content) staged.append({"name": name, "size": len(content)}) store = get_platform_store() # 标记上传完成 @@ -166,30 +323,12 @@ async def upload_source_files( ) # 自动转换并导入数据集 try: - output_dir = _output_dir(task_id) - output_dir.mkdir(parents=True, exist_ok=True) - output_path = _task_output_path(task) - # 清空旧输出（如果重新上传） - if output_path.exists(): - output_path.unlink() - input_count = 0 - output_count = 0 - for json_file in sorted(input_dir.iterdir()): - if not json_file.is_file() or not json_file.name.lower().endswith(".json"): - continue - input_count += 1 - with open(json_file, "r", encoding="utf-8") as f: - data = json.load(f) - if isinstance(data, list): - records = data - elif isinstance(data, dict): - records = [data] - else: - raise ValueError(f"JSON must be object or array: {json_file.name}") - with open(output_path, "a", encoding="utf-8") as f: - for record in records: - f.write(json.dumps(record, ensure_ascii=False) + "\n") - output_count += 1 + if _minio_enabled(): + input_count, output_count, output = _convert_from_minio( + task, task.get("created_by") or current_user.get("id") + ) + else: + input_count, output_count, output = _convert_from_local(task) with store.connect() as conn: conn.execute( "UPDATE data_convert_tasks SET status='completed', " @@ -197,12 +336,12 @@ async def upload_source_files( (input_count, output_count, task_id), ) # 自动导入数据集 - content = output_path.read_text(encoding="utf-8") + content = output.decode("utf-8") size_bytes = len(content.encode("utf-8")) dataset = store.create_dataset({ "name": task["name"], "type": "train", - "storage_type": "local", + "storage_type": "minio" if should_store_in_minio(size_bytes) else ("database" if _minio_enabled() else "local"), "source": "upload", "task_id": task_id, "size": f"{size_bytes} B", @@ -212,7 +351,20 @@ async def upload_source_files( }) dataset_id = dataset["id"] with store.connect() as conn: - store.add_dataset_file(conn, dataset_id, _safe_output_filename(task.get("output_filename")), content) + dataset_file = store.add_dataset_file(conn, dataset_id, _safe_output_filename(task.get("output_filename")), content) + if should_store_in_minio(len(output)): + output_name = _safe_output_filename(task.get("output_filename")) + object_key = f"datasets/{dataset_id}/versions/{dataset_file.get('active_version_id') or dataset_file['id']}/{output_name}" + uploaded = get_object_storage().put_bytes(object_key, output, "application/jsonl") + storage_object = store.create_storage_object({ + "resource_type": "dataset", "resource_id": dataset_id, + "version_id": dataset_file.get("active_version_id") or dataset_file["id"], + "bucket": uploaded["bucket"], "object_key": object_key, + "file_name": output_name, "content_type": "application/jsonl", + "byte_size": len(output), "checksum_sha256": hashlib.sha256(output).hexdigest(), + "status": "available", "created_by": task.get("created_by") or current_user.get("id"), + }) + store.link_dataset_file_storage_object(dataset_file["id"], storage_object["id"]) return ok({ "staged_files": staged, "auto_converted": True, @@ -248,28 +400,10 @@ def run_convert( (task_id,), ) try: - input_dir = _input_dir(task_id) - output_dir = _output_dir(task_id) - output_dir.mkdir(parents=True, exist_ok=True) - output_path = _task_output_path(task) - input_count = 0 - output_count = 0 - for json_file in sorted(input_dir.iterdir()): - if not json_file.is_file() or not json_file.name.lower().endswith(".json"): - continue - input_count += 1 - with open(json_file, "r", encoding="utf-8") as f: - data = json.load(f) - if isinstance(data, list): - records = data - elif isinstance(data, dict): - records = [data] - else: - raise ValueError(f"JSON must be object or array: {json_file.name}") - with open(output_path, "a", encoding="utf-8") as f: - for record in records: - f.write(json.dumps(record, ensure_ascii=False) + "\n") - output_count += 1 + if _minio_enabled(): + input_count, output_count, _ = _convert_from_minio(task, task.get("created_by") or current_user.get("id")) + else: + input_count, output_count, _ = _convert_from_local(task) # 更新任务状态 with store.connect() as conn: conn.execute( @@ -297,11 +431,15 @@ def download_result( raise fail(404, "task not found") if task["status"] != "completed": raise fail(400, "task is not completed") - output_path = _task_output_path(task) - if not output_path.exists(): + output = _read_output(task) + if output is None: raise fail(404, "output file not found") + if _minio_enabled(): + return Response(content=output, media_type="application/octet-stream", headers={ + "Content-Disposition": f"attachment; filename={_safe_output_filename(task.get('output_filename'))}" + }) return FileResponse( - str(output_path), + str(_task_output_path(task)), media_type="application/octet-stream", filename=_safe_output_filename(task.get("output_filename")), ) @@ -319,10 +457,10 @@ def import_as_dataset( raise fail(404, "task not found") if task["status"] != "completed": raise fail(400, "task is not completed") - output_path = _task_output_path(task) - if not output_path.exists(): + output = _read_output(task) + if output is None: raise fail(404, "output file not found") - content = output_path.read_text(encoding="utf-8") + content = output.decode("utf-8") dataset_name = str(payload.get("name") or task["name"]).strip() description = str(payload.get("description") or f"由数据类型转换任务 {task_id} 导入").strip() size_bytes = len(content.encode("utf-8")) @@ -331,7 +469,7 @@ def import_as_dataset( dataset = store.create_dataset({ "name": dataset_name, "type": "train", - "storage_type": "local", + "storage_type": "minio" if should_store_in_minio(size_bytes) else ("database" if _minio_enabled() else "local"), "source": "upload", "task_id": task_id, "size": f"{size_bytes} B", @@ -341,7 +479,20 @@ def import_as_dataset( }) dataset_id = dataset["id"] with store.connect() as conn: - store.add_dataset_file(conn, dataset_id, _safe_output_filename(task.get("output_filename")), content) + dataset_file = store.add_dataset_file(conn, dataset_id, _safe_output_filename(task.get("output_filename")), content) + if should_store_in_minio(len(output)): + output_name = _safe_output_filename(task.get("output_filename")) + object_key = f"datasets/{dataset_id}/versions/{dataset_file.get('active_version_id') or dataset_file['id']}/{output_name}" + uploaded = get_object_storage().put_bytes(object_key, output, "application/jsonl") + storage_object = store.create_storage_object({ + "resource_type": "dataset", "resource_id": dataset_id, + "version_id": dataset_file.get("active_version_id") or dataset_file["id"], + "bucket": uploaded["bucket"], "object_key": object_key, + "file_name": output_name, "content_type": "application/jsonl", + "byte_size": len(output), "checksum_sha256": hashlib.sha256(output).hexdigest(), + "status": "available", "created_by": task.get("created_by") or (current_user.get("id") if current_user else None), + }) + store.link_dataset_file_storage_object(dataset_file["id"], storage_object["id"]) return ok({"dataset_id": dataset_id, "name": dataset_name}) @@ -360,11 +511,19 @@ def delete_task( "UPDATE data_convert_tasks SET deleted_at=NOW() WHERE id=%s", (task_id,), ) - # 清理文件 - import shutil - task_dir = _task_dir(task_id) - if task_dir.exists(): - shutil.rmtree(task_dir, ignore_errors=True) + if _minio_enabled(): + for item in _task_objects(task_id): + try: + get_object_storage().delete(item["object_key"]) + store.update_storage_object(item["id"], {"status": "deleted"}) + except Exception: + pass + else: + # 旧兼容数据仍清理本地目录。 + import shutil + task_dir = _task_dir(task_id) + if task_dir.exists(): + shutil.rmtree(task_dir, ignore_errors=True) return ok({"deleted": task_id}) diff --git a/backend/app/modules/data_process/storage.py b/backend/app/modules/data_process/storage.py index ce44a24..513f32a 100644 --- a/backend/app/modules/data_process/storage.py +++ b/backend/app/modules/data_process/storage.py @@ -1,4 +1,4 @@ -"""数据处理原始源文件的受控本地对象存储。""" +"""数据处理源文件的受控暂存与分层对象存储。""" from __future__ import annotations @@ -13,6 +13,9 @@ from pathlib import Path, PurePosixPath from typing import Iterable, Iterator from urllib.parse import quote, unquote, urlsplit +from app.core.config import get_settings +from app.modules.storage.minio_store import get_object_storage + class DataProcessStorageError(ValueError): """本地对象引用或文件系统状态不安全。""" @@ -68,7 +71,7 @@ def _safe_basename(value: str) -> str: class LocalDataProcessStorage: - """只允许访问配置根目录下的版本化原始文件。""" + """Stage locally, but publish and read authoritative source files from MinIO.""" def __init__(self, root: str | os.PathLike[str] | Path | None = None) -> None: configured = Path(root) if root is not None else _configured_storage_root() @@ -132,10 +135,7 @@ class LocalDataProcessStorage: f"v{version}", basename, ) - reference = ( - "local://data-process/" - f"{task_id}/{source_file_id}/v{version}/{quote(basename, safe='')}" - ) + reference = self._reference(task_id, source_file_id, version, basename) staged = StagedSourceObject(reference, temporary_path, relative_path) self._issued_staged_objects[temporary_path] = staged return staged @@ -168,6 +168,18 @@ class LocalDataProcessStorage: expected_task_id=expected_source_task_id, expected_source_file_id=expected_source_file_id, ) + if self._is_minio_reference(source_reference): + content = self.read(source_reference) + if content is None: + raise DataProcessStorageError("original source object is not available") + return self.stage_bytes( + batch_id=batch_id, + task_id=task_id, + source_file_id=source_file_id, + version=version, + name=basename, + content=content, + ) descriptor, source_info = self._open_read_descriptor(source_relative) os.close(descriptor) @@ -193,10 +205,7 @@ class LocalDataProcessStorage: f"v{version}", basename, ) - reference = ( - "local://data-process/" - f"{task_id}/{source_file_id}/v{version}/{quote(basename, safe='')}" - ) + reference = self._reference(task_id, source_file_id, version, basename) staged = StagedSourceObject(reference, temporary_path, relative_path) self._issued_staged_objects[temporary_path] = staged return staged @@ -212,14 +221,22 @@ class LocalDataProcessStorage: raise DataProcessStorageError("duplicate staged source object") seen_temporary_paths.add(item._temporary_path) for item in staged: - final_path = self._path_for_relative(item._relative_path) - self._ensure_directory(final_path.parent) - if final_path.exists() or final_path.is_symlink(): - raise DataProcessStorageError("source storage object already exists") - os.link(item._temporary_path, final_path, follow_symlinks=False) + if self._is_minio_reference(item.reference): + content = item._temporary_path.read_bytes() + get_object_storage().put_bytes( + self.object_key(item.reference), + content, + "application/octet-stream", + ) + elif not item.reference.startswith("db://data-process/"): + final_path = self._path_for_relative(item._relative_path) + self._ensure_directory(final_path.parent) + if final_path.exists() or final_path.is_symlink(): + raise DataProcessStorageError("source storage object already exists") + os.link(item._temporary_path, final_path, follow_symlinks=False) + self._fsync_directory(final_path.parent) published.append(item) item._temporary_path.unlink() - self._fsync_directory(final_path.parent) except Exception: for item in reversed(published): try: @@ -258,7 +275,13 @@ class LocalDataProcessStorage: raise first_error def read(self, reference: str) -> bytes | None: - """读取 local 引用；旧 ``db://`` 对象返回 ``None`` 由数据库正文兜底。""" + """Read a MinIO object or legacy local reference.""" + + if self._is_minio_reference(reference): + try: + return get_object_storage().get_bytes(self.object_key(reference)) + except Exception as exc: # noqa: BLE001 - normalize object-not-found for callers + raise DataProcessStorageError("source storage object does not exist") from exc relative_path = self._relative_from_reference(reference) if relative_path is None: @@ -276,6 +299,11 @@ class LocalDataProcessStorage: ) -> int | None: """返回受控 local 对象大小；旧 ``db://`` 对象没有原始文件。""" + if self._is_minio_reference(reference): + try: + return int(get_object_storage().stat(self.object_key(reference)).get("byte_size") or 0) + except Exception as exc: # noqa: BLE001 + raise DataProcessStorageError("source storage object does not exist") from exc relative_path = self._relative_from_reference(reference) if relative_path is None: return None @@ -301,6 +329,15 @@ class LocalDataProcessStorage: ) -> Iterator[bytes]: """按范围流式读取原始文件，避免 PDF 预览把大文件整体载入内存。""" + if self._is_minio_reference(reference): + content = self.read(reference) or b"" + if start < 0 or expected_size != len(content) or start > expected_size: + raise DataProcessStorageError("source object size does not match metadata") + remaining = expected_size - start if length is None else length + if remaining < 0 or start + remaining > expected_size: + raise DataProcessStorageError("invalid source byte range") + yield content[start : start + remaining] + return relative_path = self._relative_from_reference(reference) if relative_path is None: raise DataProcessStorageError("original source object is not available") @@ -337,6 +374,12 @@ class LocalDataProcessStorage: ) -> bool: """校验 local 引用归属；旧 ``db://`` 引用无需文件系统处理。""" + if self._is_minio_reference(reference): + self._assert_minio_owner(reference, expected_task_id, expected_source_file_id) + return True + if str(reference or "").startswith("db://data-process/"): + self._assert_database_owner(reference, expected_task_id, expected_source_file_id) + return True relative_path = self._relative_from_reference(reference) if relative_path is None: return False @@ -382,6 +425,19 @@ class LocalDataProcessStorage: ) -> bool: """删除受控 local 对象；旧 ``db://`` 引用保持不变。""" + if self._is_minio_reference(reference): + if (expected_task_id is None) != (expected_source_file_id is None): + raise DataProcessStorageError("both expected storage owner fields are required") + if expected_task_id is not None and expected_source_file_id is not None: + self._assert_minio_owner(reference, expected_task_id, expected_source_file_id) + get_object_storage().delete(self.object_key(reference)) + return True + if str(reference or "").startswith("db://data-process/"): + if (expected_task_id is None) != (expected_source_file_id is None): + raise DataProcessStorageError("both expected storage owner fields are required") + if expected_task_id is not None and expected_source_file_id is not None: + self._assert_database_owner(reference, expected_task_id, expected_source_file_id) + return False relative_path = self._relative_from_reference(reference) if relative_path is None: return False @@ -426,7 +482,7 @@ class LocalDataProcessStorage: if reference.startswith("db://"): return None parsed = urlsplit(reference) - if parsed.scheme != "local" or parsed.netloc != "data-process": + if parsed.scheme not in {"local", "minio"} or parsed.netloc != "data-process": raise DataProcessStorageError("unsupported source storage reference") if parsed.query or parsed.fragment or "\\" in parsed.path: raise DataProcessStorageError("unsafe source storage reference") @@ -460,6 +516,39 @@ class LocalDataProcessStorage: basename = _safe_basename(decoded[3]) return PurePosixPath(task_id, source_file_id, f"v{version}", basename) + @staticmethod + def _is_minio_reference(reference: str) -> bool: + return str(reference or "").startswith("minio://data-process/") + + @staticmethod + def _reference(task_id: str, source_file_id: str, version: int, basename: str) -> str: + scheme = "minio" if get_settings().minio_enabled else "local" + return f"{scheme}://data-process/{task_id}/{source_file_id}/v{version}/{quote(basename, safe='')}" + + @staticmethod + def object_key(reference: str) -> str: + parsed = urlsplit(reference) + if parsed.scheme != "minio" or parsed.netloc != "data-process": + raise DataProcessStorageError("reference is not a MinIO source object") + return "data-process/" + parsed.path.lstrip("/") + + def _assert_minio_owner(self, reference: str, task_id: str, source_file_id: str) -> None: + relative = self._relative_from_reference(reference) + if relative is None: + raise DataProcessStorageError("invalid MinIO source reference") + self._assert_expected_owner(relative, expected_task_id=task_id, expected_source_file_id=source_file_id) + + @staticmethod + def _assert_database_owner(reference: str, task_id: str, source_file_id: str) -> None: + parsed = urlsplit(reference) + parts = parsed.path.lstrip("/").split("/") + if parsed.netloc != "data-process" or len(parts) != 3: + raise DataProcessStorageError("invalid database source reference") + expected_task_id = _safe_component(task_id, "expected task id") + expected_source_file_id = _safe_component(source_file_id, "expected source file id") + if tuple(parts[:2]) != (expected_task_id, expected_source_file_id) or parts[2] != "v1": + raise DataProcessStorageError("source storage object owner mismatch") + def _path_for_relative(self, relative_path: PurePosixPath) -> Path: if relative_path.is_absolute() or any( part in {"", ".", ".."} for part in relative_path.parts @@ -479,9 +568,22 @@ class LocalDataProcessStorage: raise DataProcessStorageError("invalid staged source object") if self._issued_staged_objects.get(item._temporary_path) is not item: raise DataProcessStorageError("staged source object was not issued by this storage") - expected_relative = self._relative_from_reference(item.reference) - if expected_relative is None or expected_relative != item._relative_path: - raise DataProcessStorageError("staged source object reference mismatch") + if item.reference.startswith("db://data-process/"): + parsed = urlsplit(item.reference) + parts = parsed.path.lstrip("/").split("/") + expected = item._relative_path.parts[:3] + if ( + parsed.netloc != "data-process" + or parsed.query + or parsed.fragment + or len(parts) != 3 + or tuple(parts) != expected + ): + raise DataProcessStorageError("staged source object reference mismatch") + else: + expected_relative = self._relative_from_reference(item.reference) + if expected_relative is None or expected_relative != item._relative_path: + raise DataProcessStorageError("staged source object reference mismatch") staging_root = self._root / ".staging" try: relative_temporary = item._temporary_path.relative_to(staging_root) diff --git a/backend/app/modules/data_process/store/base.py b/backend/app/modules/data_process/store/base.py index e7618bf..4784a7a 100644 --- a/backend/app/modules/data_process/store/base.py +++ b/backend/app/modules/data_process/store/base.py @@ -247,14 +247,19 @@ def _source_storage_descriptor( or f"db://data-process/{task_id}/{file_id}/v1" ) expected_local_prefix = f"local://data-process/{task_id}/{file_id}/v1/" + expected_minio_prefix = f"minio://data-process/{task_id}/{file_id}/v1/" expected_database_reference = f"db://data-process/{task_id}/{file_id}/v1" if storage_object_id.startswith(expected_local_prefix) and len(storage_object_id) > len( expected_local_prefix ): storage_backend = "local" + elif storage_object_id.startswith(expected_minio_prefix) and len(storage_object_id) > len( + expected_minio_prefix + ): + storage_backend = "minio" elif storage_object_id == expected_database_reference: storage_backend = "database" - elif storage_object_id.startswith(("local://data-process/", "db://data-process/")): + elif storage_object_id.startswith(("local://data-process/", "minio://data-process/", "db://data-process/")): raise DataProcessStoreError("source storage object owner mismatch") else: raise DataProcessStoreError("unsupported source storage object reference") diff --git a/backend/app/modules/data_process/store/datasets.py b/backend/app/modules/data_process/store/datasets.py index 4950549..edce977 100644 --- a/backend/app/modules/data_process/store/datasets.py +++ b/backend/app/modules/data_process/store/datasets.py @@ -11,6 +11,7 @@ import psycopg from app.core.config import get_settings from app.modules.storage.minio_store import get_object_storage +from app.modules.storage.policy import should_store_in_minio from .base import ( StoreBase, @@ -168,7 +169,11 @@ 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) + # Explicit local is retained for old callers/tests that request the + # legacy backend; all normal platform requests default to MinIO. + allow_minio = bool(get_settings().minio_enabled) and str( + payload.get("storage_type") or "minio" + ).lower() != "local" for split_name in split_order: split_records = [ (source_row, record) @@ -193,12 +198,15 @@ class DatasetsMixin: "storage_object_id": ( f"db://data-process/{task_id}/{file_id}/v1" ), + "store_in_minio": allow_minio and should_store_in_minio( + len(raw), content_type="application/jsonl", file_format="jsonl" + ), } ) source_result_ids = [row["id"] for row in rows] common_metadata = { "source": "data_process", - "storage_backend": "minio" if use_minio else "database", + "storage_backend": "minio" if any(spec["store_in_minio"] for spec in split_specs) else "database", "source_task_id": task_id, "output_type": _task_output_type(task), "reasoning_detail": _task_reasoning_detail(task), @@ -273,7 +281,7 @@ class DatasetsMixin: split_name = str(spec["split"]) dataset_id = dataset_ids[split_name] storage_object_id = str(spec["storage_object_id"]) - if use_minio: + if spec["store_in_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( @@ -356,7 +364,7 @@ class DatasetsMixin: ( dataset_name, dataset_types[split_name], - "minio" if use_minio else (payload.get("storage_type") or "local"), + "minio" if spec["store_in_minio"] else "database", f"{len(spec['raw'])} B", len(spec["raw"]), len(spec["records"]), @@ -386,7 +394,7 @@ class DatasetsMixin: dataset_id, dataset_name, dataset_types[split_name], - "minio" if use_minio else (payload.get("storage_type") or "local"), + "minio" if spec["store_in_minio"] else "database", task_id, task_id, f"{len(spec['raw'])} B", @@ -405,7 +413,11 @@ class DatasetsMixin: ), ).fetchone() - file_metadata = {**dataset_metadata, "file_split": split_name} + file_metadata = { + **dataset_metadata, + "file_split": split_name, + "storage_backend": "minio" if spec["store_in_minio"] else "database", + } version = { "id": spec["version_id"], "version_no": 1, diff --git a/backend/app/modules/data_process/store/source_files.py b/backend/app/modules/data_process/store/source_files.py index 19d01ac..d27f4f3 100644 --- a/backend/app/modules/data_process/store/source_files.py +++ b/backend/app/modules/data_process/store/source_files.py @@ -137,7 +137,7 @@ class SourceFilesMixin: payload["record_count"], payload["file_format"], payload["checksum_sha256"], - payload["content"], + "" if str(storage_object_id or "").startswith("minio://") else payload["content"], str(payload["content"])[:2000], json_dumps(metadata_payload), task.get("tenant_id"), @@ -223,7 +223,17 @@ class SourceFilesMixin: ).fetchone() if not row: raise NotFoundError("source file not found") - return _decode_row(row) or {} + decoded = _decode_row(row) or {} + # New source files keep only a preview in PostgreSQL. Load the + # authoritative body from MinIO on demand for existing processing code. + reference = str(decoded.get("storage_object_id") or "") + if include_content and not decoded.get("content") and reference.startswith("minio://"): + from app.modules.data_process.storage import get_data_process_storage + + decoded["content"] = (get_data_process_storage().read(reference) or b"").decode( + "utf-8", errors="replace" + ) + return decoded def source_content_window( self, task_id: str, file_id: str, offset: int, limit: int diff --git a/backend/app/modules/storage/minio_store.py b/backend/app/modules/storage/minio_store.py index 89fc43c..25de9ba 100644 --- a/backend/app/modules/storage/minio_store.py +++ b/backend/app/modules/storage/minio_store.py @@ -3,6 +3,7 @@ from __future__ import annotations from datetime import timedelta from functools import lru_cache from io import BytesIO +from collections.abc import Iterator from typing import Any from minio import Minio @@ -53,6 +54,64 @@ class MinioObjectStorage: except S3Error as exc: raise ObjectStorageError(str(exc)) from exc + def get_bytes(self, object_key: str) -> bytes: + """Read an object through the backend for small API responses and workers.""" + self._ensure_enabled() + self.ensure_bucket() + response = None + try: + response = self.client.get_object(self.bucket, object_key) + return response.read() + except S3Error as exc: + raise ObjectStorageError(str(exc)) from exc + finally: + if response is not None: + response.close() + response.release_conn() + + def iter_bytes(self, object_key: str, chunk_size: int = 256 * 1024) -> Iterator[bytes]: + """Stream an object without loading the complete file into memory.""" + self._ensure_enabled() + self.ensure_bucket() + response = None + try: + response = self.client.get_object(self.bucket, object_key) + while True: + chunk = response.read(chunk_size) + if not chunk: + break + yield chunk + except S3Error as exc: + raise ObjectStorageError(str(exc)) from exc + finally: + if response is not None: + response.close() + response.release_conn() + + def list_objects(self, prefix: str) -> list[dict[str, Any]]: + self._ensure_enabled() + self.ensure_bucket() + try: + return [ + { + "object_key": item.object_name, + "byte_size": item.size or 0, + "etag": item.etag, + "last_modified": item.last_modified.isoformat() if item.last_modified else None, + } + for item in self.client.list_objects(self.bucket, prefix=prefix, recursive=True) + ] + except S3Error as exc: + raise ObjectStorageError(str(exc)) from exc + + def delete(self, object_key: str) -> None: + self._ensure_enabled() + self.ensure_bucket() + try: + self.client.remove_object(self.bucket, object_key) + except S3Error as exc: + raise ObjectStorageError(str(exc)) from exc + def put_bytes(self, object_key: str, content: bytes, content_type: str = "application/octet-stream") -> dict[str, Any]: self._ensure_enabled() self.ensure_bucket() diff --git a/backend/app/modules/storage/policy.py b/backend/app/modules/storage/policy.py new file mode 100644 index 0000000..d56e062 --- /dev/null +++ b/backend/app/modules/storage/policy.py @@ -0,0 +1,54 @@ +"""Storage placement rules shared by dataset and data-processing flows.""" + +from __future__ import annotations + +from app.core.config import get_settings + + +_INLINE_TEXT_FORMATS = { + "txt", "text", "md", "markdown", "json", "jsonl", "csv", "tsv", + "yaml", "yml", "xml", "html", "text/plain", "application/json", + "application/jsonl", "text/csv", +} + + +def should_store_in_minio( + size_bytes: int | None, + *, + content_type: str | None = None, + file_format: str | None = None, +) -> bool: + """Return whether a file is large enough to use the shared object store. + + Small files remain inline in PostgreSQL so page previews and metadata reads + do not pay an object-storage round trip. MinIO is still mandatory for + large files when it is enabled. + """ + + if not get_settings().minio_enabled: + return False + try: + size = max(0, int(size_bytes or 0)) + except (TypeError, ValueError): + size = 0 + if size > get_settings().minio_inline_max_bytes: + return True + # Binary office/document files remain in MinIO even when small because + # their original bytes cannot be safely represented by a text DB column. + normalized_format = str(file_format or "").strip().lower().lstrip(".") + normalized_type = str(content_type or "").strip().lower().split(";", 1)[0] + if normalized_format or normalized_type: + return not ( + normalized_format in _INLINE_TEXT_FORMATS + or normalized_type in _INLINE_TEXT_FORMATS + or normalized_type.startswith("text/") + ) + return False + + +def storage_backend_for_size(size_bytes: int | None, *, requested: str | None = None) -> str: + """Return ``minio`` or ``database`` for a managed file.""" + + if str(requested or "").strip().lower() == "local": + return "database" + return "minio" if should_store_in_minio(size_bytes) else "database" diff --git a/backend/app/modules/system/router.py b/backend/app/modules/system/router.py index 7ebd5aa..d6d57f4 100644 --- a/backend/app/modules/system/router.py +++ b/backend/app/modules/system/router.py @@ -56,13 +56,15 @@ def audit_logs( actor_id: str | None = Query(default=None, description="操作人 ID"), action: str | None = Query(default=None, description="动作类型"), target_type: str | None = Query(default=None, description="目标类型"), + target_id: str | None = Query(default=None, description="目标 ID"), + keyword: str | None = Query(default=None, description="目标 ID 或详情关键字"), start_time: str | None = Query(default=None, description="ISO8601 起始时间"), end_time: str | None = Query(default=None, description="ISO8601 结束时间"), limit: int = Query(default=50, ge=1, le=200), offset: int = Query(default=0, ge=0), current_user: dict = Depends(get_current_user), ) -> dict: - """审计日志查询：按租户/项目/操作人/动作/目标类型/时间范围分页过滤。""" + """审计日志查询：按组织、操作人、动作、资源、关键字和时间范围分页过滤。""" if not is_admin(current_user): from app.api.v1.endpoints.platform import fail raise fail(403, "admin permission required") @@ -73,6 +75,8 @@ def audit_logs( actor_id=actor_id, action=action, target_type=target_type, + target_id=target_id, + keyword=keyword, start_time=start_time, end_time=end_time, limit=limit, @@ -88,6 +92,8 @@ def audit_logs_export( actor_id: str | None = Query(default=None, description="操作人 ID"), action: str | None = Query(default=None, description="动作类型"), target_type: str | None = Query(default=None, description="目标类型"), + target_id: str | None = Query(default=None, description="目标 ID"), + keyword: str | None = Query(default=None, description="目标 ID 或详情关键字"), start_time: str | None = Query(default=None, description="ISO8601 起始时间"), end_time: str | None = Query(default=None, description="ISO8601 结束时间"), current_user: dict = Depends(get_current_user), @@ -103,6 +109,8 @@ def audit_logs_export( actor_id=actor_id, action=action, target_type=target_type, + target_id=target_id, + keyword=keyword, start_time=start_time, end_time=end_time, limit=10000, diff --git a/docker/app/.env b/docker/app/.env index c5bc101..f2cf2c4 100644 --- a/docker/app/.env +++ b/docker/app/.env @@ -60,5 +60,6 @@ MINIO_ACCESS_KEY=minioadmin MINIO_SECRET_KEY=change_me_minio_secret MINIO_BUCKET=yg-ft-resources MINIO_SECURE=false +MINIO_INLINE_MAX_BYTES=262144 STORAGE_WAIT_SECONDS=300 STORAGE_CHECK_INTERVAL_SECONDS=10 diff --git a/docker/app/docker-compose.yml b/docker/app/docker-compose.yml index f57975b..c1fe42d 100644 --- a/docker/app/docker-compose.yml +++ b/docker/app/docker-compose.yml @@ -70,6 +70,7 @@ services: MINIO_SECRET_KEY: ${MINIO_SECRET_KEY:-minioadmin} MINIO_BUCKET: ${MINIO_BUCKET:-yg-ft-resources} MINIO_SECURE: ${MINIO_SECURE:-false} + MINIO_INLINE_MAX_BYTES: ${MINIO_INLINE_MAX_BYTES:-262144} STORAGE_WAIT_SECONDS: ${STORAGE_WAIT_SECONDS:-300} STORAGE_CHECK_INTERVAL_SECONDS: ${STORAGE_CHECK_INTERVAL_SECONDS:-10} DATA_PROCESS_STORAGE_DIR: ${DATA_PROCESS_STORAGE_DIR:-/data/yg-ft/data-process} diff --git a/docs/governance-user-guide.md b/docs/governance-user-guide.md index 618a39a..cc2052a 100644 --- a/docs/governance-user-guide.md +++ b/docs/governance-user-guide.md @@ -1,9 +1,9 @@ # 平台治理功能使用指南 -> 版本：v1.1 -> 日期：2026-08-13 +> 版本：v1.3 +> 日期：2026-08-19 > 适用版本：YG Fine-Tune Platform v1.0+ -> 更新说明：移除页面权限码设计，改为基于角色的简化权限模型 +> 更新说明：合并组织权限、审批和运行日志入口；取消项目空间菜单但保留旧接口兼容 --- @@ -40,28 +40,27 @@ ### 1.3 入口在哪里？ -所有治理功能集中在左侧导航栏的 **「系统设置」** 和 **「平台治理」** 分组下： +治理、组织和运维功能按职责分布在左侧导航栏的 **「平台治理」**、**「系统设置」** 和 **「算力资源」** 分组下： ``` -系统设置 -├── 用户设置 ← 用户 CRUD + 角色权限 + 密码管理（仅 admin） -├── 平台性能 ← 系统监控 -└── 查看日志 ← 日志查看 - 平台治理 -├── 租户管理 ← 组织/团队（仅 admin） -├── 项目空间 ← 项目级资源隔离（仅 admin） -├── 审批模板 ← 定义哪些操作需要审批（仅 admin） -├── 审批中心 ← 处理待审批请求（仅 admin） -└── 审计日志 ← 查看所有操作记录（仅 admin） +├── 组织与权限 ← 用户与角色、租户与配额（仅 admin） +├── 资源授权 ← 数据集、模型等资源授权（仅 admin） +└── 审批中心 ← 待审批请求与审批策略（仅 admin） + +系统设置 +├── 平台性能 ← 系统资源监控 +└── 运行日志 ← 运行日志、审计记录、操作诊断 算力资源 └── 算力节点 ← GPU 分配与管理（仅 admin） ``` -> ⚠️ 以上菜单**只有 admin 用户能看到**。普通用户登录后不会出现这些入口。 +> ⚠️ 平台治理、资源授权、审批中心和算力节点菜单仅 admin 用户能看到。运行日志入口继续沿用原权限，普通用户可查看系统/训练日志；审计记录和操作诊断页签仅 admin 可见。 > -> **重要变更（v1.1）**：非 admin 用户**默认可以访问所有业务功能菜单**（模型训练、评测、推理、数据集、数据处理等），无需管理员单独分配权限。 +> **重要变更（v1.2）**：非 admin 用户**默认可以访问所有业务功能菜单**（模型训练、评测、推理、数据集、数据处理等），无需管理员单独分配权限；治理和资源管理入口仍仅 admin 可见。 + +> **当前菜单调整（v1.3）**：平台不再提供项目空间菜单和项目级操作入口。历史项目表、接口和旧地址仅作为兼容层保留，当前资源访问以用户所有权、租户边界（如启用）和资源 ACL 为准；新建业务资源不再要求项目字段。 --- @@ -73,24 +72,25 @@ | 用户类型 | 可见菜单 | 说明 | |---------|---------|------| -| **admin（管理员）** | **全部菜单** | 包括用户设置、平台治理、算力节点等管理功能 | -| **非 admin 用户** | **除管理功能外的所有业务菜单** | 模型训练/评测/推理、数据集、数据处理、日志等 | +| **admin（管理员）** | **全部菜单** | 包括组织权限、审批、运行日志、平台治理和算力节点等管理功能 | +| **非 admin 用户** | **除管理功能外的所有业务菜单** | 模型训练/评测/推理、数据集、数据处理等 | > **核心原则**： > - 非 admin 用户**默认拥有所有业务功能的访问权限**，无需单独分配 > - 仅以下功能**仅管理员可见**： -> - `用户设置`（用户 CRUD、角色管理） -> - `平台治理`（租户管理、项目空间、审批模板/中心、审计日志） +> - `平台治理 - 组织与权限`（用户、角色、租户与配额） +> - `平台治理`（资源授权、审批中心） > - `算力节点`（GPU 分配） +> - `运行日志`中的审计记录和操作诊断 > > 资源级别的访问控制通过 **ACL（访问控制列表）** 实现，详见第 4 章。 ### 2.2 创建用户 -**路径**：`用户设置` → `创建用户` +**路径**：`平台治理` → `组织与权限` → `用户与角色` → `创建用户` 1. 以 admin 身份登录平台 -2. 进入「用户设置」页面 +2. 进入「组织与权限」页面的「用户与角色」页签 3. 点击右上角「创建用户」按钮 4. 填写信息： - **账号**：登录用户名（如 `zhangsan`） @@ -107,12 +107,10 @@ | 功能分组 | 包含菜单 | 路由前缀 | |---------|---------|----------| -| 系统设置 - 用户设置 | 用户列表、创建用户、重置密码 | `/user-settings` | -| 平台治理 - 租户管理 | 租户列表、配额设置 | `/tenants` | -| 平台治理 - 项目空间 | 项目列表、成员管理、ACL | `/projects` | -| 平台治理 - 审批模板 | 审批流程定义 | `/approval-templates` | -| 平台治理 - 审批中心 | 待审批请求处理 | `/approval-instances` | -| 平台治理 - 审计日志 | 操作记录查询与导出 | `/audit-logs` | +| 平台治理 - 组织与权限 | 用户、角色、租户与配额 | `/organization` | +| 平台治理 - 资源授权 | 数据集、模型等资源 ACL | `/resource-acl` | +| 平台治理 - 审批中心 | 待审批请求、审批历史与策略 | `/approval-instances` | +| 系统设置 - 运行日志 | 系统/训练日志；管理员可查看审计记录、操作诊断 | `/logs` | | 算力资源 - 算力节点 | GPU 分配与管理 | `/compute` | ### 2.4 重置用户密码 @@ -125,18 +123,18 @@ 3. 输入新密码，确认 **方式二：用户自行修改** -1. 用户登录后在「用户设置」页面点击「修改密码」按钮 +1. 用户登录后在「组织与权限」页面的「用户与角色」页签点击「修改密码」按钮 2. 输入旧密码 + 新密码（至少 6 位） 3. 确认修改 ### 2.5 删除用户 -**路径**：`用户设置` → 用户列表 → 操作列「删除」 +**路径**：`组织与权限` → `用户与角色` → 用户列表 → 操作列「删除」 > ⚠️ 删除用户时会**级联清理**其所有关联数据： > - 该用户创建的数据集、基座模型、微调产物、评测任务 > - 该用户的 ACL 授权记录、GPU 分配记录 -> - 该用户的审批实例、审计日志、项目成员关系、登录会话 +> - 该用户的审批实例、审计日志、历史项目成员关系、登录会话 > - **训练任务保留不删**（避免算力节点上的物理任务数据不一致） --- @@ -148,7 +146,7 @@ 当服务器有多张 GPU 卡（如 8×A800）时，需要指定**哪个用户能用哪张卡**： - 避免两个人同时选同一张卡导致训练冲突 -- 按团队/项目隔离算力资源 +- 按用户和租户边界隔离算力资源 - 控制每个用户的 GPU 配额 ### 3.2 分配 GPU（仅 admin） @@ -259,7 +257,6 @@ curl -X PUT /modelTF/resources/dataset/ds_alpaca_id/acl \ | 删除他人的数据集 | 非 admin 删除别人创建的数据集 | 创建审批实例 或 admin 直接执行 | | 删除他人的模型 | 非 admin 删除别人创建的模型 | 同上 | | 停止他人的训练任务 | 非 admin 停止别人发起的任务 | 同上 | -| 归档/删除项目空间 | 存在待审批变更时 | 拒绝执行 | **核心规则**：admin 做任何操作都直接执行（旁路）；普通用户操作他人资源时进入审批流程。 @@ -305,7 +302,7 @@ curl -X PUT /modelTF/resources/dataset/ds_alpaca_id/acl \ 3. 决策：「通过」或「拒绝」 4. 决策结果自动执行对应操作并记录审计日志 -**审批模板**（`平台治理` → `审批模板`）：定义每种操作需要几步审批、每步谁来审。默认模板都是单步（admin 审批即可）。 +**审批策略**（`平台治理` → `审批中心` → `审批策略`）：定义每种操作需要几步审批、每步谁来审。默认模板都是单步（admin 审批即可）。 --- @@ -324,24 +321,29 @@ curl -X PUT /modelTF/resources/dataset/ds_alpaca_id/acl \ ### 6.2 查询审计日志 -**路径**：`平台治理` → `审计日志` +**路径**：`系统设置` → `运行日志` → `审计记录` 支持筛选条件： | 筛选项 | 说明 | |---|---| -| 操作人 | 按用户 ID 过滤 | -| 动作类型 | 如 `user.create`, `dataset.delete`, `gpu.assign` 等 | -| 目标资源类型 | dataset / model / fine_tune_task 等 | +| 租户 | 按租户名称选择 | +| 操作人 | 按用户名称选择，不需要手工填写用户 ID | +| 动作类型 | 使用中文动作选择，例如创建数据集、删除模型、授予资源权限 | +| 目标资源类型 | 使用中文资源类型选择，例如数据集、模型、训练任务 | +| 关键词 | 模糊搜索目标 ID 或审计详情 | +| 目标 ID | 对指定资源 ID 进行精确查询 | | 时间范围 | 开始时间 ~ 结束时间 | +项目筛选已移除。底层接口仍兼容历史 `project_id` 参数，但当前平台不再提供项目菜单。 + ### 6.3 导出审计日志 -审计日志页面底部有「导出 CSV」按钮，导出的文件包含当前筛选条件下的全部记录，可用于合规审计或问题追溯。 +运行日志的「审计记录」页签提供「导出 CSV」按钮；「操作诊断」页签用于检索失败操作和接口耗时，可用于问题追溯。 ### 6.4 日志保留策略 -审计日志受**留存策略**控制（`平台治理` → 租户管理 → 绑定留存策略）。默认保留 30 天，超期自动清理。 +审计日志受**留存策略**控制（`平台治理` → `组织与权限` → `租户与配额`）。默认保留 30 天，超期自动清理。 --- @@ -351,7 +353,7 @@ curl -X PUT /modelTF/resources/dataset/ds_alpaca_id/acl \ 根据 v1.1 权限模型： 1. **业务菜单**（训练、评测、推理、数据集等）：普通用户**默认全部可见**，无需分配 -2. **管理菜单**（用户设置、租户管理、算力节点等）：**仅 admin 可见**，这是设计如此 +2. **管理菜单**（组织与权限、资源授权、审批中心、算力节点，以及运行日志中的审计/诊断页签）：**仅 admin 可见**，这是设计如此 如果普通用户看不到业务菜单，请检查： - 用户是否正常登录（token 是否有效） @@ -402,7 +404,7 @@ curl -H "Authorization: Bearer platform-token-admin" \ ### Q6: 用户忘记密码怎么办？ 两种方案： -1. **admin 重置**：在「用户设置」→ 用户列表 →「重置密码」 +1. **admin 重置**：在「组织与权限」→「用户与角色」→ 用户列表 →「重置密码」 2. **用户自助修改**：用户登录后点击「修改密码」（需知道旧密码） 如果是完全忘记且不是 admin，只能由 admin 重置。 diff --git a/docs/menu-functional-requirements.md b/docs/menu-functional-requirements.md index e73fdc3..06209fa 100644 --- a/docs/menu-functional-requirements.md +++ b/docs/menu-functional-requirements.md @@ -1,6 +1,8 @@ # 菜单与功能需求总览 > 本文根据当前前端侧边栏、路由、需求文档、接口文档、部署文档和 SQL 脚本整理。当前代码和 SQL 均按正式系统开发基线维护；Mock、Simulator 只能作为显式联调能力，不作为默认开发准则。 +> +> 当前治理版本取消“项目空间”菜单。历史项目表和接口仅保留兼容，不再作为前端业务入口或新资源的必填隔离层。 ## 1. 菜单分层 @@ -17,9 +19,11 @@ | 数据治理 | 数据处理 | `/data-process` | `data-process` | 前端页面已有，后端待完整实现 | 文档上传、切片预览、LLM 生成、结果编辑、发布数据集 | | 其他工具 | 数据类型转换 | `/data-convert` | `data-convert` | 前端页面已有，后端待实现 | JSON/JSONL/Markdown 等格式转换任务 | | 算力资源 | 算力节点 | `/compute` | `compute` | 已接入节点管理接口 | 节点地址、权重、标签、启用状态、GPU、队列、资源副本 | -| 系统设置 | 用户设置 | `/user-settings` | `user-settings` | 已接入基础用户接口 | 用户列表、创建用户、启停、页面权限 | +| 平台治理 | 组织与权限 | `/organization` | `user-settings` | 新增合并入口 | 用户与角色、租户与配额、密码管理 | +| 平台治理 | 资源授权 | `/resource-acl` | `user-settings` | 已接入 ACL 接口 | 数据集、模型等资源授权 | +| 平台治理 | 审批中心 | `/approval-instances` | `user-settings` | 新增合并入口 | 待审批请求、审批历史、审批策略 | | 系统设置 | 平台性能 | `/hardware` | `hardware` | 已有接口，需接真实采集 | CPU、内存、磁盘、GPU、进程、网络监控 | -| 系统设置 | 查看日志 | `/logs` | `logs` | 已有接口，需接真实日志文件 | 后端日志、error 日志、训练日志索引、日志内容查看 | +| 系统设置 | 运行日志 | `/logs` | `logs` | 新增合并入口 | 运行日志、训练日志；管理员可查看审计记录、操作诊断 | ### 1.2 当前二级和隐藏路由 @@ -41,19 +45,17 @@ | 数据集创建/编辑/预览 | `/dataset/create`、`/dataset/:id/edit`、`/dataset/:id/preview` | 数据集管理 | 数据集元数据、文件、版本与内容 | | 自定义工具 | `/tools`、`/tools/create`、`/tools/:id/edit` | 规划入口 | 路由存在，当前侧边栏未展示，后续可归入“其他工具” | | 算力子页 | `/compute/gpus`、`/compute/queue`、`/compute/nodes` | 算力节点 | 当前可作为页签或深链 | -| 创建用户/权限设置 | `/user-settings/create`、`/user-settings/:id/permission` | 用户设置 | 用户创建和页面权限 | +| 组织与权限内部页签 | `/user-settings`、`/tenants`、`/user-settings/create`、`/user-settings/:id/permission` | 平台治理 - 组织与权限 | 旧地址兼容，当前通过页签进入 | +| 项目旧地址 | `/projects`、`/projects/:id` | 兼容跳转 | 跳转到组织与权限，不再展示项目管理 | +| 审批策略旧地址 | `/approval-templates` | 兼容跳转 | 跳转到审批中心的策略页签 | +| 日志旧地址 | `/audit-logs`、`/operation-logs` | 兼容跳转 | 跳转到运行日志对应页签 | | 无权限页 | `/permission-denied` | 系统页 | 路由守卫无权限跳转 | -### 1.3 企业治理待补菜单 +### 1.3 后续治理扩展 | 建议菜单分组 | 菜单 | 建议路由 | 优先级 | 必要性 | | --- | --- | --- | --- | --- | -| 组织与项目 | 租户管理 | `/tenants`、`/tenants/:id` | P0 | 多租户隔离、配额、留存策略入口 | -| 组织与项目 | 项目空间 | `/projects`、`/projects/:id`、`/projects/:id/members` | P0 | 项目级模型/数据集/任务隔离 | -| 组织与项目 | 资源授权 | `/projects/:id/permissions` 或资源详情弹窗 | P0 | 模型/数据集/任务级 ACL | -| 治理中心 | 审批中心 | `/approvals`、`/approvals/:id` | P0 | 删除、发布、导出、停止他人任务等高风险动作 | -| 治理中心 | 审批设置 | `/approval-settings` | P1 | 审批模板、审批人规则、超时策略 | -| 治理中心 | 审计中心 | `/audit-logs`、`/login-logs`、`/download-logs` | P1 | 操作审计、登录审计、下载审计、导出 | +| 系统设置 | 运行日志扩展 | `/logs`、`/login-logs`、`/download-logs` | P1 | 增加登录审计、下载审计、导出审计维度 | | 运维中心 | 存储管理 | `/storage` | P1 | 本地磁盘占用、临时文件、checkpoint 清理、留存 | | 运维中心 | 训练引擎管理 | `/training-engines` | P2 | LLaMA-Factory 和后续引擎能力 schema、健康检查 | | 模型服务 | 模型服务治理 | `/model-services`、`/model-services/:id` | P1 | 测试/生产服务发布、调用统计、下线审批 | @@ -62,7 +64,7 @@ | 菜单/模块 | 主要接口 | 当前运行 SQL | 目标 SQL | | --- | --- | --- | --- | -| 登录、用户设置 | `/modelTF/login`、`/modelTF/me`、`/modelTF/users` | `users` | `users`、`login_sessions`、`permissions`、`role_permissions`、`user_permission_overrides` | +| 登录、组织与权限 | `/modelTF/login`、`/modelTF/me`、`/modelTF/users`、`/modelTF/tenants` | `users`、`tenants` | `users`、`login_sessions`、`permissions`、`role_permissions`、`user_permission_overrides`、`tenants` | | 服务看板 | `/modelTF/dashboard/overview`、`/modelTF/health` | 复用模型/数据集/任务/算力表 | `system_metric_snapshots`、`web_logs`、各业务表聚合 | | 模型管理 | `/modelTF/model-manage`、`/modelTF/model-manage/trained-models`、`/modelTF/model-manage/merge` | `models`、`trained_models` | `models`、`trained_models`、`storage_objects`、`local_import_jobs`、`resource_acl` | | 数据集管理 | `/modelTF/dataset-manage`、`/modelTF/dataset-manage/upload/{id}`、`/preview`、`/versions` | `datasets`、`dataset_files` | `datasets`、`dataset_files`、`dataset_file_versions`、`dataset_records`、`storage_objects` | @@ -70,12 +72,12 @@ | 训练日志 | `/modelTF/training-log-files`、`/modelTF/training-log-content` | 由任务表生成索引 | 日志文件元数据、`fine_tune_metrics`、`audit_logs` | | 算力节点 | `/modelTF/compute/nodes`、`/compute/gpus`、`/compute/queue`、`/compute/nodes/{id}/replicas` | `compute_nodes`、`gpus`、`resource_replicas`、`resource_sync_jobs` | `compute_nodes`、`gpu_devices`、`compute_node_engines`、`compute_jobs`、`resource_replicas`、`resource_sync_jobs` | | 平台性能 | `/modelTF/system-info`、`/modelTF/compute/gpus` | `gpus`、任务表 | `system_metric_snapshots`、`gpu_devices`、`compute_jobs` | -| 查看日志 | `/modelTF/log-files`、`/modelTF/log-content`、`/modelTF/web-log` | 文件日志 | `web_logs`、`audit_logs`，大日志进入日志平台 | +| 运行日志 | `/modelTF/log-files`、`/modelTF/log-content`、`/modelTF/web-log`、`/modelTF/audit-logs` | 文件日志 | `web_logs`、`audit_logs`，大日志进入日志平台 | | 模型评测 | `/modelTF/model-eval`、`/modelTF/dimension` | 当前运行 SQL 未覆盖 | `eval_tasks`、`eval_dimensions`、`eval_sample_results`、`eval_dimension_summaries` | | 模型推理/对比 | `/modelTF/model-compare`、`/modelTF/model-chat/*` | 当前运行 SQL 未覆盖 | `inference_tasks`、`inference_task_models`、`chat_sessions`、`chat_messages` | | 数据处理 | `/modelTF/data-process/*` | 当前运行 SQL 未覆盖 | `data_process_tasks`、`data_process_source_files`、`data_process_preview_items`、`data_process_results` | | 数据转换/自定义工具 | `/modelTF/data-convert/jobs`、`/modelTF/tools` | 当前运行 SQL 未覆盖 | `data_convert_jobs`、`custom_tools` | -| 租户/项目/资源授权 | `/modelTF/tenants`、`/modelTF/projects`、`/modelTF/resources/{type}/{id}/acl` | 当前运行 SQL 未覆盖 | `tenants`、`tenant_users`、`projects`、`project_members`、`resource_acl` | +| 租户/资源授权 | `/modelTF/tenants`、`/modelTF/resources/{type}/{id}/acl` | 当前运行 SQL 未覆盖 | `tenants`、`tenant_users`、`resource_acl`；`projects`、`project_members` 仅作兼容 | | 审批/审计/留存/配额 | `/modelTF/approvals`、`/modelTF/audit-logs`、`/modelTF/retention-policies`、`/modelTF/quotas/usage` | 当前运行 SQL 未覆盖 | `approval_templates`、`approval_instances`、`approval_steps`、`audit_logs`、`retention_policies`、`quotas`、`quota_usage` | ## 3. 文档和脚本检查结论 diff --git a/docs/permissions-design.md b/docs/permissions-design.md index 3797631..8af20be 100644 --- a/docs/permissions-design.md +++ b/docs/permissions-design.md @@ -4,6 +4,8 @@ > 日期：2026-08-02 > 状态：设计基线，供后端实现和前端联调参照 +> **当前菜单基线（2026-08-19）**：平台治理已取消“项目空间”作为用户可见菜单和新资源的业务隔离层。当前前端入口为“组织与权限、资源授权、审批中心”，系统设置下的“运行日志”承载运行日志、审计记录和操作诊断。`projects`、`project_members` 表及相关后端接口仅作历史兼容，不删除、不要求新建资源填写 `project_id`。 + --- ## 目录 @@ -129,7 +131,7 @@ | `compute` | `/compute` | 算力节点 | | `hardware` | `/hardware` | 平台性能 | | `logs` | `/logs`, `/training-log/:id` | 查看日志 | -| `user-settings` | `/user-settings`, `/tenants`, `/projects`, `/approvals`, `/audit-logs` | 系统设置与平台治理 | +| `user-settings` | `/organization`, `/resource-acl`, `/approval-instances`, `/logs` | 平台治理和运行日志（管理员） | --- diff --git a/docs/platform-governance-menu-design.md b/docs/platform-governance-menu-design.md new file mode 100644 index 0000000..0336aeb --- /dev/null +++ b/docs/platform-governance-menu-design.md @@ -0,0 +1,120 @@ +# 平台治理菜单设计与开发计划 + +> 版本：v1.0 +> 日期：2026-08-19 +> 状态：按本文档实施 + +## 1. 设计结论 + +当前项目已经有用户所有权、资源 ACL、租户接口和审计接口，但项目隔离尚未真正落地。核心资源的 `project_id` 当前没有有效业务数据，训练、评测、推理和数据集创建流程也没有统一的项目上下文。 + +因此当前版本取消项目层级设计，资源权限统一采用： + +```text +用户所有权 + 资源 ACL + 租户边界（可选） +``` + +项目相关数据库表和后端接口暂不物理删除，仅作为历史兼容能力保留，后续不再新增项目数据，也不在前端提供项目入口。 + +## 2. 最终菜单 + +```text +平台治理 +├── 组织与权限 +├── 资源授权 +└── 审批中心 + +系统设置 +├── 平台性能 +└── 运行日志 + ├── 系统日志 + ├── 训练日志 + ├── 审计记录 + └── 操作诊断 + +算力资源 +└── 算力节点 +``` + +### 2.1 组织与权限 + +使用页签统一承载： + +- 用户与角色：用户 CRUD、启停、密码、角色。 +- 租户与配额：租户、GPU 配额、存储配额和资源数量配额。 + +用户、租户和配额仍使用独立表和接口，不把组织配额字段混入用户表。单租户部署时可默认停留在“用户与角色”页签。 + +### 2.2 资源授权 + +保留当前资源 ACL 能力，支持数据集、训练模型等资源的 `read/write/execute/download/delete/admin` 权限。项目不再作为授权前置条件。 + +### 2.3 审批中心 + +统一使用页签承载： + +- 待审批/审批历史。 +- 我的申请。 +- 审批策略，仅管理员可见。 + +“审批策略”不再作为独立一级菜单。 + +### 2.4 运行日志 + +在现有系统日志、训练日志基础上增加： + +- 审计记录：写操作、授权、审批、删除、导出等敏感操作。 +- 操作诊断：失败操作、错误类型和接口耗时。 + +“审计中心”不再作为独立菜单。旧的 `/audit-logs` 和 `/operation-logs` 地址保留重定向。 + +## 3. 兼容策略 + +| 原入口 | 新入口/处理方式 | +|---|---| +| `/user-settings` | 重定向到 `/organization?tab=users` | +| `/tenants` | 重定向到 `/organization?tab=tenants` | +| `/approval-templates` | 重定向到 `/approval-instances?tab=strategies` | +| `/audit-logs` | 重定向到 `/logs?tab=audit` | +| `/operation-logs` | 重定向到 `/logs?tab=operations` | +| `/projects` | 移除前端入口；旧地址重定向到组织与权限 | + +后端的租户、项目、审批、ACL、审计 API 暂不删除，保证已有脚本和历史客户端不立即失效。数据库不执行删表操作，也不新增项目字段迁移。 + +## 4. 开发计划 + +### 阶段一：导航和页面聚合 + +1. 新增“组织与权限”聚合页面。 +2. 新增“审批中心”聚合页面。 +3. 扩展“运行日志”页面，加入审计和操作诊断页签。 +4. 调整侧边栏，只展示最终菜单。 + +### 阶段二：兼容旧入口 + +1. 旧用户、租户、审批策略、审计和操作日志路由改为重定向。 +2. 保留原页面组件、API 和后端路由，避免历史调用失效。 +3. 项目路由不再作为业务入口，不再新增项目数据。 + +### 阶段三：权限和功能检查 + +1. 管理员可以访问组织、租户、配额、审批、审计和 ACL。 +2. 普通用户不能访问平台治理菜单；运行日志基础页签继续保持原有访问权限。 +3. 审批策略页签仅管理员可见。 +4. 审计和操作诊断仍保留管理员可见能力。 +5. 数据集、模型、训练、评测、推理继续使用用户所有权和 ACL，不增加项目选择器。 + +### 阶段四：验证 + +- `npm run build`。 +- 检查旧路由重定向。 +- 检查管理员菜单显示。 +- 检查非管理员权限拦截。 +- 检查 Backend 健康接口和前端静态资源。 + +## 5. 暂不处理事项 + +- 不删除 `projects`、`project_members` 表。 +- 不删除后端项目模块，避免历史数据和接口调用中断。 +- 不把租户配额字段直接合并到 `users` 表。 +- 不改变现有数据集、模型、训练、评测、推理的业务接口格式。 diff --git a/docs/当前项目开发进度.md b/docs/当前项目开发进度.md new file mode 100644 index 0000000..06d9aa0 --- /dev/null +++ b/docs/当前项目开发进度.md @@ -0,0 +1,347 @@ +# 当前项目开发进度 + +> 评估基线：2026-08-19 当前工作区代码、数据库初始化脚本、Docker 部署文件、前端页面和现有设计文档。 +> +> 本文以代码实际情况为准。设计文档中已经提出但代码没有形成完整闭环的内容，统一标记为“部分完成”或“未完成”。 + +## 一、项目定位与总体结论 + +当前项目是一个面向多用户、多算力节点的模型训练与推理平台，主要链路为： + +```text +Vue 前端 + | +FastAPI Backend API + |-- PostgreSQL：业务元数据、权限、任务状态、小型内容和预览数据 + |-- Redis：会话、限流、短期缓存和任务辅助状态 + |-- MinIO：模型、数据集、报告和大文件的统一对象存储 + |-- Compute API / Agent：训练、推理、评测、模型合并和 GPU 执行 + | +多台算力节点 +``` + +整体判断： + +| 范围 | 当前状态 | 结论 | +|---|---|---| +| 平台基础架构 | 基本完成 | 前后端、数据库、Redis、MinIO、Compute Agent 和 Docker 部署均已具备 | +| 核心业务闭环 | 基本可用 | 数据集、数据处理、数据转换、训练、模型、推理、评测均有页面和接口 | +| 多算力节点 | 部分完成 | 节点选择、GPU 分配和缓存准备已经接入，跨节点一致性和失败恢复仍需加强 | +| 权限治理 | 部分完成 | 登录、角色、权限码、ACL、审批、审计已实现，但完整的租户/项目隔离尚未闭环 | +| MinIO 统一存储 | 部分完成 | 大文件和模型已接入，仍存在兼容性的本地路径和部分数据双写/回退路径 | +| 生产可靠性 | 未完成 | 缓存容量治理、对象清理、流式上传、归档重试、备份和高可用尚未完成 | +| 前端体验 | 基本可用，需优化 | 构建问题已持续修复，但页面响应等待、首屏体积和部分错误提示仍需优化 | + +## 二、已完成的功能 + +### 2.1 平台基础与部署 + +- 已建立 Vue 3 + TypeScript + Vite 前端工程。 +- 已建立 FastAPI 后端服务，提供登录、平台管理和模型业务接口。 +- 已建立 Compute API / Agent，用于连接算力节点并执行训练、推理、评测和模型处理任务。 +- 已使用 PostgreSQL 保存核心业务数据，Redis 提供会话、限流和缓存能力。 +- 已增加 MinIO 服务及 Backend 的 MinIO 配置，支持和 Compute Agent 分离部署。 +- 已提供 `docker/app`、`docker/compute`、`docker/minio` 和 `docker/offline` 部署目录。 +- 已考虑后端、算力服务、MinIO 分布在不同服务器时使用独立网络；Compute 节点访问 MinIO 需要配置所有节点都能访问的固定 IP 或 DNS。 +- 离线部署目录已经同步后端、算力相关源码和初始化 SQL 的主要改造内容。 + +### 2.2 登录、用户和权限基础 + +- 用户登录、退出、当前用户信息和密码修改接口已经存在。 +- 已有 Token 会话、Redis 会话记录和登录限流逻辑。 +- 已建立用户、角色、权限码和角色权限关系。 +- 已实现管理员、普通用户等基础角色分层。 +- 已实现页面路由守卫、菜单过滤和前端按钮级权限的基础能力。 +- 已建立资源 ACL 管理页面和相关接口，可对用户或角色授予资源级权限。 +- 已建立审批模板、审批实例和审批步骤的基本数据模型与页面。 +- 已建立运行日志、审计日志查询页面及审计记录写入机制。 +- 已加入软删除相关字段和部分删除逻辑，避免直接物理删除业务资源。 + +### 2.3 算力节点与 GPU 资源 + +- 已实现算力节点的新增、编辑、启用、禁用、维护/删除、连通性测试和健康检查。 +- 已实现节点列表、节点详情、节点副本/同步状态和 Compute Agent 连接。 +- 已实现 GPU 信息发现、GPU 状态查询和队列查询。 +- 已建立 `gpu_allocations`、`gpu_assignments`、调度锁等资源分配表。 +- 训练任务已经支持选择调度节点和一张或多张 GPU，并在预检阶段校验资源可用性。 +- Compute Agent 已支持训练、评测、推理和缓存准备等任务接口。 +- 已存在资源副本和同步任务模型，用于记录节点侧资源同步状态。 + +### 2.4 数据集管理 + +- 已实现数据集创建、列表、详情、编辑、删除和文件上传。 +- 已实现数据集文件下载、预览、记录列表和数据记录编辑入口。 +- 已处理 JSON 与 JSONL 的记录数差异：JSON 数组按元素计数，JSONL 按有效行计数，避免把整个 JSON 文件误按行数统计。 +- 已提供数据集版本列表、版本详情、创建版本、切换当前激活版本和删除版本接口。 +- 已增加数据集文件、版本、数据记录等初始化表结构。 +- 已支持数据集文件在数据库小内容和 MinIO 大文件之间按策略存储。 +- 已在训练预检中检查数据集文件是否存在、是否可从 MinIO 获取以及是否能准备到目标算力节点。 + +### 2.5 数据处理与数据转换 + +- 已提供结构化数据、非结构化数据、外部数据源的处理创建流程。 +- 已实现源文件上传、预览、分片/切分、生成、质量检查、去重和结果管理等数据处理流程。 +- 已支持处理结果生成数据集或导入数据集版本。 +- 已建立数据处理任务、源文件、预览项、结果等数据表。 +- 已提供 JSON、JSONL 等数据格式转换页面和后端任务接口。 +- 已将数据转换输出接入 MinIO/数据库分层存储：小型文本结果可存数据库，大文件存 MinIO。 +- 已处理输出文件下载和转换结果元数据保存问题。 + +### 2.6 模型训练 + +- 已实现训练任务创建、配置预检、命令预览、启动、停止、重试和删除。 +- 已接入 LLaMA-Factory 等训练适配逻辑。 +- 已支持选择训练数据、基座模型、算力节点和 GPU。 +- 已实现训练日志获取、训练任务概览、诊断信息、检查点和训练指标查询。 +- 前端训练详情已经具备训练曲线解析和展示逻辑，日志轮询间隔已调整为 3 秒。 +- 已增加 GPU 详情展示入口，包括显存和利用率等 Compute Agent 上报信息。 +- 已支持训练任务的 MinIO 数据准备和目标算力节点缓存准备。 + +### 2.7 模型管理与权重合并 + +- 已实现在线模型/基座模型和训练模型的列表、创建、详情、用途修改和删除。 +- 已建立模型、训练模型、模型血缘、模型产物和导出任务相关表。 +- 已提供权重合并入口，能够根据训练任务准备基座模型和 Adapter，并提交 Compute Agent 执行合并。 +- 已增加模型产物和 MinIO 对象关联字段。 +- 合并结果能够在任务完成后归档到 MinIO 的设计和主要代码路径已经建立。 + +### 2.8 模型推理与模型对比 + +- 已实现推理模型列表、创建、详情和删除入口。 +- 已实现模型加载、卸载、服务启动、服务状态查询和对话调用。 +- 已实现模型对比任务及多模型聊天相关接口。 +- 已增加推理失败重试、停止和资源释放的处理路径。 +- 已支持根据页面选择的算力节点准备模型缓存，兼容训练时所选节点优先的业务要求。 +- Compute Agent 已提供本地推理会话和模型缓存状态接口。 + +### 2.9 模型评测 + +- 已实现评测任务列表、创建、详情和删除。 +- 已实现评测维度管理和评测规则配置页面。 +- 已建立评测任务、评测维度、对比任务等数据库表。 +- 已支持选择模型、数据集、评测维度、算力节点和 GPU 的基础流程。 +- 已接入 Compute Agent 执行评测任务，并保存评测结果和报告相关元数据。 + +### 2.10 数据存储策略 + +- 已建立 `storage_objects`、`storage_cache_jobs` 等 MinIO 元数据和缓存任务表。 +- 已建立 MinIO 对象上传、下载、预签名 URL 和节点缓存准备的主要接口。 +- 已采用分层策略： + - 小型 JSON、JSONL、CSV、任务参数快照、预览数据保留在数据库，降低频繁预览的 MinIO 延迟。 + - 模型权重、训练产物、评测报告和大文件使用 MinIO。 + - 小型 PDF、DOCX、XLSX 仍优先存 MinIO，以保留原始二进制文件内容。 +- 已增加 `data_convert_tasks.output_content`，用于保存小型转换结果，避免所有小结果都依赖 MinIO。 +- Backend 和离线包中的 `000_full_init.sql` 已同步，当前两份初始化脚本内容一致。 + +## 三、部分完成、仍需完善的功能 + +### 3.1 MinIO 统一数据源尚未完全闭环 + +当前 MinIO 已成为模型、大文件和跨节点资源的主存储方向，但仍保留以下兼容路径： + +- Compute Agent 仍有本地文件上传、导入本地模型和扫描本地模型目录的旧接口。 +- 部分历史数据仍使用数据库中的 `content` 或 `output_content` 字段，这是当前已确认的小文件性能策略，不是错误，但必须统一记录来源、大小、校验值和版本。 +- 大文件在部分代码路径中仍通过 `read()` 或 `put_bytes()` 一次性读入内存，未完成流式或分片上传。 +- MinIO 对象和业务资源之间采用多态 `resource_type/resource_id` 关联，数据库没有直接外键，删除和数据一致性需要应用层保证。 +- 删除业务资源后，对应 MinIO 对象的延迟清理、失败重试和孤儿对象扫描尚未形成完整闭环。 + +### 3.2 MinIO 预签名接口的权限边界需要加强 + +当前预签名接口已经存在，但 PUT 上传场景仍需要重点补强： + +- 需要根据资源类型和资源 ID 校验当前用户的写权限，而不应只校验读取权限。 +- 需要服务端生成并校验对象 Key，避免客户端任意写入其他用户或其他资源的对象路径。 +- 需要增加上传完成确认接口，校验对象实际存在、大小和校验值后再写入业务表。 +- 需要限制允许的 Bucket、Content-Type、大小和有效期。 +- 需要记录预签名创建、上传完成、失败和过期事件，便于审计。 + +### 3.3 激活版本和跨节点资源版本仍需加强 + +数据集已经有 `active_version_id` 和版本表，但以下场景仍需补充： + +- 训练、推理和评测必须只使用资源当前激活版本，并在任务创建时固化版本 ID。 +- 同一文件名的不同版本不能只依靠文件名同步，应使用资源 ID、版本 ID 和对象 Key 组成唯一定位。 +- 已创建任务在后续切换激活版本后，不能被意外切换到新版本。 +- 需要为每个准备到算力节点的资源保存版本、对象 ETag/校验值和本地路径清单。 +- 历史版本的数据库内容回退和 MinIO 对象回退逻辑还需要补全并增加测试。 + +### 3.4 权限 2.0 尚未完全落地 + +已有用户、角色、权限码、ACL、审批和审计基础，但仍存在以下差距： + +- 租户、用户、资源、算力节点、模型、数据集之间的隔离规则没有全部在 SQL 查询层统一执行。 +- 项目空间设计已经讨论过取消，但数据库中仍保留 `projects`、`project_members` 等历史结构，需要明确兼容策略和最终迁移方式。 +- 训练创建的模型、数据集和训练任务之间的联合权限约束还没有完全统一。 +- 评测、推理、模型合并、导出、缓存准备等动作需要逐一校验资源读权限和操作权限。 +- 前端按钮权限已经有基础实现，但不能替代后端鉴权；仍需要对所有关键动作进行后端默认拒绝校验。 +- 审批拦截范围、管理员豁免规则和跨租户资源访问规则需要形成可执行矩阵。 + +### 3.5 模型合并、导出和评测报告闭环不足 + +- 权重合并前自动准备 Base Model 和 Adapter 的主要路径已建立，但失败时的清理、重试和幂等性仍需加强。 +- 合并结果归档到 MinIO 的逻辑主要依赖任务完成轮询，服务重启或轮询中断时可能需要补偿扫描。 +- 模型导出任务目前有查询模型和表结构，但完整的创建、执行、进度、失败重试和下载闭环尚未完成。 +- 评测结果和报告字段已经存在，但报告对象归档、报告下载、报告版本和报告与任务的稳定关联仍需验证。 +- 评测指标配置和执行器返回指标之间仍需要强类型映射，避免前端显示为通用的 `custom`。 + +### 3.6 GPU 资源分配需要统一到所有任务类型 + +- 训练已经有较完整的节点/GPU 选择和预检流程。 +- 推理和评测已经出现节点选择、缓存准备和 GPU 选择的接入代码，但还需要确认从页面选择到 Compute Agent 启动参数、进程环境变量和释放逻辑的全链路生效。 +- 需要防止同一张 GPU 被多个任务绕过调度锁重复占用。 +- 需要处理服务异常退出、Backend 重启、Compute Agent 重启后的分配回收和状态对账。 +- 训练详情中的显存使用量、GPU 使用率等指标依赖 Compute Agent 上报，仍需要校验采样时间、单位、空值和任务对应关系。 + +## 四、尚未完成的功能 + +以下功能在当前代码中没有形成可验收的完整闭环，或仍处于设计/基础代码阶段： + +1. **完整的租户隔离和资源继承模型**：所有列表、详情、下载、缓存、训练、推理、评测和导出接口都需要统一的租户范围过滤。 +2. **项目取消后的正式数据迁移方案**：需要决定历史项目数据如何归属到用户或租户，并提供一次性迁移脚本和回滚方案。 +3. **预签名上传完成确认和对象校验**：包括 Key 白名单、ACL、大小限制、哈希/ETag 和状态回写。 +4. **MinIO 对象生命周期管理**：软删除后的延迟删除、失败重试、孤儿对象扫描、对象引用检查和管理员清理入口。 +5. **Compute Agent 缓存治理**：容量上限、LRU/TTL、运行任务保护、磁盘占用监控、缓存清单和版本校验。 +6. **统一的资源归档编排器**：训练、合并、评测和推理相关产物需要支持断点恢复、幂等重试和服务重启补偿。 +7. **模型导出完整流程**：导出任务创建、格式/量化参数、进度、失败重试、MinIO 归档和下载权限。 +8. **流式和分片文件传输**：避免大文件上传、下载和对象复制时将完整内容读入 Backend 或 Compute Agent 内存。 +9. **生产级 MinIO 安全和高可用**：默认密钥替换、TLS、网络访问控制、管理员 Console 隔离、容量监控、备份和恢复。 +10. **完整的端到端测试和持续集成**：至少覆盖单节点、多节点、多 GPU、跨用户、跨租户、版本切换、MinIO 不可用和服务重启恢复。 +11. **统一数据库迁移体系**：当前初始化 SQL 适合新库初始化，但尚未替代正式的版本化迁移工具；已有数据库更新仍需要明确迁移脚本和执行记录。 + +## 五、需要优化的功能 + +### 5.1 后端响应性能 + +- 页面列表接口需要避免每条记录重复查询用户、资源、MinIO 元数据和 Compute 节点状态。 +- MinIO 的 Bucket 检查、对象 Head 和预签名生成应使用连接复用、短期缓存和批量查询。 +- 训练、推理、评测页面不应通过过短间隔轮询大量详情接口，应按任务状态动态退避，并在完成后停止轮询。 +- 对 dashboard、节点健康、GPU 状态等高频数据应区分实时数据和缓存数据。 +- 后端日志轮询和健康检查日志需要继续降噪，仅在状态变化、失败或达到较长周期时输出。 + +### 5.2 前端加载和交互 + +- 列表页面应区分首屏 loading、刷新 loading、操作 loading，避免整页长时间无反馈。 +- 推理、评测、训练详情应使用统一的任务状态刷新策略和超时提示。 +- 前端仍有 FontAwesome 在线资源解析警告，应清理对外部网络文件的依赖，保证离线环境打开速度。 +- 应继续拆分首屏大体积 chunk，并减少一次性加载不相关页面组件。 +- GPU 选择组件需要明确显示空闲、占用、不可达、预留和已分配状态。 +- 错误提示应携带资源名称、节点名称、版本和下一步处理建议，减少只显示 500/404 的情况。 + +### 5.3 训练、推理和评测可靠性 + +- 所有任务创建前应执行同一套资源权限、版本存在性、MinIO 可用性和 GPU 原子分配校验。 +- 任务创建接口应支持幂等键，避免前端重复点击造成重复任务。 +- 节点不可达时应快速失败或进入可见的等待状态，不能让页面长时间无反馈。 +- 失败重试应区分网络瞬时失败、资源不足、模型文件缺失、参数错误和执行器失败。 +- 任务停止后必须释放 GPU 分配、推理端口、缓存锁和临时目录。 + +### 5.4 数据和模型一致性 + +- 每个对象都应保存大小、校验值、版本 ID、来源、创建者、租户和引用状态。 +- 数据库中的小文件内容和 MinIO 对象不能同时被当作可独立修改的主副本；需要明确唯一写入入口。 +- 数据集激活版本变更需要留下审计记录，并影响后续任务创建但不改变已创建任务。 +- 模型权重、Adapter、合并结果和导出结果需要形成完整血缘关系。 + +## 六、数据库和初始化脚本状态 + +当前 `backend/app/db/sql/000_full_init.sql` 已包含以下主要类别： + +- 用户、模型、训练模型、模型血缘、模型产物、模型导出任务。 +- 数据集、数据集文件、数据集版本、数据集记录。 +- 算力节点、GPU、GPU 分配、调度锁、Compute Job。 +- 资源副本、资源同步任务、MinIO 对象、缓存任务。 +- 评测任务、评测维度、模型对比任务。 +- 租户、项目兼容表、项目成员、角色、会话、ACL。 +- 审批模板、审批实例、审批步骤、审计日志、留存策略。 +- 数据处理任务、源文件、预览项、处理结果、数据转换任务。 + +已确认的近期字段包括： + +- `model_artifacts.storage_object_id` +- `model_artifacts.storage_backend` +- `dataset_files.storage_object_id` +- `data_convert_tasks.output_content` +- `data_convert_tasks.output_storage_object_id` +- `data_convert_tasks.storage_backend` +- `eval_tasks.report_storage_object_id` + +离线包中的 `docker/offline/src/backend/app/db/sql/000_full_init.sql` 应与主工程初始化脚本保持同步。需要注意： + +- 初始化 SQL 主要用于新数据库或新数据卷；已有数据库不能仅靠重启容器自动获得全部新字段。 +- 生产/测试数据库需要执行可追踪的迁移脚本，并在迁移前备份或生成结构快照。 +- `ensure_schema` 类运行时补字段逻辑只能作为兼容兜底，不能替代正式迁移。 +- 后续如果正式移除项目设计，需要先完成数据归属迁移，再决定是否删除历史表，不能直接从初始化 SQL 中删除表。 + +## 七、当前验证结果 + +已完成的静态和局部验证： + +- Backend 和离线 Backend 源码 `compileall` 检查通过。 +- MinIO 分层策略冒烟验证通过：小型 JSON/JSONL 可落数据库，小型二进制和超过阈值的内容进入 MinIO。 +- 主工程和离线包初始化 SQL 已做同步检查，内容一致。 +- 前端此前已完成 `npm run build` 类型错误修复，构建剩余问题主要是非阻断的资源/分包警告。 +- 已对训练日志、数据集 JSON/JSONL 统计、MinIO 资源准备等重点链路进行过问题修复。 + +当前不能据此宣称“全量功能测试通过”： + +- 现有部分自动化测试仍保留旧的本地文件或旧 MinIO 行为假设，需要按当前分层存储策略更新。 +- WSL Docker 运行时验证受当前环境的 `E_ACCESSDENIED` 影响，不能在本次文档生成时完成全部容器健康、数据库字段和跨节点测试。 +- 多节点、多 GPU、MinIO 临时不可用、服务重启恢复和跨用户权限测试仍需要在可用运行环境中执行。 + +## 八、下一阶段开发计划 + +### P0：安全与数据正确性 + +1. 完善 MinIO 预签名 PUT 的资源写权限、对象 Key 白名单、大小/类型限制和上传完成确认。 +2. 统一任务创建时的资源版本固化，训练、推理、评测只使用已授权的激活版本快照。 +3. 逐一补齐评测、推理、模型合并、模型导出、缓存准备的后端权限校验和审计记录。 +4. 完成租户隔离查询范围，清理或兼容历史项目字段，补充数据迁移脚本。 + +### P1：跨节点可靠性 + +1. 建立资源清单/manifest，记录 MinIO 对象版本、校验值、目标节点路径和缓存状态。 +2. 完善训练、合并、评测和推理的准备、执行、归档、失败重试和服务重启补偿。 +3. 完善 GPU 原子分配、异常回收、节点重连对账和任务释放。 +4. 增加缓存容量、TTL/LRU、运行任务保护和磁盘占用监控。 +5. 增加 MinIO 对象引用清理、孤儿对象扫描和软删除回收任务。 + +### P2：性能与用户体验 + +1. 优化页面列表接口和高频轮询，采用批量查询、短期缓存和动态退避。 +2. 将大文件上传/下载/复制改为流式或分片传输。 +3. 统一前端任务状态组件、loading、超时、重试和错误诊断信息。 +4. 处理前端离线资源警告，继续拆分首屏 chunk。 +5. 统一 GPU 状态展示及训练指标采样时间、单位和空值处理。 + +### P3：工程化和上线准备 + +1. 建立正式数据库版本迁移机制和离线升级脚本。 +2. 增加 CI：前端类型检查/构建、Backend 单元测试、Compute Agent 测试、SQL 新库初始化测试。 +3. 增加多节点端到端测试和 MinIO 故障注入测试。 +4. 完善 MinIO TLS、密钥管理、网络隔离、监控、备份和恢复方案。 +5. 建立生产运行手册，包括首次部署、升级、回滚、数据库迁移、对象清理和故障处理。 + +## 九、阶段验收标准 + +完成下一阶段后，至少应满足： + +- 用户只能看到和操作其所属租户授权的模型、数据集、训练任务、推理服务和评测任务。 +- 任何任务创建都能明确记录用户、租户、资源版本、算力节点、GPU 列表和 MinIO 对象版本。 +- 同一个节点的同一张 GPU 不能被两个活动任务同时分配。 +- MinIO 临时不可用时，任务进入可解释的等待/失败状态，并能按策略重试，页面不会无限等待。 +- Backend 或 Compute Agent 重启后，任务、缓存、GPU 分配和归档状态可以对账恢复。 +- 训练、合并、评测和推理产物都能在 MinIO 中找到，并且可以通过权限校验后的接口下载或使用。 +- 删除资源后不会继续出现在普通列表中，关联对象能够按引用状态延迟清理并留下审计记录。 +- 新数据库初始化和已有数据库迁移后，所有业务接口不再因为缺表或缺字段启动失败。 +- 离线部署不依赖外部字体、图标或 CDN，前端首屏和核心业务操作在无网络环境下可用。 + +## 十、相关文件索引 + +- 平台架构：[platform-architecture-requirements.md](./platform-architecture-requirements.md) +- 权限设计：[permissions-design.md](./permissions-design.md) +- MinIO 与 Compute 缓存方案：[minio-compute-cache-plan.md](./minio-compute-cache-plan.md) +- 平台治理菜单设计：[platform-governance-menu-design.md](./platform-governance-menu-design.md) +- 数据处理设计：[data-process-design.md](./data-process-design.md) +- 数据库初始化脚本：[../backend/app/db/sql/000_full_init.sql](../backend/app/db/sql/000_full_init.sql) +- 离线部署目录：[../docker/offline](../docker/offline) + diff --git a/frontend/src/api/modules/audit.ts b/frontend/src/api/modules/audit.ts index 9999b1e..511d4db 100644 --- a/frontend/src/api/modules/audit.ts +++ b/frontend/src/api/modules/audit.ts @@ -20,6 +20,8 @@ export interface AuditQuery { actor_id?: string action?: string target_type?: string + target_id?: string + keyword?: string start_time?: string end_time?: string limit?: number diff --git a/frontend/src/components/AppSidebar.vue b/frontend/src/components/AppSidebar.vue index 7d188fb..64c8e8e 100644 --- a/frontend/src/components/AppSidebar.vue +++ b/frontend/src/components/AppSidebar.vue @@ -15,6 +15,10 @@ const activeMenu = computed(() => { if (seg === 'training-log') return 'fine-tune' // 维度管理归到模型评测 if (route.path.includes('model-eval/dimension')) return 'model-eval' + // 组织与权限承接用户、租户和历史治理入口 + if (route.path.startsWith('/organization') || route.path.startsWith('/user-settings') || route.path.startsWith('/tenants')) return 'organization' + // 运行日志承接审计和操作诊断两个历史入口 + if (route.path.startsWith('/logs') || route.path.startsWith('/audit-logs') || route.path.startsWith('/operation-logs')) return 'logs' // 对比对话归到模型推理 if (route.path.startsWith('/model-compare/chat')) return 'model-inference' // 合并权重归到模型管理 @@ -77,21 +81,16 @@ const menuGroups: MenuGroup[] = [ { title: '平台治理', items: [ - { key: 'tenants', label: '租户管理', icon: 'fa-building', to: '/tenants', permission: 'user-settings' }, - { key: 'projects', label: '项目空间', icon: 'fa-folder', to: '/projects', permission: 'user-settings' }, + { key: 'organization', label: '组织与权限', icon: 'fa-users', to: '/organization', permission: 'user-settings' }, { key: 'resource-acl', label: '资源授权', icon: 'fa-key', to: '/resource-acl', permission: 'user-settings' }, - { key: 'audit-logs', label: '审计日志', icon: 'fa-history', to: '/audit-logs', permission: 'user-settings' }, - { key: 'operation-logs', label: '操作日志', icon: 'fa-list', to: '/operation-logs', permission: 'user-settings' }, - { key: 'approval-templates', label: '审批模板', icon: 'fa-list-alt', to: '/approval-templates', permission: 'user-settings' }, { key: 'approval-instances', label: '审批中心', icon: 'fa-check-square', to: '/approval-instances', permission: 'user-settings' }, ], }, { title: '系统设置', items: [ - { key: 'user-settings', label: '用户设置', icon: 'fa-users', to: '/user-settings', permission: 'user-settings' }, { key: 'hardware', label: '平台性能', icon: 'fa-bar-chart', to: '/hardware', permission: 'hardware' }, - { key: 'logs', label: '查看日志', icon: 'fa-file-text', to: '/logs', permission: 'logs' }, + { key: 'logs', label: '运行日志', icon: 'fa-file-text', to: '/logs', permission: 'logs' }, ], }, ] @@ -101,9 +100,9 @@ const menuGroups: MenuGroup[] = [ * * 1. admin 用户：可以看到所有菜单 * 2. 非 admin 用户： - * - 默认可见所有业务菜单（模型训练、评测、推理、数据集、数据处理、转换、性能、日志等） + * - 默认可见所有业务菜单（模型训练、评测、推理、数据集、数据处理、转换、性能、运行日志等） * - 仅以下菜单对非 admin 不可见： - * - user-settings（用户设置、租户管理、项目空间、审批模板/中心、审计日志） + * - user-settings（组织与权限、资源授权、审批；运行日志中的审计/诊断页签） * - compute（算力节点/GPU 分配） * * 注意：移除了旧的权限码（permission code）过滤逻辑， diff --git a/frontend/src/router/index.ts b/frontend/src/router/index.ts index 710987a..ede4f42 100644 --- a/frontend/src/router/index.ts +++ b/frontend/src/router/index.ts @@ -32,11 +32,17 @@ const routes: RouteRecordRaw[] = [ meta: { title: '服务看板' }, }, // 平台治理 + { + path: 'organization', + name: 'organization', + component: () => import('@/views/governance/OrganizationPermissionView.vue'), + meta: { title: '组织与权限', pageSurface: 'self', permission: 'user-settings' }, + }, { path: 'tenants', name: 'tenants', - component: () => import('@/views/tenants/TenantListView.vue'), - meta: { title: '租户管理', permission: 'user-settings' }, + redirect: '/organization?tab=tenants', + meta: { title: '租户与配额', permission: 'user-settings' }, }, { path: 'tenants/:id', @@ -47,37 +53,37 @@ const routes: RouteRecordRaw[] = [ { path: 'projects', name: 'projects', - component: () => import('@/views/projects/ProjectListView.vue'), - meta: { title: '项目空间', permission: 'user-settings' }, + redirect: '/organization?tab=users', + meta: { title: '组织与权限', permission: 'user-settings' }, }, { path: 'projects/:id', name: 'project-detail', - component: () => import('@/views/projects/ProjectDetailView.vue'), - meta: { title: '项目详情', permission: 'user-settings' }, + redirect: '/organization?tab=users', + meta: { title: '组织与权限', permission: 'user-settings' }, }, { path: 'audit-logs', name: 'audit-logs', - component: () => import('@/views/audit/AuditLogView.vue'), - meta: { title: '审计日志', permission: 'user-settings' }, + redirect: '/logs?tab=audit', + meta: { title: '运行日志', permission: 'user-settings' }, }, { path: 'operation-logs', name: 'operation-logs', - component: () => import('@/views/audit/OperationLogView.vue'), - meta: { title: '操作日志', permission: 'user-settings' }, + redirect: '/logs?tab=operations', + meta: { title: '运行日志', permission: 'user-settings' }, }, { path: 'approval-templates', name: 'approval-templates', - component: () => import('@/views/approvals/ApprovalTemplateView.vue'), - meta: { title: '审批模板', permission: 'user-settings' }, + redirect: '/approval-instances?tab=strategies', + meta: { title: '审批中心', permission: 'user-settings' }, }, { path: 'approval-instances', name: 'approval-instances', - component: () => import('@/views/approvals/ApprovalInstanceView.vue'), + component: () => import('@/views/approvals/ApprovalCenterView.vue'), meta: { title: '审批中心', permission: 'user-settings' }, }, { @@ -302,14 +308,14 @@ const routes: RouteRecordRaw[] = [ { path: 'logs', name: 'logs', - component: () => import('@/views/system/LogsView.vue'), - meta: { title: '查看日志' }, + component: () => import('@/views/system/RuntimeLogsView.vue'), + meta: { title: '运行日志', pageSurface: 'self', permission: 'logs' }, }, { path: 'user-settings', name: 'user-settings', - component: () => import('@/views/system/UserSettingsView.vue'), - meta: { title: '用户设置', pageSurface: 'self', permission: 'user-settings' }, + redirect: '/organization?tab=users', + meta: { title: '组织与权限', pageSurface: 'self', permission: 'user-settings' }, }, { path: 'user-settings/create', @@ -357,6 +363,7 @@ const permissionBySegment: Record = { tools: 'data-convert', hardware: 'hardware', logs: 'logs', + organization: 'user-settings', 'user-settings': 'user-settings', tenants: 'user-settings', projects: 'user-settings', @@ -379,7 +386,7 @@ function requiredPermission(path: string, explicit?: unknown) { // 权限控制规则（基于 governance-user-guide.md 设计）： // - admin 用户：可以访问所有页面 // - 非 admin 用户：默认可访问所有业务页面（训练、评测、推理、数据等） -// 仅以下页面限制 admin 访问：user-settings、compute（算力节点） +// 仅治理与资源管理页面限制 admin 访问：organization、user-settings、compute router.beforeEach((to, _from, next) => { if (!to.meta.public) routeLoading.value = true const auth = useAuthStore() @@ -404,7 +411,7 @@ router.beforeEach((to, _from, next) => { if (!to.meta.skipPermission) { const permission = requiredPermission(to.path, to.meta.permission) // 仅限制管理员专属页面的访问权限 - // user-settings（用户设置、租户管理、项目空间、审批、审计日志）仅 admin 可访问 + // user-settings（组织与权限、资源授权、审批中心、运行日志）仅 admin 可访问 if (permission === 'user-settings' && !auth.isAdmin) { next({ name: 'permission-denied', replace: true }) return diff --git a/frontend/src/views/approvals/ApprovalCenterView.vue b/frontend/src/views/approvals/ApprovalCenterView.vue new file mode 100644 index 0000000..dd6817b --- /dev/null +++ b/frontend/src/views/approvals/ApprovalCenterView.vue @@ -0,0 +1,66 @@ + + + + + diff --git a/frontend/src/views/approvals/ApprovalInstanceView.vue b/frontend/src/views/approvals/ApprovalInstanceView.vue index 4664758..dcb225d 100644 --- a/frontend/src/views/approvals/ApprovalInstanceView.vue +++ b/frontend/src/views/approvals/ApprovalInstanceView.vue @@ -1,11 +1,15 @@