# AI に渡すためのエラー情報の構造化
 
import traceback
import sys
import platform
import json
 
class ErrorReporter:
    """デバッグに必要な情報を構造化して収集"""
 
    def capture_error(self, exception: Exception, context: dict = None) -> dict:
        """エラー情報を AI デバッグ用に構造化"""
        tb = traceback.extract_tb(exception.__traceback__)
 
        report = {
            "error": {
                "type": type(exception).__name__,
                "message": str(exception),
                "traceback": self._format_traceback(tb),
            },
            "environment": {
                "python_version": sys.version,
                "platform": platform.platform(),
                "packages": self._get_relevant_packages(tb),
            },
            "context": context or {},
            "source_code": self._extract_source_context(tb),
            "recent_changes": self._get_recent_git_changes(),
        }
        return report
 
    def _format_traceback(self, tb) -> list[dict]:
        """スタックトレースを構造化"""
        frames = []
        for frame in tb:
            frames.append({
                "file": frame.filename,
                "line": frame.lineno,
                "function": frame.name,
                "code": frame.line,
            })
        return frames
 
    def _extract_source_context(self, tb, context_lines=10) -> list[dict]:
        """エラー発生箇所の前後のソースコードを抽出"""
        contexts = []
        for frame in tb[-3:]:  # 直近3フレーム
            try:
                with open(frame.filename) as f:
                    lines = f.readlines()
                start = max(0, frame.lineno - context_lines - 1)
                end = min(len(lines), frame.lineno + context_lines)
                contexts.append({
                    "file": frame.filename,
                    "error_line": frame.lineno,
                    "code": "".join(lines[start:end]),
                    "start_line": start + 1,
                })
            except (FileNotFoundError, PermissionError):
                pass
        return contexts
 
    def _get_recent_git_changes(self) -> str:
        """直近の Git 変更を取得"""
        import subprocess
        try:
            result = subprocess.run(
                ["git", "log", "--oneline", "-5"],
                capture_output=True, text=True, timeout=5,
            )
            return result.stdout.strip()
        except Exception:
            return "Git 情報取得失敗"
 
    def _get_relevant_packages(self, tb) -> dict:
        """スタックトレースに関連するパッケージのバージョンを取得"""
        import importlib.metadata
        packages = {}
        for frame in tb:
            parts = frame.filename.split("/")
            for part in parts:
                if part.endswith(".egg-info") or part == "site-packages":
                    continue
                try:
                    version = importlib.metadata.version(part)
                    packages[part] = version
                except importlib.metadata.PackageNotFoundError:
                    pass
        return packages
 
 
# 使用例
reporter = ErrorReporter()
 
try:
    # バグのあるコード
    result = process_data(None)
except Exception as e:
    error_report = reporter.capture_error(
        e, context={"input": None, "user_id": 123}
    )
    print(json.dumps(error_report, indent=2, ensure_ascii=False))

# 効果的なデバッグプロンプトのテンプレート
 
DEBUG_PROMPT_TEMPLATE = """
以下のエラーの原因を特定し、修正方法を提案してください。
 
## エラー情報
エラー種別: {error_type}
メッセージ: {error_message}
 
## スタックトレース
{traceback}
 
## 関連コード
{source_code}
 
## 環境
{environment}
 
## コンテキスト
{context}
 
## 直近の変更
{recent_changes}
 
以下の形式で回答してください:
1. **原因の推定** (確率の高い順に最大3つ)
2. **各原因の根拠**
3. **確認すべきポイント** (原因を絞り込むための追加調査)
4. **修正コード** (最も可能性の高い原因に対する修正)
5. **再発防止策** (テスト追加、バリデーション強化等)
"""
 
def build_debug_prompt(error_report: dict) -> str:
    """エラーレポートからデバッグプロンプトを構築"""
    return DEBUG_PROMPT_TEMPLATE.format(
        error_type=error_report["error"]["type"],
        error_message=error_report["error"]["message"],
        traceback=format_traceback_text(error_report["error"]["traceback"]),
        source_code=format_source_contexts(error_report["source_code"]),
        environment=json.dumps(error_report["environment"], indent=2),
        context=json.dumps(error_report["context"], indent=2),
        recent_changes=error_report.get("recent_changes", "N/A"),
    )

