Codex APIで構築するカスタム開発ワークフロー

1

pip install openai flask requests python-dotenv

1
2
3
4

export OPENAI_API_KEY="sk-proj-..."
export GITHUB_TOKEN="ghp_..."
export SLACK_WEBHOOK_URL="https://hooks.slack.com/services/..."
export DISCORD_WEBHOOK_URL="https://discord.com/api/webhooks/..."

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104

from flask import Flask, request, jsonify
from openai import OpenAI
import requests
import os
import json

app = Flask(__name__)
client = OpenAI()

# Issue分類用のシステムプロンプト
TRIAGE_SYSTEM_PROMPT = """あなたはソフトウェア開発プロジェクトのIssueトリアージ担当者です。
与えられたIssueの内容を分析し、以下の情報をJSON形式で出力してください。

1. labels: 適切なラベル（配列）
   - bug: バグ報告
   - feature: 新機能リクエスト
   - documentation: ドキュメント関連
   - enhancement: 機能改善
   - question: 質問
   - security: セキュリティ関連
   - performance: パフォーマンス関連

2. priority: 優先度（high/medium/low）

3. estimated_effort: 見積もり工数（small/medium/large）

4. suggested_assignee_skill: 必要なスキル（frontend/backend/devops/fullstack）

5. summary: 50文字以内の要約

出力は必ず有効なJSONのみを返してください。"""


@app.route('/webhook/issues', methods=['POST'])
def handle_issue_webhook():
    """GitHub Issue Webhookを処理"""
    payload = request.json
    
    # Issue作成イベントのみ処理
    if payload.get('action') != 'opened':
        return jsonify({'status': 'ignored'}), 200
    
    issue = payload['issue']
    repository = payload['repository']
    
    # Issueの内容を分析
    triage_result = analyze_issue(issue)
    
    # GitHubにラベルを適用
    apply_labels(
        repository['full_name'],
        issue['number'],
        triage_result['labels']
    )
    
    # 通知を送信
    send_triage_notification(issue, triage_result, repository)
    
    return jsonify({
        'status': 'success',
        'triage_result': triage_result
    }), 200


def analyze_issue(issue: dict) -> dict:
    """Codex APIを使用してIssueを分析"""
    issue_content = f"""
    タイトル: {issue['title']}
    
    本文:
    {issue['body'] or '(本文なし)'}
    """
    
    response = client.responses.create(
        model="codex-mini-latest",
        instructions=TRIAGE_SYSTEM_PROMPT,
        input=issue_content,
        temperature=0.2,
        text={
            "format": {
                "type": "json_object"
            }
        }
    )
    
    return json.loads(response.output_text)


def apply_labels(repo_full_name: str, issue_number: int, labels: list):
    """GitHubにラベルを適用"""
    github_token = os.environ['GITHUB_TOKEN']
    url = f"https://api.github.com/repos/{repo_full_name}/issues/{issue_number}/labels"
    
    headers = {
        'Authorization': f'token {github_token}',
        'Accept': 'application/vnd.github+json'
    }
    
    response = requests.post(url, headers=headers, json={'labels': labels})
    return response.status_code == 200


if __name__ == '__main__':
    app.run(port=5000)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

# プロジェクト固有のトリアージ設定
TRIAGE_CONFIG = {
    "label_mappings": {
        "frontend": ["react", "css", "ui", "ux", "component"],
        "backend": ["api", "database", "server", "endpoint"],
        "devops": ["ci", "cd", "deploy", "docker", "kubernetes"],
        "security": ["vulnerability", "auth", "permission", "xss", "injection"]
    },
    "priority_keywords": {
        "high": ["critical", "urgent", "blocker", "crash", "data loss"],
        "medium": ["important", "should", "expected"],
        "low": ["nice to have", "minor", "cosmetic"]
    },
    "auto_assign": {
        "frontend": ["frontend-team"],
        "backend": ["backend-team"],
        "devops": ["devops-team"],
        "security": ["security-team"]
    }
}


