powerbi-to-sl

powerbi-to-sl — Power BI → Keboola 语义层迁移

读取现有 Power BI 工件 → 映射为 Keboola SL 负载 → 写入 JSON 文件 → 由用户确认 → 通过 sl-toolkit 或直接 REST 推送。这是 sl-toolkit（从零开始构建）的配套工具；此处适用于已有项目迁移。

---

步骤 0 — 识别输入

询问用户 Power BI 工件所在位置，然后自动检测格式：

INPUT_DIR="$(read -p 'Path to PowerBI artifacts: ' x; echo "$x")"
python3 -c "
import os, sys
p = sys.argv[1]
tmdl = [f for f in os.listdir(p) if f.endswith('.tmdl')] if os.path.isdir(p) else []
jsons = [f for f in os.listdir(p) if f.endswith('_semantic_layer.json')] if os.path.isdir(p) else []
if tmdl: print('tmdl')
elif jsons: print('per-table-json')
else: print('unknown')
" "$INPUT_DIR"

如果为 unknown，请用户说明情况，或让他们使用 Microsoft 的 Power BI Modeling MCP，通过 --readonly 对其数据集运行，以先生成 TMDL。

---

步骤 1 — 业务问题（映射前）

“有三个简短问题：

1. 这些表对应的 Keboola bucket 前缀是什么？（例如， in.c-pbi-migration — 用于为每个 semantic-dataset 构建 tableId）
2. 语义模型名称 + 显示名称是什么？（例如， acme-pbi-migration / “Acme Power BI Migration”）
3. 使用 Snowflake 还是 BigQuery 方言？（默认值：Snowflake）”

将答案保存为 BUCKET_PREFIX、MODEL_NAME、MODEL_DISPLAY、DIALECT。

---

步骤 2 — 运行迁移器

python3 scripts/migrate.py \
  --input "$INPUT_DIR" \
  --input-format "$FORMAT" \
  --bucket-prefix "$BUCKET_PREFIX" \
  --model-name "$MODEL_NAME" \
  --model-display "$MODEL_DISPLAY" \
  --dialect "$DIALECT" \
  --output ./out/

输出布局：

out/
  semantic-model.json
  semantic-dataset/
    <bucket-prefix>.<table>.json   # one per PowerBI table
  semantic-metric/
    <measure-slug>.json            # one per PowerBI measure
  semantic-relationship/
    <from>__<to>.json              # one per PowerBI relationship
  WARNINGS.md                      # DAX-not-translated, unmapped types, etc.

---

步骤 3 — 与用户一起查看 WARNINGS.md

迁移器会在 semantic-metric.sql 中原样保留 DAX 度量表达式。向用户展示 WARNINGS.md，并询问他们希望：

• 暂缓处理 — 保留 DAX；由下游 Kai（或人工）按需改写为 SQL。
• 自动转换简单情况 — SUM('Table'[Col]) → SUM("Table"."Col")、 COUNT、DISTINCTCOUNT、AVERAGE。任何涉及 CALCULATE、筛选器 上下文、RELATED、时间智能或变量的情况都保留标记。
• 人工处理 — 打开 WARNINGS 列表，逐行改写。

默认：暂缓处理 + 显示列表。简单情况的转换需主动选择。

---

步骤 4 — 根据 Keboola SL 架构验证（jsonschema）

python3 -c "
import json, jsonschema, glob, sys, os
schemas_dir = os.environ['SCHEMAS_DIR']
schemas = {
    name: json.load(open(f'{schemas_dir}/semantic-{name}_schema_1.0.0.json'))
    for name in ['model','dataset','metric','relationship']
}
errs = 0
for kind, schema in schemas.items():
    pattern = './out/semantic-model.json' if kind == 'model' else f'./out/semantic-{kind}/*.json'
    for f in glob.glob(pattern):
        try:
            jsonschema.validate(json.load(open(f)), schema)
        except jsonschema.ValidationError as e:
            errs += 1
            print(f'❌ {f}: {e.message}')
print(f'{errs} validation errors' if errs else '✅ all valid')
"

同样的验证也会作为 pytest 断言在 tests/test_smoke.py 中运行。将 SCHEMAS_DIR 设为包含 Keboola metastore semantic-* JSON 架构的目录。

---

步骤 5 — 推送（可选，需用户确认）

