opm-add-dlt-source - SKILL.md Agent Skill

name: opm-add-dlt-source description: "open-process-miningにdlt (data load tool) を使った外部システム連携データソースを追加します。dltソース、パイプライン、dbt stagingモデル、設定ファイルを一式生成します。「dltソースを追加」「外部APIからデータ取得」「dltパイプラインを作成」「外部システム連携」「APIからイベントログ」「dltでデータ投入」などのキーワードで発動します。" argument-hint: "連携先システム名、API仕様、またはデータの説明"

dlt外部システム連携によるデータソース追加

外部APIやデータベースからdltでデータを抽出し、dbt経由でopen-process-miningに投入するパイプラインを一式生成します。

共有リソース

パターン集: ../opm-core/references/data-pipeline-patterns.md

出力先

dltソース: dlt/sources/{system}_source.py
dltパイプライン: dlt/pipelines/{system}_pipeline.py
dlt設定: dlt/.dlt/config.toml（追記）, dlt/.dlt/secrets.toml.example（追記）
dbt staging: dbt/models/staging/{system}/stg_{system}_{entity}.sql
Bronze sources: dbt/models/bronze/_bronze__sources.yml（追記）
ユーザーマッピング: dbt/seeds/master_user_mapping.csv（追記）

ワークフロー

Step 1: 入力情報の収集

ユーザーに以下を確認する:

連携先システム名: snake_case（例: salesforce, servicenow）
API仕様: エンドポイント、認証方式、レスポンス構造
プロセスタイプ名: kebab-case（例: crm-lead-management）
イベント定義: どのデータがどのアクティビティに対応するか
組織分析の要否: ユーザー識別子フィールドの特定

Step 2: dltソースの生成

../opm-core/references/data-pipeline-patterns.md の「パターン2」を参照し:

dlt/sources/{system}_source.py を生成
- @dlt.resource デコレータで各エンティティを定義
- ページネーション対応
- 増分ロード対応（since パラメータ）
- ユーザー識別子フィールドを必ず含める（組織分析用）
@dlt.source でリソースを束ねる

Step 3: dltパイプラインの生成

dlt/pipelines/{system}_pipeline.py を生成
- dataset_name="bronze_raw" を指定
- エラーハンドリングを含める

Step 4: dlt設定ファイルの更新

dlt/.dlt/config.toml にソース設定を追記
dlt/.dlt/secrets.toml.example に認証情報のテンプレートを追記

設定キーの命名規則: ソースファイル名 {system}_source.py → セクション [sources.{system}_source]

Step 5: dbt staging モデルの生成

dbt/models/bronze/_bronze__sources.yml に Bronze テーブル定義を追加
dbt/models/staging/{system}/stg_{system}_{entity}.sql を生成
- user_mapping CTE でユーザーマッピングを実装
- case_extraction CTE で case_id を生成
- events CTE でアクティビティイベントを生成
dbt/models/staging/stg_all_events.sql に UNION ALL を追加
dbt/seeds/master_user_mapping.csv にマッピングエントリを追記

Step 6: 検証

以下のコマンドで動作確認を案内:

# dlt パイプライン実行
docker compose -f compose.dev.yml --profile dlt run --rm dlt python pipelines/{system}_pipeline.py

# dbt 変換実行
docker compose -f compose.dev.yml run --rm dbt bash -c "cd /app/dbt && dbt run"

# dbt テスト
docker compose -f compose.dev.yml run --rm dbt bash -c "cd /app/dbt && dbt test"

# Bronze層のデータ確認
docker compose -f compose.dev.yml exec postgres psql -U process_mining -d process_mining_db \
  -c "SELECT COUNT(*) FROM bronze_raw.bronze_{system}_{entity};"

# イベントログ確認
docker compose -f compose.dev.yml exec postgres psql -U process_mining -d process_mining_db \
  -c "SELECT process_type, COUNT(*) FROM fct_event_log GROUP BY process_type;"

Step 7: 結果の提示

生成したファイル一覧と次のステップを案内:

secrets.toml の認証情報設定
パイプラインの初回実行
Web UIでの分析作成

注意事項

dlt/.dlt/secrets.toml は gitignore 対象。secrets.toml.example を参考にユーザーが作成する
dltはDockerコンテナ内で実行される。--profile dlt でオプション起動
Bronze層（bronze_rawスキーマ）にはdltの管理カラム（_dlt_load_id, _dlt_id）が自動追加される
タイムスタンプはUTCに正規化すること（::timestamptz AT TIME ZONE 'UTC'）
既存のdltソース（dlt/sources/github_source.py 等）を参考実装として参照可能