audio_codebuddy/server/webhook-deploy.py

#!/usr/bin/env python3
"""
Webhook 部署脚本
接收 Gogs/GitHub webhook 请求，自动执行部署

用法:
1. 访问 https://book.rrbrr.com/webhook-deploy.py?secret=YOUR_SECRET
2. 或使用 curl: curl -X POST https://book.rrbrr.com/webhook-deploy.py -d "secret=YOUR_SECRET"
"""

import os
import sys
import hashlib
import hmac
import subprocess
from http.server import HTTPServer, BaseHTTPRequestHandler
import json

# 配置
WEBHOOK_SECRET = os.environ.get('WEBHOOK_SECRET', 'your-secret-key-change-me')
DEPLOY_SCRIPT = '/data/ai/deploy.sh'
LOG_FILE = '/tmp/webhook-deploy.log'

def log(msg):
    """写入日志"""
    timestamp = subprocess.check_output(['date', '+%Y-%m-%d %H:%M:%S']).decode().strip()
    with open(LOG_FILE, 'a') as f:
        f.write(f"[{timestamp}] {msg}\n")
    print(f"[{timestamp}] {msg}")

def verify_signature(secret, payload, signature):
    """验证请求签名"""
    if not signature:
        return False
    mac = hmac.new(secret.encode(), payload, hashlib.sha256)
    return mac.hexdigest() == signature.replace('sha256=', '')

def execute_deploy():
    """执行部署脚本"""
    try:
        log("开始执行部署...")

        # 切换到项目目录
        os.chdir('/data/ai/audio_codebuddy')

        # Git pull
        log("执行 git pull...")
        result = subprocess.run(['git', 'pull', 'origin', 'master'],
                              capture_output=True, text=True, timeout=60)
        log(f"Git pull 结果: {result.returncode}")
        if result.stdout:
            log(f"stdout: {result.stdout}")
        if result.stderr:
            log(f"stderr: {result.stderr}")

        # 执行部署脚本
        log("执行部署脚本...")
        result = subprocess.run(['bash', DEPLOY_SCRIPT],
                              capture_output=True, text=True, timeout=600)
        log(f"部署脚本返回: {result.returncode}")
        if result.stdout:
            log(f"stdout: {result.stdout[-2000:]}")  # 只保留最后2000字符
        if result.stderr:
            log(f"stderr: {result.stderr[-2000:]}")

        if result.returncode == 0:
            log("部署成功!")
            return True
        else:
            log(f"部署失败! 返回码: {result.returncode}")
            return False
    except subprocess.TimeoutExpired:
        log("部署超时!")
        return False
    except Exception as e:
        log(f"部署异常: {e}")
        return False

class WebhookHandler(BaseHTTPRequestHandler):
    def do_GET(self):
        """处理 GET 请求（健康检查）"""
        self.send_response(200)
        self.send_header('Content-type', 'text/plain')
        self.end_headers()
        self.wfile.write(b'Webhook is running!')

    def do_POST(self):
        """处理 POST 请求（webhook）"""
        # 读取请求体
        content_length = int(self.headers.get('Content-Length', 0))
        body = self.rfile.read(content_length)

        # 获取签名
        signature = self.headers.get('X-Hub-Signature-256') or \
                   self.headers.get('X-Gogs-Signature') or \
                   self.headers.get('X-Gitea-Signature')

        # 验证签名（如果有 secret）
        if WEBHOOK_SECRET != 'your-secret-key-change-me':
            if not verify_signature(WEBHOOK_SECRET, body, signature):
                log("签名验证失败!")
                self.send_response(401)
                self.send_header('Content-type', 'application/json')
                self.end_headers()
                self.wfile.write(json.dumps({'error': 'Invalid signature'}).encode())
                return

        # 解析 payload
        try:
            payload = json.loads(body)
            log(f"收到 webhook: {payload.get('ref', 'unknown')}")
        except:
            log(f"无法解析 payload: {body[:200]}")
            payload = {}

        # 执行部署
        success = execute_deploy()

        # 返回结果
        self.send_response(200 if success else 500)
        self.send_header('Content-type', 'application/json')
        self.end_headers()
        result = {'success': success, 'message': 'Deployment completed'}
        self.wfile.write(json.dumps(result).encode())

    def log_message(self, format, *args):
        """禁用默认日志"""
        pass

def main():
    port = int(os.environ.get('WEBHOOK_PORT', 8080))
    server = HTTPServer(('0.0.0.0', port), WebhookHandler)
    log(f"Webhook 服务启动在端口 {port}")
    print(f"Webhook 服务启动在端口 {port}")
    server.serve_forever()

if __name__ == '__main__':
    main()