add: тестовый скрипт test_ad.py и INSTALL.md для переноса на новый сервер

test_ad.py

@@ -0,0 +1,230 @@
+#!/usr/bin/env python3
+"""
+Тестовый скрипт для проверки подключения к Active Directory.
+Работает автономно - не требует Flask, Qdrant, Ollama, pip, dotenv.
+
+Использование:
+    python3 test_ad.py --login user123
+    python3 test_ad.py --email user123@sibcem.ru
+    python3 test_ad.py --phone 7012345
+"""
+
+import os
+import sys
+import argparse
+import re
+
+
+# ==================== АВТОНОМНЫЙ ЗАГРУЗЧИК .env ====================
+def load_env_file(filepath):
+    """Парсит .env файл без зависимостей (аналог python-dotenv)."""
+    env = {}
+    if not os.path.exists(filepath):
+        return env
+    with open(filepath, 'r', encoding='utf-8') as f:
+        for line in f:
+            line = line.strip()
+            if not line or line.startswith('#'):
+                continue
+            if '=' in line:
+                key, value = line.split('=', 1)
+                key = key.strip()
+                value = value.strip().strip('"').strip("'")
+                env[key] = value
+    return env
+
+
+# ==================== ЗАГРУЗКА КОНФИГА ====================
+BASE_DIR = os.path.abspath(os.path.dirname(__file__))
+
+# Загружаем .env из config/
+env_file = os.path.join(BASE_DIR, 'config', '.env')
+env_vars = load_env_file(env_file)
+
+# ==================== ПАРАМЕТРЫ AD ====================
+# Пути к Passwork (если есть)
+PASSWORK_DIR = '/opt/passwork'
+passwork_pool = {}
+if os.path.exists(PASSWORK_DIR):
+    try:
+        sys.path.insert(0, PASSWORK_DIR)
+        from passwork import get_passwork_secrets
+        required = [env_vars.get('PW_ID_FTP', ''), env_vars.get('PW_ID_SD', ''), env_vars.get('PW_ID_AD', '')]
+        required = [r for r in required if r]
+        passwork_pool = get_passwork_secrets(required_cards=required)
+    except Exception:
+        passwork_pool = {}
+
+
+def resolve_secret(pw_key, env_key):
+    """Получает секрет из Passwork или из .env."""
+    card_name = env_vars.get(env_key, '').strip()
+    if card_name and card_name in passwork_pool:
+        return passwork_pool[card_name].get('password', '').strip()
+    return env_vars.get(env_key, '').strip()
+
+
+# AD-параметры (дублируют config.py)
+AD_SERVER = 'ldap://172.16.20.20'
+AD_BASE = 'OU=-Пользователи,DC=sibcem,DC=ru'
+AD_CONNECT_TIMEOUT = 5
+AD_RECEIVE_TIMEOUT = 5
+AD_USER = resolve_secret('AD', 'PW_ID_AD') or env_vars.get('AD_USER', '').strip()
+AD_PASSWORD = resolve_secret('AD', 'PW_ID_AD') or env_vars.get('AD_PASSWORD', '').strip()
+
+# CITY_SUPPORT - встроенный справочник (дублирует config.py)
+CITY_SUPPORT = {
+    'Ангарск': 'Техподдержка Ангарск',
+    'Слюдянка': 'Техподдержка Ангарск',
+    'Искитим': 'Техподдержка Искитим',
+    'Новосибирск': 'Техподдержка КТЦ',
+    'Топки': 'Техподдержка Топки',
+    'Кемерово': 'Техподдержка ХК',
+    'Москва': 'Техподдержка Москва',
+    'Тимлюй': 'Техподдержка ХК',
+    'Каменск': 'Техподдержка ХК',
+    'Улан-Удэ': 'Техподдержка ХК',
+    'Татарский Ключ': 'Техподдержка ХК',
+    'Спас-Клепики': 'Техподдержка ХК',
+    'Красноярск': 'Техподдержка КЦ',
+}
+
+
+# ==================== ФУНКЦИИ ПОИСКА ====================
+def print_header(title):
+    print(f'\n{"="*60}\n  {title}\n{"="*60}')
+
+
+def print_result(label, value, success=True):
+    icon = '\u2705' if success else '\u274c'
+    print(f'  {icon} {label}: {value}')
+
+
+def search_ad(login=None, email=None, phone=None):
+    """Выполняет LDAP-поиск и возвращает данные пользователя."""
+    if login:
+        ldap_filter = f'(sAMAccountName={login})'
+        search_by = f'Логину ({login})'
+        print_header(f'Поиск по логину: {login}')
+    elif email:
+        ldap_filter = f'(mail={email})'
+        search_by = f'Почте ({email})'
+        print_header(f'Поиск по email: {email}')
+    elif phone:
+        ldap_filter = f'(|(telephoneNumber=*{phone}*)(mobile=*{phone}*)(homePhone=*{phone}*)(pager=*{phone}*)(facsimileTelephoneNumber=*{phone}*)(ipPhone=*{phone}*))'
+        search_by = f'номеру телефона ({phone})'
+        print_header(f'Поиск по телефону: {phone}')
+
+    try:
+        from ldap3 import Server, Connection, ALL
+    except ImportError:
+        print_result('Ошибка', 'Модуль ldap3 не установлен. Установите: pip install ldap3', False)
+        return None
+
+    try:
+        server = Server(AD_SERVER, get_info=ALL, connect_timeout=AD_CONNECT_TIMEOUT)
+        conn = Connection(server, user=AD_USER, password=AD_PASSWORD, auto_bind=True, receive_timeout=AD_RECEIVE_TIMEOUT)
+        conn.search(search_base=AD_BASE, search_filter=ldap_filter,
+                    attributes=['l', 'company', 'manager', 'mail', 'displayName', 'telephoneNumber'])
+    except Exception as e:
+        print_result('Ошибка подключения к AD', str(e), False)
+        return None
+
+    if not conn.entries:
+        print_result('Результат', 'Пользователь не найден', False)
+        return None
+
+    user = conn.entries[0]
+
+    # Парсим атрибуты
+    city = user.l.value if 'l' in user and user.l.value else None
+    company = user.company.value if 'company' in user and user.company.value else None
+    ad_path = user.entry_dn
+    display_name = user.displayName.value if 'displayName' in user and user.displayName.value else None
+    mail = user.mail.value if 'mail' in user and user.mail.value else None
+    phone_val = user.telephoneNumber.value if 'telephoneNumber' in user and user.telephoneNumber.value else None
+
+    # Город из OU
+    if not city and ad_path:
+        match = re.search(r'OU=([^,]+),OU=-Пользователи', ad_path, re.IGNORECASE)
+        if match:
+            city = match.group(1).strip()
+            if city.lower() == 'красноярск' and 'OU=ООО Комбинат Волна' in ad_path:
+                company = 'ООО Комбинат Волна'
+
+    # Email руководителя
+    manager_dn = user.manager.value if 'manager' in user else None
+    manager_email = None
+    if manager_dn:
+        try:
+            server2 = Server(AD_SERVER, get_info=ALL, connect_timeout=AD_CONNECT_TIMEOUT)
+            conn2 = Connection(server2, user=AD_USER, password=AD_PASSWORD, auto_bind=True, receive_timeout=AD_RECEIVE_TIMEOUT)
+            conn2.search(search_base=manager_dn, search_filter='(objectClass=user)', attributes=['mail'], search_scope='BASE')
+            if conn2.entries and 'mail' in conn2.entries[0]:
+                manager_email = conn2.entries[0].mail.value
+        except Exception:
+            pass
+
+    # Вывод результатов
+    print_result('ФИО', display_name or 'Не указано')
+    print_result('Email', mail or 'Не указано')
+    print_result('Телефон', phone_val or 'Не указано')
+    print_result('Город', city or 'Не определено', city is not None)
+    print_result('Компания', company or 'Не определена', company is not None)
+    print_result('OU-путь', ad_path)
+    print_result('Руководитель', manager_email or 'Не указан', manager_email is not None)
+    print_result('Источник', f'AD ({search_by})')
+
+    # Региональная техподдержка
+    if city:
+        target_group = CITY_SUPPORT.get(city.strip())
+        if target_group:
+            print_result('\U0001f4cd Региональная техподдержка', target_group)
+        elif city.strip() == 'Красноярск':
+            print_result('\U0001f4cd Региональная техподдержка', 'Техподдержка КЦ (по умолчанию)')
+
+    return {
+        'city': city, 'company': company, 'ad_path': ad_path,
+        'manager_email': manager_email, 'display_name': display_name,
+        'mail': mail, 'phone': phone_val, 'source': f'AD ({search_by})',
+    }
+
+
+def print_config_info():
+    """Вывод конфигурации."""
+    print_header('Конфигурация AD')
+    print_result('AD Server', AD_SERVER)
+    print_result('Base DN', AD_BASE)
+    print_result('Connect Timeout', f'{AD_CONNECT_TIMEOUT} сек')
+    print_result('Receive Timeout', f'{AD_RECEIVE_TIMEOUT} сек')
+    print_result('Bind User', AD_USER or 'Не задан')
+    print_result('.env файл', env_file)
+    print_result('Справочник городов', f'{len(CITY_SUPPORT)} городов')
+
+
+def main():
+    parser = argparse.ArgumentParser(description='Тест подключения к Active Directory')
+    group = parser.add_mutually_exclusive_group(required=True)
+    group.add_argument('--login', '-l', help='sAMAccountName пользователя')
+    group.add_argument('--email', '-e', help='Email пользователя')
+    group.add_argument('--phone', '-p', help='Номер телефона')
+    args = parser.parse_args()
+
+    print_header('SD Dispatcher - Тест подключения к AD')
+    print(f'  Проект: {BASE_DIR}')
+    print(f'  Config: {os.path.join(BASE_DIR, "config", "config.py")}')
+    print()
+    print_config_info()
+
+    result = search_ad(login=args.login, email=args.email, phone=args.phone)
+
+    if result:
+        print(f'\n{"="*60}\n  \u2705 Подключение к AD работает!\n{"="*60}')
+        return 0
+    else:
+        print(f'\n{"="*60}\n  \u274c Не удалось получить данные из AD\n{"="*60}')
+        return 1
+
+
+if __name__ == '__main__':
+    sys.exit(main())