# チーム共有のエラーパターンデータベース
 
ERROR_PATTERNS = {
    "TypeError: Cannot read properties of undefined": {
        "category": "null_reference",
        "common_causes": [
            "API レスポンスの形式変更 (フィールドが undefined)",
            "非同期処理の完了前にデータアクセス",
            "オプショナルチェーン (?.) の未使用",
        ],
        "fix_patterns": [
            "Optional Chaining: obj?.prop?.nested",
            "Nullish Coalescing: value ?? defaultValue",
            "Guard Clause: if (!obj) return;",
        ],
        "prevention": [
            "TypeScript strict mode の有効化",
            "Zod / Valibot でランタイムバリデーション",
            "API レスポンスの型定義とバリデーション",
        ],
    },
    "ECONNREFUSED": {
        "category": "connection",
        "common_causes": [
            "対象サービスが起動していない",
            "ポート番号の設定ミス",
            "Docker ネットワーク設定の問題",
            "ファイアウォール / セキュリティグループの制限",
        ],
        "diagnosis_steps": [
            "curl / telnet で接続テスト",
            "docker ps でコンテナ状態確認",
            "netstat / lsof でポート使用状況確認",
            "環境変数 (DATABASE_URL 等) の値確認",
        ],
    },
    "OOMKilled": {
        "category": "resource",
        "common_causes": [
            "メモリリーク (イベントリスナーの未解除)",
            "大量データの一括読み込み",
            "コンテナのメモリ制限が不足",
            "N+1 クエリによる大量オブジェクト生成",
        ],
        "fix_patterns": [
            "ストリーミング処理への変更",
            "ページネーション / バッチ処理",
            "メモリプロファイリング (heapdump)",
            "コンテナのリソース制限の見直し",
        ],
    },
}

# 言語・フレームワーク固有のデバッグパターン
 
LANGUAGE_DEBUG_PATTERNS = {
    "python": {
        "ImportError / ModuleNotFoundError": {
            "diagnosis": """
            1. パッケージがインストールされているか確認:
               pip list | grep <package_name>
            2. 仮想環境が正しく有効化されているか確認:
               which python
            3. PYTHONPATH にモジュールのディレクトリが含まれているか確認
            4. パッケージ名とインポート名が異なるケース:
               例: pip install Pillow → import PIL
            """,
            "ai_prompt": """
            以下のImportErrorを解決してください。
            エラー: {error_message}
            Python バージョン: {python_version}
            インストール済みパッケージ: {pip_list}
            仮想環境: {venv_info}
            """,
        },
        "asyncio.TimeoutError": {
            "diagnosis": """
            1. タイムアウト値が適切か確認
            2. 対象サービスのレスポンスタイムを計測
            3. ネットワーク遅延の有無を確認
            4. コネクションプールの枯渇を確認
            5. async/await の使い方が正しいか確認
            """,
            "common_fixes": [
                "タイムアウト値の調整",
                "リトライロジックの追加",
                "コネクションプールサイズの拡大",
                "Circuit Breaker パターンの導入",
            ],
        },
        "SQLAlchemy DetachedInstanceError": {
            "diagnosis": """
            1. セッションのスコープを確認（リクエスト単位か）
            2. Lazy Loading がセッション外で発生していないか
            3. expire_on_commit=False の設定を確認
            4. joinedload / selectinload でイーガーロードに変更
            """,
            "ai_prompt": """
            SQLAlchemy の DetachedInstanceError が発生しています。
            モデル定義: {model_code}
            クエリコード: {query_code}
            セッション設定: {session_config}
            アクセスしようとしたリレーション: {relation_name}
            """,
        },
    },
    "javascript": {
        "Unhandled Promise Rejection": {
            "diagnosis": """
            1. async関数内でtry-catchが適切にされているか
            2. .catch()がPromiseチェーンに付いているか
            3. Promise.allSettled vs Promise.all の使い分け
            4. Node.js のバージョンによる挙動の違い
            """,
            "ai_prompt": """
            Unhandled Promise Rejection が発生しています。
            エラー: {error_message}
            コード: {code}
            Node.js バージョン: {node_version}
            このPromiseがどこでrejectされたか追跡してください。
            """,
        },
        "CORS Error": {
            "diagnosis": """
            1. サーバー側のAccess-Control-Allow-Originヘッダを確認
            2. プリフライトリクエスト（OPTIONS）への応答を確認
            3. クレデンシャル（Cookie）送信時のwithCredentials設定
            4. プロキシ設定（開発時のvite.config.ts / next.config.js）
            """,
            "common_fixes": [
                "cors ミドルウェアの設定追加",
                "開発用プロキシの設定",
                "API GatewayでのCORS設定",
                "credentials: 'include' と対応するサーバー設定",
            ],
        },
    },
    "go": {
        "panic: runtime error: invalid memory address": {
            "diagnosis": """
            1. nil ポインタのデリファレンスを確認
            2. マップの初期化忘れ（var m map[string]int → make必要）
            3. インターフェースの nil チェック漏れ
            4. goroutine 間での共有データの競合
            """,
            "ai_prompt": """
            Go の nil pointer dereference パニックが発生しています。
            スタックトレース: {stacktrace}
            関連コード: {code}
            goroutine の使用有無: {uses_goroutines}
            このパニックの原因と安全な修正方法を提案してください。
            """,
        },
    },
}

