import os import sys import argparse import pandas as pd sys.path.append(os.path.dirname(os.path.abspath(__file__))) from config import DATE_TODAY, DATE_YESTERDAY, OUTPUT_DIR, DATA_DIR, normalize_fio from modules.data_validator import check_file_freshness from modules.data_loader import load_all_data from modules.excel_exporter import generate_summary_excel, generate_detailed_excel from modules.ai_verifier import ai_verify_scud_against_staff from modules.text_reporter import generate_markdown_report from modules.feedback_loop import review_ai_decisions def load_static_reason_workers(): """ Загружает реестр сотрудников со статическими/особыми причинами из static_reason_workers.csv. Возвращает словарь: {fio_clean: reason} """ 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 aggregate_scud_by_employee(df, debug=False): """ Агрегирует записи СКУД по сотруднику (fio_clean). Устраняет дубликаты карт/табельных номеров (например, Кондрашова Е.В.): Если человек пришел хотя бы по одной карте — Пришел = True. """ if 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 if debug and len(group) > 1: print(f" [DEBUG] Найдено дубликатов карт/записей для {fio_clean}: {len(group)} шт. Статус присутствия: {is_present}") aggregated.append(best_row) return pd.DataFrame(aggregated) def main(): parser = argparse.ArgumentParser(description="Система модульного контроллинга СКУД ⟷ 1С") parser.add_argument('-d', '--debug', action='store_true', help="Запуск в режиме отладки (подробный вывод)") args = parser.parse_args() DEBUG = args.debug print("=" * 60) print(f"ЗАПУСК СИСТЕМЫ МОДУЛЬНОГО КОНТРОЛЛИНГА СКУД ⟷ 1С {'[DEBUG MODE]' if DEBUG else ''}") print("=" * 60) # 0. Валидация файлов print("[0/5] Проверка актуальности и свежести входных файлов...") is_valid, warnings, errors = 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") # 1. Загрузка данных print("[1/5] Загрузка данных СКУД, 1С, реестра статических причин и исключений...") raw_scud_df, staff_fios_clean, df_absent = load_all_data() static_reasons_dict = load_static_reason_workers() if DEBUG: print(f" [DEBUG] Загружено записей СКУД: {len(raw_scud_df)}") print(f" [DEBUG] Загружено сотрудников из 1С: {len(staff_fios_clean)}") print(f" [DEBUG] Загружено докум. отсутствий 1С: {len(df_absent)}") print(f" [DEBUG] Загружено статических причин: {len(static_reasons_dict)}") # 2. ИИ-сверка ФИО print("[2/5] Проверка опечаток и сопоставление ФИО через ИИ...") scud_unrecognized = raw_scud_df[~raw_scud_df['fio_clean'].isin(staff_fios_clean)]['fio_clean'].tolist() fio_mapping = ai_verify_scud_against_staff(scud_unrecognized, staff_fios_clean) if fio_mapping: print(f" [ИИ] Найдено и исправлено совпадений/опечаток: {len(fio_mapping)}") raw_scud_df['fio_clean'] = raw_scud_df['fio_clean'].apply( lambda x: fio_mapping[x]['staff_fio'] if x in fio_mapping else x ) # 2.1 Агрегация нескольких карт/дубликатов raw_scud_df = aggregate_scud_by_employee(raw_scud_df, debug=DEBUG) # 3. Объединение данных и иерархия приоритетов print("[3/5] Объединение реестров 1С, СКУД, статических причин и исключений...") merged = raw_scud_df[raw_scud_df['fio_clean'].isin(staff_fios_clean)].copy() # 🎯 ПРИОРИТЕТ 1: Выгрузка докум. отсутствий 1С:ЗУП (Отсутствия сотрудников - выгрузка.xlsx) merged = merged.merge(df_absent, on='fio_clean', how='left') # 🎯 ПРИОРИТЕТ 2: Статические причины из static_reason_workers.csv # Применяются ТОЛЬКО если в 1С:ЗУП документа нет и сотрудник отсутствует в СКУД if static_reasons_dict: for fio_clean, reason_val in static_reasons_dict.items(): mask_apply_static = ( (merged['Пришел'] == False) & (merged['Вид_отсутствия'].isna() | (merged['Вид_отсутствия'].astype(str).str.strip() == '')) & (merged['fio_clean'] == fio_clean) ) merged.loc[mask_apply_static, 'Вид_отсутствия'] = reason_val # 🎯 ПРИОРИТЕТ 3: Исключения по должности/департаменту/ФИО (exceptions.json) # Если сотрудник попал в исключения (is_excluded = True), НЕ пришел в СКУД и НЕ имеет документа в 1С: # Присваиваем ему статус "Исключение (ОВК/Подрядчики)", чтобы он НЕ ПОПАДАЛ в неизвестные! mask_apply_exceptions = ( (merged['Пришел'] == False) & (merged['Вид_отсутствия'].isna() | (merged['Вид_отсутствия'].astype(str).str.strip() == '')) & (merged.get('is_excluded', False) == True) ) merged.loc[mask_apply_exceptions, 'Вид_отсутствия'] = 'Исключение (ОВК/Подрядчики)' absent_explained = merged[(merged['Пришел'] == False) & (merged['Вид_отсутствия'].notna())] absent_unexplained = merged[(merged['Пришел'] == False) & (merged['Вид_отсутствия'].isna())] scud_present_but_absent_in_1c = merged[(merged['Пришел'] == True) & (merged['Вид_отсутствия'].notna())] if DEBUG: print("\n--- [DEBUG STATS] ---") print(f" Всего по списку: {len(merged)}") print(f" Присутствуют на рабочем месте: {len(merged[merged['Пришел'] == True])}") print(f" Официально отсутствуют (1С + стат. причины + исключения): {len(absent_explained)}") print(f" Истинно неизвестные случаи: {len(absent_unexplained)}") if not absent_unexplained.empty: print(" Список неизвестных:") for fio in absent_unexplained['Сотрудник'].dropna().tolist(): print(f" - {fio}") print("---------------------\n") # 4. ИИ-анализ и интерактивная проверка (HITL) print("[4/5] Запуск ИИ-аудитора и проверка аномалий...") report_text = generate_markdown_report( merged_df=merged, absent_explained=absent_explained, absent_unexplained=absent_unexplained, scud_present_but_absent_in_1c=scud_present_but_absent_in_1c, raw_scud_df=raw_scud_df, date_str=DATE_TODAY ) suspicious_cases = [] if not scud_present_but_absent_in_1c.empty: for _, row in scud_present_but_absent_in_1c.iterrows(): suspicious_cases.append({ 'fio_target': row.get('Сотрудник', ''), 'reason': f"Пришел по СКУД, но в 1С зафиксировано: {row.get('Вид_отсутствия', '')}" }) review_ai_decisions(report_text, suspicious_cases) # 5. Формирование Excel и Markdown отчетов print("[5/5] Формирование итоговых Excel и Markdown отчетов...") generate_summary_excel(merged_df=merged, date_str=DATE_TODAY) generate_detailed_excel(merged_df=merged, date_str=DATE_YESTERDAY) print("\n" + "=" * 60) print("ГОТОВАЯ ТЕКСТОВАЯ СВОДКА ИИ:") print("=" * 60) print(report_text) 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}") if __name__ == "__main__": main()