manoraga/scud_orion_ai_v2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

12.08.2026 17:30 Логика формирования отчетов работает, все отсутствия учитываются, логика присутствия сотрудников в случае сбоя СКУД на работе настроена.

Browse Source
This commit is contained in:
manoraga
2026-08-12 17:33:48 +03:00
parent ffb0e18635
commit 8897b14197
13 changed files with 467 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.py
+96 -111
View File
@@ -3,7 +3,7 @@ import sys
import json
import argparse
import pandas as pd
from datetime import datetime
from datetime import datetime, timedelta
sys.path.append(os.path.dirname(os.path.abspath(__file__)))
@@ -11,7 +11,7 @@ from config import DATE_TODAY, DATE_YESTERDAY, OUTPUT_DIR, DATA_DIR, normalize_f
from services.scud_export import run_export
from services.share_copier import copy_1c_files_from_share
from services.data_validator import check_file_freshness
from services.data_loader import load_all_data, load_scud_data
from services.data_loader import load_1c_data_smart, load_scud_data
from services.excel_exporter import generate_summary_excel, generate_detailed_excel
from services.ai_verifier import ai_verify_scud_against_staff, analyze_scud_mass_failure_ai
from services.text_reporter import generate_markdown_report
@@ -24,45 +24,12 @@ from core.database import (
save_absences_to_db,
save_anomalies_to_db,
load_scud_from_db_by_snapshot,
load_staff_from_db,
load_absences_from_db,
get_latest_snapshot_time,
has_scud_logs_for_date
)
from services.ai_verifier import resolve_department_exception_ai
def apply_exceptions_from_json(df, exceptions_cfg):
"""
Размечает флаг is_excluded=True на основе правил из exceptions.json и БД SQLite.
Сверяет одновременно 1С, СКУД и использует интеллектуальный поиск отделов.
"""
if df is None or df.empty or not exceptions_cfg:
if df is not None:
df['is_excluded'] = False
return df
deps = exceptions_cfg.get("departments", [])
exact_pos = [p.lower() for p in exceptions_cfg.get("positions", [])]
pos_kw = [k.lower() for k in exceptions_cfg.get("position_keywords", [])]
exc_fios = [normalize_fio(f) for f in exceptions_cfg.get("fio", [])]
df['is_excluded'] = False
for idx, row in df.iterrows():
fio_clean = row.get('fio_clean', '')
dep_1c = row.get('Подразделение', '')
dep_scud = row.get('department_scud', row.get('department', ''))
pos = str(row.get('Должность', '')).lower()
is_fio_exc = fio_clean in exc_fios
is_pos_exc = (pos in exact_pos) or any(k in pos for k in pos_kw) if pos else False
# Двухуровневая интеллектуальная проверка отделов (1С + СКУД + ИИ-Кеш БД)
is_dep_exc = resolve_department_exception_ai(dep_1c, dep_scud, deps)
if is_fio_exc or is_dep_exc or is_pos_exc:
df.at[idx, 'is_excluded'] = True
return df
def load_exceptions_config():
"""Загружает файл exceptions.json из корня проекта."""
@@ -80,18 +47,18 @@ def load_exceptions_config():
def apply_exceptions_from_json(df, exceptions_cfg):
"""
Размечает флаг is_excluded=True на основе правил из exceptions.json.
Сверяет отдел одновременно по полному имени из 1С и по короткой аббревиатуре из СКУД.
Быстрая и эффективная разметка флага is_excluded=True на основе exceptions.json.
Выполняется мгновенно без цикличных HTTP-запросов к ИИ.
"""
if df is None or df.empty or not exceptions_cfg:
if df is not None:
df['is_excluded'] = False
return df
deps = [d.lower() for d in exceptions_cfg.get("departments", [])]
exact_pos = [p.lower() for p in exceptions_cfg.get("positions", [])]
pos_kw = [k.lower() for k in exceptions_cfg.get("position_keywords", [])]
exc_fios = [normalize_fio(f) for f in exceptions_cfg.get("fio", [])]
deps = [d.strip().lower() for d in exceptions_cfg.get("departments", []) if d]
exact_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]
exc_fios = [normalize_fio(f) for f in exceptions_cfg.get("fio", []) if f]
df['is_excluded'] = False
@@ -102,8 +69,12 @@ def apply_exceptions_from_json(df, exceptions_cfg):
pos = str(row.get('Должность', '')).lower()
is_fio_exc = fio_clean in exc_fios
is_dep_exc = any(d in dep_1c or d == dep_scud for d in deps) if (dep_1c or dep_scud) else False
is_pos_exc = (pos in exact_pos) or any(k in pos for k in pos_kw) if pos else False
is_pos_exc = (pos in exact_pos) or any(k in pos for k in pos_kw if k) if pos else False
# Быстрая проверка отделов по подстрокам
is_dep_exc = False
if deps:
is_dep_exc = any(d in dep_1c or d in dep_scud for d in deps)
if is_fio_exc or is_dep_exc or is_pos_exc:
df.at[idx, 'is_excluded'] = True
@@ -127,7 +98,6 @@ def filter_report_dataframe(merged_df):
is_not_excluded = merged_df.get('is_excluded', False) == False
is_not_no_pass = merged_df.get('no_scud_pass', False) == False
# В отчет попадают ТОЛЬКО офисные сотрудники с пропусками ИЛИ имеющие документ из 1С
filtered_df = merged_df[(is_not_excluded & is_not_no_pass) | has_1c_reason].copy()
return filtered_df
@@ -150,7 +120,6 @@ def load_static_reason_workers():
def detect_all_anomalies(merged_df, static_reasons_dict, kb_rules, scud_fios_set=None):
"""
Автоматически выявляет истинные аномалии СКУД ⟷ 1С.
Исключения (ОВК/Клининг/Охрана) НЕ попадают в аномалии СКУД, если на них нет документа из 1С.
"""
anomalies = []
kb_rules_text = " ".join(kb_rules).lower() if kb_rules else ""
@@ -168,7 +137,6 @@ def detect_all_anomalies(merged_df, static_reasons_dict, kb_rules, scud_fios_set
is_fio_whitelisted_in_kb = fio_clean.lower() in kb_rules_text
# 🚨 АНОМАЛИЯ 1: Физическое присутствие при документе отсутствия 1С
if is_present and has_1c_reason:
reason_lower = reason_1c.lower()
is_allowed_trip = any(kw in reason_lower for kw in ALLOWED_WORK_TRIP_KEYWORDS)
@@ -180,11 +148,9 @@ def detect_all_anomalies(merged_df, static_reasons_dict, kb_rules, scud_fios_set
"details": f"Сотрудник пришел по СКУД, но в 1С оформлен документ: '{reason_1c}'"
})
# ИСКЛЮЧЕНИЯ без документов 1С НЕ пропускаются дальше к аномалиям СКУД
if is_exc and not has_1c_reason:
continue
# 🚨 АНОМАЛИЯ 2: Перемещение внутри здания без отметки утреннего входа
if anom_flag == 'ANOMALY_NO_IN_HAS_ACTIVITY':
first_act = row.get('Первая_активность', '—')
anomalies.append({
@@ -193,7 +159,6 @@ def detect_all_anomalies(merged_df, static_reasons_dict, kb_rules, scud_fios_set
"details": f"Отсутствует регистрация входа на КПП при зафиксированной первой активности в {first_act}"
})
# 🚨 АНОМАЛИЯ 3: Сотрудник есть в Штате 1С, но ПОЛНОСТЬЮ ОТСУТСТВУЕТ в базе СКУД Орион Pro
if scud_fios_set is not None:
if fio_clean not in scud_fios_set and not has_1c_reason:
anomalies.append({
@@ -274,65 +239,84 @@ def main():
else:
print("[0/5] Пропуск прямого экспорта из MS SQL (чтение из базы SQLite)...")
copy_1c_files_from_share()
# ⚡️ ПРОВЕРКА ШАРЫ ТОЛЬКО ДЛЯ ОБЫЧНОГО ДНЕВНОГО ЗАПУСКА
if not args.snapshot and not args.skip_export:
copy_1c_files_from_share()
print("[1/5] Проверка актуальности и свежести входных данных...")
is_valid, warnings, errors, has_today_1c = check_file_freshness()
print("[1/5] Проверка актуальности и свежести входных данных...")
is_valid, warnings, errors, has_today_1c = check_file_freshness()
if warnings:
print("\n--- ⚠️ ПРЕДУПРЕЖДЕНИЯ ОБ АКТУАЛЬНОСТИ ---")
for w in warnings:
print(f" • {w}")
print("-" * 45)
if warnings:
print("\n--- ⚠️ ПРЕДУПРЕЖДЕНИЯ ОБ АКТУАЛЬНОСТИ ---")
for w in warnings:
print(f" • {w}")
print("-" * 45)
if not is_valid:
print("\n" + "!" * 60)
print("🛑 ОСТАНОВКА ВЫПОЛНЕНИЯ: Отсутствуют критически важные файлы за вчерашний день!")
for e in errors:
print(f" {e}")
print("!" * 60)
sys.exit(1)
if not is_valid:
print("\n" + "!" * 60)
print("🛑 ОСТАНОВКА ВЫПОЛНЕНИЯ: Отсутствуют критически важные файлы за вчерашний день!")
for e in errors:
print(f" {e}")
print("!" * 60)
sys.exit(1)
print("[✓] Проверка доступности данных успешно пройдена!\n")
print("[✓] Проверка доступности данных успешно пройдена!\n")
else:
print("[1/5] Пропуск проверки сетевой шары (все данные читаются из SQLite)...")
has_today_1c = True
print("[2/5] Загрузка данных из СКУД, 1С:ЗУП, реестра причин и исключений...")
# 🎯 ВЫЧИСЛЕНИЕ ДАТ СНАПШОТА И ДАТ НАКАНУНЕ
if args.snapshot or args.skip_export:
raw_scud_today_df = load_scud_from_db_by_snapshot(DATE_TODAY, snapshot_param)
raw_scud_yesterday_df = load_scud_from_db_by_snapshot(DATE_YESTERDAY, snapshot_param)
raw_scud_today_df = load_scud_from_db_by_snapshot(None, snapshot_param=snapshot_param)
if raw_scud_yesterday_df.empty:
raw_scud_yesterday_df = load_scud_from_db_by_snapshot(DATE_YESTERDAY)
if not raw_scud_today_df.empty and 'log_date' in raw_scud_today_df.columns:
target_date_str = str(raw_scud_today_df['log_date'].iloc[0])
else:
target_date_str = DATE_TODAY
(
_, _,
df_staff_today, df_absent_today,
df_staff_yesterday, df_absent_yesterday
) = load_all_data()
dt_target = datetime.strptime(target_date_str, "%d.%m.%Y")
if dt_target.weekday() == 0: # Понедельник -> Пятница
dt_yesterday = dt_target - timedelta(days=3)
else:
dt_yesterday = dt_target - timedelta(days=1)
yesterday_date_str = dt_yesterday.strftime("%d.%m.%Y")
print(f"[📸] СНАПШОТ ОПРЕДЕЛЕН: Целевая дата = {target_date_str}, Накануне = {yesterday_date_str}")
raw_scud_yesterday_df = load_scud_from_db_by_snapshot(yesterday_date_str, snapshot_param=None)
# Гарантированная загрузка кадров 1С за обе даты (БД + локальный фолбэк)
df_staff_today, df_absent_today = load_1c_data_smart(target_date_str, use_db=True)
df_staff_yesterday, df_absent_yesterday = load_1c_data_smart(yesterday_date_str, use_db=True)
else:
(
raw_scud_today_df, raw_scud_yesterday_df,
df_staff_today, df_absent_today,
df_staff_yesterday, df_absent_yesterday
) = load_all_data()
target_date_str = DATE_TODAY
yesterday_date_str = DATE_YESTERDAY
raw_scud_today_df = load_scud_from_db_by_snapshot(target_date_str, snapshot_param=snapshot_param)
raw_scud_yesterday_df = load_scud_from_db_by_snapshot(yesterday_date_str, snapshot_param=None)
df_staff_today, df_absent_today = load_1c_data_smart(DATE_TODAY, use_db=False)
df_staff_yesterday, df_absent_yesterday = load_1c_data_smart(DATE_YESTERDAY, use_db=False)
if df_staff_yesterday is not None:
save_staff_to_db(df_staff_yesterday, DATE_YESTERDAY)
save_staff_to_db(df_staff_yesterday, yesterday_date_str)
if df_absent_yesterday is not None:
save_absences_to_db(df_absent_yesterday, DATE_YESTERDAY)
save_absences_to_db(df_absent_yesterday, yesterday_date_str)
if df_staff_today is not None:
save_staff_to_db(df_staff_today, DATE_TODAY)
save_staff_to_db(df_staff_today, target_date_str)
if df_absent_today is not None:
save_absences_to_db(df_absent_today, DATE_TODAY)
save_absences_to_db(df_absent_today, target_date_str)
staff_fios_yesterday_clean = df_staff_yesterday['fio_clean'].dropna().tolist() if df_staff_yesterday is not None else []
static_reasons_dict = load_static_reason_workers()
# ============================================================
# 🎯 ЧАСТЬ 1: ДЕТАЛЬНЫЙ ОТЧЕТ ЗА ВЧЕРА (ГАРАНТИРОВАННЫЙ РАСЧЕТ)
# 🎯 ЧАСТЬ 1: ДЕТАЛЬНЫЙ ОТЧЕТ ЗА ВЧЕРА (ДЕНЬ НАКАНУНЕ)
# ============================================================
print(f"\n[3/5] Обработка и построение детального отчета за ВЧЕРА ({DATE_YESTERDAY})...")
print(f"\n[3/5] Обработка и построение детального отчета за ВЧЕРА ({yesterday_date_str})...")
if not raw_scud_yesterday_df.empty and 'Пришел' not in raw_scud_yesterday_df.columns:
raw_scud_yesterday_df['Пришел'] = raw_scud_yesterday_df['is_present'].astype(int) == 1 if 'is_present' in raw_scud_yesterday_df.columns else False
@@ -356,7 +340,9 @@ def main():
raw_scud_yesterday_df = aggregate_scud_by_employee(raw_scud_yesterday_df, debug=DEBUG)
merged_yesterday = df_staff_yesterday.copy() if df_staff_yesterday is not None else pd.DataFrame()
# ⭐️ ФОРМИРУЕМ МЕРДЖ ИСКЛЮЧИТЕЛЬНО НА БАЗЕ ШТАТА ЗА ПРОШЛЫЙ ДЕНЬ
merged_yesterday = df_staff_yesterday.copy() if (df_staff_yesterday is not None and not df_staff_yesterday.empty) else (df_staff_today.copy() if df_staff_today is not None else pd.DataFrame())
if not merged_yesterday.empty:
if not raw_scud_yesterday_df.empty:
if 'Подразделение' in raw_scud_yesterday_df.columns:
@@ -371,8 +357,16 @@ def main():
on='fio_clean', how='left'
)
if 'Вид_отсутствия' in merged_yesterday.columns:
merged_yesterday = merged_yesterday.drop(columns=['Вид_отсутствия'])
# Присоединяем отсутствия СТРОГО за дату yesterday_date_str
if df_absent_yesterday is not None and not df_absent_yesterday.empty:
merged_yesterday = merged_yesterday.merge(df_absent_yesterday[['fio_clean', 'Вид_отсутствия']], on='fio_clean', how='left')
merged_yesterday = merged_yesterday.merge(
df_absent_yesterday[['fio_clean', 'Вид_отсутствия']],
on='fio_clean',
how='left'
)
if 'Пришел' not in merged_yesterday.columns:
merged_yesterday['Пришел'] = False
@@ -399,7 +393,6 @@ def main():
)
merged_yesterday.loc[mask_exc_yesterday, 'Вид_отсутствия'] = 'Исключение (ОВК/Подрядчики)'
# Расчет флага no_scud_pass для вчера
scud_fios_yesterday_set = set(raw_scud_yesterday_df['fio_clean'].dropna().tolist()) if not raw_scud_yesterday_df.empty else set()
merged_yesterday['no_scud_pass'] = (
(~merged_yesterday['fio_clean'].isin(scud_fios_yesterday_set)) &
@@ -407,28 +400,25 @@ def main():
(merged_yesterday.get('is_excluded', False) == False)
)
# 🚨 Рассчитываем аномалии за ВЧЕРА
anomalies_yesterday_list = detect_all_anomalies(merged_yesterday, static_reasons_dict, kb_rules, scud_fios_set=scud_fios_yesterday_set)
save_anomalies_to_db(anomalies_yesterday_list, DATE_YESTERDAY)
save_anomalies_to_db(anomalies_yesterday_list, yesterday_date_str)
# ⚡️ Фильтрация: исключения НЕ попадают в вчерашний отчет (если нет отпуска из 1С)
filtered_yesterday = filter_report_dataframe(merged_yesterday)
generate_detailed_excel(merged_df=filtered_yesterday, date_str=DATE_YESTERDAY)
generate_detailed_excel(merged_df=filtered_yesterday, date_str=yesterday_date_str)
# Фиксируем вчерашний снапшот строго на 22:00:00
yesterday_dt_obj = datetime.strptime(DATE_YESTERDAY, "%d.%m.%Y")
yesterday_dt_obj = datetime.strptime(yesterday_date_str, "%d.%m.%Y")
yesterday_22_str = yesterday_dt_obj.strftime("%Y-%m-%d 22:00:00")
save_scud_to_db(merged_yesterday, DATE_YESTERDAY, snapshot_time=yesterday_22_str, is_yesterday=True)
print(f"[✓] Детальный отчет за вчера сформирован и зафиксирован в SQLite со снапшотом на {yesterday_22_str}")
save_scud_to_db(merged_yesterday, yesterday_date_str, snapshot_time=yesterday_22_str, is_yesterday=True)
print(f"[✓] Детальный отчет за вчера сформирован и зафиксирован в SQLite за {yesterday_date_str}")
# ============================================================
# 🎯 ЧАСТЬ 2: ЕЖЕДНЕВНАЯ СВОДКА ЗА СЕГОДНЯ
# 🎯 ЧАСТЬ 2: ЕЖЕДНЕВНАЯ СВОДКА (ЗА ЦЕЛЕВОЙ ДЕНЬ СНАПШОТА)
# ============================================================
if not args.snapshot and not args.skip_export:
save_scud_to_db(raw_scud_today_df, DATE_TODAY, snapshot_time=snapshot_param, is_yesterday=False)
save_scud_to_db(raw_scud_today_df, target_date_str, snapshot_time=snapshot_param, is_yesterday=False)
if has_today_1c and df_staff_today is not None and df_absent_today is not None:
print("\n[4/5] Обработка данных за СЕГОДНЯ...")
print(f"\n[4/5] Обработка и построение Ежедневной сводки за {target_date_str}...")
staff_fios_today_clean = df_staff_today['fio_clean'].dropna().tolist()
if not raw_scud_today_df.empty and 'Пришел' not in raw_scud_today_df.columns:
@@ -493,7 +483,6 @@ def main():
)
merged_today.loc[mask_exc_today, 'Вид_отсутствия'] = 'Исключение (ОВК/Подрядчики)'
# Расчет флага no_scud_pass для сегодня
scud_fios_today_set = set(raw_scud_today_df['fio_clean'].dropna().tolist()) if not raw_scud_today_df.empty else set()
merged_today['no_scud_pass'] = (
(~merged_today['fio_clean'].isin(scud_fios_today_set)) &
@@ -504,7 +493,6 @@ def main():
scud_fios_set = set(raw_scud_today_df['fio_clean'].dropna().tolist()) if not raw_scud_today_df.empty else set()
anomalies_list = detect_all_anomalies(merged_today, static_reasons_dict, kb_rules, scud_fios_set=scud_fios_set)
# 🚨 ОПЕРАТИВНЫЙ МОНИТОРИНГ ОБОРУДОВАНИЯ СКУД ЗА СЕГОДНЯ:
mass_failure_today = analyze_scud_mass_failure_ai(raw_scud_today_df)
if mass_failure_today and mass_failure_today.get("is_mass_failure"):
print("\n" + "!" * 60)
@@ -512,9 +500,8 @@ def main():
print(mass_failure_today["alert_text"])
print("!" * 60 + "\n")
save_anomalies_to_db(anomalies_list, DATE_TODAY)
save_anomalies_to_db(anomalies_list, target_date_str)
# Флаги фильтрации для ИИ-репортера
is_no_pass_today = merged_today['no_scud_pass'] == True if 'no_scud_pass' in merged_today.columns else False
is_exc_today = merged_today.get('is_excluded', False) == True
@@ -524,7 +511,6 @@ def main():
(~merged_today['Вид_отсутствия'].astype(str).str.startswith('Исключение'))
]
# Истинно неизвестные случаи (исключаем тех, у кого нет пропуска, и спецотделы ОВК/Охрана)
absent_unexplained = merged_today[
(merged_today['Пришел'] == False) &
(merged_today['Вид_отсутствия'].isna() | (merged_today['Вид_отсутствия'].astype(str).str.strip() == '')) &
@@ -540,7 +526,6 @@ def main():
]
print("[5/5] Запуск ИИ-аудитора и построение Ежедневной сводки...")
# Фильтруем аномалии СКУД для текстового отчета (убираем аномалии "ОТСУТСТВУЕТ В СКУД", т.к. они ушли в раздел "Нет пропуска")
filtered_anomalies_list = [
a for a in anomalies_list
if "ОТСУТСТВУЕТ В СКУД" not in a.get('type', '')
@@ -554,12 +539,12 @@ def main():
anomalies_list=filtered_anomalies_list,
raw_scud_df=raw_scud_today_df,
raw_absent_df=df_absent_today,
date_str=DATE_TODAY
date_str=target_date_str
)
generate_summary_excel(merged_df=merged_today, date_str=DATE_TODAY)
generate_summary_excel(merged_df=merged_today, date_str=target_date_str)
md_report_path = os.path.join(OUTPUT_DIR, f"Сводка_контроллинга_{DATE_TODAY}.md")
md_report_path = os.path.join(OUTPUT_DIR, f"Сводка_контроллинга_{target_date_str}.md")
with open(md_report_path, "w", encoding="utf-8") as f:
f.write(report_text)
print(f"\n[✓] Текстовый отчет сохранен в: {md_report_path}")
@@ -580,7 +565,7 @@ def main():
print("=" * 60)
print(report_text)
else:
print(f"\n[ℹ️] Формирование Ежедневной сводки за {DATE_TODAY} ПРОПУЩЕНО.")
print(f"\n[ℹ️] Формирование Ежедневной сводки за {target_date_str} ПРОПУЩЕНО.")
if __name__ == "__main__":