// ブラウザ環境のデバッグ情報収集
 
interface BrowserDebugInfo {
  url: string;
  userAgent: string;
  viewport: { width: number; height: number };
  networkErrors: NetworkError[];
  consoleErrors: ConsoleError[];
  performanceMetrics: PerformanceMetrics;
  reactComponentTree?: ComponentInfo[];
}
 
class FrontendDebugCollector {
  /**
   * ブラウザ環境のデバッグ情報を包括的に収集する
   * AIデバッグに最適な形式で構造化
   */
 
  collectDebugInfo(): BrowserDebugInfo {
    return {
      url: window.location.href,
      userAgent: navigator.userAgent,
      viewport: {
        width: window.innerWidth,
        height: window.innerHeight,
      },
      networkErrors: this.getNetworkErrors(),
      consoleErrors: this.getConsoleErrors(),
      performanceMetrics: this.getPerformanceMetrics(),
      reactComponentTree: this.getReactTree(),
    };
  }
 
  private getNetworkErrors(): NetworkError[] {
    // Performance API からネットワークエラーを抽出
    const entries = performance.getEntriesByType("resource") as PerformanceResourceTiming[];
    return entries
      .filter((entry) => entry.responseStatus >= 400 || entry.responseStatus === 0)
      .map((entry) => ({
        url: entry.name,
        status: entry.responseStatus,
        duration: entry.duration,
        initiatorType: entry.initiatorType,
        timestamp: entry.startTime,
      }));
  }
 
  private getConsoleErrors(): ConsoleError[] {
    // 事前にオーバーライドされた console.error のログを取得
    return (window as any).__debugConsoleErrors || [];
  }
 
  private getPerformanceMetrics(): PerformanceMetrics {
    const navigation = performance.getEntriesByType("navigation")[0] as PerformanceNavigationTiming;
    const paint = performance.getEntriesByType("paint");
 
    return {
      domContentLoaded: navigation?.domContentLoadedEventEnd - navigation?.startTime,
      loadComplete: navigation?.loadEventEnd - navigation?.startTime,
      firstPaint: paint.find((p) => p.name === "first-paint")?.startTime,
      firstContentfulPaint: paint.find((p) => p.name === "first-contentful-paint")?.startTime,
      jsHeapSize: (performance as any).memory?.usedJSHeapSize,
      jsHeapLimit: (performance as any).memory?.jsHeapSizeLimit,
    };
  }
 
