Compare commits
30 Commits
2d9c50d0b9
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| b269e7d6ec | |||
| 23d073f9d8 | |||
| 9c80ff716e | |||
| 00630cd18c | |||
| 748b262936 | |||
| f03cd70b48 | |||
| 16b0b9f063 | |||
| 23a18b7396 | |||
| 08220a6858 | |||
| 6c8c79cf70 | |||
| a1d391cf5f | |||
| b7b099cb38 | |||
| 6494ed1e17 | |||
| 46cbcde05b | |||
| a980bea083 | |||
| f974f678f8 | |||
| 3dff2c2a95 | |||
| 167e79fa2c | |||
| 8aa13afad9 | |||
| 08f6b67971 | |||
| 42335ce015 | |||
| 13f99579f5 | |||
| 3396e3e65b | |||
| a48b484ea1 | |||
| 1ee2b6387a | |||
| d14930aa39 | |||
| 4ad93cfdd5 | |||
| cf1d526304 | |||
| 074379aa22 | |||
| 179f1b1a74 |
@@ -115,6 +115,7 @@ ENABLE_FUZZY_CORRECTION = True # Нечёткое исправление по
|
|||||||
SERVER_URL = "https://servicedesk.sibcem.ru:8080"
|
SERVER_URL = "https://servicedesk.sibcem.ru:8080"
|
||||||
ENDPOINT = f"{SERVER_URL}/api/v3/requests"
|
ENDPOINT = f"{SERVER_URL}/api/v3/requests"
|
||||||
|
|
||||||
|
|
||||||
# Системный email-адрес учетной записи ИИ-сортировщика в Service Desk
|
# Системный email-адрес учетной записи ИИ-сортировщика в Service Desk
|
||||||
DEFAULT_REQUESTER = "ai@sibcem.ru"
|
DEFAULT_REQUESTER = "ai@sibcem.ru"
|
||||||
|
|
||||||
|
|||||||
@@ -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
@@ -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())
|
|
||||||
@@ -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()
|
|
||||||
Reference in New Issue
Block a user