Add auto-deployment scripts for Gitea webhook
- deploy.sh: Script to pull code and restart Docker containers - webhook-server.py: HTTP server to receive Gitea push events - shorts-maker-webhook.service: Systemd service for webhook server - setup-autodeploy.sh: Setup script for the server 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
131
webhook-server.py
Normal file
131
webhook-server.py
Normal file
@@ -0,0 +1,131 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Simple webhook server for Gitea auto-deployment.
|
||||
Listens for push events and triggers deploy.sh
|
||||
"""
|
||||
|
||||
import subprocess
|
||||
import hmac
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
from http.server import HTTPServer, BaseHTTPRequestHandler
|
||||
from datetime import datetime
|
||||
|
||||
# Configuration
|
||||
WEBHOOK_SECRET = os.environ.get('WEBHOOK_SECRET', 'your-secret-here')
|
||||
DEPLOY_SCRIPT = '/home/bini/shorts-maker/deploy.sh'
|
||||
PORT = 9000
|
||||
LOG_FILE = '/var/log/webhook-server.log'
|
||||
|
||||
def log(message):
|
||||
timestamp = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
||||
log_line = f"[{timestamp}] {message}"
|
||||
print(log_line)
|
||||
try:
|
||||
with open(LOG_FILE, 'a') as f:
|
||||
f.write(log_line + '\n')
|
||||
except:
|
||||
pass
|
||||
|
||||
def verify_signature(payload, signature):
|
||||
"""Verify Gitea webhook signature"""
|
||||
if not signature:
|
||||
return False
|
||||
expected = hmac.new(
|
||||
WEBHOOK_SECRET.encode(),
|
||||
payload,
|
||||
hashlib.sha256
|
||||
).hexdigest()
|
||||
return hmac.compare_digest(f"sha256={expected}", signature)
|
||||
|
||||
class WebhookHandler(BaseHTTPRequestHandler):
|
||||
def do_POST(self):
|
||||
if self.path != '/webhook':
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
return
|
||||
|
||||
content_length = int(self.headers.get('Content-Length', 0))
|
||||
payload = self.rfile.read(content_length)
|
||||
|
||||
# Verify signature if secret is configured
|
||||
if WEBHOOK_SECRET != 'your-secret-here':
|
||||
signature = self.headers.get('X-Gitea-Signature')
|
||||
if not verify_signature(payload, signature):
|
||||
log("ERROR: Invalid webhook signature")
|
||||
self.send_response(403)
|
||||
self.end_headers()
|
||||
self.wfile.write(b'Invalid signature')
|
||||
return
|
||||
|
||||
try:
|
||||
data = json.loads(payload)
|
||||
ref = data.get('ref', '')
|
||||
|
||||
# Only deploy on push to main branch
|
||||
if ref == 'refs/heads/main':
|
||||
log(f"Received push to main from {data.get('pusher', {}).get('name', 'unknown')}")
|
||||
log(f"Commit: {data.get('after', 'unknown')[:8]}")
|
||||
|
||||
# Run deploy script
|
||||
log("Starting deployment...")
|
||||
result = subprocess.run(
|
||||
['bash', DEPLOY_SCRIPT],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=600 # 10 minute timeout
|
||||
)
|
||||
|
||||
if result.returncode == 0:
|
||||
log("Deployment successful!")
|
||||
self.send_response(200)
|
||||
self.end_headers()
|
||||
self.wfile.write(b'Deployment successful')
|
||||
else:
|
||||
log(f"Deployment failed: {result.stderr}")
|
||||
self.send_response(500)
|
||||
self.end_headers()
|
||||
self.wfile.write(f'Deployment failed: {result.stderr}'.encode())
|
||||
else:
|
||||
log(f"Ignoring push to {ref}")
|
||||
self.send_response(200)
|
||||
self.end_headers()
|
||||
self.wfile.write(b'Ignored (not main branch)')
|
||||
|
||||
except json.JSONDecodeError:
|
||||
log("ERROR: Invalid JSON payload")
|
||||
self.send_response(400)
|
||||
self.end_headers()
|
||||
self.wfile.write(b'Invalid JSON')
|
||||
except subprocess.TimeoutExpired:
|
||||
log("ERROR: Deployment timed out")
|
||||
self.send_response(500)
|
||||
self.end_headers()
|
||||
self.wfile.write(b'Deployment timed out')
|
||||
except Exception as e:
|
||||
log(f"ERROR: {str(e)}")
|
||||
self.send_response(500)
|
||||
self.end_headers()
|
||||
self.wfile.write(f'Error: {str(e)}'.encode())
|
||||
|
||||
def do_GET(self):
|
||||
if self.path == '/health':
|
||||
self.send_response(200)
|
||||
self.end_headers()
|
||||
self.wfile.write(b'OK')
|
||||
else:
|
||||
self.send_response(404)
|
||||
self.end_headers()
|
||||
|
||||
def log_message(self, format, *args):
|
||||
log(f"HTTP: {args[0]}")
|
||||
|
||||
if __name__ == '__main__':
|
||||
log(f"Starting webhook server on port {PORT}")
|
||||
server = HTTPServer(('0.0.0.0', PORT), WebhookHandler)
|
||||
try:
|
||||
server.serve_forever()
|
||||
except KeyboardInterrupt:
|
||||
log("Shutting down webhook server")
|
||||
server.shutdown()
|
||||
Reference in New Issue
Block a user