This commit is contained in:
2025-10-01 18:37:59 +02:00
parent 85b035fd27
commit 696d0e61f1
46 changed files with 1750 additions and 0 deletions

View File

23
backend/account/admin.py Normal file
View File

@@ -0,0 +1,23 @@
from django.contrib import admin
from django.contrib.auth import get_user_model
from django.utils.translation import gettext_lazy as _
User = get_user_model()
@admin.register(User)
class UserAdmin(admin.ModelAdmin):
list_display = (
"id", "email", "role", "phone_number", "city", "street", "postal_code", "gdpr", "is_active", "email_verified", "create_time"
)
list_filter = ("role", "gdpr", "is_active", "email_verified", "city", "postal_code")
search_fields = ("email", "phone_number", "city", "street", "postal_code")
ordering = ("-create_time",)
fieldsets = (
(None, {"fields": ("email", "password", "role")}),
(_("Personal info"), {"fields": ("phone_number", "city", "street", "postal_code")}),
(_("Permissions"), {"fields": ("gdpr", "is_active", "email_verified")}),
(_("Important dates"), {"fields": ("create_time",)}),
)
readonly_fields = ("create_time",)
# Register your models here.

6
backend/account/apps.py Normal file
View File

@@ -0,0 +1,6 @@
from django.apps import AppConfig
class AccountConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'account'

View File

@@ -0,0 +1,24 @@
import django_filters
from django.contrib.auth import get_user_model
User = get_user_model()
class UserFilter(django_filters.FilterSet):
role = django_filters.CharFilter(field_name="role", lookup_expr="exact")
email = django_filters.CharFilter(field_name="email", lookup_expr="icontains")
phone_number = django_filters.CharFilter(field_name="phone_number", lookup_expr="icontains")
city = django_filters.CharFilter(field_name="city", lookup_expr="icontains")
street = django_filters.CharFilter(field_name="street", lookup_expr="icontains")
postal_code = django_filters.CharFilter(field_name="postal_code", lookup_expr="exact")
gdpr = django_filters.BooleanFilter(field_name="gdpr")
is_active = django_filters.BooleanFilter(field_name="is_active")
email_verified = django_filters.BooleanFilter(field_name="email_verified")
create_time_after = django_filters.IsoDateTimeFilter(field_name="create_time", lookup_expr="gte")
create_time_before = django_filters.IsoDateTimeFilter(field_name="create_time", lookup_expr="lte")
class Meta:
model = User
fields = [
"role", "email", "phone_number", "city", "street", "postal_code", "gdpr", "is_active", "email_verified",
"create_time_after", "create_time_before"
]

View File

@@ -0,0 +1,54 @@
# Generated by Django 5.2.5 on 2025-08-13 23:19
import account.models
import django.contrib.auth.validators
import django.core.validators
import django.utils.timezone
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
('auth', '0012_alter_user_first_name_max_length'),
]
operations = [
migrations.CreateModel(
name='CustomUser',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('password', models.CharField(max_length=128, verbose_name='password')),
('last_login', models.DateTimeField(blank=True, null=True, verbose_name='last login')),
('is_superuser', models.BooleanField(default=False, help_text='Designates that this user has all permissions without explicitly assigning them.', verbose_name='superuser status')),
('username', models.CharField(error_messages={'unique': 'A user with that username already exists.'}, help_text='Required. 150 characters or fewer. Letters, digits and @/./+/-/_ only.', max_length=150, unique=True, validators=[django.contrib.auth.validators.UnicodeUsernameValidator()], verbose_name='username')),
('first_name', models.CharField(blank=True, max_length=150, verbose_name='first name')),
('last_name', models.CharField(blank=True, max_length=150, verbose_name='last name')),
('is_staff', models.BooleanField(default=False, help_text='Designates whether the user can log into this admin site.', verbose_name='staff status')),
('date_joined', models.DateTimeField(default=django.utils.timezone.now, verbose_name='date joined')),
('is_deleted', models.BooleanField(default=False)),
('deleted_at', models.DateTimeField(blank=True, null=True)),
('role', models.CharField(blank=True, choices=[('admin', 'Administrátor'), ('user', 'Uživatel')], max_length=32, null=True)),
('email_verified', models.BooleanField(default=False)),
('phone_number', models.CharField(blank=True, max_length=16, unique=True, validators=[django.core.validators.RegexValidator('^\\+?\\d{9,15}$', message='Zadejte platné telefonní číslo.')])),
('email', models.EmailField(db_index=True, max_length=254, unique=True)),
('create_time', models.DateTimeField(auto_now_add=True)),
('city', models.CharField(blank=True, max_length=100, null=True)),
('street', models.CharField(blank=True, max_length=200, null=True)),
('postal_code', models.CharField(blank=True, max_length=5, null=True, validators=[django.core.validators.RegexValidator(code='invalid_postal_code', message='Postal code must contain exactly 5 digits.', regex='^\\d{5}$')])),
('gdpr', models.BooleanField(default=False)),
('is_active', models.BooleanField(default=False)),
('groups', models.ManyToManyField(blank=True, help_text='The groups this user belongs to.', related_name='customuser_set', related_query_name='customuser', to='auth.group')),
('user_permissions', models.ManyToManyField(blank=True, help_text='Specific permissions for this user.', related_name='customuser_set', related_query_name='customuser', to='auth.permission')),
],
options={
'abstract': False,
},
managers=[
('objects', account.models.CustomUserActiveManager()),
('all_objects', account.models.CustomUserAllManager()),
],
),
]

