Improve AI Table dashboard creation workflow

2026-08-13 10:28:30 +08:00
3 changed files with 329 additions and 13 deletions
@@ -61,6 +61,7 @@ metadata:
 | 更新记录 | `dws aitable record update --base-id <baseId> --table-id <tableId> --records '[{"recordId":"<id>","cells":{"<fieldId>":<值>}}]' --format json` | 先 query 拿 recordId；只传需改字段；取 `data.recordIds[]` 后回读 |
 | 删除记录 | 先 `dws aitable +record-query ...` 定位，再 `dws aitable record delete --base-id <baseId> --table-id <tableId> --record-ids <ids>` | 展示目标与影响，得到明确确认后才加 `--yes` |
 | 创建 Base / Table | `dws aitable base create --name "<名>"` / `dws aitable table create --base-id <id> --name "<名>" --fields '[...]'` | 使用创建返回的真实 ID；系统改名/加后缀时不得继续猜原名 |
+| 创建仪表盘 / 常用图表 | `python3 <本 Skill 绝对目录>/scripts/create_dashboard_chart.py <baseId> "<仪表盘名>" [--chart-specs <workspace内JSON>]` | 这是 dashboard/chart 创建的唯一首选 recipe；脚本创建、串联真实 ID、最终回读并输出可评分 ledger。图表 JSON 参数见对应 reference |
 | 复制视图 | `dws aitable view duplicate --base-id <baseId> --table-id <tableId> --view-id <源viewId> --new-name "<新名称>" --format json` | 源 viewId 来自当前表的真实返回；不要复制数据表或创建仪表盘替代 |
 | 批量追加 CSV / JSON 到已有表 | `python3 scripts/import_records.py <baseId> <tableId> <file> [batch_size]` | CSV 表头必须是 fieldId；脚本返回不完整 ledger 时不得宣称全成功 |
 | 文件导入为新数据表 | `python3 scripts/aitable_import_via_task.py <baseId> <file>` | 与“追加已有 table”不同；走 prepare → PUT → import task |
@@ -1,18 +1,43 @@
 # dashboard & chart — 仪表盘与图表

-## 建议操作顺序
+## 创建首选流程

 ```bash
-# 1) 先看配置模板（JSONC）
-dws aitable dashboard config-example --format json
-dws aitable chart widgets-example --format json
+# 仅建仪表盘
+python3 <本 Skill 绝对目录>/scripts/create_dashboard_chart.py <BASE_ID> "<仪表盘名>"

-# 2) 先拿 dashboard，再拿 chart 详情
-dws aitable dashboard get --base-id <BASE_ID> --dashboard-id <DASHBOARD_ID> --format json
-dws aitable chart get --base-id <BASE_ID> --dashboard-id <DASHBOARD_ID> --chart-id <CHART_ID> --format json
+# 建仪表盘和常用图表
+python3 <本 Skill 绝对目录>/scripts/create_dashboard_chart.py <BASE_ID> "<仪表盘名>" \
+  --chart-specs <workspace内/charts.json>
 ```

-## 要点
+脚本是 `dashboard create → chart create（可选）→ dashboard get` 的唯一首选
+recipe，并输出 `dws-skill-script-ledger/v1`。不要在脚本前调用 config-example 或
+widgets-example，也不要在成功后重复创建或回读。
+
+`charts.json` 是 1–6 项数组，每项参数：
+
+| 参数 | 要求 |
+|---|---|
+| `name` | 必填，图表名 |
+| `chart_type` | 必填：`AREA`、`BAR`、`HISTOGRAM`、`LINE`、`PIE`、`STATISTICS` |
+| `table_id` | 必填，当前链路的真实 tableId |
+| `measure_type` | `record-count`（默认）或 `field` |
+| `measure_field_id` | `measure_type=field` 时必填 |
+| `dimension_field_id` | 分组、分类或时间维度需要时填写 |
+| `aggregation` | 可选：`sum`、`count`、`count_distinct`、`average`、`min`、`max` |
+| `view_id` | 可选；不用视图时省略 |
+
+例如按状态统计记录数：
+
+```json
+[{"name":"跟进状态记录数","chart_type":"HISTOGRAM","table_id":"<tableId>","measure_type":"record-count","dimension_field_id":"<状态fieldId>"}]
+```
+
+脚本不支持的图表类型或完整高级配置才走下方原子命令；这种例外至多读取一次
+`chart widgets-example`，再按真实 tableId/fieldId 构造配置。
+
+## 查询与管理要点

 - `dashboard get` 返回的 `charts[].chartId` 可直接给 `chart get` 使用
 - `dashboard share get` 可能返回 `404`（资源不存在或未开通），需按可重试错误处理，不要误判为参数拼错
