# 月間処理量別のコスト比較
cost_comparison = {
    "monthly_tasks_1000": {
        "zapier": {"plan": "Starter", "cost_usd": 20, "included_tasks": 750,
                   "overage": "$0.01/task"},
        "make": {"plan": "Core", "cost_usd": 9, "included_ops": 10000,
                 "overage": "N/A（余裕あり）"},
        "n8n_cloud": {"plan": "Starter", "cost_usd": 20, "included_execs": 2500,
                      "overage": "N/A"},
        "n8n_self": {"plan": "Self-hosted", "cost_usd": 5,
                     "note": "VPS費用のみ（Hetzner等）"}
    },
    "monthly_tasks_10000": {
        "zapier": {"plan": "Professional", "cost_usd": 49, "included_tasks": 2000,
                   "overage": "$80追加で10000タスク"},
        "make": {"plan": "Core", "cost_usd": 9, "included_ops": 10000,
                 "overage": "ギリギリ"},
        "n8n_cloud": {"plan": "Starter", "cost_usd": 20, "note": "余裕あり"},
        "n8n_self": {"plan": "Self-hosted", "cost_usd": 10,
                     "note": "VPS増強が必要な場合あり"}
    },
    "monthly_tasks_100000": {
        "zapier": {"plan": "Team", "cost_usd": 400, "note": "追加課金必要"},
        "make": {"plan": "Teams", "cost_usd": 99, "note": "追加Ops購入必要"},
        "n8n_cloud": {"plan": "Pro", "cost_usd": 50, "note": "余裕あり"},
        "n8n_self": {"plan": "Self-hosted", "cost_usd": 30,
                     "note": "高性能VPS必要"}
    }
}

# Zapier フロー定義（概念的にPythonで表現）
zapier_flow = {
    "name": "顧客問い合わせ自動分類・対応",
    "trigger": {
        "app": "Gmail",
        "event": "新規メール受信",
        "filter": "from:*@customer.com"
    },
    "steps": [
        {
            "app": "ChatGPT (OpenAI)",
            "action": "Conversation",
            "config": {
                "model": "gpt-4",
                "prompt": """
以下のメールを分析し、JSON形式で返答してください:
- category: billing/technical/sales/other
- urgency: high/medium/low
- summary: 50文字以内の要約
- suggested_response: 推奨返信文
 
メール本文:
{{trigger.body}}
""",
                "memory_key": "customer_{{trigger.from}}"
            }
        },
        {
            "app": "Formatter",
            "action": "Text - Extract JSON",
            "config": {"input": "{{step1.response}}"}
        },
        {
            "app": "Paths",
            "conditions": [
                {
                    "name": "緊急対応",
                    "condition": "{{step2.urgency}} == high",
                    "actions": [
                        {"app": "Slack", "channel": "#urgent-support"},
                        {"app": "PagerDuty", "severity": "high"}
                    ]
                },
                {
                    "name": "通常対応",
                    "condition": "{{step2.urgency}} != high",
                    "actions": [
                        {"app": "Notion", "database": "Support Tickets"},
                        {"app": "Gmail", "draft": "{{step2.suggested_response}}"}
                    ]
                }
            ]
        }
    ]
}

# Zapier マルチステップ AI ワークフロー
zapier_advanced = {
    "name": "営業リードスコアリング＆自動フォローアップ",
    "trigger": {
        "app": "Typeform",
        "event": "新規回答受信"
    },
    "steps": [
        {
            "step": 1,
            "app": "ChatGPT",
            "action": "リードスコアリング",
            "prompt": """
以下のフォーム回答からリードの質を0-100でスコアリング:
- 会社名: {{trigger.company}}
- 役職: {{trigger.title}}
- 従業員数: {{trigger.employees}}
- 予算: {{trigger.budget}}
- 導入時期: {{trigger.timeline}}
- 課題: {{trigger.challenge}}
 
JSON形式で返答: {score, reasoning, segment, recommended_action}
"""
        },
        {
            "step": 2,
            "app": "Formatter",
            "action": "JSON解析"
        },
        {
            "step": 3,
            "app": "Paths",
            "conditions": [
                {
                    "name": "ホットリード（80+）",
                    "condition": "score >= 80",
                    "actions": [
                        {"app": "Salesforce", "action": "リード作成", "priority": "Hot"},
                        {"app": "Slack", "channel": "#sales-hot-leads"},
                        {"app": "Calendly", "action": "即日ミーティングリンク送信"},
                        {"app": "Gmail", "action": "パーソナライズドメール送信",
                         "template": "hot_lead_template"}
                    ]
                },
                {
                    "name": "ウォームリード（50-79）",
                    "condition": "50 <= score < 80",
                    "actions": [
                        {"app": "HubSpot", "action": "コンタクト作成",
                         "lifecycle": "MQL"},
                        {"app": "Mailchimp", "action": "ナーチャリングシーケンス登録"},
                        {"app": "Slack", "channel": "#sales-pipeline"}
                    ]
                },
                {
                    "name": "コールドリード（50未満）",
                    "condition": "score < 50",
                    "actions": [
                        {"app": "Mailchimp", "action": "一般メルマガ登録"},
                        {"app": "Google Sheets", "action": "記録のみ"}
                    ]
                }
            ]
        }
    ]
}

