import logging import asyncio import os import sys import json import httpx from trueconf import Router, Message from trueconf.types import FSInputFile from utils.texts import ( UNKNOWN_MAIN_CMD_TEXT, EMOJI_DIGITS, INSTRUCT_MAIN_MENU_TEXT, INSTRUCT_TRUECONF_TEXT, INSTRUCT_TRUECONF_WITH_AD, INSTRUCT_TRUECONF_PENDING_TEXT, INSTRUCT_EMAIL_TEXT, INSTRUCT_ERROR_MSG, ) 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, SD_INSTRUCT_GROUP_NAME, SD_INSTRUCT_TECHNICIAN_NAME 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 SD_INSTRUCT_GROUP_NAME = custom_config.SD_INSTRUCT_GROUP_NAME SD_INSTRUCT_TECHNICIAN_NAME = custom_config.SD_INSTRUCT_TECHNICIAN_NAME 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", "userPrincipalName"]) 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 # Fallback: используем UPN если mail пустой if not mail: raw_upn = str(entry.userPrincipalName) if 'userPrincipalName' in entry else "" if raw_upn and raw_upn not in ("[]", "N/A", ""): mail = raw_upn 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_cn: str, requester_email: str, subject: str, description: str, city: str ) -> str | None: """ Автономная прямая отправка HTTP-запроса в Service Desk API v3. Умеет привязывать пользователя по ФИО, Email или Fallback. """ # 1. Формируем структуру заявителя (приоритет: ФИО -> Email -> ИИ) if requester_cn and requester_cn != "N/A": requester_payload = {"name": requester_cn} elif requester_email and requester_email not in (DEFAULT_REQUESTER, "ai@sibcem.ru"): requester_payload = {"email_id": requester_email} else: requester_payload = {"name": "Искусственный Интеллект"} # 2. Поиск ID группы (пробуем /groups и /group) group_id = None try: headers = {"TECHNICIAN_KEY": SD_TOKEN} async with httpx.AsyncClient(verify=False) as client: for endpoint in ["/api/v3/groups", "/api/v3/group"]: url_groups = f"{SD_URL.rstrip('/')}{endpoint}" resp = await client.get(url_groups, headers=headers, timeout=10.0) if resp.status_code == 200: groups = resp.json().get("groups", []) or resp.json().get("group", []) for g in groups: if g.get("name") == SD_INSTRUCT_GROUP_NAME: group_id = g.get("id") logger.info(f"SD group found: name={SD_INSTRUCT_GROUP_NAME}, id={group_id}") break if group_id: break except Exception as e: logger.warning(f"Failed to fetch SD groups: {e}") # 3. Поиск ID техника tech_id = None try: url_techs = f"{SD_URL.rstrip('/')}/api/v3/technicians" headers = {"TECHNICIAN_KEY": SD_TOKEN} async with httpx.AsyncClient(verify=False) as client: resp = await client.get(url_techs, headers=headers, timeout=10.0) if resp.status_code == 200: techs = resp.json().get("technicians", []) for t in techs: if t.get("display_name") == SD_INSTRUCT_TECHNICIAN_NAME: tech_id = t.get("id") logger.info(f"SD technician found: name={SD_INSTRUCT_TECHNICIAN_NAME}, id={tech_id}") break except Exception as e: logger.warning(f"Failed to fetch SD technicians: {e}") # 4. Формируем тело запроса payload = { "request": { "subject": subject, "description": description, "requester": requester_payload, "udf_fields": { "udf_pick_301": city } } } if group_id: payload["request"]["group"] = {"id": group_id} if tech_id: payload["request"]["assigned_to"] = {"id": tech_id} 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 # Fallback: Если по ФИО не нашло, пробуем через DEFAULT_REQUESTER elif response.status_code == 400 and requester_payload != {"name": "Искусственный Интеллект"}: logger.warning( f"ServiceDesk не смог привязать requester={requester_payload}. " f"Повторяем запрос от имени системного заявителя..." ) payload["request"]["requester"] = {"name": "Искусственный Интеллект"} payload["request"]["description"] = f"[Пользователь: {requester_cn}]\n\n{description}" data_fb = {"input_data": json.dumps(payload)} resp_fb = await client.post(url, headers=headers, data=data_fb, timeout=15.0) if resp_fb.status_code in (200, 201): ticket_id = resp_fb.json().get("request", {}).get("id") return str(ticket_id) if ticket_id else None logger.error(f"Service Desk API Error ({response.status_code}): {response.text}") return None async def _find_existing_trueconf_ticket(cn: str, requester_email: str = None) -> str | None: """ Найти существующую открытую заявку с темой 'Запрос доступа к Trueconf' для пользователя. Ищет по ФИО (name), затем по email, затем по CN в описании. Статусы: Open (Открыто) и InProgress (В работе). """ subject = "Запрос доступа к Trueconf" try: # Забираем все заявки с темой (без фильтра по статусу — проверяем в коде) url = f"{SD_URL.rstrip('/')}/api/v3/requests?filter=subject:{subject}&fields=id,subject,requester,status" headers = {"TECHNICIAN_KEY": SD_TOKEN} async with httpx.AsyncClient(verify=False) as client: resp = await client.get(url, headers=headers, timeout=10.0) if resp.status_code == 200: data = resp.json() requests_list = data.get("requests", []) logger.info(f"Found {len(requests_list)} tickets with subject='{subject}', searching for CN={cn}") for req in requests_list: req_status = req.get("status", {}) status_name = req_status.get("name", "") if isinstance(req_status, dict) else str(req_status) req_id = req.get("id") req_name = req_status.get("requester", {}).get("name", "") if isinstance(req_status.get("requester"), dict) else "" logger.info(f"Ticket {req_id}: status={status_name}") # Пропускаем закрытые заявки if status_name in ("Resolved", "Rejected", "Closed", "Solved"): continue # Получаем детали заявки для поиска по описанию detail_url = f"{SD_URL.rstrip('/')}/api/v3/requests/{req_id}" detail_resp = await client.get(detail_url, headers=headers, timeout=10.0) if detail_resp.status_code == 200: detail = detail_resp.json() requester = detail.get("requester", {}) req_name = requester.get("name", "") or requester.get("display_name", "") req_email = requester.get("email", "") or requester.get("email_id", "") req_desc = (detail.get("description", "") or "")[:500] logger.info(f"Detail ticket {req_id}: name='{req_name}', email='{req_email}', desc_preview='{req_desc[:100]}'") # 1. Поиск по ФИО в имени заявителя if cn and cn in req_name: logger.info(f"Match ticket {req_id} for CN={cn} (by name: {req_name})") return str(req_id) # 2. Поиск по email if req_email == requester_email and requester_email and requester_email not in (DEFAULT_REQUESTER, "ai@sibcem.ru"): logger.info(f"Match ticket {req_id} for {requester_email} (by email)") return str(req_id) # 3. Поиск по системному заявщику с пометкой пользователя в описании if f"[Пользователь: {cn}]" in req_desc: logger.info(f"Match ticket {req_id} for CN={cn} (by description note)") return str(req_id) # 4. Поиск по CN в описании заявки if cn in req_desc: logger.info(f"Match ticket {req_id} for CN={cn} (CN in description)") return str(req_id) except Exception as e: logger.warning(f"Failed to search existing tickets: {e}") 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) requester_email = (ad_info.get("mail") if ad_info else None) or DEFAULT_REQUESTER city = ad_info.get("city", "Кемерово") if ad_info else "Кемерово" description = ( f"Пользователь {cn} ({user_id}) запросил инструкцию по Trueconf, " f"но не имеет группы 2FA. Требуется выдача доступа." ) try: logger.info(f"Calling SD API directly: cn='{cn}', email='{requester_email}', city='{city}'") ticket_id = await _send_sd_api_request(cn, requester_email, subject, description, city) if ticket_id: logger.info(f"SD ticket created for {cn}: #{ticket_id}") 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 @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") # Получаем 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") if ad_check.get("in_group", False): # Есть доступ — сразу показываем инструкцию set_state(user_id, "INSTRUCT_TRUECONF_VIEW") text = INSTRUCT_TRUECONF_WITH_AD(EMOJI_DIGITS, cn, None) if os.path.isfile(TRUECONF_INSTRUCT_PATH): await msg.bot.send_document( chat_id=msg.chat_id, file=FSInputFile(TRUECONF_INSTRUCT_PATH, file_name="Инструкция.docx"), caption=text, parse_mode="html" ) else: await msg.answer(text, parse_mode="html") else: # Нет доступа — проверяем, есть ли уже открытая заявка ad_info = get_ad_user_info(user_id) requester_email = (ad_info.get("mail") if ad_info else None) or DEFAULT_REQUESTER existing_ticket = await _find_existing_trueconf_ticket(cn, requester_email) if existing_ticket: # Заявка уже существует — показываем информацию set_state(user_id, "INSTRUCT_TRUECONF_VIEW") text = INSTRUCT_TRUECONF_WITH_AD(EMOJI_DIGITS, cn, existing_ticket) await msg.answer(text, parse_mode="html") else: # Заявки нет — показываем промпт с запросом set_state(user_id, "INSTRUCT_TRUECONF_PENDING") text = INSTRUCT_TRUECONF_PENDING_TEXT(EMOJI_DIGITS, cn) await msg.answer(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 == "INSTRUCT_TRUECONF_PENDING": 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") elif cmd == "1": # Запросить доступ cn = get_user_cn(user_id) if cn is None: cn = user_id ticket_id = await create_sd_ticket_for_trueconf(user_id, cn) if ticket_id: text = INSTRUCT_TRUECONF_WITH_AD(EMOJI_DIGITS, cn, ticket_id) await msg.answer(text, parse_mode="html") else: await msg.answer(INSTRUCT_ERROR_MSG, 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")