From 179f1b1a74115ff4d536d242cc39c72bea802566 Mon Sep 17 00:00:00 2001 From: dddennnisss Date: Sat, 4 Jul 2026 16:16:35 +0700 Subject: [PATCH] chore: remove test_ad.py from main --- test_ad.py | 230 ----------------------------------------------------- 1 file changed, 230 deletions(-) delete mode 100644 test_ad.py diff --git a/test_ad.py b/test_ad.py deleted file mode 100644 index 3b4a14a..0000000 --- a/test_ad.py +++ /dev/null @@ -1,230 +0,0 @@ -#!/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())