В этом примере мы продемонстрируем, как создать серверный вебхук, который принимает данные через HTTP-запросы, обрабатывает их с помощью Python-скрипта и записывает информацию о событиях в лог-файл.
Пример также доступен по ссылке
webhook.php
<?php
# Команда для выполнения Python скрипта
$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):
# Проверяем что событие - изменение статуса задачи, data при этом содержит ключ prev_status
# Другие события будут содержать свои уникальные ключи
if 'prev_status' in data.keys():
log('Событие изменения статуса задачи')
log('Имя задачи: ' + str(data['task_name']))
log('ID задачи: ' + str(data['task_id']))
# Вы можете использовать параметры для получения других данных через API Cerebro
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))
Чтобы узнать больше о работе с серверными хуки в Cerebro, перейдите по ссылке