# Zapier Tables を AI ナレッジベースとして活用
zapier_tables_ai = {
    "name": "AI FAQ自動応答 + ナレッジベース学習",
    "architecture": {
        "zapier_tables": {
            "faq_table": {
                "columns": ["question", "answer", "category",
                            "usage_count", "last_updated", "feedback_score"],
                "purpose": "FAQ データベース"
            },
            "feedback_table": {
                "columns": ["query", "ai_response", "user_rating",
                            "corrected_answer", "timestamp"],
                "purpose": "フィードバック収集"
            }
        },
        "flow": [
            "1. ユーザーが質問をSubmit",
            "2. Zapier Tables から類似FAQ検索",
            "3. 見つかった → そのまま回答（APIコストゼロ）",
            "4. 見つからない → AI生成 → 回答 → Tablesに追加",
            "5. ユーザーがフィードバック → 品質改善ループ"
        ]
    },
    "cost_savings": {
        "without_cache": "月$500（全問AI呼び出し）",
        "with_tables_cache": "月$100（80%キャッシュヒット）",
        "savings": "80% ($400/月)"
    }
}

# docker-compose.yml - n8n セルフホスト
version: '3.8'
services:
  n8n:
    image: n8nio/n8n:latest
    ports:
      - "5678:5678"
    environment:
      - N8N_BASIC_AUTH_ACTIVE=true
      - N8N_BASIC_AUTH_USER=admin
      - N8N_BASIC_AUTH_PASSWORD=${N8N_PASSWORD}
      - N8N_ENCRYPTION_KEY=${ENCRYPTION_KEY}
      - OPENAI_API_KEY=${OPENAI_API_KEY}
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
    volumes:
      - n8n_data:/home/node/.n8n
    restart: unless-stopped
 
  postgres:
    image: postgres:16
    environment:
      - POSTGRES_DB=n8n
      - POSTGRES_USER=n8n
      - POSTGRES_PASSWORD=${DB_PASSWORD}
    volumes:
      - postgres_data:/var/lib/postgresql/data
 
volumes:
  n8n_data:
  postgres_data:

{
  "name": "AI Document Summarizer",
  "nodes": [
    {
      "type": "n8n-nodes-base.webhook",
      "name": "Document Upload",
      "parameters": {
        "httpMethod": "POST",
        "path": "summarize",
        "responseMode": "lastNode"
      }
    },
    {
      "type": "n8n-nodes-base.code",
      "name": "Preprocess",
      "parameters": {
        "jsCode": "// ドキュメントをチャンクに分割\nconst text = $input.first().json.document;\nconst chunkSize = 3000;\nconst chunks = [];\nfor (let i = 0; i < text.length; i += chunkSize) {\n  chunks.push({ chunk: text.slice(i, i + chunkSize), index: i / chunkSize });\n}\nreturn chunks.map(c => ({ json: c }));"
      }
    },
    {
      "type": "@n8n/n8n-nodes-langchain.openAi",
      "name": "Summarize Chunks",
      "parameters": {
        "model": "gpt-4",
        "prompt": "以下のテキストを日本語で3行に要約:\n\n{{$json.chunk}}"
      }
    },
    {
      "type": "n8n-nodes-base.code",
      "name": "Merge Summaries",
      "parameters": {
        "jsCode": "const summaries = $input.all().map(i => i.json.text).join('\\n');\nreturn [{ json: { combined: summaries } }];"
      }
    },
    {
      "type": "@n8n/n8n-nodes-langchain.openAi",
      "name": "Final Summary",
      "parameters": {
        "model": "gpt-4",
        "prompt": "以下の部分要約を統合し、構造化された最終要約を作成:\n\n{{$json.combined}}"
      }
    }
  ]
}

