trueconf_bot/utils/otp_service.py

# /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}")