lawless-client/scripts/webhook-server.py

#!/usr/bin/env python3
"""
Gitea Webhook 接收器 - 自动部署文档站
使用方法: python3 webhook-server.py
"""

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

# 配置
WEBHOOK_SECRET = os.environ.get('WEBHOOK_SECRET', 'lawless-docs-deploy')
DEPLOY_DIR = os.environ.get('DEPLOY_DIR', '/var/www/docs.xuqinmin.com/site')
REPO_URL = os.environ.get('REPO_URL', 'https://xuqinmin.com/xuqinmin12/lawless.git')
BRANCH = os.environ.get('BRANCH', 'main')
PORT = int(os.environ.get('WEBHOOK_PORT', '9000'))

def log(message):
    """日志输出"""
    print(f"[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}] {message}", flush=True)

def verify_signature(payload, signature):
    """验证 Gitea webhook 签名"""
    if not signature:
        return False
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(signature, expected)

def deploy():
    """执行部署"""
    log("Starting deployment...")

    tmp_dir = tempfile.mkdtemp()
    try:
        # 克隆仓库
        log("Cloning repository...")
        subprocess.run(
            ['git', 'clone', '--depth', '1', '--branch', BRANCH, REPO_URL, 'repo'],
            cwd=tmp_dir,
            check=True,
            capture_output=True
        )

        repo_dir = os.path.join(tmp_dir, 'repo')

        # 安装依赖
        log("Installing dependencies...")
        subprocess.run(
            ['pip3', 'install', 'mkdocs-material', '--quiet'],
            cwd=repo_dir,
            capture_output=True
        )

        # 构建文档
        log("Building docs...")
        subprocess.run(
            ['mkdocs', 'build', '--clean', '--site-dir', 'site'],
            cwd=repo_dir,
            check=True,
            capture_output=True
        )

        site_dir = os.path.join(repo_dir, 'site')

        # 部署到目标目录
        log(f"Deploying to {DEPLOY_DIR}...")
        os.makedirs(DEPLOY_DIR, exist_ok=True)
        subprocess.run(
            ['sudo', 'rsync', '-avz', '--delete', site_dir + '/', DEPLOY_DIR + '/'],
            check=True,
            capture_output=True
        )

        log("Deployment completed successfully!")
        return True

    except Exception as e:
        log(f"Deployment failed: {e}")
        return False

    finally:
        # 清理临时目录
        shutil.rmtree(tmp_dir, ignore_errors=True)

class WebhookHandler(BaseHTTPRequestHandler):
    """HTTP 请求处理器"""

    def do_POST(self):
        """处理 POST 请求"""
        if self.path == '/deploy':
            # 读取请求体
            content_length = int(self.headers.get('Content-Length', 0))
            payload = self.rfile.read(content_length)

            # 获取签名
            signature = self.headers.get('X-Gitea-Signature', '')

            # 验证签名
            if not verify_signature(payload, signature):
                self.send_response(403)
                self.send_header('Content-Type', 'text/plain')
                self.end_headers()
                self.wfile.write(b'Invalid signature')
                return

            # 解析 payload
            try:
                data = json.loads(payload)
                ref = data.get('ref', '')

                # 检查是否是目标分支
                if ref == f'refs/heads/{BRANCH}':
                    log(f"Push to {BRANCH} detected, triggering deploy...")
                    # 异步执行部署
                    import threading
                    threading.Thread(target=deploy).start()

                    self.send_response(200)
                    self.send_header('Content-Type', 'text/plain')
                    self.end_headers()
                    self.wfile.write(b'Deployment triggered')
                else:
                    log(f"Ignoring push to {ref}")
                    self.send_response(200)
                    self.send_header('Content-Type', 'text/plain')
                    self.end_headers()
                    self.wfile.write(b'Ignored')

            except json.JSONDecodeError:
                self.send_response(400)
                self.send_header('Content-Type', 'text/plain')
                self.end_headers()
                self.wfile.write(b'Invalid JSON')

        elif self.path == '/health':
            self.send_response(200)
            self.send_header('Content-Type', 'text/plain')
            self.end_headers()
            self.wfile.write(b'OK')

        else:
            self.send_response(404)
            self.send_header('Content-Type', 'text/plain')
            self.end_headers()
            self.wfile.write(b'Not Found')

    def do_GET(self):
        """处理 GET 请求"""
        if self.path == '/health':
            self.send_response(200)
            self.send_header('Content-Type', 'text/plain')
            self.end_headers()
            self.wfile.write(b'OK')
        else:
            self.send_response(404)
            self.send_header('Content-Type', 'text/plain')
            self.end_headers()
            self.wfile.write(b'Not Found')

    def log_message(self, format, *args):
        """重写日志格式"""
        log(f"{self.client_address[0]} - {format % args}")

def main():
    """主函数"""
    log(f"Starting webhook server on port {PORT}...")
    log(f"Deploy directory: {DEPLOY_DIR}")
    log(f"Repository: {REPO_URL}")
    log(f"Branch: {BRANCH}")

    server = HTTPServer(('0.0.0.0', PORT), WebhookHandler)

    try:
        server.serve_forever()
    except KeyboardInterrupt:
        log("Shutting down server...")
        server.shutdown()

if __name__ == '__main__':
    main()