import logging import asyncio import os import sys import json import httpx import smtplib from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart 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_AVAILABLE, INSTRUCT_EMAIL_UNAVAILABLE, 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, SMTP_SERVER, SMTP_PORT, DEFAULT_EMAIL_FROM, ALERTS_SUPPORT_EMAILS ) 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 SMTP_SERVER = getattr(custom_config, "SMTP_SERVER", "srvkem-mail.sibcem.ru") SMTP_PORT = getattr(custom_config, "SMTP_PORT", 25) DEFAULT_EMAIL_FROM = getattr(custom_config, "DEFAULT_EMAIL_FROM", "ai@sibcem.ru") ALERTS_SUPPORT_EMAILS = getattr(custom_config, "ALERTS_SUPPORT_EMAILS", ["ds.krivochenko@sibcem.ru"]) logger = logging.getLogger(__name__) router = Router() TRUECONF_INSTRUCT_PATH = os.path.join( os.path.dirname(__file__), "Проверка_наличия_и_авторизация_на_мобильном_устройстве_Trueconf.docx" ) # ВСПОМОГАТЕЛЬНАЯ ФУНКЦИЯ ОТПРАВКИ EMAIL УВЕДОМЛЕНИЙ def _send_email_sync(to_emails: list[str], subject: str, body_html: str): """Синхронная отправка письма по SMTP""" try: msg = MIMEMultipart("alternative") msg["Subject"] = subject msg["From"] = DEFAULT_EMAIL_FROM msg["To"] = ", ".join(to_emails) part_html = MIMEText(body_html, "html", "utf-8") msg.attach(part_html) with smtplib.SMTP(SMTP_SERVER, SMTP_PORT, timeout=10) as server: server.sendmail(DEFAULT_EMAIL_FROM, to_emails, msg.as_string()) logger.info(f"📧 Уведомление о заявке 2FA успешно отправлено на {to_emails}") except Exception as e: logger.error(f"❌ Ошибка отправки почтового уведомления 2FA: {e}", exc_info=True) async def send_2fa_alert_email(cn: str, user_id: str, ticket_id: str): """Фоновая асинхронная задача формирования и отправки уведомления""" if not ALERTS_SUPPORT_EMAILS: logger.warning("ALERTS_SUPPORT_EMAILS пуст. Уведомление на почту пропущено.") return clean_user_id = user_id.replace("@tcs.sibcem.ru", "@sibcem.ru") subject = f"🎫 [2FA Request] Создана заявка #{ticket_id} на доступ к Trueconf ({cn})" ticket_link = f"{SD_URL.rstrip('/')}/WorkOrder.do?woMode=viewWO&woID={ticket_id}" body_html = f"""

🎫 Создана новая заявка на доступ к Trueconf (2FA)

Сотрудник: {cn}

Логин / Email: {clean_user_id}

Номер заявки в ServiceDesk: #{ticket_id}

Назначенная группа: {SD_INSTRUCT_GROUP_NAME}

Исполнитель: {SD_INSTRUCT_TECHNICIAN_NAME}

Сообщение сгенерировано автоматически цифровым ассистентом КЛЕВЕР.