  /**
   * 収集した情報をAIデバッグ用のプロンプトに変換
   */
  buildAIPrompt(info: BrowserDebugInfo, userDescription: string): string {
    return `
## フロントエンドバグレポート
 
### ユーザー報告
${userDescription}
 
### 環境情報
- URL: ${info.url}
- ブラウザ: ${info.userAgent}
- ビューポート: ${info.viewport.width}x${info.viewport.height}
 
### ネットワークエラー
${JSON.stringify(info.networkErrors, null, 2)}
 
### コンソールエラー
${JSON.stringify(info.consoleErrors, null, 2)}
 
### パフォーマンスメトリクス
${JSON.stringify(info.performanceMetrics, null, 2)}
 
### 分析依頼
1. エラーの根本原因を推定してください
2. ネットワークエラーとコンソールエラーの関連性を分析してください
3. パフォーマンスに問題がある場合、その原因を指摘してください
4. 修正方法を具体的なコードで示してください
`;
  }
}

# AI 解析に適した構造化ログの設計
import structlog
import logging
from datetime import datetime
 
# structlog の設定
structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.add_log_level,
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer(),
    ],
)
 
logger = structlog.get_logger()
 
# 構造化ログの出力例
def process_order(order_id: str, user_id: str):
    log = logger.bind(order_id=order_id, user_id=user_id)
 
    log.info("order_processing_started", step="validation")
 
    try:
        # バリデーション
        order = validate_order(order_id)
        log.info("order_validated", items_count=len(order.items))
 
        # 支払い処理
        payment = process_payment(order)
        log.info("payment_processed",
                 amount=payment.amount,
                 method=payment.method,
                 duration_ms=payment.duration_ms)
 
        # 在庫確認
        inventory = check_inventory(order)
        log.info("inventory_checked",
                 all_available=inventory.all_available,
                 unavailable_items=inventory.unavailable_items)
 
    except PaymentError as e:
        log.error("payment_failed",
                  error_code=e.code,
                  error_message=str(e),
                  retry_possible=e.retryable)
        raise
 
    except Exception as e:
        log.error("order_processing_failed",
                  error_type=type(e).__name__,
                  error_message=str(e),
                  exc_info=True)
        raise

# 大量ログから異常パターンを AI で抽出
 
class LogAnalyzer:
    """AI を活用したログ分析エンジン"""
 
    def analyze_error_patterns(self, logs: list[dict],
                                time_window_minutes: int = 60) -> dict:
        """エラーログのパターンを分析"""
 
        # 1. エラーの集計
        error_groups = {}
        for log in logs:
            if log.get("level") in ("error", "critical"):
                key = f"{log.get('error_type', 'unknown')}:{log.get('error_message', '')[:100]}"
                error_groups.setdefault(key, []).append(log)
 
        # 2. エラーの時系列分析
        timeline = self._build_error_timeline(logs, time_window_minutes)
 
        # 3. 相関分析
        correlations = self._find_correlations(logs)
 
        return {
            "error_summary": {
                key: {
                    "count": len(entries),
                    "first_seen": entries[0].get("timestamp"),
                    "last_seen": entries[-1].get("timestamp"),
                    "sample": entries[0],
                }
                for key, entries in sorted(
                    error_groups.items(), key=lambda x: -len(x[1])
                )[:10]
            },
            "timeline": timeline,
            "correlations": correlations,
            "ai_analysis_prompt": self._build_analysis_prompt(
                error_groups, timeline, correlations
            ),
        }
 
    def _find_correlations(self, logs: list[dict]) -> list[dict]:
        """エラー間の相関関係を検出"""
        correlations = []
 
        # 同一リクエスト内のエラーチェーン
        request_logs = {}
        for log in logs:
            req_id = log.get("request_id")
            if req_id:
                request_logs.setdefault(req_id, []).append(log)
 
        for req_id, req_log_list in request_logs.items():
            errors = [l for l in req_log_list if l.get("level") == "error"]
            if len(errors) >= 2:
                correlations.append({
                    "type": "error_chain",
                    "request_id": req_id,
                    "errors": [e.get("error_message", "")[:100] for e in errors],
                    "root_cause_candidate": errors[0].get("error_message", ""),
                })
 
        return correlations
 
    def _build_analysis_prompt(self, error_groups, timeline, correlations) -> str:
        """AI に渡すログ分析プロンプトを構築"""
        return f"""
以下のログ分析結果から、根本原因と対策を提案してください。
 
## エラーサマリー（発生頻度順）
{json.dumps(list(error_groups.keys())[:10], ensure_ascii=False, indent=2)}
 
## 時系列パターン
{json.dumps(timeline, ensure_ascii=False, indent=2)}
 
## エラーの相関関係
{json.dumps(correlations[:5], ensure_ascii=False, indent=2)}
 
以下の形式で回答してください:
1. 根本原因の推定（最も可能性が高い順に）
2. 影響範囲の評価
3. 即座に実施すべき対策
4. 中長期的な改善策
"""

