Обновить handlers.py — универсальный текст заявки 2FA

instruct/handlers.py

@@ -1,264 +1,56 @@
-import logging
+import re
-import asyncio
-import os
-import sys
-import json
-import httpx
 
-from trueconf import Router, Message
+path = "/opt/trueconf_bot/instruct/handlers.py"
-from trueconf.types import FSInputFile
+with open(path, "r") as f:
+    content = f.read()
 
-from utils.texts import (
+old = '''        description = (
-    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})")
 
-        if ad_info and not ad_info.get("mail"):
+new = '''        description = (
-            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 = (
                 f"Пользователь {cn} ({user_id}) найден в Active Directory, но у него не указан email. "
                 f"Требуется ручная проверка и выдача доступа к Trueconf."
-            )
+            )'''
-        else:
+
-            description = (
+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 = (
                 f"Пользователь {cn} ({user_id}) не найден в Active Directory. "
                 f"Требуется ручная проверка и выдача доступа к Trueconf."
-            )
+            )'''
 
-    try:
+new3 = '''            description = (
-        logger.info(f"Calling SD API directly: requester={requester_email}, city={city}")
+                f"Пользователь {cn} ({user_id}) отсутствует группа 2FA."
-        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:
-@router.message()
+    content = content.replace(old3, new3)
-async def instruct_router_handler(msg: Message):
+    with open(path, "w") as f:
-    user_id = msg.from_user.id
+        f.write(content)
-    user_name = getattr(msg.from_user, 'name', 'N/A')
+    print("OK 3 replaced")
-    user_username = getattr(msg.from_user, 'username', 'N/A')
+else:
-
+    print("OLD 3 NOT FOUND")
-    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.")
-
-            # Показываем текст с результатом
-            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))
-        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")