248 lines
8.4 KiB
Python
248 lines
8.4 KiB
Python
import json
|
|
import logging
|
|
|
|
from django.utils import timezone
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
from django.utils.decorators import method_decorator
|
|
from profiles.models import UserProfile
|
|
from rest_framework import status
|
|
from rest_framework.authentication import BaseAuthentication, TokenAuthentication
|
|
from rest_framework.exceptions import AuthenticationFailed
|
|
from rest_framework.permissions import IsAuthenticated
|
|
from rest_framework.response import Response
|
|
from rest_framework.views import APIView
|
|
from scrobbles.scrobblers import (
|
|
emacs_scrobble_task,
|
|
emacs_scrobble_update_task,
|
|
todoist_scrobble_task,
|
|
todoist_scrobble_update_task,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TodoistTokenAuthentication(BaseAuthentication):
|
|
keyword = "Bearer"
|
|
|
|
def authenticate(self, request):
|
|
auth_header = request.META.get("HTTP_AUTHORIZATION", "")
|
|
|
|
logger.info(
|
|
"[todoist_auth] Raw auth header: %s",
|
|
auth_header[:20] if auth_header else "None",
|
|
)
|
|
|
|
if not auth_header:
|
|
return None
|
|
|
|
parts = auth_header.split()
|
|
|
|
if len(parts) != 2 or parts[0] != self.keyword:
|
|
logger.warning(
|
|
"[todoist_auth] Invalid auth header format: parts=%s",
|
|
parts,
|
|
)
|
|
return None
|
|
|
|
token = parts[1]
|
|
|
|
profile = UserProfile.objects.filter(todoist_auth_key=token).first()
|
|
|
|
if not profile:
|
|
logger.warning("[todoist_auth] No profile found for token")
|
|
raise AuthenticationFailed("Invalid token")
|
|
|
|
if not profile.user:
|
|
logger.warning("[todoist_auth] No user associated with profile")
|
|
raise AuthenticationFailed("No user associated with token")
|
|
|
|
logger.info("[todoist_auth] Authenticated user: %s", profile.user.id)
|
|
return (profile.user, profile)
|
|
|
|
def authenticate_header(self, request):
|
|
return self.keyword
|
|
|
|
|
|
class TodoistWebhookView(APIView):
|
|
authentication_classes = []
|
|
permission_classes = []
|
|
|
|
@method_decorator(csrf_exempt)
|
|
def dispatch(self, request, *args, **kwargs):
|
|
return super().dispatch(request, *args, **kwargs)
|
|
|
|
@property
|
|
def logger(self):
|
|
return logger
|
|
|
|
def post(self, request):
|
|
logger.info(
|
|
"[todoist_webhook] Raw auth header: %s",
|
|
request.META.get("HTTP_AUTHORIZATION", "None"),
|
|
)
|
|
|
|
user_profile = UserProfile.objects.filter(
|
|
todoist_user_id=request.data.get("user_id", None)
|
|
).first()
|
|
|
|
if not user_profile:
|
|
logger.warning(
|
|
"[todoist_webhook] No user profile found for user_id: %s",
|
|
request.data.get("user_id"),
|
|
)
|
|
return Response(
|
|
{"error": "No user found for this webhook"},
|
|
status=status.HTTP_401_UNAUTHORIZED,
|
|
)
|
|
|
|
post_data = request.data
|
|
self.logger.info(
|
|
"[todoist_webhook] called",
|
|
extra={"post_data": post_data},
|
|
)
|
|
todoist_task = {}
|
|
todoist_note = {}
|
|
todoist_type, todoist_event = post_data.get("event_name").split(":")
|
|
event_data = post_data.get("event_data", {})
|
|
is_item_type = todoist_type == "item"
|
|
is_note_type = todoist_type == "note"
|
|
new_labels = event_data.get("labels", [])
|
|
old_labels = (
|
|
post_data.get("event_data_extra", {}).get("old_item", {}).get("labels", [])
|
|
)
|
|
is_updated = todoist_event in ["updated"]
|
|
is_added = todoist_event in ["added"]
|
|
|
|
task_started = "inprogress" in new_labels and "inprogress" not in old_labels
|
|
task_stopped = "inprogress" not in new_labels and "inprogress" in old_labels
|
|
|
|
if is_item_type and is_updated and (task_started or task_stopped):
|
|
todoist_task = {
|
|
"todoist_id": event_data.get("id"),
|
|
"todoist_label_list": event_data.get("labels"),
|
|
"todoist_type": todoist_type,
|
|
"todoist_event": todoist_event,
|
|
"updated_at": event_data.get("updated_at"),
|
|
"todoist_project_id": event_data.get("project_id"),
|
|
"description": event_data.get("content"),
|
|
"details": event_data.get("description"),
|
|
}
|
|
if is_note_type and is_added:
|
|
task_data = event_data.get("item", {})
|
|
todoist_note = {
|
|
"task_id": event_data.get("item_id"),
|
|
"todoist_id": event_data.get("id"),
|
|
"todoist_label_list": task_data.get("labels"),
|
|
"todoist_type": todoist_type,
|
|
"todoist_event": todoist_event,
|
|
"updated_at": task_data.get("updated_at"),
|
|
"posted_at": event_data.get("posted_at"),
|
|
"details": task_data.get("description"),
|
|
"notes": event_data.get("content"),
|
|
"is_deleted": (
|
|
True if event_data.get("is_deleted") == "true" else False
|
|
),
|
|
}
|
|
|
|
if (is_added and not todoist_note) or (is_updated and not todoist_task):
|
|
self.logger.info(
|
|
"[todoist_webhook] ignoring wrong todoist type, event or labels",
|
|
extra={
|
|
"todoist_type": todoist_type,
|
|
"todoist_event": todoist_event,
|
|
"task_started": task_started,
|
|
"task_stopped": task_stopped,
|
|
"new_labels": new_labels,
|
|
"old_labels": old_labels,
|
|
},
|
|
)
|
|
return Response({}, status=status.HTTP_304_NOT_MODIFIED)
|
|
|
|
scrobble = None
|
|
if todoist_task:
|
|
scrobble = todoist_scrobble_task(
|
|
todoist_task,
|
|
user_profile.user_id,
|
|
stopped=task_stopped,
|
|
user_context_list=user_profile.task_context_tags,
|
|
)
|
|
|
|
if todoist_note:
|
|
scrobble = todoist_scrobble_update_task(todoist_note, user_profile.user_id)
|
|
|
|
if not scrobble:
|
|
self.logger.info(
|
|
"[todoist_webhook] finished with no note or task found",
|
|
extra={"scrobble_id": None},
|
|
)
|
|
return Response(
|
|
{"error": "No scrobble found to be updated"},
|
|
status=status.HTTP_304_NOT_MODIFIED,
|
|
)
|
|
|
|
self.logger.info(
|
|
"[todoist_webhook] finished",
|
|
extra={"scrobble_id": scrobble.id},
|
|
)
|
|
return Response({"scrobble_id": scrobble.id}, status=status.HTTP_200_OK)
|
|
|
|
|
|
class EmacsWebhookView(APIView):
|
|
authentication_classes = [TokenAuthentication]
|
|
permission_classes = [IsAuthenticated]
|
|
|
|
@property
|
|
def logger(self):
|
|
return logger
|
|
|
|
def post(self, request):
|
|
try:
|
|
post_data = json.loads(request.data)
|
|
except TypeError:
|
|
post_data = request.data
|
|
self.logger.info(
|
|
"[emacs_webhook] called",
|
|
extra={"post_data": post_data},
|
|
)
|
|
task_in_progress = post_data.get("state") == "STRT"
|
|
task_stopped = post_data.get("state") == "DONE"
|
|
post_data["source_id"] = post_data.pop("emacs_id")
|
|
|
|
user_id = request.user.id
|
|
|
|
user_profile = UserProfile.objects.filter(user_id=user_id).first()
|
|
|
|
scrobble = None
|
|
if post_data.get("source_id"):
|
|
scrobble = emacs_scrobble_task(
|
|
post_data,
|
|
user_id,
|
|
started=task_in_progress,
|
|
stopped=task_stopped,
|
|
user_context_list=user_profile.task_context_tags,
|
|
)
|
|
|
|
if not scrobble:
|
|
self.logger.info(
|
|
"[emacs_webhook] finished with no note or task found",
|
|
extra={"scrobble_id": None},
|
|
)
|
|
return Response(
|
|
{"error": "No scrobble found to be updated"},
|
|
status=status.HTTP_304_NOT_MODIFIED,
|
|
)
|
|
|
|
if scrobble and scrobble.in_progress:
|
|
emacs_scrobble_update_task(
|
|
post_data.get("source_id"),
|
|
post_data.get("notes") or [],
|
|
user_id,
|
|
description=post_data.get("body"),
|
|
)
|
|
|
|
self.logger.info(
|
|
"[emacs_webhook] finished",
|
|
extra={"scrobble_id": scrobble.id},
|
|
)
|
|
return Response({"scrobble_id": scrobble.id}, status=status.HTTP_200_OK)
|