# OpenTelemetry + AI 分析の統合
 
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
 
tracer = trace.get_tracer("debug-service")
 
class DistributedDebugger:
    """分散トレーシングと AI を組み合わせたデバッグ"""
 
    def analyze_slow_request(self, trace_id: str) -> dict:
        """遅いリクエストのトレースを AI で分析"""
        spans = self._get_trace_spans(trace_id)
 
        # スパンの所要時間を分析
        bottlenecks = []
        for span in spans:
            duration_ms = (span["end_time"] - span["start_time"]) / 1_000_000
            if duration_ms > 100:  # 100ms 以上のスパン
                bottlenecks.append({
                    "service": span["service_name"],
                    "operation": span["operation_name"],
                    "duration_ms": duration_ms,
                    "attributes": span.get("attributes", {}),
                })
 
        # AI に分析を依頼
        analysis_prompt = f"""
以下の分散トレースを分析し、パフォーマンスのボトルネックを特定してください。
 
トレース ID: {trace_id}
総所要時間: {self._get_total_duration(spans)} ms
 
ボトルネック候補（100ms超のスパン）:
{json.dumps(bottlenecks, indent=2, ensure_ascii=False)}
 
スパン数: {len(spans)}
サービス数: {len(set(s['service_name'] for s in spans))}
"""
        return {
            "bottlenecks": bottlenecks,
            "analysis_prompt": analysis_prompt,
        }

# アプリケーションメトリクスの異常をAIで検知・分析
 
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from statistics import mean, stdev
from typing import Optional
 
@dataclass
class MetricPoint:
    """メトリクスデータポイント"""
    timestamp: datetime
    value: float
    labels: dict = field(default_factory=dict)
 
@dataclass
class AnomalyReport:
    """異常検知レポート"""
    metric_name: str
    anomaly_type: str  # "spike", "drop", "trend_change", "pattern_break"
    severity: str      # "critical", "warning", "info"
    detected_at: datetime
    current_value: float
    expected_range: tuple[float, float]
    context: dict = field(default_factory=dict)
 
