""" Вспомогательные функции со списками пользователей, статистикой и т.д. """ import logging from datetime import timedelta from typing import Union from django.contrib.auth import get_user_model 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: Пользователь, меняющий роль :return: Пользователь с обновленной ролью """ 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: """ Функция **get_users_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) -> list: """ Функция возвращает список тикетов пользователя Zendesk """ return TicketListRequester().get_tickets_list_for_user(zenpy.get_user(email)) def get_tickets_list_for_group(group_name): """ Функция возвращает список неназначенных, нерешённых тикетов группы Zendesk """ 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) 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 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: """ Функция проверяет, верны ли входные данные. :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: Обновленный профиль пользователя """ 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 """ 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: timedelta, end_date: timedelta) -> 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): 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, admin=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)