import os import pandas as pd import warnings from config import ( normalize_fio, clean_scud_fio_light, load_exceptions, DATE_TODAY, DATE_YESTERDAY, find_dated_file, ZUP_1C_DIR, SCUD_DIR ) # Импортируем прямую SQL-выгрузку отсутствий из 1С и функции работы с SQLite from services.zup_extractor import fetch_zup_absences_from_sql from core.database import ( load_scud_from_db_by_snapshot, load_staff_from_db, load_absences_from_db ) warnings.filterwarnings('ignore', category=UserWarning, module='pandas') def is_excluded(fio, dept, pos, exceptions): """Проверяет, входит ли сотрудник или должность в список исключений.""" fio_clean = normalize_fio(fio) dept_str = str(dept).strip().upper() if pd.notna(dept) else "" pos_str = str(pos).strip().lower() if pd.notna(pos) else "" if fio_clean in exceptions["fio"] or dept_str in exceptions["departments"] or pos_str in exceptions["positions"]: return True if any(keyword in pos_str for keyword in exceptions["position_keywords"]): return True return False def load_staff_data(date_str): """ Загружает файл Штатного расписания 1С за конкретную дату из папки 1c. """ filepath = find_dated_file("Штат", date_str) if not filepath: fallback_path = os.path.join(ZUP_1C_DIR, "штат.xlsx") if os.path.exists(fallback_path): filepath = fallback_path else: return None try: df_raw = pd.read_excel(filepath, skiprows=8) # Индексы столбцов в файле 1С: 1 — ФИО, 5 — Подразделение, 12 — Должность df_staff = df_raw.iloc[:, [1, 5, 12]].copy() df_staff.columns = ['ФИО', 'Подразделение', 'Должность'] df_staff = df_staff.dropna(subset=['ФИО']).reset_index(drop=True) df_staff = df_staff[~df_staff['ФИО'].astype(str).str.contains('Всего|Организация|Сотрудник|ФИО', case=False, na=False)] df_staff['fio_clean'] = df_staff['ФИО'].apply(normalize_fio) return df_staff except Exception as e: print(f"[❌] Ошибка загрузки штата из {filepath}: {e}") return None def load_absent_data(date_str): """ Загружает документально подтвержденные отсутствия сотрудников НАПРЯМУЮ из базы MS SQL 1С:ЗУП за указанную дату, а также автоматически обогащает их удаленщиками из static_reason_workers.csv. """ df_absent = None try: df_absent = fetch_zup_absences_from_sql(date_str) if df_absent is not None and not df_absent.empty: df_absent = df_absent.dropna(subset=['ФИО', 'Вид_отсутствия']) df_absent = df_absent[~df_absent['ФИО'].astype(str).str.contains('АО "ЛЕНМОРНИИПРОЕКТ"|Сотрудник', case=False, na=False)] df_absent['fio_clean'] = df_absent['ФИО'].apply(normalize_fio) df_absent = df_absent[['fio_clean', 'Вид_отсутствия']].dropna(subset=['fio_clean']) else: df_absent = pd.DataFrame(columns=['fio_clean', 'Вид_отсутствия']) except Exception as e: print(f"[⚠️] Ошибка SQL-выгрузки отсутствий за {date_str}: {e}") df_absent = pd.DataFrame(columns=['fio_clean', 'Вид_отсутствия']) # ⭐️ ОБОГАЩЕНИЕ УДАЛЕНЩИКАМИ ИЗ static_reason_workers.csv try: from config import DATA_DIR static_path = os.path.join(DATA_DIR, "static_reason_workers.csv") if os.path.exists(static_path): df_static = pd.read_csv(static_path, encoding='utf-8') if 'fio' in df_static.columns and 'reason' in df_static.columns: df_static['fio_clean'] = df_static['fio'].apply(normalize_fio) existing_fios = set(df_absent['fio_clean'].dropna().tolist()) if not df_absent.empty else set() new_rows = [] for _, s_row in df_static.iterrows(): if s_row['fio_clean'] not in existing_fios: new_rows.append({ 'fio_clean': s_row['fio_clean'], 'Вид_отсутствия': s_row['reason'] }) if new_rows: df_absent = pd.concat([df_absent, pd.DataFrame(new_rows)], ignore_index=True) except Exception as e: print(f"[⚠️] Ошибка обогащения удаленщиками в load_absent_data: {e}") return df_absent if not df_absent.empty else None def load_1c_data_smart(date_str, use_db=False): """ Универсальный загрузчик данных 1С (Штат и Отсутствия). Если use_db=True, приоритетно извлекает сохраненные данные из SQLite БД за целевую дату указанного снапшота. """ df_staff = None df_absent = None if use_db: df_staff = load_staff_from_db(date_str) df_absent = load_absences_from_db(date_str) # Если в базе нет записей за эту дату или use_db=False — загружаем из файлов/SQL if df_staff is None: df_staff = load_staff_data(date_str) if df_absent is None: df_absent = load_absent_data(date_str) return df_staff, df_absent def process_scud_anomalies(df_scud, exceptions=None): """ Вычисляет 'Первую активность' и маркирует аномалии СКУД. """ if exceptions is None: exceptions = load_exceptions() start_col = 'Начало_дня' if 'Начало_дня' in df_scud.columns else ('Начало дня' if 'Начало дня' in df_scud.columns else None) end_col = 'Конец_дня' if 'Конец_дня' in df_scud.columns else ('Конец дня' if 'Конец дня' in df_scud.columns else None) status_col = 'Статус' if 'Статус' in df_scud.columns else None first_activities = [] anomaly_flags = [] is_present_flags = [] for _, row in df_scud.iterrows(): fio = str(row.get('Сотрудник', row.get('fio_raw', ''))).strip() fio_clean = normalize_fio(fio) dept = str(row.get('Подразделение', '')).strip() pos = str(row.get('Должность', '')).strip() val_start = str(row.get(start_col, '')).strip() if start_col else '' val_end = str(row.get(end_col, '')).strip() if end_col else '' val_status = str(row.get(status_col, '')).strip().lower() if status_col else '' raw_first_event = row.get('raw_first_event', None) if pd.isna(raw_first_event) or str(raw_first_event).strip() in ['', 'None', 'nan', '00:00:00']: raw_first_event = None has_no_in = (val_start in ['нет входа', 'none', 'nan', '', '00:00:00', '00:00']) has_no_out = (val_end in ['нет выхода', 'none', 'nan', '', '00:00:00', '00:00']) first_act = '—' anom_flag = 'NONE' is_present = not ('отсутствовал' in val_status or has_no_in) if has_no_in: if raw_first_event is not None or not has_no_out: first_act = str(raw_first_event) if raw_first_event else val_end anom_flag = 'ANOMALY_NO_IN_HAS_ACTIVITY' if not is_excluded(fio, dept, pos, exceptions): is_present = False else: first_act = '—' is_present = False first_activities.append(first_act) anomaly_flags.append(anom_flag) is_present_flags.append(is_present) df_scud['Первая_активность'] = first_activities df_scud['anomaly_flag'] = anomaly_flags df_scud['Пришел'] = is_present_flags return df_scud def load_scud_data(target_date_type="today", snapshot_param=None): """ Загружает логи СКУД за 'today' или 'yesterday' НАПРЯМУЮ из локальной базы SQLite. """ target_date = DATE_TODAY if target_date_type == "today" else DATE_YESTERDAY df_scud = load_scud_from_db_by_snapshot(target_date, snapshot_param=snapshot_param) if df_scud is None or df_scud.empty: df_scud = load_scud_from_db_by_snapshot(DATE_TODAY, snapshot_param=snapshot_param) if df_scud is None or df_scud.empty: print(f"[ℹ️] В базе SQLite / файлах отсутствует СКУД за {target_date}. Возвращаем пустой набор.") return pd.DataFrame() exceptions = load_exceptions() if 'fio_clean' not in df_scud.columns: fio_raw_col = 'fio' if 'fio' in df_scud.columns else ('Сотрудник' if 'Сотрудник' in df_scud.columns else df_scud.columns[0]) df_scud['fio_raw'] = df_scud[fio_raw_col].astype(str) df_scud['fio_clean'] = df_scud['fio_raw'].apply(clean_scud_fio_light) if 'Начало_дня' not in df_scud.columns and 'time_in' in df_scud.columns: df_scud['Начало_дня'] = df_scud['time_in'] if 'Конец_дня' not in df_scud.columns and 'time_out' in df_scud.columns: df_scud['Конец_дня'] = df_scud['time_out'] if 'Первая_активность' not in df_scud.columns and 'first_activity' in df_scud.columns: df_scud['Первая_активность'] = df_scud['first_activity'] if 'Пришел' not in df_scud.columns and 'is_present' in df_scud.columns: df_scud['Пришел'] = df_scud['is_present'].astype(bool) df_scud['is_excluded'] = df_scud.apply( lambda r: is_excluded(r.get('fio', r.get('Сотрудник', '')), r.get('department', r.get('Подразделение', '')), r.get('position', r.get('Должность', '')), exceptions), axis=1 ) return df_scud def load_all_data(): """ Загружает слитные датасеты за Сегодня и за Вчера напрямую из базы SQLite и 1С. """ df_scud_today = load_scud_data(target_date_type="today") try: df_scud_yesterday = load_scud_data(target_date_type="yesterday") except FileNotFoundError: df_scud_yesterday = df_scud_today.copy() df_staff_today, df_absent_today = load_1c_data_smart(DATE_TODAY) df_staff_yesterday, df_absent_yesterday = load_1c_data_smart(DATE_YESTERDAY) return ( df_scud_today, df_scud_yesterday, df_staff_today, df_absent_today, df_staff_yesterday, df_absent_yesterday )