{
  "name": "RAG Customer Support Bot",
  "description": "ベクトルDBを活用したAIカスタマーサポート",
  "nodes": [
    {
      "type": "n8n-nodes-base.webhook",
      "name": "Chat Webhook",
      "parameters": {
        "httpMethod": "POST",
        "path": "chat",
        "responseMode": "lastNode"
      }
    },
    {
      "type": "n8n-nodes-base.code",
      "name": "Query Preprocessing",
      "parameters": {
        "jsCode": "const query = $input.first().json.message;\nconst userId = $input.first().json.user_id;\n\n// 会話履歴の取得\nconst history = await $getWorkflowStaticData('global');\nconst userHistory = history[userId] || [];\n\n// 直近5件の会話を保持\nconst context = userHistory.slice(-5).map(h => `${h.role}: ${h.content}`).join('\\n');\n\nreturn [{ json: { query, userId, context, userHistory } }];"
      }
    },
    {
      "type": "@n8n/n8n-nodes-langchain.vectorStore",
      "name": "Knowledge Base Search",
      "parameters": {
        "mode": "retrieve",
        "topK": 5,
        "query": "{{$json.query}}"
      }
    },
    {
      "type": "@n8n/n8n-nodes-langchain.openAi",
      "name": "Generate Response",
      "parameters": {
        "model": "gpt-4",
        "systemPrompt": "あなたはカスタマーサポートAIです。ナレッジベースの情報に基づいて正確に回答してください。不明な場合は正直に伝え、人間のサポートへのエスカレーションを提案してください。",
        "prompt": "会話履歴:\n{{$json.context}}\n\n関連情報:\n{{$json.documents}}\n\n質問: {{$json.query}}"
      }
    },
    {
      "type": "n8n-nodes-base.code",
      "name": "Save History & Route",
      "parameters": {
        "jsCode": "const response = $input.first().json.text;\nconst confidence = response.includes('不明') || response.includes('確認') ? 'low' : 'high';\n\n// エスカレーション判定\nif (confidence === 'low') {\n  return [{ json: { response, action: 'escalate', channel: '#support-escalation' } }];\n}\n\nreturn [{ json: { response, action: 'reply', confidence } }];"
      }
    }
  ]
}

# docker-compose.production.yml - 本番環境用
version: '3.8'
services:
  n8n:
    image: n8nio/n8n:1.30.0  # バージョン固定
    ports:
      - "5678:5678"
    environment:
      - N8N_BASIC_AUTH_ACTIVE=true
      - N8N_BASIC_AUTH_USER=${N8N_USER}
      - N8N_BASIC_AUTH_PASSWORD=${N8N_PASSWORD}
      - N8N_ENCRYPTION_KEY=${ENCRYPTION_KEY}
      - N8N_HOST=n8n.yourdomain.com
      - N8N_PORT=5678
      - N8N_PROTOCOL=https
      - WEBHOOK_URL=https://n8n.yourdomain.com/
      - DB_TYPE=postgresdb
      - DB_POSTGRESDB_HOST=postgres
      - DB_POSTGRESDB_PORT=5432
      - DB_POSTGRESDB_DATABASE=n8n
      - DB_POSTGRESDB_USER=n8n
      - DB_POSTGRESDB_PASSWORD=${DB_PASSWORD}
      - EXECUTIONS_DATA_PRUNE=true
      - EXECUTIONS_DATA_MAX_AGE=168  # 7日
      - EXECUTIONS_DATA_SAVE_ON_ERROR=all
      - EXECUTIONS_DATA_SAVE_ON_SUCCESS=none  # 成功時は保存しない
      - GENERIC_TIMEZONE=Asia/Tokyo
      - N8N_METRICS=true  # Prometheus メトリクス有効化
    volumes:
      - n8n_data:/home/node/.n8n
    depends_on:
      postgres:
        condition: service_healthy
    deploy:
      resources:
        limits:
          memory: 2G
          cpus: '2'
    restart: always
    healthcheck:
      test: ["CMD", "wget", "-q", "--spider", "http://localhost:5678/healthz"]
      interval: 30s
      timeout: 10s
      retries: 3
 
  postgres:
    image: postgres:16-alpine
    environment:
      - POSTGRES_DB=n8n
      - POSTGRES_USER=n8n
      - POSTGRES_PASSWORD=${DB_PASSWORD}
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U n8n"]
      interval: 10s
      timeout: 5s
      retries: 5
    deploy:
      resources:
        limits:
          memory: 1G
    restart: always
 
  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data
    restart: always
 
  caddy:
    image: caddy:2-alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./Caddyfile:/etc/caddy/Caddyfile
      - caddy_data:/data
    restart: always
 
volumes:
  n8n_data:
  postgres_data:
  redis_data:
  caddy_data:

