import re import orjson from django.contrib.auth.mixins import LoginRequiredMixin from django.http import HttpResponse, HttpResponseBadRequest from pydantic import ValidationError from rest_framework.parsers import JSONParser from rest_framework.views import APIView from core.forms import HookForm from core.lib import market from core.lib.schemas.drakdoo_s import DrakdooCallback from core.models import Callback, Hook, Signal from core.util import logs from core.views import ObjectCreate, ObjectDelete, ObjectList, ObjectUpdate log = logs.get_logger(__name__) def extract_price(message): result = re.findall("\d+\.\d+", message) # noqa if len(result) != 1: log.error(f"Could not extract price from message: {message}") return False try: log.debug(f"Extracted {result[0]} from '{message}'") return float(result[0]) except ValueError: return False class HookAPI(APIView): parser_classes = [JSONParser] def post(self, request, hook_name): log.debug(f"HookAPI POST: {request.data}") # Try loading the JSON # try: # loaded_json = orjson.loads(request.data) # except orjson.JSONDecodeError: # return HttpResponseBadRequest("Invalid JSON") # Try validating the JSON try: hook_resp = DrakdooCallback(**request.data) except ValidationError as e: log.error(f"HookAPI POST: {e}") return HttpResponseBadRequest(e) if hasattr(hook_resp, "market.price"): try: price = float(hook_resp.market.price) except ValueError: log.debug( f"Could not extract price from message: {hook_resp.market.price}" ) return HttpResponseBadRequest("Could not extract price from message") else: price = extract_price(hook_resp.message) if not price: log.debug(f"Could not extract price from message: {hook_resp.message}") return HttpResponseBadRequest("Could not extract price from message") base = hook_resp.market.item quote = hook_resp.market.currency symbol = f"{base.upper()}/{quote.upper()}" data = { "title": hook_resp.title, "message": hook_resp.message, "period": hook_resp.period, "sent": hook_resp.timestamp.sent, "trade": hook_resp.timestamp.trade, "exchange": hook_resp.market.exchange, "base": hook_resp.market.item, "quote": hook_resp.market.currency, "symbol": symbol, "contract": hook_resp.market.contract, "price": price, } log.debug("HookAPI callback: data: %s", data) # Try getting the hook try: hook = Hook.objects.get(hook=hook_name) except Hook.DoesNotExist: return HttpResponseBadRequest("Hook does not exist.") # Try getting the signal # AUCAD 3M StochExit: C$2419.64 at OANDA message_first_half = hook_resp.message.split(":")[0] signal_name = message_first_half.split(" ")[-1] try: signal = Signal.objects.get(signal=signal_name) except Signal.DoesNotExist: return HttpResponseBadRequest("Signal does not exist.") # For uptime checks/testing if data["exchange"].lower() == "n/a": log.debug("HookAPI callback: exchange is N/A, skipping") return HttpResponse("OK") # Set the signal relation data["signal"] = signal # Create the callback object callback = Callback.objects.create(hook=hook, **data) callback.save() market.process_callback(callback) # Bump received count hook.received = hook.received + 1 signal.received = signal.received + 1 hook.save() signal.save() return HttpResponse("OK") def get(self, request, hook_name): try: hook = Hook.objects.get(name=hook_name) except Hook.DoesNotExist: return HttpResponseBadRequest("Hook does not exist.") return_data = {"name": hook.name, "hook": hook.hook, "hook_id": hook.id} return HttpResponse(orjson.dumps(return_data), content_type="application/json") class HookList(LoginRequiredMixin, ObjectList): # window_content = "window-content/hooks.html" list_template = "partials/hook-list.html" model = Hook page_title = "List of active URL endpoints for receiving hooks." page_subtitle = ( "Add URLs here to receive Drakdoo callbacks. " "Make then unique and hard to guess!" ) list_url_name = "hooks" list_url_args = ["type"] submit_url_name = "hook_create" class HookCreate(LoginRequiredMixin, ObjectCreate): model = Hook form_class = HookForm list_url_name = "hooks" list_url_args = ["type"] submit_url_name = "hook_create" class HookUpdate(LoginRequiredMixin, ObjectUpdate): model = Hook form_class = HookForm list_url_name = "hooks" list_url_args = ["type"] submit_url_name = "hook_update" class HookDelete(LoginRequiredMixin, ObjectDelete): model = Hook list_url_name = "hooks" list_url_args = ["type"]