diff --git a/instruct/handlers.py b/instruct/handlers.py index 28aa5a8..f4936fc 100644 --- a/instruct/handlers.py +++ b/instruct/handlers.py @@ -1,56 +1,266 @@ -import re +import logging +import asyncio +import os +import sys +import json +import httpx -path = "/opt/trueconf_bot/instruct/handlers.py" -with open(path, "r") as f: - content = f.read() +from trueconf import Router, Message +from trueconf.types import FSInputFile -old = ''' description = ( +from utils.texts import ( + UNKNOWN_MAIN_CMD_TEXT, + EMOJI_DIGITS, + INSTRUCT_MAIN_MENU_TEXT, + INSTRUCT_TRUECONF_TEXT, + INSTRUCT_TRUECONF_WITH_AD, + INSTRUCT_EMAIL_TEXT, +) +from utils.states import get_state, set_state, clear_state +from utils.stats_logger import log_menu_stats + +# Безопасный импорт конфигурации +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" +) + + +def get_user_cn(user_id: str) -> str | None: + """ + Получить CN пользователя из AD по его user_id (email или логин). + user_id из TrueConf — это обычно email, например 'ds.krivochenko@tcs.sibcem.ru' + """ + from utils.ad_search import search_by_user_id + + try: + 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 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"]) + + if entries: + entry = entries[0] + cn = str(entry.cn) + mail = str(entry.mail) if 'mail' in entry else "" + sam = str(entry.sAMAccountName) if 'sAMAccountName' in entry else "N/A" + upn = str(entry.userPrincipalName) if 'userPrincipalName' in entry else "N/A" + logger.info(f"AD found: cn={cn}, mail={mail}, sam={sam}, upn={upn}") + return cn + + logger.warning(f"AD not found for user_id={user_id}, search_id={search_id}, short={short_username}") + except Exception as e: + logger.error(f"Ошибка получения CN из AD для {user_id}: {e}", exc_info=True) + return None + + +def get_ad_user_info(user_id: str) -> dict | None: + """ + Получить полную информацию о пользователе из AD. + Возвращает {cn, mail, city} или None. + """ + from utils.ad_search import search_by_user_id + + try: + search_id = user_id.replace("@tcs.sibcem.ru", "@sibcem.ru") + entries = search_by_user_id(search_id, ["cn", "mail", "l", "userAccountControl"]) + + if entries: + entry = entries[0] + cn = str(entry.cn) + 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 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) + return None + + +async def _send_sd_api_request(requester_email: str, subject: str, description: str, city: str) -> str | None: + """ + Автономная прямая отправка HTTP-запроса в Service Desk API v3. + """ + # Если запрашивает системный ИИ или не указан email — ищем по ИМЕНИ + if requester_email == DEFAULT_REQUESTER or requester_email == "ai@sibcem.ru": + requester_payload = {"name": "Искусственный Интеллект"} + else: + requester_payload = {"email_id": requester_email} + + payload = { + "request": { + "subject": subject, + "description": description, + "requester": requester_payload, + "udf_fields": { + "udf_pick_301": city + } + } + } + + url = f"{SD_URL.rstrip('/')}/api/v3/requests" + headers = {"TECHNICIAN_KEY": SD_TOKEN} + data = {"input_data": json.dumps(payload)} + + async with httpx.AsyncClient(verify=False) as client: + response = await client.post(url, headers=headers, data=data, timeout=15.0) + + if response.status_code in (200, 201): + res_json = response.json() + ticket_id = res_json.get("request", {}).get("id") + return str(ticket_id) if ticket_id else None + else: + logger.error(f"Service Desk API Error ({response.status_code}): {response.text}") + return None + + +async def create_sd_ticket_for_trueconf(user_id: str, cn: str) -> str | None: + """ + Создать заявку в Service Desk для запроса доступа к Trueconf. + Полностью автономная функция. + """ + 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"): + requester_email = ad_info["mail"] + city = ad_info.get("city", "Кемерово") + description = ( f"Пользователь {cn} ({user_id}) запросил инструкцию по Trueconf, " f"но не имеет группы 2FA. Требуется выдача доступа." - )''' + ) + else: + requester_email = DEFAULT_REQUESTER + city = "Кемерово" + logger.info(f"Using DEFAULT_REQUESTER ({requester_email}) for user {user_id} (cn={cn})") -new = ''' description = ( - f"Пользователь {cn} ({user_id}) отсутствует группа 2FA." - )''' - -if old in content: - content = content.replace(old, new) - with open(path, "w") as f: - f.write(content) - print("OK 1 replaced") -else: - print("OLD 1 NOT FOUND") - -old2 = ''' description = ( + if ad_info and not ad_info.get("mail"): + description = ( f"Пользователь {cn} ({user_id}) найден в Active Directory, но у него не указан email. " f"Требуется ручная проверка и выдача доступа к Trueconf." - )''' - -new2 = ''' description = ( - f"Пользователь {cn} ({user_id}) отсутствует группа 2FA." - )''' - -if old2 in content: - content = content.replace(old2, new2) - with open(path, "w") as f: - f.write(content) - print("OK 2 replaced") -else: - print("OLD 2 NOT FOUND") - -old3 = ''' description = ( + ) + else: + description = ( f"Пользователь {cn} ({user_id}) не найден в Active Directory. " f"Требуется ручная проверка и выдача доступа к Trueconf." - )''' + ) -new3 = ''' description = ( - f"Пользователь {cn} ({user_id}) отсутствует группа 2FA." - )''' + try: + logger.info(f"Calling SD API directly: requester={requester_email}, city={city}") + ticket_id = await _send_sd_api_request(requester_email, subject, description, city) + if ticket_id: + logger.info(f"SD ticket created for {cn}: #{ticket_id} (requester={requester_email})") + else: + logger.error(f"Failed to create SD ticket for {cn}") + return ticket_id + except Exception as e: + logger.error(f"Ошибка создания SD заявки для {cn}: {e}", exc_info=True) + return None -if old3 in content: - content = content.replace(old3, new3) - with open(path, "w") as f: - f.write(content) - print("OK 3 replaced") -else: - print("OLD 3 NOT FOUND") + +@router.message() +async def instruct_router_handler(msg: Message): + user_id = msg.from_user.id + user_name = getattr(msg.from_user, 'name', 'N/A') + user_username = getattr(msg.from_user, 'username', 'N/A') + + logger.info(f"Instruct: user_id={user_id}, name={user_name}, username={user_username}") + + current_state = get_state(user_id) + + if not msg.text: + return + + cmd = msg.text.strip().lower() + + for raw_num, emoji_num in EMOJI_DIGITS.items(): + if cmd == emoji_num: + cmd = raw_num + break + + if current_state is None: + from utils.menu import MENU_MAP + if MENU_MAP.get(cmd) == "INSTRUCT": + msg.handled = True + set_state(user_id, "INSTRUCT_MODE") + await msg.answer(INSTRUCT_MAIN_MENU_TEXT, parse_mode="html") + return + return + + if current_state == "INSTRUCT_MODE": + msg.handled = True + if cmd == "0": + log_menu_stats(user_id, "Инструкции", "Выход в главное меню") + clear_state(user_id) + from utils.menu import MENU_TEXT + await msg.answer(MENU_TEXT, parse_mode="html") + 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 + + # Проверяем наличие группы 2FA в AD перед созданием заявки + from utils.ad_checker import check_group_membership + ad_check = check_group_membership(cn, "2FA") + + ticket_id = None + # Создаём заявку в SD ТОЛЬКО если у пользователя НЕТ группы 2FA (или он не найден) + if not ad_check.get("in_group", False): + ticket_id = await create_sd_ticket_for_trueconf(user_id, cn) + else: + logger.info(f"User {cn} already has 2FA group. SD ticket creation skipped.") + + # Показываем текст с результатом и отправляем документ с caption + text = INSTRUCT_TRUECONF_WITH_AD(EMOJI_DIGITS, cn, ticket_id) + if os.path.isfile(TRUECONF_INSTRUCT_PATH): + await msg.bot.send_document( + chat_id=msg.chat_id, + file=FSInputFile(TRUECONF_INSTRUCT_PATH), + caption=text, + parse_mode="html" + ) + elif cmd == "2": + log_menu_stats(user_id, "Инструкции", "Почта") + set_state(user_id, "INSTRUCT_EMAIL_VIEW") + await msg.answer(INSTRUCT_EMAIL_TEXT, parse_mode="html") + else: + await msg.answer(UNKNOWN_MAIN_CMD_TEXT, parse_mode="html") + + elif current_state in ["INSTRUCT_TRUECONF_VIEW", "INSTRUCT_EMAIL_VIEW"]: + msg.handled = True + if cmd == "0": + clear_state(user_id) + from utils.menu import MENU_TEXT + await msg.answer(MENU_TEXT, parse_mode="html") + elif cmd == "9": + 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") \ No newline at end of file