# Make HTTP モジュール設定（概念的表現）
make_scenario = {
    "modules": [
        {
            "type": "RSS - Watch Items",
            "config": {
                "url": "https://myblog.com/feed",
                "limit": 5
            }
        },
        {
            "type": "HTTP - Make a Request",
            "config": {
                "url": "https://api.openai.com/v1/chat/completions",
                "method": "POST",
                "headers": {
                    "Authorization": "Bearer {{OPENAI_API_KEY}}",
                    "Content-Type": "application/json"
                },
                "body": {
                    "model": "gpt-4",
                    "messages": [{
                        "role": "user",
                        "content": "以下のブログ記事からSNS投稿を3種類生成:\n"
                                   "- Twitter用(140文字)\n"
                                   "- LinkedIn用(300文字、ビジネス調)\n"
                                   "- Facebook用(200文字、カジュアル)\n\n"
                                   "記事: {{1.title}} - {{1.description}}"
                    }]
                }
            }
        },
        {
            "type": "JSON - Parse",
            "config": {
                "input": "{{2.data.choices[0].message.content}}"
            }
        }
    ]
}

# Make eコマース注文AI処理シナリオ
make_ecommerce = {
    "name": "AI注文処理・不正検知・顧客対応",
    "trigger": {
        "module": "Shopify - Watch Orders",
        "config": {"status": "any"}
    },
    "scenario": [
        {
            "step": 1,
            "module": "HTTP - OpenAI API",
            "purpose": "注文の不正リスク判定",
            "prompt": """
以下の注文情報から不正リスクを0-100で判定:
- 注文額: {{order.total_price}}
- 配送先: {{order.shipping_address}}
- 請求先: {{order.billing_address}}
- メール: {{order.email}}
- IP国: {{order.browser_ip_country}}
- 過去注文数: {{customer.orders_count}}
- アカウント作成日: {{customer.created_at}}
 
JSON: {risk_score, risk_factors, recommendation}
"""
        },
        {
            "step": 2,
            "module": "Router",
            "routes": [
                {
                    "name": "高リスク（80+）",
                    "filter": "risk_score >= 80",
                    "actions": [
                        "Shopify: 注文を保留",
                        "Slack: #fraud-alert に通知",
                        "Email: 確認メール送信"
                    ]
                },
                {
                    "name": "中リスク（50-79）",
                    "filter": "50 <= risk_score < 80",
                    "actions": [
                        "Shopify: 手動レビューフラグ設定",
                        "Slack: #orders-review に通知"
                    ]
                },
                {
                    "name": "低リスク（50未満）",
                    "filter": "risk_score < 50",
                    "actions": [
                        "Shopify: 自動承認",
                        "Email: 注文確認メール（AIパーソナライズ）",
                        "Slack: #orders に記録"
                    ]
                }
            ]
        },
        {
            "step": 3,
            "module": "HTTP - OpenAI API",
            "purpose": "パーソナライズド確認メール生成",
            "prompt": """
以下の顧客に合わせた注文確認メールを生成:
- 名前: {{customer.first_name}}
- 注文商品: {{order.line_items}}
- 過去の購入回数: {{customer.orders_count}}
- リピーターか: {{customer.orders_count > 1}}
 
トーン: 親しみやすく、ブランドに合った表現
含めること: おすすめ商品（購入履歴ベース）
"""
        }
    ]
}

# Make 組み込み関数の活用パターン
make_data_transforms = {
    "text_operations": {
        "trim": "{{trim(data.text)}}",
        "lower": "{{lower(data.text)}}",
        "replace": '{{replace(data.text; "old"; "new")}}',
        "split": '{{split(data.text; ",")}}',
        "substring": "{{substring(data.text; 0; 100)}}",
        "length": "{{length(data.text)}}"
    },
    "date_operations": {
        "now": "{{now}}",
        "format": '{{formatDate(now; "YYYY-MM-DD")}}',
        "add_days": '{{addDays(now; 7)}}',
        "parse": '{{parseDate(data.date; "DD/MM/YYYY")}}'
    },
    "array_operations": {
        "map": "{{map(data.items; 'name')}}",
        "filter": '{{filter(data.items; "status"; "active")}}',
        "join": '{{join(data.items; ", ")}}',
        "first": "{{first(data.items)}}",
        "last": "{{last(data.items)}}",
        "count": "{{length(data.items)}}"
    },
    "conditional": {
        "if": '{{if(data.score > 80; "high"; "low")}}',
        "ifempty": '{{ifempty(data.name; "Unknown")}}',
        "switch": '{{switch(data.status; "active"; "Active"; "inactive"; "Inactive"; "Unknown")}}'
    },
    "math": {
        "round": "{{round(data.price; 2)}}",
        "ceil": "{{ceil(data.price)}}",
        "floor": "{{floor(data.price)}}",
        "min": "{{min(data.a; data.b)}}",
        "max": "{{max(data.a; data.b)}}"
    }
}

