feat(reports): add simplified report builder, fix nan/inf crash and separate unhired/no-pass staff

This commit is contained in:
2026-09-24 09:25:57 +03:00
parent 9c89b6592e
commit 30651672ab
10 changed files with 437 additions and 89 deletions
+33 -9
View File
@@ -18,12 +18,17 @@ def generate_detailed_excel(merged_df, date_str="20.08.2026", filename=None):
if not filename:
filename = f"{format_date_ru(date_clean)} отчет.xlsx"
df_export = merged_df[merged_df.get('is_excluded', False) == False].copy() if merged_df is not None and not merged_df.empty else pd.DataFrame()
is_exc = merged_df.get('is_excluded', False) == True
is_not_hired = merged_df.get('not_hired_yet', False) == True
# Исключаем из отчета за вчера сотрудников-исключений И тех, кто еще не принят на работу
df_export = merged_df[(~is_exc) & (~is_not_hired)].copy() if merged_df is not None and not merged_df.empty else pd.DataFrame()
target_dir = get_dated_reports_dir(date_clean)
output_path = os.path.join(target_dir, filename)
wb = xlsxwriter.Workbook(output_path)
# Включаем опцию защиты nan_inf_to_errors
wb = xlsxwriter.Workbook(output_path, {'nan_inf_to_errors': True})
ws = wb.add_worksheet("Детальный_отчет")
fmt_date_lbl = wb.add_format({'font_name': 'Arial', 'font_size': 10, 'bold': True})
@@ -46,18 +51,32 @@ def generate_detailed_excel(merged_df, date_str="20.08.2026", filename=None):
for idx, row in df_export.reset_index(drop=True).iterrows():
row_num = 4 + idx
is_present = row.get('Пришел', False)
is_present = bool(row.get('Пришел', False))
absence_reason = row.get('Вид_отсутствия', '')
has_reason = pd.notna(absence_reason) and str(absence_reason).strip() != ''
has_reason = pd.notna(absence_reason) and str(absence_reason).strip() not in ['', 'nan', 'None']
in_val = str(row.get(start_col, 'Нет входа')).strip()
if in_val.lower() in ['nan', 'none']: in_val = 'Нет входа'
out_val = str(row.get(end_col, 'Нет выхода')).strip()
in_building_str = str(row.get(hours_col, '00:00'))
if out_val.lower() in ['nan', 'none']: out_val = 'Нет выхода'
in_building_str = str(row.get(hours_col, '00:00')).strip()
if in_building_str.lower() in ['nan', 'none']: in_building_str = '00:00'
first_act_val = str(row.get('Первая_активность', '—')).strip()
if first_act_val.lower() in ['nan', 'none']: first_act_val = '—'
has_first_act = first_act_val not in ['—', '', 'None', 'nan']
deviation_val = calculate_deviation(in_building_str, reason=absence_reason if has_reason else "", norm_hours=8, lunch_minutes=30)
dept_scud_val = row.get('department_scud', row.get('department', row.get('Подразделение', '')))
dept_raw = row.get('department_scud', row.get('department', row.get('Подразделение', '')))
dept_scud_val = str(dept_raw).strip() if pd.notna(dept_raw) else '—'
if dept_scud_val.lower() in ['nan', 'none']: dept_scud_val = '—'
fio_raw = row.get('Сотрудник', '')
fio_val = str(fio_raw).strip() if pd.notna(fio_raw) else ''
if fio_val.lower() in ['nan', 'none']: fio_val = ''
row_color = None
if is_present and has_reason:
@@ -67,24 +86,29 @@ def generate_detailed_excel(merged_df, date_str="20.08.2026", filename=None):
elif not is_present and not has_reason and not has_first_act:
row_color = '#FCE4D6'
val_h_str = str(absence_reason) if has_reason else ""
val_h_str = str(absence_reason).strip() if has_reason else ""
if val_h_str.lower() in ['nan', 'none']: val_h_str = ""
lines_count = math.ceil(len(val_h_str) / chars_per_line_h) if len(val_h_str) > chars_per_line_h else 1
ws.set_row(row_num, max(lines_count * 18, 20))
values = [
(idx + 1, 'center', False),
(row.get('Сотрудник', ''), 'left', False),
(fio_val, 'left', False),
(dept_scud_val, 'center', False),
(in_val, 'center', False),
(first_act_val, 'center', False),
(out_val, 'center', False),
(in_building_str, 'center', False),
(absence_reason if has_reason else '', 'left', True),
(val_h_str, 'left', True),
(8, 'center', False),
(deviation_val, 'center', False)
]
for col_idx, (val, align_type, is_wrap) in enumerate(values):
# Санитизация для исключения передачи float('nan') в _write_number
if pd.isna(val) or val is None or str(val).strip().lower() == 'nan':
val = ""
fmt = create_xlsx_format(wb, font_name='Arial', font_size=10, bg_color=row_color, align=align_type, wrap=is_wrap)
ws.write(row_num, col_idx, val, fmt)
+147
View File
@@ -0,0 +1,147 @@
"""
===============================================================================
FILE: services/reports/simplified_builder.py
ROLE: Генератор книги "Упрощенный отчет за ДД.ММ.ГГГГг..xlsx"
на основе агрегированных данных СКУД и 1С.
===============================================================================
"""
import os
import openpyxl
from openpyxl.styles import Font, Alignment, Border, Side
from datetime import datetime, time
import pandas as pd
from config import REPORTS_DIR
from services.reports.styles import get_dated_reports_dir
def format_fio_initials(full_fio: str) -> str:
"""Преобразует 'Иванов Иван Иванович' в 'Иванов И.И.'"""
if not full_fio or pd.isna(full_fio):
return ""
parts = str(full_fio).strip().split()
if len(parts) >= 3:
return f"{parts[0]} {parts[1][0]}.{parts[2][0]}."
elif len(parts) == 2:
return f"{parts[0]} {parts[1][0]}."
return str(full_fio).strip()
def parse_time_cell(val, default_empty="Нет входа (0:00)"):
"""Преобразует строку времени СКУД в time(hh, mm) либо оставляет текстовую заглушку."""
if not val or pd.isna(val):
return default_empty
s = str(val).strip()
if s in ["Нет входа", "—", "", "nan", "None", "00:00:00", "00:00"]:
return default_empty
if s in ["Нет выхода"]:
return "Нет выхода (23:59)"
# Если уже пришёл time
if isinstance(val, time):
return val
try:
parts = s.split(":")
h = int(parts[0])
m = int(parts[1]) if len(parts) > 1 else 0
return time(h, m)
except Exception:
return s
def generate_simplified_excel(df_merged: pd.DataFrame, date_str: str, filename: str = None) -> str:
"""
Генерирует Excel-файл упрощенного отчета точно по образцу Ориона.
Структура:
A: № | B: Подразделение | C: Сотрудник | D: Должность | E: Начало дня | F: Конец дня
"""
clean_date = str(date_str).replace('_', '.')
if not filename:
filename = f"Упрощенный отчет за {clean_date}г..xlsx"
target_dir = get_dated_reports_dir(clean_date)
output_path = os.path.join(target_dir, filename)
# Исключаем тех, кто не принят на работу и исключения
is_exc = df_merged.get('is_excluded', False) == True
is_not_hired = df_merged.get('not_hired_yet', False) == True
df_src = df_merged[(~is_exc) & (~is_not_hired)].copy()
# Подготовка данных
rows_to_sort = []
for _, r in df_src.iterrows():
fio_raw = r.get('Сотрудник', r.get('ФИО', ''))
dept = str(r.get('Подразделение', '—')).strip()
pos = str(r.get('Должность', '—')).strip()
if pos.lower() in ['nan', 'none', '']: pos = '—'
if dept.lower() in ['nan', 'none', '']: dept = '—'
fio_short = format_fio_initials(fio_raw)
t_in = parse_time_cell(r.get('Начало_дня'), default_empty="Нет входа (0:00)")
t_out = parse_time_cell(r.get('Конец_дня'), default_empty="Нет выхода (23:59)")
rows_to_sort.append({
'dept': dept,
'fio_short': fio_short,
'pos': pos,
't_in': t_in,
't_out': t_out
})
# Сортировка: Подразделение (А-Я), затем Сотрудник (А-Я)
rows_sorted = sorted(rows_to_sort, key=lambda x: (x['dept'].lower(), x['fio_short'].lower()))
# Создание книги openpyxl для точного соблюдения структуры образца
wb = openpyxl.Workbook()
ws = wb.active
ws.title = "Лист1"
# Стили по образцу
font_main = Font(name="Calibri", size=11, bold=False)
align_center = Alignment(horizontal="center", vertical="center")
border_thin = Border(
left=Side(style="thin"),
right=Side(style="thin"),
top=Side(style="thin"),
bottom=Side(style="thin")
)
# Заголовок
ws["B1"] = f"Упрощенный отчет: c {clean_date} по {clean_date}"
ws["B1"].font = font_main
# Шапка таблицы (Строка 3)
headers = [None, "Подразделение", "Сотрудник", "Должность", "Начало дня", "Конец дня"]
for col_idx, h_text in enumerate(headers, start=1):
cell = ws.cell(row=3, column=col_idx, value=h_text)
cell.font = font_main
cell.alignment = align_center
cell.border = border_thin
# Заполнение строк данных (с 4 строки)
for idx, item in enumerate(rows_sorted, start=1):
row_num = 3 + idx
vals = [idx, item['dept'], item['fio_short'], item['pos'], item['t_in'], item['t_out']]
for col_idx, val in enumerate(vals, start=1):
cell = ws.cell(row=row_num, column=col_idx, value=val)
cell.font = font_main
cell.alignment = align_center
cell.border = border_thin
# Числовой формат времени h:mm
if isinstance(val, time):
cell.number_format = "h:mm"
# Настройка ширины колонок по образцу
widths = {'A': 13.0, 'B': 25.0, 'C': 27.0, 'D': 80.0, 'E': 21.5, 'F': 23.3}
for col_letter, w in widths.items():
ws.column_dimensions[col_letter].width = w
# Создание пустых Лист2 и Лист3 как в оригинальном шаблоне
wb.create_sheet("Лист2")
wb.create_sheet("Лист3")
wb.save(output_path)
return output_path
+52 -24
View File
@@ -20,7 +20,7 @@ def generate_summary_excel(merged_df, date_str="21.08.2026", filename=None):
target_dir = get_dated_reports_dir(date_clean)
output_path = os.path.join(target_dir, filename)
wb = xlsxwriter.Workbook(output_path)
wb = xlsxwriter.Workbook(output_path, {'nan_inf_to_errors': True})
ws = wb.add_worksheet("Лист_1")
ws.outline_settings(visible=True, symbols_below=False, symbols_right=False, auto_style=False)
@@ -38,18 +38,24 @@ def generate_summary_excel(merged_df, date_str="21.08.2026", filename=None):
ws.write(1, 0, "", fmt_empty)
ws.write(1, 1, "", fmt_empty)
is_not_hired = merged_df.get('not_hired_yet', False) == True
is_no_pass = merged_df.get('no_scud_pass', False) == True
is_exc = merged_df.get('is_excluded', False) == True
# "По списку" строго по официальному штату 1С
staff_total = len(merged_df[~is_not_hired])
ws.set_row(2, 20)
ws.write(2, 0, "По списку", fmt_tot_l)
ws.write(2, 1, len(merged_df), fmt_tot_r)
ws.write(2, 1, staff_total, fmt_tot_r)
current_row = 3
is_no_pass = merged_df['no_scud_pass'] == True if 'no_scud_pass' in merged_df.columns else False
is_exc = merged_df.get('is_excluded', False) == True
# 1. Неизвестно
unexplained = merged_df[
(~is_not_hired) &
(merged_df['Пришел'] == False) &
(merged_df['Вид_отсутствия'].isna() | (merged_df['Вид_отсутствия'].astype(str).str.strip() == '')) &
(merged_df['Вид_отсутствия'].isna() | (merged_df['Вид_отсутствия'].astype(str).str.strip().isin(['', 'nan', 'None']))) &
(~is_no_pass) & (~is_exc)
]
fmt_unexp_hl = create_xlsx_format(wb, bg_color='#FCE4D6', bold=True, align="left")
@@ -64,34 +70,54 @@ def generate_summary_excel(merged_df, date_str="21.08.2026", filename=None):
for fio in sorted(unexplained['Сотрудник'].dropna().unique()):
ws.set_row(current_row, 20, None, {'level': 1, 'hidden': False})
ws.write(current_row, 0, fio, fmt_unexp_rl)
ws.write(current_row, 0, str(fio), fmt_unexp_rl)
ws.write(current_row, 1, "", fmt_unexp_rr)
current_row += 1
# 2. Нет пропуска
no_pass_df = merged_df[is_no_pass & (~is_exc)] if 'no_scud_pass' in merged_df.columns else pd.DataFrame()
# 2. Нет пропуска (в штате 1С есть, но карты СКУД нет)
no_pass_df = merged_df[is_no_pass & (~is_exc) & (~is_not_hired)]
fmt_np_hl = create_xlsx_format(wb, bg_color='#E1F5FE', bold=True, align="left")
fmt_np_hr = create_xlsx_format(wb, bg_color='#E1F5FE', bold=True, align="right")
fmt_np_rl = create_xlsx_format(wb, bg_color='#E1F5FE', bold=False, align="left")
fmt_np_rr = create_xlsx_format(wb, bg_color='#E1F5FE', bold=False, align="right")
ws.set_row(current_row, 20)
ws.write(current_row, 0, "Нет пропуска", fmt_np_hl)
ws.write(current_row, 0, "Нет в СКУД", fmt_np_hl)
ws.write(current_row, 1, len(no_pass_df), fmt_np_hr)
current_row += 1
if not no_pass_df.empty:
for fio in sorted(no_pass_df['Сотрудник'].dropna().unique()):
ws.set_row(current_row, 20, None, {'level': 1, 'hidden': False})
ws.write(current_row, 0, fio, fmt_np_rl)
ws.write(current_row, 0, str(fio), fmt_np_rl)
ws.write(current_row, 1, "", fmt_np_rr)
current_row += 1
# 3. Официальные отсутствия
# 3. Не приняты на работу (в СКУД есть, но в 1С приказа еще нет)
not_hired_df = merged_df[is_not_hired & (~is_exc)]
fmt_nh_hl = create_xlsx_format(wb, bg_color='#FFF0F5', bold=True, align="left")
fmt_nh_hr = create_xlsx_format(wb, bg_color='#FFF0F5', bold=True, align="right")
fmt_nh_rl = create_xlsx_format(wb, bg_color='#FFF0F5', bold=False, align="left")
fmt_nh_rr = create_xlsx_format(wb, bg_color='#FFF0F5', bold=False, align="right")
ws.set_row(current_row, 20)
ws.write(current_row, 0, "Нет в ЗУП", fmt_nh_hl)
ws.write(current_row, 1, len(not_hired_df), fmt_nh_hr)
current_row += 1
if not not_hired_df.empty:
for fio in sorted(not_hired_df['Сотрудник'].dropna().unique()):
ws.set_row(current_row, 20, None, {'level': 1, 'hidden': False})
ws.write(current_row, 0, str(fio), fmt_nh_rl)
ws.write(current_row, 1, "", fmt_nh_rr)
current_row += 1
# 4. Официальные отсутствия
reason_clean = merged_df['Вид_отсутствия'].astype(str).str.lower()
is_remote_reason = reason_clean.str.contains('удален|дистанцион', regex=True, na=False)
absent_only = merged_df[
(~is_not_hired) &
(merged_df['Пришел'] == False) &
(merged_df['Вид_отсутствия'].notna()) &
(~merged_df['Вид_отсутствия'].astype(str).str.startswith('Исключение')) &
@@ -108,7 +134,7 @@ def generate_summary_excel(merged_df, date_str="21.08.2026", filename=None):
fmt_cat_rr = create_xlsx_format(wb, bg_color=hex_c, bold=False, align="right")
ws.set_row(current_row, 20)
ws.write(current_row, 0, cat_name, fmt_cat_hl)
ws.write(current_row, 0, str(cat_name), fmt_cat_hl)
ws.write(current_row, 1, len(group), fmt_cat_hr)
current_row += 1
@@ -116,14 +142,16 @@ def generate_summary_excel(merged_df, date_str="21.08.2026", filename=None):
for _, row in group.sort_values(by='Сотрудник').iterrows():
fio = row.get('Сотрудник', '')
detail_val = row.get('detailed_reason', row.get('comment', '')) if is_other_category else ""
if pd.isna(detail_val):
detail_val = ""
ws.set_row(current_row, 20, None, {'level': 1, 'hidden': True, 'collapsed': True})
ws.write(current_row, 0, fio, fmt_cat_rl)
ws.write(current_row, 1, detail_val, fmt_cat_rr)
ws.write(current_row, 0, str(fio), fmt_cat_rl)
ws.write(current_row, 1, str(detail_val), fmt_cat_rr)
current_row += 1
# 4. Итого на работе
exc_without_doc = merged_df[is_exc & (merged_df['Вид_отсутствия'].isna() | (merged_df['Вид_отсутствия'].astype(str).str.strip().isin(['', 'nan', 'Исключение'])))]
present_scud = merged_df[(merged_df['Пришел'] == True) & (~is_exc)]
# 5. Итого на работе (только принятые)
exc_without_doc = merged_df[(~is_not_hired) & is_exc & (merged_df['Вид_отсутствия'].isna() | (merged_df['Вид_отсутствия'].astype(str).str.strip().isin(['', 'nan', 'Исключение'])))]
present_scud = merged_df[(~is_not_hired) & (merged_df['Пришел'] == True) & (~is_exc)]
total_present_count = len(present_scud) + len(exc_without_doc)
fmt_pres_hl = create_xlsx_format(wb, bg_color='#E2EFDA', bold=True, align="left")
@@ -134,8 +162,8 @@ def generate_summary_excel(merged_df, date_str="21.08.2026", filename=None):
ws.write(current_row, 1, total_present_count, fmt_pres_hr)
current_row += 1
# 5. Удаленная работа
remote_home = merged_df[(merged_df['Пришел'] == False) & is_remote_reason & (~is_exc)]
# 6. Удаленная работа
remote_home = merged_df[(~is_not_hired) & (merged_df['Пришел'] == False) & is_remote_reason & (~is_exc)]
fmt_rem_hl = create_xlsx_format(wb, bg_color='#E8F8F5', bold=True, align="left")
fmt_rem_hr = create_xlsx_format(wb, bg_color='#E8F8F5', bold=True, align="right")
fmt_rem_rl = create_xlsx_format(wb, bg_color='#E8F8F5', bold=False, align="left")
@@ -149,13 +177,13 @@ def generate_summary_excel(merged_df, date_str="21.08.2026", filename=None):
if not remote_home.empty:
for fio in sorted(remote_home['Сотрудник'].dropna().unique()):
ws.set_row(current_row, 20, None, {'level': 1, 'hidden': True, 'collapsed': True})
ws.write(current_row, 0, fio, fmt_rem_rl)
ws.write(current_row, 0, str(fio), fmt_rem_rl)
ws.write(current_row, 1, "", fmt_rem_rr)
current_row += 1
# 6. Аномалии СКУД и 1С
# 7. Аномалии СКУД и 1С
anomalies = merged_df[
(~is_exc) & (
(~is_not_hired) & (~is_exc) & (
((merged_df['Пришел'] == True) & (merged_df['Вид_отсутствия'].notna()) &
(~merged_df['Вид_отсутствия'].astype(str).str.startswith('Исключение')) &
(~is_remote_reason) &
@@ -182,8 +210,8 @@ def generate_summary_excel(merged_df, date_str="21.08.2026", filename=None):
lines_count = math.ceil(len(reason_text) / chars_per_line_b) if len(reason_text) > chars_per_line_b else 1
ws.set_row(current_row, max(lines_count * 18, 20), None, {'level': 1, 'hidden': True, 'collapsed': True})
ws.write(current_row, 0, fio, fmt_anom_rl)
ws.write(current_row, 1, reason_text, fmt_anom_rr)
ws.write(current_row, 0, str(fio), fmt_anom_rl)
ws.write(current_row, 1, str(reason_text), fmt_anom_rr)
current_row += 1
ws.set_column(0, 0, 45)
+121 -33
View File
@@ -138,7 +138,8 @@ def merge_scud_and_1c(
return pd.DataFrame(columns=[
'Сотрудник', 'fio_clean', 'Подразделение', 'Должность',
'Начало_дня', 'Первая_активность', 'Конец_дня', 'Находился_в_здании',
'Пришел', 'anomaly_flag', 'причина отсутствия', 'Вид_отсутствия', 'is_excluded'
'Пришел', 'anomaly_flag', 'причина отсутствия', 'Вид_отсутствия',
'is_excluded', 'not_hired_yet', 'no_scud_pass'
])
synonyms = get_department_synonyms_dict()
@@ -147,15 +148,63 @@ def merge_scud_and_1c(
df_scud_agg = aggregate_scud_by_person(df_scud)
df_staff_agg = select_best_zup_position(df_staff_1c, df_scud_agg)
df_res = df_scud_agg.copy() if df_scud_agg is not None and not df_scud_agg.empty else df_staff_agg.copy()
staff_fios = set(df_staff_agg['fio_clean'].dropna().tolist()) if df_staff_agg is not None and not df_staff_agg.empty else set()
scud_dict = df_scud_agg.set_index('fio_clean').to_dict('index') if df_scud_agg is not None and not df_scud_agg.empty else {}
if "Сотрудник" in df_res.columns:
df_res["fio_clean"] = df_res["Сотрудник"].apply(normalize_fio)
elif "ФИО" in df_res.columns:
merged_rows = []
# 1. Формируем строки по официальному штату 1С
if df_staff_agg is not None and not df_staff_agg.empty:
for _, s_row in df_staff_agg.iterrows():
fio_c = s_row.get('fio_clean', '')
r = dict(s_row)
if 'Сотрудник' not in r or pd.isna(r['Сотрудник']) or str(r['Сотрудник']).strip() == '':
r['Сотрудник'] = r.get('ФИО', fio_c)
if fio_c in scud_dict:
# Сотрудник есть в СКУД — берем короткую аббревиатуру отдела из СКУД
scud_data = scud_dict[fio_c]
dept_scud = str(scud_data.get('Подразделение', '')).strip()
if dept_scud and dept_scud.lower() not in ['nan', 'none', '—', 'без подразделения']:
r['Подразделение'] = dept_scud
r['Начало_дня'] = scud_data.get('Начало_дня', 'Нет входа')
r['Первая_активность'] = scud_data.get('Первая_активность', '—')
r['Конец_дня'] = scud_data.get('Конец_дня', 'Нет выхода')
r['Находился_в_здании'] = scud_data.get('Находился_в_здании', '00:00')
r['Пришел'] = bool(scud_data.get('Пришел', False))
r['anomaly_flag'] = scud_data.get('anomaly_flag', 'NONE')
r['no_scud_pass'] = False
r['not_hired_yet'] = False
else:
# Сотрудника нет в СКУД (Нет пропуска)
r['Начало_дня'] = 'Нет входа'
r['Первая_активность'] = '—'
r['Конец_дня'] = 'Нет выхода'
r['Находился_в_здании'] = '00:00'
r['Пришел'] = False
r['anomaly_flag'] = 'NONE'
r['no_scud_pass'] = True
r['not_hired_yet'] = False
merged_rows.append(r)
# 2. Сотрудники из СКУД, которых еще нет в 1С (Не приняты на работу)
if df_scud_agg is not None and not df_scud_agg.empty:
for _, scud_row in df_scud_agg.iterrows():
fio_c = scud_row.get('fio_clean', '')
if fio_c not in staff_fios:
r = dict(scud_row)
r['not_hired_yet'] = True
r['no_scud_pass'] = False
if 'Должность' not in r or pd.isna(r['Должность']):
r['Должность'] = '—'
merged_rows.append(r)
df_res = pd.DataFrame(merged_rows)
if "Сотрудник" not in df_res.columns and "ФИО" in df_res.columns:
df_res["Сотрудник"] = df_res["ФИО"]
df_res["fio_clean"] = df_res["ФИО"].apply(normalize_fio)
elif "fio_clean" not in df_res.columns:
df_res["fio_clean"] = ""
for col, default_val in [
('Начало_дня', 'Нет входа'),
@@ -163,20 +212,46 @@ def merge_scud_and_1c(
('Конец_дня', 'Нет выхода'),
('Находился_в_здании', '00:00'),
('Пришел', False),
('anomaly_flag', 'NONE')
('anomaly_flag', 'NONE'),
('not_hired_yet', False),
('no_scud_pass', False)
]:
if col not in df_res.columns:
df_res[col] = default_val
# Словарь синонимов и принудительное сокращение длинных отделов 1С до аббревиатур
reverse_synonyms = {v.lower(): k.upper() for k, v in synonyms.items()}
direct_synonyms = {k.lower(): k.upper() for k in synonyms.keys()}
all_dept_map = {**reverse_synonyms, **direct_synonyms, "отдел внутреннего контроля": "ОВК", "отдел вневедомственного контроля": "ОВК"}
all_dept_map = {
**reverse_synonyms,
**direct_synonyms,
"отдел внутреннего контроля": "ОВК",
"отдел вневедомственного контроля": "ОВК",
"отдел авторского надзора и технического аудита": "ОАН",
"отдел инженерных изысканий": "ОИЗ",
"правовое управление": "ПУ",
"макетная мастерская": "ММ",
"отдел автоматизации": "ОА",
"испытательная геотехническая лаборатория лабораторного центра": "ИГТЛЛ",
"отдел экономики, смет и организации строительства": "ОЭС",
"отдел электротехники, связи и пожарной автоматики": "ОЭСС",
"бетонная лаборатория": "БЛ",
"управление главных инженеров проектов №1": "УГИП №1",
"управление главных инженеров проектов №2": "УГИП №2",
"строительный отдел": "СО",
"гидротехническая экспедиция": "ГЭ",
"планово-экономический отдел": "ПЭО",
"конструкторский отдел": "КО",
"отдел тепловодоснабжения и канализации": "ОТВК",
"электротехнический отдел": "ЭТО"
}
if "Подразделение" in df_res.columns:
df_res["Подразделение"] = df_res["Подразделение"].apply(
lambda d: all_dept_map.get(str(d).strip().lower(), str(d).strip())
lambda d: all_dept_map.get(str(d).strip().lower(), str(d).strip()) if pd.notna(d) else "—"
)
# Привязка кадровых документов 1С
absences_map = {}
if df_absences_1c is not None and not df_absences_1c.empty:
fio_col = next((c for c in ["fio_clean", "ФИО", "Сотрудник"] if c in df_absences_1c.columns), None)
@@ -186,13 +261,12 @@ def merge_scud_and_1c(
for _, row in df_absences_1c.iterrows():
fio = normalize_fio(str(row[fio_col]))
reason = str(row[reason_col]).strip()
if reason and reason.lower() != "nan":
if reason and reason.lower() not in ["nan", "none"]:
absences_map[fio] = reason
manual_reasons_map = {}
try:
from services.manual_absences_repo import get_active_manual_absences_for_date
# date_clean берется из даты контекста либо из текущих суток
target_date_val = df_res.get('Дата', pd.Series()).iloc[0] if 'Дата' in df_res.columns and not df_res.empty else None
if target_date_val:
m_records = get_active_manual_absences_for_date(str(target_date_val))
@@ -205,6 +279,7 @@ def merge_scud_and_1c(
df_res["причина отсутствия"] = df_res["fio_clean"].map(absences_map)
df_res["Вид_отсутствия"] = df_res["причина отсутствия"]
# Исключения и белый список
exc_fios = [normalize_fio(f) for f in exceptions_cfg.get("fio", []) if f]
exc_depts = [d.strip().upper() for d in exceptions_cfg.get("departments", []) if d]
exc_pos = [p.strip().lower() for p in exceptions_cfg.get("positions", []) if p]
@@ -226,10 +301,7 @@ def merge_scud_and_1c(
is_match_exc = (fio in exc_fios or dep in exc_depts or any(d in dep for d in exc_depts) or pos in exc_pos or any(k in pos for k in pos_kw))
if is_match_exc:
if has_official_absence:
df_res.at[idx, "is_excluded"] = False
else:
df_res.at[idx, "is_excluded"] = True
df_res.at[idx, "is_excluded"] = not has_official_absence
mask_exc = (df_res["is_excluded"] == True) & (df_res["Вид_отсутствия"].isna() | (df_res["Вид_отсутствия"] == ""))
df_res.loc[mask_exc, "Вид_отсутствия"] = "Исключение"
@@ -239,38 +311,54 @@ def merge_scud_and_1c(
def calculate_summary_metrics(df_merged: pd.DataFrame) -> Dict[str, Any]:
total_staff = len(df_merged)
# Создаем независимую копию среза штата с собственным непрерывным индексом
df_staff_only = df_merged[~df_merged.get('not_hired_yet', False)].copy().reset_index(drop=True)
total_staff = len(df_staff_only)
came_to_office_mask = (df_merged["Начало_дня"].astype(str).str.strip().ne("Нет входа")) & (df_merged.get("is_excluded", False) == False)
exc_without_doc_mask = (df_merged.get("is_excluded", False) == True) & (
df_merged["Вид_отсутствия"].isna() |
df_merged["Вид_отсутствия"].astype(str).str.strip().isin(["", "nan", "Исключение"])
is_exc_staff = df_staff_only.get('is_excluded', False) == True
is_no_pass_staff = df_staff_only.get('no_scud_pass', False) == True
came_to_office_mask = (df_staff_only["Начало_дня"].astype(str).str.strip().ne("Нет входа")) & (~is_exc_staff)
exc_without_doc_mask = (is_exc_staff) & (
df_staff_only["Вид_отсутствия"].isna() |
df_staff_only["Вид_отсутствия"].astype(str).str.strip().isin(["", "nan", "Исключение"])
)
working_in_office_count = len(df_merged[came_to_office_mask | exc_without_doc_mask])
working_in_office_count = int((came_to_office_mask | exc_without_doc_mask).sum())
df_not_working = df_merged[~came_to_office_mask & ~exc_without_doc_mask]
df_not_working = df_staff_only[~came_to_office_mask & ~exc_without_doc_mask].copy().reset_index(drop=True)
reason_series = df_not_working["причина отсутствия"].astype(str).str.lower()
is_remote_mask = reason_series.str.contains("удален|дистанцион", regex=True, na=False)
remote_home = df_not_working[is_remote_mask]
remote_home_count = len(remote_home)
remote_home_count = int(is_remote_mask.sum())
df_remaining_absent = df_not_working[~is_remote_mask]
has_doc_mask = df_remaining_absent["причина отсутствия"].notna() & \
df_remaining_absent["причина отсутствия"].ne("") & \
df_remaining_absent["причина отсутствия"].ne("nan") & \
(~df_remaining_absent["причина отсутствия"].astype(str).str.startswith("Исключение"))
official_absent_count = len(df_remaining_absent[has_doc_mask])
df_remaining_absent = df_not_working[~is_remote_mask].copy().reset_index(drop=True)
has_doc_mask = (
df_remaining_absent["причина отсутствия"].notna() &
df_remaining_absent["причина отсутствия"].ne("") &
df_remaining_absent["причина отсутствия"].ne("nan") &
(~df_remaining_absent["причина отсутствия"].astype(str).str.startswith("Исключение"))
)
official_absent_count = int(has_doc_mask.sum())
unknown = df_remaining_absent[~has_doc_mask]
# Неизвестно: среди тех, у кого нет официального документа и кто имеет пропуск
is_no_pass_remaining = df_remaining_absent.get('no_scud_pass', False) == True
unknown = df_remaining_absent[~has_doc_mask & ~is_no_pass_remaining]
unknown_count = len(unknown)
no_pass_count = int((is_no_pass_staff & ~is_exc_staff).sum())
is_not_hired_all = df_merged.get('not_hired_yet', False) == True
is_exc_all = df_merged.get('is_excluded', False) == True
not_hired_count = int((is_not_hired_all & ~is_exc_all).sum())
return {
"total_staff": total_staff,
"working_in_office_count": working_in_office_count,
"remote_home_count": remote_home_count,
"official_absent_count": official_absent_count,
"no_pass_count": no_pass_count,
"not_hired_count": not_hired_count,
"unknown_count": unknown_count,
"unknown_list": unknown[["fio_clean", "Подразделение", "Должность"]].to_dict(orient="records") if not unknown.empty else []
}