chore: remove test_ad.py from main
This commit is contained in:
-230
@@ -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())
|
||||
Reference in New Issue
Block a user