dddennnisss/trueconf_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Обновить instruct/handlers.py

Browse Source
This commit is contained in:
dddennnisss
2026-07-30 12:03:33 +07:00
parent 8aea92b84f
commit 8bb0e27028
1 changed files with 155 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
instruct/handlers.py
+155 -112
View File
@@ -4,6 +4,9 @@ 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
@@ -23,7 +26,11 @@ 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
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")
@@ -35,23 +42,79 @@ except ImportError:
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
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"""
<html>
<body style="font-family: Arial, sans-serif; color: #333;">
<h2>🎫 Создана новая заявка на доступ к Trueconf (2FA)</h2>
<p><b>Сотрудник:</b> {cn}</p>
<p><b>Логин / Email:</b> {clean_user_id}</p>
<p><b>Номер заявки в ServiceDesk:</b> <a href="{ticket_link}">#{ticket_id}</a></p>
<p><b>Назначенная группа:</b> {SD_INSTRUCT_GROUP_NAME}</p>
<p><b>Исполнитель:</b> {SD_INSTRUCT_TECHNICIAN_NAME}</p>
<hr style="border: 0; border-top: 1px solid #ccc;">
<p style="font-size: 12px; color: #777;"><i>Сообщение сгенерировано автоматически цифровым ассистентом КЛЕВЕР.</i></p>
</body>
</html>
"""
# Запускаем отправку в отдельном потоке, чтобы не блокировать 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 (email или логин).
user_id из TrueConf — это обычно email, например 'ds.krivochenko@tcs.sibcem.ru'
"""
"""Получить CN пользователя из AD по его user_id."""
from utils.ad_search import search_by_user_id
try:
@@ -78,10 +141,7 @@ def get_user_cn(user_id: str) -> str | None:
def get_ad_user_info(user_id: str) -> dict | None:
"""
Получить полную информацию о пользователе из AD.
Возвращает {cn, mail, city} или None.
"""
"""Получить полную информацию о пользователе из AD."""
from utils.ad_search import search_by_user_id
try:
@@ -92,9 +152,7 @@ def get_ad_user_info(user_id: str) -> dict | None:
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", ""):
@@ -113,11 +171,7 @@ async def _send_sd_api_request(
description: str,
city: str
) -> str | None:
"""
Автономная прямая отправка HTTP-запроса в Service Desk API v3.
Умеет привязывать пользователя по ФИО, Email или Fallback.
"""
# 1. Формируем структуру заявителя (приоритет: ФИО -> Email -> ИИ)
"""Автономная прямая отправка 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"):
@@ -125,44 +179,6 @@ async def _send_sd_api_request(
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,
@@ -173,10 +189,12 @@ async def _send_sd_api_request(
}
}
}
if group_id:
payload["request"]["group"] = {"id": group_id}
if tech_id:
payload["request"]["assigned_to"] = {"id": tech_id}
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}
@@ -190,7 +208,6 @@ async def _send_sd_api_request(
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}. "
@@ -210,83 +227,111 @@ async def _send_sd_api_request(
async def _find_existing_trueconf_ticket(cn: str, requester_email: str = None) -> str | None:
"""
Найти существующую открытую заявку с темой 'Запрос доступа к Trueconf' для пользователя.
Ищет по ФИО (name), затем по email, затем по CN в описании.
Статусы: Open (Открыто) и InProgress (В работе).
"""
subject = "Запрос доступа к Trueconf"
"""Находит открытую или находящуюся в работе заявку по Trueconf."""
subject_keyword = "запрос доступа к trueconf"
CLOSED_STATUSES = {
"resolved", "rejected", "closed", "solved", "canceled", "cancelled",
"решена", "закрыта", "отклонена", "выполнена", "отменена", "закрыто", "решено"
}
try:
# Забираем все заявки без фильтра — API v3 некорректно работает с русскими буквами в filter
url = f"{SD_URL.rstrip('/')}/api/v3/requests?fields=id,subject,requester,status"
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, timeout=10.0)
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"Found {len(requests_list)} total tickets, searching for CN={cn}")
logger.info(f"Scanning {len(requests_list)} recent tickets for CN='{cn}', Email='{requester_email}'")
for req in requests_list:
# Фильтруем по subject в коде
if subject not in req.get("subject", ""):
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)
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 ("Resolved", "Rejected", "Closed", "Solved"):
logger.info(f"Skipping ticket {req_id}: status={status_name}")
if status_name in CLOSED_STATUSES:
logger.info(f"Skipping closed ticket #{req_id} (status: '{status_name}')")
continue
logger.info(f"Checking ticket {req_id}: status={status_name}")
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()
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})")
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)
# 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)")
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)
# 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)")
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}")
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.
"""
"""Создать заявку в Service Desk для запроса доступа к Trueconf."""
subject = "Запрос доступа к Trueconf"
# Получаем данные пользователя из AD
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} ({user_id}) запросил инструкцию по Trueconf, "
f"Пользователь {cn} ({clean_user_id}) запросил инструкцию по Trueconf, "
f"но не имеет группы 2FA. Требуется выдача доступа."
)
@@ -295,6 +340,8 @@ async def create_sd_ticket_for_trueconf(user_id: str, cn: str) -> str | None:
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
@@ -341,17 +388,14 @@ async def instruct_router_handler(msg: Message):
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
cn = user_id
# Проверяем наличие группы 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):
@@ -364,17 +408,15 @@ async def instruct_router_handler(msg: Message):
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")
@@ -395,12 +437,13 @@ async def instruct_router_handler(msg: Message):
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: