219 lines
11 KiB
Python
219 lines
11 KiB
Python
from django.shortcuts import render
|
|
from django.http import JsonResponse
|
|
from django.db import connection
|
|
|
|
from django.middleware.csrf import get_token
|
|
from django.core.signing import TimestampSigner
|
|
from django.utils.encoding import force_str, force_bytes
|
|
from django.utils.http import urlsafe_base64_decode, urlsafe_base64_encode
|
|
from django.contrib.auth.hashers import PBKDF2PasswordHasher
|
|
|
|
from django.views.decorators.csrf import csrf_exempt
|
|
import re
|
|
|
|
|
|
def call_func(name: str, parameters: dict):
|
|
with connection.cursor() as cursor:
|
|
if parameters:
|
|
this_params = []
|
|
for param in list(parameters.values()):
|
|
if type(param) == str:
|
|
this_params.append(f"'{param}'")
|
|
elif param is None:
|
|
this_params.append("null")
|
|
else:
|
|
this_params.append(param)
|
|
|
|
cursor.execute(f"SELECT * FROM {name}({', '.join(this_params)})")
|
|
else:
|
|
cursor.execute(f"SELECT * FROM {name}()")
|
|
columns = [col[0] for col in cursor.description]
|
|
return [dict(zip(columns, row)) for row in cursor.fetchall()]
|
|
|
|
def api_get_xcsrf(request):
|
|
data = { 'X-CSFRToken': get_token(request) }
|
|
return JsonResponse(data, safe=False, status=200)
|
|
|
|
def is_auth(request):
|
|
if not request.headers.get('Authorization'):
|
|
return False
|
|
token = request.headers.get('Authorization').split(' ')[-1]
|
|
signer = TimestampSigner(salt='django.core.signing')
|
|
try:
|
|
login = signer.unsign(force_str(urlsafe_base64_decode(token)), max_age=1000)
|
|
print(login)
|
|
except:
|
|
return False
|
|
else:
|
|
user = call_func('nn_db.user_get', {'search_value': login})[0]
|
|
return user
|
|
|
|
def api_applications(request):
|
|
user = is_auth(request)
|
|
if not user:
|
|
return JsonResponse({"code": 403, "message":"Доступ запрещен" }, safe=False, status=403)
|
|
if request.method == 'GET':
|
|
params = {}
|
|
if user['is_admin']:
|
|
params['pk_user'] = None
|
|
else:
|
|
params['pk_user'] = str(user['pk_user'])
|
|
# params['pk_user'] = None
|
|
results = call_func('nn_db.application_get', params)
|
|
|
|
applications = []
|
|
keys = []
|
|
for result in results:
|
|
if result['pk_application'] not in keys:
|
|
application = {}
|
|
application['pk_application'] = result['pk_application']
|
|
application['car_number'] = result['car_number']
|
|
application['description'] = result['description']
|
|
application['fk_status'] = result['fk_status']
|
|
application['title'] = result['title']
|
|
application['fk_user'] = result['fk_user']
|
|
application['username'] = result['username']
|
|
application['images'] = []
|
|
|
|
if result['pk_image'] is not None:
|
|
image = {}
|
|
image['pk_image'] = result['pk_image']
|
|
image['file_name'] = result['file_name']
|
|
image['file_ext'] = result['file_ext']
|
|
application['images'].append(image)
|
|
|
|
applications.append(application)
|
|
keys.append(application['pk_application'])
|
|
else:
|
|
for application in applications:
|
|
if application['pk_application'] == result['pk_application']:
|
|
image = {}
|
|
image['pk_image'] = result['pk_image']
|
|
image['file_name'] = result['file_name']
|
|
image['file_ext'] = result['file_ext']
|
|
application['images'].append(image)
|
|
return JsonResponse({"code": 200, "data": applications}, safe=False, status=200)
|
|
if request.method == 'POST':
|
|
params = {}
|
|
params['car_number'] = request.POST.get('car_number')
|
|
params['description'] = request.POST.get('description')
|
|
params['fk_user'] = user['pk_user']
|
|
confirm = call_func('nn_db.application_post', params)
|
|
return JsonResponse({"code": 201, "data": confirm}, safe=False, status=201)
|
|
|
|
def api_users(request):
|
|
if is_auth(request):
|
|
return JsonResponse({ "code": 403, "message":"Доступ запрещен" }, safe=False, status=403)
|
|
if request.method == 'POST':
|
|
params = []
|
|
|
|
|
|
lastname = request.POST.get('lastname')
|
|
firstname = request.POST.get('firstname')
|
|
middlename = request.POST.get('middlename')
|
|
username = request.POST.get('username')
|
|
password = request.POST.get('password')
|
|
email = request.POST.get('email')
|
|
phone = request.POST.get('phone')
|
|
|
|
errors = []
|
|
|
|
if lastname is None:
|
|
errors.append({ "lastname": "Параметр обязателен для заполнения" })
|
|
if firstname is None:
|
|
errors.append({ "firstname": "Параметр обязателен для заполнения" })
|
|
if username is None:
|
|
errors.append({ "username": "Параметр обязателен для заполнения" })
|
|
if password is None:
|
|
errors.append({ "password": "Параметр обязателен для заполнения" })
|
|
if email is None:
|
|
errors.append({ "email": "Параметр обязателен для заполнения" })
|
|
if phone is None:
|
|
errors.append({ "phone": "Параметр обязателен для заполнения" })
|
|
|
|
if errors:
|
|
return JsonResponse({"code": 409, "errors": errors}, safe=False, status=409)
|
|
|
|
if len(lastname) < 6:
|
|
errors.append({ "lastname": "Фамилия должна быть длинее 6 символов" })
|
|
elif len(lastname) > 60:
|
|
errors.append({ "lastname": "Фамилия должна быть короче 60 символов" })
|
|
if len(firstname) < 6:
|
|
errors.append({ "firstname": "Имя должна быть длинее 6 символов" })
|
|
elif len(firstname) > 60:
|
|
errors.append({ "firstname": "Имя должна быть короче 60 символов" })
|
|
if middlename is not None:
|
|
if len(middlename) != 0:
|
|
if len(middlename) < 6:
|
|
errors.append({ "middlename": "Отчество должна быть длинее 6 символов" })
|
|
elif len(middlename) > 60:
|
|
errors.append({ "middlename": "Отчество должна быть короче 60 символов" })
|
|
|
|
if len(username) < 6:
|
|
errors.append({ "username": "Имя пользователя должно быть длинее 6 символов" })
|
|
elif len(username) > 60:
|
|
errors.append({ "username": "Имя пользователя должно быть короче 60 символов" })
|
|
if len(password) < 8:
|
|
errors.append({ "password": "Пароль должен быть длинее 8 символов" })
|
|
|
|
|
|
if len(email) < 5:
|
|
errors.append({ "email": "Адрес электронной почты должен быть длинее 5 символов" })
|
|
elif re.match(r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$', email) is None:
|
|
errors.append({ "email": "Адрес электронной почты имеет некорректный формат" })
|
|
if len(phone) < 11:
|
|
errors.append({ "phone": "Номер телефона должен быть длинее 10 символов" })
|
|
elif len(phone) > 12:
|
|
errors.append({ "phone": "Номер телефона должен быть короче 12 символов" })
|
|
elif re.match(r'^(?:\+79|89)\d{9}$', phone) is None:
|
|
errors.append({ "phone": "Номер телефона имеет некорректный формат" })
|
|
|
|
if errors:
|
|
return JsonResponse({"code": 409, "errors": errors}, safe=False, status=409)
|
|
|
|
|
|
|
|
user_username = call_func('nn_db.user_get', {"search_value": username})
|
|
user_email = call_func('nn_db.user_get', {"search_value": email})
|
|
if not(user_username) and not(user_email):
|
|
hasher = PBKDF2PasswordHasher()
|
|
password_hash = hasher.encode(password, salt='extra')
|
|
pk_user = call_func("nn_db.user_post", {"lastname": lastname, "firstname": firstname, "middlename": middlename, "username": username, "password_hash": password_hash, "email": email, "phone": phone})[0]['pk_user']
|
|
user = call_func("nn_db.user_get", {"search_value": username})[0]
|
|
return JsonResponse(user, safe=False, status=201)
|
|
else:
|
|
return JsonResponse({"code": 409, "message":"Пользователь c таким именем или адресом электронной почты существует"}, safe=False, status=409)
|
|
|
|
def api_auth(request):
|
|
if is_auth(request):
|
|
return JsonResponse({ "code": 403, "message":"Доступ запрещен" }, safe=False, status=403)
|
|
if request.method == 'POST':
|
|
response = {}
|
|
login = request.POST.get('login')
|
|
password = request.POST.get('password')
|
|
|
|
errors = []
|
|
|
|
if len(login) < 6:
|
|
errors.append({ "login": "Имя пользователя должно быть длинее 6 символов" })
|
|
elif len(login) > 60:
|
|
errors.append({ "login": "Имя пользователя должно быть короче 60 символов" })
|
|
if len(password) < 8:
|
|
errors.append({ "password": "Пароль должен быть длинее 8 символов" })
|
|
|
|
if errors:
|
|
return JsonResponse({"code": 409, "errors": errors}, safe=False, status=409)
|
|
|
|
user = call_func('nn_db.user_get', {'search_value': login})
|
|
if user:
|
|
user = user[0]
|
|
hasher = PBKDF2PasswordHasher()
|
|
if not hasher.verify(password, user['password_hash']):
|
|
return JsonResponse({ "code": 401, "message":"Неверное имя пользователя или пароль" }, safe=False, status=401)
|
|
response = user
|
|
signer = TimestampSigner(salt='django.core.signing')
|
|
token = urlsafe_base64_encode(force_bytes(signer.sign(user['username'])))
|
|
response['token'] = token
|
|
return JsonResponse(response, safe=False, status=200)
|
|
else:
|
|
return JsonResponse({ "code": 401, "message":"Неверное имя пользователя или пароль" }, safe=False, status=401) |