feat: SPA routing, projects/resume sections, deployment infra
- Add client-side router with History API pushState (router.ts, pages.ts) - Extend overlay with projects grid/category/detail views and resume section - Add Gitea CI/CD workflow and webhook deployment server - Improve Docker Compose with nginx reverse proxy config - Add deployment script and .env.example
This commit is contained in:
138
webhook/server.py
Executable file
138
webhook/server.py
Executable file
@@ -0,0 +1,138 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Gitea webhook receiver — listens for push events and triggers a docker-compose
|
||||
redeploy with zero-downtime (rebuild + restart only the changed service).
|
||||
|
||||
Usage:
|
||||
WEBHOOK_SECRET=... REPO_DIR=/home/djoser/webgpu-portfolio python3 server.py
|
||||
|
||||
Environment:
|
||||
WEBHOOK_SECRET Gitea webhook secret (HMAC-SHA256) [required]
|
||||
REPO_DIR Path to the repo with docker-compose.yml [default: .]
|
||||
LISTEN_HOST Bind address [default: 0.0.0.0]
|
||||
LISTEN_PORT Port [default: 9000]
|
||||
"""
|
||||
|
||||
import hashlib
|
||||
import hmac
|
||||
import http.server
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
# ---- config ----
|
||||
|
||||
SECRET = os.environ.get("WEBHOOK_SECRET", "").encode()
|
||||
REPO_DIR = Path(os.environ.get("REPO_DIR", Path.cwd()))
|
||||
HOST = os.environ.get("LISTEN_HOST", "0.0.0.0")
|
||||
PORT = int(os.environ.get("LISTEN_PORT", "9000"))
|
||||
|
||||
|
||||
def verify_signature(body: bytes, header: str) -> bool:
|
||||
"""Verify Gitea's X-Gitea-Signature header against the shared secret."""
|
||||
if not SECRET:
|
||||
print("[webhook] WARNING: WEBHOOK_SECRET is empty — accepting all requests", flush=True)
|
||||
return True
|
||||
if not header:
|
||||
return False
|
||||
expected = hmac.new(SECRET, body, hashlib.sha256).hexdigest()
|
||||
return hmac.compare_digest(expected, header)
|
||||
|
||||
|
||||
def deploy() -> tuple[int, str]:
|
||||
"""Run 'git pull' + 'docker compose up -d --build' in REPO_DIR."""
|
||||
steps = [
|
||||
# Pull latest code
|
||||
(["git", "pull", "origin", "master"], "git pull"),
|
||||
# Rebuild + restart only the web container (nginx-proxy doesn't change)
|
||||
(["docker", "compose", "up", "-d", "--build", "--remove-orphans"], "docker compose up"),
|
||||
]
|
||||
|
||||
for cmd, label in steps:
|
||||
result = subprocess.run(
|
||||
cmd,
|
||||
cwd=REPO_DIR,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=120,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
return 1, f"{label} failed (exit {result.returncode}):\n{result.stderr.strip()}"
|
||||
print(f"[webhook] {label}: ok", flush=True)
|
||||
|
||||
return 0, "deploy ok"
|
||||
|
||||
|
||||
class WebhookHandler(http.server.BaseHTTPRequestHandler):
|
||||
def log_message(self, fmt: str, *args) -> None:
|
||||
print(f"[webhook] {self.client_address[0]} — {fmt % args}", flush=True)
|
||||
|
||||
def do_POST(self) -> None:
|
||||
content_length = int(self.headers.get("Content-Length", "0"))
|
||||
body = self.rfile.read(content_length)
|
||||
|
||||
# Verify signature
|
||||
signature = self.headers.get("X-Gitea-Signature", "")
|
||||
if not verify_signature(body, signature):
|
||||
self.send_error(403, "Bad signature")
|
||||
return
|
||||
|
||||
# Parse event
|
||||
event_type = self.headers.get("X-Gitea-Event", "")
|
||||
if event_type != "push":
|
||||
self.send_response(200)
|
||||
self.end_headers()
|
||||
self.wfile.write(f'{{"status":"ignored","event":"{event_type}"}}\n'.encode())
|
||||
return
|
||||
|
||||
try:
|
||||
payload = json.loads(body)
|
||||
except json.JSONDecodeError:
|
||||
self.send_error(400, "Invalid JSON")
|
||||
return
|
||||
|
||||
ref = payload.get("ref", "")
|
||||
print(f"[webhook] push on {ref} — deploying...", flush=True)
|
||||
|
||||
exit_code, msg = deploy()
|
||||
if exit_code == 0:
|
||||
self.send_response(200)
|
||||
self.end_headers()
|
||||
self.wfile.write(b'{"status":"ok"}\n')
|
||||
else:
|
||||
self.send_response(500)
|
||||
self.end_headers()
|
||||
self.wfile.write(f'{{"status":"error","detail":{json.dumps(msg)}}}\n'.encode())
|
||||
|
||||
def do_GET(self) -> None:
|
||||
"""Health-check endpoint."""
|
||||
if self.path == "/health":
|
||||
self.send_response(200)
|
||||
self.end_headers()
|
||||
self.wfile.write(b"ok\n")
|
||||
else:
|
||||
self.send_error(405, "Method not allowed")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
if not SECRET:
|
||||
print(
|
||||
"[webhook] WARNING: WEBHOOK_SECRET is empty. Set it for security.\n"
|
||||
" Generate one: openssl rand -hex 32",
|
||||
flush=True,
|
||||
)
|
||||
|
||||
server = http.server.HTTPServer((HOST, PORT), WebhookHandler)
|
||||
print(f"[webhook] Listening on {HOST}:{PORT} (repo dir: {REPO_DIR})", flush=True)
|
||||
try:
|
||||
server.serve_forever()
|
||||
except KeyboardInterrupt:
|
||||
print("\n[webhook] Shutting down.", flush=True)
|
||||
server.shutdown()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user