def get_enhanced_triage_prompt(config: dict) -> str:
    """プロジェクト設定に基づいた拡張トリアージプロンプトを生成"""
    return f"""あなたはソフトウェア開発プロジェクトのIssueトリアージ担当者です。
以下のプロジェクト固有のルールに従ってIssueを分類してください。

## ラベルマッピング
{json.dumps(config['label_mappings'], ensure_ascii=False, indent=2)}

## 優先度キーワード
{json.dumps(config['priority_keywords'], ensure_ascii=False, indent=2)}

## 自動アサイン設定
{json.dumps(config['auto_assign'], ensure_ascii=False, indent=2)}

出力は以下のJSON形式で返してください:
{{
    "labels": ["ラベル1", "ラベル2"],
    "priority": "high|medium|low",
    "area": "frontend|backend|devops|security|other",
    "suggested_teams": ["チーム名"],
    "summary": "50文字以内の要約"
}}
"""

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198

from dataclasses import dataclass
from typing import Optional
import base64

@dataclass
class ReviewComment:
    """レビューコメントを表すデータクラス"""
    path: str
    line: int
    body: str
    severity: str  # error, warning, suggestion


@dataclass
class ReviewResult:
    """レビュー結果を表すデータクラス"""
    approved: bool
    summary: str
    comments: list[ReviewComment]
    security_issues: list[str]
    performance_issues: list[str]


# コードレビュー用のシステムプロンプト
CODE_REVIEW_PROMPT = """あなたは経験豊富なシニアソフトウェアエンジニアです。
提出されたコードの差分をレビューし、以下の観点から問題点や改善点を指摘してください。

## レビュー観点
1. コードの品質と可読性
2. バグの可能性
3. セキュリティ上の問題
4. パフォーマンスの懸念
5. ベストプラクティスへの準拠
6. テストの網羅性

## 出力形式（JSON）
{
    "approved": true/false,
    "summary": "全体的な評価コメント",
    "comments": [
        {
            "path": "ファイルパス",
            "line": 行番号,
            "body": "指摘内容",
            "severity": "error|warning|suggestion"
        }
    ],
    "security_issues": ["セキュリティ上の問題点"],
    "performance_issues": ["パフォーマンス上の問題点"]
}

厳密にJSONのみを出力してください。"""


@app.route('/webhook/pull_request', methods=['POST'])
def handle_pr_webhook():
    """GitHub Pull Request Webhookを処理"""
    payload = request.json
    
    # PR作成・更新イベントのみ処理
    if payload.get('action') not in ['opened', 'synchronize']:
        return jsonify({'status': 'ignored'}), 200
    
    pr = payload['pull_request']
    repository = payload['repository']
    
    # PRの変更内容を取得
    diff_content = get_pr_diff(repository['full_name'], pr['number'])
    
    # Codex APIでコードレビュー
    review_result = review_code(diff_content, pr)
    
    # レビュー結果をGitHubに投稿
    post_review(repository['full_name'], pr['number'], review_result)
    
    # 通知を送信
    send_review_notification(pr, review_result, repository)
    
    return jsonify({
        'status': 'success',
        'review_result': {
            'approved': review_result.approved,
            'summary': review_result.summary
        }
    }), 200


def get_pr_diff(repo_full_name: str, pr_number: int) -> str:
    """PRの差分を取得"""
    github_token = os.environ['GITHUB_TOKEN']
    url = f"https://api.github.com/repos/{repo_full_name}/pulls/{pr_number}"
    
    headers = {
        'Authorization': f'token {github_token}',
        'Accept': 'application/vnd.github.diff'
    }
    
    response = requests.get(url, headers=headers)
    return response.text


def get_pr_files(repo_full_name: str, pr_number: int) -> list:
    """PRで変更されたファイル一覧を取得"""
    github_token = os.environ['GITHUB_TOKEN']
    url = f"https://api.github.com/repos/{repo_full_name}/pulls/{pr_number}/files"
    
    headers = {
        'Authorization': f'token {github_token}',
        'Accept': 'application/vnd.github+json'
    }
    
    response = requests.get(url, headers=headers)
    return response.json()


def review_code(diff_content: str, pr: dict) -> ReviewResult:
    """Codex APIを使用してコードレビューを実行"""
    
    # PRの情報とdiffを組み合わせたプロンプト
    review_input = f"""
    ## Pull Request情報
    タイトル: {pr['title']}
    説明: {pr['body'] or '(説明なし)'}
    ベースブランチ: {pr['base']['ref']}
    ヘッドブランチ: {pr['head']['ref']}
    
    ## コード差分
    {diff_content[:50000]}  # トークン制限を考慮
    """
    
    response = client.responses.create(
        model="codex-mini-latest",
        instructions=CODE_REVIEW_PROMPT,
        input=review_input,
        temperature=0.1,
        max_output_tokens=4096,
        text={
            "format": {
                "type": "json_object"
            }
        }
    )
    
    result = json.loads(response.output_text)
    
    return ReviewResult(
        approved=result['approved'],
        summary=result['summary'],
        comments=[
            ReviewComment(**comment) for comment in result.get('comments', [])
        ],
        security_issues=result.get('security_issues', []),
        performance_issues=result.get('performance_issues', [])
    )


