276 lines
12 KiB
Python
276 lines
12 KiB
Python
"""
|
|
===============================================================================
|
|
FILE: services/scud_etl/merger.py
|
|
ROLE: Агрегация реестров, выбор совместителей 1С по отделу СКУД и авто-связки.
|
|
Распределение исключений в общий рабочий пул при отсутствии справок.
|
|
===============================================================================
|
|
"""
|
|
|
|
import logging
|
|
from typing import Dict, Any, List
|
|
import pandas as pd
|
|
|
|
from services.knowledge.service import get_department_synonyms_dict
|
|
from config import normalize_fio, load_exceptions
|
|
from core.connection import get_connection
|
|
|
|
logger = logging.getLogger("SCUD_MERGER")
|
|
|
|
|
|
def load_identity_mappings() -> Dict[str, str]:
|
|
with get_connection() as conn:
|
|
cursor = conn.cursor()
|
|
cursor.execute("""
|
|
CREATE TABLE IF NOT EXISTS person_identity_mapping (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
scud_fio TEXT NOT NULL,
|
|
zup_fio TEXT NOT NULL,
|
|
scud_dept TEXT,
|
|
zup_dept TEXT,
|
|
match_source TEXT DEFAULT 'AI',
|
|
status TEXT DEFAULT 'ACTIVE',
|
|
confidence REAL DEFAULT 1.0,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
UNIQUE(scud_fio, zup_fio)
|
|
);
|
|
""")
|
|
cursor.execute("""
|
|
SELECT scud_fio, zup_fio
|
|
FROM person_identity_mapping
|
|
WHERE status = 'ACTIVE'
|
|
""")
|
|
return {r[0]: r[1] for r in cursor.fetchall()}
|
|
|
|
|
|
def aggregate_scud_by_person(df_scud: pd.DataFrame) -> pd.DataFrame:
|
|
if df_scud is None or df_scud.empty:
|
|
return df_scud
|
|
|
|
df = df_scud.copy()
|
|
if 'fio_clean' not in df.columns:
|
|
fio_col = 'Сотрудник' if 'Сотрудник' in df.columns else 'ФИО'
|
|
df['fio_clean'] = df[fio_col].apply(normalize_fio)
|
|
|
|
mapping_dict = load_identity_mappings()
|
|
if mapping_dict:
|
|
df['fio_clean'] = df['fio_clean'].apply(lambda f: mapping_dict.get(f, f))
|
|
|
|
aggregated_rows = []
|
|
for fio_clean, group in df.groupby('fio_clean'):
|
|
if len(group) == 1:
|
|
aggregated_rows.append(group.iloc[0].to_dict())
|
|
continue
|
|
|
|
base_row = group.sort_values(by='Пришел', ascending=False).iloc[0].to_dict()
|
|
|
|
valid_ins = [
|
|
str(t).strip() for t in group['Начало_дня']
|
|
if str(t).strip() not in ['Нет входа', '—', '', 'nan', 'None', '00:00:00', '00:00']
|
|
]
|
|
base_row['Начало_дня'] = min(valid_ins) if valid_ins else 'Нет входа'
|
|
|
|
valid_outs = [
|
|
str(t).strip() for t in group['Конец_дня']
|
|
if str(t).strip() not in ['Нет выхода', '—', '', 'nan', 'None', '00:00:00', '00:00']
|
|
]
|
|
base_row['Конец_дня'] = max(valid_outs) if valid_outs else 'Нет выхода'
|
|
|
|
valid_acts = [
|
|
str(t).strip() for t in group['Первая_активность']
|
|
if str(t).strip() not in ['—', '', 'nan', 'None', '00:00:00']
|
|
]
|
|
base_row['Первая_активность'] = min(valid_acts) if valid_acts else '—'
|
|
base_row['Пришел'] = any(group['Пришел'] == True) or (base_row['Начало_дня'] != 'Нет входа')
|
|
|
|
durations = [str(d) for d in group['Находился_в_здании'] if str(d) not in ['00:00', '', 'nan']]
|
|
if durations:
|
|
base_row['Находился_в_здании'] = max(durations)
|
|
|
|
aggregated_rows.append(base_row)
|
|
|
|
return pd.DataFrame(aggregated_rows)
|
|
|
|
|
|
def select_best_zup_position(df_staff_1c: pd.DataFrame, df_scud_agg: pd.DataFrame) -> pd.DataFrame:
|
|
if df_staff_1c is None or df_staff_1c.empty:
|
|
return pd.DataFrame()
|
|
|
|
df_staff = df_staff_1c.copy()
|
|
if 'fio_clean' not in df_staff.columns:
|
|
f_col = 'ФИО' if 'ФИО' in df_staff.columns else 'Сотрудник'
|
|
df_staff['fio_clean'] = df_staff[f_col].apply(normalize_fio)
|
|
|
|
scud_dept_map = {}
|
|
if df_scud_agg is not None and not df_scud_agg.empty:
|
|
for _, r in df_scud_agg.iterrows():
|
|
scud_dept_map[r.get('fio_clean', '')] = str(r.get('Подразделение', '')).strip().upper()
|
|
|
|
best_rows = []
|
|
for fio, group in df_staff.groupby('fio_clean'):
|
|
if len(group) == 1:
|
|
best_rows.append(group.iloc[0].to_dict())
|
|
continue
|
|
|
|
target_scud_dept = scud_dept_map.get(fio, "")
|
|
matched_row = None
|
|
|
|
if target_scud_dept:
|
|
for _, r in group.iterrows():
|
|
dept_1c = str(r.get('Подразделение', '')).strip().upper()
|
|
if dept_1c == target_scud_dept or target_scud_dept in dept_1c or dept_1c in target_scud_dept:
|
|
matched_row = r.to_dict()
|
|
break
|
|
|
|
if not matched_row:
|
|
matched_row = group.iloc[0].to_dict()
|
|
|
|
best_rows.append(matched_row)
|
|
|
|
return pd.DataFrame(best_rows)
|
|
|
|
|
|
def merge_scud_and_1c(
|
|
df_scud: pd.DataFrame,
|
|
df_staff_1c: pd.DataFrame,
|
|
df_absences_1c: pd.DataFrame
|
|
) -> pd.DataFrame:
|
|
if (df_scud is None or df_scud.empty) and (df_staff_1c is None or df_staff_1c.empty):
|
|
return pd.DataFrame(columns=[
|
|
'Сотрудник', 'fio_clean', 'Подразделение', 'Должность',
|
|
'Начало_дня', 'Первая_активность', 'Конец_дня', 'Находился_в_здании',
|
|
'Пришел', 'anomaly_flag', 'причина отсутствия', 'Вид_отсутствия', 'is_excluded'
|
|
])
|
|
|
|
synonyms = get_department_synonyms_dict()
|
|
exceptions_cfg = load_exceptions()
|
|
|
|
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()
|
|
|
|
if "Сотрудник" in df_res.columns:
|
|
df_res["fio_clean"] = df_res["Сотрудник"].apply(normalize_fio)
|
|
elif "ФИО" 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 [
|
|
('Начало_дня', 'Нет входа'),
|
|
('Первая_активность', '—'),
|
|
('Конец_дня', 'Нет выхода'),
|
|
('Находился_в_здании', '00:00'),
|
|
('Пришел', False),
|
|
('anomaly_flag', 'NONE')
|
|
]:
|
|
if col not in df_res.columns:
|
|
df_res[col] = default_val
|
|
|
|
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, "отдел внутреннего контроля": "ОВК", "отдел вневедомственного контроля": "ОВК"}
|
|
|
|
if "Подразделение" in df_res.columns:
|
|
df_res["Подразделение"] = df_res["Подразделение"].apply(
|
|
lambda d: all_dept_map.get(str(d).strip().lower(), str(d).strip())
|
|
)
|
|
|
|
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)
|
|
reason_col = next((c for c in ["Вид_отсутствия", "Причина", "причина отсутствия"] if c in df_absences_1c.columns), None)
|
|
|
|
if fio_col and reason_col:
|
|
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":
|
|
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))
|
|
for mr in m_records:
|
|
manual_reasons_map[mr['fio_clean']] = mr['reason']
|
|
except Exception:
|
|
pass
|
|
|
|
df_res["detailed_reason"] = df_res["fio_clean"].map(manual_reasons_map).fillna("")
|
|
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]
|
|
pos_kw = [k.strip().lower() for k in exceptions_cfg.get("position_keywords", []) if k]
|
|
whitelist_fios = [normalize_fio(f) for f in exceptions_cfg.get("include_fio", []) if f]
|
|
|
|
df_res["is_excluded"] = False
|
|
for idx, row in df_res.iterrows():
|
|
fio = row.get("fio_clean", "")
|
|
has_official_absence = pd.notna(row.get("Вид_отсутствия")) and str(row.get("Вид_отсутствия")).strip() not in ["", "nan", "None", "Исключение"]
|
|
|
|
if fio in whitelist_fios:
|
|
df_res.at[idx, "is_excluded"] = False
|
|
continue
|
|
|
|
dep = str(row.get("Подразделение", "")).upper()
|
|
pos = str(row.get("Должность", "")).lower()
|
|
|
|
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
|
|
|
|
mask_exc = (df_res["is_excluded"] == True) & (df_res["Вид_отсутствия"].isna() | (df_res["Вид_отсутствия"] == ""))
|
|
df_res.loc[mask_exc, "Вид_отсутствия"] = "Исключение"
|
|
df_res.loc[mask_exc, "причина отсутствия"] = "Исключение"
|
|
|
|
return df_res
|
|
|
|
|
|
def calculate_summary_metrics(df_merged: pd.DataFrame) -> Dict[str, Any]:
|
|
total_staff = len(df_merged)
|
|
|
|
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", "Исключение"])
|
|
)
|
|
|
|
working_in_office_count = len(df_merged[came_to_office_mask | exc_without_doc_mask])
|
|
|
|
df_not_working = df_merged[~came_to_office_mask & ~exc_without_doc_mask]
|
|
|
|
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)
|
|
|
|
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])
|
|
|
|
unknown = df_remaining_absent[~has_doc_mask]
|
|
unknown_count = len(unknown)
|
|
|
|
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,
|
|
"unknown_count": unknown_count,
|
|
"unknown_list": unknown[["fio_clean", "Подразделение", "Должность"]].to_dict(orient="records") if not unknown.empty else []
|
|
} |