View File

152
backend/account/models.py Normal file
View File

@@ -0,0 +1,152 @@
import uuid
from django.db import models
from django.contrib.auth.models import AbstractUser, Group, Permission
from django.core.validators import RegexValidator, MinLengthValidator, MaxValueValidator, MinValueValidator
from django.conf import settings
from django.db import models
from django.utils import timezone
from datetime import timedelta
from vontor_cz.models import SoftDeleteModel
from django.contrib.auth.models import UserManager
import logging
logger = logging.getLogger(__name__)
# Custom User Manager to handle soft deletion
class CustomUserActiveManager(UserManager):
def get_queryset(self):
return super().get_queryset().filter(is_deleted=False)
# Custom User Manager to handle all users, including soft deleted
class CustomUserAllManager(UserManager):
def get_queryset(self):
return super().get_queryset()
class CustomUser(SoftDeleteModel, AbstractUser):
groups = models.ManyToManyField(
Group,
related_name="customuser_set", # <- přidáš related_name
blank=True,
help_text="The groups this user belongs to.",
related_query_name="customuser",
)
user_permissions = models.ManyToManyField(
Permission,
related_name="customuser_set", # <- přidáš related_name
blank=True,
help_text="Specific permissions for this user.",
related_query_name="customuser",
)
ROLE_CHOICES = (
('admin', 'Administrátor'),
('user', 'Uživatel'),
)
role = models.CharField(max_length=32, choices=ROLE_CHOICES, null=True, blank=True)
"""ACCOUNT_TYPES = (
('company', 'Firma'),
('individual', 'Fyzická osoba')
)
account_type = models.CharField(max_length=32, choices=ACCOUNT_TYPES, null=True, blank=True)"""
email_verified = models.BooleanField(default=False)
phone_number = models.CharField(
unique=True,
max_length=16,
blank=True,
validators=[RegexValidator(r'^\+?\d{9,15}$', message="Zadejte platné telefonní číslo.")]
)
email = models.EmailField(unique=True, db_index=True)
create_time = models.DateTimeField(auto_now_add=True)
"""company_id = models.CharField(
max_length=8,
blank=True,
null=True,
validators=[
RegexValidator(
regex=r'^\d{8}$',
message="Company ID must contain exactly 8 digits.",
code='invalid_company_id'
)
]
)"""
"""personal_id = models.CharField(
max_length=11,
blank=True,
null=True,
validators=[
RegexValidator(
regex=r'^\d{6}/\d{3,4}$',
message="Personal ID must be in the format 123456/7890.",
code='invalid_personal_id'
)
]
)"""
city = models.CharField(null=True, blank=True, max_length=100)
street = models.CharField(null=True, blank=True, max_length=200)
postal_code = models.CharField(
max_length=5,
blank=True,
null=True,
validators=[
RegexValidator(
regex=r'^\d{5}$',
message="Postal code must contain exactly 5 digits.",
code='invalid_postal_code'
)
]
)
gdpr = models.BooleanField(default=False)
is_active = models.BooleanField(default=False)
objects = CustomUserActiveManager()
all_objects = CustomUserAllManager()
REQUIRED_FIELDS = ['email', "username", "password"]
def __str__(self):
return f"{self.email} at {self.create_time.strftime('%d-%m-%Y %H:%M:%S')}"
def delete(self, *args, **kwargs):
self.is_active = False
#self.orders.all().update(is_deleted=True, deleted_at=timezone.now())
return super().delete(*args, **kwargs)
def save(self, *args, **kwargs):
is_new = self.pk is None # check BEFORE saving
if is_new:
if self.is_superuser or self.role == "admin":
self.is_active = True
if self.role == 'admin':
self.is_staff = True
self.is_superuser = True
if self.is_superuser:
self.role = 'admin'
else:
self.is_staff = False
return super().save(*args, **kwargs)

View File