def post_review(repo_full_name: str, pr_number: int, review: ReviewResult):
    """GitHubにレビュー結果を投稿"""
    github_token = os.environ['GITHUB_TOKEN']
    url = f"https://api.github.com/repos/{repo_full_name}/pulls/{pr_number}/reviews"
    
    headers = {
        'Authorization': f'token {github_token}',
        'Accept': 'application/vnd.github+json'
    }
    
    # レビュー本文を構築
    body_parts = [f"## 自動コードレビュー結果\n\n{review.summary}"]
    
    if review.security_issues:
        body_parts.append("\n### セキュリティ上の懸念")
        for issue in review.security_issues:
            body_parts.append(f"- {issue}")
    
    if review.performance_issues:
        body_parts.append("\n### パフォーマンス上の懸念")
        for issue in review.performance_issues:
            body_parts.append(f"- {issue}")
    
    # レビューコメントを整形
    comments = [
        {
            'path': comment.path,
            'line': comment.line,
            'body': f"**[{comment.severity.upper()}]** {comment.body}"
        }
        for comment in review.comments
    ]
    
    # レビューを投稿
    review_data = {
        'body': '\n'.join(body_parts),
        'event': 'APPROVE' if review.approved else 'REQUEST_CHANGES',
        'comments': comments
    }
    
    response = requests.post(url, headers=headers, json=review_data)
    return response.status_code == 200

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

# プロジェクト固有のレビュー設定
REVIEW_CONFIG = {
    "coding_standards": {
        "python": {
            "max_line_length": 88,
            "max_function_length": 50,
            "required_docstrings": True,
            "type_hints_required": True
        },
        "typescript": {
            "max_line_length": 100,
            "prefer_const": True,
            "strict_null_checks": True
        }
    },
    "security_checks": [
        "SQL injection",
        "XSS",
        "CSRF",
        "Hardcoded credentials",
        "Insecure dependencies"
    ],
    "auto_approve_conditions": {
        "max_files_changed": 3,
        "max_lines_changed": 100,
        "allowed_file_patterns": ["*.md", "*.txt", "*.json"]
    }
}


def get_project_review_prompt(config: dict, file_extension: str) -> str:
    """プロジェクト設定に基づいたレビュープロンプトを生成"""
    language_config = config['coding_standards'].get(
        file_extension.lstrip('.'),
        {}
    )
    
    return f"""あなたは経験豊富なシニアソフトウェアエンジニアです。
以下のプロジェクト固有のルールに従ってコードレビューを行ってください。

## コーディング規約
{json.dumps(language_config, ensure_ascii=False, indent=2)}

## 必須セキュリティチェック
{json.dumps(config['security_checks'], ensure_ascii=False, indent=2)}

## 自動承認条件
以下の条件をすべて満たす場合は自動承認を推奨:
{json.dumps(config['auto_approve_conditions'], ensure_ascii=False, indent=2)}

レビュー結果はJSON形式で出力してください。
"""

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311

from abc import ABC, abstractmethod
from enum import Enum


class NotificationType(Enum):
    ISSUE_TRIAGED = "issue_triaged"
    PR_REVIEWED = "pr_reviewed"
    SECURITY_ALERT = "security_alert"


class NotificationService(ABC):
    """通知サービスの抽象基底クラス"""
    
    @abstractmethod
    def send(self, notification_type: NotificationType, data: dict) -> bool:
        pass