# ワークフローエラーハンドリングの3層構造
class WorkflowErrorHandler:
    """ワークフロー共通エラーハンドリング"""
 
    def __init__(self):
        self.retry_count = 0
        self.max_retries = 3
        self.dead_letter_queue = []
 
    def handle_error(self, error, step_name: str, input_data: dict):
        """3層エラーハンドリング"""
        # Layer 1: リトライ
        if self.retry_count < self.max_retries:
            self.retry_count += 1
            wait_time = 2 ** self.retry_count
            print(f"[{step_name}] リトライ {self.retry_count}/{self.max_retries} "
                  f"({wait_time}秒後)")
            time.sleep(wait_time)
            return "retry"
 
        # Layer 2: フォールバック
        if hasattr(self, f"fallback_{step_name}"):
            print(f"[{step_name}] フォールバック実行")
            return getattr(self, f"fallback_{step_name}")(input_data)
 
        # Layer 3: Dead Letter Queue
        self.dead_letter_queue.append({
            "step": step_name,
            "error": str(error),
            "input": input_data,
            "timestamp": time.time()
        })
        self.notify_admin(step_name, error)
        return "failed"
 
    def fallback_ai_summarize(self, input_data):
        """AI要約のフォールバック: 安いモデルで再試行"""
        return call_ai(input_data["text"], model="gpt-3.5-turbo")
 
    def notify_admin(self, step_name, error):
        """管理者通知"""
        send_slack(f"ワークフロー障害: {step_name} - {error}")

import asyncio
from datetime import datetime, timedelta
from collections import deque
 
class RateLimitedWorkflow:
    """レート制限対応ワークフロー実行エンジン"""
 
    def __init__(self, max_requests_per_minute: int = 60,
                 max_tokens_per_minute: int = 100000):
        self.max_rpm = max_requests_per_minute
        self.max_tpm = max_tokens_per_minute
        self.request_timestamps = deque()
        self.token_usage = deque()
 
    async def execute_with_rate_limit(self, tasks: list[dict]) -> list[dict]:
        """レート制限を守りながらタスクを実行"""
        results = []
        for task in tasks:
            # レート制限チェック
            await self._wait_for_rate_limit(task.get("estimated_tokens", 1000))
 
            try:
                result = await self._execute_task(task)
                results.append({"status": "success", "result": result})
            except Exception as e:
                if "rate_limit" in str(e).lower():
                    # レート制限エラー: 指数バックオフで再試行
                    await asyncio.sleep(60)
                    result = await self._execute_task(task)
                    results.append({"status": "success", "result": result})
                else:
                    results.append({"status": "error", "error": str(e)})
 
            self._record_request(task.get("estimated_tokens", 1000))
 
        return results
 
    async def _wait_for_rate_limit(self, estimated_tokens: int):
        """レート制限の待機"""
        now = datetime.now()
        one_minute_ago = now - timedelta(minutes=1)
 
        # 古いエントリを削除
        while self.request_timestamps and self.request_timestamps[0] < one_minute_ago:
            self.request_timestamps.popleft()
        while self.token_usage and self.token_usage[0][0] < one_minute_ago:
            self.token_usage.popleft()
 
        # RPMチェック
        if len(self.request_timestamps) >= self.max_rpm:
            wait_time = (self.request_timestamps[0] - one_minute_ago).total_seconds()
            await asyncio.sleep(max(0, wait_time) + 1)
 
        # TPMチェック
        current_tokens = sum(t[1] for t in self.token_usage)
        if current_tokens + estimated_tokens > self.max_tpm:
            await asyncio.sleep(60)
 
    def _record_request(self, tokens: int):
        """リクエスト記録"""
        now = datetime.now()
        self.request_timestamps.append(now)
        self.token_usage.append((now, tokens))

