""" Вспомогательные функции. """ import logging from datetime import timedelta, date from typing import Union, Optional from django.contrib.auth import get_user_model from django.contrib.auth.models import Permission from django.contrib.contenttypes.models import ContentType from django.core.exceptions import ObjectDoesNotExist from django.core.handlers.wsgi import WSGIRequest from django.http import HttpResponseRedirect, HttpResponsePermanentRedirect from django.shortcuts import redirect from django.utils import timezone from zenpy import Zenpy from zenpy.lib.api_objects import User as ZenpyUser, Ticket as ZenpyTicket from zenpy.lib.exception import APIException from zenpy.lib.generator import SearchResultGenerator from access_controller.settings import ZENDESK_ROLES as ROLES, ACTRL_ZENDESK_SUBDOMAIN from main.models import UserProfile, RoleChangeLogs, UnassignedTicket, UnassignedTicketStatus from main.requester import TicketListRequester from main.zendesk_admin import zenpy def update_role(user_profile: UserProfile, role: int, who_changes: get_user_model()) -> None: """ Функция меняет роль пользователя. :param user_profile: Профиль пользователя :param role: Новая роль :param who_changes: Пользователь, меняющий роль """ zendesk = zenpy user = zendesk.get_user(user_profile.user.email) user.custom_role_id = role user_profile.custom_role_id = role user_profile.save() log(user_profile, who_changes.userprofile) zendesk.update_user(user) def make_engineer(user_profile: UserProfile, who_changes: get_user_model()) -> None: """ Функция устанавливает пользователю роль инженера. :param user_profile: Профиль пользователя :return: Вызов функции **update_role** с параметрами: профиль пользователя, роль "engineer" """ update_role(user_profile, ROLES['engineer'], who_changes) def make_light_agent(user_profile: UserProfile, who_changes: get_user_model()) -> None: """ Функция устанавливает пользователю роль легкого агента. :param user_profile: Профиль пользователя :return: Вызов функции **update_role** с параметрами: профиль пользователя, роль "light_agent". Предварительно снимаем тикеты, находящие в работы у пользователя. """ tickets: SearchResultGenerator = get_tickets_list(user_profile.user.email) ticket: ZenpyTicket for ticket in tickets: UnassignedTicket.objects.create( assignee=user_profile.user, ticket_id=ticket.id, status=UnassignedTicketStatus.SOLVED if ticket.status == 'solved' else UnassignedTicketStatus.UNASSIGNED ) if ticket.status == 'solved': ticket.assignee_id = zenpy.solved_tickets_user_id else: ticket.assignee = None ticket.group_id = zenpy.buffer_group_id if tickets: zenpy.admin.tickets.update(tickets) attempts, success = 20, False while not success and attempts != 0: try: update_role(user_profile, ROLES['light_agent'], who_changes) success = True except APIException as e: attempts -= 1 if attempts == 0: raise e def get_users_list() -> list: """ Функция возвращает список пользователей Zendesk, относящихся к организации SYSTEM. """ zendesk = zenpy # У пользователей должна быть организация SYSTEM org = next(zendesk.admin.search(type='organization', name='SYSTEM')) users = zendesk.admin.organizations.users(org) return users def get_tickets_list(email: str) -> Optional[list]: """ Функция возвращает список тикетов пользователя Zendesk. :param email: Email пользователя :return: Список тикетов пользователя """ return TicketListRequester().get_tickets_list_for_user(zenpy.get_user(email)) def get_tickets_list_for_group(group_name: str) -> Optional[list]: """ Функция возвращает список не назначенных, не решённых тикетов группы Zendesk. :param group_name: Название группы пользователя :return: Список тикетов группы """ return TicketListRequester().get_tickets_list_for_group(zenpy.get_group(group_name)) def update_profile(user_profile: UserProfile) -> None: """ Функция обновляет профиль пользователя в БД в соответствии с текущим в Zendesk. :param user_profile: Профиль пользователя :return: Обновленный, в соответствие с текущими данными в Zendesk, профиль пользователя """ user = zenpy.get_user(user_profile.user.email) update_permission(user_profile, user) user_profile.name = user.name user_profile.role = user.role user_profile.custom_role_id = user.custom_role_id if user.custom_role_id else 0 user_profile.image = user.photo['content_url'] if user.photo else None user_profile.save() def update_permission(user_profile: UserProfile, user: ZenpyUser): """ Функция обновляет права доступа пользователя в БД. :param user_profile: Профиль пользователя :param user: Данные пользователя в Zendesk """ if user_profile.role != user.role: user_profile.role = user.role user_profile.save() set_permission(user_profile.user) del_permission(user_profile.user) def set_permission(user: get_user_model()) -> 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) user.save() def del_permission(user: get_user_model()) -> None: """ Функция забирает разрешение на просмотр страница администратора, если пользователь не имеет роль admin. :param user: Авторизованный пользователь (теряет разрешение, не имея роль "admin") """ if user.userprofile.role == 'agent' and user.has_perm('main.has_control_access'): content_type = ContentType.objects.get_for_model(UserProfile) permission = Permission.objects.get( codename='has_control_access', content_type=content_type, ) user.user_permissions.remove(permission) user.save() def check_user_exist(email: str) -> bool: """ Функция проверяет, существует ли пользователь. :param email: Email пользователя :return: Зарегистрирован ли пользователь в Zendesk """ return zenpy.check_user(email) def get_user_organization(email: str) -> str: """ Функция возвращает организацию пользователя. :param email: Email пользователя :return: Организация пользователя """ return zenpy.get_user_org(email) def check_user_auth(email: str, password: str) -> bool: """ Функция проверяет, верны ли входные данные. :param email: Email пользователя :param password: Пароль пользователя :return: Существует ли пользователь :raise: :class:`APIException`: исключение, вызываемое если пользователь не аутентифицирован """ creds = { 'email': email, 'password': password, 'subdomain': ACTRL_ZENDESK_SUBDOMAIN, } try: user = Zenpy(**creds) user.search(email, type='user') except APIException: return False return True def update_user_in_model(profile: UserProfile, zendesk_user: ZenpyUser) -> None: """ Функция обновляет профиль пользователя в модели при изменении данных пользователя на Zendesk. :param profile: Профиль пользователя :param zendesk_user: Данные пользователя в Zendesk :return: Обновленный профиль пользователя """ update_permission(profile, zendesk_user) profile.name = zendesk_user.name profile.role = zendesk_user.role profile.image = zendesk_user.photo['content_url'] if zendesk_user.photo else None if zendesk_user.custom_role_id is not None: profile.custom_role_id = int(zendesk_user.custom_role_id) profile.save() def count_users(users: list) -> tuple: """ Функция подсчета количества сотрудников с ролями engineer и light_agent. :param users: Список пользователей :return: Количество инженеров, количество light_agents """ engineers, light_agents = 0, 0 for user in users: if user.custom_role_id == ROLES['engineer']: engineers += 1 elif user.custom_role_id == ROLES['light_agent']: light_agents += 1 return engineers, light_agents def update_users_in_model() -> list: """ Обновляет пользователей в модели UserProfile по списку пользователей в организации. """ users = get_users_list() for user in users: try: profile = get_user_model().objects.get(email=user.email).userprofile update_user_in_model(profile, user) except ObjectDoesNotExist: pass return users def daterange(start_date: date, end_date: date) -> list: """ Функция возвращает список дней с start_date по end_date, исключая правую границу. :param start_date: Начальная дата :param end_date: Конечная дата :return: Список дней, не включая конечную дату """ dates = [] for n in range(int((end_date - start_date).days)): dates.append(start_date + timedelta(n)) return dates def get_timedelta(current_log: RoleChangeLogs, time: timedelta = None) -> timedelta: """ Функция возвращает объект класса timedelta, который хранит промежуток времени от начала суток до момента, который находится в log (объект класса RoleChangeLogs) или в time(datetime.time), если введён. :param current_log: Лог :param time: Время :return: Сколько времени прошло от начала суток до события """ if time is None: time = current_log.change_time.time() time = timedelta(hours=time.hour, minutes=time.minute, seconds=time.second) return time def last_day_of_month(day: int) -> int: """ Функция возвращает последний день текущего месяца. :param day: Текущий день :return: Последний день месяца """ next_month = day.replace(day=28) + timedelta(days=4) return next_month - timedelta(days=next_month.day) class DatabaseHandler(logging.Handler): """ Класс записи изменений ролей в базу данных. """ def __init__(self): logging.Handler.__init__(self) def emit(self, record: logging.LogRecord) -> None: """ Функция записи в базу данных лога с изменением роли пользователя. :param record: Лог смены роли пользователя :return: Запись в БД лога по смене роли пользователя с указанием новой и старой роли, а также автора изменения """ database = RoleChangeLogs() users = record.msg if users[1]: user = users[0] admin = users[1] elif not users[1]: user = users[0] admin = users[0] database.name = user.name database.user = user.user database.changed_by = admin.user if user.custom_role_id == ROLES['engineer']: database.old_role = ROLES['light_agent'] elif user.custom_role_id == ROLES['light_agent']: database.old_role = ROLES['engineer'] database.new_role = user.custom_role_id database.save() class CsvFormatter(logging.Formatter): """ Класс преобразования смены ролей пользователей в строковый формат. """ def __init__(self): logging.Formatter.__init__(self) def format(self, record: logging.LogRecord) -> str: """ Функция форматирует запись смены роли пользователя в строку. :param record: Лог смены роли пользователя. :return: Строка с записью смены пользователя. """ users = record.msg if users[1]: user = users[0] admin = users[1] elif not users[1]: user = users[0] admin = users[0] msg = '' msg += user.name if user.custom_role_id == ROLES['engineer']: msg += ',engineer,' elif user.custom_role_id == ROLES['light_agent']: msg += ',light_agent,' time = str(timezone.now().today()) msg += time[:16] msg += ',' msg += admin.name return msg def log(user: get_user_model(), admin: get_user_model() = None) -> None: """ Функция осуществляет запись логов в базу данных и csv файл. :param admin: Админ, который меняет роль :param user: Пользователь, которому изменена роль """ users = [user, admin] logger = logging.getLogger('MY_LOGGER') if not logger.hasHandlers(): dbhandler = DatabaseHandler() csvformatter = CsvFormatter() csvhandler = logging.FileHandler('logs/logs.csv', "a") csvhandler.setFormatter(csvformatter) logger.addHandler(dbhandler) logger.addHandler(csvhandler) logger.setLevel('INFO') logger.info(users) def set_session_params_for_work_page(request: WSGIRequest, count: int = None, is_confirm: bool = True) -> \ Union[HttpResponsePermanentRedirect, HttpResponseRedirect]: """ Функция для страницы получения прав, устанавливает данные сессии о успешности запроса и количестве назначенных тикетов. :param request: Получение данных с рабочей страницы пользователя :param count: Количество запрошенных тикетов :param is_confirm: Назначены ли тикеты :return: Перезагрузка страницы "Управление правами" соответствующего пользователя """ request.session['is_confirm'] = is_confirm request.session['count_tickets'] = count return redirect('work', request.user.id)