class SlackNotificationService(NotificationService):
    """Slack通知サービス"""
    
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url
    
    def send(self, notification_type: NotificationType, data: dict) -> bool:
        message = self._build_message(notification_type, data)
        response = requests.post(self.webhook_url, json=message)
        return response.status_code == 200
    
    def _build_message(self, notification_type: NotificationType, data: dict) -> dict:
        if notification_type == NotificationType.ISSUE_TRIAGED:
            return self._build_triage_message(data)
        elif notification_type == NotificationType.PR_REVIEWED:
            return self._build_review_message(data)
        elif notification_type == NotificationType.SECURITY_ALERT:
            return self._build_security_alert_message(data)
    
    def _build_triage_message(self, data: dict) -> dict:
        issue = data['issue']
        triage = data['triage_result']
        repo = data['repository']
        
        priority_emoji = {
            'high': ':red_circle:',
            'medium': ':large_orange_circle:',
            'low': ':large_green_circle:'
        }
        
        return {
            "blocks": [
                {
                    "type": "header",
                    "text": {
                        "type": "plain_text",
                        "text": "New Issue Triaged"
                    }
                },
                {
                    "type": "section",
                    "fields": [
                        {
                            "type": "mrkdwn",
                            "text": f"*Repository:*\n{repo['full_name']}"
                        },
                        {
                            "type": "mrkdwn",
                            "text": f"*Issue:*\n<{issue['html_url']}|#{issue['number']}>"
                        },
                        {
                            "type": "mrkdwn",
                            "text": f"*Priority:*\n{priority_emoji.get(triage['priority'], '')} {triage['priority']}"
                        },
                        {
                            "type": "mrkdwn",
                            "text": f"*Labels:*\n{', '.join(triage['labels'])}"
                        }
                    ]
                },
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": f"*Summary:* {triage['summary']}"
                    }
                }
            ]
        }
    
    def _build_review_message(self, data: dict) -> dict:
        pr = data['pull_request']
        review = data['review_result']
        repo = data['repository']
        
        status_emoji = ':white_check_mark:' if review.approved else ':x:'
        
        blocks = [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": "PR Review Completed"
                }
            },
            {
                "type": "section",
                "fields": [
                    {
                        "type": "mrkdwn",
                        "text": f"*Repository:*\n{repo['full_name']}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*PR:*\n<{pr['html_url']}|#{pr['number']}>"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*Status:*\n{status_emoji} {'Approved' if review.approved else 'Changes Requested'}"
                    },
                    {
                        "type": "mrkdwn",
                        "text": f"*Comments:*\n{len(review.comments)} issues found"
                    }
                ]
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Summary:* {review.summary}"
                }
            }
        ]
        
        # セキュリティ問題がある場合は警告を追加
        if review.security_issues:
            blocks.append({
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f":warning: *Security Issues:*\n" + '\n'.join(f"• {issue}" for issue in review.security_issues)
                }
            })
        
        return {"blocks": blocks}
    
    def _build_security_alert_message(self, data: dict) -> dict:
        return {
            "blocks": [
                {
                    "type": "header",
                    "text": {
                        "type": "plain_text",
                        "text": ":rotating_light: Security Alert"
                    }
                },
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": f"*{data['title']}*\n{data['description']}"
                    }
                },
                {
                    "type": "section",
                    "text": {
                        "type": "mrkdwn",
                        "text": f"<{data['url']}|View Details>"
                    }
                }
            ]
        }


class DiscordNotificationService(NotificationService):
    """Discord通知サービス"""
    
    def __init__(self, webhook_url: str):
        self.webhook_url = webhook_url
    
    def send(self, notification_type: NotificationType, data: dict) -> bool:
        message = self._build_message(notification_type, data)
        response = requests.post(self.webhook_url, json=message)
        return response.status_code in [200, 204]
    
    def _build_message(self, notification_type: NotificationType, data: dict) -> dict:
        if notification_type == NotificationType.ISSUE_TRIAGED:
            return self._build_triage_embed(data)
        elif notification_type == NotificationType.PR_REVIEWED:
            return self._build_review_embed(data)
        elif notification_type == NotificationType.SECURITY_ALERT:
            return self._build_security_alert_embed(data)
    
    def _build_triage_embed(self, data: dict) -> dict:
        issue = data['issue']
        triage = data['triage_result']
        repo = data['repository']
        
        priority_color = {
            'high': 0xFF0000,
            'medium': 0xFFA500,
            'low': 0x00FF00
        }
        
        return {
            "embeds": [{
                "title": f"Issue Triaged: {issue['title']}",
                "url": issue['html_url'],
                "color": priority_color.get(triage['priority'], 0x808080),
                "fields": [
                    {"name": "Repository", "value": repo['full_name'], "inline": True},
                    {"name": "Priority", "value": triage['priority'], "inline": True},
                    {"name": "Labels", "value": ', '.join(triage['labels']), "inline": False},
                    {"name": "Summary", "value": triage['summary'], "inline": False}
                ],
                "footer": {"text": "Codex Auto-Triage"}
            }]
        }
    
    def _build_review_embed(self, data: dict) -> dict:
        pr = data['pull_request']
        review = data['review_result']
        repo = data['repository']
        
        color = 0x00FF00 if review.approved else 0xFF0000
        
        return {
            "embeds": [{
                "title": f"PR Review: {pr['title']}",
                "url": pr['html_url'],
                "color": color,
                "fields": [
                    {"name": "Repository", "value": repo['full_name'], "inline": True},
                    {"name": "Status", "value": "Approved" if review.approved else "Changes Requested", "inline": True},
                    {"name": "Issues Found", "value": str(len(review.comments)), "inline": True},
                    {"name": "Summary", "value": review.summary, "inline": False}
                ],
                "footer": {"text": "Codex Auto-Review"}
            }]
        }
    
    def _build_security_alert_embed(self, data: dict) -> dict:
        return {
            "embeds": [{
                "title": f"Security Alert: {data['title']}",
                "url": data['url'],
                "color": 0xFF0000,
                "description": data['description'],
                "footer": {"text": "Codex Security Scanner"}
            }]
        }