任何写入操作前都要与用户确认。有两种方式：

方式 A — 通过 sl-toolkit（如果已安装该插件，优先选择此方式；它负责规范的生成–验证–推送流程）：

“要交由 sl-toolkit 处理吗？它负责任何 SL 模型的规范推送流程。还是直接推送到 metastore REST API？”

如果选择 sl-toolkit，使用 --source ./out/ 调用其推送步骤。

方式 B — 直接 REST（sl-toolkit 不可用时）：

import json, os, urllib.request, glob
from concurrent.futures import ThreadPoolExecutor
TOKEN = os.environ['KBC_STORAGE_TOKEN']
METASTORE = os.environ['KBC_METASTORE_URL']  # e.g. https://metastore.us-east4.gcp.keboola.com

def post(kind, payload):
    req = urllib.request.Request(
        f'{METASTORE}/api/v1/repository/semantic-{kind}',
        data=json.dumps(payload).encode(),
        headers={'X-StorageAPI-Token': TOKEN, 'Content-Type': 'application/json'},
        method='POST')
    return urllib.request.urlopen(req, timeout=30).read()

# Model first — get the UUID, propagate to children
model_uuid = json.loads(post('model', json.load(open('./out/semantic-model.json'))))['data']['id']
for kind in ['dataset','metric','relationship']:
    payloads = []
    for f in glob.glob(f'./out/semantic-{kind}/*.json'):
        p = json.load(open(f))
        p['modelUUID'] = model_uuid
        payloads.append(p)
    with ThreadPoolExecutor(max_workers=10) as ex:
        list(ex.map(lambda p: post(kind, p), payloads))

---

映射参考

Power BI	Keboola SL
tables[].name	semantic-dataset.name + tableId
tables[].description	semantic-dataset.description
tables[].columns[].name	semantic-dataset.fields[].name
tables[].columns[].dataType	semantic-dataset.fields[].type *
tables[].columns[].description	semantic-dataset.fields[].description
PK column hints	semantic-dataset.primaryKey
tables[].measures[].expression (DAX)	semantic-metric.sql（原样保留）
tables[].measures[].description	semantic-metric.description
relationships[]	semantic-relationship

类型映射：string→string、int64/integer→integer、 double/decimal/currency→decimal、dateTime/date→datetime/date、 boolean→boolean。未知类型会回退为 string 并记录日志。

---

已知注意事项

• **逐表 JSON 会丢失 Power Query M 步骤。**较早的 get_semantic_model 输出不会捕获 M 表达式。如果需要完整保真度，请使用 Microsoft 的 Power BI Modeling MCP --readonly 标志重新提取，以生成 TMDL；它能原生捕获 DAX 和 M。
• **DAX ≠ SQL。**默认保留原文；由下游的 Kai 或人工进行转换。CALCULATE、RELATED、时间智能函数和变量无法通过简单转换保留。
• **PowerBI 表名包含空格和非 ASCII 字符。**文件名需转换为 slug；在 name/displayName 中保留原名。UTF-8 路径在 macOS/Linux 上没有问题，但在文件系统特性特殊的环境中需留意。
• **主键检测尽力而为。**Power BI 并不总是显式标记主键；迁移工具会：(a) 使用 TMDL 中显式的 isKey 标记，(b) 将关系中的 to 列作为启发式依据，(c) 若仍无法确定，则使用空的 primaryKey，并添加一条 WARNINGS 条目，请用户自行选择。
• 关系基数。manyToOne 和 oneToOne → inner；oneToMany → 从“多”端开始的 left。manyToMany 会生成一条 WARNINGS 条目——SL 模型要求适合星型架构。

---

为什么存在两个技能（powerbi-to-sl 与 sl-toolkit）

方面	sl-toolkit	powerbi-to-sl（本技能）
方向	全新构建（Keboola 表 → SL）	基于现有模型迁移（Power BI → SL）
输入	Keboola 存储 + SQL 转换	TMDL 文件夹或逐表 JSON
生成	基于 LLM，根据架构 + KPI 生成	机械式结构转换
验证	自动修复 + LLM 评审	jsonschema 验证 + WARNINGS
推送	负责推送管道	将推送交给 sl-toolkit

同时安装这两个技能时，本技能会生成 SL 负载，并将推送委托给 sl-toolkit。这是复用，而非重叠。