This is Pluto, the webhook server.
Vous ne pouvez pas sélectionner plus de 25 sujets Les noms de sujets doivent commencer par une lettre ou un nombre, peuvent contenir des tirets ('-') et peuvent comporter jusqu'à 35 caractères.
 
 
 
 

222 lignes
8.0 KiB

import time,string
from flask import Flask, render_template, redirect, url_for, request, g, make_response
from model import *
from util import *
app = Flask('pluto')
app.debug = True
@app.before_request
def init_globals():
for cls in type.__subclasses__(DBObject):
setattr(g, cls.__name__, cls)
app.jinja_env.filters['ctime'] = time.ctime
app.jinja_env.filters['jloads'] = jloads
app.jinja_env.filters['hloads'] = header_loads
app.jinja_env.filters['split'] = string.split
app.jinja_env.globals['safe_load'] = safe_load
@app.errorhandler(404)
def error_404(error):
return render_template('404.html', entity='page'), 404
@app.route('/')
def root():
return render_template('index.html')
@app.route('/hook', methods=['GET', 'POST'])
def hook():
response = make_response(render_template('api_hit.txt'))
response.headers['Content-type'] = 'text/json'
triggered = filter(lambda hook: hook.trigger(request.path, request.headers, jloads(request.data), response), Hook.all())
Log.create(time.time(), request.path, header_dumps(request.headers), request.data, ','.join(str(hook.rowid) for hook in triggered))
return response
@app.route('/logs')
def logs():
n = request.values.get('n', 10)
return render_template('logs.html', logs=Log.most_recent(n), n=n)
@app.route('/debuglogs', methods=['GET', 'POST'])
def debuglogs():
n = request.values.get('n', 10)
return render_template('debuglogs.html', logs=DebugLog.most_recent(n), n=n)
@app.route('/hooks')
def hooks():
return render_template('hooks.html', hooks=Hook.all())
@app.route('/hooks/new', methods=['GET', 'POST'])
def hooks_new():
if request.method == 'POST':
hook = Hook.create(
request.values['name'], None, # FIXME no author yet
checkbox(request, 'disabled'),
checkbox(request, 'debugged'),
)
return redirect('/hook/%d'%(hook.rowid,))
return render_template('hooks_new.html')
@app.route('/hook/<int:hookid>')
def hook_obj(hookid):
try:
hook = Hook.get_one(rowid=hookid)
except NoSuchEntity:
return render_template('404.html', entity='hook'), 404
except TooManyEntities:
return render_template('500.html', cause='too many hooks'), 500
return render_template('hook.html', hook=hook)
@app.route('/hook/<int:hookid>/edit', methods=['GET', 'POST'])
def hook_edit(hookid):
try:
hook = Hook.get_one(rowid=hookid)
except NoSuchEntity:
return render_template('404.html', entity='hook'), 404
except TooManyEntities:
return render_template('500.html', cause='too many hooks'), 500
if request.method == 'POST':
hook.name = request.values['name']
hook.disabled = checkbox(request, 'disabled')
hook.debugged = checkbox(request, 'debugged')
hook.update()
return redirect('/hook/%d'%(hook.rowid,))
return render_template('hook_edit.html', hook=hook)
@app.route('/hook/<int:hookid>/newcond', methods=['GET', 'POST'])
def hook_new_cond(hookid):
try:
hook = Hook.get_one(rowid=hookid)
except NoSuchEntity:
return render_template('404.html', entity='hook'), 404
except TooManyEntities:
return render_template('500.html', cause='too many hooks'), 500
if request.method == 'POST':
cond = Condition.create(
hook.rowid,
request.values['selector'], request.values['s1'], request.values['s2'], request.values['s3'],
request.values['test'], request.values['t1'], request.values['t2'], request.values['t3'],
checkbox(request, 'invert'),
)
return redirect('/cond/%d'%(cond.rowid,))
return render_template('hook_new_cond.html', hook=hook)
@app.route('/hook/<int:hookid>/newact', methods=['GET', 'POST'])
def hook_new_act(hookid):
try:
hook = Hook.get_one(rowid=hookid)
except NoSuchEntity:
return render_template('404.html', entity='hook'), 404
except TooManyEntities:
return render_template('500.html', cause='too many hooks'), 500
if request.method == 'POST':
act = Action.create(
hook.rowid,
request.values['action'], request.values['a1'], request.values['a2'], request.values['a3'],
)
return redirect('/act/%d'%(act.rowid,))
return render_template('hook_new_act.html', hook=hook)
@app.route('/hook/<int:hookid>/delete', methods=['GET', 'POST'])
def hook_delete(hookid):
try:
hook = Hook.get_one(rowid=hookid)
except NoSuchEntity:
return render_template('404.html', entity='hook'), 404
except TooManyEntities:
return render_template('500.html', cause='too many hooks'), 500
if request.method == 'POST':
hook.delete()
return redirect(url_for('hooks'))
return render_template('hook_delete.html', hook=hook)
@app.route('/cond/<int:condid>')
def cond_obj(condid):
try:
cond = Condition.get_one(rowid=condid)
except NoSuchEntity:
return render_template('404.html', entity='condition'), 404
except TooManyEntities:
return render_template('500.html', cause='too many conditions'), 500
return render_template('cond.html', cond=cond)
@app.route('/cond/<int:condid>/edit', methods=['GET', 'POST'])
def cond_edit(condid):
try:
cond = Condition.get_one(rowid=condid)
except NoSuchEntity:
return render_template('404.html', entity='condition'), 404
except TooManyEntities:
return render_template('500.html', cause='too many conditions'), 500
if request.method == 'POST':
cond.selector = request.values['selector']
cond.s1 = request.values['s1']
cond.s2 = request.values['s2']
cond.s3 = request.values['s3']
cond.test = request.values['test']
cond.t1 = request.values['t1']
cond.t2 = request.values['t2']
cond.t3 = request.values['t3']
cond.invert = checkbox(request, 'invert')
cond.update()
return redirect('/cond/%d'%(cond.rowid,))
return render_template('cond_edit.html', cond=cond)
@app.route('/cond/<int:condid>/delete', methods=['GET', 'POST'])
def cond_delete(condid):
try:
cond = Condition.get_one(rowid=condid)
except NoSuchEntity:
return render_template('404.html', entity='condition'), 404
except TooManyEntities:
return render_template('500.html', cause='too many conditions'), 500
if request.method == 'POST':
cond.delete()
return redirect('/hook/%d'%(cond.hook,))
return render_template('cond_delete.html', cond=cond)
@app.route('/act/<int:actid>')
def act_obj(actid):
try:
act = Action.get_one(rowid=actid)
except NoSuchEntity:
return render_template('404.html', entity='action'), 404
except TooManyEntities:
return render_template('500.html', cause='too many actions'), 500
return render_template('act.html', act=act)
@app.route('/act/<int:actid>/edit', methods=['GET', 'POST'])
def act_edit(actid):
try:
act = Action.get_one(rowid=actid)
except NoSuchEntity:
return render_template('404.html', entity='action'), 404
except TooManyEntities:
return render_template('500.html', cause='too many actions'), 500
if request.method == 'POST':
act.action = request.values['action']
act.a1 = request.values['a1']
act.a2 = request.values['a2']
act.a3 = request.values['a3']
act.update()
return redirect('/act/%d'%(act.rowid,))
return render_template('act_edit.html', act=act)
@app.route('/act/<int:actid>/delete', methods=['GET', 'POST'])
def act_delete(actid):
try:
act = Action.get_one(rowid=actid)
except NoSuchEntity:
return render_template('404.html', entity='action'), 404
except TooManyEntities:
return render_template('500.html', cause='too many actions'), 500
if request.method == 'POST':
act.delete()
return redirect('/hook/%d'%(act.hook,))
return render_template('act_delete.html', act=act)
if __name__ == '__main__':
app.run(host='0.0.0.0', port=8080)