# 通知マネージャー
class NotificationManager:
    """複数の通知サービスを管理"""
    
    def __init__(self):
        self.services: list[NotificationService] = []
    
    def add_service(self, service: NotificationService):
        self.services.append(service)
    
    def notify_all(self, notification_type: NotificationType, data: dict):
        results = []
        for service in self.services:
            try:
                result = service.send(notification_type, data)
                results.append(result)
            except Exception as e:
                print(f"Notification failed: {e}")
                results.append(False)
        return all(results)


# 通知ヘルパー関数
def send_triage_notification(issue: dict, triage_result: dict, repository: dict):
    """トリアージ結果の通知を送信"""
    manager = NotificationManager()
    
    slack_url = os.environ.get('SLACK_WEBHOOK_URL')
    if slack_url:
        manager.add_service(SlackNotificationService(slack_url))
    
    discord_url = os.environ.get('DISCORD_WEBHOOK_URL')
    if discord_url:
        manager.add_service(DiscordNotificationService(discord_url))
    
    manager.notify_all(NotificationType.ISSUE_TRIAGED, {
        'issue': issue,
        'triage_result': triage_result,
        'repository': repository
    })


def send_review_notification(pr: dict, review_result: ReviewResult, repository: dict):
    """レビュー結果の通知を送信"""
    manager = NotificationManager()
    
    slack_url = os.environ.get('SLACK_WEBHOOK_URL')
    if slack_url:
        manager.add_service(SlackNotificationService(slack_url))
    
    discord_url = os.environ.get('DISCORD_WEBHOOK_URL')
    if discord_url:
        manager.add_service(DiscordNotificationService(discord_url))
    
    manager.notify_all(NotificationType.PR_REVIEWED, {
        'pull_request': pr,
        'review_result': review_result,
        'repository': repository
    })

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179

from datetime import datetime, timedelta
from collections import defaultdict
import sqlite3