class DataNormalizer:
    """ワークフロー間データ正規化"""
 
    @staticmethod
    def normalize_contact(source: str, data: dict) -> dict:
        """異なるソースからの連絡先データを統一形式に変換"""
        normalizers = {
            "gmail": lambda d: {
                "email": d.get("from", ""),
                "name": d.get("from_name", ""),
                "company": DataNormalizer._extract_company_from_email(d.get("from", "")),
                "source": "email",
                "timestamp": d.get("date"),
                "content": d.get("body", "")
            },
            "typeform": lambda d: {
                "email": d.get("email", ""),
                "name": f"{d.get('first_name', '')} {d.get('last_name', '')}".strip(),
                "company": d.get("company", ""),
                "source": "form",
                "timestamp": d.get("submitted_at"),
                "content": str(d.get("answers", {}))
            },
            "slack": lambda d: {
                "email": d.get("user_email", ""),
                "name": d.get("user_name", ""),
                "company": "",
                "source": "chat",
                "timestamp": d.get("ts"),
                "content": d.get("text", "")
            },
            "intercom": lambda d: {
                "email": d.get("email", ""),
                "name": d.get("name", ""),
                "company": d.get("company", {}).get("name", ""),
                "source": "support",
                "timestamp": d.get("created_at"),
                "content": d.get("body", "")
            }
        }
 
        normalizer = normalizers.get(source)
        if not normalizer:
            raise ValueError(f"未対応のソース: {source}")
 
        normalized = normalizer(data)
        # 共通バリデーション
        normalized["email"] = normalized["email"].lower().strip()
        normalized["name"] = normalized["name"].strip()
        return normalized
 
    @staticmethod
    def _extract_company_from_email(email: str) -> str:
        """メールアドレスから会社名を推定"""
        domain = email.split("@")[-1] if "@" in email else ""
        free_domains = {"gmail.com", "yahoo.co.jp", "hotmail.com", "outlook.com"}
        if domain in free_domains:
            return ""
        return domain.split(".")[0].capitalize()

class WorkflowTestSuite:
    """ワークフローのテスト実行フレームワーク"""
 
    def __init__(self, workflow_config: dict):
        self.config = workflow_config
        self.test_results = []
 
    def run_unit_tests(self):
        """各ステップの単体テスト"""
        test_cases = {
            "email_classification": [
                {
                    "input": {"subject": "請求書について", "body": "先月の請求額が違います"},
                    "expected": {"category": "billing", "priority": "high"}
                },
                {
                    "input": {"subject": "新機能の提案", "body": "こんな機能があると便利です"},
                    "expected": {"category": "sales", "priority": "low"}
                },
                {
                    "input": {"subject": "システムエラー", "body": "ログインできません"},
                    "expected": {"category": "technical", "priority": "high"}
                }
            ],
            "lead_scoring": [
                {
                    "input": {"company": "大手株式会社", "employees": 500,
                              "budget": "500万円以上", "timeline": "今月中"},
                    "expected_range": {"score_min": 70, "score_max": 100}
                },
                {
                    "input": {"company": "個人", "employees": 1,
                              "budget": "検討中", "timeline": "未定"},
                    "expected_range": {"score_min": 0, "score_max": 40}
                }
            ]
        }
 
        for test_name, cases in test_cases.items():
            for i, case in enumerate(cases):
                result = self._execute_step(test_name, case["input"])
                passed = self._validate_result(result, case)
                self.test_results.append({
                    "test": f"{test_name}_{i+1}",
                    "passed": passed,
                    "input": case["input"],
                    "result": result,
                    "expected": case.get("expected") or case.get("expected_range")
                })
 
        return self.test_results
 
    def run_integration_test(self, test_data: dict):
        """エンドツーエンドの統合テスト"""
        print("=== 統合テスト開始 ===")
        results = []
        for step in self.config["steps"]:
            result = self._execute_step(step["name"], test_data)
            results.append({"step": step["name"], "result": result})
            test_data = {**test_data, **result}  # 次のステップに結果を渡す
        print(f"=== 統合テスト完了: {len(results)}ステップ ===")
        return results
 
    def run_load_test(self, concurrent_requests: int = 10):
        """負荷テスト"""
        import time
        start_time = time.time()
        results = []
 
        for i in range(concurrent_requests):
            test_data = self._generate_test_data()
            result = self.run_integration_test(test_data)
            results.append(result)
 
        elapsed = time.time() - start_time
        return {
            "total_requests": concurrent_requests,
            "total_time_sec": round(elapsed, 2),
            "avg_time_per_request": round(elapsed / concurrent_requests, 2),
            "success_rate": sum(1 for r in results if r) / len(results) * 100
        }
 
    def _execute_step(self, step_name: str, input_data: dict) -> dict:
        """ステップ実行（モック対応）"""
        # テスト環境ではAI APIをモックに差し替え
        return {"status": "success", "mock": True}
 
    def _validate_result(self, result: dict, expected: dict) -> bool:
        """結果の検証"""
        if "expected" in expected:
            return all(result.get(k) == v for k, v in expected["expected"].items())
        if "expected_range" in expected:
            score = result.get("score", 0)
            return expected["expected_range"]["score_min"] <= score <= expected["expected_range"]["score_max"]
        return True
 
    def _generate_test_data(self) -> dict:
        """テストデータ生成"""
        return {
            "subject": "テストメール",
            "body": "これはテスト用のメール本文です。",
            "sender": "test@example.com",
            "date": "2025-01-01T00:00:00Z"
        }