@@ -0,0 +1,73 @@
from rest_framework.permissions import BasePermission, SAFE_METHODS
from rest_framework.permissions import IsAuthenticated
from rest_framework_api_key.permissions import HasAPIKey
#Podle svého uvážení (NEPOUŽÍVAT!!!)
class RolePermission(BasePermission):
allowed_roles = []
def has_permission(self, request, view):
# Je uživatel přihlášený a má roli z povolených?
user_has_role = (
request.user and
request.user.is_authenticated and
getattr(request.user, "role", None) in self.allowed_roles
)
# Má API klíč?
has_api_key = HasAPIKey().has_permission(request, view)
return user_has_role or has_api_key
#TOHLE POUŽÍT!!!
#Prostě stačí vložit: RoleAllowed('seller','cityClerk')
def RoleAllowed(*roles):
"""
Allows safe methods for any authenticated user.
Allows unsafe methods only for users with specific roles.
Args:
RolerAllowed('admin', 'user')
"""
class SafeOrRolePermission(BasePermission):
def has_permission(self, request, view):
# Allow safe methods for any authenticated user
if request.method in SAFE_METHODS:
return IsAuthenticated().has_permission(request, view)
# Otherwise, check the user's role
user = request.user
return user and user.is_authenticated and getattr(user, "role", None) in roles
return SafeOrRolePermission
def OnlyRolesAllowed(*roles):
class SafeOrRolePermission(BasePermission):
"""
Allows all methods only for users with specific roles.
"""
def has_permission(self, request, view):
# Otherwise, check the user's role
user = request.user
return user and user.is_authenticated and getattr(user, "role", None) in roles
return SafeOrRolePermission
# For Settings.py
class AdminOnly(BasePermission):
""" Allows access only to users with the 'admin' role.
Args:
BasePermission (rest_framework.permissions.BasePermission): Base class for permission classes.
"""
def has_permission(self, request, view):
return request.user and request.user.is_authenticated and getattr(request.user, 'role', None) == 'admin'

View File

@@ -0,0 +1,210 @@
import re
from django.utils.text import slugify
from django.core.validators import MinValueValidator, MaxValueValidator
from rest_framework import serializers
from rest_framework.exceptions import NotFound
from django.contrib.auth import get_user_model
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer
from django.utils.translation import gettext_lazy as _
from django.utils.text import slugify
from .permissions import *
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer
from rest_framework.exceptions import PermissionDenied
User = get_user_model()
class CustomUserSerializer(serializers.ModelSerializer):
class Meta:
model = User
fields = [
"id",
"username",
"first_name",
"last_name",
"email",
"role",
"account_type",
"email_verified",
"phone_number",
"create_time",
"var_symbol",
"bank_account",
"ICO",
"RC",
"city",
"street",
"PSC",
"GDPR",
"is_active",
]
read_only_fields = ["id", "create_time", "GDPR", "username"] # <-- removed "account_type"
def update(self, instance, validated_data):
user = self.context["request"].user
staff_only_fields = ["role", "email_verified", "var_symbol", "is_active"]
if user.role not in ["admin", "cityClerk"]:
unauthorized = [f for f in staff_only_fields if f in validated_data]
if unauthorized:
raise PermissionDenied(f"You are not allowed to modify: {', '.join(unauthorized)}")
return super().update(instance, validated_data)
# Token obtaining Default Serializer
class CustomTokenObtainPairSerializer(TokenObtainPairSerializer):
username_field = User.USERNAME_FIELD
def validate(self, attrs):
login = attrs.get("username")
password = attrs.get("password")
# Allow login by username or email
user = User.objects.filter(email__iexact=login).first() or \
User.objects.filter(username__iexact=login).first()
if user is None or not user.check_password(password):
raise serializers.ValidationError(_("No active account found with the given credentials"))
# Call the parent validation to create token
data = super().validate({
self.username_field: user.username,
"password": password
})
data["user_id"] = user.id
data["username"] = user.username
data["email"] = user.email
return data
# user creating section start ------------------------------------------
class UserRegistrationSerializer(serializers.ModelSerializer):
password = serializers.CharField(
write_only=True,
help_text="Heslo musí mít alespoň 8 znaků, obsahovat velká a malá písmena a číslici."
)
class Meta:
model = User
fields = [
'first_name', 'last_name', 'email', 'phone_number', 'password',
'city', 'street', 'postal_code', 'gdpr'
]
extra_kwargs = {
'first_name': {'required': True, 'help_text': 'Křestní jméno uživatele'},
'last_name': {'required': True, 'help_text': 'Příjmení uživatele'},
'email': {'required': True, 'help_text': 'Emailová adresa uživatele'},
'phone_number': {'required': True, 'help_text': 'Telefonní číslo uživatele'},
'city': {'required': True, 'help_text': 'Město uživatele'},
'street': {'required': True, 'help_text': 'Ulice uživatele'},
'postal_code': {'required': True, 'help_text': 'PSČ uživatele'},
'gdpr': {'required': True, 'help_text': 'Souhlas se zpracováním osobních údajů'},
}
def validate_password(self, value):
if len(value) < 8:
raise serializers.ValidationError("Password must be at least 8 characters long.")
if not re.search(r"[A-Z]", value):
raise serializers.ValidationError("Password must contain at least one uppercase letter.")
if not re.search(r"[a-z]", value):
raise serializers.ValidationError("Password must contain at least one lowercase letter.")
if not re.search(r"\d", value):
raise serializers.ValidationError("Password must contain at least one digit.")
return value
def validate(self, data):
email = data.get("email")
phone = data.get("phone_number")
dgpr = data.get("GDPR")
if not dgpr:
raise serializers.ValidationError({"GDPR": "You must agree to the GDPR to register."})
if User.objects.filter(email=email).exists():
raise serializers.ValidationError({"email": "Account with this email already exists."})
if phone and User.objects.filter(phone_number=phone).exists():
raise serializers.ValidationError({"phone_number": "Account with this phone number already exists."})
return data
def create(self, validated_data):
password = validated_data.pop("password")
username = validated_data.get("username", "")
user = User.objects.create(
username=username,
is_active=False, #uživatel je defaultně deaktivovaný
**validated_data
)
user.set_password(password)
user.save()
return user
class UserActivationSerializer(serializers.Serializer):
user_id = serializers.IntegerField()
var_symbol = serializers.IntegerField(validators=[MinValueValidator(0), MaxValueValidator(9999999999)])
def save(self, **kwargs):
try:
user = User.objects.get(pk=self.validated_data['user_id'])
except User.DoesNotExist:
raise NotFound("User with this ID does not exist.")
user.var_symbol = self.validated_data['var_symbol']
user.is_active = True
user.save()
return user
def to_representation(self, instance):
return {
"id": instance.id,
"email": instance.email,
"var_symbol": instance.var_symbol,
"is_active": instance.is_active,
}
class Meta:
model = User
fields = [
'user_id', 'var_symbol'
]
extra_kwargs = {
'user_id': {'required': True, 'help_text': 'ID uživatele'},
'var_symbol': {'required': True, 'help_text': 'Variablní symbol, zadán úředníkem'},
}
# user creating section end --------------------------------------------
class PasswordResetRequestSerializer(serializers.Serializer):
email = serializers.EmailField(
help_text="E-mail registrovaného a aktivního uživatele, na který bude zaslán reset hesla."
)
def validate_email(self, value):
if not User.objects.filter(email=value, is_active=True).exists():
raise serializers.ValidationError("Účet s tímto emailem neexistuje nebo není aktivní.")
return value
class PasswordResetConfirmSerializer(serializers.Serializer):
password = serializers.CharField(
write_only=True,
help_text="Nové heslo musí mít alespoň 8 znaků, obsahovat velká a malá písmena a číslici."
)
def validate_password(self, value):
import re
if len(value) < 8:
raise serializers.ValidationError("Heslo musí mít alespoň 8 znaků.")
if not re.search(r"[A-Z]", value):
raise serializers.ValidationError("Musí obsahovat velké písmeno.")
if not re.search(r"[a-z]", value):
raise serializers.ValidationError("Musí obsahovat malé písmeno.")
if not re.search(r"\d", value):
raise serializers.ValidationError("Musí obsahovat číslici.")
return value

