import os import sys import json import argparse import pandas as pd from datetime import datetime sys.path.append(os.path.dirname(os.path.abspath(__file__))) from config import DATE_TODAY, DATE_YESTERDAY, OUTPUT_DIR, DATA_DIR, normalize_fio 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 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 from services.feedback_loop import review_ai_decisions from services.knowledge_base import load_knowledge_base from core.database import ( init_db, save_scud_to_db, save_staff_to_db, save_absences_to_db, save_anomalies_to_db, load_scud_from_db_by_snapshot, get_latest_snapshot_time, has_scud_logs_for_date ) def load_exceptions_config(): """Загружает файл exceptions.json из корня проекта.""" root_dir = os.path.dirname(os.path.abspath(__file__)) json_path = os.path.join(root_dir, "exceptions.json") if not os.path.exists(json_path): return {} try: with open(json_path, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: print(f"[⚠️] Ошибка чтения exceptions.json: {e}") return {} def apply_exceptions_from_json(df, exceptions_cfg): """ Размечает флаг is_excluded=True на основе правил из exceptions.json. Сверяет отдел одновременно по полному имени из 1С и по аббревиатуре из СКУД. """ if df is None or df.empty or not exceptions_cfg: 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", [])] df['is_excluded'] = False for idx, row in df.iterrows(): fio_clean = row.get('fio_clean', '') dep_1c = str(row.get('Подразделение', '')).lower() dep_scud = str(row.get('department_scud', row.get('department', ''))).lower() 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 if is_fio_exc or is_dep_exc or is_pos_exc: df.at[idx, 'is_excluded'] = True return df def load_static_reason_workers(): """Загружает реестр удалёнщиков и статических причин из CSV.""" static_path = os.path.join(DATA_DIR, "static_reason_workers.csv") if not os.path.exists(static_path): return {} try: 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) return dict(zip(df_static['fio_clean'], df_static['reason'])) except Exception as e: print(f"[⚠️] Ошибка чтения static_reason_workers.csv: {e}") return {} def detect_all_anomalies(merged_df, static_reasons_dict, kb_rules): """Автоматически находит ИСТИННЫЕ аномалии.""" anomalies = [] kb_rules_text = " ".join(kb_rules).lower() if kb_rules else "" ALLOWED_WORK_TRIP_KEYWORDS = ['командировк', 'разъездн', 'поездк'] for idx, row in merged_df.iterrows(): fio = row.get('Сотрудник', row.get('fio_clean', '')) fio_clean = row.get('fio_clean', '') is_present = row.get('Пришел', False) reason_1c = str(row.get('Вид_отсутствия', '')).strip() has_1c_reason = pd.notna(row.get('Вид_отсутствия')) and reason_1c != '' and reason_1c != 'Исключение (ОВК/Подрядчики)' anom_flag = row.get('anomaly_flag', 'NONE') 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) if not is_allowed_trip and not is_fio_whitelisted_in_kb: anomalies.append({ "type": "ФИЗИЧЕСКОЕ ПРИСУТСТВИЕ ПРИ ОФИЦИАЛЬНОМ ОТСУТСТВИИ", "fio": fio, "details": f"Сотрудник пришел по СКУД, но в 1С оформлен документ: '{reason_1c}'" }) # АНОМАЛИЯ 2: Перемещение внутри здания без отметки утреннего входа if anom_flag == 'ANOMALY_NO_IN_HAS_ACTIVITY': first_act = row.get('Первая_активность', '—') anomalies.append({ "type": "АНОМАЛИЯ СКУД: ПЕРЕМЕЩЕНИЕ БЕЗ ВХОДА", "fio": fio, "details": f"Отсутствует регистрация входа на КПП при зафиксированной первой активности в {first_act}" }) return anomalies def aggregate_scud_by_employee(df, debug=False): """Агрегирует проходы СКУД по уникальным сотрудникам.""" if df is None or df.empty or 'fio_clean' not in df.columns: return df aggregated = [] for fio_clean, group in df.groupby('fio_clean', sort=False): is_present = group['Пришел'].any() if 'Пришел' in group.columns else False if is_present and 'Пришел' in group.columns: present_rows = group[group['Пришел'] == True] best_row = present_rows.iloc[0].to_dict() if not present_rows.empty else group.iloc[0].to_dict() else: best_row = group.iloc[0].to_dict() best_row['Пришел'] = is_present aggregated.append(best_row) return pd.DataFrame(aggregated) def main(): help_text = """ Система автоматизированного контроллинга СКУД ⟷ 1С:ЗУП (scud_orion_ai_v2) ПРИМЕРЫ ЗАПУСКА: python main.py -- Обычный дневной запуск python main.py --skip-export -- Расчет отчета на момент ПОСЛЕДНЕГО имеющегося снапшота из SQLite python main.py --snapshot 20260805-001 -- Расчет отчета строго по составному ID снапшота python main.py --snapshot Y20260805-001 -- Расчет отчета по вчерашнему фиксированному снапшоту python main.py --snapshot "2026-08-05 09:23:31" -- Расчет отчета по точной метке времени создания python main.py -d -- Запуск в режиме расширенной отладки (DEBUG) """ parser = argparse.ArgumentParser( description=help_text, formatter_class=argparse.RawDescriptionHelpFormatter ) parser.add_argument('-d', '--debug', action='store_true', help="Запуск в режиме отладки с выводом подробных логов") parser.add_argument('--skip-export', action='store_true', help="Пропустить выгрузку СКУД из MS SQL и построить отчет на момент последнего имеющегося снапшота из SQLite") parser.add_argument('--snapshot', type=str, default=None, help="Составной ID (например, '20260805-001', 'Y20260805-001') или время создания конкретного снапшота") parser.add_argument('--with-xlsx', action='store_true', help="Сохранять дублирующие сырые XLSX-файлы в папке data/") args = parser.parse_args() DEBUG = args.debug print("=" * 60) print(f"ЗАПУСК СИСТЕМЫ МОДУЛЬНОГО КОНТРОЛЛИНГА СКУД ⟷ 1С {'[DEBUG MODE]' if DEBUG else ''}") print("=" * 60) init_db() kb_data = load_knowledge_base() kb_rules = kb_data.get("rules", []) exceptions_cfg = load_exceptions_config() if args.snapshot: snapshot_param = args.snapshot print(f"[📸] РЕЖИМ СНАПШОТА: Расчет отчета строго по срезу '{snapshot_param}'") elif args.skip_export: snapshot_param = get_latest_snapshot_time() print(f"[📸] РЕЖИМ --skip-export: Используем самый последний снапшот из SQLite ('{snapshot_param}')") else: snapshot_param = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[📸] СФОРМИРОВАН НОВЫЙ СНАПШОТ: '{snapshot_param}'") if not args.skip_export and not args.snapshot: try: run_export(save_xlsx=True, debug=DEBUG) except Exception as e: print(f"[⚠️] Ошибка автоэкспорта из БД: {e}. Переходим к имеющимся записям в SQLite.") else: print("[0/5] Пропуск прямого экспорта из MS SQL (чтение из базы SQLite)...") copy_1c_files_from_share() 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 not is_valid: print("\n" + "!" * 60) print("🛑 ОСТАНОВКА ВЫПОЛНЕНИЯ: Отсутствуют критически важные файлы за вчерашний день!") for e in errors: print(f" {e}") print("!" * 60) sys.exit(1) print("[✓] Проверка доступности данных успешно пройдена!\n") 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) if raw_scud_yesterday_df.empty: raw_scud_yesterday_df = load_scud_from_db_by_snapshot(DATE_YESTERDAY) ( _, _, df_staff_today, df_absent_today, df_staff_yesterday, df_absent_yesterday ) = load_all_data() else: ( raw_scud_today_df, raw_scud_yesterday_df, df_staff_today, df_absent_today, df_staff_yesterday, df_absent_yesterday ) = load_all_data() if df_staff_yesterday is not None: save_staff_to_db(df_staff_yesterday, DATE_YESTERDAY) if df_absent_yesterday is not None: save_absences_to_db(df_absent_yesterday, DATE_YESTERDAY) if df_staff_today is not None: save_staff_to_db(df_staff_today, DATE_TODAY) if df_absent_today is not None: save_absences_to_db(df_absent_today, DATE_TODAY) 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: ДЕТАЛЬНЫЙ ОТЧЕТ ЗА ВЧЕРА # ============================================================ if has_scud_logs_for_date(DATE_YESTERDAY) and not args.snapshot: print(f"\n[3/5] Вчерашние данные за {DATE_YESTERDAY} уже обработаны и зафиксированы в SQLite. Повторный расчет пропущен.") else: print("\n[3/5] Обработка данных за ВЧЕРА...") 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 scud_yesterday_unrecognized = raw_scud_yesterday_df[~raw_scud_yesterday_df['fio_clean'].isin(staff_fios_yesterday_clean)]['fio_clean'].tolist() if not raw_scud_yesterday_df.empty else [] fio_mapping_scud_y = ai_verify_scud_against_staff(scud_yesterday_unrecognized, staff_fios_yesterday_clean) if fio_mapping_scud_y: raw_scud_yesterday_df['fio_clean'] = raw_scud_yesterday_df['fio_clean'].apply( lambda x: fio_mapping_scud_y[x]['staff_fio'] if x in fio_mapping_scud_y else x ) if df_absent_yesterday is not None and not df_absent_yesterday.empty: absent_unrecognized_yesterday = df_absent_yesterday[~df_absent_yesterday['fio_clean'].isin(staff_fios_yesterday_clean)]['fio_clean'].tolist() if absent_unrecognized_yesterday: fio_mapping_absent_y = ai_verify_scud_against_staff(absent_unrecognized_yesterday, staff_fios_yesterday_clean) if fio_mapping_absent_y: df_absent_yesterday['fio_clean'] = df_absent_yesterday['fio_clean'].apply( lambda x: fio_mapping_absent_y[x]['staff_fio'] if x in fio_mapping_absent_y else x ) 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() if not merged_yesterday.empty: if not raw_scud_yesterday_df.empty: if 'Подразделение' in raw_scud_yesterday_df.columns: raw_scud_yesterday_df['department_scud'] = raw_scud_yesterday_df['Подразделение'] elif 'department' in raw_scud_yesterday_df.columns: raw_scud_yesterday_df['department_scud'] = raw_scud_yesterday_df['department'] else: raw_scud_yesterday_df['department_scud'] = '' merged_yesterday = merged_yesterday.merge( raw_scud_yesterday_df[['fio_clean', 'Пришел', 'Начало_дня', 'Первая_активность', 'Конец_дня', 'Находился_в_здании', 'anomaly_flag', 'department_scud']], on='fio_clean', how='left' ) # 1. ПРИОРИТЕТ 1: 1С-документ 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') if 'Пришел' not in merged_yesterday.columns: merged_yesterday['Пришел'] = False else: merged_yesterday['Пришел'] = merged_yesterday['Пришел'].fillna(False) if 'Сотрудник' not in merged_yesterday.columns: merged_yesterday['Сотрудник'] = merged_yesterday.get('ФИО', merged_yesterday['fio_clean']) # 2. ПРИОРИТЕТ 2: Реестр статических причин if static_reasons_dict: for fio_clean, reason_val in static_reasons_dict.items(): mask_yesterday = ( (merged_yesterday['Пришел'] == False) & (merged_yesterday['Вид_отсутствия'].isna() | (merged_yesterday['Вид_отсутствия'].astype(str).str.strip() == '')) & (merged_yesterday['fio_clean'] == fio_clean) ) merged_yesterday.loc[mask_yesterday, 'Вид_отсутствия'] = reason_val # 3. ПРИОРИТЕТ 3: Исключения из exceptions.json (ТОЛЬКО при пустом 1С) merged_yesterday = apply_exceptions_from_json(merged_yesterday, exceptions_cfg) mask_exc_yesterday = ( (merged_yesterday['Пришел'] == False) & (merged_yesterday['Вид_отсутствия'].isna() | (merged_yesterday['Вид_отсутствия'].astype(str).str.strip() == '')) & (merged_yesterday.get('is_excluded', False) == True) ) merged_yesterday.loc[mask_exc_yesterday, 'Вид_отсутствия'] = 'Исключение (ОВК/Подрядчики)' generate_detailed_excel(merged_df=merged_yesterday, date_str=DATE_YESTERDAY) save_scud_to_db(merged_yesterday, DATE_YESTERDAY, snapshot_time=snapshot_param if not args.snapshot and not args.skip_export else None, is_yesterday=True) # ============================================================ # 🎯 ЧАСТЬ 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) if has_today_1c and df_staff_today is not None and df_absent_today is not None: print("\n[4/5] Обработка данных за СЕГОДНЯ...") 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: raw_scud_today_df['Пришел'] = raw_scud_today_df['is_present'].astype(int) == 1 if 'is_present' in raw_scud_today_df.columns else False scud_unrecognized = raw_scud_today_df[~raw_scud_today_df['fio_clean'].isin(staff_fios_today_clean)]['fio_clean'].tolist() if not raw_scud_today_df.empty else [] fio_mapping_scud = ai_verify_scud_against_staff(scud_unrecognized, staff_fios_today_clean) if fio_mapping_scud: raw_scud_today_df['fio_clean'] = raw_scud_today_df['fio_clean'].apply( lambda x: fio_mapping_scud[x]['staff_fio'] if x in fio_mapping_scud else x ) absent_unrecognized_today = df_absent_today[~df_absent_today['fio_clean'].isin(staff_fios_today_clean)]['fio_clean'].tolist() if absent_unrecognized_today: fio_mapping_absent = ai_verify_scud_against_staff(absent_unrecognized_today, staff_fios_today_clean) if fio_mapping_absent: df_absent_today['fio_clean'] = df_absent_today['fio_clean'].apply( lambda x: fio_mapping_absent[x]['staff_fio'] if x in fio_mapping_absent else x ) raw_scud_today_df = aggregate_scud_by_employee(raw_scud_today_df, debug=DEBUG) merged_today = df_staff_today.copy() if not raw_scud_today_df.empty: if 'Подразделение' in raw_scud_today_df.columns: raw_scud_today_df['department_scud'] = raw_scud_today_df['Подразделение'] elif 'department' in raw_scud_today_df.columns: raw_scud_today_df['department_scud'] = raw_scud_today_df['department'] else: raw_scud_today_df['department_scud'] = '' merged_today = merged_today.merge( raw_scud_today_df[['fio_clean', 'Пришел', 'Начало_дня', 'Первая_активность', 'Конец_дня', 'Находился_в_здании', 'anomaly_flag', 'department_scud']], on='fio_clean', how='left' ) # 1. ПРИОРИТЕТ 1: 1С-документы if df_absent_today is not None and not df_absent_today.empty: merged_today = merged_today.merge(df_absent_today[['fio_clean', 'Вид_отсутствия']], on='fio_clean', how='left') if 'Пришел' not in merged_today.columns: merged_today['Пришел'] = False else: merged_today['Пришел'] = merged_today['Пришел'].fillna(False) if 'Сотрудник' not in merged_today.columns: merged_today['Сотрудник'] = merged_today.get('ФИО', merged_today['fio_clean']) # 2. ПРИОРИТЕТ 2: Реестр статических причин if static_reasons_dict: for fio_clean, reason_val in static_reasons_dict.items(): mask_today = ( (merged_today['Пришел'] == False) & (merged_today['Вид_отсутствия'].isna() | (merged_today['Вид_отсутствия'].astype(str).str.strip() == '')) & (merged_today['fio_clean'] == fio_clean) ) merged_today.loc[mask_today, 'Вид_отсутствия'] = reason_val # 3. ПРИОРИТЕТ 3: Исключения из exceptions.json (ТОЛЬКО при пустом 1С) merged_today = apply_exceptions_from_json(merged_today, exceptions_cfg) mask_exc_today = ( (merged_today['Пришел'] == False) & (merged_today['Вид_отсутствия'].isna() | (merged_today['Вид_отсутствия'].astype(str).str.strip() == '')) & (merged_today.get('is_excluded', False) == True) ) merged_today.loc[mask_exc_today, 'Вид_отсутствия'] = 'Исключение (ОВК/Подрядчики)' anomalies_list = detect_all_anomalies(merged_today, static_reasons_dict, kb_rules) mass_failure_result = analyze_scud_mass_failure_ai(merged_today) if mass_failure_result and mass_failure_result.get("is_mass_failure"): print("\n" + "!" * 60) print(f"🚨 ВНИМАНИЕ! ИИ ОБНАРУЖИЛ МАССОВЫЙ СБОЙ ТУРНИКЕТОВ ВХОДА ({mass_failure_result['anomaly_percent']}% СМЕНЫ)") print(mass_failure_result["alert_text"]) print("!" * 60 + "\n") save_anomalies_to_db(anomalies_list, DATE_TODAY) absent_explained = merged_today[(merged_today['Пришел'] == False) & (merged_today['Вид_отсутствия'].notna())] absent_unexplained = merged_today[(merged_today['Пришел'] == False) & (merged_today['Вид_отсутствия'].isna())] scud_present_but_absent_in_1c = merged_today[(merged_today['Пришел'] == True) & (merged_today['Вид_отсутствия'].notna())] print("[5/5] Запуск ИИ-аудитора и построение Ежедневной сводки...") report_text = generate_markdown_report( merged_df=merged_today, absent_explained=absent_explained, absent_unexplained=absent_unexplained, scud_present_but_absent_in_1c=scud_present_but_absent_in_1c, anomalies_list=anomalies_list, raw_scud_df=raw_scud_today_df, raw_absent_df=df_absent_today, date_str=DATE_TODAY ) generate_summary_excel(merged_df=merged_today, date_str=DATE_TODAY) md_report_path = os.path.join(OUTPUT_DIR, f"Сводка_контроллинга_{DATE_TODAY}.md") with open(md_report_path, "w", encoding="utf-8") as f: f.write(report_text) print(f"\n[✓] Текстовый отчет сохранен в: {md_report_path}") suspicious_cases = [] for a in anomalies_list: if "ОФИЦИАЛЬНОМ ОТСУТСТВИИ" not in a.get('type', ''): suspicious_cases.append({ 'fio_target': a['fio'], 'reason': f"{a['type']}: {a['details']}" }) if suspicious_cases: review_ai_decisions(report_text, suspicious_cases) print("\n" + "=" * 60) print("ГОТОВАЯ ТЕКСТОВАЯ СВОДКА ИИ-АУДИТОРА:") print("=" * 60) print(report_text) else: print(f"\n[ℹ️] Формирование Ежедневной сводки за {DATE_TODAY} ПРОПУЩЕНО.") if __name__ == "__main__": main()