pit-router/webhook-server.py

#!/usr/bin/env python3
"""
智队中枢 Webhook 服务
接收 Gitea Webhook 请求，触发自动更新
"""

import os
import hmac
import hashlib
import subprocess
import logging
from flask import Flask, request, jsonify
from datetime import datetime

app = Flask(__name__)

# 配置
WEBHOOK_SECRET = os.environ.get('WEBHOOK_SECRET', 'pit-router-webhook-secret-2026')
UPDATE_SCRIPT = '/www/wwwroot/pit-router/auto-update.sh'
LOG_FILE = '/var/log/pit-router-webhook.log'

# 日志配置
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler(LOG_FILE),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)


def verify_signature(payload: bytes, signature: str) -> bool:
    """验证 Gitea Webhook 签名"""
    if not signature:
        return False
    
    expected = 'sha256=' + hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    
    return hmac.compare_digest(expected, signature)


def log_request(event: str, data: dict):
    """记录请求日志"""
    ref = data.get('ref', '')
    commits = data.get('commits', [])
    pusher = data.get('pusher', {}).get('login', 'unknown')
    
    logger.info(f"Event: {event} | Ref: {ref} | Pusher: {pusher} | Commits: {len(commits)}")
    
    for commit in commits[:3]:  # 只记录前3个 commit
        msg = commit.get('message', '').split('\n')[0][:50]
        logger.info(f"  - {commit.get('id', '')[:7]}: {msg}")


@app.route('/health', methods=['GET'])
def health():
    """健康检查端点"""
    return jsonify({'status': 'ok', 'service': 'pit-router-webhook'})


@app.route('/webhook/pit-router', methods=['POST'])
def handle_webhook():
    """处理 Gitea Webhook"""
    try:
        # 1. 验证签名
        signature = request.headers.get('X-Gitea-Signature', '')
        if not verify_signature(request.data, signature):
            logger.warning("Invalid signature")
            return jsonify({'error': 'Invalid signature'}), 401
        
        # 2. 解析事件
        event = request.headers.get('X-Gitea-Event', '')
        data = request.json or {}
        
        # 记录请求
        log_request(event, data)
        
        # 3. 只处理 push 事件
        if event != 'push':
            logger.info(f"Event '{event}' ignored")
            return jsonify({'message': f'Event {event} ignored'}), 200
        
        # 4. 只处理 main 分支
        ref = data.get('ref', '')
        if ref != 'refs/heads/main':
            logger.info(f"Branch '{ref}' ignored")
            return jsonify({'message': f'Branch {ref} ignored'}), 200
        
        # 5. 执行更新脚本
        logger.info("🚀 Triggering auto-update...")
        result = subprocess.run(
            [UPDATE_SCRIPT],
            capture_output=True,
            text=True,
            timeout=600  # 10 分钟超时
        )
        
        # 记录输出
        if result.stdout:
            logger.info(f"Update output:\n{result.stdout[-2000:]}")  # 只记录最后 2000 字符
        if result.stderr:
            logger.error(f"Update error:\n{result.stderr[-1000:]}")
        
        if result.returncode == 0:
            logger.info("✅ Update completed successfully")
            return jsonify({
                'status': 'success',
                'message': 'Update completed',
                'timestamp': datetime.utcnow().isoformat()
            })
        else:
            logger.error(f"❌ Update failed with code {result.returncode}")
            return jsonify({
                'status': 'error',
                'message': 'Update failed',
                'details': result.stderr[-500:] if result.stderr else 'Unknown error'
            }), 500
            
    except subprocess.TimeoutExpired:
        logger.error("Update timeout")
        return jsonify({'error': 'Update timeout'}), 500
    except Exception as e:
        logger.exception(f"Unexpected error: {e}")
        return jsonify({'error': str(e)}), 500


@app.route('/webhook/test', methods=['GET'])
def test_webhook():
    """测试端点（手动触发更新）"""
    try:
        logger.info("🧪 Manual test trigger")
        result = subprocess.run(
            [UPDATE_SCRIPT],
            capture_output=True,
            text=True,
            timeout=600
        )
        
        return jsonify({
            'status': 'success' if result.returncode == 0 else 'error',
            'output': result.stdout[-1000:],
            'error': result.stderr[-500:] if result.stderr else None
        })
    except Exception as e:
        return jsonify({'error': str(e)}), 500


if __name__ == '__main__':
    logger.info("🚀 Starting PIT Router Webhook Server on port 5001")
    app.run(host='0.0.0.0', port=5001, debug=False)