Обновить instruct/handlers.py — файл с текстом в одном сообщении
This commit is contained in:
Денис Кривоченко
+253
-43
@@ -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. Требуется выдача доступа."
|
||||
)'''
|
||||
|
||||
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")
|
||||
requester_email = DEFAULT_REQUESTER
|
||||
city = "Кемерово"
|
||||
logger.info(f"Using DEFAULT_REQUESTER ({requester_email}) for user {user_id} (cn={cn})")
|
||||
|
||||
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 = (
|
||||
description = (
|
||||
f"Пользователь {cn} ({user_id}) не найден в Active Directory. "
|
||||
f"Требуется ручная проверка и выдача доступа к Trueconf."
|
||||
)'''
|
||||
)
|
||||
|
||||
new3 = ''' description = (
|
||||
f"Пользователь {cn} ({user_id}) отсутствует группа 2FA."
|
||||
)'''
|
||||
|
||||
if old3 in content:
|
||||
content = content.replace(old3, new3)
|
||||
with open(path, "w") as f:
|
||||
f.write(content)
|
||||
print("OK 3 replaced")
|
||||
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:
|
||||
print("OLD 3 NOT FOUND")
|
||||
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")
|
||||
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")
|
||||
Reference in New Issue
Block a user