"""Сервисы для directory identity lookup.""" from __future__ import annotations from dataclasses import dataclass from django.conf import settings from employees.models import LdapSyncState @dataclass(frozen=True) class DirectoryLoginResult: username: str | None source: str is_cached: bool is_ldap_managed: bool def _find_unique_by_email(ldap_user_model, email: str): if not email: return None for lookup in ("mail", "user_principal_name"): candidates = list(ldap_user_model.objects.filter(**{lookup: email})) if len(candidates) == 1: return candidates[0] return None def _find_by_sync_state(ldap_user_model, employee_number: str): sync_state = LdapSyncState.objects.filter( model="employee", object_pk=employee_number, ).first() if not sync_state or not sync_state.ldap_dn: return None try: return ldap_user_model.objects.get(dn=sync_state.ldap_dn) except ldap_user_model.DoesNotExist: return None def _normalized_ldap_email(ldap_user, attribute: str) -> str: return (getattr(ldap_user, attribute, "") or "").strip().lower() def _matching_email_candidate(candidates, email: str): for candidate in candidates: if email in { _normalized_ldap_email(candidate, "mail"), _normalized_ldap_email(candidate, "user_principal_name"), }: return candidate return None def _candidate_sort_key(candidate, email: str): return ( 0 if _normalized_ldap_email(candidate, "mail") == email else 1, 0 if _normalized_ldap_email(candidate, "user_principal_name") == email else 1, str(getattr(candidate, "dn", "")).lower(), ) def _find_ldap_user(employee): from employees.ldap.orm_models import LdapUser employee_number = str(employee.pk) employee_email = (employee.email or "").strip().lower() sync_state_match = _find_by_sync_state(LdapUser, employee_number) if sync_state_match: return sync_state_match if not employee.is_ldap_managed: email_match = _find_unique_by_email(LdapUser, employee_email) if email_match is not None: return email_match candidates = list(LdapUser.objects.filter(employee_number=employee_number)) if not candidates: return _find_unique_by_email(LdapUser, employee_email) if len(candidates) == 1: return candidates[0] email_match = _matching_email_candidate(candidates, employee_email) if email_match: return email_match candidates.sort(key=lambda item: _candidate_sort_key(item, employee_email)) return candidates[0] def resolve_directory_login(employee, *, force_refresh: bool = False): cached_username = (employee.username or "").strip() if cached_username and not force_refresh: return DirectoryLoginResult( username=cached_username, source="db", is_cached=True, is_ldap_managed=employee.is_ldap_managed, ) if not employee.is_ldap_managed: return DirectoryLoginResult( username=cached_username or None, source="db" if cached_username else "none", is_cached=bool(cached_username), is_ldap_managed=False, ) if not getattr(settings, "LDAP_ENABLED", False): return DirectoryLoginResult( username=cached_username or None, source="db" if cached_username else "none", is_cached=bool(cached_username), is_ldap_managed=True, ) ldap_user = _find_ldap_user(employee) if not ldap_user or not (ldap_user.sam_account_name or "").strip(): return DirectoryLoginResult( username=None, source="ldap_not_found", is_cached=False, is_ldap_managed=True, ) username = ldap_user.sam_account_name.strip() if employee.username != username: employee.username = username employee.save(update_fields=["username"]) return DirectoryLoginResult( username=username, source="ldap", is_cached=False, is_ldap_managed=True, )