@@ -26,7 +51,7 @@ dws aitable chart get --base-id <BASE_ID> --dashboard-id <DASHBOARD_ID> --chart-
 | `dashboard create` | 创建仪表盘 | `--base-id` + (`--config` 或 `--name`) | `--name` 简化版创建空看板；`--config` 传完整 JSON |
 | `dashboard update` | 更新仪表盘 | `--base-id` `--dashboard-id` + (`--config` 或 `--name`) | `--name` 仅改名；`--config` 更新完整配置 |
 | `dashboard delete` | 删除仪表盘 | `--base-id` `--dashboard-id` `--yes` | — |
-| `dashboard config-example` | 查看仪表盘配置模板 | 无 | 创建前先调此命令了解 config 结构 |
+| `dashboard config-example` | 查看仪表盘配置模板 | 无 | 仅脚本不支持的高级配置按需读取一次 |
 | `dashboard arrange` | 自动重排图表布局 | `--base-id` `--dashboard-id` | 把图表按行铺满网格，避免某行只占半幅、留下大片空白；返回 `{totalColumns, layout, alignedChartCount}` |

 ## chart 子命令
@@ -37,8 +62,7 @@ dws aitable chart get --base-id <BASE_ID> --dashboard-id <DASHBOARD_ID> --chart-
 | `chart create` | 创建图表 | `--base-id` `--dashboard-id` `--config` `--layout` |
 | `chart update` | 更新图表配置 | `--base-id` `--dashboard-id` `--chart-id` `--config` |
 | `chart delete` | 删除图表 | `--base-id` `--dashboard-id` `--chart-id` `--yes` |
-| `chart widgets-example` | 查看图表 widgets 配置模板 | 无 |
+| `chart widgets-example` | 查看图表 widgets 配置模板 | 无；返回很大 | 仅脚本不支持的高级图表按需读取一次 |

