57 lines
1.8 KiB
Python
57 lines
1.8 KiB
Python
import os
|
|
import subprocess
|
|
from flask import Flask, request, jsonify
|
|
|
|
app = Flask(__name__)
|
|
|
|
WEBBUMP_DIR = "/etc/webbump"
|
|
|
|
@app.route('/status', methods=['GET'])
|
|
def status():
|
|
"""Minimal endpoint for health check."""
|
|
return "OK", 200
|
|
|
|
@app.route("/webhook", methods=["POST"])
|
|
def gitea_webhook():
|
|
data = request.json
|
|
if not data:
|
|
print("Received webhook without JSON")
|
|
return jsonify({"error": "Invalid JSON"}), 400
|
|
|
|
# Repository name
|
|
repo_name = data.get("repository", {}).get("name")
|
|
if not repo_name:
|
|
print("Could not get Repository name from JSON")
|
|
return jsonify({"error": "Repository name not found"}), 400
|
|
|
|
# Determine event type from header
|
|
event_type = request.headers.get("X-Gitea-Event", "").lower()
|
|
if event_type == "push":
|
|
args = []
|
|
|
|
elif event_type == "release":
|
|
version = data.get("release", {}).get("tag_name")
|
|
if not version:
|
|
return jsonify({"error": "Release version not found"}), 400
|
|
args = [version]
|
|
else:
|
|
return jsonify({"message": f"Ignored event type: {event_type}"}), 200
|
|
|
|
script_path = os.path.join(WEBBUMP_DIR, repo_name, event_type)
|
|
if not os.path.isfile(script_path) or not os.access(script_path, os.X_OK):
|
|
return jsonify({"message": f"No scripts executed for event: {event_type}"}), 200
|
|
|
|
# Run the script
|
|
try:
|
|
subprocess.run([script_path, *args], check=True)
|
|
except subprocess.CalledProcessError as e:
|
|
print(repo_name + " " + script_name + " script failed with exit code: " + e)
|
|
return jsonify({"error": f"Script failed: {e}"}), 500
|
|
|
|
return jsonify({"message": f"Executed {script_path}"}), 200
|
|
|
|
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', port=5000, debug=True)
|