85
backend/account/tasks.py Normal file
View File

@@ -0,0 +1,85 @@
from celery import shared_task
from celery.utils.log import get_task_logger
from django.core.mail import send_mail
from django.conf import settings
from django.utils.http import urlsafe_base64_encode
from django.utils.encoding import force_bytes
from django.template.loader import render_to_string
from .tokens import *
from .models import CustomUser
logger = get_task_logger(__name__)
@shared_task
def send_password_reset_email_task(user_id):
try:
user = CustomUser.objects.get(pk=user_id)
except CustomUser.DoesNotExist:
error_msg = f"Task send_password_reset_email has failed. Invalid User ID was sent."
logger.error(error_msg)
raise Exception(error_msg)
uid = urlsafe_base64_encode(force_bytes(user.pk))
token = password_reset_token.make_token(user)
reset_url = f"{settings.FRONTEND_URL}/reset-password/{uid}/{token}"
html_message = render_to_string(
'emails/password_reset.html',
{'reset_url': reset_url}
)
if settings.EMAIL_BACKEND == 'django.core.mail.backends.console.EmailBackend':
logger.debug("\nEMAIL OBSAH:\n", html_message, "\nKONEC OBSAHU")
send_email_with_context(
recipients=user.email,
subject="Obnova hesla",
message=None,
html_message=html_message
)
# Only email verification for user registration
@shared_task
def send_email_verification_task(user_id):
try:
user = CustomUser.objects.get(pk=user_id)
except CustomUser.DoesNotExist:
error_msg = f"Task send_email_verification_task has failed. Invalid User ID was sent."
logger.error(error_msg)
raise Exception(error_msg)
uid = urlsafe_base64_encode(force_bytes(user.pk))
token = account_activation_token.make_token(user)
verification_url = f"{settings.FRONTEND_URL}/email-verification/?uidb64={uid}&token={token}"
html_message = render_to_string(
'emails/email_verification.html',
{'verification_url': verification_url}
)
if settings.EMAIL_BACKEND == 'django.core.mail.backends.console.EmailBackend':
logger.debug("\nEMAIL OBSAH:\n", html_message, "\nKONEC OBSAHU")
send_email_with_context(
recipients=user.email,
subject="Ověření e-mailu",
message=None,
html_message=html_message
)
def send_email_with_context(recipients, subject, message=None, html_message=None):
"""
General function to send emails with a specific context.
"""
if isinstance(recipients, str):
recipients = [recipients]
try:
send_mail(
subject=subject,
message=message if message else '',
from_email=None,
recipient_list=recipients,
fail_silently=False,
html_message=html_message
)
if settings.EMAIL_BACKEND == 'django.core.mail.backends.console.EmailBackend':
logger.debug("\nEMAIL OBSAH:\n", html_message if html_message else message, "\nKONEC OBSAHU")
return True
except Exception as e:
logger.error(f"E-mail se neodeslal: {e}")
return False

