""" =============================================================================== FILE: services/scud_etl/anomaly_detector.py PROJECT: SCUD Orion AI (Unified Architecture) MODULE: services / scud_etl ROLE: Детектирование истинных аномалий и конфликтов реестров. (Приход удаленщиков и командированных в офис аномалией НЕ является). =============================================================================== """ import logging from typing import List, Dict, Any import pandas as pd logger = logging.getLogger("SCUD_ANOMALY") def detect_registry_anomalies(df_merged: pd.DataFrame) -> List[Dict[str, Any]]: """ Выявляет реальные аномалии: - Приход в офис во время отпуска или больничного листа. - Ошибки считывателей СКУД (наличие выхода при отсутствии отметки входа). """ anomalies = [] for _, row in df_merged.iterrows(): fio = row.get("fio_clean") or row.get("Сотрудник", "") start_day = str(row.get("Начало_дня", "")).strip() end_day = str(row.get("Конец_дня", "")).strip() reason = str(row.get("причина отсутствия", "")).strip() reason_lower = reason.lower() # 1. Приход в офис при отпуске / больничном if start_day != "Нет входа" and reason and reason != "nan": # Удаленная работа и командировки разрешены для работы в офисе if not ("удален" in reason_lower or "дистанцион" in reason_lower or "командировк" in reason_lower or "поездк" in reason_lower): anomalies.append({ "fio": fio, "type": "PHYSICAL_PRESENCE_DURING_ABSENCE", "description": f"Сотрудник пришел по СКУД ({start_day}), но в 1С оформлен документ: '{reason}'." }) # 2. Аномалия оборудования (есть выход без входа) if start_day == "Нет входа" and end_day != "Нет выхода": anomalies.append({ "fio": fio, "type": "SCUD_EQUIPMENT_ANOMALY", "description": f"Отсутствует отметка утреннего входа при наличии выхода ({end_day})." }) return anomalies