Добавлены QR коды и управление работниками

src/business/admin.py

@@ -0,0 +1,3 @@
+from django.contrib import admin
+
+# Register your models here.

src/business/apps.py

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

src/business/migrations/0001_initial.py

@@ -0,0 +1,35 @@
+# Generated by Django 5.0.4 on 2024-04-07 02:02
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        ('organizations', '0001_initial'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='OrganizationAccount',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('organization', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='organizations.organization')),
+                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
+            ],
+        ),
+        migrations.CreateModel(
+            name='QRCodeUsing',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('used_dt', models.DateTimeField(auto_now=True)),
+                ('client', models.ForeignKey(default=None, null=True, on_delete=django.db.models.deletion.SET_NULL, to=settings.AUTH_USER_MODEL)),
+                ('organization', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to='organizations.organization')),
+            ],
+        ),
+    ]

src/business/models.py

@@ -0,0 +1,15 @@
+from django.conf import settings
+from django.db import models
+
+# Create your models here.
+
+
+class OrganizationAccount(models.Model):
+    user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
+    organization = models.ForeignKey('organizations.Organization', on_delete=models.CASCADE)
+
+
+class QRCodeUsing(models.Model):
+    organization = models.ForeignKey('organizations.Organization', on_delete=models.CASCADE)
+    used_dt = models.DateTimeField(auto_now=True)
+    client = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.SET_NULL, null=True, default=None)

src/business/repositories.py

@@ -0,0 +1,49 @@
+from datetime import datetime, timedelta
+
+from django.db.models.functions import TruncDate
+
+from core.repositories import BaseRepository, ObjectDoesNotExist
+from .models import OrganizationAccount, QRCodeUsing
+
+
+class OrganizationAccountRepository(BaseRepository):
+    model = OrganizationAccount
+
+    @classmethod
+    def get(cls, **kwargs):
+        try:
+            return cls.model.objects.get(**kwargs)
+        except ObjectDoesNotExist:
+            return None
+
+    @classmethod
+    def filter_by_organization(cls, organization_id):
+        return cls.model.objects.filter(organization__id=organization_id)
+
+    @classmethod
+    def none(cls):
+        return cls.model.objects.none()
+
+
+class QRCodeUsingRepository(BaseRepository):
+    model = QRCodeUsing
+
+    @staticmethod
+    def __get_dates_from_timestamp(timestamp):
+        return datetime.fromtimestamp(timestamp).date(), datetime.fromtimestamp(timestamp).date() + timedelta(days=1)
+
+    @classmethod
+    def get_by_timestamp(cls, organization_id, timestamp):
+        start_date, end_date = cls.__get_dates_from_timestamp(timestamp)
+        return cls.model.objects.filter(organization__pk=organization_id, used_dt__gte=start_date,
+                                        used_dt__lt=end_date)
+
+    @classmethod
+    def get_with_dates(cls, organization_id):
+        return cls.model.objects.filter(organization__pk=organization_id).values(date=TruncDate('used_dt')).distinct()
+
+    @classmethod
+    def count(cls, organization_id, timestamp):
+        start_date, end_date = cls.__get_dates_from_timestamp(timestamp)
+        return cls.model.objects.filter(organization__pk=organization_id, used_dt__gte=start_date,
+                                        used_dt__lt=end_date).count()

src/business/serializers.py

@@ -0,0 +1,39 @@
+from datetime import datetime, time
+from django.conf import settings
+from rest_framework import serializers
+from pytz import timezone
+
+from .models import QRCodeUsing, OrganizationAccount
+from .repositories import QRCodeUsingRepository
+
+
+class ListUsingQRCodeSerializer(serializers.Serializer):
+    date = serializers.SerializerMethodField(method_name='get_timestamp', read_only=True)
+    scansCount = serializers.SerializerMethodField(method_name='get_scans_count', read_only=True)
+
+    def get_timestamp(self, obj):
+        return datetime.combine(obj['date'], time=time(), tzinfo=timezone(settings.TIME_ZONE)).timestamp()
+
+    def get_scans_count(self, obj):
+        pk = int(''.join([i for i in self.context['request'].path if i.isdigit()]))
+        return QRCodeUsingRepository.count(pk, self.get_timestamp(obj))
+
+
+class DetailedUsingQRCodeSerializer(serializers.ModelSerializer):
+
+    timestamp = serializers.SerializerMethodField(method_name='get_timestamp', read_only=True)
+    user = serializers.StringRelatedField(source='client', read_only=True)
+    discountPercent = serializers.IntegerField(source='discount')
+
+    class Meta:
+        model = QRCodeUsing
+        fields = ('id', 'user', 'organization_id', 'timestamp')
+
+    def get_timestamp(self, obj):
+        return obj.used_dt.timestamp()
+
+
+class OrganizationAccountSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = OrganizationAccount
+        fields = ('user', 'organization')