View File

@@ -0,0 +1,19 @@
<!DOCTYPE html>
<html lang="cs">
<head>
<meta charset="UTF-8">
<title>Ověření e-mailu</title>
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet">
</head>
<body>
<div class="container mt-5">
<div class="card">
<div class="card-body">
<h2 class="card-title">Ověření e-mailu</h2>
<p class="card-text">Ověřte svůj e-mail kliknutím na odkaz níže:</p>
<a href="{{ verification_url }}" class="btn btn-success">Ověřit e-mail</a>
</div>
</div>
</div>
</body>
</html>

View File

@@ -0,0 +1,19 @@
<!DOCTYPE html>
<html lang="cs">
<head>
<meta charset="UTF-8">
<title>Obnova hesla</title>
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css" rel="stylesheet">
</head>
<body>
<div class="container mt-5">
<div class="card">
<div class="card-body">
<h2 class="card-title">Obnova hesla</h2>
<p class="card-text">Pro obnovu hesla klikněte na následující odkaz:</p>
<a href="{{ reset_url }}" class="btn btn-primary">Obnovit heslo</a>
</div>
</div>
</div>
</body>
</html>

3
backend/account/tests.py Normal file
View File

@@ -0,0 +1,3 @@
from django.test import TestCase
# Create your tests here.

33
backend/account/tokens.py Normal file
View File

@@ -0,0 +1,33 @@
from django.contrib.auth.tokens import PasswordResetTokenGenerator
# Subclass PasswordResetTokenGenerator to create a separate token generator
# for account activation. This allows future customization specific to activation tokens,
# even though it currently behaves exactly like the base class.
class AccountActivationTokenGenerator(PasswordResetTokenGenerator):
pass # No changes yet; inherits all behavior from PasswordResetTokenGenerator
# Create an instance of AccountActivationTokenGenerator to be used for generating
# and validating account activation tokens throughout the app.
account_activation_token = AccountActivationTokenGenerator()
# Create an instance of the base PasswordResetTokenGenerator to be used
# for password reset tokens.
password_reset_token = PasswordResetTokenGenerator()
from rest_framework_simplejwt.authentication import JWTAuthentication
#NEMĚNIT CUSTOM SBÍRANÍ COOKIE TOKENU
class CookieJWTAuthentication(JWTAuthentication):
def authenticate(self, request):
raw_token = request.COOKIES.get('access_token')
if not raw_token:
return None
validated_token = self.get_validated_token(raw_token)
return self.get_user(validated_token), validated_token

27
backend/account/urls.py Normal file
View File

@@ -0,0 +1,27 @@
from django.urls import path
from . import views
from django.urls import path, include
from rest_framework.routers import DefaultRouter
router = DefaultRouter()
router.register(r'users', views.UserView, basename='user')
urlpatterns = [
# Auth endpoints
path('login/', views.CookieTokenObtainPairView.as_view(), name='login'),
path('token/refresh/', views.CookieTokenRefreshView.as_view(), name='token-refresh'),
path('logout/', views.LogoutView.as_view(), name='logout'),
path('user/me/', views.CurrentUserView.as_view(), name='user-detail'),
# Registration & email endpoints
path('register/', views.UserRegistrationViewSet.as_view({'post': 'create'}), name='register'),
path('verify-email/<uidb64>/<token>/', views.EmailVerificationView.as_view(), name='verify-email'),
path('activate/', views.UserActivationViewSet.as_view(), name='activate-user'),
# Password reset endpoints
path('password-reset/', views.PasswordResetRequestView.as_view(), name='password-reset-request'),
path('password-reset-confirm/<uidb64>/<token>/', views.PasswordResetConfirmView.as_view(), name='password-reset-confirm'),
# User CRUD (list, retrieve, update, delete)
path('', include(router.urls)), #/users/
]