class MetricsCollector:
    """メトリクス収集クラス"""
    
    def __init__(self, db_path: str = 'metrics.db'):
        self.db_path = db_path
        self._init_db()
    
    def _init_db(self):
        """データベースを初期化"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS triage_events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                repository TEXT,
                issue_number INTEGER,
                priority TEXT,
                labels TEXT,
                processing_time_ms INTEGER
            )
        ''')
        
        cursor.execute('''
            CREATE TABLE IF NOT EXISTS review_events (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                repository TEXT,
                pr_number INTEGER,
                approved BOOLEAN,
                issues_found INTEGER,
                security_issues INTEGER,
                processing_time_ms INTEGER
            )
        ''')
        
        conn.commit()
        conn.close()
    
    def record_triage(self, repository: str, issue_number: int, 
                      priority: str, labels: list, processing_time_ms: int):
        """トリアージイベントを記録"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO triage_events (repository, issue_number, priority, labels, processing_time_ms)
            VALUES (?, ?, ?, ?, ?)
        ''', (repository, issue_number, priority, ','.join(labels), processing_time_ms))
        
        conn.commit()
        conn.close()
    
    def record_review(self, repository: str, pr_number: int, approved: bool,
                      issues_found: int, security_issues: int, processing_time_ms: int):
        """レビューイベントを記録"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        cursor.execute('''
            INSERT INTO review_events (repository, pr_number, approved, issues_found, security_issues, processing_time_ms)
            VALUES (?, ?, ?, ?, ?, ?)
        ''', (repository, pr_number, approved, issues_found, security_issues, processing_time_ms))
        
        conn.commit()
        conn.close()
    
    def get_dashboard_data(self, days: int = 30) -> dict:
        """ダッシュボード用のデータを取得"""
        conn = sqlite3.connect(self.db_path)
        cursor = conn.cursor()
        
        since = datetime.now() - timedelta(days=days)
        since_str = since.isoformat()
        
        # トリアージ統計
        cursor.execute('''
            SELECT 
                COUNT(*) as total,
                AVG(processing_time_ms) as avg_time,
                priority,
                COUNT(*) as count
            FROM triage_events
            WHERE timestamp > ?
            GROUP BY priority
        ''', (since_str,))
        triage_stats = cursor.fetchall()
        
        # レビュー統計
        cursor.execute('''
            SELECT 
                COUNT(*) as total,
                SUM(CASE WHEN approved THEN 1 ELSE 0 END) as approved_count,
                AVG(issues_found) as avg_issues,
                AVG(processing_time_ms) as avg_time
            FROM review_events
            WHERE timestamp > ?
        ''', (since_str,))
        review_stats = cursor.fetchone()
        
        # 日別トレンド
        cursor.execute('''
            SELECT 
                DATE(timestamp) as date,
                COUNT(*) as triage_count
            FROM triage_events
            WHERE timestamp > ?
            GROUP BY DATE(timestamp)
            ORDER BY date
        ''', (since_str,))
        triage_trend = cursor.fetchall()
        
        cursor.execute('''
            SELECT 
                DATE(timestamp) as date,
                COUNT(*) as review_count,
                SUM(CASE WHEN approved THEN 1 ELSE 0 END) as approved_count
            FROM review_events
            WHERE timestamp > ?
            GROUP BY DATE(timestamp)
            ORDER BY date
        ''', (since_str,))
        review_trend = cursor.fetchall()
        
        conn.close()
        
        return {
            'triage': {
                'by_priority': {row[2]: row[3] for row in triage_stats},
                'total': sum(row[3] for row in triage_stats),
                'avg_processing_time_ms': triage_stats[0][1] if triage_stats else 0
            },
            'review': {
                'total': review_stats[0] if review_stats else 0,
                'approved_count': review_stats[1] if review_stats else 0,
                'approval_rate': (review_stats[1] / review_stats[0] * 100) if review_stats and review_stats[0] > 0 else 0,
                'avg_issues_found': review_stats[2] if review_stats else 0,
                'avg_processing_time_ms': review_stats[3] if review_stats else 0
            },
            'trends': {
                'triage': [{'date': row[0], 'count': row[1]} for row in triage_trend],
                'review': [{'date': row[0], 'count': row[1], 'approved': row[2]} for row in review_trend]
            }
        }


# ダッシュボードAPIエンドポイント
metrics_collector = MetricsCollector()


@app.route('/api/dashboard', methods=['GET'])
def get_dashboard():
    """ダッシュボードデータを取得"""
    days = request.args.get('days', 30, type=int)
    data = metrics_collector.get_dashboard_data(days)
    return jsonify(data)


@app.route('/api/dashboard/summary', methods=['GET'])
def get_summary():
    """サマリー情報を取得"""
    data = metrics_collector.get_dashboard_data(7)
    
    summary = {
        'issues_triaged_this_week': data['triage']['total'],
        'prs_reviewed_this_week': data['review']['total'],
        'approval_rate': data['review']['approval_rate'],
        'high_priority_issues': data['triage']['by_priority'].get('high', 0),
        'avg_triage_time_seconds': data['triage']['avg_processing_time_ms'] / 1000,
        'avg_review_time_seconds': data['review']['avg_processing_time_ms'] / 1000
    }
    
    return jsonify(summary)

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114

@app.route('/dashboard')
def dashboard():
    """ダッシュボードページを表示"""
    return '''