""" # Запускаем отправку в отдельном потоке, чтобы не блокировать event loop asyncio await asyncio.to_thread(_send_email_sync, ALERTS_SUPPORT_EMAILS, subject, body_html) # ОСНОВНАЯ ЛОГИКА АКТИВНОСТИ def get_user_cn(user_id: str) -> str | None: """Получить CN пользователя из AD по его user_id.""" 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.""" 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 = raw_mail if raw_mail and raw_mail not in ("[]", "N/A", "") else None 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.""" 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": "Искусственный Интеллект"} payload = { "request": { "subject": subject, "description": description, "requester": requester_payload, "udf_fields": { "udf_pick_301": city } } } if SD_INSTRUCT_GROUP_NAME: payload["request"]["group"] = {"name": SD_INSTRUCT_GROUP_NAME} if SD_INSTRUCT_TECHNICIAN_NAME: payload["request"]["technician"] = {"name": SD_INSTRUCT_TECHNICIAN_NAME} 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 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.""" subject_keyword = "запрос доступа к trueconf" CLOSED_STATUSES = { "resolved", "rejected", "closed", "solved", "canceled", "cancelled", "решена", "закрыта", "отклонена", "выполнена", "отменена", "закрыто", "решено" } try: url = f"{SD_URL.rstrip('/')}/api/v3/requests" headers = {"TECHNICIAN_KEY": SD_TOKEN} input_data = { "list_info": { "row_count": 100, "sort_field": "created_time", "sort_order": "desc" } } async with httpx.AsyncClient(verify=False) as client: resp = await client.get(url, headers=headers, params={"input_data": json.dumps(input_data)}, timeout=10.0) if resp.status_code != 200: logger.warning(f"SD GET list_info returned status {resp.status_code}. Retrying pure GET...") 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"Scanning {len(requests_list)} recent tickets for CN='{cn}', Email='{requester_email}'") for req in requests_list: req_subject = str(req.get("subject", "")).lower() if subject_keyword not in req_subject: continue req_status = req.get("status", {}) status_name = ( req_status.get("name", "") if isinstance(req_status, dict) else str(req_status) ).strip().lower() req_id = req.get("id") if status_name in CLOSED_STATUSES: logger.info(f"Skipping closed ticket #{req_id} (status: '{status_name}')") continue logger.info(f"Checking candidate open ticket #{req_id} (status: '{status_name}')") 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().get("request", {}) requester = detail.get("requester", {}) req_name = str(requester.get("name", "") or requester.get("display_name", "")).strip().lower() req_email = str(requester.get("email", "") or requester.get("email_id", "")).strip().lower() req_desc = str(detail.get("description", "") or "").lower() cn_clean = cn.strip().lower() if cn_clean and cn_clean in req_name: logger.info(f"Matched ticket #{req_id} by requester name ('{req_name}')") return str(req_id) if ( requester_email and requester_email.lower() not in (DEFAULT_REQUESTER.lower(), "ai@sibcem.ru") and req_email == requester_email.lower() ): logger.info(f"Matched ticket #{req_id} by email ('{req_email}')") return str(req_id) if cn_clean and (f"[пользователь: {cn_clean}]" in req_desc or cn_clean in req_desc): logger.info(f"Matched ticket #{req_id} by user mention in description") return str(req_id) else: logger.error(f"Failed to fetch SD requests list completely, HTTP {resp.status_code}: {resp.text[:200]}") except Exception as e: logger.warning(f"Failed to search existing tickets: {e}", exc_info=True) return None async def create_sd_ticket_for_trueconf(user_id: str, cn: str) -> str | None: """Создать заявку в Service Desk для запроса доступа к Trueconf.""" subject = "Запрос доступа к Trueconf" clean_user_id = user_id.replace("@tcs.sibcem.ru", "@sibcem.ru") 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 "Кемерово" # Защита от дублей existing_ticket = await _find_existing_trueconf_ticket(cn, requester_email) if existing_ticket: logger.info(f"Ticket creation skipped: open ticket #{existing_ticket} already exists for {cn}") return existing_ticket description = ( f"Пользователь {cn} ({clean_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}") # 📧 ОТПРАВЛЯЕМ ПОЧТОВОЕ УВЕДОМЛЕНИЕ В ФОНЕ asyncio.create_task(send_2fa_alert_email(cn, user_id, 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 = get_user_cn(user_id) if cn is None: cn = user_id 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 logger.info(f"[DEBUG] Checking existing ticket for cn={cn}, email={requester_email}") existing_ticket = await _find_existing_trueconf_ticket(cn, requester_email) logger.info(f"[DEBUG] existing_ticket result: {existing_ticket}") if existing_ticket: set_state(user_id, "INSTRUCT_TRUECONF_VIEW") text = INSTRUCT_TRUECONF_WITH_AD(EMOJI_DIGITS, cn, existing_ticket) 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: 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, "Инструкции", "Почта") cn = get_user_cn(user_id) if cn is None: cn = user_id ad_info = get_ad_user_info(user_id) mail = ad_info.get("mail") if ad_info else None if mail: text = INSTRUCT_EMAIL_AVAILABLE(EMOJI_DIGITS, cn, mail) else: text = INSTRUCT_EMAIL_UNAVAILABLE(EMOJI_DIGITS, cn) await msg.answer(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: set_state(user_id, "INSTRUCT_TRUECONF_VIEW") 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")