webhook_agent.py 文件源码

def webhook_emit(self, request):
        name = request.match_info.get('name', None)
        token = request.match_info.get('token', None)
        post_data = await request.post()
        if not name:
            if 'name' not in request.GET or 'name' not in post_data:
                return web.json_response({'success': False, 'message': 'Error: required parameter "name" not found.'})
        if not token and not any(['token' not in request.GET, 'X-Hub-Signature' not in request.headers]):
            return web.json_response({'success': False, 'message': 'Error: required parameter "token" not found.'})
        if not name:
            name = request.GET.get('name', None) or post_data['name']
        if not token:
            token = request.GET.get('token', None)
        if token != TOKEN is False:
            return web.json_response({'success': False, 'message': 'Error: authentication failed. Invalid token.'})
        data = {k: v for k, v in request.GET.items() if k not in ['name', 'token']}
        if post_data:
            data.update({k: v for k, v in post_data.items() if k not in ['name', 'token']})
        self.emit(name, data=data)
        return web.json_response({'success': True})