137 lines
6.0 KiB
Python
137 lines
6.0 KiB
Python
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С
|
|
from modules.zup_extractor import fetch_zup_absences_from_sql
|
|
|
|
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С:ЗУП за указанную дату.
|
|
"""
|
|
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)
|
|
return df_absent[['fio_clean', 'Вид_отсутствия']].dropna(subset=['fio_clean'])
|
|
else:
|
|
return None
|
|
except Exception as e:
|
|
print(f"[⚠️] Ошибка SQL-выгрузки отсутствий за {date_str}: {e}")
|
|
return None
|
|
|
|
|
|
def load_scud_data(target_date_type="today"):
|
|
"""Загружает файл СКУД за 'today' или 'yesterday'."""
|
|
target_date = DATE_TODAY if target_date_type == "today" else DATE_YESTERDAY
|
|
scud_path = find_dated_file("Сотрудники", target_date, search_dirs=[SCUD_DIR])
|
|
|
|
if not scud_path:
|
|
scud_path = find_dated_file("Сотрудники", DATE_TODAY, search_dirs=[SCUD_DIR])
|
|
if not scud_path:
|
|
raise FileNotFoundError(f"Файл СКУД [{target_date_type}] за {target_date} не найден в {SCUD_DIR}!")
|
|
|
|
df_scud = pd.read_excel(scud_path)
|
|
exceptions = load_exceptions()
|
|
|
|
df_scud['fio_raw'] = df_scud['Сотрудник'].astype(str) if 'Сотрудник' in df_scud.columns else df_scud[df_scud.columns[0]].astype(str)
|
|
df_scud['fio_clean'] = df_scud['fio_raw'].apply(clean_scud_fio_light)
|
|
|
|
start_col = 'Начало_дня' if 'Начало_дня' in df_scud.columns else ('Начало дня' if 'Начало дня' in df_scud.columns else None)
|
|
status_col = 'Статус' if 'Статус' in df_scud.columns else None
|
|
|
|
def is_person_present(row):
|
|
val_status = str(row.get(status_col, '')).strip().lower() if status_col else ''
|
|
val_time = str(row.get(start_col, '')).strip() if start_col else ''
|
|
|
|
if 'отсутствовал' in val_status or val_time in ['нет входа', 'none', 'nan', '', '00:00']:
|
|
return False
|
|
return True
|
|
|
|
df_scud['Пришел'] = df_scud.apply(is_person_present, axis=1)
|
|
|
|
df_scud['is_excluded'] = df_scud.apply(
|
|
lambda r: is_excluded(r.get('Сотрудник', ''), r.get('Подразделение', ''), r.get('Должность', ''), exceptions),
|
|
axis=1
|
|
)
|
|
return df_scud
|
|
|
|
|
|
def load_all_data():
|
|
"""
|
|
Загружает слитные датасеты за Сегодня и за Вчера.
|
|
Если данных за Сегодня нет, возвращает для них None.
|
|
"""
|
|
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()
|
|
|
|
# Загружаем 1С за Сегодня (может быть None)
|
|
df_staff_today = load_staff_data(DATE_TODAY)
|
|
df_absent_today = load_absent_data(DATE_TODAY)
|
|
|
|
# Загружаем 1С за Вчера (гарантированно есть)
|
|
df_staff_yesterday = load_staff_data(DATE_YESTERDAY)
|
|
df_absent_yesterday = load_absent_data(DATE_YESTERDAY)
|
|
|
|
return (
|
|
df_scud_today,
|
|
df_scud_yesterday,
|
|
df_staff_today,
|
|
df_absent_today,
|
|
df_staff_yesterday,
|
|
df_absent_yesterday
|
|
) |