src/business/services.py

@@ -0,0 +1,14 @@
+from crypto.services import decrypt
+from jwtauth.repositories import UserRepository, ObjectDoesNotExist
+
+
+def qr_prove(cryptred_str):
+    try:
+        data = decrypt(cryptred_str)
+    except Exception:
+        return None
+    try:
+        email = data
+        return UserRepository.get(email)
+    except (KeyError, ObjectDoesNotExist):
+        return None

src/business/tests.py

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

src/business/urls.py

@@ -0,0 +1,13 @@
+from django.urls import path
+
+from .views import *
+
+
+urlpatterns = [
+    path('organizations/accounts/<int:id>/', OrganizationAccountDestroyAPIView.as_view()),
+    path('organization/<int:id>/accounts/', OrganizationAccountsListAPIView.as_view()),
+    path('organization/<int:id>/accounts/', OrganizationAccountCreateAPIView.as_view()),
+    path('organization/<int:id>/scans/<int:timestamp>/', QRCodeUsingDetailedListAPIView.as_view()),
+    path('organization/<int:id>/scans/', QRCodeUsingListAPIView.as_view()),
+    path('qr-code/check/', QrCodeProveAPIView.as_view()),
+]

src/business/views.py

@@ -0,0 +1,97 @@
+from drf_yasg import openapi
+from drf_yasg.utils import swagger_auto_schema
+from rest_framework import status
+from rest_framework.generics import ListAPIView, CreateAPIView, DestroyAPIView
+from rest_framework.pagination import PageNumberPagination
+from rest_framework.permissions import IsAuthenticated
+from rest_framework.response import Response
+from rest_framework.views import APIView
+
+from core.permissions import IsOrganizationOwner
+from .repositories import QRCodeUsingRepository, OrganizationAccountRepository
+from .serializers import DetailedUsingQRCodeSerializer, ListUsingQRCodeSerializer, OrganizationAccount
+from .services import qr_prove
+
+# Create your views here.
+
+
+class QRCodeUsingListAPIView(ListAPIView):
+    pagination_class = PageNumberPagination
+    serializer_class = ListUsingQRCodeSerializer
+    lookup_field = 'id'
+
+    def get_queryset(self):
+        return QRCodeUsingRepository.get_with_dates(self.kwargs.get('id'))
+
+
+class QRCodeUsingDetailedListAPIView(ListAPIView):
+    pagination_class = PageNumberPagination
+    serializer_class = DetailedUsingQRCodeSerializer
+    lookup_field = 'id'
+
+    def get_queryset(self):
+        return QRCodeUsingRepository.get_by_timestamp(self.kwargs.get('id'), int(self.kwargs.get('timestamp')))
+
+
+class QrCodeProveAPIView(APIView):
+    permission_classes = (IsAuthenticated,)
+
+    @swagger_auto_schema(request_body=openapi.Schema(
+        type=openapi.TYPE_OBJECT,
+        properties={
+            'qrCodeData': openapi.Schema(type=openapi.TYPE_STRING, description='access_token'),
+        },
+        required=['qrCodeData']
+    ),
+        responses={
+            status.HTTP_200_OK: openapi.Schema(
+                type=openapi.TYPE_OBJECT
+            ),
+            status.HTTP_400_BAD_REQUEST: openapi.Schema(
+                type=openapi.TYPE_OBJECT
+            ),
+            status.HTTP_403_FORBIDDEN: openapi.Schema(
+                type=openapi.TYPE_OBJECT
+            )
+        }
+    )
+    def post(self, request):
+        self.check_permissions(request)
+        oa = OrganizationAccountRepository.get(user=self.request.user)
+        if oa:
+            try:
+                client = qr_prove(self.request.data['qrCodeData'])
+            except KeyError:
+                return Response(status=status.HTTP_400_BAD_REQUEST)
+            if client:
+                QRCodeUsingRepository.create(
+                    organization_id=oa.organization.pk,
+                    client=client
+                )
+                return Response(status=status.HTTP_200_OK)
+            return Response(status=status.HTTP_400_BAD_REQUEST)
+        return Response(status=status.HTTP_403_FORBIDDEN)
+
+
+class OrganizationAccountCreateAPIView(CreateAPIView):
+    serializer_class = OrganizationAccount
+    permission_classes = [IsOrganizationOwner]
+    queryset = OrganizationAccountRepository.all()
+
+
+class OrganizationAccountDestroyAPIView(DestroyAPIView):
+    serializer_class = OrganizationAccount
+    permission_classes = [IsOrganizationOwner]
+    queryset = OrganizationAccountRepository.all()
+
+
+class OrganizationAccountsListAPIView(ListAPIView):
+    serializer_class = OrganizationAccount
+    permission_classes = [IsOrganizationOwner]
+
+    def get_queryset(self):
+        pk = self.kwargs.get('id')
+        if pk:
+            return OrganizationAccountRepository.filter_by_organization(int(pk))
+        else:
+            return OrganizationAccountRepository.none()

