from smtplib import SMTPException from django.contrib import messages from django.contrib.auth.decorators import login_required from django.contrib.auth.mixins import LoginRequiredMixin, PermissionRequiredMixin from django.contrib.auth.models import User, Permission from django.contrib.auth.tokens import default_token_generator from django.contrib.auth.forms import PasswordResetForm from django.contrib.auth.views import LoginView from django.contrib.contenttypes.models import ContentType from django.contrib.messages.views import SuccessMessageMixin from django.core.handlers.wsgi import WSGIRequest from django.http import HttpResponseRedirect, HttpResponse from django.shortcuts import render, redirect, get_list_or_404 from django.urls import reverse_lazy, reverse from django.views.generic import FormView from django_registration.views import RegistrationView # Django REST from rest_framework import viewsets from rest_framework.response import Response from zenpy.lib.api_objects import User as ZenpyUser from access_controller.settings import DEFAULT_FROM_EMAIL, ZENDESK_ROLES, ZENDESK_MAX_AGENTS from main.extra_func import check_user_exist, update_profile, get_user_organization, \ make_engineer, make_light_agent, get_users_list, update_users_in_model, count_users, \ StatisticData, log, ZendeskAdmin from main.forms import AdminPageUsers, CustomRegistrationForm, CustomAuthenticationForm, StatisticForm from main.serializers import ProfileSerializer, ZendeskUserSerializer from .models import UserProfile def setup_context(profile_lit: bool = False, control_lit: bool = False, work_lit: bool = False, registration_lit: bool = False, login_lit: bool = False): context = { 'profile_lit': profile_lit, 'control_lit': control_lit, 'work_lit': work_lit, 'registration_lit': registration_lit, 'login_lit': login_lit, } return context class CustomRegistrationView(RegistrationView): """ Отображение и логика работы страницы регистрации пользователя. :param form_class: Форма, которую необходимо заполнить для регистрации :type form_class: :class:`forms.CustomRegistrationForm` :param template_name: Указание пути к html-странице django регистрации :type template_name: :class:`str` :param success_url: Указание пути к html-странице завершения регистрации :type success_url: :class:`django.utils.functional.lazy..__proxy__` :param is_allowed: Определение зарегистрирован ли пользователь с введенным email на Zendesk и принадлежит ли он к организации SYSTEM :type is_allowed: :class:`bool` """ extra_context = setup_context(registration_lit=True) form_class = CustomRegistrationForm template_name = 'django_registration/registration_form.html' urls = { 'done': reverse_lazy('password_reset_done'), 'invalid_zendesk_email': reverse_lazy('django_registration_disallowed'), 'email_sending_error': reverse_lazy('registration_email_error'), } redirect_url = 'done' def register(self, form: CustomRegistrationForm) -> User: """ Функция регистрации пользователя. 1. Ввод email пользователя, указанный на Zendesk 2. В случае если пользователь с данным паролем зарегистрирован на Zendesk и относится к организации SYSTEM, происходит сброс ссылки с установлением пароля на указанный email 3. Создается пользователь class User, а также его профиль. :param form: Email пользователя на Zendesk :return: user """ self.redirect_url = 'done' if check_user_exist(form.data['email']) and get_user_organization(form.data['email']) == 'SYSTEM': forms = PasswordResetForm(self.request.POST) if forms.is_valid(): opts = { 'use_https': self.request.is_secure(), 'token_generator': default_token_generator, 'from_email': DEFAULT_FROM_EMAIL, 'email_template_name': 'registration/password_reset_email.html', 'subject_template_name': 'registration/password_reset_subject.txt', 'request': self.request, 'html_email_template_name': None, 'extra_email_context': None, } user = User.objects.create_user( username=form.data['email'], email=form.data['email'], password=User.objects.make_random_password(length=50) ) try: update_profile(user.userprofile) self.set_permission(user) forms.save(**opts) return user except SMTPException: self.redirect_url = 'email_sending_error' else: raise ValueError('Непредвиденная ошибка') else: self.redirect_url = 'invalid_zendesk_email' @staticmethod def set_permission(user: User) -> None: """ Функция дает разрешение на просмотр страница администратора, если пользователь имеет роль admin. :param user: авторизованный пользователь (получает разрешение, имея роль "admin") """ if user.userprofile.role == 'admin': content_type = ContentType.objects.get_for_model(UserProfile) permission = Permission.objects.get( codename='has_control_access', content_type=content_type, ) user.user_permissions.add(permission) def get_success_url(self, user: User = None): """ Функция возвращает url-адрес страницы, куда нужно перейти после успешной/не успешной регистрации. Используется самой django-registration. :param user: пользователь, пытающийся зарегистрироваться :return: адресация на страницу успешной регистрации """ return self.urls[self.redirect_url] def registration_error(request): return render(request, 'django_registration/registration_error.html') @login_required() def profile_page(request: WSGIRequest) -> HttpResponse: """ Функция отображения страницы профиля. :param request: данные пользователя из БД :return: адресация на страницу пользователя """ user_profile: UserProfile = request.user.userprofile update_profile(user_profile) context = setup_context(profile_lit=True) context.update({ 'profile': user_profile, 'pagename': 'Страница профиля', 'ZENDESK_ROLES': ZENDESK_ROLES, }) return render(request, 'pages/profile.html', context) def auth_user(request: WSGIRequest) -> ZenpyUser: """ Функция возвращает профиль пользователя на Zendesk. :param request: email, subdomain и token пользователя :return: объект пользователя Zendesk """ admin = ZendeskAdmin().admin zenpy_user: ZenpyUser = admin.users.search(request.user.email).values[0] return zenpy_user, admin @login_required() def work_page(request: WSGIRequest, id: int) -> HttpResponse: """ Функция отображения страницы "Управления правами" для текущего пользователя (login_required). :param request: объект пользователя :param id: id пользователя, используется для динамической адресации :return: адресация на страницу "Управления правами" (либо на страницу "Авторизации", если id и user.id не совпадают """ users = get_users_list() if request.user.id == id: engineers = [] light_agents = [] for user in users: if user.custom_role_id == ZENDESK_ROLES['engineer']: engineers.append(user) elif user.custom_role_id == ZENDESK_ROLES['light_agent']: light_agents.append(user) context = setup_context(work_lit=True) context.update({ 'engineers': engineers, 'agents': light_agents, 'messages': messages.get_messages(request), 'licences_remaining': max(0, ZENDESK_MAX_AGENTS - len(engineers)), 'pagename': 'Управление правами' }) return render(request, 'pages/work.html', context) return redirect("login") @login_required() def work_hand_over(request: WSGIRequest) -> HttpResponseRedirect: """ Функция позволяет текущему пользователю (login_required) сдать права, а именно сменить в Zendesk роль с "engineer" на "light agent" и установить роль "agent" в БД. Действия выполняются, если исходная роль пользователя "engineer". :param request: данные текущего пользователя (login_required) :return: перезагрузка текущей страницы после выполнения смены роли """ make_light_agent(request.user.userprofile,request.user) return HttpResponseRedirect(reverse('work', args=(request.user.id,))) @login_required() def work_become_engineer(request: WSGIRequest) -> HttpResponseRedirect: """ Функция меняет роль пользователя в Zendesk на "engineer" и присваивает роль "agent" в БД (в случае, если исходная роль пользователя была "light_agent"). :param request: данные текущего пользователя (login_required) :return: перезагрузка текущей страницы после выполнения смены роли """ zenpy_user, admin = auth_user(request) make_engineer(request.user.userprofile,request.user) return HttpResponseRedirect(reverse('work', args=(request.user.id,))) @login_required() def work_get_tickets(request): zenpy_user, admin = auth_user(request) count_tickets = int(request.GET["count_tickets"]) tickets = [ticket for ticket in admin.search(type="ticket") if ticket.group.name == 'Сменная группа' and ticket.assignee is None] for i in range(len(tickets)): if i == count_tickets: return HttpResponseRedirect(reverse('work', args=(request.user.id,))) tickets[i].assignee = zenpy_user admin.tickets.update(tickets[i]) return HttpResponseRedirect(reverse('work', args=(request.user.id,))) def main_page(request: WSGIRequest) -> HttpResponse: """ Функция переадресации на главную страницу. """ return render(request, 'pages/index.html') class AdminPageView(LoginRequiredMixin, PermissionRequiredMixin, SuccessMessageMixin, FormView): """ Класс отображения страницы администратора. :param permission_required: Права доступа к странице администратора :type permission_required: :class:`str` :param template_name: HTML-шаблон страницы администратора :type template_name: :class:`str` :param form_class: Форма страницы администратора :type form_class: :class:`forms.AdminPageUsersForm` :param success_url: Адрес страницы администратора :type success_url: :class:`HttpResponseRedirect` """ permission_required = 'main.has_control_access' template_name = 'pages/adm_ruleset.html' form_class = AdminPageUsers success_url = '/control/' success_message = "Права были изменены." def form_valid(self, form: AdminPageUsers) -> AdminPageUsers: """ Функция обновления страницы AdminPageUsers. :param form: Форма страницы администратора :return: Обновленная страница (пользователям проставлены новые статусы) """ users = form.cleaned_data['users'] if 'engineer' in self.request.POST: self.make_engineers(users) elif 'light_agent' in self.request.POST: self.make_light_agents(users) return super().form_valid(form) def make_engineers(self, users): """ Функция проходит по списку пользователей, проставляя статус "engineer". :param users: Список пользователей :return: Обновленный список пользователей """ for user in users: make_engineer(user, self.request.user) log(user, self.request.user.userprofile) def make_light_agents(self, users): """ Функция проходит по списку пользователей, проставляя статус "light agent". :param users: Список пользователей :return: Обновленный список пользователей """ for user in users: make_light_agent(user, self.request.user) log(user, self.request.user.userprofile) class CustomLoginView(LoginView): """ Отображение страницы авторизации пользователя """ extra_context = setup_context(login_lit=True) form_class = CustomAuthenticationForm class UsersViewSet(viewsets.ReadOnlyModelViewSet): """ Класс для получения пользователей с помощью api """ queryset = UserProfile.objects.filter(role='agent') serializer_class = ProfileSerializer def list(self, request, *args, **kwargs): users = update_users_in_model() count = count_users(users.values) profiles = UserProfile.objects.filter(role='agent') serializer = self.get_serializer(profiles, many=True) res = { 'users': serializer.data, 'engineers': count[0], 'light_agents': count[1], 'zendesk_users': self.get_zendesk_users(self.choose_users(users.values, profiles)), 'max_agents': ZENDESK_MAX_AGENTS } return Response(res) @staticmethod def choose_users(zendesk, model): users = [] for zendesk_user in zendesk: if zendesk_user.name not in [user.name for user in model]: users.append(zendesk_user) return users @staticmethod def get_zendesk_users(users): zendesk_users = ZendeskUserSerializer( data=[user for user in users if user.role != 'admin'], many=True ) zendesk_users.is_valid() return zendesk_users.data @login_required() def statistic_page(request: WSGIRequest) -> HttpResponse: """ Функция отображения страницы статистики (для "superuser"). :param request: данные о пользователе: email, время и интервал работы. Данные получаем через forms.StatisticForm :return: адресация на страницу статистики """ # if not request.user.has_perm('main.has_control_access'): # raise PermissionDenied # context = { if not request.user.is_superuser: return redirect('index') context = setup_context() context.update({ 'pagename': 'страница статистики', 'errors': list(), }) if request.method == "POST": form = StatisticForm(request.POST) if form.is_valid(): start_date, end_date = form.cleaned_data['range_start'], form.cleaned_data['range_end'] interval, show = form.cleaned_data['interval'], form.cleaned_data['display_format'] Data = StatisticData(start_date, end_date, form.cleaned_data['email']) Data.set_display(show) Data.set_interval(interval) stats = Data.get_statistic() if Data.errors: context['errors'] = Data.errors if Data.warnings: context['warnings'] = Data.warnings context['log_stats'] = stats if not context['errors'] else None if request.method == 'GET': form = StatisticForm() context['form'] = form return render(request, 'pages/statistic.html', context)