Денис Кривоченко
410 lines
21 KiB
Python
410 lines
21 KiB
Python
import logging
|
|
from logging.handlers import RotatingFileHandler
|
|
import os
|
|
import json
|
|
import asyncio
|
|
import httpx
|
|
import re
|
|
import smtplib
|
|
import urllib3
|
|
import importlib.util
|
|
from email.message import EmailMessage
|
|
from ldap3 import Server, Connection, ALL
|
|
from trueconf import Router, Message
|
|
|
|
# --- SD WORKFLOW LOGGING SETUP ---
|
|
log_dir = os.path.join(os.path.dirname(__file__), 'logs')
|
|
os.makedirs(log_dir, exist_ok=True)
|
|
sd_workflow_logger = logging.getLogger('sd_workflow')
|
|
sd_workflow_logger.setLevel(logging.INFO)
|
|
if not sd_workflow_logger.handlers:
|
|
handler = RotatingFileHandler(os.path.join(log_dir, 'sd_workflow.log'), maxBytes=10*1024*1024, backupCount=5)
|
|
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
|
|
handler.setFormatter(formatter)
|
|
sd_workflow_logger.addHandler(handler)
|
|
# ----------------------------------
|
|
|
|
# Абсолютный импорт конфигурации
|
|
spec = importlib.util.spec_from_file_location("custom_config", os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config", "config.py"))
|
|
custom_config = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(custom_config)
|
|
globals().update(vars(custom_config))
|
|
|
|
from utils import states
|
|
from utils.menu import MENU_TEXT
|
|
from utils.texts import (
|
|
EMOJI_DIGITS,
|
|
SD_MAIN_MENU_TEXT,
|
|
SD_UNKNOWN_CMD_TEXT,
|
|
SD_TEXT_REQUIRED,
|
|
SD_CREATING,
|
|
sd_ticket_created,
|
|
sd_ticket_create_error,
|
|
sd_system_error,
|
|
sd_uploading_file,
|
|
sd_file_attached,
|
|
sd_file_upload_error,
|
|
)
|
|
from utils.texts import UNKNOWN_MAIN_CMD_TEXT
|
|
# 🔌 Импортируем централизованную функцию сбора статистики из main
|
|
from utils.stats_logger import log_menu_stats
|
|
|
|
# Таймаут ожидания вложений после текста (секунды)
|
|
ATTACHMENT_WAIT_TIMEOUT = 2.0
|
|
|
|
SD_DISPATCHER_SYSTEM_PROMPT = (
|
|
"Ты — старший ИИ-диспетчер Service Desk холдинга. Твоя задача — сформулировать краткую и понятную тему ИТ-заявки по-русски (от 3 до 7 слов) на основе текста пользователя.\n"
|
|
"ПРАВИЛА:\n"
|
|
"1. Назови проблему и оборудование/программу (например: 'Самопроизвольное выключение ПК', 'Неисправность принтера', 'Сбой авторизации в почте').\n"
|
|
"2. Отвечать на английском языке ЗАПРЕЩЕНО.\n"
|
|
"3. Тебе категорически запрещено решать проблему или писать мануалы по настройке.\n"
|
|
"4. Не пиши префиксы 'Тема:', 'Заголовок:' в ответе.\n"
|
|
"5. Если в тексте только приветствие или мат — выведи ровно одно слово: СПАМ.")
|
|
|
|
# Глушим системные предупреждения
|
|
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
|
logger = logging.getLogger(__name__)
|
|
router = Router()
|
|
sd_sessions = {}
|
|
|
|
# ================================================
|
|
# БИЗНЕС-ЛОГИКА
|
|
# ================================================
|
|
def get_ad_user_sync(login: str):
|
|
try:
|
|
from utils.ad_search import search_by_login
|
|
entries = search_by_login(login, ["displayName", "mail", "l", "userAccountControl"])
|
|
if entries:
|
|
user = entries[0]
|
|
uac = user.userAccountControl.value if 'userAccountControl' in user else 0
|
|
return {
|
|
"name": user.displayName.value if 'displayName' in user else login,
|
|
"mail": user.mail.value if 'mail' in user else None,
|
|
"city": user.l.value if 'l' in user else "Кемерово",
|
|
"is_disabled": bool(uac & 2)
|
|
}
|
|
except Exception as e:
|
|
logger.error(f"Ошибка LDAP: {e}")
|
|
return None
|
|
|
|
async def generate_smart_subject(text: str) -> str:
|
|
import re, traceback, importlib.util, httpx
|
|
spec = importlib.util.spec_from_file_location("custom_config", os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "config", "config.py"))
|
|
custom_config = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(custom_config)
|
|
OLLAMA_SUBJECT_URL = custom_config.OLLAMA_SUBJECT_URL
|
|
OLLAMA_SUBJECT_MODEL = custom_config.OLLAMA_SUBJECT_MODEL
|
|
def has_chinese(t): return bool(re.search(r'[\u4e00-\u9fff]', str(t)))
|
|
if not text or len(text.split()) < 2: return "Заявка из КЛЕВЕР"
|
|
|
|
user_msg = f"Сформулируй краткую тему для следующей ИТ-заявки:\n{text}"
|
|
|
|
messages = [
|
|
{"role": "system", "content": SD_DISPATCHER_SYSTEM_PROMPT},
|
|
{"role": "user", "content": user_msg}
|
|
]
|
|
|
|
try:
|
|
payload = {"model": OLLAMA_SUBJECT_MODEL, "messages": messages, "temperature": 0.0}
|
|
sd_workflow_logger.info(f"📝 [Subject Gen] Payload: {payload}")
|
|
endpoint = OLLAMA_SUBJECT_URL.replace("/api/generate", "/v1/chat/completions")
|
|
async with httpx.AsyncClient() as client:
|
|
response = await client.post(endpoint, json=payload, timeout=120.0)
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
if "choices" in data and len(data["choices"]) > 0:
|
|
res = data["choices"][0]["message"]["content"].strip(" .'\"`")
|
|
res = re.sub(r'^(тема|заголовок|subject):\s*', '', res, flags=re.IGNORECASE).strip()
|
|
if res and not has_chinese(res) and not any(m in res.upper() for m in ["СПАМ", "НЕ ПОНЯЛ", "Я ИСКУССТВЕННЫЙ", "Я МОДЕЛЬ"]):
|
|
return res[0].upper() + res[1:] if len(res) > 0 else "Заявка из КЛЕВЕР"
|
|
else:
|
|
sd_workflow_logger.info(f"📝 [Subject Gen Error] Validation Failed: Received '{res}'")
|
|
else:
|
|
sd_workflow_logger.info(f"📝 [Subject Gen Error] Empty choices in LLM response")
|
|
else:
|
|
sd_workflow_logger.info(f"📝 [Subject Gen Error] API Error: Status {response.status_code} - {response.text}")
|
|
except Exception as e:
|
|
sd_workflow_logger.error(f"📝 [Subject Gen Error] Exception: {e}\n{traceback.format_exc()}")
|
|
return "Заявка из КЛЕВЕР"
|
|
|
|
async def create_ticket_in_sd(requester_email: str, subject: str, description: str, city: str):
|
|
headers = {"authtoken": SD_TOKEN, "Accept": "application/vnd.manageengine.sdp.v3+json"}
|
|
endpoint = f"{SD_URL}/api/v3/requests"
|
|
html_desc = f"<p>{description.replace(chr(10), '<br>')}</p><br><hr><p style='color:#555;font-size:12px;'><i>Создано через TrueConf КЛЕВЕР</i></p>"
|
|
async def _send_req(email, current_city):
|
|
payload = {"request": {"subject": subject, "description": html_desc, "requester": {"email_id": email}, "udf_fields": {"udf_pick_301": current_city}}}
|
|
async with httpx.AsyncClient(verify=False) as client:
|
|
return await client.post(endpoint, headers=headers, data={"input_data": json.dumps(payload)}, timeout=15.0)
|
|
resp = await _send_req(requester_email, city)
|
|
data = resp.json() if resp.status_code in [200, 201] else {}
|
|
if data.get("response_status", {}).get("status_code") != 2000:
|
|
resp = await _send_req(requester_email, "Кемерово")
|
|
data = resp.json() if resp.status_code in [200, 201] else {}
|
|
if data.get("response_status", {}).get("status_code") == 2000:
|
|
return data.get("request", {}).get("id")
|
|
return None
|
|
|
|
async def process_and_upload_file(file_id: str, filename: str, ticket_id: str):
|
|
import io
|
|
downloaded_path = None
|
|
try:
|
|
if isinstance(filename, bytes) or (isinstance(filename, str) and filename.startswith("b'")):
|
|
safe_filename = f"attachment_{ticket_id}.dat"
|
|
else:
|
|
safe_filename = str(filename)
|
|
downloaded_file = await states.tc_bot.download_file_by_id(file_id)
|
|
if isinstance(downloaded_file, bytes):
|
|
file_obj = io.BytesIO(downloaded_file)
|
|
downloaded_path = None
|
|
else:
|
|
downloaded_path = str(downloaded_file)
|
|
file_obj = open(downloaded_path, 'rb')
|
|
try:
|
|
upload_url = f"{SD_URL}/api/v3/requests/{ticket_id}/upload"
|
|
async with httpx.AsyncClient(verify=False) as client:
|
|
attach_resp = await client.put(
|
|
upload_url,
|
|
headers={"authtoken": SD_TOKEN},
|
|
files={'input_file': (safe_filename, file_obj, 'application/octet-stream')},
|
|
timeout=30.0
|
|
)
|
|
return attach_resp.status_code in [200, 201]
|
|
finally:
|
|
if not isinstance(downloaded_file, bytes):
|
|
file_obj.close()
|
|
except Exception as e:
|
|
logger.error(f"Ошибка загрузки файла {filename}: {e}")
|
|
return False
|
|
finally:
|
|
if downloaded_path and os.path.exists(downloaded_path):
|
|
os.remove(downloaded_path)
|
|
|
|
def _extract_text_from_content(content):
|
|
"""Извлечение текста из msg.content."""
|
|
if isinstance(content, dict):
|
|
return content.get("text", "").strip()
|
|
elif hasattr(content, "text"):
|
|
return str(content.text).strip()
|
|
return ""
|
|
|
|
def _extract_attachments_from_content(content, is_attachment_type=False):
|
|
"""Извлечение file_id и file_name из msg.content."""
|
|
attachments = []
|
|
if isinstance(content, dict):
|
|
atts = content.get("attachments") or content.get("files")
|
|
single_fid = content.get("file_id")
|
|
single_fname = content.get("file_name")
|
|
if atts:
|
|
if isinstance(atts, list):
|
|
for a in atts:
|
|
if isinstance(a, dict):
|
|
fid = a.get("file_id") or a.get("id")
|
|
fname = a.get("file_name") or a.get("name", "attachment")
|
|
if fid: attachments.append({"file_id": fid, "file_name": fname})
|
|
elif isinstance(a, str):
|
|
attachments.append({"file_id": a, "file_name": "attachment"})
|
|
elif isinstance(atts, dict):
|
|
fid = atts.get("file_id") or atts.get("id")
|
|
fname = atts.get("file_name") or atts.get("name", "attachment")
|
|
if fid: attachments.append({"file_id": fid, "file_name": fname})
|
|
if single_fid:
|
|
attachments.append({"file_id": single_fid, "file_name": single_fname or "attachment"})
|
|
elif is_attachment_type:
|
|
if hasattr(content, "file_id"):
|
|
fid = content.file_id
|
|
fname = getattr(content, "file_name", "attachment")
|
|
attachments.append({"file_id": fid, "file_name": fname})
|
|
return attachments
|
|
|
|
async def _create_ticket_and_attach_files(user_id, msg_text, session, msg, login):
|
|
"""Логика генерации темы, отправки в SD и загрузки всех очередей вложений."""
|
|
session["step"] = "creating_ticket"
|
|
try:
|
|
ad_user = await asyncio.to_thread(get_ad_user_sync, login)
|
|
sd_workflow_logger.info(f"👤 [AD Lookup] User: {login} -> Found: {ad_user is not None}")
|
|
sender_email = ad_user.get("mail") if ad_user else DEFAULT_REQUESTER
|
|
city = ad_user.get("city") if ad_user else "Кемерово"
|
|
subject = await generate_smart_subject(msg_text)
|
|
sd_workflow_logger.info(f"📝 [Subject Gen] Text: {msg_text[:50]}... -> Subject: {subject}")
|
|
|
|
ticket_id = await create_ticket_in_sd(sender_email, subject, msg_text, city)
|
|
sd_workflow_logger.info(f"🎫 [Ticket Created] ID: {ticket_id}")
|
|
|
|
if ticket_id:
|
|
session["ticket_id"] = ticket_id
|
|
|
|
# 1. Загрузка основных вложений (накопленных до и во время таймаута)
|
|
queued_files = list(session.get("files_queue", []))
|
|
if queued_files:
|
|
sd_workflow_logger.info(f"📎 [Upload] Uploading {len(queued_files)} file(s) to ticket #{ticket_id}")
|
|
for f in queued_files:
|
|
await process_and_upload_file(f['file_id'], f['file_name'], ticket_id)
|
|
sd_workflow_logger.info(f"✅ [Upload] All queued file(s) uploaded")
|
|
|
|
# 2. Загрузка вложений, прилетевших во время выполнения API-запросов (из "черной дыры")
|
|
post_files = list(session.get("post_create_queue", []))
|
|
if post_files:
|
|
sd_workflow_logger.info(f"📎 [Upload Extra] Uploading {len(post_files)} late file(s) to ticket #{ticket_id}")
|
|
for f in post_files:
|
|
await process_and_upload_file(f['file_id'], f['file_name'], ticket_id)
|
|
|
|
session["files_queue"] = []
|
|
session["post_create_queue"] = []
|
|
session["step"] = "ticket_created"
|
|
|
|
await msg.answer(
|
|
sd_ticket_created(ticket_id, subject, msg_text),
|
|
parse_mode="html"
|
|
)
|
|
log_menu_stats(user_id, "Service Desk", f"Создание заявки #{ticket_id}")
|
|
else:
|
|
session["step"] = "need_text"
|
|
await msg.answer(system_error_text(EMOJI_DIGITS), parse_mode="html")
|
|
except Exception as e:
|
|
logger.exception(f"Ошибка: {e}")
|
|
sd_workflow_logger.error(f"❌ [SD Error] {e}")
|
|
session["step"] = "need_text"
|
|
await msg.answer(system_error_text(EMOJI_DIGITS), parse_mode="html")
|
|
|
|
async def _wait_for_attachments_and_create(user_id, session):
|
|
"""Фоновый таймер с возможностью отмены (Debounce)."""
|
|
try:
|
|
await asyncio.sleep(ATTACHMENT_WAIT_TIMEOUT)
|
|
except asyncio.CancelledError:
|
|
return # Таймер сброшен, прерываем выполнение текущей таски
|
|
|
|
if session.get("step") != "waiting_for_attachments":
|
|
return
|
|
|
|
sd_workflow_logger.info(f"⏰ [SD] Timeout reached. Initiating ticket creation for user {user_id}")
|
|
msg = session.get("msg")
|
|
login = session.get("login")
|
|
if msg:
|
|
await _create_ticket_and_attach_files(user_id, session["msg_text"], session, msg, login)
|
|
|
|
@router.message()
|
|
async def sd_module_handler(msg: Message):
|
|
try:
|
|
user_id = msg.from_user.id
|
|
sd_workflow_logger.info(f"🚀 [SD Workflow Start] User: {user_id}")
|
|
if states.get_state(user_id) != "SD_MODE":
|
|
return
|
|
|
|
msg_text = _extract_text_from_content(msg.content)
|
|
is_attachment = hasattr(msg.type, "name") and msg.type.name == "ATTACHMENT"
|
|
inline_attachments = _extract_attachments_from_content(msg.content, is_attachment)
|
|
|
|
sd_workflow_logger.info(f"🚀 [SD Debug] msg.type={msg.type}, content_type={type(msg.content).__name__}, msg_text='{msg_text}', is_attachment={is_attachment}, inline_attachments={len(inline_attachments)}")
|
|
|
|
msg.handled = True
|
|
login = user_id.split("@")[0] if "@" in user_id else user_id
|
|
cmd = msg_text.lower()
|
|
|
|
# 🔄 НОРМАЛИЗАЦИЯ КНОПОК ВК-ЭМОДЗИ
|
|
for raw_num, emoji_num in EMOJI_DIGITS.items():
|
|
if cmd == emoji_num:
|
|
cmd = raw_num
|
|
break
|
|
|
|
# --- ВЫХОД (Кнопки 0 и 9 возвращают в общее меню) ---
|
|
if cmd in ["0", "9", "/start", "меню"]:
|
|
log_menu_stats(user_id, "Service Desk", "Выход в главное меню")
|
|
states.clear_state(user_id)
|
|
sd_sessions.pop(user_id, None)
|
|
await msg.answer(MENU_TEXT, parse_mode="html")
|
|
return
|
|
|
|
if user_id not in sd_sessions:
|
|
sd_sessions[user_id] = {"step": "need_text", "files_queue": [], "post_create_queue": []}
|
|
session = sd_sessions[user_id]
|
|
|
|
# --- ШАГ 1: ОЖИДАНИЕ ТЕКСТА ---
|
|
if session["step"] == "need_text":
|
|
if not msg_text.strip() and not inline_attachments:
|
|
await msg.answer(SD_UNKNOWN_CMD_TEXT, parse_mode="html")
|
|
return
|
|
|
|
# Если только вложение без текста — сохраняем и ждем текст
|
|
if not msg_text.strip() and inline_attachments:
|
|
session["files_queue"].extend(inline_attachments)
|
|
sd_workflow_logger.info(f"📎 [SD] Saved {len(inline_attachments)} attachment(s), waiting for text")
|
|
await msg.answer(SD_TEXT_REQUIRED, parse_mode="html")
|
|
return
|
|
|
|
# Если пришел текст (с вложениями или без)
|
|
if inline_attachments:
|
|
session["files_queue"].extend(inline_attachments)
|
|
sd_workflow_logger.info(f"📎 [SD] Added {len(inline_attachments)} inline attachment(s) to queue")
|
|
|
|
# Настраиваем параметры сессии для ожидания
|
|
session["step"] = "waiting_for_attachments"
|
|
session["msg_text"] = msg_text
|
|
session["msg"] = msg
|
|
session["user_id"] = user_id
|
|
session["login"] = login
|
|
|
|
# Сразу выдаем ОДНО сообщение пользователю, чтобы он видел реакцию бота
|
|
await msg.answer(SD_CREATING, parse_mode="html")
|
|
|
|
# Запускаем фоновый таймер с механизмом сброса (Debounce)
|
|
if "timer_task" in session and not session["timer_task"].done():
|
|
session["timer_task"].cancel()
|
|
session["timer_task"] = asyncio.create_task(_wait_for_attachments_and_create(user_id, session))
|
|
session["timer_task"].add_done_callback(lambda t: _on_sd_timer_done(t, user_id))
|
|
sd_workflow_logger.info(f"⏳ [SD] Timer started: waiting {ATTACHMENT_WAIT_TIMEOUT}s for potentially more attachments...")
|
|
return
|
|
|
|
# --- ШАГ 2: РЕЖИМ ОЖИДАНИЯ ДОП. ВЛОЖЕНИЙ (Сброс таймаута) ---
|
|
if session["step"] == "waiting_for_attachments":
|
|
if inline_attachments:
|
|
session["files_queue"].extend(inline_attachments)
|
|
sd_workflow_logger.info(f"📎 [SD] Received {len(inline_attachments)} more files. Total queue: {len(session['files_queue'])}")
|
|
if msg_text.strip():
|
|
session["msg_text"] = session["msg_text"] + " " + msg_text.strip()
|
|
sd_workflow_logger.info(f"📝 [SD] Additional text appended")
|
|
|
|
# Перезапускаем таймер (пользователь активен, сдвигаем окно создания вперед)
|
|
if "timer_task" in session and not session["timer_task"].done():
|
|
session["timer_task"].cancel()
|
|
session["timer_task"] = asyncio.create_task(_wait_for_attachments_and_create(user_id, session))
|
|
session["timer_task"].add_done_callback(lambda t: _on_sd_timer_done(t, user_id))
|
|
sd_workflow_logger.info(f"🔄 [SD] Timer reset due to user activity. Waiting another {ATTACHMENT_WAIT_TIMEOUT}s")
|
|
return
|
|
|
|
# --- ШАГ 3: ЗАЯВКА СОЗДАЕТСЯ (Перехват файлов из "черной дыры") ---
|
|
if session["step"] == "creating_ticket":
|
|
if inline_attachments:
|
|
if "post_create_queue" not in session:
|
|
session["post_create_queue"] = []
|
|
session["post_create_queue"].extend(inline_attachments)
|
|
sd_workflow_logger.info(f"📥 [SD Hole-Fix] Captured {len(inline_attachments)} file(s) DURING ticket creation API call.")
|
|
return
|
|
|
|
# --- ШАГ 4: ПРИЕМ ДОП. ФАЙЛОВ К УЖЕ СОЗДАННОЙ ЗАЯВКЕ ---
|
|
if session["step"] == "ticket_created":
|
|
if inline_attachments:
|
|
ticket_id = session["ticket_id"]
|
|
for att in inline_attachments:
|
|
await msg.answer(sd_uploading_file(ticket_id), parse_mode="html")
|
|
ok = await process_and_upload_file(att['file_id'], att['file_name'], ticket_id)
|
|
sd_workflow_logger.info(f"📎 [Late File Upload] Name: {att['file_name']} -> Success: {ok}")
|
|
if ok:
|
|
log_menu_stats(user_id, "Service Desk", f"Добавление файла к заявке #{ticket_id}")
|
|
await msg.answer(sd_file_attached(), parse_mode="html")
|
|
else:
|
|
await msg.answer(system_error_text(EMOJI_DIGITS), parse_mode="html")
|
|
elif msg_text.strip():
|
|
await msg.answer(UNKNOWN_MAIN_CMD_TEXT, parse_mode="html")
|
|
return
|
|
except Exception as e:
|
|
logger.exception(f"❌ [SD Fatal Error] user={getattr(msg.from_user, 'id', 'unknown')}: {e}")
|
|
sd_workflow_logger.error(f"❌ [SD Fatal] {e}")
|
|
try:
|
|
await msg.answer(system_error_text(EMOJI_DIGITS), parse_mode="html")
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
|
|
|