class MetricsAnomalyDetector:
    """メトリクスの異常をAIで検知するシステム"""
 
    def __init__(self, lookback_window: int = 60):
        """
        Args:
            lookback_window: 異常検知の基準とする過去の分数
        """
        self.lookback_window = lookback_window
        self.metrics_history: dict[str, list[MetricPoint]] = {}
 
    def detect_anomalies(self, current_metrics: dict[str, float]) -> list[AnomalyReport]:
        """現在のメトリクスから異常を検知"""
        anomalies = []
 
        for metric_name, current_value in current_metrics.items():
            history = self.metrics_history.get(metric_name, [])
            if len(history) < 10:
                continue  # データ不足
 
            historical_values = [p.value for p in history]
            avg = mean(historical_values)
            std = stdev(historical_values) if len(historical_values) > 1 else 0
 
            # Z-score ベースの異常検知
            z_score = (current_value - avg) / std if std > 0 else 0
 
            if abs(z_score) > 3:
                anomaly_type = "spike" if z_score > 0 else "drop"
                severity = "critical" if abs(z_score) > 5 else "warning"
 
                anomalies.append(AnomalyReport(
                    metric_name=metric_name,
                    anomaly_type=anomaly_type,
                    severity=severity,
                    detected_at=datetime.now(),
                    current_value=current_value,
                    expected_range=(avg - 2 * std, avg + 2 * std),
                    context={
                        "z_score": z_score,
                        "historical_avg": avg,
                        "historical_std": std,
                        "data_points": len(historical_values),
                    },
                ))
 
        return anomalies
 
    def build_ai_analysis_prompt(self, anomalies: list[AnomalyReport]) -> str:
        """異常検知結果をAI分析用プロンプトに変換"""
        if not anomalies:
            return "異常は検知されませんでした。"
 
        anomaly_details = []
        for a in anomalies:
            anomaly_details.append(f"""
- メトリクス: {a.metric_name}
  種類: {a.anomaly_type} ({a.severity})
  現在値: {a.current_value:.2f}
  期待範囲: {a.expected_range[0]:.2f} - {a.expected_range[1]:.2f}
  Z-score: {a.context['z_score']:.2f}
""")
 
        return f"""
以下のメトリクス異常を分析し、原因と対策を提案してください。
 
## 検知された異常（{len(anomalies)}件）
{''.join(anomaly_details)}
 
## 分析依頼
1. 異常間の相関関係（複数の異常が同一原因に起因している可能性）
2. 推定される根本原因（最も可能性が高い順に）
3. 即座に確認すべき項目
4. 緩和策と恒久対策
"""

# CI で失敗したテストを AI で自動分析・修正提案
 
class AutoFixSuggester:
    """テスト失敗時の自動修正提案"""
 
    def analyze_test_failure(self, test_output: str, source_files: dict) -> dict:
        """テスト失敗の原因分析と修正提案"""
 
        prompt = f"""
以下のテスト失敗を分析し、修正を提案してください。
 
## テスト出力

# デバッグセッションのナレッジベース化
 
class DebugSessionRecorder:
    """デバッグセッションを記録してチームで共有"""
 
    def record_session(self, session: dict) -> dict:
        """デバッグセッションを構造化して記録"""
        record = {
            "id": generate_session_id(),
            "timestamp": datetime.now().isoformat(),
            "error": {
                "type": session["error_type"],
                "message": session["error_message"],
                "stacktrace": session["stacktrace"],
            },
            "investigation": {
                "hypotheses": session["hypotheses"],
                "steps_taken": session["investigation_steps"],
                "ai_suggestions": session["ai_suggestions"],
                "time_spent_minutes": session["time_spent"],
            },
            "resolution": {
                "root_cause": session["root_cause"],
                "fix_description": session["fix_description"],
                "fix_commit": session["commit_hash"],
                "tests_added": session["tests_added"],
            },
            "lessons": {
                "what_helped": session.get("what_helped", []),
                "what_hindered": session.get("what_hindered", []),
                "prevention_measures": session.get("prevention", []),
            },
            "tags": session.get("tags", []),
        }
        return record
 
    def search_similar_issues(self, error_message: str,
                               knowledge_base: list[dict]) -> list[dict]:
        """類似のデバッグ記録を検索"""
        # テキスト類似度に基づく検索
        results = []
        for record in knowledge_base:
            similarity = compute_text_similarity(
                error_message,
                record["error"]["message"]
            )
            if similarity > 0.6:
                results.append({
                    "record": record,
                    "similarity": similarity,
                    "resolution_summary": record["resolution"]["root_cause"],
                })
        return sorted(results, key=lambda x: -x["similarity"])[:5]

# GitHub Actions + AI デバッグの統合例
 
class CIDebugIntegration:
    """CI/CDパイプラインでのAIデバッグ自動化"""
 
    def analyze_ci_failure(self, workflow_run: dict) -> dict:
        """CIの失敗を自動分析してPRにコメント"""
 
        # 失敗したジョブの情報収集
        failed_jobs = [
            job for job in workflow_run["jobs"]
            if job["conclusion"] == "failure"
        ]
 
        analysis_results = []
        for job in failed_jobs:
            # ログの取得と解析
            log_content = self._fetch_job_logs(job["id"])
            error_sections = self._extract_error_sections(log_content)
 
            # 変更されたファイルとの照合
            changed_files = self._get_pr_changed_files(workflow_run["pr_number"])
            related_changes = self._correlate_errors_with_changes(
                error_sections, changed_files
            )
 
            analysis_results.append({
                "job_name": job["name"],
                "errors": error_sections,
                "related_changes": related_changes,
                "ai_prompt": self._build_ci_debug_prompt(
                    job, error_sections, related_changes
                ),
            })
 
        return {
            "workflow_run_id": workflow_run["id"],
            "analyses": analysis_results,
            "pr_comment": self._format_pr_comment(analysis_results),
        }
 
    def _build_ci_debug_prompt(self, job: dict, errors: list,
                                changes: list) -> str:
        """CI失敗のAI分析用プロンプトを構築"""
        return f"""