<!DOCTYPE html>
<html>
<head>
    <title>Codex Automation Dashboard</title>
    <script src="https://cdn.jsdelivr.net/npm/chart.js"></script>
    <style>
        body { font-family: Arial, sans-serif; margin: 20px; background: #f5f5f5; }
        .container { max-width: 1200px; margin: 0 auto; }
        .header { text-align: center; margin-bottom: 30px; }
        .cards { display: grid; grid-template-columns: repeat(auto-fit, minmax(200px, 1fr)); gap: 20px; margin-bottom: 30px; }
        .card { background: white; padding: 20px; border-radius: 8px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); }
        .card h3 { margin: 0 0 10px 0; color: #666; font-size: 14px; }
        .card .value { font-size: 32px; font-weight: bold; color: #333; }
        .charts { display: grid; grid-template-columns: 1fr 1fr; gap: 20px; }
        .chart-container { background: white; padding: 20px; border-radius: 8px; box-shadow: 0 2px 4px rgba(0,0,0,0.1); }
    </style>
</head>
<body>
    <div class="container">
        <div class="header">
            <h1>Codex Automation Dashboard</h1>
        </div>
        
        <div class="cards">
            <div class="card">
                <h3>Issues Triaged (7 days)</h3>
                <div class="value" id="issues-triaged">-</div>
            </div>
            <div class="card">
                <h3>PRs Reviewed (7 days)</h3>
                <div class="value" id="prs-reviewed">-</div>
            </div>
            <div class="card">
                <h3>Approval Rate</h3>
                <div class="value" id="approval-rate">-</div>
            </div>
            <div class="card">
                <h3>Avg Triage Time</h3>
                <div class="value" id="avg-triage-time">-</div>
            </div>
        </div>
        
        <div class="charts">
            <div class="chart-container">
                <h3>Triage Trend</h3>
                <canvas id="triageChart"></canvas>
            </div>
            <div class="chart-container">
                <h3>Review Trend</h3>
                <canvas id="reviewChart"></canvas>
            </div>
        </div>
    </div>
    
    <script>
        async function loadDashboard() {
            const summaryRes = await fetch('/api/dashboard/summary');
            const summary = await summaryRes.json();
            
            document.getElementById('issues-triaged').textContent = summary.issues_triaged_this_week;
            document.getElementById('prs-reviewed').textContent = summary.prs_reviewed_this_week;
            document.getElementById('approval-rate').textContent = summary.approval_rate.toFixed(1) + '%';
            document.getElementById('avg-triage-time').textContent = summary.avg_triage_time_seconds.toFixed(1) + 's';
            
            const dashboardRes = await fetch('/api/dashboard?days=30');
            const dashboard = await dashboardRes.json();
            
            // Triage Chart
            new Chart(document.getElementById('triageChart'), {
                type: 'line',
                data: {
                    labels: dashboard.trends.triage.map(d => d.date),
                    datasets: [{
                        label: 'Issues Triaged',
                        data: dashboard.trends.triage.map(d => d.count),
                        borderColor: '#4CAF50',
                        tension: 0.1
                    }]
                }
            });
            
            // Review Chart
            new Chart(document.getElementById('reviewChart'), {
                type: 'line',
                data: {
                    labels: dashboard.trends.review.map(d => d.date),
                    datasets: [
                        {
                            label: 'PRs Reviewed',
                            data: dashboard.trends.review.map(d => d.count),
                            borderColor: '#2196F3',
                            tension: 0.1
                        },
                        {
                            label: 'Approved',
                            data: dashboard.trends.review.map(d => d.approved),
                            borderColor: '#4CAF50',
                            tension: 0.1
                        }
                    ]
                }
            });
        }
        
        loadDashboard();
        setInterval(loadDashboard, 60000); // 1分ごとに更新
    </script>
</body>
</html>
    '''

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

import hmac
import hashlib


def verify_github_signature(payload: bytes, signature: str, secret: str) -> bool:
    """GitHub Webhookの署名を検証"""
    expected_signature = 'sha256=' + hmac.new(
        secret.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(expected_signature, signature)


@app.before_request
def verify_webhook():
    """Webhookリクエストの署名を検証"""
    if request.path.startswith('/webhook/'):
        signature = request.headers.get('X-Hub-Signature-256')
        secret = os.environ.get('GITHUB_WEBHOOK_SECRET')
        
        if not signature or not secret:
            return jsonify({'error': 'Unauthorized'}), 401
        
        if not verify_github_signature(request.data, signature, secret):
            return jsonify({'error': 'Invalid signature'}), 401