src/conf/settings.py

@@ -34,6 +34,7 @@ ALLOWED_HOSTS = ['*']
 
 INSTALLED_APPS = [
     'activities',
+    'business',
     'jwtauth',
     'organizations',
     'jwt',

src/conf/urls.py

@@ -25,7 +25,8 @@ from core.openapi import schema_view
 api_urls = [
     path('organizations/', include('organizations.urls')),
     path('user/', include('jwtauth.urls')),
-    path('activity/', include('activities.urls'))
+    path('activity/', include('activities.urls')),
+    path('business/', include('business.urls'))
 ]
 
 urlpatterns = [

src/core/permissions.py

@@ -1,5 +1,7 @@
 from rest_framework.permissions import BasePermission, SAFE_METHODS
 
+from organizations.repositories import OrganizationRepository
+
 
 class IsAuthorOrReadOnly(BasePermission):
 
@@ -7,4 +9,10 @@ class IsAuthorOrReadOnly(BasePermission):
         if request.method in SAFE_METHODS:
             return True
 
-        return obj.user == request.user
+        return obj.user == request.user
+
+
+class IsOrganizationOwner(BasePermission):
+
+    def has_object_permission(self, request, view, obj):
+        return request.user == OrganizationRepository.get(request.kwargs.get('id'))

src/crypto/crypto_system.py

@@ -0,0 +1,23 @@
+class BaseCryptoMSDU:
+    """
+    class that describes any CryptoSystem
+    encrypt_function is partial solution of MSDU using as encode function
+    private_key is secret key for encode and decode
+    decrypt_function is function to decrypt message
+    for async systems
+    U is module
+    w is open key
+    v is another private key
+    """
+
+    encrypt_function = None
+    secret_key = None
+    decrypt_function = None
+
+    @classmethod
+    def encrypt(cls, msg: int) -> int:
+        return cls.encrypt_function(msg, cls.secret_key)
+
+    @classmethod
+    def decrypt(cls, msg: int) -> int:
+        return cls.decrypt_function(msg, cls.secret_key)

src/crypto/k_gram_coder.py

@@ -0,0 +1,39 @@
+class KGramCoder:
+    """
+    class to describe any k-grams coding,
+    step is ascii letter that start in ur alphabet(better user ord(letter)),
+    alphabet power is length of your alphabet,
+    alphabet - just for add or override codes,
+    use_default_alphabet - if True using ansi coding(and adding your alphabet), else you need to describe your alphabet
+    """
+    k: int = None
+    start_symbol: int = 1
+    alphabet_power: int = None
+    alphabet: dict = {}
+    use_default_alphabet: bool = True
+
+    @classmethod
+    def encode(cls, k_gram: str) -> int:
+        if cls.use_default_alphabet:
+            return sum([cls.alphabet_power ** i * (ord(char) - cls.start_symbol) if char not in
+                                                                                    cls.alphabet else
+                        cls.alphabet_power ** i * cls.alphabet[char] for i, char in enumerate(list(k_gram[::-1]))])
+        else:
+            return sum([cls.alphabet_power ** i * cls.alphabet[char] for i, char in enumerate(list(k_gram[::-1]))])
+
+    @classmethod
+    def decode(cls, encrypted_k_gram: int) -> str:
+        alphabet_codes = dict((v, k) for k, v in cls.alphabet.items())
+        k_gram_list = []
+        for i in range(cls.k):
+            if cls.use_default_alphabet:
+                res = encrypted_k_gram % cls.alphabet_power
+                if res in alphabet_codes.keys():
+                    k_gram_list.append(alphabet_codes[res])
+                else:
+                    k_gram_list.append(str(chr(res + cls.start_symbol)))
+            else:
+                k_gram_list.append(str(alphabet_codes[encrypted_k_gram % cls.alphabet_power]))
+            encrypted_k_gram //= cls.alphabet_power
+
+        return ''.join(k_gram_list)[::-1]

src/crypto/message.py

@@ -0,0 +1,11 @@
+from dataclasses import dataclass
+
+
+@dataclass
+class Message:
+    """
+    dataclass for Message
+    """
+    string: str
+    k: int
+    delimiter: str = ' '

src/crypto/message_coder.py

@@ -0,0 +1,23 @@
+from typing import Type
+
+from .crypto_system import BaseCryptoMSDU
+from .k_gram_coder import KGramCoder
+from .message import Message
+
+
+class MessageCoder:
+    """
+    class for code or decode Message
+    """
+
+    @staticmethod
+    def encode(msg: Message, encrypter_class: Type[BaseCryptoMSDU], k_gram_coder: Type[KGramCoder]):
+        k_gram_list = [msg.string[i:i+2] for i in range(0, len(msg.string), msg.k)]
+        return msg.delimiter.join(map(lambda x: str(encrypter_class.encrypt(x)),
+                                      [k_gram_coder.encode(k_gram) for k_gram in k_gram_list]))
+
+    @staticmethod
+    def decode(msg: Message, decrypter_class: Type[BaseCryptoMSDU], k_gram_coder: Type[KGramCoder]):
+        encrypted_k_gram_list = list(map(int, msg.string.split(msg.delimiter)))
+        return ''.join([k_gram_coder.decode(decrypter_class.decrypt(int(encrypted_k_gram)))
+                        for encrypted_k_gram in encrypted_k_gram_list])

src/crypto/services.py

@@ -0,0 +1,41 @@
+from ast import literal_eval
+
+from crypto.crypto_system import BaseCryptoMSDU
+from crypto.k_gram_coder import KGramCoder
+from crypto.message import Message
+from crypto.message_coder import MessageCoder
+
+
+def encrypt_function(msg: int, key: int) -> int:
+    power = 5
+    return (msg + 3 * key) ** power + (9 * msg + 4 * key) ** power + (12 * msg + 19 * key) ** power + (
+            28 * msg + 21 * key) ** power + (31 * msg + 36 * key) ** power + (39 * msg + 37 * key) ** power - (
+            4 * msg + 9 * key) ** power - (19 * msg + 12 * key) ** power - (21 * msg + 28 * key) ** power - (
+            36 * msg + 31 * key) ** power - (37 * msg + 39 * key) ** power
+
+
+def decrypt_function(coded_msg: int, key: int) -> int:
+    return int(coded_msg ** (1 / 5) - key) // 3
+
+
+class SymmMSDUPowerFive(BaseCryptoMSDU):
+    encrypt_function = encrypt_function
+    secret_key = 9
+    decrypt_function = decrypt_function
+
+
+class BiGramCoder(KGramCoder):
+    k = 2
+    alphabet_power = 128
+
+
+def encrypt(data):
+    str_data = str(data)
+    if len(str_data) % 2:
+        str_data += ' '
+    base_msg = Message(str_data, 2)
+    return MessageCoder.encode(base_msg, SymmMSDUPowerFive, BiGramCoder)
+
+
+def decrypt(st):
+    return literal_eval(MessageCoder.decode(Message(st, 2), SymmMSDUPowerFive, BiGramCoder))

src/jwtauth/migrations/0002_qrcode.py

@@ -0,0 +1,23 @@
+# Generated by Django 5.0.4 on 2024-04-07 02:02
+
+import django.db.models.deletion
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('jwtauth', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='QRCode',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('qr_str', models.CharField(blank=True, default='', max_length=1024)),
+                ('user', models.OneToOneField(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
+            ],
+        ),
+    ]

src/jwtauth/models.py

@@ -6,6 +6,7 @@ from django.contrib.auth.models import PermissionsMixin
 from django.db import models
 from pytz import timezone
 
+from crypto.services import encrypt
 from .managers import CustomUserManager
 from .token_generators import generate_jwt
 
@@ -63,3 +64,14 @@ class Friend(models.Model):
     first = models.ForeignKey(Profile, on_delete=models.CASCADE, related_name='first_friend')
     second = models.ForeignKey(Profile, on_delete=models.CASCADE, related_name='second_friend')
     approved = models.BooleanField(default=False)
+
+
+class QRCode(models.Model):
+    qr_str = models.CharField(max_length=1024, default='', blank=True)
+    user = models.OneToOneField(CustomUser, on_delete=models.CASCADE)
+
+    def save(
+        self, force_insert=False, force_update=False, using=None, update_fields=None
+    ):
+        self.qr_str = encrypt(self.user.email)
+

src/jwtauth/repositories.py

@@ -1,5 +1,5 @@
 from core.repositories import BaseRepository, ObjectDoesNotExist
-from .models import CustomUser, Profile, RefreshToken
+from .models import CustomUser, Profile, RefreshToken, QRCode
 
 
 class UserRepository(BaseRepository):
@@ -37,3 +37,11 @@ class RefreshTokenRepository(BaseRepository):
     @classmethod
     def get(cls, token):
         return cls.model.objects.get(token=token)
+
+
+class QRCodeRepository(BaseRepository):
+    model = QRCode
+
+    @classmethod
+    def get(cls, user):
+        return cls.model.objects.get(user=user)

src/jwtauth/serializers.py

@@ -2,7 +2,7 @@ from django.contrib.auth import authenticate
 
 from rest_framework import serializers
 
-from .models import CustomUser, Profile, RefreshToken
+from .models import CustomUser, Profile, QRCode
 from .repositories import UserRepository, ProfileRepository, RefreshTokenRepository
 from .token_generators import generate_rt
 
@@ -116,3 +116,9 @@ class ProfileSerializer(serializers.ModelSerializer):
     class Meta:
         model = Profile
         fields = ('user', 'first_name', 'last_name', 'title', 'avatar', 'count_friends', 'count_subscribers')
+
+
+class QRCodeSerializer(serializers.ModelSerializer):
+    class Meta:
+        model = QRCode
+        fields = ('qr_str',)

src/jwtauth/urls.py

@@ -1,7 +1,7 @@
 from django.urls import path
 from rest_framework.routers import DefaultRouter
 
-from .views import RegistrationAPIView, LoginAPIView, CustomUserRetrieveUpdateAPIView, RefreshAPIView, ProfileRetrieveUpdateDestroyAPIView, ProfileRetrieveAPIView, ProfileListAPIView
+from .views import *
 
 app_name = 'jwtauth'
 
@@ -12,5 +12,6 @@ urlpatterns = [
     path('profile/<int:id>', ProfileRetrieveAPIView.as_view()),
     path('profiles/', ProfileListAPIView.as_view()),
     path('profile/', ProfileRetrieveUpdateDestroyAPIView.as_view()),
+    path('qr/', QRCodeRetrieveAPIView.as_view()),
     path('', CustomUserRetrieveUpdateAPIView.as_view()),
 ]

src/jwtauth/views.py

@@ -14,8 +14,8 @@ from rest_framework.views import APIView
 
 from core.permissions import IsAuthorOrReadOnly
 from .models import CustomUser, RefreshToken
-from .repositories import RefreshTokenRepository, ProfileRepository, UserRepository
-from .serializers import RegistrationSerializer, LoginSerializer, CustomUserSerializer, ProfileSerializer
+from .repositories import RefreshTokenRepository, ProfileRepository, UserRepository, QRCodeRepository
+from .serializers import RegistrationSerializer, LoginSerializer, CustomUserSerializer, ProfileSerializer, QRCodeSerializer
 from .renderers import CustomUserJSONRenderer
 from .token_generators import generate_rt, generate_jwt
 
@@ -216,3 +216,11 @@ class ProfileListAPIView(ListAPIView):
     serializer_class = ProfileSerializer
     permission_classes = [IsAdminUser]
     pagination_class = PageNumberPagination
+
+
+class QRCodeRetrieveAPIView(RetrieveAPIView):
+    serializer_class = QRCodeSerializer
+    queryset = QRCodeRepository.all()
+
+    def get_object(self):
+        return QRCodeRepository.get(self.request.user)