class WorkflowAlertManager:
    """ワークフローアラート管理"""
 
    def __init__(self):
        self.alert_rules = [
            {
                "name": "成功率低下",
                "condition": lambda metrics: metrics["success_rate"] < 95,
                "severity": "warning",
                "channel": "slack:#workflow-alerts",
                "message": "ワークフロー成功率が{success_rate}%に低下"
            },
            {
                "name": "成功率危機",
                "condition": lambda metrics: metrics["success_rate"] < 80,
                "severity": "critical",
                "channel": "pagerduty",
                "message": "ワークフロー成功率が{success_rate}%に急落"
            },
            {
                "name": "APIコスト超過",
                "condition": lambda metrics: metrics["daily_api_cost"] > 50,
                "severity": "warning",
                "channel": "slack:#cost-alerts",
                "message": "日次APIコストが${daily_api_cost}に到達"
            },
            {
                "name": "レスポンス遅延",
                "condition": lambda metrics: metrics["avg_latency_sec"] > 10,
                "severity": "warning",
                "channel": "slack:#workflow-alerts",
                "message": "平均レスポンス時間が{avg_latency_sec}秒に悪化"
            },
            {
                "name": "Dead Letter Queue 蓄積",
                "condition": lambda metrics: metrics["dlq_count"] > 10,
                "severity": "warning",
                "channel": "slack:#workflow-alerts",
                "message": "未処理のDLQメッセージが{dlq_count}件に蓄積"
            }
        ]
 
    def check_alerts(self, metrics: dict):
        """メトリクスを確認してアラートを発火"""
        for rule in self.alert_rules:
            if rule"condition":
                self._send_alert(rule, metrics)
 
    def _send_alert(self, rule: dict, metrics: dict):
        """アラート送信"""
        message = rule["message"].format(**metrics)
        if rule["severity"] == "critical":
            send_pagerduty(message)
            send_slack(f"[CRITICAL] {message}", channel="#workflow-alerts")
        else:
            send_slack(f"[WARNING] {message}", channel="#workflow-alerts")

# BAD: 全判断をAIに委ねる
def process_order(order):
    decision = call_ai(f"この注文を処理すべきか?: {order}")
    if "はい" in decision:
        charge_customer(order)  # AIの回答で課金処理
 
# GOOD: AIは補助、ルールベースと併用
def process_order(order):
    # ルールベースのバリデーション（確実）
    if order.amount > 100000:
        return flag_for_review(order)
    if order.customer.is_blocked:
        return reject(order)
 
    # AIは追加の異常検知にのみ使用
    fraud_score = call_ai(f"不正スコアを0-100で判定: {order}")
    if int(fraud_score) > 80:
        return flag_for_review(order)
 
    return approve(order)

# BAD: 開発環境で数回テストしただけで本番投入
def deploy_workflow_bad():
    workflow = build_workflow()
    deploy_to_production(workflow)  # いきなり全トラフィック
    # → 予期しないデータでエラー続出
 
# GOOD: 段階的デプロイ
def deploy_workflow_good():
    workflow = build_workflow()
 
    # 1. ユニットテスト
    run_unit_tests(workflow)
 
    # 2. ステージング環境で統合テスト
    deploy_to_staging(workflow)
    run_integration_tests(workflow, test_data=generate_diverse_test_data(100))
 
    # 3. カナリアデプロイ（10%のトラフィック）
    deploy_canary(workflow, traffic_percentage=10)
    wait_and_monitor(duration_hours=24)
 
    # 4. メトリクス確認後に全展開
    if get_canary_metrics()["success_rate"] > 99:
        deploy_full(workflow)
    else:
        rollback(workflow)
        alert("カナリアデプロイ失敗 — ロールバック実行")

# BAD: APIキーをワークフローにハードコード
zapier_step = {
    "url": "https://api.openai.com/v1/chat/completions",
    "headers": {"Authorization": "Bearer sk-abc123..."}  # ハードコード
}
 
# GOOD: 環境変数・シークレット管理
zapier_step = {
    "url": "https://api.openai.com/v1/chat/completions",
    "headers": {"Authorization": "Bearer {{env.OPENAI_API_KEY}}"}  # 環境変数参照
}
 
# n8nの場合: Credentials機能を使用
n8n_credentials = {
    "type": "openAiApi",
    "name": "Production OpenAI",
    "data": {
        "apiKey": "{{$credentials.openAiApiKey}}"  # 暗号化保存
    }
}

