From 179f1b1a74115ff4d536d242cc39c72bea802566 Mon Sep 17 00:00:00 2001 From: dddennnisss Date: Sat, 4 Jul 2026 16:16:35 +0700 Subject: [PATCH 1/3] 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()) From 074379aa22ef4baecb237ae80341faf9fef468b1 Mon Sep 17 00:00:00 2001 From: dddennnisss Date: Sat, 4 Jul 2026 16:16:36 +0700 Subject: [PATCH 2/3] chore: remove test_ad_lookup.py from main --- test_ad_lookup.py | 283 ---------------------------------------------- 1 file changed, 283 deletions(-) delete mode 100644 test_ad_lookup.py diff --git a/test_ad_lookup.py b/test_ad_lookup.py deleted file mode 100644 index f078d75..0000000 --- a/test_ad_lookup.py +++ /dev/null @@ -1,283 +0,0 @@ -#!/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() From 1949a461b2e456c469497fc4810433936c5921af Mon Sep 17 00:00:00 2001 From: Denis Krivchenko Date: Sat, 4 Jul 2026 17:55:35 +0700 Subject: [PATCH 3/3] feat: add department exclusion for auto-approval - Add department attribute to LDAP search - Add APPROVAL_EXCLUDED_DEPARTMENTS config - Skip auto-approval if user department is excluded --- config/config.py | 8 ++++++++ sd_dispatcher.py | 18 ++++++++++++++++-- 2 files changed, 24 insertions(+), 2 deletions(-) diff --git a/config/config.py b/config/config.py index ca8bec4..1355dac 100644 --- a/config/config.py +++ b/config/config.py @@ -287,6 +287,14 @@ APPROVAL_CATEGORIES = { "1с": ["*"] } +# Отделы, исключающие пользователя из автосогласования +# Если department пользователя совпадает с любым из этих значений — +# автосогласование НЕ запускается, даже если все остальные условия выполнены. +APPROVAL_EXCLUDED_DEPARTMENTS = [ + # "IT-Администраторы", + # "Системные аналитики", +] + # План «Б» маршрутизации: жесткое определение локации по телефонному префиксу внутренней АТС PREFIX_ROUTING = { "700": {"city": "Красноярск", "company": "ООО Комбинат Волна"}, diff --git a/sd_dispatcher.py b/sd_dispatcher.py index acaefff..be1c6bd 100644 --- a/sd_dispatcher.py +++ b/sd_dispatcher.py @@ -170,7 +170,7 @@ def find_user_in_ad(requester, caller_phone=None, ticket_id=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"]) + conn.search(search_base=AD_BASE, search_filter=ldap_filter, attributes=["l", "company", "manager", "department"]) if not conn.entries: log_to_file(f" [AD] ❌ Объект ({search_by}) не найден in AD!", ticket_id) @@ -195,7 +195,8 @@ def find_user_in_ad(requester, caller_phone=None, ticket_id=None): if conn.entries and 'mail' in conn.entries[0]: manager_email = conn.entries[0].mail.value - return {"city": city, "company": company, "ad_path": ad_path, "manager_email": manager_email, "source": f"AD ({search_by})"} + department = user.department.value if "department" in user and user.department.value else None + return {"city": city, "company": company, "department": department, "ad_path": ad_path, "manager_email": manager_email, "source": f"AD ({search_by})"} except Exception as e: log_to_file(f"❌ Ошибка LDAP или таймаут сети AD: {e}", ticket_id) return None @@ -764,6 +765,7 @@ def process_ticket_logic(req_id): ad_path = user_info["ad_path"] if user_info else None manager_email = user_info["manager_email"] if user_info else None info_source = user_info["source"] if user_info else "Не определен" + user_department = user_info["department"] if user_info else None prefix_used = False if caller_phone and not user_info: @@ -892,6 +894,18 @@ def process_ticket_logic(req_id): if "*" in allowed_subcats or subcat_name in allowed_subcats: is_cat_ok = True + # Проверка: отдел пользователя в списке исключений + is_excluded = False + if user_department and APPROVAL_EXCLUDED_DEPARTMENTS: + for dept in APPROVAL_EXCLUDED_DEPARTMENTS: + if dept.lower() in user_department.lower() or user_department.lower() in dept.lower(): + is_excluded = True + t_log(f" ⚠️ Отдел '{user_department}' в списке исключений — автосогласование отменено") + break + if is_excluded: + t_log(" ⏭ Пропуск: пользователь в исключённом отделе.") + continue + if is_city_ok and (is_group_ok or is_cat_ok): if manager_email: t_log(f"✔ Условия совпали. Запуск процесса согласования для {manager_email}...")