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, 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)