142 lines
6.5 KiB
Python
142 lines
6.5 KiB
Python
"""
|
|
===============================================================================
|
|
FILE: services/data_loader.py
|
|
ROLE: Надежная загрузка штата и отсутствий (MS SQL ЗУП -> Резервный Excel).
|
|
===============================================================================
|
|
"""
|
|
|
|
import os
|
|
import warnings
|
|
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, 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)
|
|
print(f" [✓] Реестр удаленщиков: добавлено {len(new_rows)} чел. из static_reason_workers.csv")
|
|
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 |