# カスタマーサポート自動化レシピ
support_automation = {
    "name": "AI カスタマーサポート自動化",
    "platform": "n8n（推奨）/ Zapier / Make",
    "components": {
        "tier_1_auto_reply": {
            "description": "FAQ・よくある質問への自動回答",
            "trigger": "Intercom / Zendesk 新規チケット",
            "process": [
                "受信メッセージを分類",
                "FAQデータベースから類似質問を検索",
                "類似度90%以上 → 自動回答",
                "類似度70-90% → ドラフト生成 + 人間レビュー",
                "類似度70%未満 → エスカレーション"
            ],
            "expected_automation_rate": "40-60%",
            "implementation_time": "2-3日"
        },
        "sentiment_routing": {
            "description": "感情分析によるルーティング",
            "trigger": "チケット作成/更新",
            "process": [
                "テキストの感情分析（positive/neutral/negative）",
                "negative + urgent → シニアスタッフに即座にルーティング",
                "negative + not urgent → 優先対応キューに追加",
                "positive → 通常キュー + アップセル機会フラグ"
            ],
            "expected_benefit": "顧客満足度15%向上、エスカレーション25%削減"
        },
        "multi_language": {
            "description": "多言語自動対応",
            "trigger": "非日本語メッセージの受信",
            "process": [
                "言語検出",
                "日本語に翻訳（内部処理用）",
                "AI回答生成（日本語）",
                "元言語に翻訳して返信",
                "翻訳品質チェック（信頼度スコア）"
            ],
            "supported_languages": "英語、中国語、韓国語、スペイン語、フランス語",
            "implementation_time": "1-2日"
        }
    }
}

# マーケティング自動化レシピ
marketing_automation = {
    "name": "AI マーケティング自動化パイプライン",
    "workflows": {
        "content_repurposing": {
            "trigger": "新規ブログ記事の公開（WordPress Webhook）",
            "steps": [
                "ブログ記事の全文取得",
                "AI: 記事から5つのSNS投稿を生成",
                "  - Twitter: 3ツイート（スレッド形式）",
                "  - LinkedIn: 1投稿（ビジネス向け）",
                "  - Instagram: 1キャプション + ハッシュタグ",
                "AI: メルマガ用要約を生成",
                "Buffer/Hootsuite: 投稿をスケジュール",
                "Mailchimp: メルマガドラフト作成"
            ],
            "monthly_time_saved": "20時間",
            "api_cost": "月$15程度"
        },
        "competitor_monitoring": {
            "trigger": "毎日9:00 AM（スケジュール）",
            "steps": [
                "RSS: 競合ブログの新記事を取得",
                "AI: 競合記事の要約と分析",
                "AI: 自社との差分分析",
                "AI: 対策コンテンツの提案",
                "Slack: #competitive-intel に日次レポート送信",
                "Notion: 競合データベースに記録"
            ],
            "monthly_time_saved": "10時間",
            "api_cost": "月$20程度"
        },
        "review_monitoring": {
            "trigger": "新規レビュー検出（G2, Capterra, App Store）",
            "steps": [
                "レビュー内容の取得",
                "AI: 感情分析 + カテゴリ分類",
                "negative → Slack #reviews-alert に即通知",
                "AI: 返信ドラフト生成",
                "positive → 社内Slackに共有（モチベーション）",
                "Google Sheets: レビューデータベースに記録"
            ],
            "monthly_time_saved": "5時間",
            "api_cost": "月$5程度"
        }
    }
}

# データパイプライン自動化レシピ
data_pipeline = {
    "name": "AI データ処理パイプライン",
    "workflows": {
        "invoice_processing": {
            "description": "請求書の自動処理",
            "trigger": "Gmail: 添付ファイル付きメール受信（件名に「請求書」含む）",
            "steps": [
                "添付PDF/画像を取得",
                "OCR: テキスト抽出（Cloud Vision API）",
                "AI: 構造化データ抽出（金額、日付、取引先、項目）",
                "バリデーション: 金額チェック、重複チェック",
                "会計ソフト（freee/MFクラウド）: 仕訳データ作成",
                "Google Sheets: 管理台帳に記録",
                "Slack: 処理完了通知"
            ],
            "accuracy": "95%以上（人間レビュー推奨）",
            "processing_time": "1請求書あたり30秒"
        },
        "crm_enrichment": {
            "description": "CRMデータの自動エンリッチメント",
            "trigger": "HubSpot/Salesforce: 新規コンタクト作成",
            "steps": [
                "メールドメインから会社情報を取得",
                "LinkedIn API: 会社規模、業界を取得",
                "AI: コンタクトのスコアリングと推奨アクション生成",
                "CRM: フィールド更新（会社規模、業界、スコア）",
                "Slack: 高スコアリードを営業チームに通知"
            ],
            "enrichment_rate": "80%のコンタクトを自動エンリッチ",
            "monthly_cost": "$30程度"
        }
    }
}