以下のCIジョブの失敗を分析してください。
 
## ジョブ情報
- 名前: {job['name']}
- ステップ: {job.get('failed_step', 'N/A')}
 
## エラー内容
{chr(10).join(errors[:5])}
 
## PR で変更されたファイル
{chr(10).join(f'- {c["filename"]} (+{c["additions"]} -{c["deletions"]})' for c in changes[:10])}
 
## 分析依頼
1. 失敗の原因（変更との関連）
2. 修正方法
3. ローカルでの再現手順
"""
 
    def _format_pr_comment(self, analyses: list) -> str:
        """PRコメント用のフォーマット"""
        comment = "## AI デバッグ分析結果\n\n"
        for analysis in analyses:
            comment += f"### {analysis['job_name']}\n"
            comment += f"**検出されたエラー**: {len(analysis['errors'])}件\n"
            if analysis['related_changes']:
                comment += f"**関連する変更ファイル**: "
                comment += ", ".join(
                    f"`{c['filename']}`" for c in analysis['related_changes'][:3]
                )
                comment += "\n"
            comment += "\n"
        return comment

# インシデント後のポストモーテム作成をAIで支援
 
POSTMORTEM_AI_PROMPT = """
以下のインシデント情報に基づいて、ポストモーテムのドラフトを作成してください。
 
## インシデント概要
- 発生日時: {incident_start}
- 復旧日時: {incident_end}
- 影響範囲: {impact}
- 深刻度: {severity}
 
## タイムライン
{timeline}
 
## 検知方法
{detection}
 
## 根本原因
{root_cause}
 
## 対応アクション
{actions_taken}
 
## テンプレート
以下の形式でポストモーテムを出力してください:
 
