From f0be37b18fdaf9db9567af244ec84b5fe02b69e3 Mon Sep 17 00:00:00 2001 From: dddennnisss Date: Thu, 23 Jul 2026 15:49:59 +0700 Subject: [PATCH] =?UTF-8?q?=D0=9E=D0=B1=D0=BD=D0=BE=D0=B2=D0=B8=D1=82?= =?UTF-8?q?=D1=8C=20instruct/handlers.py?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- instruct/handlers.py | 97 +++++++++++++++++++++++++------------------- 1 file changed, 55 insertions(+), 42 deletions(-) diff --git a/instruct/handlers.py b/instruct/handlers.py index c5b090d..482afd7 100644 --- a/instruct/handlers.py +++ b/instruct/handlers.py @@ -1,3 +1,11 @@ +import logging +import asyncio +import os +import sys + +from trueconf import Router, Message +from trueconf.types import FSInputFile + from utils.texts import ( UNKNOWN_MAIN_CMD_TEXT, EMOJI_DIGITS, @@ -6,54 +14,46 @@ from utils.texts import ( INSTRUCT_TRUECONF_WITH_AD, INSTRUCT_EMAIL_TEXT, ) -import logging -import asyncio -import os - -from trueconf import Router, Message -from trueconf.types import FSInputFile - from utils.states import get_state, set_state, clear_state - -# 🔌 Импортируем централизованную функцию сбора статистики из main from utils.stats_logger import log_menu_stats - -# 🔌 Импортируем рабочую функцию создания заявки из SD модуля from service_desk.handlers import create_ticket_in_sd -# Импортируем конфиг для DEFAULT_REQUESTER и SD_URL -import importlib.util -config_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config", "config.py") -spec = importlib.util.spec_from_file_location("custom_config", config_path) -custom_config = importlib.util.module_from_spec(spec) -spec.loader.exec_module(custom_config) -DEFAULT_REQUESTER = custom_config.DEFAULT_REQUESTER -SD_URL = custom_config.SD_URL -SD_TOKEN = custom_config.SD_TOKEN +# Безопасный импорт конфигурации +try: + from config.config import DEFAULT_REQUESTER, SD_URL, SD_TOKEN +except ImportError: + import importlib.util + config_path = os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config", "config.py") + spec = importlib.util.spec_from_file_location("custom_config", config_path) + custom_config = importlib.util.module_from_spec(spec) + spec.loader.exec_module(custom_config) + DEFAULT_REQUESTER = custom_config.DEFAULT_REQUESTER + SD_URL = custom_config.SD_URL + SD_TOKEN = custom_config.SD_TOKEN logger = logging.getLogger(__name__) router = Router() # Путь к файлу инструкции Trueconf -TRUECONF_INSTRUCT_PATH = os.path.join(os.path.dirname(__file__), "Проверка_наличия_и_авторизация_на_мобильном_устройстве_Trueconf.docx") +TRUECONF_INSTRUCT_PATH = os.path.join( + os.path.dirname(__file__), + "Проверка_наличия_и_авторизация_на_мобильном_устройстве_Trueconf.docx" +) def get_user_cn(user_id: str) -> str | None: """ Получить CN пользователя из AD по его user_id (email или логин). user_id из TrueConf — это обычно email, например 'ds.krivochenko@tcs.sibcem.ru' - Заменяет tcs.sibcem.ru на sibcem.ru для поиска в AD. - Ищет по всем OU из AD_BASES. """ from utils.ad_search import search_by_user_id try: - # Заменяем tcs.sibcem.ru на sibcem.ru search_id = user_id.replace("@tcs.sibcem.ru", "@sibcem.ru") short_username = search_id.split("@")[0] if "@" in search_id else user_id - logger.info(f"AD lookup: user_id={user_id}, search_id={search_id}, short={short_username}") + logger.info(f"AD lookup CN: user_id={user_id}, search_id={search_id}, short={short_username}") entries = search_by_user_id(search_id, ["cn", "sAMAccountName", "mail", "userPrincipalName", "l", "userAccountControl"]) @@ -89,7 +89,7 @@ def get_ad_user_info(user_id: str) -> dict | None: raw_mail = str(entry.mail) if 'mail' in entry else "" # mail может быть "[]" (пустой LDAP объект), "" или "N/A" — фильтруем mail = raw_mail if raw_mail and raw_mail not in ("[]", "N/A", "") else None - city = str(entry.l) if 'l' in entry else "Кемерово" + city = str(entry.l) if 'l' in entry and entry.l.value else "Кемерово" return {"cn": cn, "mail": mail, "city": city} except Exception as e: logger.error(f"Ошибка получения AD info для {user_id}: {e}", exc_info=True) @@ -99,30 +99,43 @@ def get_ad_user_info(user_id: str) -> dict | None: async def create_sd_ticket_for_trueconf(user_id: str, cn: str) -> str | None: """ Создать заявку в Service Desk для запроса доступа к Trueconf. - - Если пользователь не найден в AD — создаёт заявку от ai@sibcem.ru. - Если пользователь найден, но не в группе 2FA — создаёт заявку от его email (или ai@sibcem.ru если mail нет). + + Если пользователь не найден в AD или у него нет email — создаёт заявку от DEFAULT_REQUESTER (ai@sibcem.ru). + Если пользователь найден — создаёт заявку от его email. """ subject = "Запрос доступа к Trueconf" - + # Пытаемся получить информацию из AD ad_info = get_ad_user_info(user_id) - + logger.info(f"AD search result for SD ticket ({user_id}): {ad_info}") + if ad_info and ad_info.get("mail"): # Пользователь найден в AD и имеет email requester_email = ad_info["mail"] - city = ad_info["city"] - description = f"Пользователь {cn} ({user_id}) запросил инструкцию по Trueconf, но не имеет группы 2FA. Требуется выдача доступа." + city = ad_info.get("city", "Кемерово") + description = ( + f"Пользователь {cn} ({user_id}) запросил инструкцию по Trueconf, " + f"но не имеет группы 2FA. Требуется выдача доступа." + ) else: - # Пользователь не найден в AD ИЛИ не имеет email — создаём от ai@sibcem.ru + # Пользователь не найден в AD ИЛИ не имеет email — создаём от DEFAULT_REQUESTER (ai@sibcem.ru) requester_email = DEFAULT_REQUESTER city = "Кемерово" + logger.info(f"Using DEFAULT_REQUESTER ({requester_email}) for user {user_id} (cn={cn})") + if ad_info and not ad_info.get("mail"): - description = f"Пользователь {cn} ({user_id}) не найден в Active Directory (нет email). Требуется ручная проверка и выдача доступа к Trueconf." + description = ( + f"Пользователь {cn} ({user_id}) найден в Active Directory, но у него не указан email. " + f"Требуется ручная проверка и выдача доступа к Trueconf." + ) else: - description = f"Пользователь {cn} ({user_id}) не найден в Active Directory. Требуется ручная проверка и выдача доступа к Trueconf." - + description = ( + f"Пользователь {cn} ({user_id}) не найден в Active Directory. " + f"Требуется ручная проверка и выдача доступа к Trueconf." + ) + try: + logger.info(f"Calling create_ticket_in_sd: requester={requester_email}, city={city}") ticket_id = await create_ticket_in_sd(requester_email, subject, description, city) if ticket_id: logger.info(f"SD ticket created for {cn}: #{ticket_id} (requester={requester_email})") @@ -173,19 +186,19 @@ async def instruct_router_handler(msg: Message): elif cmd == "1": log_menu_stats(user_id, "Инструкции", "Trueconf") set_state(user_id, "INSTRUCT_TRUECONF_VIEW") - + # Получаем CN пользователя cn = get_user_cn(user_id) if cn is None: cn = user_id # fallback - + # Создаём заявку в SD ticket_id = await create_sd_ticket_for_trueconf(user_id, cn) - + # Показываем текст с результатом text = INSTRUCT_TRUECONF_WITH_AD(EMOJI_DIGITS, cn, ticket_id) await msg.answer(text, parse_mode="html") - + # Отправляем документ-инструкцию if os.path.isfile(TRUECONF_INSTRUCT_PATH): await msg.answer_document(FSInputFile(TRUECONF_INSTRUCT_PATH)) @@ -206,4 +219,4 @@ async def instruct_router_handler(msg: Message): set_state(user_id, "INSTRUCT_MODE") await msg.answer(INSTRUCT_MAIN_MENU_TEXT, parse_mode="html") else: - await msg.answer(UNKNOWN_MAIN_CMD_TEXT, parse_mode="html") + await msg.answer(UNKNOWN_MAIN_CMD_TEXT, parse_mode="html") \ No newline at end of file