DataOps-platform/app/core/workbench/registry.py

from __future__ import annotations

from typing import Any

from app.core.system.permissions import READ_GOVERNANCE


WIDGET_REGISTRY = {
    "pending_reviews": {"title": "待审核数量", "min_w": 2, "max_w": 6, "min_h": 2, "max_h": 4, "permission": READ_GOVERNANCE},
    "order_status": {"title": "数据订单状态", "min_w": 3, "max_w": 8, "min_h": 2, "max_h": 6, "permission": READ_GOVERNANCE},
    "product_stats": {"title": "数据产品统计", "min_w": 2, "max_w": 6, "min_h": 2, "max_h": 4, "permission": READ_GOVERNANCE},
    "datasource_health": {"title": "数据源健康", "min_w": 3, "max_w": 8, "min_h": 2, "max_h": 6, "permission": READ_GOVERNANCE},
    "n8n_executions": {"title": "n8n 执行概览", "min_w": 3, "max_w": 8, "min_h": 2, "max_h": 6, "permission": READ_GOVERNANCE},
}

DEFAULT_LAYOUT = [
    {"id": "pending_reviews", "x": 0, "y": 0, "w": 3, "h": 2},
    {"id": "order_status", "x": 3, "y": 0, "w": 5, "h": 3},
    {"id": "product_stats", "x": 8, "y": 0, "w": 3, "h": 2},
    {"id": "datasource_health", "x": 0, "y": 3, "w": 5, "h": 3},
    {"id": "n8n_executions", "x": 5, "y": 3, "w": 6, "h": 3},
]


def validate_layout(widgets: list[dict[str, Any]]) -> list[dict[str, Any]]:
    if not isinstance(widgets, list):
        raise ValueError("widgets must be a list")
    seen = set()
    cleaned = []
    for item in widgets:
        widget_id = item.get("id")
        spec = WIDGET_REGISTRY.get(widget_id)
        if not spec or widget_id in seen:
            raise ValueError("unknown or duplicate widget")
        seen.add(widget_id)
        values = {key: int(item.get(key, 0)) for key in ("x", "y", "w", "h")}
        if values["x"] < 0 or values["y"] < 0 or not (spec["min_w"] <= values["w"] <= spec["max_w"]) or not (spec["min_h"] <= values["h"] <= spec["max_h"]):
            raise ValueError(f"invalid size or position for {widget_id}")
        cleaned.append({"id": widget_id, **values})
    return cleaned