Initial commit: TrueConf Chatbot КЛЕВЕР
This commit is contained in:
@@ -0,0 +1,133 @@
|
||||
# /opt/trueconf_bot/utils/otp_service.py
|
||||
import re
|
||||
import secrets
|
||||
import logging
|
||||
import httpx
|
||||
import urllib3
|
||||
from ldap3 import Server, Connection, ALL
|
||||
from ldap3.utils.conv import escape_filter_chars
|
||||
import config.config as config
|
||||
|
||||
# Отключаем во внутреннем контуре предупреждения об SSL
|
||||
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def generate_otp_code(length: int = 4) -> str:
|
||||
"""Генерирует криптографически безопасный цифровой код"""
|
||||
return "".join(secrets.choice("0123456789") for _ in range(length))
|
||||
|
||||
async def send_otp_via_api(phone: str, code: str) -> tuple:
|
||||
"""
|
||||
Асинхронно отправляет код на шлюз apisms.
|
||||
Парсит XML-ответ и определяет фактический канал доставки (MAX, Telegram, СМС).
|
||||
Возвращает: (True/False, "канал_доставки или текст_ошибки")
|
||||
"""
|
||||
url = "https://mobileapp.sibcem.ru/MobDriveService/MobDriveService/MobService.asmx/SendSMS"
|
||||
params = {
|
||||
"_number": phone,
|
||||
"_message": f"Код авторизации Сибцем : {code}"
|
||||
}
|
||||
try:
|
||||
async with httpx.AsyncClient(verify=False) as client:
|
||||
response = await client.get(url, params=params, timeout=40.0)
|
||||
|
||||
if response.status_code in [200, 201]:
|
||||
xml_content = response.text
|
||||
logger.info(f"Сырой ответ OTP-шлюза: {xml_content}")
|
||||
|
||||
# Извлекаем текст внутри тега <string>...</string>
|
||||
match = re.search(r'<string[^>]*>(.*?)</string>', xml_content, re.DOTALL)
|
||||
if match:
|
||||
inner_text = match.group(1).strip()
|
||||
inner_lower = inner_text.lower()
|
||||
|
||||
# Умное определение канала на основе ключевых слов шлюза Сибцема
|
||||
if "max" in inner_lower:
|
||||
channel_info = "в MAX"
|
||||
elif "telegram" in inner_lower or "телеграм" in inner_lower:
|
||||
channel_info = "в Telegram"
|
||||
elif "смс" in inner_lower or "sms" in inner_lower or "yeastar" in inner_lower or "шлюз" in inner_lower:
|
||||
channel_info = "по СМС"
|
||||
else:
|
||||
channel_info = "по СМС" # Дефолтный фоллбэк, если текст изменится
|
||||
else:
|
||||
channel_info = "по СМС"
|
||||
|
||||
return True, channel_info
|
||||
|
||||
logger.error(f"❌ Шлюз API вернул ошибку {response.status_code}: {response.text}")
|
||||
return False, f"Ошибка сервера шлюза ({response.status_code})"
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"❌ Критический сбой сети при отправке OTP на {phone}: {e}")
|
||||
return False, str(e)
|
||||
|
||||
def get_card_id_from_ad(user_id: str) -> str:
|
||||
"""Синхронно вытягивает CARD_ID сотрудника из Active Directory (extensionAttribute2)"""
|
||||
try:
|
||||
server = Server(config.AD_SERVER, get_info=ALL)
|
||||
conn = Connection(server, user=config.AD_USER, password=config.AD_PASSWORD, auto_bind=True)
|
||||
safe_user_id = escape_filter_chars(user_id)
|
||||
short_username = user_id.split('@')[0]
|
||||
|
||||
search_filter = f"(&(objectClass=user)(|(mail={safe_user_id})(userPrincipalName={safe_user_id})(userPrincipalName={escape_filter_chars(f'{short_username}@sibcem.ru')})(sAMAccountName={escape_filter_chars(short_username)})))"
|
||||
conn.search(config.AD_BASE, search_filter, attributes=['extensionAttribute2'])
|
||||
|
||||
if conn.entries and 'extensionAttribute2' in conn.entries[0] and conn.entries[0].extensionAttribute2.value:
|
||||
return str(conn.entries[0].extensionAttribute2.value).strip()
|
||||
raise Exception("Поле extensionAttribute2 не заполнено в AD.")
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка получения CARD_ID из AD для {user_id}: {e}")
|
||||
raise Exception(f"Не удалось извлечь CARD_ID из AD: {e}")
|
||||
|
||||
def get_phone_from_db(card_id: str) -> str:
|
||||
"""Синхронно вытягивает телефон сотрудника из БД BossCopy (таблица UOV_SELFSERVICE_PR_CARD) по CARD_ID"""
|
||||
import pymssql
|
||||
try:
|
||||
conn = pymssql.connect(
|
||||
server=config.SQL_SERVER,
|
||||
user=config.DB_USER,
|
||||
password=config.DB_PASSWORD,
|
||||
database=config.SQL_DB_NAME,
|
||||
charset="cp1251",
|
||||
)
|
||||
cursor = conn.cursor()
|
||||
|
||||
# Телефон берётся из поля PHONE таблицы UOV_SELFSERVICE_PR_CARD
|
||||
query = """
|
||||
SELECT PHONE FROM UOV_SELFSERVICE_PR_CARD WHERE ID = %s
|
||||
"""
|
||||
cursor.execute(query, (card_id,))
|
||||
row = cursor.fetchone()
|
||||
conn.close()
|
||||
|
||||
if not row or not row[0]:
|
||||
raise Exception(f"Телефон не найден для CARD_ID {card_id} в UOV_SELFSERVICE_PR_CARD")
|
||||
|
||||
phone = str(row[0]).strip()
|
||||
# Очищаем номер от пробелов, дефисов, скобок
|
||||
phone = phone.replace(" ", "").replace("-", "").replace("(", "").replace(")", "")
|
||||
return phone
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка получения телефона из БД для CARD_ID {card_id}: {e}")
|
||||
raise Exception(f"Не удалось извлечь номер телефона из БД BossCopy: {e}")
|
||||
|
||||
def get_user_phone_from_ad(user_id: str) -> str:
|
||||
"""
|
||||
Синхронно вытягивает телефон сотрудника:
|
||||
1. Получает CARD_ID из Active Directory (extensionAttribute2)
|
||||
2. Получает телефон из БД BossCopy (таблица UOV_SELFSERVICE_PR_CARD) по CARD_ID
|
||||
"""
|
||||
try:
|
||||
# Шаг 1: Получаем CARD_ID из AD
|
||||
card_id = get_card_id_from_ad(user_id)
|
||||
logger.info(f"✅ CARD_ID для {user_id}: {card_id}")
|
||||
|
||||
# Шаг 2: Получаем телефон из БД по CARD_ID
|
||||
phone = get_phone_from_db(card_id)
|
||||
logger.info(f"✅ Телефон для CARD_ID {card_id}: {phone}")
|
||||
|
||||
return phone
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка получения телефона для {user_id}: {e}")
|
||||
raise Exception(f"Не удалось извлечь номер телефона: {e}")
|
||||
Reference in New Issue
Block a user