343 lines
20 KiB
Python
343 lines
20 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.http.request import QueryDict
|
||
|
||
from django.views.decorators.csrf import csrf_exempt
|
||
import re, json
|
||
|
||
|
||
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-CSRFToken': 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)
|
||
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 # Заполняем параметр null-значением (в Python - None)
|
||
else:
|
||
params['pk_user'] = str(user['pk_user']) # Иначе заполняем параметр идентификатором пользователя
|
||
|
||
# Запрашиваем данные по заявкам из базы данных по указанному пользователю
|
||
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 = {} # Инициализируем список параметров для запроса и заполняем его данными из POST-запроса
|
||
params['car_number'] = request.POST.get('car_number')
|
||
params['description'] = request.POST.get('description')
|
||
params['fk_user'] = str(user['pk_user'])
|
||
|
||
# Инициализируем список ошибок валидации данных
|
||
errors = []
|
||
# Автоматическая проверка параметров на заполненность
|
||
for key in params:
|
||
if params[key] is None or params[key] == '':
|
||
errors.append({ key: "Параметр обязателен для заполнения" })
|
||
|
||
# Если ошибки есть - то возвращаем ответ с ошибкой
|
||
if errors:
|
||
return JsonResponse({"code": 409, "errors": errors}, safe=False, status=409)
|
||
|
||
# Проверяем введенные данные на валидность
|
||
if re.match(r'^[YETOPAHKXCBMУЕТОРАНКХСВМ]\d{3}[YETOPAHKXCBMУЕТОРАНКХСВМ]{2}\d{1,3}$', params['car_number']) is None:
|
||
errors.append({ "car_number": "Государственный номер автомобиля имеет некорректный формат" })
|
||
if len(params['description']) < 15:
|
||
errors.append({ "description": "Описание нарушения должно быть длинее 5 символов" })
|
||
|
||
# Если ошибки есть - то возвращаем ответ с ошибкой
|
||
if errors:
|
||
return JsonResponse({"code": 409, "errors": errors}, safe=False, status=409)
|
||
|
||
# Отправляем запрос на добавление данных
|
||
confirm = call_func('nn_db.application_post', params)
|
||
|
||
# Проверяем, пришел ли в ответ на запрос первичный ключ добавленной заявки
|
||
if any(confirm) and 'pk_application' in confirm[0]:
|
||
# Если пришёл - то возвращаем данные
|
||
return JsonResponse({"code": 201, "data": confirm}, safe=False, status=201)
|
||
else:
|
||
# Если нет - то возвращаем ошибку с сообщением
|
||
return JsonResponse({"code": 520, "message": "Не удалось добавить запись в БД, попробуйте позднее"}, safe=False, status=520)
|
||
|
||
# Если запрос на изменение заявки:
|
||
if request.method == 'PATCH':
|
||
if not user['is_admin']: # Если пользователь - не администратор
|
||
return JsonResponse({"code": 403, "message":"Доступ запрещен" }, safe=False, status=403) # По ТЗ смена статуса - только у администратора
|
||
|
||
# Запросы PUT, PATCH и DELETE не работают с Form-data нативно, по этому в теле запроса должен быть JSON объект
|
||
data = json.loads(request.body)
|
||
|
||
params = {} # Инициализируем список параметров для запроса и заполняем его данными из JSON объекта
|
||
params['pk_application'] = str(data.get('pk_application'))
|
||
params['fk_status'] = str(data.get('fk_status'))
|
||
|
||
# Инициализируем список ошибок валидации данных
|
||
errors = []
|
||
# Автоматическая проверка параметров на заполненность
|
||
for key in params:
|
||
if params[key] is None or params[key] == '':
|
||
errors.append({ key: "Параметр обязателен для заполнения" })
|
||
|
||
# Если ошибки есть - то возвращаем ответ с ошибкой
|
||
if errors:
|
||
return JsonResponse({"code": 409, "errors": errors}, safe=False, status=409)
|
||
|
||
confirm = call_func('nn_db.application_patch', params)
|
||
|
||
# Если ответ не пустой и есть поле подтверждения
|
||
if any(confirm) and 'confirm' in confirm[0]:
|
||
if confirm[0]['confirm']:
|
||
# Если запись обновлена - возвращаем 202 ответ с сообщением
|
||
return JsonResponse({"code": 202, "message": "Данные обновлены"}, safe=False, status=202)
|
||
else:
|
||
# Если запись не обновлена - то возвращаем ответ 304 с сообщением
|
||
return JsonResponse({"code": 304, "message": "Данные не обновлены"}, safe=False, status=304)
|
||
else:
|
||
# Если нет - то возвращаем ошибку с сообщением
|
||
return JsonResponse({"code": 520, "message": "Не удалось обновить запись в БД, попробуйте позднее"}, safe=False, status=520)
|
||
|
||
def api_statuses(request):
|
||
user = is_auth(request)
|
||
if not user:
|
||
return JsonResponse({"code": 403, "message":"Доступ запрещен" }, safe=False, status=403)
|
||
|
||
if request.method == 'GET':
|
||
results = call_func('nn_db.status_get', {})
|
||
return JsonResponse({"code": 200, "data": results}, safe=False, status=200)
|
||
|
||
if request.method == 'POST':
|
||
params = {}
|
||
params['title'] = request.POST.get('title')
|
||
|
||
if params['title'] is None:
|
||
return JsonResponse({"code": 409, "errors": [{"title": "Параметр обязателен для заполнения"}]}, safe=False, status=409)
|
||
|
||
if len(params['title']) <= 4:
|
||
return JsonResponse({"code": 409, "errors": [{"title": "Описание статуса должно быть длинее 4 символов"}]}, safe=False, status=409)
|
||
|
||
confirm = call_func('nn_db.status_post', params)
|
||
|
||
# Проверяем, пришел ли в ответ на запрос первичный ключ добавленной заявки
|
||
if any(confirm) and 'pk_status' in confirm[0]:
|
||
# Если пришёл - то возвращаем данные
|
||
return JsonResponse({"code": 201, "data": confirm}, safe=False, status=201)
|
||
else:
|
||
# Если нет - то возвращаем ошибку с сообщением
|
||
return JsonResponse({"code": 520, "message": "Не удалось добавить запись в БД, попробуйте позднее"}, safe=False, status=520)
|
||
|
||
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) |