""" =============================================================================== FILE: services/data_loader.py ROLE: Надежная загрузка штата и отсутствий (MS SQL ЗУП -> Резервный Excel). =============================================================================== """ import os import warnings from datetime import datetime # <-- ДОБАВИТЬ ЭТУ СТРОКУ import pandas as pd from config import ( normalize_fio, clean_scud_fio_light, load_exceptions, DATE_TODAY, DATE_YESTERDAY, find_dated_file, ZUP_1C_DIR, SCUD_DIR, DATA_DIR ) 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 load_staff_data(date_str): """Загружает файл Штатного расписания 1С.""" 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) 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_from_excel(date_str): """Резервное чтение файла Отсутствия_ДД_ММ_ГГГГ.xlsx.""" filepath = find_dated_file("Отсутствия", date_str) if not filepath: return None try: for skip in [0, 1, 2, 3, 4, 5, 8]: df_try = pd.read_excel(filepath, skiprows=skip) fio_col = None reason_col = None for col in df_try.columns: c_str = str(col).lower() if ('фио' in c_str or 'сотрудник' in c_str) and fio_col is None: fio_col = col if ('вид' in c_str or 'причина' in c_str or 'отсутств' in c_str) and reason_col is None: reason_col = col if fio_col and reason_col: df_res = df_try[[fio_col, reason_col]].copy() df_res.columns = ['ФИО', 'Вид_отсутствия'] df_res = df_res.dropna(subset=['ФИО', 'Вид_отсутствия']) df_res = df_res[~df_res['ФИО'].astype(str).str.contains('Всего|Организация|Сотрудник|ФИО|ЛЕНМОРНИИПРОЕКТ', case=False, na=False)] df_res['fio_clean'] = df_res['ФИО'].apply(normalize_fio) print(f" [✓] Резервный Excel: загружено {len(df_res)} записей отсутствий из {os.path.basename(filepath)}") return df_res[['fio_clean', 'Вид_отсутствия']].dropna(subset=['fio_clean']) except Exception as e: print(f" [⚠️] Ошибка чтения резервного Excel отсутствий {filepath}: {e}") return None def load_absent_data(date_str): """ Загружает отсутствия из MS SQL 1С:ЗУП, а при сбое — из локального Excel. """ df_absent = None try: df_absent = fetch_zup_absences_from_sql(date_str) if df_absent is not None and not df_absent.empty: print(f" [✓] MS SQL ЗУП: получено {len(df_absent)} записей отсутствий за {date_str}") 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']) except Exception as e: print(f" [⚠️] Ошибка подключения к MS SQL ЗУП за {date_str}: {e}") # Резервный источник: Excel файл из 1С с сетевой шары if df_absent is None or df_absent.empty: df_absent = load_absent_from_excel(date_str) if df_absent is None: df_absent = pd.DataFrame(columns=['fio_clean', 'Вид_отсутствия']) # Обогащение удаленщиками из static_reason_workers.csv с ротацией просроченных записей try: static_path = os.path.join(DATA_DIR, "static_reason_workers.csv") if os.path.exists(static_path): df_static = pd.read_csv(static_path, dtype=str, on_bad_lines='skip').fillna("") for col in ['fio', 'reason', 'department', 'date_from', 'date_to']: if col not in df_static.columns: df_static[col] = "" today_date = datetime.now().date() # Разбор целевой даты отчета clean_target_str = str(date_str).replace('_', '.') try: clean_target_date = datetime.strptime(clean_target_str, "%d.%m.%Y").date() except ValueError: clean_target_date = today_date def parse_date_safe(d_val): if not d_val or str(d_val).lower() in ['nan', 'none', '', 'nat']: return None s = str(d_val).strip().replace('_', '.') for fmt in ("%d.%m.%Y", "%Y-%m-%d"): try: return datetime.strptime(s, fmt).date() except ValueError: pass return None active_for_file = [] new_rows_for_report = [] file_changed = False existing_fios = set(df_absent['fio_clean'].dropna().tolist()) if not df_absent.empty else set() for _, s_row in df_static.iterrows(): fio_raw = str(s_row.get('fio', '')).strip() if not fio_raw: continue fio_c = normalize_fio(fio_raw) reason = str(s_row.get('reason', '')).strip() or "Дистанционная работа" d_from = parse_date_safe(s_row.get('date_from')) d_to = parse_date_safe(s_row.get('date_to')) # 1. Физическая ротация просроченных: только если текущий реальный день (today) строго больше date_to if d_to is not None and today_date > d_to: print(f" [🧹] Удаленка истекла: {fio_raw} (до {d_to.strftime('%d.%m.%Y')}). Удалена из CSV.") file_changed = True continue active_for_file.append(s_row.to_dict()) # 2. Проверка действия удаленки на дату формируемого отчета: # Если date_from не указана — действует всегда до date_to is_after_start = (d_from is None) or (clean_target_date >= d_from) is_before_end = (d_to is None) or (clean_target_date <= d_to) if is_after_start and is_before_end: if fio_c not in existing_fios: new_rows_for_report.append({ 'fio_clean': fio_c, 'Вид_отсутствия': reason }) existing_fios.add(fio_c) # Перезаписываем CSV только если реально были удалены просроченные сотрудники if file_changed: pd.DataFrame(active_for_file).to_csv(static_path, index=False, encoding='utf-8') print(" [✓] Файл static_reason_workers.csv синхронизирован без просроченных записей.") if new_rows_for_report: df_absent = pd.concat([df_absent, pd.DataFrame(new_rows_for_report)], ignore_index=True) print(f" [✓] Реестр удаленщиков: добавлено {len(new_rows_for_report)} чел. в отчет за {date_str}") except Exception as e: print(f" [⚠️] Ошибка обработки static_reason_workers.csv: {e}") return df_absent if not df_absent.empty else None def load_1c_data_smart(date_str, use_db=False): 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) 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