-## 配置获取流程
-
-创建图表前至多调用一次 `chart widgets-example`，不要为解析其 JSONC 重复调用或编写临时解析脚本；根据真实 tableId 和 fieldId 填充 `--config`。`chart create` 必须同时传 `--layout '{"x":0,"y":0,"w":6,"h":4}'`，多图表使用互不重叠的位置。用户未指定数量时只创建足以表达目标的 1–2 个图表。取返回 chartId 后立即 `chart get`，或最后 `dashboard get` 核对 charts；回读成功即答复，缺失时不得声称成功或重复提交同一条缺参命令。
+原子 `chart create` 必须同时传 `--layout`。返回 chartId 后用 `chart get`，或最后
+一次 `dashboard get` 核对；回读成功即停止。脚本已自动完成这些动作，不再重复执行。
@@ -0,0 +1,291 @@
+#!/usr/bin/env python3
+"""Create an AI Table dashboard and optional common charts deterministically.
+
+Examples:
+  python3 create_dashboard_chart.py BASE_ID "状态分析仪表盘"
+  python3 create_dashboard_chart.py BASE_ID "状态分析仪表盘" --chart-specs charts.json
+
+charts.json is a JSON array. Each item accepts:
+  name, chart_type, table_id, measure_type, measure_field_id,
+  dimension_field_id, aggregation, view_id.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any, Optional
+
+LEDGER_SCHEMA_VERSION = "dws-skill-script-ledger/v1"
+SCRIPT_NAME = "create_dashboard_chart.py"
+RESOURCE_ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]{3,128}$")
+SUPPORTED_CHART_TYPES = {"AREA", "BAR", "HISTOGRAM", "LINE", "PIE", "STATISTICS"}
+SUPPORTED_AGGREGATIONS = {"sum", "count", "count_distinct", "average", "avg", "min", "max"}
+MAX_CHARTS = 6
+
+
+def run_dws(dws_bin: str, args: list[str]) -> tuple[Optional[dict[str, Any]], str]:
+    try:
+        completed = subprocess.run(
+            [dws_bin, *args], capture_output=True, text=True, timeout=120
+        )
+    except subprocess.TimeoutExpired:
+        return None, "dws command timeout after 120 seconds"
+    except FileNotFoundError:
+        return None, f"dws binary not found: {dws_bin}"
+    if completed.returncode != 0:
+        return None, (completed.stderr or completed.stdout).strip()[:800]
+    try:
+        payload = json.loads(completed.stdout)
+    except json.JSONDecodeError as exc:
+        return None, f"dws returned non-JSON output: {exc}"
+    if not isinstance(payload, dict) or payload.get("status") != "success":
+        return None, f"dws business failure: {payload}"
+    return payload, ""
+
+
+def safe_json_file(value: str) -> Any:
+    root = Path(os.environ.get("OPENCLAW_WORKSPACE", os.getcwd())).resolve()
+    source = Path(value).expanduser()
+    source = source.resolve() if source.is_absolute() else (Path.cwd() / source).resolve()
+    try:
+        source.relative_to(root)
+    except ValueError as exc:
+        raise ValueError(f"chart specs must be inside the workspace: {source}") from exc
+    if not source.is_file() or source.stat().st_size > 1024 * 1024:
+        raise ValueError("chart specs must be a readable JSON file no larger than 1 MiB")
+    with source.open("r", encoding="utf-8") as stream:
+        return json.load(stream)
+
+
+def validate_specs(value: Any) -> list[dict[str, Any]]:
+    if not isinstance(value, list) or not value or len(value) > MAX_CHARTS:
+        raise ValueError(f"chart specs must contain 1-{MAX_CHARTS} items")
+    specs: list[dict[str, Any]] = []
+    for index, item in enumerate(value, start=1):
+        if not isinstance(item, dict):
+            raise ValueError(f"chart spec #{index} must be an object")
+        name = str(item.get("name") or "").strip()
+        chart_type = str(item.get("chart_type") or "").strip().upper()
+        table_id = str(item.get("table_id") or "").strip()
+        measure_type = str(item.get("measure_type") or "record-count").strip()
+        measure_field_id = str(item.get("measure_field_id") or "").strip()
+        dimension_field_id = str(item.get("dimension_field_id") or "").strip()
+        aggregation = str(item.get("aggregation") or "sum").strip().lower()
+        view_id = str(item.get("view_id") or "").strip()
+        if not name or len(name) > 80:
+            raise ValueError(f"chart spec #{index} needs a 1-80 character name")
+        if chart_type not in SUPPORTED_CHART_TYPES:
+            raise ValueError(f"chart spec #{index} has unsupported chart_type: {chart_type}")
+        if not RESOURCE_ID_PATTERN.fullmatch(table_id):
+            raise ValueError(f"chart spec #{index} has invalid table_id")
+        if measure_type not in {"record-count", "field"}:
+            raise ValueError(f"chart spec #{index} has invalid measure_type")
+        if measure_type == "field" and not RESOURCE_ID_PATTERN.fullmatch(measure_field_id):
+            raise ValueError(f"chart spec #{index} needs measure_field_id")
+        if aggregation not in SUPPORTED_AGGREGATIONS:
+            raise ValueError(f"chart spec #{index} has unsupported aggregation")
+        for label, resource_id in (
+            ("dimension_field_id", dimension_field_id), ("view_id", view_id)
+        ):
+            if resource_id and not RESOURCE_ID_PATTERN.fullmatch(resource_id):
+                raise ValueError(f"chart spec #{index} has invalid {label}")
+        specs.append(
+            {
+                "name": name,
+                "chart_type": chart_type,
+                "table_id": table_id,
+                "measure_type": measure_type,
+                "measure_field_id": measure_field_id,
+                "dimension_field_id": dimension_field_id,
+                "aggregation": "average" if aggregation == "avg" else aggregation,
+                "view_id": view_id,
+            }
+        )
+    return specs
+
+
+def chart_config(spec: dict[str, Any]) -> dict[str, Any]:
+    config: dict[str, Any] = {
+        "chartType": spec["chart_type"],
+        "name": spec["name"],
+        "sheet": spec["table_id"],
+        "view": spec["view_id"] or None,
+        "measureType": spec["measure_type"],
+        "measure": [],
+        "filter": [],
+    }
+    if spec["measure_type"] == "field":
+        config["measure"] = [
+            {
+                "value": spec["measure_field_id"],
+                "externalValue": [{"type": "formula", "value": spec["aggregation"]}],
+            }
+        ]
+    if spec["dimension_field_id"]:
+        config["dimension"] = [
+            {"value": spec["dimension_field_id"], "externalValue": []}
+        ]
+    if spec["chart_type"] in {"AREA", "BAR", "HISTOGRAM", "LINE", "PIE"}:
+        config.update({"colors": "COLOR_PALETTE_1", "legend": "top", "label": True})
+    if spec["chart_type"] in {"AREA", "BAR", "HISTOGRAM", "LINE"}:
+        config.update({"xAxisShow": True, "yAxisShow": True})
+    if spec["chart_type"] == "PIE":
+        config.update({"innerRadius": 0, "outerRadius": 60})
+    return config
+
+
+def layout(index: int, total: int) -> dict[str, int]:
+    width = 12 if total == 1 else 6 if total in {2, 3, 4} else 4
+    per_row = 12 // width
+    return {"x": (index % per_row) * width, "y": (index // per_row) * 5, "w": width, "h": 5}
+
+
+def extract_id(payload: dict[str, Any], key: str) -> str:
+    data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
+    value = data.get(key)
+    return str(value or "").strip()
+
+
+def dashboard_chart_ids(payload: dict[str, Any]) -> set[str]:
+    data = payload.get("data") if isinstance(payload.get("data"), dict) else {}
+    charts = data.get("charts") if isinstance(data.get("charts"), list) else []
+    return {
+        str(item.get("chartId"))
+        for item in charts
+        if isinstance(item, dict) and item.get("chartId")
+    }
+
+
+def ledger_step(
+    cli_path: str, status: str, params: dict[str, Any], output_ids: Optional[dict[str, str]] = None,
+    error: str = "",
+) -> dict[str, Any]:
+    return {
+        "cli_path": cli_path,
+        "status": status,
+        "params": params,
+        "output_ids": output_ids or {},
+        "error": error,
+    }
+
+
+def emit(status: str, ledger: list[dict[str, Any]], **result: Any) -> None:
+    print(
+        json.dumps(
+            {
+                "schema_version": LEDGER_SCHEMA_VERSION,
+                "script": SCRIPT_NAME,
+                "status": status,
+                "result": result,
+                "ledger": ledger,
+            },
+            ensure_ascii=False,
+        )
+    )
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("base_id", help="Target AI Table baseId")
+    parser.add_argument("dashboard_name", help="Dashboard name")
+    parser.add_argument("--chart-specs", help="Workspace-local JSON chart spec file")
+    parser.add_argument("--dws", default="dws", help="dws executable")
+    args = parser.parse_args()
+
+    base_id = args.base_id.strip()
+    dashboard_name = args.dashboard_name.strip()
+    if not RESOURCE_ID_PATTERN.fullmatch(base_id) or not dashboard_name:
+        parser.error("base_id and dashboard_name are required and must be valid")
+    try:
+        specs = validate_specs(safe_json_file(args.chart_specs)) if args.chart_specs else []
+    except (OSError, ValueError, json.JSONDecodeError) as exc:
+        parser.error(str(exc))
+
+    ledger: list[dict[str, Any]] = []
+    dashboard_params = {"base-id": base_id, "name": dashboard_name, "format": "json"}
+    dashboard, error = run_dws(
+        args.dws,
+        ["aitable", "dashboard", "create", "--base-id", base_id, "--name", dashboard_name, "--format", "json"],
+    )
+    if not dashboard:
+        ledger.append(ledger_step("aitable dashboard create", "failed", dashboard_params, error=error))
+        emit("failed", ledger, error=error)
+        return 1
+    dashboard_id = extract_id(dashboard, "dashboardId")
+    if not dashboard_id:
+        error = "dashboard create returned no dashboardId"
+        ledger.append(ledger_step("aitable dashboard create", "failed", dashboard_params, error=error))
+        emit("failed", ledger, error=error)
+        return 1
+    ledger.append(
+        ledger_step(
+            "aitable dashboard create", "success", dashboard_params,
+            {"dashboardId": dashboard_id},
+        )
+    )
+
+    chart_ids: list[str] = []
+    for index, spec in enumerate(specs):
+        config = chart_config(spec)
+        chart_layout = layout(index, len(specs))
+        params = {
+            "base-id": base_id,
+            "dashboard-id": dashboard_id,
+            "config": config,
+            "layout": chart_layout,
+            "format": "json",
+        }
+        chart, error = run_dws(
+            args.dws,
+            [
+                "aitable", "chart", "create", "--base-id", base_id,
+                "--dashboard-id", dashboard_id,
+                "--config", json.dumps(config, ensure_ascii=False, separators=(",", ":")),
+                "--layout", json.dumps(chart_layout, separators=(",", ":")),
+                "--format", "json",
+            ],
+        )
+        if not chart:
+            ledger.append(ledger_step("aitable chart create", "failed", params, error=error))
+            emit("failed", ledger, dashboardId=dashboard_id, chartIds=chart_ids, error=error)
+            return 1
+        chart_id = extract_id(chart, "chartId")
+        if not chart_id:
+            error = "chart create returned no chartId"
+            ledger.append(ledger_step("aitable chart create", "failed", params, error=error))
+            emit("failed", ledger, dashboardId=dashboard_id, chartIds=chart_ids, error=error)
+            return 1
+        chart_ids.append(chart_id)
+        ledger.append(
+            ledger_step("aitable chart create", "success", params, {"chartId": chart_id})
+        )
+
+    get_params = {"base-id": base_id, "dashboard-id": dashboard_id, "format": "json"}
+    verified, error = run_dws(
+        args.dws,
+        ["aitable", "dashboard", "get", "--base-id", base_id, "--dashboard-id", dashboard_id, "--format", "json"],
+    )
+    if not verified:
+        ledger.append(ledger_step("aitable dashboard get", "failed", get_params, error=error))
+        emit("failed", ledger, dashboardId=dashboard_id, chartIds=chart_ids, error=error)
+        return 1
+    missing_chart_ids = set(chart_ids) - dashboard_chart_ids(verified)
+    if missing_chart_ids:
+        error = "dashboard verification missing chartIds: " + ",".join(sorted(missing_chart_ids))
+        ledger.append(ledger_step("aitable dashboard get", "failed", get_params, error=error))
+        emit("failed", ledger, dashboardId=dashboard_id, chartIds=chart_ids, error=error)
+        return 1
+    ledger.append(
+        ledger_step("aitable dashboard get", "success", get_params, {"dashboardId": dashboard_id})
+    )
+    emit("success", ledger, dashboardId=dashboard_id, chartIds=chart_ids)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())