Compare commits

..

7 Commits

Author SHA1 Message Date
David Bruno Vontor
4791bbc92c websockets + chat app (django) 2025-10-31 13:32:39 +01:00
8dd4f6e731 converter 2025-10-30 01:58:28 +01:00
dd9d076bd2 okay 2025-10-29 00:58:37 +01:00
73da41b514 commit 2025-10-28 03:21:01 +01:00
10796dcb31 integrace api, stripe, vytvoření commecre app 2025-10-05 23:41:14 +02:00
f5cf8bbaa7 style changes 2025-10-03 01:48:36 +02:00
d0227e4539 fixed components 2025-10-02 02:10:07 +02:00
576 changed files with 10721 additions and 354345 deletions

119
.github/copilot-instructions.md vendored Normal file
View File

@@ -0,0 +1,119 @@
# Copilot Instructions for Vontor CZ
## Overview
This monorepo contains a Django backend and a Vite/React frontend, orchestrated via Docker Compose. The project is designed for a Czech e-marketplace, with custom payment integrations and real-time features.
## Architecture
- **backend/**: Django project (`vontor_cz`), custom `account` app, and third-party payment integrations (`thirdparty/`).
- Uses Django REST Framework, Channels (ASGI), Celery, and S3/static/media via `django-storages`.
- Custom user model: `account.CustomUser`.
- API docs: DRF Spectacular (`/api/schema/`).
- **frontend/**: Vite + React + TypeScript app.
- Organized by `src/api/`, `components/`, `features/`, `layouts/`, `pages/`, `routes/`.
- Uses React Router layouts and nested routes (see `src/layouts/`, `src/routes/`).
- Uses Tailwind CSS for styling (configured via `src/index.css` with `@import "tailwindcss";`). Prefer utility classes over custom CSS.
## Developer Workflows
- **Backend**
- Local dev: `python manage.py runserver` (or use Docker Compose)
- Migrations: `python manage.py makemigrations && python manage.py migrate`
- Celery: `celery -A vontor_cz worker -l info`
- Channels: Daphne/ASGI (see Docker Compose command)
- Env config: `.env` files in `backend/` (see `.gitignore` for secrets)
- **Frontend**
- Install: `npm install`
- Dev server: `npm run dev`
- Build: `npm run build`
- Preview: `npm run preview`
- Static assets: `src/assets/` (import in JS/CSS), `public/` (referenced in HTML)
## Conventions & Patterns
- **Backend**
- Use environment variables for secrets and config (see `settings.py`).
- Static/media files: S3 in production, local in dev (see `settings.py`).
- API versioning and docs: DRF Spectacular config in `settings.py`.
- Custom permissions, filters, and serializers in each app.
- **Frontend**
- Use React Router layouts for shared UI (see `src/layouts/`, `LAYOUTS.md`).
- API calls and JWT handling in `src/api/`.
- Route definitions and guards in `src/routes/` (`ROUTES.md`).
- Use TypeScript strict mode (see `tsconfig.*.json`).
- Linting: ESLint config in `eslint.config.js`.
- Styling: Tailwind CSS is present. Prefer utility classes; keep minimal component-scoped CSS. Global/base styles live in `src/index.css`. Avoid inline styles and CSS-in-JS unless necessary.
### Frontend API Client (required)
All frontend API calls must use the shared client at frontend/src/api/Client.ts.
- Client.public: no cookies, no Authorization header (for public Django endpoints).
- Client.auth: sends cookies and includes Bearer token; auto-refreshes on 401 (retries up to 2x).
- Centralized error handling: subscribe via Client.onError to show toasts/snackbars.
- Tokens are stored in cookies by Client.setTokens and cleared by Client.clearTokens.
Example usage (TypeScript)
```ts
import Client from "@/api/Client";
// Public request (no credentials)
async function listPublicItems() {
const res = await Client.public.get("/api/public/items/");
return res.data;
}
// Login (obtain tokens and persist to cookies)
async function login(username: string, password: string) {
// Default SimpleJWT endpoint (adjust if your backend differs)
const res = await Client.public.post("/api/token/", { username, password });
const { access, refresh } = res.data;
Client.setTokens(access, refresh);
}
// Authenticated requests (auto Bearer + refresh on 401)
async function fetchProfile() {
const res = await Client.auth.get("/api/users/me/");
return res.data;
}
function logout() {
Client.clearTokens();
window.location.assign("/login");
}
// Global error toasts
import { useEffect } from "react";
function useApiErrors(showToast: (msg: string) => void) {
useEffect(() => {
const unsubscribe = Client.onError((e) => {
const { message, status } = e.detail;
showToast(status ? `${status}: ${message}` : message);
});
return unsubscribe;
}, [showToast]);
}
```
Vite env used by the client:
- VITE_API_BASE_URL (default: http://localhost:8000)
- VITE_API_REFRESH_URL (default: /api/token/refresh/)
- VITE_LOGIN_PATH (default: /login)
Notes
- Public client never sends cookies or Authorization.
- Ensure Django CORS settings allow your frontend origin. See backend/vontor_cz/settings.py.
- Use React Router layouts and guards as documented in frontend/src/routes/ROUTES.md and frontend/src/layouts/LAYOUTS.md.
## Integration Points
- **Payments**: `thirdparty/` contains custom integrations for Stripe, GoPay, Trading212.
- **Real-time**: Django Channels (ASGI, Redis) for websockets.
- **Task queue**: Celery + Redis for async/background jobs.
- **API**: REST endpoints, JWT auth, API key support.
## References
- [frontend/REACT.md](../frontend/REACT.md): Frontend structure, workflows, and conventions.
- [frontend/src/layouts/LAYOUTS.md](../frontend/src/layouts/LAYOUTS.md): Layout/component patterns.
- [frontend/src/routes/ROUTES.md](../frontend/src/routes/ROUTES.md): Routing conventions.
- [backend/vontor_cz/settings.py](../backend/vontor_cz/settings.py): All backend config, env, and integration details.
- [docker-compose.yml](../docker-compose.yml): Service orchestration and dev workflow.
---
**When in doubt, check the referenced markdown files and `settings.py` for project-specific logic and patterns.**

View File

@@ -1,165 +0,0 @@
#
*.mp4
backups/
collectedstaticfiles/
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/latest/usage/project/#working-with-version-control
.pdm.toml
.pdm-python
.pdm-build/
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/

15
.vscode/launch.json vendored Normal file
View File

@@ -0,0 +1,15 @@
{
// Use IntelliSense to learn about possible attributes.
// Hover to view descriptions of existing attributes.
// For more information, visit: https://go.microsoft.com/fwlink/?linkid=830387
"version": "0.2.0",
"configurations": [
{
"type": "chrome",
"request": "launch",
"name": "Launch Chrome against localhost",
"url": "http://localhost:8080",
"webRoot": "${workspaceFolder}"
}
]
}

View File

@@ -1,23 +0,0 @@
# vontor-cz
## venv
- windows
```
Set-ExecutionPolicy -Scope Process -ExecutionPolicy RemoteSigned
python -m venv venv
.\venv\Scripts\Activate
#start server
daphne -b localhost -p 8000 vontor_cz.asgi:application
```
## docker compose
spuštění dockeru pro lokální hosting, s instantníma změnami během editace ve vscodu.
```docker-compose up --build```
## dns reset windows
```ipconfig /flushdns```

View File

@@ -1,37 +0,0 @@
from django.contrib import admin
from .models import MyModel # Importuj modely
# Příklad přizpůsobení zobrazení modelu v administrátorské sekci
class MyModelAdmin(admin.ModelAdmin):
# Určují se pole, která se zobrazí v seznamu (list view)
list_display = ('field1', 'field2', 'field3')
# Určuje, podle kterých polí lze vyhledávat
search_fields = ('field1', 'field2')
# Aktivuje filtrování podle hodnoty pole v pravém postranním panelu
list_filter = ('field1', 'field2')
# Určuje pole, která se zobrazí ve formuláři při detailním pohledu na model
fields = ('field1', 'field2', 'field3')
# Definuje rozložení polí ve formuláři
fieldsets = (
(None, {
'fields': ('field1', 'field2'),
}),
('Další informace', {
'classes': ('collapse',),
'fields': ('field3',),
}),
)
# Nastavuje výchozí řazení záznamů při jejich zobrazení
ordering = ('field1',)
# Určuje počet záznamů zobrazených na jedné stránce
list_per_page = 10
# Definuje akce dostupné pro vybrané objekty
actions = ['custom_action']
# Příklad vlastní akce
def custom_action(self, request, queryset):
# Vlastní logika pro akci
queryset.update(field1='Updated Value')
# Registrování modelu s vlastními nastaveními administrátorského rozhraní
admin.site.register(MyModel, MyModelAdmin)

View File

@@ -1,14 +0,0 @@
from django.db import models
# Create your models here.
from django.contrib.auth.models import User
class UserProfile(models.Model):
user = models.OneToOneField(User, on_delete=models.CASCADE)
# Add custom fields here
bio = models.TextField(blank=True)
birthdate = models.DateField(null=True, blank=True)
profile_picture = models.ImageField(upload_to='profile_pics/', null=True, blank=True)
def __str__(self):
return f'{self.user.username} Profile'

View File

@@ -1,7 +0,0 @@
from rest_framework_api_key.permissions import HasAPIKey
class UserEditAPIKeyPermissions(HasAPIKey):
"""
Custom permision for restricting access using API key.
"""
pass

View File

@@ -1,21 +0,0 @@
from rest_framework import serializers
from .models import User
#Serializers are for what views can show fields of models
class PublicUserSerializers(serializers.ModelSerializer):
"""
Serializer for public User fields
"""
class Meta:
model = User
fields = ['id', 'username']
class SecureUserSerializers(serializers.ModelSerializer):
"""
Serializer for all User fields
Requires API key
"""
class Meta:
model = User
fields = ['id', 'username', 'email', 'first_name', 'last_name', 'bio']

View File

@@ -1,10 +0,0 @@
from django.urls import path
from .views import PublicUserView, SecureUserUpdateView
urlpatterns = [
# URL for the public view to list users with public fields
path('users/', PublicUserView.as_view(), name='public-user-list'),
# URL for secure view to retrieve and update user with all fields
path('users/<int:pk>/', SecureUserUpdateView.as_view(), name='secure-user-update'),
]

View File

@@ -1,20 +0,0 @@
from django.shortcuts import render
# Create your views here.
from rest_framework import generics, permissions
from .models import User
from .serializers import PublicUserSerializers, SecureUserSerializers
from .permissions import UserEditAPIKeyPermissions
#Public view: List users with only public fields
class PublicUserView(generics.ListAPIView):
queryset = User.objects.all()
serializer_class = PublicUserSerializers
permission_classes = [permissions.AllowAny]
#Secure view for retrive/update user all fields (API key)
class SecureUserUpdateView(generics.RetrieveUpdateAPIView):
queryset = User.objects.all()
serializer_class = SecureUserSerializers
permission_classes = [UserEditAPIKeyPermissions]

12
backend/Dockerfile Normal file
View File

@@ -0,0 +1,12 @@
FROM python:3.12-slim
WORKDIR /app
RUN apt update && apt install ffmpeg -y
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000

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.

View File

@@ -1,6 +1,6 @@
from django.apps import AppConfig
class HomeConfig(AppConfig):
class AccountConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'home'
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.7 on 2025-10-28 22:28
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(choices=[('admin', 'Admin'), ('mod', 'Moderator'), ('regular', 'Regular')], default='regular', max_length=20)),
('phone_number', models.CharField(blank=True, max_length=16, null=True, unique=True, validators=[django.core.validators.RegexValidator('^\\+?\\d{9,15}$', message='Zadejte platné telefonní číslo.')])),
('email_verified', models.BooleanField(default=False)),
('email', models.EmailField(db_index=True, max_length=254, unique=True)),
('gdpr', models.BooleanField(default=False)),
('is_active', models.BooleanField(default=False)),
('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}$')])),
('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.CustomUserManager()),
('active', account.models.ActiveUserManager()),
],
),
]

View File

@@ -0,0 +1,23 @@
# Generated by Django 5.2.7 on 2025-10-31 07:36
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('account', '0001_initial'),
]
operations = [
migrations.AddField(
model_name='customuser',
name='email_verification_sent_at',
field=models.DateTimeField(blank=True, null=True),
),
migrations.AddField(
model_name='customuser',
name='email_verification_token',
field=models.CharField(blank=True, db_index=True, max_length=128, null=True),
),
]

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

@@ -0,0 +1,160 @@
import uuid
from django.db import models
from django.contrib.auth.models import AbstractUser, UserManager, Group, Permission
from django.core.validators import RegexValidator
from django.utils.crypto import get_random_string
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__)
class CustomUserManager(UserManager):
# Inherit get_by_natural_key and all auth behaviors
use_in_migrations = True
class ActiveUserManager(CustomUserManager):
def get_queryset(self):
return super().get_queryset().filter(is_active=True)
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",
)
class Role(models.TextChoices):
ADMIN = "admin", "Admin"
MANAGER = "mod", "Moderator"
CUSTOMER = "regular", "Regular"
role = models.CharField(max_length=20, choices=Role.choices, default=Role.CUSTOMER)
phone_number = models.CharField(
null=True,
blank=True,
unique=True,
max_length=16,
validators=[RegexValidator(r'^\+?\d{9,15}$', message="Zadejte platné telefonní číslo.")]
)
email_verified = models.BooleanField(default=False)
email = models.EmailField(unique=True, db_index=True)
# + fields for email verification flow
email_verification_token = models.CharField(max_length=128, null=True, blank=True, db_index=True)
email_verification_sent_at = models.DateTimeField(null=True, blank=True)
gdpr = models.BooleanField(default=False)
is_active = models.BooleanField(default=False)
create_time = models.DateTimeField(auto_now_add=True)
city = models.CharField(null=True, blank=True, max_length=100)
street = models.CharField(null=True, blank=True, max_length=200)
postal_code = models.CharField(
blank=True,
null=True,
max_length=5,
validators=[
RegexValidator(
regex=r'^\d{5}$',
message="Postal code must contain exactly 5 digits.",
code='invalid_postal_code'
)
]
)
USERNAME_FIELD = "username"
REQUIRED_FIELDS = [
"email"
]
# Ensure default manager has get_by_natural_key
objects = CustomUserManager()
# Optional convenience manager for active users only
active = ActiveUserManager()
def delete(self, *args, **kwargs):
self.is_active = False
return super().delete(*args, **kwargs)
def save(self, *args, **kwargs):
is_new = self._state.adding # True if object hasn't been saved yet
# Pre-save flags for new users
if is_new:
if self.is_superuser or self.role == "admin":
# ensure admin flags are consistent
self.is_active = True
self.is_staff = True
self.is_superuser = True
self.role = "admin"
else:
self.is_staff = False
# First save to obtain a primary key
super().save(*args, **kwargs)
# Assign group after we have a PK
if is_new:
from django.contrib.auth.models import Group
group, _ = Group.objects.get_or_create(name=self.role)
# Use add/set now that PK exists
self.groups.set([group])
return super().save(*args, **kwargs)
def generate_email_verification_token(self, length: int = 48, save: bool = True) -> str:
token = get_random_string(length=length)
self.email_verification_token = token
self.email_verification_sent_at = timezone.now()
if save:
self.save(update_fields=["email_verification_token", "email_verification_sent_at"])
return token
def verify_email_token(self, token: str, max_age_hours: int = 48, save: bool = True) -> bool:
if not token or not self.email_verification_token:
return False
# optional expiry check
if self.email_verification_sent_at:
age = timezone.now() - self.email_verification_sent_at
if age > timedelta(hours=max_age_hours):
return False
if token != self.email_verification_token:
return False
if not self.email_verified:
self.email_verified = True
# clear token after success
self.email_verification_token = None
self.email_verification_sent_at = None
if save:
self.save(update_fields=["email_verified", "email_verification_token", "email_verification_sent_at"])
return True

View File

@@ -0,0 +1,57 @@
from urllib import request
from rest_framework.permissions import BasePermission, SAFE_METHODS
from rest_framework.permissions import IsAuthenticated
from rest_framework_api_key.permissions import HasAPIKey
#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.
Allows access if a valid API key is provided.
Args:
RoleAllowed('admin', 'user')
"""
class SafeOrRolePermission(BasePermission):
def has_permission(self, request, view):
# Má API klíč?
has_api_key = HasAPIKey().has_permission(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,205 @@
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",
"email_verified",
"phone_number",
"create_time",
"city",
"street",
"postal_code",
"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

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

@@ -0,0 +1,152 @@
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__)
def send_email_with_context(recipients, subject, message=None, template_name=None, html_template_name=None, context=None):
"""
General function to send emails with a specific context.
Supports rendering plain text and HTML templates.
Converts `user` in context to a plain dict to avoid template access to the model.
"""
if isinstance(recipients, str):
recipients = [recipients]
html_message = None
if template_name or html_template_name:
# Best effort to resolve both templates if only one provided
if not template_name and html_template_name:
template_name = html_template_name.replace(".html", ".txt")
if not html_template_name and template_name:
html_template_name = template_name.replace(".txt", ".html")
ctx = dict(context or {})
# Sanitize user if someone passes the model by mistake
if "user" in ctx and not isinstance(ctx["user"], dict):
try:
ctx["user"] = _build_user_template_ctx(ctx["user"])
except Exception:
ctx["user"] = {}
message = render_to_string(template_name, ctx)
html_message = render_to_string(html_template_name, ctx)
try:
send_mail(
subject=subject,
message=message or "",
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(f"\nEMAIL OBSAH:\n{message}\nKONEC OBSAHU")
return True
except Exception as e:
logger.error(f"E-mail se neodeslal: {e}")
return False
def _build_user_template_ctx(user: CustomUser) -> dict:
"""
Return a plain dict for templates instead of passing the DB model.
Provides aliases to avoid template errors (firstname vs first_name).
Adds a backward-compatible key 'get_full_name' for templates using `user.get_full_name`.
"""
first_name = getattr(user, "first_name", "") or ""
last_name = getattr(user, "last_name", "") or ""
full_name = f"{first_name} {last_name}".strip()
return {
"id": user.pk,
"email": getattr(user, "email", "") or "",
"first_name": first_name,
"firstname": first_name, # alias for templates using `firstname`
"last_name": last_name,
"lastname": last_name, # alias for templates using `lastname`
"full_name": full_name,
"get_full_name": full_name, # compatibility for templates using method-style access
}
#----------------------------------------------------------------------------------------------------
# This function sends an email to the user for email verification after registration.
@shared_task
def send_email_verification_task(user_id):
try:
user = CustomUser.objects.get(pk=user_id)
except CustomUser.DoesNotExist:
logger.info(f"Task send_email_verification has failed. Invalid User ID was sent.")
return 0
uid = urlsafe_base64_encode(force_bytes(user.pk))
# {changed} generate and store a per-user token
token = user.generate_email_verification_token()
verify_url = f"{settings.FRONTEND_URL}/email-verification/?uidb64={uid}&token={token}"
context = {
"user": _build_user_template_ctx(user),
"action_url": verify_url,
"frontend_url": settings.FRONTEND_URL,
"cta_label": "Ověřit email",
}
send_email_with_context(
recipients=user.email,
subject="Ověření emailu",
template_name="email/email_verification.txt",
html_template_name="email/email_verification.html",
context=context,
)
@shared_task
def send_email_test_task(email):
context = {
"action_url": settings.FRONTEND_URL,
"frontend_url": settings.FRONTEND_URL,
"cta_label": "Otevřít aplikaci",
}
send_email_with_context(
recipients=email,
subject="Testovací email",
template_name="email/test.txt",
html_template_name="email/test.html",
context=context,
)
@shared_task
def send_password_reset_email_task(user_id):
try:
user = CustomUser.objects.get(pk=user_id)
except CustomUser.DoesNotExist:
logger.info(f"Task send_password_reset_email has failed. Invalid User ID was sent.")
return 0
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}"
context = {
"user": _build_user_template_ctx(user),
"action_url": reset_url,
"frontend_url": settings.FRONTEND_URL,
"cta_label": "Obnovit heslo",
}
send_email_with_context(
recipients=user.email,
subject="Obnova hesla",
template_name="email/password_reset.txt",
html_template_name="email/password_reset.html",
context=context,
)

View File

@@ -0,0 +1,46 @@
<!doctype html>
<html lang="cs">
<body style="margin:0; padding:0; background-color:#f5f7fb;">
<table role="presentation" cellpadding="0" cellspacing="0" border="0" width="100%" style="background-color:#f5f7fb;">
<tr>
<td align="center" style="padding:24px;">
<table role="presentation" cellpadding="0" cellspacing="0" border="0" width="600" style="width:600px; max-width:100%; background-color:#ffffff; border:1px solid #e5e7eb;">
<tr>
<td style="background-color:#111827; color:#ffffff; font-family:Arial, Helvetica, sans-serif; font-size:18px; font-weight:bold; padding:16px 20px;">
Ověření emailu
</td>
</tr>
<tr>
<td style="padding:20px; font-family:Arial, Helvetica, sans-serif; font-size:14px; color:#1f2937; line-height:1.6;">
{% with name=user.first_name|default:user.firstname|default:user.get_full_name %}
<p style="margin:0 0 12px 0;">Dobrý den{% if name %} {{ name }}{% endif %},</p>
{% endwith %}
<p style="margin:0 0 16px 0;">Děkujeme za registraci. Prosíme, ověřte svou emailovou adresu kliknutím na tlačítko níže.</p>
{% if action_url and cta_label %}
<table role="presentation" cellpadding="0" cellspacing="0" border="0" style="margin:16px 0;">
<tr>
<td bgcolor="#2563eb" style="border-radius:6px;">
<a href="{{ action_url }}" target="_blank" style="display:inline-block; padding:10px 16px; font-family:Arial, Helvetica, sans-serif; font-size:14px; color:#ffffff; text-decoration:none; font-weight:bold;">
{{ cta_label }}
</a>
</td>
</tr>
</table>
<p style="margin:0; color:#6b7280; font-size:12px;">Pokud tlačítko nefunguje, zkopírujte do prohlížeče tento odkaz:<br><span style="word-break:break-all;">{{ action_url }}</span></p>
{% endif %}
</td>
</tr>
</table>
<table role="presentation" cellpadding="0" cellspacing="0" border="0" width="600" style="width:600px; max-width:100%; margin-top:12px;">
<tr>
<td align="center" style="font-family:Arial, Helvetica, sans-serif; font-size:12px; color:#6b7280;">
Tento email byl odeslán z aplikace etržnice.
</td>
</tr>
</table>
</td>
</tr>
</table>
</body>
</html>

View File

@@ -0,0 +1,7 @@
{% with name=user.first_name|default:user.firstname|default:user.get_full_name %}Dobrý den{% if name %} {{ name }}{% endif %},{% endwith %}
Děkujeme za registraci. Prosíme, ověřte svou emailovou adresu kliknutím na následující odkaz:
{{ action_url }}
Pokud jste účet nevytvořili vy, tento email ignorujte.

View File

@@ -0,0 +1,46 @@
<!doctype html>
<html lang="cs">
<body style="margin:0; padding:0; background-color:#f5f7fb;">
<table role="presentation" cellpadding="0" cellspacing="0" border="0" width="100%" style="background-color:#f5f7fb;">
<tr>
<td align="center" style="padding:24px;">
<table role="presentation" cellpadding="0" cellspacing="0" border="0" width="600" style="width:600px; max-width:100%; background-color:#ffffff; border:1px solid #e5e7eb;">
<tr>
<td style="background-color:#111827; color:#ffffff; font-family:Arial, Helvetica, sans-serif; font-size:18px; font-weight:bold; padding:16px 20px;">
Obnova hesla
</td>
</tr>
<tr>
<td style="padding:20px; font-family:Arial, Helvetica, sans-serif; font-size:14px; color:#1f2937; line-height:1.6;">
{% with name=user.first_name|default:user.firstname|default:user.get_full_name %}
<p style="margin:0 0 12px 0;">Dobrý den{% if name %} {{ name }}{% endif %},</p>
{% endwith %}
<p style="margin:0 0 12px 0;">Obdrželi jste tento email, protože byla požádána obnova hesla k vašemu účtu. Pokud jste o změnu nepožádali, tento email ignorujte.</p>
{% if action_url and cta_label %}
<table role="presentation" cellpadding="0" cellspacing="0" border="0" style="margin:16px 0;">
<tr>
<td bgcolor="#2563eb" style="border-radius:6px;">
<a href="{{ action_url }}" target="_blank" style="display:inline-block; padding:10px 16px; font-family:Arial, Helvetica, sans-serif; font-size:14px; color:#ffffff; text-decoration:none; font-weight:bold;">
{{ cta_label }}
</a>
</td>
</tr>
</table>
<p style="margin:0; color:#6b7280; font-size:12px;">Pokud tlačítko nefunguje, zkopírujte do prohlížeče tento odkaz:<br><span style="word-break:break-all;">{{ action_url }}</span></p>
{% endif %}
</td>
</tr>
</table>
<table role="presentation" cellpadding="0" cellspacing="0" border="0" width="600" style="width:600px; max-width:100%; margin-top:12px;">
<tr>
<td align="center" style="font-family:Arial, Helvetica, sans-serif; font-size:12px; color:#6b7280;">
Tento email byl odeslán z aplikace etržnice.
</td>
</tr>
</table>
</td>
</tr>
</table>
</body>
</html>

View File

@@ -0,0 +1,7 @@
{% with name=user.first_name|default:user.firstname|default:user.get_full_name %}Dobrý den{% if name %} {{ name }}{% endif %},{% endwith %}
Obdrželi jste tento email, protože byla požádána obnova hesla k vašemu účtu.
Pokud jste o změnu nepožádali, tento email ignorujte.
Pro nastavení nového hesla použijte tento odkaz:
{{ action_url }}

View File

@@ -0,0 +1,44 @@
<!doctype html>
<html lang="cs">
<body style="margin:0; padding:0; background-color:#f5f7fb;">
<table role="presentation" cellpadding="0" cellspacing="0" border="0" width="100%" style="background-color:#f5f7fb;">
<tr>
<td align="center" style="padding:24px;">
<table role="presentation" cellpadding="0" cellspacing="0" border="0" width="600" style="width:600px; max-width:100%; background-color:#ffffff; border:1px solid #e5e7eb;">
<tr>
<td style="background-color:#111827; color:#ffffff; font-family:Arial, Helvetica, sans-serif; font-size:18px; font-weight:bold; padding:16px 20px;">
Testovací email
</td>
</tr>
<tr>
<td style="padding:20px; font-family:Arial, Helvetica, sans-serif; font-size:14px; color:#1f2937; line-height:1.6;">
<p style="margin:0 0 12px 0;">Dobrý den,</p>
<p style="margin:0 0 16px 0;">Toto je testovací email z aplikace etržnice.</p>
{% if action_url and cta_label %}
<table role="presentation" cellpadding="0" cellspacing="0" border="0" style="margin:16px 0;">
<tr>
<td bgcolor="#2563eb" style="border-radius:6px;">
<a href="{{ action_url }}" target="_blank" style="display:inline-block; padding:10px 16px; font-family:Arial, Helvetica, sans-serif; font-size:14px; color:#ffffff; text-decoration:none; font-weight:bold;">
{{ cta_label }}
</a>
</td>
</tr>
</table>
<p style="margin:0; color:#6b7280; font-size:12px;">Pokud tlačítko nefunguje, zkopírujte do prohlížeče tento odkaz:<br><span style="word-break:break-all;">{{ action_url }}</span></p>
{% endif %}
</td>
</tr>
</table>
<table role="presentation" cellpadding="0" cellspacing="0" border="0" width="600" style="width:600px; max-width:100%; margin-top:12px;">
<tr>
<td align="center" style="font-family:Arial, Helvetica, sans-serif; font-size:12px; color:#6b7280;">
Tento email byl odeslán z aplikace etržnice.
</td>
</tr>
</table>
</td>
</tr>
</table>
</body>
</html>

View File

@@ -0,0 +1,6 @@
Dobrý den,
Toto je testovací email z aplikace etržnice.
Odkaz na aplikaci:
{{ action_url }}

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

@@ -0,0 +1,28 @@
from django.test import TestCase
from django.contrib.auth import get_user_model
from rest_framework.test import APIClient
class UserViewAnonymousTests(TestCase):
def setUp(self):
self.client = APIClient()
User = get_user_model()
self.target_user = User.objects.create_user(
username="target",
email="target@example.com",
password="pass1234",
is_active=True,
)
def test_anonymous_update_user_is_forbidden_and_does_not_crash(self):
url = f"/api/account/users/{self.target_user.id}/"
payload = {"username": "newname", "email": self.target_user.email}
resp = self.client.put(url, data=payload, format="json")
# Expect 403 Forbidden (permission denied), but most importantly no 500 error
self.assertEqual(resp.status_code, 403, msg=f"Unexpected status: {resp.status_code}, body={getattr(resp, 'data', resp.content)}")
def test_anonymous_retrieve_user_is_unauthorized(self):
url = f"/api/account/users/{self.target_user.id}/"
resp = self.client.get(url)
# Retrieve requires authentication per view; expect 401 Unauthorized
self.assertEqual(resp.status_code, 401, msg=f"Unexpected status: {resp.status_code}, body={getattr(resp, 'data', resp.content)}")

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/
]

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

@@ -0,0 +1,425 @@
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)
#---------------------------------------------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']:
user = getattr(self, 'request', None) and getattr(self.request, 'user', None)
# Admins can modify any user
if user and getattr(user, 'is_authenticated', False) and getattr(user, 'role', None) == 'admin':
return [OnlyRolesAllowed("admin")()]
# Users can modify their own record
if user and getattr(user, 'is_authenticated', False) and self.kwargs.get('pk') and str(getattr(user, 'id', '')) == self.kwargs['pk']:
return [IsAuthenticated()]
# Fallback - deny access (prevents AttributeError for AnonymousUser)
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)

View File

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

View File

@@ -0,0 +1,3 @@
from django.db import models
# Create your models here.

View File

@@ -0,0 +1,2 @@
#udělat zasílaní reklamních emailů uživatelům.
#newletter --> když se vytvoří nový record s reklamou email se uloží pomocí zaškrtnutí tlačítka v záznamu

View File

@@ -0,0 +1,3 @@
from django.shortcuts import render
# Create your views here.

14
backend/commerce/admin.py Normal file
View File

@@ -0,0 +1,14 @@
from django.contrib import admin
from .models import Carrier, Product
# Register your models here.
@admin.register(Carrier)
class CarrierAdmin(admin.ModelAdmin):
list_display = ("name", "base_price", "is_active")
@admin.register(Product)
class ProductAdmin(admin.ModelAdmin):
list_display = ("name", "price", "currency", "stock", "is_active")
search_fields = ("name", "description")

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

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

View File

@@ -0,0 +1,41 @@
# Generated by Django 5.2.7 on 2025-10-28 22:28
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
]
operations = [
migrations.CreateModel(
name='Carrier',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=100)),
('base_price', models.DecimalField(decimal_places=2, default=0, max_digits=10)),
('delivery_time', models.CharField(blank=True, max_length=100)),
('is_active', models.BooleanField(default=True)),
('logo', models.ImageField(blank=True, null=True, upload_to='carriers/')),
('external_id', models.CharField(blank=True, max_length=50, null=True)),
],
),
migrations.CreateModel(
name='Product',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('name', models.CharField(max_length=200)),
('description', models.TextField(blank=True)),
('price', models.DecimalField(decimal_places=2, max_digits=10)),
('currency', models.CharField(default='czk', max_length=10)),
('stock', models.PositiveIntegerField(default=0)),
('is_active', models.BooleanField(default=True)),
('created_at', models.DateTimeField(auto_now_add=True)),
('default_carrier', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, related_name='default_for_products', to='commerce.carrier')),
],
),
]

View File

@@ -0,0 +1,40 @@
from django.db import models
class Product(models.Model):
name = models.CharField(max_length=200)
description = models.TextField(blank=True)
price = models.DecimalField(max_digits=10, decimal_places=2)
currency = models.CharField(max_length=10, default="czk")
stock = models.PositiveIntegerField(default=0)
is_active = models.BooleanField(default=True)
default_carrier = models.ForeignKey(
"Carrier", on_delete=models.SET_NULL, null=True, blank=True, related_name="default_for_products"
)
created_at = models.DateTimeField(auto_now_add=True)
@property
def available(self):
return self.is_active and self.stock > 0
def __str__(self):
return f"{self.name} ({self.price} {self.currency.upper()})"
# Dopravci a způsoby dopravy
from django.db import models
class Carrier(models.Model):
name = models.CharField(max_length=100) # název dopravce (Zásilkovna, Česká pošta…)
base_price = models.DecimalField(max_digits=10, decimal_places=2, default=0) # základní cena dopravy
delivery_time = models.CharField(max_length=100, blank=True) # např. "23 pracovní dny"
is_active = models.BooleanField(default=True)
# pole pro logo
logo = models.ImageField(upload_to="carriers/", blank=True, null=True)
# pole pro propojení s externím API (např. ID služby u Zásilkovny)
external_id = models.CharField(max_length=50, blank=True, null=True)
def __str__(self):
return f"{self.name} ({self.base_price} Kč)"

View File

@@ -0,0 +1,26 @@
from rest_framework import serializers
from .models import Carrier
class CarrierSerializer(serializers.ModelSerializer):
class Meta:
model = Carrier
fields = [
"id", "name", "base_price", "delivery_time",
"is_active", "logo", "external_id"
]
from rest_framework import serializers
from .models import Product, Carrier, Order
class ProductSerializer(serializers.ModelSerializer):
class Meta:
model = Product
fields = "__all__"
class CarrierSerializer(serializers.ModelSerializer):
class Meta:
model = Carrier
fields = "__all__"

View File

@@ -0,0 +1,3 @@
from django.shortcuts import render
# Create your views here.

59
backend/env Normal file
View File

@@ -0,0 +1,59 @@
# ------------------ NGINX ------------------
FRONTEND_URL=http://192.168.67.98
#FRONTEND_URL=http://localhost:5173
# ------------------ CORE ------------------
DEBUG=True
SSL=False
DJANGO_SECRET_KEY=CHANGE_ME_SECURE_RANDOM_KEY
# ------------------ DATABASE (Postgres in Docker) ------------------
USE_DOCKER_DB=True
DATABASE_ENGINE=django.db.backends.postgresql
DATABASE_HOST=db
DATABASE_PORT=5432
POSTGRES_DB=djangoDB
POSTGRES_USER=dockerDBuser
POSTGRES_PASSWORD=AWSJeMocDrahaZalezitost
# Legacy/unused (was: USE_PRODUCTION_DB) removed
# ------------------ MEDIA / STATIC ------------------
#MEDIA_URL=http://192.168.67.98/media/
# ------------------ REDIS / CACHING / CHANNELS ------------------
# Was REDIS=... (not used). Docker expects REDIS_PASSWORD.
REDIS_PASSWORD=passwd
# ------------------ CELERY ------------------
CELERY_BROKER_URL=redis://redis:6379/0
CELERY_RESULT_BACKEND=redis://redis:6379/0
CELERY_ACCEPT_CONTENT=json
CELERY_TASK_SERIALIZER=json
CELERY_TIMEZONE=Europe/Prague
CELERY_BEAT_SCHEDULER=django_celery_beat.schedulers:DatabaseScheduler
# ------------------ EMAIL (dev/prod logic in settings) ------------------
EMAIL_HOST_DEV=kerio4.vitkovice.cz
EMAIL_PORT_DEV=465
EMAIL_USER_DEV=Test.django@vitkovice.cz
EMAIL_USER_PASSWORD_DEV=PRneAP0819b
# DEFAULT_FROM_EMAIL_DEV unused in settings; kept for reference
DEFAULT_FROM_EMAIL_DEV=Test.django@vitkovice.cz
# ------------------ AWS (disabled unless USE_AWS=True) ------------------
USE_AWS=False
AWS_ACCESS_KEY_ID=
AWS_SECRET_ACCESS_KEY=
AWS_STORAGE_BUCKET_NAME=
AWS_S3_REGION_NAME=eu-central-1
# ------------------ JWT / TOKENS (lifetimes defined in code) ------------------
# (No env vars needed; kept placeholder section)
# ------------------ MISC ------------------
# FRONTEND_URL_DEV not used; rely on FRONTEND_URL
# Add any extra custom vars below

91
backend/requirements.txt Normal file
View File

@@ -0,0 +1,91 @@
# -- BASE --
requests
pip
python-dotenv # .env support
virtualenv #venv
Django
numpy # NumPy je knihovna programovacího jazyka Python, která poskytuje infrastrukturu pro práci s vektory, maticemi a obecně vícerozměrnými poli.
# -- DATABASE --
sqlparse #non-validating SQL parser for Python. It provides support for parsing, splitting and formatting SQL statements.
tzdata #timezone
psycopg[binary] #PostgreSQL database adapter for the Python
django-filter
django-constance #allows you to store and manage settings of page in the Django admin interface!!!!
# -- OBJECT STORAGE --
Pillow #adds image processing capabilities to your Python interpreter
whitenoise #pomáha se spuštěním serveru a načítaní static files
django-cleanup #odstraní zbytečné media soubory které nejsou v databázi/modelu
django-storages # potřeba k S3 bucket storage
boto3
# -- PROTOCOLS (asgi, websockets) --
redis
channels_redis
channels #django channels
#channels requried package
uvicorn[standard]
daphne
gunicorn
# -- REST API --
djangorestframework #REST Framework
djangorestframework-api-key #API key
djangorestframework-simplejwt #JWT authentication for Django REST Framework
PyJWT #JSON Web Token implementation in Python
asgiref #ASGI reference implementation, to be used with Django Channels
pytz
# pytz brings the Olson tz database into Python and allows
# accurate and cross platform timezone calculations.
# It also solves the issue of ambiguous times at the end of daylight saving time.
#documentation for frontend dev
drf-spectacular
# -- APPS --
django-tinymce
django-cors-headers #csfr
celery #slouží k vytvaření asynchoních úkolu (třeba každou hodinu vyčistit cache atd.)
django-celery-beat #slouží k plánování úkolů pro Celery
# -- EDITING photos, gifs, videos --
#aiofiles
#opencv-python #moviepy use this better instead of pillow
#moviepy
yt-dlp
weasyprint #tvoření PDFek z html dokumentu + css styly
## -- MISCELLANEOUS --
faker #generates fake data for testing purposes
## -- api --
stripe
gopay

View File

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

View File

@@ -0,0 +1,8 @@
from django.apps import AppConfig
class ChatConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'social.chat'
label = "chat"

View File

@@ -0,0 +1,27 @@
# chat/consumers.py
import json
from account.models import UserProfile
from channels.db import database_sync_to_async
from channels.generic.websocket import AsyncWebsocketConsumer
class ChatConsumer(AsyncWebsocketConsumer):
async def connect(self):
await self.accept()
async def disconnect(self, close_code):
pass
async def receive(self, text_data):
text_data_json = json.loads(text_data)
message = text_data_json["message"]
await self.send(text_data=json.dumps({"message": message}))
@database_sync_to_async
def get_user_profile(user_id):
return UserProfile.objects.get(pk=user_id)

View File

@@ -0,0 +1,3 @@
from django.db import models
# Create your models here.

View File

@@ -0,0 +1,8 @@
# chat/routing.py
from django.urls import re_path
from . import consumers
websocket_urlpatterns = [
re_path(r"ws/chat/(?P<room_name>\w+)/$", consumers.ChatConsumer.as_asgi()),
]

View File

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

View File

@@ -0,0 +1,3 @@
from django.shortcuts import render
# Create your views here.

10
backend/thirdparty/downloader/admin.py vendored Normal file
View File

@@ -0,0 +1,10 @@
from django.contrib import admin
from .models import DownloaderRecord
@admin.register(DownloaderRecord)
class DownloaderRecordAdmin(admin.ModelAdmin):
list_display = ("id", "url", "format", "length_of_media", "file_size", "download_time")
list_filter = ("format",)
search_fields = ("url",)
ordering = ("-download_time",)
readonly_fields = ("download_time",)

10
backend/thirdparty/downloader/apps.py vendored Normal file
View File

@@ -0,0 +1,10 @@
from django.apps import AppConfig
class DownloaderConfig(AppConfig):
# Ensure stable default primary key type
default_auto_field = "django.db.models.BigAutoField"
# Must be the full dotted path of this app
name = "thirdparty.downloader"
# Keep a short, stable label (used in migrations/admin)
label = "downloader"

View File

@@ -0,0 +1,30 @@
# Generated by Django 5.2.7 on 2025-10-29 14:53
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
]
operations = [
migrations.CreateModel(
name='DownloaderRecord',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('is_deleted', models.BooleanField(default=False)),
('deleted_at', models.DateTimeField(blank=True, null=True)),
('url', models.URLField()),
('download_time', models.DateTimeField(auto_now_add=True)),
('format', models.CharField(max_length=50)),
('length_of_media', models.IntegerField(help_text='Length of media in seconds')),
('file_size', models.BigIntegerField(help_text='File size in bytes')),
],
options={
'abstract': False,
},
),
]

15
backend/thirdparty/downloader/models.py vendored Normal file
View File

@@ -0,0 +1,15 @@
from django.db import models
from django.conf import settings
from vontor_cz.models import SoftDeleteModel
# 7áznamy pro donwloader, co lidé nejvíc stahujou a v jakém formátu
class DownloaderRecord(SoftDeleteModel):
url = models.URLField()
download_time = models.DateTimeField(auto_now_add=True)
format = models.CharField(max_length=50)
length_of_media = models.IntegerField(help_text="Length of media in seconds")
file_size = models.BigIntegerField(help_text="File size in bytes")

View File

@@ -0,0 +1,9 @@
from rest_framework import serializers
class DownloaderStatsSerializer(serializers.Serializer):
total_downloads = serializers.IntegerField()
avg_length_of_media = serializers.FloatField(allow_null=True)
avg_file_size = serializers.FloatField(allow_null=True)
total_length_of_media = serializers.IntegerField(allow_null=True)
total_file_size = serializers.IntegerField(allow_null=True)
most_common_format = serializers.CharField(allow_null=True)

View File

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

9
backend/thirdparty/downloader/urls.py vendored Normal file
View File

@@ -0,0 +1,9 @@
from django.urls import path
from .views import Downloader, DownloaderStats
urlpatterns = [
# Probe formats for a URL (size-checked)
path("download/", Downloader.as_view(), name="downloader-download"),
path("stats/", DownloaderStats.as_view(), name="downloader-stats"),
]

305
backend/thirdparty/downloader/views.py vendored Normal file
View File

@@ -0,0 +1,305 @@
# ---------------------- Inline serializers for documentation only ----------------------
# Using inline_serializer to avoid creating new files.
import yt_dlp
import tempfile
import os
import shutil
from rest_framework import serializers
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated, AllowAny
from drf_spectacular.utils import extend_schema, inline_serializer
from drf_spectacular.types import OpenApiTypes
from django.conf import settings
from django.http import StreamingHttpResponse
from django.utils.text import slugify
# NEW: aggregations and timeseries helpers
from django.db import models
from django.utils import timezone
from django.db.models.functions import TruncDay, TruncHour
from .models import DownloaderRecord
# Allowed container formats for output/remux
FORMAT_CHOICES = ("mp4", "mkv", "webm", "flv", "mov", "avi", "ogg")
FORMAT_HELP = (
"Choose container format: "
"mp4 (H.264 + AAC, most compatible), "
"mkv (flexible, lossless container), "
"webm (VP9/AV1 + Opus), "
"flv (legacy), mov (Apple-friendly), "
"avi (older), ogg (mostly obsolete)."
)
# Minimal mime map by extension
MIME_BY_EXT = {
"mp4": "video/mp4",
"mkv": "video/x-matroska",
"webm": "video/webm",
"flv": "video/x-flv",
"mov": "video/quicktime",
"avi": "video/x-msvideo",
"ogg": "video/ogg",
}
class Downloader(APIView):
permission_classes = [AllowAny]
authentication_classes = []
@extend_schema(
tags=["downloader"],
summary="Get video info from URL",
parameters=[
inline_serializer(
name="VideoInfoParams",
fields={
"url": serializers.URLField(help_text="Video URL to analyze"),
},
)
],
responses={
200: inline_serializer(
name="VideoInfoResponse",
fields={
"title": serializers.CharField(),
"duration": serializers.IntegerField(allow_null=True),
"thumbnail": serializers.URLField(allow_null=True),
"video_resolutions": serializers.ListField(child=serializers.CharField()),
"audio_resolutions": serializers.ListField(child=serializers.CharField()),
},
),
400: inline_serializer(
name="ErrorResponse",
fields={"error": serializers.CharField()},
),
},
)
def get(self, request):
url = request.data.get("url") or request.query_params.get("url")
if not url:
return Response({"error": "URL is required"}, status=400)
ydl_options = {
"quiet": True,
}
try:
with yt_dlp.YoutubeDL(ydl_options) as ydl:
info = ydl.extract_info(url, download=False)
except Exception:
return Response({"error": "Failed to retrieve video info"}, status=400)
formats = info.get("formats", []) or []
# Video: collect unique heights and sort desc
heights = {
int(f.get("height"))
for f in formats
if f.get("vcodec") != "none" and isinstance(f.get("height"), int)
}
video_resolutions = [f"{h}p" for h in sorted(heights, reverse=True)]
# Audio: collect unique bitrates (abr kbps), fallback to tbr when abr missing
bitrates = set()
for f in formats:
if f.get("acodec") != "none" and f.get("vcodec") == "none":
abr = f.get("abr")
tbr = f.get("tbr")
val = None
if isinstance(abr, (int, float)):
val = int(abr)
elif isinstance(tbr, (int, float)):
val = int(tbr)
if val and val > 0:
bitrates.add(val)
audio_resolutions = [f"{b}kbps" for b in sorted(bitrates, reverse=True)]
return Response(
{
"title": info.get("title"),
"duration": info.get("duration"),
"thumbnail": info.get("thumbnail"),
"video_resolutions": video_resolutions,
"audio_resolutions": audio_resolutions,
},
status=200,
)
@extend_schema(
tags=["downloader"],
summary="Download video from URL",
request=inline_serializer(
name="DownloadRequest",
fields={
"url": serializers.URLField(help_text="Video URL to download"),
"ext": serializers.ChoiceField(
choices=FORMAT_CHOICES,
required=False,
default="mp4",
help_text=FORMAT_HELP,
),
"format": serializers.ChoiceField(
choices=FORMAT_CHOICES,
required=False,
help_text="Alias of 'ext' (deprecated)."
),
"video_quality": serializers.IntegerField(
required=True,
help_text="Target max video height (e.g. 1080)."
),
"audio_quality": serializers.IntegerField(
required=True,
help_text="Target max audio bitrate in kbps (e.g. 160)."
),
},
),
responses={
200: OpenApiTypes.BINARY,
400: inline_serializer(
name="DownloadErrorResponse",
fields={
"error": serializers.CharField(),
"allowed": serializers.ListField(child=serializers.CharField(), required=False),
},
),
},
)
def post(self, request):
url = request.data.get("url")
# Accept ext or legacy format param
ext = (request.data.get("ext") or request.data.get("format") or "mp4").lower()
try:
video_quality = int(request.data.get("video_quality")) # height, e.g., 1080
audio_quality = int(request.data.get("audio_quality")) # abr kbps, e.g., 160
except Exception:
return Response({"error": "Invalid quality parameters, not integers!"}, status=400)
if not url:
return Response({"error": "URL is required"}, status=400)
if ext not in FORMAT_CHOICES:
return Response({"error": f"Unsupported extension '{ext}'", "allowed": FORMAT_CHOICES}, status=400)
# Ensure base tmp dir exists
os.makedirs(settings.DOWNLOADER_TMP_DIR, exist_ok=True)
tmpdir = tempfile.mkdtemp(prefix="downloader_", dir=settings.DOWNLOADER_TMP_DIR)
outtmpl = os.path.join(tmpdir, "download.%(ext)s")
# Build a format selector using requested quality caps
# Example: "bv[height<=1080]+ba[abr<=160]/b"
video_part = f"bv[height<={video_quality}]" if video_quality else "bv*"
audio_part = f"ba[abr<={audio_quality}]" if audio_quality else "ba"
format_selector = f"{video_part}+{audio_part}/b"
ydl_options = {
"format": format_selector, # select by requested quality
"merge_output_format": ext, # container
"outtmpl": outtmpl, # temp dir
"quiet": True,
"max_filesize": settings.DOWNLOADER_MAX_SIZE_BYTES,
"socket_timeout": settings.DOWNLOADER_TIMEOUT,
# remux to container without re-encoding where possible
"postprocessors": [
{"key": "FFmpegVideoRemuxer", "preferedformat": ext}
],
}
file_path = ""
try:
with yt_dlp.YoutubeDL(ydl_options) as ydl:
info = ydl.extract_info(url, download=True)
base = ydl.prepare_filename(info)
file_path = base if base.endswith(f".{ext}") else os.path.splitext(base)[0] + f".{ext}"
# Stats before streaming
duration = int((info or {}).get("duration") or 0)
size = os.path.getsize(file_path) if os.path.exists(file_path) else 0
DownloaderRecord.objects.create(
url=url,
format=ext,
length_of_media=duration,
file_size=size,
)
# Streaming generator that deletes file & temp dir after send (or on abort)
def stream_and_cleanup(path: str, temp_dir: str, chunk_size: int = 8192):
try:
with open(path, "rb") as f:
while True:
chunk = f.read(chunk_size)
if not chunk:
break
yield chunk
finally:
try:
if os.path.exists(path):
os.remove(path)
finally:
shutil.rmtree(temp_dir, ignore_errors=True)
safe_title = slugify(info.get("title") or "video")
filename = f"{safe_title}.{ext}"
content_type = MIME_BY_EXT.get(ext, "application/octet-stream")
response = StreamingHttpResponse(
streaming_content=stream_and_cleanup(file_path, tmpdir),
content_type=content_type,
)
if size:
response["Content-Length"] = str(size)
response["Content-Disposition"] = f'attachment; filename="{filename}"'
return response
except Exception as e:
shutil.rmtree(tmpdir, ignore_errors=True)
return Response({"error": str(e)}, status=400)
# ---------------- STATS FOR GRAPHS ----------------
from .serializers import DownloaderStatsSerializer
from django.db.models import Count, Avg, Sum
class DownloaderStats(APIView):
"""
Vrací agregované statistiky z tabulky DownloaderRecord.
"""
authentication_classes = []
permission_classes = [AllowAny]
@extend_schema(
tags=["downloader"],
summary="Get aggregated downloader statistics",
responses={200: DownloaderStatsSerializer},
)
def get(self, request):
# agregace číselných polí
agg = DownloaderRecord.objects.aggregate(
total_downloads=Count("id"),
avg_length_of_media=Avg("length_of_media"),
avg_file_size=Avg("file_size"),
total_length_of_media=Sum("length_of_media"),
total_file_size=Sum("file_size"),
)
# zjištění nejčastějšího formátu
most_common = (
DownloaderRecord.objects.values("format")
.annotate(count=Count("id"))
.order_by("-count")
.first()
)
agg["most_common_format"] = most_common["format"] if most_common else None
serializer = DownloaderStatsSerializer(agg)
return Response(serializer.data)

3
backend/thirdparty/gopay/admin.py vendored Normal file
View File

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

View File

@@ -1,6 +1,6 @@
from django.apps import AppConfig
class ApiConfig(AppConfig):
class GopayConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'api'
name = 'gopay'

3
backend/thirdparty/gopay/models.py vendored Normal file
View File

@@ -0,0 +1,3 @@
from django.db import models
# Create your models here.

37
backend/thirdparty/gopay/serializers.py vendored Normal file
View File

@@ -0,0 +1,37 @@
from rest_framework import serializers
class GoPayCreatePaymentRequestSerializer(serializers.Serializer):
amount = serializers.DecimalField(max_digits=12, decimal_places=2, min_value=0.01)
currency = serializers.CharField(required=False, default="CZK")
order_number = serializers.CharField(required=False, allow_blank=True, default="order-001")
order_description = serializers.CharField(required=False, allow_blank=True, default="Example GoPay payment")
return_url = serializers.URLField(required=False)
notify_url = serializers.URLField(required=False)
preauthorize = serializers.BooleanField(required=False, default=False)
class GoPayPaymentCreatedResponseSerializer(serializers.Serializer):
id = serializers.IntegerField()
state = serializers.CharField()
gw_url = serializers.URLField(required=False, allow_null=True)
class GoPayStatusResponseSerializer(serializers.Serializer):
id = serializers.IntegerField()
state = serializers.CharField()
class GoPayRefundRequestSerializer(serializers.Serializer):
amount = serializers.DecimalField(max_digits=12, decimal_places=2, required=False, min_value=0.01)
class GoPayCaptureRequestSerializer(serializers.Serializer):
amount = serializers.DecimalField(max_digits=12, decimal_places=2, required=False, min_value=0.01)
class GoPayCreateRecurrenceRequestSerializer(serializers.Serializer):
amount = serializers.DecimalField(max_digits=12, decimal_places=2, min_value=0.01)
currency = serializers.CharField(required=False, default="CZK")
order_number = serializers.CharField(required=False, allow_blank=True, default="recur-001")
order_description = serializers.CharField(required=False, allow_blank=True, default="Recurring payment")

3
backend/thirdparty/gopay/tests.py vendored Normal file
View File

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

20
backend/thirdparty/gopay/urls.py vendored Normal file
View File

@@ -0,0 +1,20 @@
from django.urls import path
from .views import (
GoPayPaymentView,
GoPayPaymentStatusView,
GoPayRefundPaymentView,
GoPayCaptureAuthorizationView,
GoPayVoidAuthorizationView,
GoPayCreateRecurrenceView,
GoPayPaymentInstrumentsView,
)
urlpatterns = [
path('payment/', GoPayPaymentView.as_view(), name='gopay-payment'),
path('payment/<int:payment_id>/status/', GoPayPaymentStatusView.as_view(), name='gopay-payment-status'),
path('payment/<int:payment_id>/refund/', GoPayRefundPaymentView.as_view(), name='gopay-refund-payment'),
path('payment/<int:payment_id>/capture/', GoPayCaptureAuthorizationView.as_view(), name='gopay-capture-authorization'),
path('payment/<int:payment_id>/void/', GoPayVoidAuthorizationView.as_view(), name='gopay-void-authorization'),
path('payment/<int:payment_id>/recurrence/', GoPayCreateRecurrenceView.as_view(), name='gopay-create-recurrence'),
path('payment-instruments/', GoPayPaymentInstrumentsView.as_view(), name='gopay-payment-instruments'),
]

233
backend/thirdparty/gopay/views.py vendored Normal file
View File

@@ -0,0 +1,233 @@
from django.shortcuts import render
# Create your views here.
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
import gopay
from gopay.enums import TokenScope, Language
import os
from drf_spectacular.utils import extend_schema, OpenApiResponse, OpenApiExample, OpenApiParameter
from .serializers import (
GoPayCreatePaymentRequestSerializer,
GoPayPaymentCreatedResponseSerializer,
GoPayStatusResponseSerializer,
GoPayRefundRequestSerializer,
GoPayCaptureRequestSerializer,
GoPayCreateRecurrenceRequestSerializer,
)
class GoPayClientMixin:
"""Shared helpers for configuring GoPay client and formatting responses."""
def get_gopay_client(self):
gateway_url = os.getenv("GOPAY_GATEWAY_URL", "https://gw.sandbox.gopay.com/api")
return gopay.payments({
"goid": os.getenv("GOPAY_GOID"),
"client_id": os.getenv("GOPAY_CLIENT_ID"),
"client_secret": os.getenv("GOPAY_CLIENT_SECRET"),
"gateway_url": gateway_url,
"scope": TokenScope.ALL,
"language": Language.CZECH,
})
def _to_response(self, sdk_response):
# The GoPay SDK returns a response object with has_succeed(), json, errors, status_code
try:
if hasattr(sdk_response, "has_succeed") and sdk_response.has_succeed():
return Response(getattr(sdk_response, "json", {}))
status = getattr(sdk_response, "status_code", 400)
errors = getattr(sdk_response, "errors", None)
if errors is None and hasattr(sdk_response, "json"):
errors = sdk_response.json
if errors is None:
errors = {"detail": "GoPay request failed"}
return Response({"errors": errors}, status=status)
except Exception as e:
return Response({"errors": str(e)}, status=500)
class GoPayPaymentView(GoPayClientMixin, APIView):
permission_classes = [IsAuthenticated]
@extend_schema(
tags=["GoPay"],
summary="Create GoPay payment",
description="Creates a GoPay payment and returns gateway URL and payment info.",
request=GoPayCreatePaymentRequestSerializer,
responses={
200: OpenApiResponse(response=GoPayPaymentCreatedResponseSerializer, description="Payment created"),
400: OpenApiResponse(description="Validation error or SDK error"),
},
examples=[
OpenApiExample(
"Create payment",
value={
"amount": 123.45,
"currency": "CZK",
"order_number": "order-001",
"order_description": "Example GoPay payment",
"return_url": "https://yourfrontend.com/success",
"notify_url": "https://yourbackend.com/gopay/notify",
"preauthorize": False,
},
request_only=True,
)
]
)
def post(self, request):
amount = request.data.get("amount")
currency = request.data.get("currency", "CZK")
order_number = request.data.get("order_number", "order-001")
order_description = request.data.get("order_description", "Example GoPay payment")
return_url = request.data.get("return_url", "https://yourfrontend.com/success")
notify_url = request.data.get("notify_url", "https://yourbackend.com/gopay/notify")
preauthorize = bool(request.data.get("preauthorize", False))
if not amount:
return Response({"error": "Amount is required"}, status=400)
payments = self.get_gopay_client()
payment_data = {
"payer": {
"allowed_payment_instruments": ["PAYMENT_CARD"],
"default_payment_instrument": "PAYMENT_CARD",
"allowed_swifts": ["FIOB"],
"contact": {
"first_name": getattr(request.user, "first_name", ""),
"last_name": getattr(request.user, "last_name", ""),
"email": getattr(request.user, "email", ""),
},
},
"amount": int(float(amount) * 100), # GoPay expects amount in cents
"currency": currency,
"order_number": order_number,
"order_description": order_description,
"items": [
{"name": "Example Item", "amount": int(float(amount) * 100)}
],
"callback": {"return_url": return_url, "notify_url": notify_url},
"preauthorize": preauthorize,
}
resp = payments.create_payment(payment_data)
return self._to_response(resp)
class GoPayPaymentStatusView(GoPayClientMixin, APIView):
permission_classes = [IsAuthenticated]
@extend_schema(
tags=["GoPay"],
summary="Get GoPay payment status",
parameters=[OpenApiParameter(name="payment_id", required=True, type=int, location=OpenApiParameter.PATH)],
responses={200: OpenApiResponse(response=GoPayStatusResponseSerializer, description="Payment status")},
)
def get(self, request, payment_id: int):
payments = self.get_gopay_client()
resp = payments.get_status(payment_id)
return self._to_response(resp)
class GoPayRefundPaymentView(GoPayClientMixin, APIView):
permission_classes = [IsAuthenticated]
@extend_schema(
tags=["GoPay"],
summary="Refund GoPay payment",
parameters=[OpenApiParameter(name="payment_id", required=True, type=int, location=OpenApiParameter.PATH)],
request=GoPayRefundRequestSerializer,
responses={200: OpenApiResponse(description="Refund processed")},
)
def post(self, request, payment_id: int):
amount = request.data.get("amount") # optional for full refund
payments = self.get_gopay_client()
if amount is None or amount == "":
# Full refund
resp = payments.refund_payment(payment_id)
else:
resp = payments.refund_payment(payment_id, int(float(amount) * 100))
return self._to_response(resp)
class GoPayCaptureAuthorizationView(GoPayClientMixin, APIView):
permission_classes = [IsAuthenticated]
@extend_schema(
tags=["GoPay"],
summary="Capture GoPay authorization",
parameters=[OpenApiParameter(name="payment_id", required=True, type=int, location=OpenApiParameter.PATH)],
request=GoPayCaptureRequestSerializer,
responses={200: OpenApiResponse(description="Capture processed")},
)
def post(self, request, payment_id: int):
amount = request.data.get("amount") # optional for partial capture
payments = self.get_gopay_client()
if amount is None or amount == "":
resp = payments.capture_authorization(payment_id)
else:
resp = payments.capture_authorization(payment_id, int(float(amount) * 100))
return self._to_response(resp)
class GoPayVoidAuthorizationView(GoPayClientMixin, APIView):
permission_classes = [IsAuthenticated]
@extend_schema(
tags=["GoPay"],
summary="Void GoPay authorization",
parameters=[OpenApiParameter(name="payment_id", required=True, type=int, location=OpenApiParameter.PATH)],
responses={200: OpenApiResponse(description="Authorization voided")},
)
def post(self, request, payment_id: int):
payments = self.get_gopay_client()
resp = payments.void_authorization(payment_id)
return self._to_response(resp)
class GoPayCreateRecurrenceView(GoPayClientMixin, APIView):
permission_classes = [IsAuthenticated]
@extend_schema(
tags=["GoPay"],
summary="Create GoPay recurrence",
parameters=[OpenApiParameter(name="payment_id", required=True, type=int, location=OpenApiParameter.PATH)],
request=GoPayCreateRecurrenceRequestSerializer,
responses={200: OpenApiResponse(description="Recurrence created")},
)
def post(self, request, payment_id: int):
amount = request.data.get("amount")
currency = request.data.get("currency", "CZK")
order_number = request.data.get("order_number", "recur-001")
order_description = request.data.get("order_description", "Recurring payment")
if not amount:
return Response({"error": "Amount is required"}, status=400)
payments = self.get_gopay_client()
recurrence_payload = {
"amount": int(float(amount) * 100),
"currency": currency,
"order_number": order_number,
"order_description": order_description,
}
resp = payments.create_recurrence(payment_id, recurrence_payload)
return self._to_response(resp)
class GoPayPaymentInstrumentsView(GoPayClientMixin, APIView):
permission_classes = [IsAuthenticated]
@extend_schema(
tags=["GoPay"],
summary="Get GoPay payment instruments",
parameters=[OpenApiParameter(name="currency", required=False, type=str, location=OpenApiParameter.QUERY)],
responses={200: OpenApiResponse(description="Available payment instruments returned")},
)
def get(self, request):
currency = request.query_params.get("currency", "CZK")
goid = os.getenv("GOPAY_GOID")
if not goid:
return Response({"error": "GOPAY_GOID is not configured"}, status=500)
payments = self.get_gopay_client()
resp = payments.get_payment_instruments(goid, currency)
return self._to_response(resp)

23
backend/thirdparty/stripe/admin.py vendored Normal file
View File

@@ -0,0 +1,23 @@
from django.contrib import admin
from .models import Order
@admin.register(Order)
class OrderAdmin(admin.ModelAdmin):
list_display = ("id", "amount", "currency", "status", "created_at")
list_filter = ("status", "currency", "created_at")
search_fields = ("id", "stripe_session_id", "stripe_payment_intent")
readonly_fields = ("created_at", "stripe_session_id", "stripe_payment_intent")
fieldsets = (
(None, {
"fields": ("amount", "currency", "status")
}),
("Stripe info", {
"fields": ("stripe_session_id", "stripe_payment_intent"),
"classes": ("collapse",),
}),
("Metadata", {
"fields": ("created_at",),
}),
)
ordering = ("-created_at",)

7
backend/thirdparty/stripe/apps.py vendored Normal file
View File

@@ -0,0 +1,7 @@
from django.apps import AppConfig
class StripeConfig(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'thirdparty.stripe'
label = "stripe"

View File

@@ -0,0 +1,26 @@
# Generated by Django 5.2.7 on 2025-10-28 22:28
from django.db import migrations, models
class Migration(migrations.Migration):
initial = True
dependencies = [
]
operations = [
migrations.CreateModel(
name='Order',
fields=[
('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
('amount', models.DecimalField(decimal_places=2, max_digits=10)),
('currency', models.CharField(default='czk', max_length=10)),
('status', models.CharField(choices=[('pending', 'Pending'), ('paid', 'Paid'), ('failed', 'Failed'), ('cancelled', 'Cancelled')], default='pending', max_length=20)),
('stripe_session_id', models.CharField(blank=True, max_length=255, null=True)),
('stripe_payment_intent', models.CharField(blank=True, max_length=255, null=True)),
('created_at', models.DateTimeField(auto_now_add=True)),
],
),
]

21
backend/thirdparty/stripe/models.py vendored Normal file
View File

@@ -0,0 +1,21 @@
from django.db import models
# Create your models here.
class Order(models.Model):
STATUS_CHOICES = [
("pending", "Pending"),
("paid", "Paid"),
("failed", "Failed"),
("cancelled", "Cancelled"),
]
amount = models.DecimalField(max_digits=10, decimal_places=2)
currency = models.CharField(max_length=10, default="czk")
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default="pending")
stripe_session_id = models.CharField(max_length=255, blank=True, null=True)
stripe_payment_intent = models.CharField(max_length=255, blank=True, null=True)
created_at = models.DateTimeField(auto_now_add=True)
def __str__(self):
return f"Order {self.id} - {self.status}"

View File

@@ -0,0 +1,29 @@
from rest_framework import serializers
from .models import Order
class OrderSerializer(serializers.ModelSerializer):
# Nested read-only representations
# product = ProductSerializer(read_only=True)
# carrier = CarrierSerializer(read_only=True)
# Write-only foreign keys
# product_id = serializers.PrimaryKeyRelatedField(
# queryset=Product.objects.all(), source="product", write_only=True
# )
# carrier_id = serializers.PrimaryKeyRelatedField(
# queryset=Carrier.objects.all(), source="carrier", write_only=True
# )
class Meta:
model = Order
fields = [
"id",
"amount",
"currency",
"status",
"stripe_session_id",
"stripe_payment_intent",
"created_at",
]
read_only_fields = ("created_at",)

3
backend/thirdparty/stripe/tests.py vendored Normal file
View File

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

6
backend/thirdparty/stripe/urls.py vendored Normal file
View File

@@ -0,0 +1,6 @@
from django.urls import path
from .views import CreateCheckoutSessionView
urlpatterns = [
path("orders/create-checkout/", CreateCheckoutSessionView.as_view(), name="create-checkout"),
]

78
backend/thirdparty/stripe/views.py vendored Normal file
View File

@@ -0,0 +1,78 @@
from django.views.decorators.csrf import csrf_exempt
from django.conf import settings
from django.http import HttpResponse
from rest_framework import generics
from rest_framework.response import Response
from rest_framework.views import APIView
from drf_spectacular.utils import extend_schema
from .models import Order
from .serializers import OrderSerializer
import os
import stripe
stripe.api_key = os.getenv("STRIPE_SECRET_KEY")
class CreateCheckoutSessionView(APIView):
@extend_schema(
tags=["stripe"],
)
def post(self, request):
serializer = OrderSerializer(data=request.data) #obecný serializer
serializer.is_valid(raise_exception=True)
order = Order.objects.create(
amount=serializer.validated_data["amount"],
currency=serializer.validated_data.get("currency", "czk"),
)
# Vytvoření Stripe Checkout Session
session = stripe.checkout.Session.create(
payment_method_types=["card"],
line_items=[{
"price_data": {
"currency": order.currency,
"product_data": {"name": f"Order {order.id}"},
"unit_amount": int(order.amount * 100), # v centech
},
"quantity": 1,
}],
mode="payment",
success_url=request.build_absolute_uri(f"/payment/success/{order.id}"),
cancel_url=request.build_absolute_uri(f"/payment/cancel/{order.id}"),
)
order.stripe_session_id = session.id
order.stripe_payment_intent = session.payment_intent
order.save()
data = OrderSerializer(order).data
data["checkout_url"] = session.url
return Response(data)
@csrf_exempt
def stripe_webhook(request):
payload = request.body
sig_header = request.META.get("HTTP_STRIPE_SIGNATURE")
event = None
try:
event = stripe.Webhook.construct_event(
payload, sig_header, settings.STRIPE_WEBHOOK_SECRET
)
except stripe.error.SignatureVerificationError:
return HttpResponse(status=400)
if event["type"] == "checkout.session.completed":
session = event["data"]["object"]
order = Order.objects.filter(stripe_session_id=session.get("id")).first()
if order:
order.status = "paid"
order.save()
return HttpResponse(status=200)

View File

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

7
backend/thirdparty/trading212/apps.py vendored Normal file
View File

@@ -0,0 +1,7 @@
from django.apps import AppConfig
class Trading212Config(AppConfig):
default_auto_field = 'django.db.models.BigAutoField'
name = 'thirdparty.trading212'
label = "trading212"

View File

@@ -0,0 +1,3 @@
from django.db import models
# Create your models here.

View File

@@ -0,0 +1,11 @@
# thirdparty/trading212/serializers.py
from rest_framework import serializers
class Trading212AccountCashSerializer(serializers.Serializer):
blocked = serializers.FloatField()
free = serializers.FloatField()
invested = serializers.FloatField()
pieCash = serializers.FloatField()
ppl = serializers.FloatField()
result = serializers.FloatField()
total = serializers.FloatField()

View File

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

6
backend/thirdparty/trading212/urls.py vendored Normal file
View File

@@ -0,0 +1,6 @@
from django.urls import path
from .views import Trading212AccountCashView # Replace with actual view class
urlpatterns = [
path('your-endpoint/', Trading212AccountCashView.as_view(), name='trading212-endpoint'),
]

38
backend/thirdparty/trading212/views.py vendored Normal file
View File

@@ -0,0 +1,38 @@
# thirdparty/trading212/views.py
import os
import requests
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
from .serializers import Trading212AccountCashSerializer
from drf_spectacular.utils import extend_schema
class Trading212AccountCashView(APIView):
permission_classes = [IsAuthenticated]
@extend_schema(
tags=["trading212"],
summary="Get Trading212 account cash",
responses=Trading212AccountCashSerializer
)
def get(self, request):
api_key = os.getenv("API_KEY_TRADING212")
headers = {
"Authorization": f"Bearer {api_key}",
"Accept": "application/json",
}
url = "https://api.trading212.com/api/v0/equity/account/cash"
try:
resp = requests.get(url, headers=headers, timeout=10)
resp.raise_for_status()
except requests.RequestException as exc:
return Response({"error": str(exc)}, status=400)
data = resp.json()
serializer = Trading212AccountCashSerializer(data=data)
serializer.is_valid(raise_exception=True)
return Response(serializer.data)

View File

@@ -0,0 +1,3 @@
from .celery import app as celery_app
__all__ = ["celery_app"]

View File

@@ -4,26 +4,22 @@ ASGI config for vontor_cz project.
It exposes the ASGI callable as a module-level variable named ``application``.
For more information on this file, see
https://docs.djangoproject.com/en/5.1/howto/deployment/asgi/
https://docs.djangoproject.com/en/5.2/howto/deployment/asgi/
"""
import os
import django
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
from channels.security.websocket import AllowedHostsOriginValidator
from channels.auth import AuthMiddlewareStack
#import myapp.routing # your app's routing
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'vontor_cz.settings')
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'trznice.settings')
application = ProtocolTypeRouter({
"http": get_asgi_application(),
"websocket": AllowedHostsOriginValidator(
AuthMiddlewareStack(
URLRouter(__import__('home.routing').routing.websocket_urlpatterns)
)
),
})
# "websocket": AuthMiddlewareStack(
# URLRouter(
# #myapp.routing.websocket_urlpatterns
# )
# ),
})

View File

@@ -0,0 +1,8 @@
import os
from celery import Celery
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "backend.settings")
app = Celery("backend")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()

View File

@@ -0,0 +1,61 @@
from django.db import models
from django.utils import timezone
class ActiveManager(models.Manager):
def get_queryset(self):
return super().get_queryset().filter(is_deleted=False)
class AllManager(models.Manager):
def get_queryset(self):
return super().get_queryset()
# How to use custom object Managers: add these fields to your model, to override objects behaviour and all_objects behaviour
# objects = ActiveManager()
# all_objects = AllManager()
class SoftDeleteModel(models.Model):
is_deleted = models.BooleanField(default=False)
deleted_at = models.DateTimeField(null=True, blank=True)
def delete(self, using=None, keep_parents=False):
self.is_deleted = True
self.deleted_at = timezone.now()
self.save()
objects = ActiveManager()
all_objects = AllManager()
class Meta:
abstract = True
def delete(self, *args, **kwargs):
# Soft delete self
self.is_deleted = True
self.deleted_at = timezone.now()
self.save()
def hard_delete(self, using=None, keep_parents=False):
super().delete(using=using, keep_parents=keep_parents)
# SiteSettings model for managing site-wide settings
"""class SiteSettings(models.Model):
bank = models.CharField(max_length=100, blank=True)
support_email = models.EmailField(blank=True)
logo = models.ImageField(upload_to='settings/', blank=True, null=True)
def __str__(self):
return "Site Settings"
class Meta:
verbose_name = "Site Settings"
verbose_name_plural = "Site Settings"
@classmethod
def get_solo(cls):
obj, created = cls.objects.get_or_create(id=1)
return obj
"""

Some files were not shown because too many files have changed in this diff Show More