86 lines
3.2 KiB
Python
86 lines
3.2 KiB
Python
"""
|
|
This module manages webhook communication
|
|
"""
|
|
import logging
|
|
from typing import Any, Dict
|
|
|
|
from requests import RequestException, post
|
|
|
|
from freqtrade.rpc import RPC, RPCHandler, RPCMessageType
|
|
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
logger.debug('Included module rpc.webhook ...')
|
|
|
|
|
|
class Webhook(RPCHandler):
|
|
""" This class handles all webhook communication """
|
|
|
|
def __init__(self, rpc: RPC, config: Dict[str, Any]) -> None:
|
|
"""
|
|
Init the Webhook class, and init the super class RPCHandler
|
|
:param rpc: instance of RPC Helper class
|
|
:param config: Configuration object
|
|
:return: None
|
|
"""
|
|
super().__init__(rpc, config)
|
|
|
|
self._url = self._config['webhook']['url']
|
|
|
|
self._format = self._config['webhook'].get('format', 'form')
|
|
|
|
if self._format != 'form' and self._format != 'json':
|
|
raise NotImplementedError('Unknown webhook format `{}`, possible values are '
|
|
'`form` (default) and `json`'.format(self._format))
|
|
|
|
def cleanup(self) -> None:
|
|
"""
|
|
Cleanup pending module resources.
|
|
This will do nothing for webhooks, they will simply not be called anymore
|
|
"""
|
|
pass
|
|
|
|
def send_msg(self, msg: Dict[str, Any]) -> None:
|
|
""" Send a message to telegram channel """
|
|
try:
|
|
|
|
if msg['type'] == RPCMessageType.BUY_NOTIFICATION:
|
|
valuedict = self._config['webhook'].get('webhookbuy', None)
|
|
elif msg['type'] == RPCMessageType.BUY_CANCEL_NOTIFICATION:
|
|
valuedict = self._config['webhook'].get('webhookbuycancel', None)
|
|
elif msg['type'] == RPCMessageType.SELL_NOTIFICATION:
|
|
valuedict = self._config['webhook'].get('webhooksell', None)
|
|
elif msg['type'] == RPCMessageType.SELL_CANCEL_NOTIFICATION:
|
|
valuedict = self._config['webhook'].get('webhooksellcancel', None)
|
|
elif msg['type'] in (RPCMessageType.STATUS_NOTIFICATION,
|
|
RPCMessageType.STARTUP_NOTIFICATION,
|
|
RPCMessageType.WARNING_NOTIFICATION):
|
|
valuedict = self._config['webhook'].get('webhookstatus', None)
|
|
else:
|
|
raise NotImplementedError('Unknown message type: {}'.format(msg['type']))
|
|
if not valuedict:
|
|
logger.info("Message type '%s' not configured for webhooks", msg['type'])
|
|
return
|
|
|
|
payload = {key: value.format(**msg) for (key, value) in valuedict.items()}
|
|
self._send_msg(payload)
|
|
except KeyError as exc:
|
|
logger.exception("Problem calling Webhook. Please check your webhook configuration. "
|
|
"Exception: %s", exc)
|
|
|
|
def _send_msg(self, payload: dict) -> None:
|
|
"""do the actual call to the webhook"""
|
|
|
|
if self._format == 'form':
|
|
kwargs = {'data': payload}
|
|
elif self._format == 'json':
|
|
kwargs = {'json': payload}
|
|
else:
|
|
raise NotImplementedError('Unknown format: {}'.format(self._format))
|
|
|
|
try:
|
|
post(self._url, **kwargs)
|
|
except RequestException as exc:
|
|
logger.warning("Could not call webhook url. Exception: %s", exc)
|