Slack sql Bot を作る：llm × ガードレール × 監査の最小実装

import jsonfrom pathlib import Path
manifest = json.loads(Path("target/manifest.json").read_text())
# dbt model の meta.published == True のものだけを Bot に開放allowed_tables = {}for node_id, node in manifest["nodes"].items():    if node["resource_type"] != "model":        continue    if not node.get("meta", {}).get("published"):        continue    fqn = f"{node['database']}.{node['schema']}.{node['name']}"    allowed_tables[fqn] = {        "description": node.get("description", ""),        "columns": {            c: cfg.get("description", "")            for c, cfg in node.get("columns", {}).items()        },    }
print(f"許可テーブル数: {len(allowed_tables)}")

import anthropic
client = anthropic.Anthropic()
def generate_sql(question: str, allowed_tables: dict) -> str:    # メタデータを構造化テキストで埋め込む    schema_doc = "\n\n".join(        f"-- TABLE: {t}\n-- DESC: {info['description']}\n"        + "\n".join(f"--   {c}: {d}" for c, d in info["columns"].items())        for t, info in allowed_tables.items()    )
    prompt = f"""あなたは BigQuery SQL を書くアナリストです。以下のテーブルだけが使えます。これ以外のテーブルを参照する SQL は書かないでください。
{schema_doc}
ルール:- SELECT 文のみ。INSERT/UPDATE/DELETE/DDL は不可- LIMIT 1000 を必ず付ける- WHERE 句で _PARTITIONTIME を 30 日以内に絞る- カラム名を必ず指定（SELECT * は禁止）- 結果を返すだけでなく簡単なコメントで何を計算したかを書く
質問: {question}"""    msg = client.messages.create(        model="claude-opus-4-7",        max_tokens=2000,        messages=[{"role": "user", "content": prompt}],    )    return msg.content[0].text

import reimport sqlparse
DENY_PATTERNS = [    (r"\bSELECT\s+\*", "SELECT * は禁止です。カラム名を明示してください"),    (r"\b(INSERT|UPDATE|DELETE|DROP|TRUNCATE|CREATE|ALTER|GRANT|REVOKE)\b",     "DML/DDL は禁止です。SELECT 文のみ受け付けます"),    (r"\bINFORMATION_SCHEMA\b", "メタデータは Bot 経由で参照してください"),]
def static_check(sql: str, allowed_tables: set[str]) -> tuple[bool, str]:    sql_upper = sql.upper()
    for pat, msg in DENY_PATTERNS:        if re.search(pat, sql_upper):            return False, msg
    # 参照テーブル抽出（簡易: FROM/JOIN の直後）    refs = re.findall(r"\b(?:FROM|JOIN)\s+\`?([\w\.\-]+)\`?", sql_upper)    for ref in refs:        if ref.lower() not in {t.lower() for t in allowed_tables}:            return False, f"テーブル {ref} は許可されていません"
    # LIMIT があるか    if not re.search(r"\bLIMIT\s+\d+", sql_upper):        return False, "LIMIT 句が必要です（最大 1000）"
    return True, "OK"

from google.cloud import bigquery
bq = bigquery.Client()COST_USD_LIMIT = 1.0  # 1 クエリ $1 を超えたら確認を求めるPRICE_PER_TB = 5.0
def estimate_cost(sql: str) -> dict:    job = bq.query(sql, job_config=bigquery.QueryJobConfig(dry_run=True))    bytes_billed = job.total_bytes_processed    cost_usd = (bytes_billed / 1e12) * PRICE_PER_TB    return {"bytes": bytes_billed, "tb": bytes_billed / 1e12, "usd": cost_usd}
def execute_with_guard(sql: str, user: str):    est = estimate_cost(sql)    if est["usd"] > COST_USD_LIMIT:        return {            "status": "needs_confirmation",            "message": f"このクエリは約 ${est['usd']:.2f} かかります（{est['tb']:.2f} TB スキャン）。続行しますか？",            "estimate": est,        }    job = bq.query(sql, job_config=bigquery.QueryJobConfig(maximum_bytes_billed=int(est["bytes"] * 1.5)))    return {"status": "ok", "rows": [dict(r) for r in job.result()], "estimate": est}

import re
PATTERNS = {    "email":   re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+"),    "phone":   re.compile(r"\b0\d{1,4}-?\d{1,4}-?\d{4}\b"),    "card":    re.compile(r"\b\d{4}[-\s]?\d{4}[-\s]?\d{4}[-\s]?\d{4}\b"),}
def mask_row(row: dict) -> dict:    out = {}    for k, v in row.items():        sv = str(v) if v is not None else ""        for label, pat in PATTERNS.items():            sv = pat.sub(f"<{label}_masked>", sv)        out[k] = sv    return out

import json, hashlibfrom datetime import datetimefrom google.cloud import bigquery
audit_table = "analytics.bot_audit_log"
def audit_log(*, user: str, question: str, sql: str, result_rows: int,              cost_usd: float, status: str, masked_sample: list[dict]):    bq = bigquery.Client()    sample_hash = hashlib.sha256(json.dumps(masked_sample, sort_keys=True).encode()).hexdigest()[:16]    bq.insert_rows_json(audit_table, [{        "ts": datetime.utcnow().isoformat(),        "user": user,        "question": question,        "sql": sql,        "result_rows": result_rows,        "cost_usd": cost_usd,        "status": status,        "result_sample_hash": sample_hash,  # 内容ではなくハッシュだけ残す（PII漏洩防止）    }])

def handle_slack_message(event):    user = event["user"]    question = event["text"]
    # 1) SQL 生成    sql = generate_sql(question, allowed_tables)
    # 2) 静的検査    ok, msg = static_check(sql, set(allowed_tables.keys()))    if not ok:        post_to_slack(user, f"❌ {msg}\n生成された SQL:\n\`\`\`{sql}\`\`\`")        audit_log(user=user, question=question, sql=sql, result_rows=0,                  cost_usd=0.0, status="rejected", masked_sample=[])        return
    # 3) DRY RUN + 実行    result = execute_with_guard(sql, user)    if result["status"] == "needs_confirmation":        post_to_slack(user, result["message"] + "\n\`\`\`{sql}\`\`\`\nReact 👍 で実行")        # 実行ボタンの実装は省略。React で確認後に execute() を呼び戻す        return
    # 4) マスキング    masked = [mask_row(r) for r in result["rows"][:50]]
    # 5) 投稿 + 監査    post_to_slack(user, format_table(masked))    audit_log(user=user, question=question, sql=sql,              result_rows=len(result["rows"]),              cost_usd=result["estimate"]["usd"],              status="ok", masked_sample=masked[:5])