from django.conf import settings
from djoser import views as djoser_views
from djoser.conf import settings as djoser_settings
from djoser.permissions import CurrentUserOrAdmin
from djoser.utils import login_user
from rest_framework import views, status
from rest_framework.decorators import action
from rest_framework.exceptions import NotFound, ValidationError, MethodNotAllowed
from rest_framework.permissions import AllowAny
from rest_framework.renderers import StaticHTMLRenderer
from rest_framework.response import Response
from account.models import User
from account.serializers import SetInitialPasswordSerializer, BonusProgramTransactionSerializer, \
UserBalanceUpdateSerializer, TelegramCallbackSerializer
from tg_bot.handlers.start import request_phone_sync
from tg_bot.messages import TGCoreMessage
from tg_bot.bot import bot_sync
class UserViewSet(djoser_views.UserViewSet):
""" Replacement for Djoser's UserViewSet """
def permission_denied(self, request, **kwargs):
if (
djoser_settings.HIDE_USERS
and request.user.is_authenticated
and self.action in ["balance"]
):
raise NotFound()
super().permission_denied(request, **kwargs)
def get_permissions(self):
if self.action == "set_initial_password":
self.permission_classes = djoser_settings.PERMISSIONS.set_password
return super().get_permissions()
def get_serializer_class(self):
if self.action == "set_initial_password":
return SetInitialPasswordSerializer
return super().get_serializer_class()
@action(["post"], detail=False)
def set_initial_password(self, request, *args, **kwargs):
return super().set_password(request, *args, **kwargs)
@action(["get", "patch"], detail=True, permission_classes=[CurrentUserOrAdmin])
def balance(self, request, *args, **kwargs):
user = self.get_object()
if request.method == "GET":
serializer = BonusProgramTransactionSerializer(user.bonus_history, many=True)
return Response(serializer.data)
elif request.method == "PATCH":
if not request.user.is_superuser:
return self.permission_denied(request)
# No balance underflow or dummy transactions allowed, no error will be raised
serializer = UserBalanceUpdateSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
data = serializer.data
user.update_balance(amount=data['amount'], bonus_type=data['type'], comment=data['comment'])
list_serializer = BonusProgramTransactionSerializer(user.bonus_history, many=True)
return Response(list_serializer.data)
@action(["get"], url_path="me/balance", detail=False, permission_classes=[CurrentUserOrAdmin])
def me_balance(self, request, *args, **kwargs):
self.get_object = self.get_instance
return self.balance(request, *args, **kwargs)
class TelegramLoginForm(views.APIView):
permission_classes = [AllowAny]
def get_renderers(self):
if self.request.method == "GET" and settings.DEBUG:
return [StaticHTMLRenderer()]
return super().get_renderers()
def get(self, request, *args, **kwargs):
if not settings.DEBUG:
raise MethodNotAllowed(request.method)
source = """
"""
return Response(source)
def post(self, request, *args, **kwargs):
serializer = TelegramCallbackSerializer(data=request.data)
try:
serializer.is_valid(raise_exception=True)
except ValidationError as e:
return Response(e.detail, status=status.HTTP_401_UNAUTHORIZED)
except:
return Response(status=status.HTTP_401_UNAUTHORIZED)
data = serializer.data
# Authenticate user with given tg_user_id
tg_user_id = data["id"]
user: User = User.objects.filter(tg_user_id=tg_user_id).first()
if not user:
# Sign up user
user = User.objects.create_draft_user(tg_user_id=tg_user_id)
# Request the phone through the bot
request_phone_sync(tg_user_id, TGCoreMessage.SIGN_UP_SHARE_PHONE)
token = login_user(request, user)
return Response({"auth_token": token.key})