完善部分平台治理功能,及修改看板缺陷

This commit is contained in:
wangjiming
2026-08-10 11:41:16 +08:00
parent 75cc105ebc
commit f809825a7d
18 changed files with 1873 additions and 362 deletions

View File

@@ -1,4 +1,4 @@
from __future__ import annotations
from __future__ import annotations
import json
import uuid
@@ -442,16 +442,14 @@ async def dashboard_stats() -> dict[str, Any]:
running_statuses = {"syncing", "queued", "running"}
running_ft = [t for t in tasks if t.get("status") in running_statuses]
failed_ft = [t for t in tasks if t.get("status") == "failed"]
all_ft = tasks # 全部训练任务(含已完成/异常)
online_nodes = [n for n in nodes if n.get("scheduler_status") == "online"]
# 评测中运行的任务
running_eval = [e for e in eval_tasks if e.get("status") in running_statuses]
# 数据处理中运行的任务
# 评测中运行的任务
eval_running = 0
try:
dp_running = int(dp_store.list_tasks(page=1, page_size=1, status="running").get("total", 0))
eval_tasks = store.eval_tasks()
eval_running = len([e for e in eval_tasks if e.get("status") in running_statuses])
except Exception:
dp_running = 0
eval_running = 0
# 近 7 天训练统计(按创建日期分桶)
now = datetime.now(timezone.utc)
@@ -472,45 +470,30 @@ async def dashboard_stats() -> dict[str, Any]:
}
)
# 服务状态 —— 通过对应接口连通性判断是否正常
# 服务状态 —— 每个服务的"实例数"含义:
# 模型训练 → 训练任务总数
# 模型评测 → 评测任务总数
# 模型推理 → 推理/对比任务实例数
# 模型管理 → 基座模型注册总数
# 数据集管理 → 数据集总数
# 数据处理 → 数据处理任务总数
# 数据类型转换 → 数据转换任务总数
service_checks = [
("模型训练", "/fine-tune", "模型训练"),
("模型评测", "/model-eval", "模型评测"),
("模型推理", "/model-inference", "模型推理"),
("模型管理", "/model-manage", "模型管理"),
("数据集管理", "/dataset-manage", "数据集管理"),
("数据处理", "/data-process", "数据处理"),
("数据类型转换", "/data-convert", "数据类型转换"),
("模型训练", "fine-tune", len(tasks)),
("模型评测", "model-eval", len(eval_tasks)),
("模型推理", "model-inference", len(store.compare_tasks())),
("模型管理", "model-manage", len(store.models())),
("数据集管理", "dataset-manage", len(datasets)),
("数据处理", "data-process", dp_count),
("数据类型转换", "data-convert", dp_count),
]
service_status = []
for svc_type, _path, _label in service_checks:
try:
svc_count = 0
if svc_type == "模型训练":
svc_count = len(tasks)
elif svc_type == "模型评测":
svc_count = len(eval_tasks)
elif svc_type == "模型推理":
svc_count = len(online_nodes)
elif svc_type == "模型管理":
svc_count = len(store.models())
elif svc_type == "数据集管理":
svc_count = len(datasets)
elif svc_type == "数据处理":
svc_count = dp_count
elif svc_type == "数据类型转换":
svc_count = dp_count
service_status.append({
"type": svc_type,
"status": "normal",
"count": svc_count,
})
except Exception:
service_status.append({
"type": svc_type,
"status": "error",
"count": 0,
})
for svc_type, _path, svc_count in service_checks:
service_status.append({
"type": svc_type,
"status": "normal",
"count": svc_count,
})
# 训练任务状态归一化
status_map = {
@@ -590,12 +573,16 @@ async def dashboard_stats() -> dict[str, Any]:
]
# 登录时长排行(本月),只取 top 5
login_duration_rank = store.login_duration_rank(limit=5)
login_duration_rank = []
try:
login_duration_rank = store.login_duration_rank(limit=5)
except Exception:
pass
return ok(
{
"online_services": sum(s["count"] for s in service_status),
"running_tasks": len(running_ft) + len(running_eval) + dp_running,
"running_tasks": len(running_ft) + eval_running,
"pending_alerts": 0,
"training_7d": training_7d,
"service_status": service_status,
@@ -656,6 +643,29 @@ async def reset_user_password(
raise fail(400, str(exc))
@router.post("/users/me/password")
async def change_my_password(
payload: dict[str, Any] = Body(...),
current_user: dict = Depends(get_current_user),
) -> dict[str, Any]:
"""用户自行修改密码:验证旧密码后设置新密码。"""
old_password = payload.get("old_password") or ""
new_password = payload.get("new_password") or ""
if not old_password or not new_password:
raise fail(400, "old_password and new_password are required")
if len(new_password) < 6:
raise fail(400, "new password must be at least 6 characters")
try:
success = get_platform_store().change_password(
current_user["id"], old_password, new_password
)
except KeyError:
raise fail(404, "user not found")
if not success:
raise fail(400, "old password is incorrect")
return ok({"changed": True})
@router.get("/model-manage/local-models")
async def local_models() -> dict[str, Any]:
store = get_platform_store()
@@ -685,8 +695,15 @@ async def local_models() -> dict[str, Any]:
@router.get("/model-manage/trained-models")
async def trained_models() -> dict[str, Any]:
return ok({"models": get_platform_store().trained_models()})
async def trained_models(current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
all_models = get_platform_store().trained_models()
if is_admin(current_user):
return ok({"models": all_models})
# 普通用户只能看到自己创建的 + ACL 授权的
user_id = current_user.get("id")
accessible = set(filter_accessible_resource_ids("trained_model", [m["id"] for m in all_models], current_user))
result = [m for m in all_models if m.get("created_by") == user_id or m["id"] in accessible]
return ok({"models": result})
@router.delete("/model-manage/trained-models/{model_id}")
@@ -720,16 +737,13 @@ async def model_by_name(name: str) -> dict[str, Any]:
@router.get("/model-manage")
async def model_list(current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
models = get_platform_store().models()
if current_user.get("role") == "admin" or current_user.get("protected"):
return ok(models)
# 普通用户只返回有 ACL 授权的模型
accessible = set(filter_accessible_resource_ids("model", [m["id"] for m in models], current_user))
return ok([m for m in models if m["id"] in accessible])
# 基座模型是平台共享资源,所有登录用户均可查看
return ok(get_platform_store().models())
@router.post("/model-manage")
async def create_model(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
async def create_model(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
payload.setdefault("created_by", current_user.get("id"))
try:
return ok(get_platform_store().create_model(payload))
except KeyError as exc:
@@ -1047,14 +1061,18 @@ async def download_dataset_file(dataset_id: str, file_id: str, version_id: str |
@router.get("/dataset-manage")
async def dataset_list(current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
datasets = get_platform_store().datasets()
if current_user.get("role") == "admin" or current_user.get("protected"):
if is_admin(current_user):
return ok(datasets)
# 普通用户可见:自己创建的 + ACL 授权的
user_id = current_user.get("id")
accessible = set(filter_accessible_resource_ids("dataset", [d["id"] for d in datasets], current_user))
return ok([d for d in datasets if d["id"] in accessible])
result = [d for d in datasets if d.get("created_by") == user_id or d["id"] in accessible]
return ok(result)
@router.post("/dataset-manage")
async def create_dataset(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
async def create_dataset(payload: dict[str, Any] = Body(...), current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
payload.setdefault("created_by", current_user.get("id"))
dataset = get_platform_store().create_dataset(payload)
return ok({"id": dataset["id"]})
@@ -1127,8 +1145,20 @@ async def create_fine_tune(payload: dict[str, Any] = Body(...)) -> dict[str, Any
@router.post("/fine-tune/start")
async def start_fine_tune(payload: dict[str, Any] = Body(...)) -> dict[str, Any]:
async def start_fine_tune(
payload: dict[str, Any] = Body(...),
current_user: dict = Depends(get_current_user),
) -> dict[str, Any]:
store = get_platform_store()
# GPU 权限校验:普通用户只能使用被分配的 GPU
if not is_admin(current_user):
node_id = payload.get("compute_node_id") or payload.get("node_id")
gpu_indices = payload.get("gpus") or []
if node_id and gpu_indices:
if not store.check_gpu_access(current_user["id"], node_id, gpu_indices):
raise fail(403, "无权使用所选 GPU请联系管理员分配")
# 记录创建者
payload.setdefault("created_by", current_user.get("id"))
try:
return ok(await _submit_fine_tune_task(store, payload))
except KeyError:
@@ -1267,10 +1297,14 @@ async def update_fine_tune(task_id: str, payload: dict[str, Any] = Body(...)) ->
@router.post("/fine-tune/stop/{task_id}")
async def stop_fine_tune(task_id: str) -> dict[str, Any]:
async def stop_fine_tune(task_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
store = get_platform_store()
try:
task = store.task(task_id)
# 审批拦截:非 admin 停止他人任务需审批
pending = _require_approval_or_admin("fine_tune_task", task_id, current_user, f"停止训练任务 {task_id}")
if pending:
return pending
node = _node_for_task(task)
if task.get("compute_job_id") and node and get_settings().compute_mode != "simulator":
job = await ComputeNodeClient(node["api_base_url"]).stop_job(task["compute_job_id"])
@@ -1281,8 +1315,8 @@ async def stop_fine_tune(task_id: str) -> dict[str, Any]:
@router.post("/fine-tune/{task_id}/stop")
async def stop_fine_tune_alt(task_id: str) -> dict[str, Any]:
return await stop_fine_tune(task_id)
async def stop_fine_tune_alt(task_id: str, current_user: dict = Depends(get_current_user)) -> dict[str, Any]:
return await stop_fine_tune(task_id, current_user)
@router.post("/fine-tune/{task_id}/retry")

View File

@@ -1,4 +1,4 @@
from fastapi import APIRouter
from fastapi import APIRouter
from app.api.v1.endpoints.data_process import router as data_process_router
from app.api.v1.endpoints.platform import router as platform_router
@@ -9,6 +9,7 @@ from app.modules.approval.router import router as approval_router
from app.modules.system.router import router as system_router
from app.modules.retention.router import router as retention_router
from app.modules.resource.router import router as resource_router
from app.modules.gpu.router import router as gpu_router
from app.modules.data_convert.router import router as data_convert_router
api_router = APIRouter()
@@ -21,4 +22,5 @@ api_router.include_router(project_router, tags=["project"])
api_router.include_router(approval_router, tags=["approval"])
api_router.include_router(retention_router, tags=["retention"])
api_router.include_router(resource_router, tags=["resource"])
api_router.include_router(gpu_router, tags=["gpu-assignment"])
api_router.include_router(data_convert_router, tags=["data-convert"])