Compare commits

...

30 Commits

Author SHA1 Message Date
dddennnisss b269e7d6ec fix: CSV QUOTE_ALL for proper column separation with commas in values 2026-07-27 12:30:25 +07:00
dddennnisss 23d073f9d8 fix: replace newlines in descriptions for proper CSV table format 2026-07-27 12:27:54 +07:00
dddennnisss 9c80ff716e fix: CSV encoding utf-8-sig for proper Cyrillic display in Excel 2026-07-27 12:24:01 +07:00
dddennnisss 00630cd18c feat: add JSON and CSV file output with argparse 2026-07-27 12:20:49 +07:00
dddennnisss 748b262936 fix: get API_KEY directly from passwork, remove unused config import 2026-07-27 11:57:13 +07:00
dddennnisss f03cd70b48 fix: get API_KEY directly from passwork on srvkem-ii-02 2026-07-27 11:54:17 +07:00
dddennnisss 16b0b9f063 fix: safe config import with getattr + env fallback 2026-07-27 11:49:01 +07:00
dddennnisss 23a18b7396 fix: add fallback for SERVER_URL and API_KEY, add logging 2026-07-27 11:47:58 +07:00
dddennnisss 08220a6858 fix: add fallback for SERVER_URL if config import fails 2026-07-27 11:47:28 +07:00
dddennnisss 6c8c79cf70 fix: config path for /opt/GigaAM/sd/ (parent dir in sys.path) 2026-07-27 11:38:55 +07:00
dddennnisss a1d391cf5f feat: fetch tickets from SD using requests + config (SERVER_URL, API_KEY from passwork) 2026-07-27 11:29:17 +07:00
dddennnisss b7b099cb38 feat: use SERVER_URL and API_KEY from config (passwork), remove LOCAL_SD 2026-07-27 10:42:43 +07:00
dddennnisss 6494ed1e17 feat: fetch tickets from LOCAL_SD config using from config import * 2026-07-27 00:32:25 +07:00
dddennnisss 46cbcde05b feat: fetch tickets from LOCAL_SD config using from config import * 2026-07-27 00:29:19 +07:00
dddennnisss a980bea083 feat: fetch tickets from LOCAL_SD config using from config import * 2026-07-27 00:28:23 +07:00
dddennnisss f974f678f8 feat: fetch tickets from LOCAL_SD config using from config import * 2026-07-27 00:27:14 +07:00
dddennnisss 3dff2c2a95 feat: fetch tickets from LOCAL_SD config using from config import * 2026-07-27 00:26:13 +07:00
dddennnisss 167e79fa2c feat: fetch tickets from LOCAL_SD config 2026-07-27 00:24:58 +07:00
dddennnisss 8aa13afad9 feat: fetch tickets from LOCAL_SD config 2026-07-27 00:23:49 +07:00
dddennnisss 08f6b67971 feat: add LOCAL_SD config, fetch tickets with group/desc/assignee/category 2026-07-27 00:23:14 +07:00
dddennnisss 42335ce015 refactor: use requests sync, graceful config import 2026-07-27 00:14:39 +07:00
dddennnisss 13f99579f5 revert: remove LOCAL_SD from config.py 2026-07-26 23:35:33 +07:00
dddennnisss 3396e3e65b revert: remove LOCAL_SD from config.py 2026-07-26 23:35:09 +07:00
dddennnisss a48b484ea1 feat: read SD config from config.py, add category/assignee, support date arg 2026-07-26 23:20:32 +07:00
dddennnisss 1ee2b6387a fix: get category and assignee from ticket detail API 2026-07-26 23:13:20 +07:00
dddennnisss d14930aa39 feat: add fetch_tickets.py with group, description, assignee, category 2026-07-26 23:10:57 +07:00
dddennnisss 4ad93cfdd5 revert cf1d526304
revert Clean main branch
2026-07-14 17:01:24 +07:00
dddennnisss cf1d526304 Clean main branch 2026-07-14 09:46:58 +00:00
dddennnisss 074379aa22 chore: remove test_ad_lookup.py from main 2026-07-04 16:16:36 +07:00
dddennnisss 179f1b1a74 chore: remove test_ad.py from main 2026-07-04 16:16:35 +07:00
4 changed files with 169 additions and 513 deletions
+1
View File
@@ -115,6 +115,7 @@ ENABLE_FUZZY_CORRECTION = True # Нечёткое исправление по
SERVER_URL = "https://servicedesk.sibcem.ru:8080"
ENDPOINT = f"{SERVER_URL}/api/v3/requests"
# Системный email-адрес учетной записи ИИ-сортировщика в Service Desk
DEFAULT_REQUESTER = "ai@sibcem.ru"
+168
View File
@@ -0,0 +1,168 @@
#!/usr/bin/env python3
"""Получение списка заявок из ServiceDesk Plus. Экспорт: Группа, Описание, Назначено, Категория в JSONL/JSON/CSV."""
import requests
import json
import csv
import urllib3
from datetime import datetime
from html import unescape
import re
import sys
import os
# Отключаем ворнинги для чистой работы консоли
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
# --- ПОЛУЧЕНИЕ API_KEY ИЗ PASSWORK ---
SERVER_URL = "https://servicedesk.sibcem.ru:8080"
API_KEY = ""
# 1. Пробуем passwork напрямую
try:
sys.path.insert(0, "/opt/passwork")
from passwork import get_passwork_secrets
passwork_pool = get_passwork_secrets(required_cards=["sdesktoken"])
card = passwork_pool.get("sdesktoken")
if card:
API_KEY = card.get("password", "").strip()
print("LOG: API_KEY получен из passwork (sdesktoken)")
else:
print("LOG: Карточка sdesktoken не найдена в passwork", file=sys.stderr)
except Exception as e:
print(f"LOG: Passwork ошибка: {e}", file=sys.stderr)
# 2. Пробуем config.py
if not API_KEY:
try:
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import config as cfg
API_KEY = getattr(cfg, 'API_KEY', "")
if API_KEY:
print("LOG: API_KEY получен из config.py")
except Exception:
pass
# 3. Пробуем env var
if not API_KEY:
API_KEY = os.environ.get("SD_API_KEY", "")
if API_KEY:
print("LOG: API_KEY получен из SD_API_KEY")
if not API_KEY:
print("LOG: ОШИБКА: API_KEY не найден нигде! Нужен passwork или SD_API_KEY", file=sys.stderr)
sys.exit(1)
print(f"LOG: Подключение к {SERVER_URL}")
def get_ts(req):
ct = req.get("created_time", {})
if isinstance(ct, dict):
v = ct.get("value", "0")
return int(v) if v else 0
return 0
def clean_html(text):
if not text:
return ""
text = re.sub(r'<[^>]+>', '', text)
text = unescape(text)
return text.strip()
def get_name(obj):
if not obj:
return ""
if isinstance(obj, dict):
return obj.get("name", "") or obj.get("display_value", "")
return str(obj)
def get_ticket_detail(ticket_id):
try:
url = f"{SERVER_URL}/api/v3/requests/{ticket_id}"
headers = {"authtoken": API_KEY, "Accept": "application/vnd.manageengine.sdp.v3+json"}
resp = requests.get(url, headers=headers, verify=False, timeout=15.0)
resp.raise_for_status()
data = resp.json()
return data.get("request", data)
except Exception as e:
return {}
def list_tickets(from_date="2026-07-20", json_file=None, csv_file=None):
url = f"{SERVER_URL}/api/v3/requests"
headers = {"authtoken": API_KEY, "Accept": "application/vnd.manageengine.sdp.v3+json"}
date_obj = datetime.strptime(from_date, "%Y-%m-%d")
from_date_ms = int(date_obj.replace(tzinfo=__import__('datetime').timezone.utc).timestamp() * 1000)
print(f"LOG: Загрузка заявок с {from_date}...")
resp = requests.get(url, headers=headers, verify=False, timeout=30.0)
resp.raise_for_status()
data = resp.json()
all_reqs = data.get("requests", [])
from_date_reqs = [r for r in all_reqs if get_ts(r) >= from_date_ms]
print(f"LOG: Найдено заявок с {from_date}: {len(from_date_reqs)}")
rows = []
for i, req in enumerate(from_date_reqs):
if (i + 1) % 10 == 0:
print(f" Обработано {i + 1}/{len(from_date_reqs)}...")
req_id = req.get("request_id", req.get("id", "?"))
detail = get_ticket_detail(req_id)
group = req.get("group", {})
group_name = get_name(group)
desc = detail.get("description", "")
desc_clean = clean_html(desc).replace('\n', ' ').replace('\r', ' ')
tech = detail.get("technician", {})
assignee = get_name(tech)
cat = detail.get("category", {})
category = get_name(cat)
rows.append({
"group": group_name,
"description": desc_clean,
"assignee": assignee,
"category": category
})
# Определяем путь для файлов по умолчанию (та же директория что и скрипт)
script_dir = os.path.dirname(os.path.abspath(__file__))
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
if json_file is None:
json_file = os.path.join(script_dir, f"tickets_{timestamp}.json")
if csv_file is None:
csv_file = os.path.join(script_dir, f"tickets_{timestamp}.csv")
# Сохраняем JSON
with open(json_file, 'w', encoding='utf-8') as f:
json.dump(rows, f, ensure_ascii=False, indent=2)
print(f"LOG: JSON сохранён: {json_file}")
# Сохраняем CSV (utf-8-sig для корректного отображения кириллицы в Excel)
with open(csv_file, 'w', encoding='utf-8-sig', newline='') as f:
writer = csv.DictWriter(f, fieldnames=["group", "description", "assignee", "category"], quoting=csv.QUOTE_ALL)
writer.writeheader()
writer.writerows(rows)
print(f"LOG: CSV сохранён: {csv_file}")
# Выводим JSONL в консоль
for row in rows:
print(json.dumps(row, ensure_ascii=False))
print(f"LOG: Выведено {len(rows)} заявок", file=sys.stderr)
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser(description="Выгрузка заявок из ServiceDesk Plus")
parser.add_argument("--from", dest="from_date", default="2026-07-20", help="Дата начала (YYYY-MM-DD)")
parser.add_argument("--json", dest="json_file", default=None, help="Путь к JSON файлу")
parser.add_argument("--csv", dest="csv_file", default=None, help="Путь к CSV файлу")
args = parser.parse_args()
list_tickets(from_date=args.from_date, json_file=args.json_file, csv_file=args.csv_file)
-230
View File
@@ -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())
-283
View File
@@ -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()