### 1. サマリー（3行以内）
### 2. インパクト（定量的に）
### 3. タイムライン（時系列）
### 4. 根本原因分析（5 Whys）
### 5. 修正内容
### 6. 再発防止策（短期・中期・長期）
### 7. 学んだこと（Good / Bad / Lucky）
"""

# 演習1: 基本実装のテンプレート
class Exercise1:
    """基本的な実装パターンの演習"""
 
    def __init__(self):
        self.data = []
 
    def validate_input(self, value):
        """入力値の検証"""
        if value is None:
            raise ValueError("入力値がNoneです")
        return True
 
    def process(self, value):
        """データ処理のメインロジック"""
        self.validate_input(value)
        self.data.append(value)
        return self.data
 
    def get_results(self):
        """処理結果の取得"""
        return {
            'count': len(self.data),
            'data': self.data
        }
 
# テスト
def test_exercise1():
    ex = Exercise1()
    assert ex.process(1) == [1]
    assert ex.process(2) == [1, 2]
    assert ex.get_results()['count'] == 2
 
    try:
        ex.process(None)
        assert False, "例外が発生するべき"
    except ValueError:
        pass
 
    print("全テスト合格!")
 
test_exercise1()

# 演習2: 応用パターン
from typing import List, Dict, Optional
from datetime import datetime
 
class AdvancedExercise:
    """応用パターンの演習"""
 
    def __init__(self, max_size: int = 100):
        self._items: List[Dict] = []
        self._max_size = max_size
        self._created_at = datetime.now()
 
    def add(self, key: str, value: any) -> bool:
        """アイテムの追加（サイズ制限付き）"""
        if len(self._items) >= self._max_size:
            return False
        self._items.append({
            'key': key,
            'value': value,
            'timestamp': datetime.now().isoformat()
        })
        return True
 
    def find(self, key: str) -> Optional[Dict]:
        """キーによる検索"""
        for item in reversed(self._items):
            if item['key'] == key:
                return item
        return None
 
    def remove(self, key: str) -> bool:
        """キーによる削除"""
        for i, item in enumerate(self._items):
            if item['key'] == key:
                self._items.pop(i)
                return True
        return False
 
    def stats(self) -> Dict:
        """統計情報"""
        return {
            'total_items': len(self._items),
            'max_size': self._max_size,
            'usage_percent': len(self._items) / self._max_size * 100,
            'uptime': str(datetime.now() - self._created_at)
        }
 
# テスト
def test_advanced():
    ex = AdvancedExercise(max_size=3)
    assert ex.add("a", 1) == True
    assert ex.add("b", 2) == True
    assert ex.add("c", 3) == True
    assert ex.add("d", 4) == False  # サイズ制限
    assert ex.find("b")['value'] == 2
    assert ex.remove("b") == True
    assert ex.find("b") is None
    stats = ex.stats()
    assert stats['total_items'] == 2
    print("応用テスト全合格!")
 
test_advanced()

# 演習3: パフォーマンス最適化
import time
from functools import lru_cache
 
# 最適化前（O(n^2)）
def slow_search(data: list, target: int) -> int:
    """非効率な検索"""
    for i in range(len(data)):
        for j in range(i + 1, len(data)):
            if data[i] + data[j] == target:
                return (i, j)
    return (-1, -1)
 
# 最適化後（O(n)）
def fast_search(data: list, target: int) -> tuple:
    """ハッシュマップを使った効率的な検索"""
    seen = {}
    for i, num in enumerate(data):
        complement = target - num
        if complement in seen:
            return (seen[complement], i)
        seen[num] = i
    return (-1, -1)
 
# ベンチマーク
def benchmark():
    import random
    data = list(range(5000))
    random.shuffle(data)
    target = data[100] + data[4000]
 
    start = time.time()
    result1 = slow_search(data, target)
    slow_time = time.time() - start
 
    start = time.time()
    result2 = fast_search(data, target)
    fast_time = time.time() - start
 
    print(f"非効率版: {slow_time:.4f}秒")
    print(f"効率版:   {fast_time:.6f}秒")
    print(f"高速化率: {slow_time/fast_time:.0f}倍")
 
benchmark()

# デバッグ用ユーティリティ
import logging
import traceback
from functools import wraps
 
# ロガーの設定
logging.basicConfig(
    level=logging.DEBUG,
    format='%(asctime)s [%(levelname)s] %(name)s: %(message)s'
)
logger = logging.getLogger(__name__)
 
def debug_decorator(func):
    """関数の入出力をログ出力するデコレータ"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        logger.debug(f"呼び出し: {func.__name__}(args={args}, kwargs={kwargs})")
        try:
            result = func(*args, **kwargs)
            logger.debug(f"戻り値: {func.__name__} -> {result}")
            return result
        except Exception as e:
            logger.error(f"例外発生: {func.__name__}: {e}")
            logger.error(traceback.format_exc())
            raise
    return wrapper
 
@debug_decorator
def process_data(items):
    """データ処理（デバッグ対象）"""
    if not items:
        raise ValueError("空のデータ")
    return [item * 2 for item in items]