In this example, we will demonstrate how to create a server-side webhook that receives data via HTTP requests, processes it using a Python script, and logs event information to a log file.
The example is also available at this link.
webhook.php
<?php
# Command to run a Python script
$PY_EXECUTE_CMD = "python3 webhook_script.py $1";
$payload = @file_get_contents("php://input");
$cmd = str_replace("$1", mb_escapeshellarg($payload), $PY_EXECUTE_CMD);
exec($cmd . " > /dev/null &");
function mb_escapeshellarg($arg)
{
return "'" . str_replace("'", "'\\''", $arg) . "'";
}
webhook_script.py
import os, sys
import datetime, time
import json
from querystring_parser import parser
scriptDir = os.path.realpath(__file__).replace('\\', '/').rsplit('/', 1)[0] #
def log(message, is_error = False):
l_path = scriptDir + '/logs'
if not os.path.exists(l_path):
os.makedirs(l_path)
f = open(l_path + '/script.' + time.strftime('%Y-%m-%d') + '.log.txt', 'a')
f.write('{0}: {1}\n'.format(time.strftime('%Y-%m-%d %H.%M.%S'), str(message)))
f.close()
def main(data):
# We check that the event is a task status change, and that data contains the prev_status key.
# Other events will contain their own unique keys.
if 'prev_status' in data.keys():
log('Task status change event')
log('Task name: ' + str(data['task_name']))
log('Task ID: ' + str(data['task_id']))
# You can use parameters to get other data via Cerebro API
try:
if len(sys.argv) > 1:
data = None
try:
data = json.loads(sys.argv[1])
except Exception as ex:
data = parser.parse(sys.argv[1], normalized=True)
main(data)
except Exception as err:
log('ERROR: {0}'.format(err))
To learn more about server-side webhooks setting up follolw the link