421
backend/account/views.py Normal file
View File

@@ -0,0 +1,421 @@
from django.contrib.auth import get_user_model, authenticate
from django.utils.http import urlsafe_base64_encode, urlsafe_base64_decode
from django.utils.encoding import force_bytes, force_str
from .serializers import *
from .permissions import *
from .models import CustomUser
from .tokens import *
from .tasks import send_password_reset_email_task
from django.conf import settings
import logging
logger = logging.getLogger(__name__)
from .filters import UserFilter
from rest_framework import generics, permissions, status, viewsets
from rest_framework.response import Response
from rest_framework_simplejwt.tokens import RefreshToken
from rest_framework.views import APIView
from rest_framework.viewsets import ModelViewSet
from rest_framework.permissions import IsAuthenticated, AllowAny
from rest_framework_simplejwt.tokens import RefreshToken
from rest_framework_simplejwt.exceptions import TokenError, AuthenticationFailed
from django_filters.rest_framework import DjangoFilterBackend
from drf_spectacular.utils import extend_schema, OpenApiResponse, OpenApiExample, OpenApiParameter
User = get_user_model()
#general user view API
from rest_framework_simplejwt.views import TokenObtainPairView
#---------------------------------------------TOKENY------------------------------------------------
# Custom Token obtaining view
@extend_schema(
tags=["Authentication"],
summary="Obtain JWT access and refresh tokens (cookie-based)",
description="Authenticate user and obtain JWT access and refresh tokens. You can use either email or username.",
request=CustomTokenObtainPairSerializer,
responses={
200: OpenApiResponse(response=CustomTokenObtainPairSerializer, description="Tokens returned successfully."),
401: OpenApiResponse(description="Invalid credentials or inactive user."),
},
)
class CookieTokenObtainPairView(TokenObtainPairView):
serializer_class = CustomTokenObtainPairSerializer
def post(self, request, *args, **kwargs):
response = super().post(request, *args, **kwargs)
# Získáme tokeny z odpovědi
access = response.data.get("access")
refresh = response.data.get("refresh")
if not access or not refresh:
return response # Např. při chybě přihlášení
jwt_settings = settings.SIMPLE_JWT
# Access token cookie
response.set_cookie(
key=jwt_settings.get("AUTH_COOKIE", "access_token"),
value=access,
httponly=jwt_settings.get("AUTH_COOKIE_HTTP_ONLY", True),
secure=jwt_settings.get("AUTH_COOKIE_SECURE", not settings.DEBUG),
samesite=jwt_settings.get("AUTH_COOKIE_SAMESITE", "Lax"),
path=jwt_settings.get("AUTH_COOKIE_PATH", "/"),
max_age=5 * 60, # 5 minut
)
# Refresh token cookie
response.set_cookie(
key="refresh_token",
value=refresh,
httponly=True,
secure=not settings.DEBUG,
samesite="Lax",
path="/",
max_age=7 * 24 * 60 * 60, # 7 dní
)
return response
def validate(self, attrs):
username = attrs.get("username")
password = attrs.get("password")
# Přihlaš uživatele ručně
user = authenticate(request=self.context.get('request'), username=username, password=password)
if not user:
raise AuthenticationFailed("Špatné uživatelské jméno nebo heslo.")
if not user.is_active:
raise AuthenticationFailed("Uživatel je deaktivován.")
# Nastav validní uživatele (přebere další logiku ze SimpleJWT)
self.user = user
# Vrátí access a refresh token jako obvykle
return super().validate(attrs)
@extend_schema(
tags=["Authentication"],
summary="Refresh JWT token using cookie",
description="Refresh JWT access and refresh tokens using the refresh token stored in cookie.",
responses={
200: OpenApiResponse(description="Tokens refreshed successfully."),
400: OpenApiResponse(description="Refresh token cookie not found."),
401: OpenApiResponse(description="Invalid refresh token."),
},
)
class CookieTokenRefreshView(APIView):
def post(self, request):
refresh_token = request.COOKIES.get('refresh_token')
if not refresh_token:
return Response({"detail": "Refresh token cookie not found."}, status=status.HTTP_400_BAD_REQUEST)
try:
refresh = RefreshToken(refresh_token)
access_token = str(refresh.access_token)
new_refresh_token = str(refresh) # volitelně nový refresh token
response = Response({
"access": access_token,
"refresh": new_refresh_token,
})
# Nastav nové HttpOnly cookies
# Access token cookie (např. 5 minut platnost)
response.set_cookie(
"access_token",
access_token,
httponly=True,
secure=not settings.DEBUG,
samesite="Lax",
max_age=5 * 60,
path="/",
)
# Refresh token cookie (delší platnost, např. 7 dní)
response.set_cookie(
"refresh_token",
new_refresh_token,
httponly=True,
secure=not settings.DEBUG,
samesite="Lax",
max_age=7 * 24 * 60 * 60,
path="/",
)
return response
except TokenError:
return Response({"detail": "Invalid refresh token."}, status=status.HTTP_401_UNAUTHORIZED)
#---------------------------------------------LOGIN/LOGOUT------------------------------------------------
@extend_schema(
tags=["Authentication"],
summary="Logout user (delete access and refresh token cookies)",
description="Logs out the user by deleting access and refresh token cookies.",
responses={
200: OpenApiResponse(description="Logout successful."),
},
)
class LogoutView(APIView):
permission_classes = [AllowAny]
def post(self, request):
response = Response({"detail": "Logout successful"}, status=status.HTTP_200_OK)
# Smazání cookies
response.delete_cookie("access_token", path="/")
response.delete_cookie("refresh_token", path="/")
return response
#--------------------------------------------------------------------------------------------------------------
@extend_schema(
tags=["User"],
summary="List, retrieve, update, and delete users.",
description="Displays all users with filtering and ordering options. Requires authentication and appropriate role.",
responses={
200: OpenApiResponse(response=CustomUserSerializer, description="User(s) retrieved successfully."),
403: OpenApiResponse(description="Permission denied."),
},
)
class UserView(viewsets.ModelViewSet):
queryset = User.objects.all()
serializer_class = CustomUserSerializer
filter_backends = [DjangoFilterBackend]
filterset_class = UserFilter
# Require authentication and role permission
permission_classes = [IsAuthenticated]
class Meta:
model = CustomUser
extra_kwargs = {
"email": {"help_text": "Unikátní e-mailová adresa uživatele."},
"phone_number": {"help_text": "Telefonní číslo ve formátu +420123456789."},
"role": {"help_text": "Role uživatele určující jeho oprávnění v systému."},
"account_type": {"help_text": "Typ účtu firma nebo fyzická osoba."},
"email_verified": {"help_text": "Určuje, zda je e-mail ověřen."},
"create_time": {"help_text": "Datum a čas registrace uživatele (pouze pro čtení).", "read_only": True},
"var_symbol": {"help_text": "Variabilní symbol pro platby, pokud je vyžadován."},
"bank_account": {"help_text": "Číslo bankovního účtu uživatele."},
"ICO": {"help_text": "IČO firmy, pokud se jedná o firemní účet."},
"RC": {"help_text": "Rodné číslo pro fyzické osoby."},
"city": {"help_text": "Město trvalého pobytu / sídla."},
"street": {"help_text": "Ulice a číslo popisné."},
"PSC": {"help_text": "PSČ místa pobytu / sídla."},
"GDPR": {"help_text": "Souhlas se zpracováním osobních údajů."},
"is_active": {"help_text": "Stav aktivace uživatele."},
}
def get_permissions(self):
# Only admin can list or create users
if self.action in ['list', 'create']:
return [OnlyRolesAllowed("admin")()]
# Only admin or the user themselves can update or delete
elif self.action in ['update', 'partial_update', 'destroy']:
if self.request.user.role == 'admin':
return [OnlyRolesAllowed("admin")()]
elif self.kwargs.get('pk') and str(self.request.user.id) == self.kwargs['pk']:
return [IsAuthenticated()]
else:
# fallback - deny access
return [OnlyRolesAllowed("admin")()]
# Any authenticated user can retrieve (view) any user's profile
elif self.action == 'retrieve':
return [IsAuthenticated()]
return super().get_permissions()
# Get current user data
@extend_schema(
tags=["User"],
summary="Get current authenticated user",
description="Returns details of the currently authenticated user based on JWT token or session.",
responses={
200: OpenApiResponse(response=CustomUserSerializer, description="Current user details."),
401: OpenApiResponse(description="Unauthorized, user is not authenticated."),
}
)
class CurrentUserView(APIView):
permission_classes = [IsAuthenticated]
def get(self, request):
serializer = CustomUserSerializer(request.user)
return Response(serializer.data)
#------------------------------------------------REGISTRACE--------------------------------------------------------------
#1. registration API
@extend_schema(
tags=["User Registration"],
summary="Register a new user (company or individual)",
description="Register a new user (company or individual). The user will receive an email with a verification link.",
request=UserRegistrationSerializer,
responses={
201: OpenApiResponse(response=UserRegistrationSerializer, description="User registered successfully."),
400: OpenApiResponse(description="Invalid registration data."),
},
)
class UserRegistrationViewSet(ModelViewSet):
queryset = CustomUser.objects.all()
serializer_class = UserRegistrationSerializer
http_method_names = ['post']
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
user = serializer.save()
try:
send_email_verification_task.delay(user.id) # posílaní emailu pro potvrzení registrace - CELERY TASK
except Exception as e:
logger.error(f"Celery not available, using fallback. Error: {e}")
send_email_verification_task(user.id) # posílaní emailu pro potvrzení registrace
headers = self.get_success_headers(serializer.data)
return Response(serializer.data, status=status.HTTP_201_CREATED, headers=headers)
#2. confirming email
@extend_schema(
tags=["User Registration"],
summary="Verify user email via link",
description="Verify user email using the link with uid and token.",
parameters=[
OpenApiParameter(name='uidb64', type=str, location=OpenApiParameter.PATH, description="User ID encoded in base64 from email link."),
OpenApiParameter(name='token', type=str, location=OpenApiParameter.PATH, description="User token from email link."),
],
responses={
200: OpenApiResponse(description="Email successfully verified."),
400: OpenApiResponse(description="Invalid or expired token."),
},
)
class EmailVerificationView(APIView):
def get(self, request, uidb64, token):
try:
uid = force_str(urlsafe_base64_decode(uidb64))
user = User.objects.get(pk=uid)
except (User.DoesNotExist, ValueError, TypeError):
return Response({"error": "Neplatný odkaz."}, status=400)
if account_activation_token.check_token(user, token):
user.email_verified = True
user.save()
return Response({"detail": "E-mail byl úspěšně ověřen. Účet čeká na schválení."})
else:
return Response({"error": "Token je neplatný nebo expirovaný."}, status=400)
#3. seller activation API (var_symbol)
@extend_schema(
tags=["User Registration"],
summary="Activate user and set variable symbol (admin/cityClerk only)",
description="Activate user and set variable symbol. Only accessible by admin or cityClerk.",
request=UserActivationSerializer,
responses={
200: OpenApiResponse(response=UserActivationSerializer, description="User activated successfully."),
400: OpenApiResponse(description="Invalid activation data."),
404: OpenApiResponse(description="User not found."),
},
)
class UserActivationViewSet(APIView):
permission_classes = [OnlyRolesAllowed('cityClerk', 'admin')]
def patch(self, request, *args, **kwargs):
serializer = UserActivationSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
user = serializer.save()
try:
send_email_clerk_accepted_task.delay(user.id) # posílaní emailu pro informování uživatele o dokončení registrace, uředník doplnil variabilní symbol - CELERY TASK
except Exception as e:
logger.error(f"Celery not available, using fallback. Error: {e}")
send_email_clerk_accepted_task(user.id) # posílaní emailu pro informování uživatele o dokončení registrace, uředník doplnil variabilní symbol
return Response(serializer.to_representation(user), status=status.HTTP_200_OK)
#-------------------------------------------------END REGISTRACE-------------------------------------------------------------
#1. PasswordReset + send Email
@extend_schema(
tags=["User password reset"],
summary="Request password reset (send email)",
description="Request password reset by providing registered email. An email with instructions will be sent.",
request=PasswordResetRequestSerializer,
responses={
200: OpenApiResponse(description="Email with instructions sent."),
400: OpenApiResponse(description="Invalid email or request data."),
},
)
class PasswordResetRequestView(APIView):
def post(self, request):
serializer = PasswordResetRequestSerializer(data=request.data)
if serializer.is_valid():
try:
user = User.objects.get(email=serializer.validated_data['email'])
except User.DoesNotExist:
# Always return 200 even if user doesn't exist to avoid user enumeration
return Response({"detail": "E-mail s odkazem byl odeslán."})
try:
send_password_reset_email_task.delay(user.id) # posílaní emailu pro obnovení hesla - CELERY TASK
except Exception as e:
logger.error(f"Celery not available, using fallback. Error: {e}")
send_password_reset_email_task(user.id) # posílaní emailu pro obnovení hesla registrace
return Response({"detail": "E-mail s odkazem byl odeslán."})
return Response(serializer.errors, status=400)
#2. Confirming reset
@extend_schema(
tags=["User password reset"],
summary="Confirm password reset via token",
description="Confirm password reset using token from email.",
request=PasswordResetConfirmSerializer,
parameters=[
OpenApiParameter(name='uidb64', type=str, location=OpenApiParameter.PATH, description="User ID encoded in base64 from email link."),
OpenApiParameter(name='token', type=str, location=OpenApiParameter.PATH, description="Password reset token from email link."),
],
responses={
200: OpenApiResponse(description="Password changed successfully."),
400: OpenApiResponse(description="Invalid token or request data."),
},
)
class PasswordResetConfirmView(APIView):
def post(self, request, uidb64, token):
try:
uid = force_str(urlsafe_base64_decode(uidb64))
user = User.objects.get(pk=uid)
except (TypeError, ValueError, OverflowError, User.DoesNotExist):
return Response({"error": "Neplatný odkaz."}, status=400)
if not password_reset_token.check_token(user, token):
return Response({"error": "Token je neplatný nebo expirovaný."}, status=400)
serializer = PasswordResetConfirmSerializer(data=request.data)
if serializer.is_valid():
user.set_password(serializer.validated_data['password'])
user.save()
return Response({"detail": "Heslo bylo úspěšně změněno."})
return Response(serializer.errors, status=400)