test_ad_lookup.py

@@ -0,0 +1,283 @@
+#!/usr/bin/env python3
+"""
+Тестовый скрипт для проверки обращения к Active Directory.
+Аналог find_user_in_ad() из sd_dispatcher.py — standalone.
+
+Установка зависимостей (на сервере 192.168.1.106):
+  python3 -m pip install ldap3
+
+Использование:
+  # 1) Через аргументы:
+  python3 test_ad_lookup.py --user "sibcem\\svc_dispatcher" --pass "password" --login ivanov
+
+  # 2) Через env-переменные:
+  export AD_USER="sibcem\\svc_dispatcher"
+  export AD_PASS="password"
+  python3 test_ad_lookup.py --login ivanov
+
+  # 3) По email:
+  python3 test_ad_lookup.py --email ivanov@sibcem.ru
+
+  # 4) По телефону:
+  python3 test_ad_lookup.py --phone 79131234567
+
+  # 5) Тест: найти первого попавшегося пользователя и проверить все методы:
+  python3 test_ad_lookup.py --test-all
+"""
+
+import argparse
+import os
+import re
+import sys
+
+# ---------------------------------------------------------------------------
+# 1. Загрузка учётных данных
+# ---------------------------------------------------------------------------
+
+def get_credentials():
+    """Возвращает (user, password, base_dn)."""
+    user = os.getenv("AD_USER", "").strip()
+    password = os.getenv("AD_PASS", "").strip()
+    base = os.getenv("AD_BASE", "").strip() or "OU=-Пользователи,DC=sibcem,DC=ru"
+
+    return user, password, base
+
+
+# ---------------------------------------------------------------------------
+# 2. Подключение к LDAP
+# ---------------------------------------------------------------------------
+
+def connect_to_ad(ad_user, ad_pass, ad_base):
+    """Возвращает подключённый ldap3.Connection."""
+    from ldap3 import Server, Connection
+
+    server = Server(
+        "ldap://172.16.20.20",
+        get_info=None,  # не тянем schema — экономим время
+        connect_timeout=5,
+    )
+    conn = Connection(
+        server,
+        user=ad_user,
+        password=ad_pass,
+        auto_bind=True,
+        receive_timeout=5,
+    )
+    return conn, ad_base
+
+
+# ---------------------------------------------------------------------------
+# 3. Поиск пользователя
+# ---------------------------------------------------------------------------
+
+def find_user(conn, ad_base, login=None, email=None, phone=None):
+    """
+    Ищет пользователя в AD. Приоритет: телефон > логин > email.
+    Возвращает dict с атрибутами или None.
+    """
+    if phone:
+        filt = (
+            f"(|(telephoneNumber=*{phone}*)(mobile=*{phone}*) "
+            f"(homePhone=*{phone}*)(pager=*{phone}*) "
+            f"(facsimileTelephoneNumber=*{phone}*)(ipPhone=*{phone}*))"
+        )
+        by = f"телефону {phone}"
+    elif login:
+        clean = login.split("\\")[-1].split("/")[-1]
+        filt = f"(sAMAccountName={clean})"
+        by = f"логину {clean}"
+    elif email:
+        filt = f"(mail={email})"
+        by = f"почте {email}"
+    else:
+        print("❌ Укажите --login, --email или --phone")
+        sys.exit(1)
+
+    print(f"\n[LDAP] Поиск: {filt}")
+    print(f"[LDAP] Base DN: {ad_base}")
+
+    # Запрос 1 — основной
+    conn.search(
+        search_base=ad_base,
+        search_filter=filt,
+        attributes=["l", "company", "manager", "department", "ou",
+                     "displayName", "mail", "sAMAccountName", "title",
+                     "telephoneNumber", "mobile", "givenName", "sn"],
+    )
+
+    if not conn.entries:
+        print(f"\n❌ Пользователь не найден по {by}")
+        return None
+
+    user = conn.entries[0]
+    print(f"\n✅ Найдено: {user.entry_dn}")
+
+    result = {
+        "sam": user.sAMAccountName.value if "sAMAccountName" in user else None,
+        "given_name": user.givenName.value if "givenName" in user else None,
+        "surname": user.sn.value if "sn" in user else None,
+        "display_name": user.displayName.value if "displayName" in user else None,
+        "mail": user.mail.value if "mail" in user else None,
+        "title": user.title.value if "title" in user else None,
+        "department": user.department.value if "department" in user else None,
+        "city": user.l.value if "l" in user and user.l.value else None,
+        "company": user.company.value if "company" in user and user.company.value else None,
+        "dn": user.entry_dn,
+        "phone": user.telephoneNumber.value if "telephoneNumber" in user else None,
+        "mobile": user.mobile.value if "mobile" in user else None,
+    }
+
+    # Город из OU-пути если атрибут l пустой
+    if not result["city"]:
+        match = re.search(r"OU=([^,]+),OU=-Пользователи", result["dn"], re.IGNORECASE)
+        if match:
+            result["city"] = match.group(1).strip()
+            if result["city"].lower() == "красноярск" and "OU=ООО Комбинат Волна" in result["dn"]:
+                result["company"] = "ООО Комбинат Волна"
+            print(f"   Город извлечён из OU: {result['city']}")
+
+    # Запрос 2 — email руководителя
+    manager_dn = user.manager.value if "manager" in user else None
+    if manager_dn:
+        conn.search(
+            search_base=manager_dn,
+            search_filter="(objectClass=user)",
+            attributes=["mail", "displayName"],
+            search_scope="BASE",
+        )
+        if conn.entries and "mail" in conn.entries[0]:
+            result["manager_email"] = conn.entries[0].mail.value
+            result["manager_name"] = conn.entries[0].displayName.value if "displayName" in conn.entries[0] else None
+        else:
+            result["manager_email"] = None
+    else:
+        result["manager_email"] = None
+
+    return result
+
+
+# ---------------------------------------------------------------------------
+# 4. Форматирование вывода
+# ---------------------------------------------------------------------------
+
+def print_report(data):
+    """Красивый вывод результатов."""
+    print("\n" + "=" * 60)
+    print("  РЕЗУЛЬТАТ ПОИСКА В ACTIVE DIRECTORY")
+    print("=" * 60)
+    print(f"  sAMAccountName:  {data['sam']}")
+    print(f"  Имя:             {data['given_name']}")
+    print(f"  Фамилия:         {data['surname']}")
+    print(f"  ФИО:             {data['display_name']}")
+    print(f"  Email:           {data['mail']}")
+    print(f"  Должность:       {data['title']}")
+    print(f"  Отдел:           {data['department']}")
+    print(f"  Город:           {data['city'] or '—'}")
+    print(f"  Компания:        {data['company'] or '—'}")
+    print(f"  Телефон:         {data['phone'] or '—'}")
+    print(f"  Мобильный:       {data['mobile'] or '—'}")
+    print(f"  Руководитель:    {data['manager_name'] or '—'}")
+    if data.get("manager_email"):
+        print(f"  Email руково.:   {data['manager_email']}")
+    print("-" * 60)
+    print(f"  DN: {data['dn']}")
+
+    # OU-дерево
+    dn = data["dn"]
+    parts = [p.strip() for p in dn.split(",")]
+    ou_parts = [p for p in parts if p.startswith("OU=")]
+    if ou_parts:
+        print(f"\n  OU-структура:")
+        for i, ou in enumerate(reversed(ou_parts)):
+            indent = "    " * i
+            print(f"  {indent}└── {ou}")
+
+    print("=" * 60)
+
+
+# ---------------------------------------------------------------------------
+# 5. Тестирование всех методов поиска
+# ---------------------------------------------------------------------------
+
+def test_all_methods(conn, ad_base):
+    """Находит первого пользователя и проверяет все методы поиска."""
+    print("\n" + "─" * 60)
+    print("  РЕЖИМ ТЕСТИРОВАНИЯ: поиск по всем методам")
+    print("─" * 60)
+
+    conn.search(
+        search_base=ad_base,
+        search_filter="(objectClass=user)",
+        attributes=["sAMAccountName", "mail", "telephoneNumber", "displayName",
+                     "l", "company", "department"],
+        size_limit=1,
+    )
+
+    if not conn.entries:
+        print("❌ В AD нет пользователей.")
+        return
+
+    sample = conn.entries[0]
+    sam = sample.sAMAccountName.value
+    mail = sample.mail.value if "mail" in sample else None
+    phone = sample.telephoneNumber.value if "telephoneNumber" in sample else None
+
+    print(f"\n📌 Тестовый пользователь: {sam} ({mail})")
+
+    for label, kwargs in [("логину", {"login": sam}),
+                           ("email", {"email": mail}) if mail else None,
+                           ("телефону", {"phone": phone}) if phone else None]:
+        if kwargs is None:
+            continue
+        print(f"\n[Тест] Поиск по {label}...")
+        r = find_user(conn, ad_base, **kwargs)
+        if r:
+            print_report(r)
+
+    print("\n" + "─" * 60)
+    print("  ✅ Тестирование завершено")
+    print("─" * 60)
+
+
+# ---------------------------------------------------------------------------
+# MAIN
+# ---------------------------------------------------------------------------
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Тестовый поиск пользователя в AD")
+    parser.add_argument("--user", help="LDAP user (sAMAccountName или DN)")
+    parser.add_argument("--pass", dest="password", help="LDAP password")
+    parser.add_argument("--base", help="Base DN (по умолчанию OU=-Пользователи,DC=sibcem,DC=ru)")
+    group = parser.add_mutually_exclusive_group(required=True)
+    group.add_argument("--login", help="sAMAccountName пользователя для поиска")
+    group.add_argument("--email", help="Email пользователя для поиска")
+    group.add_argument("--phone", help="Номер телефона для поиска")
+    parser.add_argument("--test-all", action="store_true",
+                        help="Найти первого пользователя и протестировать все методы поиска")
+    args = parser.parse_args()
+
+    # Учётные данные: аргументы > env > prompt
+    ad_user = args.user or os.getenv("AD_USER", "")
+    ad_pass = args.password or os.getenv("AD_PASS", "")
+    ad_base = args.base or os.getenv("AD_BASE", "OU=-Пользователи,DC=sibcem,DC=ru")
+
+    if not ad_user or not ad_pass:
+        # Пробуем получить из .env файла проекта
+        env_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "sd_dispatcher", "config", ".env")
+        if os.path.exists(env_path):
+            print(f"[.env] Загружено: {env_path}")
+        else:
+            print("❌ Укажите --user и --pass, или задайте AD_USER / AD_PASS")
+            sys.exit(1)
+
+    conn, ad_base = connect_to_ad(ad_user, ad_pass, ad_base)
+    print(f"[LDAP] Подключено к ldap://172.16.20.20 как {ad_user}")
+
+    if args.test_all:
+        test_all_methods(conn, ad_base)
+    else:
+        data = find_user(conn, ad_base, login=args.login, email=args.email, phone=args.phone)
+        if data:
+            print_report(data)
+
+    conn.unbind()