# 📦 ETL-СЛЕПОК ИСХОДНОГО КОДА (СКУД ⟷ 1С & DB CORE) ## File: `./config.py` ```py import os import re from datetime import datetime, timedelta BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DATA_DIR = os.path.join(BASE_DIR, "data") SCUD_DIR = os.path.join(DATA_DIR, "scud") ZUP_1C_DIR = os.path.join(DATA_DIR, "1c") OUTPUT_DIR = os.path.join(BASE_DIR, "output") REPORTS_DIR = os.path.join(OUTPUT_DIR, "reports") SHARE_1C_DIR = "/mnt/scud_share" for folder in [DATA_DIR, SCUD_DIR, ZUP_1C_DIR, OUTPUT_DIR, REPORTS_DIR]: os.makedirs(folder, exist_ok=True) NOW = datetime.now() DATE_TODAY = NOW.strftime("%d.%m.%Y") if NOW.weekday() == 0: DATE_YESTERDAY = (NOW - timedelta(days=3)).strftime("%d.%m.%Y") else: DATE_YESTERDAY = (NOW - timedelta(days=1)).strftime("%d.%m.%Y") OLLAMA_URL = "http://10.121.17.227:11434/api/generate" OLLAMA_MODEL = "qwen2.5:14b" MODEL_NAME = OLLAMA_MODEL KNOWLEDGE_BASE_PATH = os.path.join(DATA_DIR, "knowledge_base.json") EXCEPTIONS_PATH = os.path.join(BASE_DIR, "exceptions.json") ZUP_SQL_CONFIG = { "driver": "{ODBC Driver 18 for SQL Server}", "server": os.getenv("ZUP_SQL_SERVER", "ACCOUNT-01"), "database": os.getenv("ZUP_SQL_DB", "ZUP30"), "user": os.getenv("ZUP_SQL_USER", "scud_reader"), "password": os.getenv("ZUP_SQL_PASS", "Rhfcysq90"), "trust_server_certificate": "yes", "encrypt": "no" } def find_dated_file(prefix, date_str, search_dirs=None): if search_dirs is None: search_dirs = [ZUP_1C_DIR, SCUD_DIR, DATA_DIR, "."] date_dots = str(date_str).replace('_', '.') date_underscores = date_dots.replace('.', '_') for d in search_dirs: if not os.path.exists(d): continue for f in os.listdir(d): if f.endswith('.xlsx') or f.endswith('.csv'): if f.lower().startswith(prefix.lower()): if date_dots in f or date_underscores in f: return os.path.join(d, f) return None def normalize_fio(fio): if not fio or not isinstance(fio, str): return "" fio_clean = re.sub(r'\(.*?\)', '', fio) fio_clean = fio_clean.replace('\xa0', ' ') parts = fio_clean.strip().split() return " ".join(parts).title() def clean_scud_fio_light(fio_str): return normalize_fio(fio_str) def load_exceptions(): """ Приоритетно читает исключения и белый список из SQLite таблицы exceptions_registry. При отсутствии таблицы или пустой базе выполняет fallback на exceptions.json. """ try: from services.exceptions_repo import get_all_exceptions_from_db db_exc = get_all_exceptions_from_db() if any(db_exc.values()): return db_exc except Exception: pass import json if os.path.exists(EXCEPTIONS_PATH): try: with open(EXCEPTIONS_PATH, "r", encoding="utf-8") as f: return json.load(f) except Exception: pass return {"fio": [], "departments": [], "positions": [], "position_keywords": [], "include_fio": []} ``` ## File: `./exceptions.json` ```json { "departments": [ "ОВК" ], "positions": [ "Уборщик производственных помещений", "Уборщик служебных помещений" ], "fio": [ "Таткало Валерий Валерьевич", "Петренюк Андрей Германович", "Михалев Сергей Геннадьевич" ], "position_keywords": [ "уборщик", "клинер", "дворник", "гардероб", "рабочий по обслуживанию" ], "include_fio": [ "Тарасенко Александр Александрович", "Журиков Михаил Николаевич" ] } ``` ## File: `./main_etl.py` ```py """ =============================================================================== FILE: main_etl.py PROJECT: SCUD Orion AI (Unified Architecture) ROLE: Главная точка входа ETL-конвейера СКУД ⟷ 1С:ЗУП. =============================================================================== """ import os import sys import argparse import logging import pandas as pd from datetime import datetime, timedelta from services.scud_etl.pipeline import load_best_snapshot_for_date, load_1c_files_for_date from services.scud_etl.merger import merge_scud_and_1c, calculate_summary_metrics from services.scud_etl.anomaly_detector import detect_registry_anomalies from services.scud_etl.svodka_generator import generate_svodka_service from services.scud_etl.otchet_generator import generate_otchet_service from services.text_reporter import generate_markdown_report from services.snapshots.retention import cleanup_old_intermediate_snapshots from services.scud_export import run_export from services.share_copier import copy_1c_files_from_share from services.excel_exporter import export_raw_scud logging.basicConfig(level=logging.INFO, format="[%(asctime)s] [%(levelname)s] %(message)s") def print_help(): print(""" =============================================================================== 🛠️ SCUD ORION AI — СИСТЕМА КОНТРОЛЛИНГА И СВОДНЫХ ОТЧЕТОВ =============================================================================== Использование: python main_etl.py [ОПЦИИ] Доступные аргументы: -h, --help, help Показать эту справку и выйти --date ДД.ММ.ГГГГ Дата расчета (по умолчанию: текущий рабочий день) --time ЧЧ:ММ Время среза для сводки (например: 14:30) Ищет ближайший срез (±20 мин) или запрашивает On-Demand экспорт --snapshot ID Точный ID снапшота для расчета (например: 20260827-002) --skip-export Пропустить выгрузку СКУД из MS SQL (работать только с SQLite) --export-only ТОЛЬКО сделать экспорт/снапшот СКУД в БД без построения отчетов -d, --debug Включить режим расширенной отладки Примеры использования: python main_etl.py 👉 Полный суточный цикл: экспорт -> отчет за вчера -> сводка за сегодня -> ИИ. python main_etl.py --export-only 👉 Почасовой тихий срез в БД (для cron) без генерации отчетов. python main_etl.py --date 27.08.2026 --time 12:15 --skip-export 👉 Построить сводку за 27.08 на 12:15 без запроса к внешнему MS SQL. python main_etl.py --snapshot 20260827-001 --skip-export 👉 Расчет отчетов строго по выбранному снапшоту из SQLite. =============================================================================== """) def main(): if len(sys.argv) > 1 and sys.argv[1] in ("-h", "--help", "help"): print_help() sys.exit(0) parser = argparse.ArgumentParser(add_help=False) parser.add_argument("-h", "--help", action="store_true") parser.add_argument("-d", "--debug", action="store_true") parser.add_argument("--skip-export", action="store_true") parser.add_argument("--export-only", action="store_true") parser.add_argument("--date", type=str, default=None) parser.add_argument("--time", type=str, default=None) parser.add_argument("--snapshot", type=str, default=None) args = parser.parse_args() if args.help: print_help() sys.exit(0) print("=" * 60) print(f"ЗАПУСК СИСТЕМЫ МОДУЛЬНОГО КОНТРОЛЛИНГА СКУД ⟷ 1С {'[DEBUG]' if args.debug else ''}") print("=" * 60) # Автоматическая ротация архивных почасовых срезов if not args.skip_export and not args.snapshot: deleted_count = cleanup_old_intermediate_snapshots(days_to_keep_all=2) if deleted_count > 0: print(f"[🧹] Ротация БД: очищено {deleted_count} строк промежуточных архивных срезов.") now = datetime.now() if args.date: today_str = args.date.replace('_', '.') dt_target = datetime.strptime(today_str, "%d.%m.%Y") days_back = 3 if dt_target.weekday() == 0 else 1 yesterday_str = (dt_target - timedelta(days=days_back)).strftime("%d.%m.%Y") else: today_str = now.strftime("%d.%m.%Y") if now.weekday() == 0: yesterday_str = (now - timedelta(days=3)).strftime("%d.%m.%Y") else: yesterday_str = (now - timedelta(days=1)).strftime("%d.%m.%Y") # [Этап 0] Выгрузка свежих данных СКУД if not args.skip_export and not args.snapshot: print(f"\n[0/5] Экспорт данных СКУД за {today_str} и {yesterday_str}...") run_export(input_date=args.date, debug=args.debug, save_xlsx=True) else: print("\n[0/5] Пропуск прямого экспорта СКУД из MS SQL (--skip-export)...") # Если запрошен режим тихого почасового среза — выходим без тяжелых генераций if args.export_only: print(f"\n[✓] Режим --export-only: срез зафиксирован в SQLite. Генерация отчетов пропущена.") sys.exit(0) # [Этап 0.5] Синхронизация файлов с шары 1С if not args.snapshot and not args.skip_export: print(f"\n[0.5/5] Проверка и копирование файлов 1С с шары...") copy_1c_files_from_share() else: print("\n[0.5/5] Пропуск синхронизации с шары (чтение локальных данных)...") # [Этап 1-2] Загрузка данных print(f"\n[1-2/5] Загрузка срезов: Сегодня = {today_str}, Накануне = {yesterday_str}...") df_scud_yesterday = load_best_snapshot_for_date(yesterday_str, prefer_final_y=True) df_scud_today = load_best_snapshot_for_date(today_str, prefer_final_y=False) df_staff_yesterday, df_abs_yesterday = load_1c_files_for_date(yesterday_str) df_staff_today, df_abs_today = load_1c_files_for_date(today_str) if df_scud_today is not None and not df_scud_today.empty: export_raw_scud(df_scud_today, filename=f"СКУД_Сырые_данные_{today_str}.xlsx") # [Этап 3] Детальный отчет за вчера через otchet_generator print(f"\n[3/5] Обработка и построение детального отчета за ВЧЕРА ({yesterday_str})...") res_otchet = generate_otchet_service(target_date=yesterday_str) if res_otchet.get("status") == "success": print(f"[✓] {res_otchet.get('message')}: {res_otchet.get('filepath')}") # [Этап 4] Сводка за сегодня через svodka_generator print(f"\n[4/5] Обработка и построение Ежедневной сводки за {today_str} {args.time or ''}...") res_svodka = generate_svodka_service( target_date=today_str, target_time=args.time, snapshot_id=args.snapshot ) if res_svodka.get("status") == "success": print(f"[✓] {res_svodka.get('message')}: {res_svodka.get('filepath')}") if res_svodka.get("note"): print(f" ℹ️ {res_svodka.get('note')}") # [Этап 5] Формирование Markdown-сводки через ИИ-аудитора (Ollama) print(f"\n[5/5] Формирование Markdown-сводки через ИИ-аудитора (Ollama)...") df_merged_today = merge_scud_and_1c(df_scud_today, df_staff_today, df_abs_today) anomalies_today = detect_registry_anomalies(df_merged_today, df_raw_scud=df_scud_today) absent_explained = df_merged_today[ (df_merged_today['Пришел'] == False) & (df_merged_today['Вид_отсутствия'].notna()) & (~df_merged_today['Вид_отсутствия'].astype(str).str.startswith('Исключение')) ] absent_unexplained = df_merged_today[ (df_merged_today['Пришел'] == False) & (df_merged_today['Вид_отсутствия'].isna() | (df_merged_today['Вид_отсутствия'].astype(str).str.strip() == '')) & (df_merged_today.get('is_excluded', False) == False) ] summary_md = generate_markdown_report( merged_df=df_merged_today[df_merged_today.get('is_excluded', False) == False], absent_explained=absent_explained, absent_unexplained=absent_unexplained, scud_present_but_absent_in_1c=pd.DataFrame(), anomalies_list=anomalies_today, raw_scud_df=df_scud_today, raw_staff_df=df_staff_today, raw_absent_df=df_abs_today, date_str=today_str ) os.makedirs("output", exist_ok=True) md_file_path = f"output/Сводка_контроллинга_{today_str}.md" with open(md_file_path, "w", encoding="utf-8") as f: f.write(summary_md) print(f"[✓] Текстовый отчет сохранен в: {md_file_path}") print("\n" + "=" * 60) print("ГОТОВАЯ ТЕКСТОВАЯ СВОДКА ИИ-АУДИТОРА:") print("=" * 60) print(summary_md) if __name__ == "__main__": main() ``` ## File: `./run_cron_etl.sh` ```bash #!/bin/bash set -e # Переход в папку проекта cd /home/puh/projects/scud_ai # Создание папки для логов mkdir -p /home/puh/projects/scud_ai/logs # Фиксация старта echo "==================================================" >> /home/puh/projects/scud_ai/logs/cron_etl.log echo "[CRON START] $(date '+%Y-%m-%d %H:%M:%S')" >> /home/puh/projects/scud_ai/logs/cron_etl.log echo "==================================================" >> /home/puh/projects/scud_ai/logs/cron_etl.log # Запуск main_etl.py через интерпретатор окружения /home/puh/scud_orion_ai_v2/venv/bin/python /home/puh/projects/scud_ai/main_etl.py >> /home/puh/projects/scud_ai/logs/cron_etl.log 2>&1 # Фиксация окончания echo "[CRON FINISH] $(date '+%Y-%m-%d %H:%M:%S')" >> /home/puh/projects/scud_ai/logs/cron_etl.log echo "" >> /home/puh/projects/scud_ai/logs/cron_etl.log ``` ## File: `./scripts/db_cli.py` ```py import os import sys import argparse import sqlite3 import pandas as pd from datetime import datetime sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from config import DATA_DIR, DATE_TODAY, OUTPUT_DIR, EXCEPTIONS_PATH, normalize_fio from core.database import ( get_connection, get_available_snapshots, get_all_rules_from_db, load_scud_from_db_by_snapshot, get_latest_snapshot_time ) from services.exceptions_repo import ( get_all_exceptions_from_db, add_exception_to_db, remove_exception_from_db, sync_json_to_db ) DB_PATH = os.path.join(DATA_DIR, "scud_orion_ai.db") def cmd_mapping(args_list): """Управление подтвержденными сопоставлениями ФИО (СКУД <-> 1С:ЗУП).""" with get_connection() as conn: cursor = conn.cursor() if not args_list or args_list[0] in ["list", "show"]: cursor.execute("SELECT id, scud_fio, zup_fio, match_source, status FROM person_identity_mapping ORDER BY id DESC") rows = cursor.fetchall() print("\n🔗 СОХРАНЕННЫЕ СОПОСТАВЛЕНИЯ ФИО (person_identity_mapping):") print("=" * 80) if rows: for r in rows: print(f" #{r[0]} [{r[4]}] СКУД: '{r[1]}' ⟷ 1С: '{r[2]}' ({r[3]})") else: print(" — сопоставлений пока нет") print("=" * 80 + "\n") return subcmd = args_list[0] if subcmd == "add": if len(args_list) < 3: print("Использование: python scripts/db_cli.py mapping add 'ФИО в СКУД' 'ФИО в 1С'") return scud_f, zup_f = args_list[1], args_list[2] cursor.execute(""" INSERT OR REPLACE INTO person_identity_mapping (scud_fio, zup_fio, match_source, status) VALUES (?, ?, 'MANUAL', 'ACTIVE') """, (scud_f.strip(), zup_f.strip())) conn.commit() print(f"✅ Успешно добавлена связка: '{scud_f}' ⟷ '{zup_f}'") elif subcmd in ["del", "delete", "remove"]: if len(args_list) < 2: print("Использование: python scripts/db_cli.py mapping del 'ФИО в СКУД'") return cursor.execute("DELETE FROM person_identity_mapping WHERE scud_fio = ?", (args_list[1].strip(),)) conn.commit() print(f"✅ Связка для '{args_list[1]}' удалена.") def print_tool_actions(): """Выводит реестр декларативных действий инструментов и шаблоны кнопок.""" print("\n" + "=" * 110) print("🧰 ДЕКЛАРАТИВНЫЙ РЕЕСТР ДЕЙСТВИЙ ИНСТРУМЕНТОВ (tool_action_registry):") print("=" * 110) with get_connection() as conn: cursor = conn.cursor() try: cursor.execute(""" SELECT id, tool_name, category, bypass_llm, success_template, follow_up_question, buttons_json FROM tool_action_registry WHERE is_active = 1 ORDER BY id ASC """) rows = cursor.fetchall() if not rows: print("Таблица tool_action_registry пуста.") else: for r in rows: print(f"ID: {r[0]} | Tool: [{r[1]}] | Категория: {r[2]} | Bypass LLM: {'ДА (0.05с)' if r[3] else 'НЕТ'}") print(f" • Сообщение: {r[4]}") if r[5]: print(f" • Вопрос: {r[5]}") print(f" • Кнопки: {r[6]}") print("-" * 110) except Exception as e: print(f"Таблица tool_action_registry недоступна: {e}") print("=" * 110 + "\n") def print_stats(): """Выводит общую статистику по записям в таблицах БД.""" print("\n" + "=" * 60) print("📊 СТАТИСТИКА БАЗЫ ДАННЫХ SQLITE (scud_orion_ai.db):") print("=" * 60) with get_connection() as conn: cursor = conn.cursor() tables = [ 'scud_logs', 'zup_staff', 'zup_absences', 'anomalies_history', 'ai_knowledge_base', 'chat_messages', 'session_states', 'system_prompt_nodes', 'tasks', 'exceptions_registry' ] for t in tables: try: cursor.execute(f"SELECT COUNT(*) FROM {t}") cnt = cursor.fetchone()[0] print(f" • Таблица [{t:<22}]: {cnt:>6} записей") except Exception: pass print("=" * 60 + "\n") def print_snapshots_list(date_str=None): """Выводит реестр снапшотов с отображением даты и точного времени среза.""" rows = get_available_snapshots(date_str) print("\n" + "=" * 105) print(f"📸 РЕЕСТР СОХРАНЕННЫХ СНАПШОТОВ (СВЕРХУ СВЕЖИЕ) {'ЗА ЛОГИ ' + date_str if date_str else ''}:") print("=" * 105) header = f"{'ID снапшота':<16} | {'Дата снапшота (создания)':<24} | {'Дата и время среза':<20} | {'Записей':<8}" print(header) print("-" * 105) if not rows: print("Снапшотов пока нет.") print("=" * 105 + "\n") return def snapshot_sort_key(row): snap_id = row[0] or "" snap_time = row[2] or "" seq_num = 0 if "-" in snap_id: parts = snap_id.replace("Y", "").split("-") if len(parts) > 1 and parts[1].isdigit(): seq_num = int(parts[1]) return (snap_time, seq_num) sorted_rows = sorted(rows, key=snapshot_sort_key, reverse=True) for r in sorted_rows: snap_id = r[0] if r[0] else '----------' log_date = r[1] if r[1] else '—' snap_time = r[2] if r[2] else '—' count = r[3] time_part = "—" if snap_time and " " in snap_time: time_part = snap_time.split(" ")[1] slice_datetime_str = f"{log_date} {time_part}" if time_part != "—" else log_date formatted_snap_id = f" {snap_id}" if not snap_id.startswith("Y") else snap_id print(f"{formatted_snap_id:<16} | {snap_time:<24} | {slice_datetime_str:<20} | {count:<8}") print("=" * 105 + "\n") def delete_snapshot_by_id(snapshot_id: str): """Удаляет конкретный снапшот из таблицы scud_logs по его ID.""" with get_connection() as conn: cursor = conn.cursor() cursor.execute("DELETE FROM scud_logs WHERE snapshot_id = ?", (snapshot_id,)) deleted_count = cursor.rowcount conn.commit() print(f"\n[✓] Успешно удален снапшот [{snapshot_id}]. Удалено строк: {deleted_count}\n") return deleted_count def delete_snapshots_by_date(date_str: str): """Удаляет все снапшоты за указанную дату (например, '04.08.2026').""" with get_connection() as conn: cursor = conn.cursor() cursor.execute("DELETE FROM scud_logs WHERE log_date = ? OR snapshot_id LIKE ?", (date_str, f"%{date_str.replace('.', '')}%")) deleted_count = cursor.rowcount conn.commit() print(f"\n[✓] Успешно удалены все снапшоты за дату [{date_str}]. Удалено строк: {deleted_count}\n") return deleted_count def inspect_scud(snapshot_id=None, date_str=None, export_xlsx=None): """Инспектирует логи СКУД за выбранный снапшот или дату и опционально сохраняет XLSX.""" target_date = date_str if date_str else DATE_TODAY print("\n" + "=" * 115) if snapshot_id: print(f"🔍 ИНСПЕКЦИЯ СКУД ПО СНАПШОТУ [{snapshot_id}] (Дата среза: {target_date}):") else: print(f"🔍 ИНСПЕКЦИЯ СКУД ЗА ТЕКУЩУЮ ДАТУ [{target_date}] (ПОСЛЕДНИЙ СРЕЗ):") print("=" * 115) df = load_scud_from_db_by_snapshot(target_date, snapshot_param=snapshot_id) if df is None or df.empty: print("Записи СКУД не найдены.") print("=" * 115 + "\n") return fio_col = next((c for c in ['Сотрудник', 'fio', 'fio_clean'] if c in df.columns), None) dept_col = next((c for c in ['Подразделение', 'department_scud', 'department'] if c in df.columns), None) pos_col = next((c for c in ['Должность', 'position'] if c in df.columns), None) in_col = next((c for c in ['Начало_дня', 'time_in'] if c in df.columns), None) first_act_col = next((c for c in ['Первая_активность', 'first_activity'] if c in df.columns), None) out_col = next((c for c in ['Конец_дня', 'time_out'] if c in df.columns), None) dur_col = next((c for c in ['Находился_в_здании', 'time_in_building', 'duration'] if c in df.columns), None) present_col = next((c for c in ['Пришел', 'is_present'] if c in df.columns), None) anom_col = 'anomaly_flag' if 'anomaly_flag' in df.columns else None snap_col = 'snapshot_id' if 'snapshot_id' in df.columns else None total = len(df) if present_col: present_cnt = len(df[df[present_col].astype(str).str.lower().isin(['true', '1'])]) else: present_cnt = 0 absent_cnt = total - present_cnt print(f"Всего записей: {total} | Присутствовали: {present_cnt} | Отсутствовали: {absent_cnt}") print("-" * 115) display_cols = [c for c in [fio_col, dept_col, in_col, first_act_col, out_col, dur_col, present_col, anom_col, snap_col] if c] print(df[display_cols].head(30).to_string(index=False)) if len(df) > 30: print(f"\n... и ещё {len(df) - 30} строк.") if export_xlsx: out_path = export_xlsx if export_xlsx.endswith('.xlsx') else f"{export_xlsx}.xlsx" if not os.path.isabs(out_path): out_path = os.path.join(OUTPUT_DIR, out_path) df.to_excel(out_path, index=False) print("\n" + "*" * 115) print(f"[✓] УСПЕШНЫЙ ЭКСПОРТ ДЕБАГ-ФАЙЛА В EXCEL: {out_path}") print("*" * 115) print("=" * 115 + "\n") def print_absences(date_str=None): """Выводит список официально отсутствующих сотрудников из 1С:ЗУП за выбранный день.""" target_date = date_str if date_str else DATE_TODAY print("\n" + "=" * 90) print(f"📋 ОФИЦИАЛЬНЫЕ ОТСУТСТВИЯ ИЗ 1С:ЗУП ЗА ДАТУ [{target_date}]:") print("=" * 90) with get_connection() as conn: df = pd.read_sql_query( "SELECT fio as 'ФИО', absence_type as 'Причина отсутствия 1С' FROM zup_absences WHERE absence_date = ? ORDER BY absence_type, fio", conn, params=(target_date,) ) if df.empty: print(f"Записи об отсутствиях 1С за {target_date} в базе не найдены.") else: print(f"Всего зафиксировано документов 1С: {len(df)}") print("-" * 90) print(df.to_string(index=False)) print("=" * 90 + "\n") def print_anomalies(): """Выводит список аномалий СКУД из БД.""" print("\n" + "=" * 80) print("🚨 ИСТОРИЯ НАЙДЕННЫХ АНОМАЛИЙ СКУД ⟷ 1С:") print("=" * 80) with get_connection() as conn: df = pd.read_sql_query("SELECT anomaly_date, fio, anomaly_type, details FROM anomalies_history ORDER BY id DESC LIMIT 50", conn) if df.empty: print("Аномалии не найдены.") else: print(df.to_string(index=False)) print("=" * 80 + "\n") def print_rules(): """Выводит правила базы знаний ИИ.""" rules = get_all_rules_from_db() print("\n" + "=" * 80) print("🧠 ПРАВИЛА БАЗЫ ЗНАНИЙ ИИ:") print("=" * 80) if not rules: print("База знаний пуста.") else: for idx, r in enumerate(rules, 1): print(f" {idx}. {r}") print("=" * 80 + "\n") def dump_all_to_excel(out_filename="db_dump_full.xlsx"): """Дампит всю базу SQLite во многостраничный Excel.""" out_path = os.path.join(OUTPUT_DIR, out_filename) print(f"\n[🔄] Создание полного дампа БД в файл: {out_path} ...") with get_connection() as conn, pd.ExcelWriter(out_path, engine='openpyxl') as writer: for table in ['scud_logs', 'zup_staff', 'zup_absences', 'anomalies_history', 'ai_knowledge_base', 'chat_messages', 'session_states', 'system_prompt_nodes', 'tasks', 'exceptions_registry']: try: df = pd.read_sql_query(f"SELECT * FROM {table}", conn) df.to_excel(writer, sheet_name=table[:31], index=False) except Exception: pass print(f"[✓] Дамп успешно сохранен: {out_path}\n") def print_system_prompts(): """Выводит все системные промпты из базы данных.""" print("\n" + "=" * 80) print("📝 СИСТЕМНЫЕ ПРОМПТЫ (system_prompt_nodes):") print("=" * 80) with get_connection() as conn: cursor = conn.cursor() try: cursor.execute("SELECT section_id, item_id, content, is_active, updated_at FROM system_prompt_nodes ORDER BY section_id ASC, item_id ASC") rows = cursor.fetchall() except Exception: rows = [] if not rows: print("Таблица system_prompt_nodes пуста.") else: for r in rows: print(f"Раздел {r[0]}.{r[1]} | Active: {r[3]} | Updated: {r[4]}") print(f" {r[2]}") print("-" * 80) print("=" * 80 + "\n") def print_session_states(): """Выводит текущие активные сессии и превью (session_states).""" print("\n" + "=" * 80) print("🔄 АКТИВНЫЕ СЕССИИ И ПРЕВЬЮ (session_states):") print("=" * 80) with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT session_id, state_type, updated_at, pending_data FROM session_states") rows = cursor.fetchall() if not rows: print("Таблица session_states пуста (нет активных превью).") else: for r in rows: print(f"Session: {r[0]} | Type: {r[1]} | Updated: {r[2]}") print("-" * 80) print(f"Pending Data:\n{r[3]}\n") print("=" * 80 + "\n") def print_chat_messages(session_id=None, limit=50): """Выводит таблицу истории сообщений чата (контекст).""" print("\n" + "=" * 105) print(f"💬 ИСТОРИЯ СООБЩЕНИЙ ЧАТА (chat_messages) {f'для сессии: {session_id}' if session_id else 'все сессии'}:") print("=" * 105) with get_connection() as conn: cursor = conn.cursor() if session_id: cursor.execute(""" SELECT id, session_id, role, content, is_ephemeral, created_at FROM chat_messages WHERE session_id = ? ORDER BY id ASC LIMIT ? """, (session_id, limit)) else: cursor.execute(""" SELECT id, session_id, role, content, is_ephemeral, created_at FROM chat_messages ORDER BY id DESC LIMIT ? """, (limit,)) rows = cursor.fetchall() if not rows: print("Таблица chat_messages пуста.") else: if not session_id: rows = list(reversed(rows)) for r in rows: msg_id, sess, role, content, ephemeral, created = r eph_marker = " [ЭФЕМЕРНОЕ]" if ephemeral else "" print(f"[{msg_id}] {created} | Сессия: {sess} | Роль: {role.upper()}{eph_marker}") print("-" * 105) content_preview = content if content else "" print(f"{content_preview}") print("=" * 105) print("\n") def purge_chat_context(session_id=None, purge_all=False): """ Очистка контекста сообщений: - По умолчанию: удаляет эфемерные сообщения, осиротевшие превью и сбрасывает стейты сессий. - purge_all=True (--all): полностью очищает всю таблицу chat_messages и сбрасывает сессии. """ with get_connection() as conn: cursor = conn.cursor() if purge_all: if session_id: cursor.execute("DELETE FROM chat_messages WHERE session_id = ?", (session_id,)) cursor.execute("DELETE FROM session_states WHERE session_id = ?", (session_id,)) else: cursor.execute("DELETE FROM chat_messages") cursor.execute("DELETE FROM session_states") deleted_msgs = cursor.rowcount conn.commit() print(f"\n[✓] Полная очистка истории выполнена! Удалено сообщений: {deleted_msgs}\n") return query = """ DELETE FROM chat_messages WHERE is_ephemeral = 1 OR content LIKE '%Предпросмотр изменений%' OR content LIKE '%Удален пункт:%' OR content LIKE '%добавлен пункт:%' """ if session_id: cursor.execute(query + " AND session_id = ?", (session_id,)) cursor.execute("DELETE FROM session_states WHERE session_id = ?", (session_id,)) else: cursor.execute(query) cursor.execute("DELETE FROM session_states") deleted_msgs = cursor.rowcount conn.commit() print(f"\n[✓] Умная зачистка контекста выполнена! Удалено сообщений: {deleted_msgs}\n") # ⭐️ Новые функции управления исключениями (Exceptions & Whitelist) def print_exceptions(): """Выводит реестр исключений и белый список сотрудников из базы SQLite.""" exc = get_all_exceptions_from_db() print("\n" + "=" * 80) print("📋 РЕЕСТР ИСКЛЮЧЕНИЙ И БЕЛЫЙ СПИСОК (exceptions_registry):") print("=" * 80) for cat, items in exc.items(): print(f"[{cat.upper()}] ({len(items)} шт.):") if items: for it in items: print(f" • {it}") else: print(" — пусто") print("-" * 80) print("=" * 80 + "\n") HELP_TEXT = """ CLI-утилита инспекции и управления SQLite базой данных СКУД (scud_orion_ai.db) ДОСТУПНЫЕ КОМАНДЫ: stats -- Общая статистика строк по всем таблицам БД snapshots [ДД.ММ.ГГГГ] -- Посмотреть реестр снапшотов (опционально за конкретную дату) scud [ДД.ММ.ГГГГ] [--snapshot ID] [--export-xlsx NAME] -- Инспекция логов СКУД по дате/снапшоту и экспорт в Excel absences [ДД.ММ.ГГГГ] -- Посмотреть список официально отсутствующих из 1С:ЗУП anomalies -- Посмотреть историю найденных аномалий СКУД ⟷ 1С rules -- Посмотреть правила Базы Знаний ИИ из SQLite prompts -- Посмотреть узлы системного промпта (system_prompt_nodes) tools -- Посмотреть реестр действий инструментов и кнопок (tool_action_registry) sessions -- Посмотреть активные сессии и превью (session_states) context [session_id] [--limit N] -- Посмотреть таблицу контекста сообщений чата (chat_messages) context purge [session_id] [--all] -- Очистить контекст (умная зачистка или полная с флагом --all) dump [output.xlsx] -- Полный дамп всех таблиц БД в многостраничный Excel snapshot del [ID] или [--day ДД.ММ.ГГГГ] -- Удаление снапшота по ID или всех за выбранный день exceptions [list] -- Посмотреть реестр исключений и белый список (SQLite) exceptions add -c CATEGORY -v VALUE [-m COMMENT] -- Добавить исключение (fio, include_fio, departments, positions, position_keywords) exceptions del -c CATEGORY -v VALUE -- Удалить исключение из БД exceptions sync -- Синхронизировать exceptions.json -> SQLite ПРИМЕРЫ ЗАПУСКА: python scripts/db_cli.py stats python scripts/db_cli.py snapshots 06.08.2026 python scripts/db_cli.py scud 06.08.2026 --export-xlsx срез_четверг python scripts/db_cli.py absences 07.08.2026 python scripts/db_cli.py prompts python scripts/db_cli.py tools python scripts/db_cli.py sessions python scripts/db_cli.py context web_session_main --limit 20 python scripts/db_cli.py context purge python scripts/db_cli.py context purge --all python scripts/db_cli.py context purge web_session_main --all python scripts/db_cli.py snapshot del Y20260805-007 python scripts/db_cli.py dump my_dump.xlsx python scripts/db_cli.py exceptions python scripts/db_cli.py exceptions add -c include_fio -v "Тарасенко Александр Александрович" python scripts/db_cli.py exceptions add -c fio -v "Михалев Сергей Геннадьевич" -m "Уборщик" python scripts/db_cli.py exceptions del -c fio -v "Михалев Сергей Геннадьевич" python scripts/db_cli.py exceptions sync """ def main(): if any(arg in sys.argv for arg in ['-h', '--help']): print(HELP_TEXT) sys.exit(0) parser = argparse.ArgumentParser( description=HELP_TEXT, formatter_class=argparse.RawDescriptionHelpFormatter, add_help=False ) parser.add_argument('command', nargs='?', default=None, choices=[ 'stats', 'snapshots', 'scud', 'absences', 'anomalies', 'rules', 'prompts', 'sessions', 'dump', 'snapshot', 'tools', 'context', 'exceptions' ], help="Основная команда") parser.add_argument('action', nargs='?', default=None, help="Действие ('del', 'purge', 'add', 'sync') или дата/сессия") parser.add_argument('param', nargs='?', default=None, help="Параметр (дата, ID снапшота, session_id, имя файла)") parser.add_argument('-c', '--category', type=str, default=None, choices=['departments', 'positions', 'fio', 'position_keywords', 'include_fio'], help="Категория исключения") parser.add_argument('-v', '--value', type=str, default=None, help="Значение исключения (ФИО, отдел, должность)") parser.add_argument('-m', '--comment', type=str, default="", help="Комментарий к исключению") parser.add_argument('--snapshot', type=str, default=None, help="ID конкретного снапшота для инспекции") parser.add_argument('--export-xlsx', type=str, default=None, help="Экспорт среза СКУД в Excel-файл") parser.add_argument('--day', type=str, default=None, help="Удалить снапшоты за конкретный день (ДД.ММ.ГГГГ)") parser.add_argument('--limit', type=int, default=50, help="Лимит выводимых сообщений чата (для команды context)") parser.add_argument('--all', action='store_true', help="Полная очистка всех сообщений (для команды context purge)") if len(sys.argv) == 1: print_stats() return args = parser.parse_args() if args.command == 'stats': print_stats() elif args.command == 'snapshots': date_val = args.action or args.param print_snapshots_list(date_str=date_val) elif args.command == 'scud': date_val = args.action or args.param inspect_scud(snapshot_id=args.snapshot, date_str=date_val, export_xlsx=args.export_xlsx) elif args.command == 'absences': date_val = args.action or args.param print_absences(date_str=date_val) elif args.command == 'anomalies': print_anomalies() elif args.command == 'rules': print_rules() elif args.command == 'prompts': print_system_prompts() elif args.command == 'sessions': print_session_states() elif args.command == 'tools': print_tool_actions() elif args.command == 'context': if args.action in ['purge', 'clear']: is_all = args.all or (args.param == '--all') sess_id = None if (args.param == '--all' or not args.param) else args.param purge_chat_context(session_id=sess_id, purge_all=is_all) else: sess_id = args.action if args.action else None print_chat_messages(session_id=sess_id, limit=args.limit) elif args.command == 'dump': filename = args.action or args.param or "db_dump_full.xlsx" dump_all_to_excel(filename) elif args.command == 'snapshot': if args.action == 'del': if args.day: delete_snapshots_by_date(args.day) elif args.param: delete_snapshot_by_id(args.param) else: print("\n[❌] Ошибка: Не указан ID снапшота или параметр --day для удаления.") print("Пример: python scripts/db_cli.py snapshot del Y20260805-007\n") else: print(f"\n[❌] Ошибка: Неизвестное действие '{args.action}' для команды snapshot.") print("Используйте: python scripts/db_cli.py snapshot del [ID или --day 'ДД.ММ.ГГГГ']\n") elif args.command == 'exceptions': if args.action == 'add': if not args.category or not args.value: print("\n[❌] Ошибка: Для добавления исключения укажите флаги -c/--category и -v/--value") print("Пример: python scripts/db_cli.py exceptions add -c include_fio -v \"Тарасенко Александр Александрович\"\n") return if add_exception_to_db(args.category, args.value, args.comment): print(f"\n[✓] Успешно добавлено исключение: [{args.category}] {args.value}\n") else: print(f"\n[❌] Ошибка добавления исключения [{args.category}] {args.value}\n") elif args.action in ['del', 'delete', 'remove']: if not args.category or not args.value: print("\n[❌] Ошибка: Для удаления исключения укажите флаги -c/--category и -v/--value") print("Пример: python scripts/db_cli.py exceptions del -c fio -v \"Михалев Сергей Геннадьевич\"\n") return if remove_exception_from_db(args.category, args.value): print(f"\n[✓] Успешно удалено исключение: [{args.category}] {args.value}\n") else: print(f"\n[⚠️] Запись не найдена в базе: [{args.category}] {args.value}\n") elif args.action == 'sync': sync_json_to_db() print("\n[✓] Синхронизация exceptions.json -> SQLite успешно завершена.\n") else: print_exceptions() else: print("\n[❌] Ошибка: Неизвестная команда.") print(HELP_TEXT) if __name__ == "__main__": main() ``` ## File: `./core/connection.py` ```py """ =============================================================================== FILE: core/connection.py PROJECT: SCUD Orion AI (Unified Architecture) ROLE: Единый менеджер подключений к базе данных SQLite (WAL mode, timeouts). =============================================================================== """ import os import sqlite3 from config import DATA_DIR DB_PATH = os.path.join(DATA_DIR, "scud_orion_ai.db") def get_connection(row_factory: bool = False) -> sqlite3.Connection: """ Создает оптимизированное подключение к SQLite. row_factory=True возвращает sqlite3.Row для доступа к полям по имени. """ conn = sqlite3.connect(DB_PATH, timeout=30.0) if row_factory: conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON;") conn.execute("PRAGMA journal_mode = WAL;") conn.execute("PRAGMA synchronous = NORMAL;") return conn ``` ## File: `./core/database.py` ```py """ =============================================================================== FILE: core/database.py ROLE: Фасад ядра базы данных с полной обратной совместимостью импортов. =============================================================================== """ from core.connection import get_connection, DB_PATH from core.schema import init_all_tables from core.repositories.scud_repo import ( has_scud_logs_for_date, has_yesterday_final_snapshot, get_or_create_snapshot_id, save_scud_to_db, get_latest_snapshot_time, load_scud_from_db_by_snapshot, get_available_snapshots, delete_snapshot_by_id, delete_snapshots_by_date ) from core.repositories.zup_repo import ( save_staff_to_db, load_staff_from_db, save_absences_to_db, load_absences_from_db, save_anomalies_to_db, get_all_rules_from_db, add_rule_to_db, get_department_synonyms_dict, add_department_synonym_to_db ) init_db = init_all_tables ``` ## File: `./core/schema.py` ```py """ =============================================================================== FILE: core/schema.py PROJECT: SCUD Orion AI (Unified Architecture) ROLE: DDL-схемы таблиц, создание индексов и инициализация базы данных. =============================================================================== """ import logging from core.connection import get_connection logger = logging.getLogger("DB_SCHEMA") def init_all_tables() -> None: """Инициализирует все таблицы и индексы системы.""" with get_connection() as conn: cursor = conn.cursor() # 1. Логи СКУД cursor.execute(""" CREATE TABLE IF NOT EXISTS scud_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, log_date TEXT NOT NULL, fio TEXT NOT NULL, fio_clean TEXT NOT NULL, department TEXT, position TEXT, time_in TEXT, first_activity TEXT DEFAULT '—', time_out TEXT, time_in_building TEXT, is_present INTEGER NOT NULL, anomaly_flag TEXT DEFAULT 'NONE', snapshot_time TEXT DEFAULT NULL, snapshot_id TEXT DEFAULT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) # 2. Кадровые реестры 1С cursor.execute(""" CREATE TABLE IF NOT EXISTS zup_staff ( id INTEGER PRIMARY KEY AUTOINCREMENT, snapshot_date TEXT NOT NULL, fio TEXT NOT NULL, fio_clean TEXT NOT NULL, department TEXT, position TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) cursor.execute(""" CREATE TABLE IF NOT EXISTS zup_absences ( id INTEGER PRIMARY KEY AUTOINCREMENT, absence_date TEXT NOT NULL, fio TEXT NOT NULL, fio_clean TEXT NOT NULL, absence_type TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) # 3. Аномалии, база знаний и синонимы cursor.execute(""" CREATE TABLE IF NOT EXISTS anomalies_history ( id INTEGER PRIMARY KEY AUTOINCREMENT, anomaly_date TEXT NOT NULL, fio TEXT NOT NULL, anomaly_type TEXT NOT NULL, details TEXT NOT NULL, human_status TEXT DEFAULT 'Pending', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) cursor.execute(""" CREATE TABLE IF NOT EXISTS ai_knowledge_base ( id INTEGER PRIMARY KEY AUTOINCREMENT, rule_text TEXT UNIQUE NOT NULL, added_by TEXT DEFAULT 'Human', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) cursor.execute(""" CREATE TABLE IF NOT EXISTS department_synonyms ( id INTEGER PRIMARY KEY AUTOINCREMENT, short_name TEXT UNIQUE NOT NULL, full_name TEXT NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) # 4. Исключения и Кэш сопоставлений личностей (ИИ / Ручной) cursor.execute(""" CREATE TABLE IF NOT EXISTS exceptions_registry ( id INTEGER PRIMARY KEY AUTOINCREMENT, category TEXT NOT NULL, value TEXT NOT NULL, comment TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(category, value) ); """) 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', -- 'AI', 'MANUAL', 'EXACT' status TEXT DEFAULT 'ACTIVE', -- 'ACTIVE', 'PROPOSED', 'REJECTED' confidence REAL DEFAULT 1.0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(scud_fio, zup_fio) ); """) # 5. Узлы системного промпта, сессии, сообщения и задачи cursor.execute(""" CREATE TABLE IF NOT EXISTS system_prompt_nodes ( id INTEGER PRIMARY KEY AUTOINCREMENT, prompt_name TEXT DEFAULT 'main_agent', section_id INTEGER NOT NULL, item_id INTEGER NOT NULL, content TEXT NOT NULL, is_active INTEGER DEFAULT 1, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(prompt_name, section_id, item_id) ); """) cursor.execute(""" CREATE TABLE IF NOT EXISTS session_states ( session_id TEXT PRIMARY KEY, state_type TEXT NOT NULL, pending_data TEXT, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) cursor.execute(""" CREATE TABLE IF NOT EXISTS chat_messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, session_id TEXT NOT NULL, role TEXT NOT NULL, content TEXT NOT NULL, is_ephemeral INTEGER DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) cursor.execute(""" CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id TEXT, module TEXT DEFAULT 'general', title TEXT NOT NULL, priority TEXT DEFAULT 'MEDIUM', status TEXT DEFAULT 'BACKLOG', due_date TEXT, user_id INTEGER DEFAULT 1, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """) # 6. Индексы cursor.execute("CREATE INDEX IF NOT EXISTS idx_scud_date ON scud_logs(log_date);") cursor.execute("CREATE INDEX IF NOT EXISTS idx_scud_fio ON scud_logs(fio_clean);") cursor.execute("CREATE INDEX IF NOT EXISTS idx_mapping_scud ON person_identity_mapping(scud_fio);") conn.commit() ``` ## File: `./core/repositories/scud_repo.py` ```py """ =============================================================================== FILE: core/repositories/scud_repo.py ROLE: Репозиторий логов СКУД, сохранение и загрузка снапшотов. =============================================================================== """ from datetime import datetime import pandas as pd from core.connection import get_connection def has_scud_logs_for_date(date_str: str) -> bool: with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT 1 FROM scud_logs WHERE log_date = ? LIMIT 1", (date_str,)) return cursor.fetchone() is not None def has_yesterday_final_snapshot(date_str: str) -> bool: with get_connection() as conn: cursor = conn.cursor() cursor.execute( "SELECT 1 FROM scud_logs WHERE log_date = ? AND (snapshot_id LIKE 'Y%' OR snapshot_time LIKE '%23:59:59' OR snapshot_time LIKE '%22:00:00') LIMIT 1", (date_str,) ) return cursor.fetchone() is not None def get_or_create_snapshot_id(snapshot_time: str, date_str: str = None, is_yesterday: bool = False) -> str: try: dt_snap = datetime.strptime(snapshot_time, "%Y-%m-%d %H:%M:%S").date() date_prefix = dt_snap.strftime("%Y%m%d") except (ValueError, TypeError): dt_snap = datetime.now().date() date_prefix = dt_snap.strftime("%Y%m%d") if date_str: try: dt_log = datetime.strptime(date_str, "%d.%m.%Y").date() if dt_log < dt_snap: is_yesterday = True except Exception: pass prefix = "Y" if is_yesterday else "" with get_connection() as conn: cursor = conn.cursor() # 1. Проверяем, существует ли уже срез с точно таким же временем и датой if date_str: cursor.execute( "SELECT snapshot_id FROM scud_logs WHERE log_date = ? AND snapshot_time = ? AND snapshot_id IS NOT NULL LIMIT 1", (date_str, snapshot_time) ) else: cursor.execute( "SELECT snapshot_id FROM scud_logs WHERE snapshot_time = ? AND snapshot_id IS NOT NULL LIMIT 1", (snapshot_time,) ) row = cursor.fetchone() if row and row[0]: return row[0] # 2. Извлекаем ВСЕ существующие ID за текущие календарные сутки cursor.execute(""" SELECT DISTINCT snapshot_id FROM scud_logs WHERE snapshot_id LIKE ? OR snapshot_id LIKE ? """, (f"{date_prefix}-%", f"Y{date_prefix}-%")) rows = cursor.fetchall() max_seq = 0 for (s_id,) in rows: if not s_id: continue try: # Извлекаем число после последнего дефиса parts = str(s_id).split('-') if len(parts) >= 2 and parts[-1].isdigit(): num = int(parts[-1]) if num > max_seq: max_seq = num except Exception: continue next_seq = max_seq + 1 return f"{prefix}{date_prefix}-{next_seq:03d}" def save_scud_to_db(df_scud: pd.DataFrame, date_str: str, snapshot_time: str = None, is_yesterday: bool = False) -> None: if df_scud is None or df_scud.empty: return if not snapshot_time: snapshot_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") snapshot_id = get_or_create_snapshot_id(snapshot_time, date_str=date_str, is_yesterday=is_yesterday) data_to_insert = [ ( date_str, r.get('Сотрудник', r.get('fio_raw', '')), r.get('fio_clean', ''), r.get('Подразделение', ''), r.get('Должность', ''), r.get('Начало_дня', 'Нет входа'), r.get('Первая_активность', '—'), r.get('Конец_дня', 'Нет выхода'), r.get('Находился_в_здании', '00:00'), 1 if r.get('Пришел', False) else 0, r.get('anomaly_flag', 'NONE'), snapshot_time, snapshot_id ) for _, r in df_scud.iterrows() ] with get_connection() as conn: cursor = conn.cursor() cursor.execute("DELETE FROM scud_logs WHERE log_date = ? AND snapshot_time = ?", (date_str, snapshot_time)) cursor.executemany(""" INSERT INTO scud_logs ( log_date, fio, fio_clean, department, position, time_in, first_activity, time_out, time_in_building, is_present, anomaly_flag, snapshot_time, snapshot_id ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, data_to_insert) conn.commit() def get_latest_snapshot_time(date_str: str = None): with get_connection() as conn: cursor = conn.cursor() if date_str: cursor.execute("SELECT snapshot_time FROM scud_logs WHERE log_date = ? AND snapshot_time IS NOT NULL ORDER BY snapshot_time DESC LIMIT 1", (date_str,)) else: cursor.execute("SELECT snapshot_time FROM scud_logs WHERE snapshot_time IS NOT NULL ORDER BY snapshot_time DESC LIMIT 1") row = cursor.fetchone() return row[0] if row else None def load_scud_from_db_by_snapshot(date_str: str, snapshot_param: str = None) -> pd.DataFrame: with get_connection() as conn: df = pd.DataFrame() # 1. Если передан конкретный ID снапшота (например 'Y20260820-004') if snapshot_param: df = pd.read_sql_query( "SELECT * FROM scud_logs WHERE snapshot_id = ?", conn, params=(str(snapshot_param),) ) # 2. Если ищем за дату (для вчерашнего дня строго ищем Y-снапшот) if df.empty and date_str: cursor = conn.cursor() # ⭐️ Жесткий приоритет 1: Ищем снапшот с префиксом 'Y' cursor.execute( "SELECT snapshot_id FROM scud_logs WHERE log_date = ? AND snapshot_id LIKE 'Y%' ORDER BY snapshot_time DESC, id DESC LIMIT 1", (date_str,) ) row = cursor.fetchone() # Приоритет 2: Если Y нет (например, за сегодня), берем самый свежий по времени if not row: cursor.execute( "SELECT snapshot_id FROM scud_logs WHERE log_date = ? ORDER BY snapshot_time DESC, id DESC LIMIT 1", (date_str,) ) row = cursor.fetchone() if row and row[0]: target_id = row[0] df = pd.read_sql_query( "SELECT * FROM scud_logs WHERE snapshot_id = ?", conn, params=(target_id,) ) if not df.empty: rename_map = { 'department': 'department_scud', 'position': 'Должность', 'fio': 'Сотрудник', 'time_in': 'Начало_дня', 'first_activity': 'Первая_активность', 'time_out': 'Конец_дня', 'time_in_building': 'Находился_в_здании', 'is_present': 'Пришел' } df = df.rename(columns={k: v for k, v in rename_map.items() if k in df.columns}) if 'department_scud' in df.columns and 'Подразделение' not in df.columns: df['Подразделение'] = df['department_scud'] for col in ['Пришел', 'Начало_дня', 'Первая_активность', 'Конец_дня', 'Находился_в_здании', 'anomaly_flag']: if col not in df.columns: df[col] = False if col == 'Пришел' else '—' if 'Пришел' in df.columns: df['Пришел'] = df['Пришел'].astype(bool) return df def get_available_snapshots(date_str: str = None): with get_connection() as conn: cursor = conn.cursor() query = """ SELECT snapshot_id, log_date, snapshot_time, COUNT(*) as cnt FROM scud_logs WHERE snapshot_time IS NOT NULL """ params = [] if date_str: query += " AND log_date = ?" params.append(date_str) query += " GROUP BY snapshot_id, log_date, snapshot_time ORDER BY snapshot_time DESC" cursor.execute(query, params) return cursor.fetchall() def delete_snapshot_by_id(snapshot_id: str) -> int: with get_connection() as conn: cursor = conn.cursor() cursor.execute("DELETE FROM scud_logs WHERE snapshot_id = ?", (snapshot_id,)) cnt = cursor.rowcount conn.commit() return cnt def delete_snapshots_by_date(date_str: str) -> int: with get_connection() as conn: cursor = conn.cursor() cursor.execute("DELETE FROM scud_logs WHERE log_date = ? OR snapshot_id LIKE ?", (date_str, f"%{date_str.replace('.', '')}%")) cnt = cursor.rowcount conn.commit() return cnt ``` ## File: `./core/repositories/zup_repo.py` ```py """ =============================================================================== FILE: core/repositories/zup_repo.py ROLE: Репозиторий кадровых данных 1С:ЗУП, аномалий и базы знаний. =============================================================================== """ import pandas as pd from typing import List, Dict, Any from core.connection import get_connection def save_staff_to_db(df_staff: pd.DataFrame, date_str: str) -> None: if df_staff is None or df_staff.empty: return data = [ (date_str, r.get('ФИО', ''), r.get('fio_clean', ''), r.get('Подразделение', ''), r.get('Должность', '')) for _, r in df_staff.iterrows() ] with get_connection() as conn: cursor = conn.cursor() cursor.execute("DELETE FROM zup_staff WHERE snapshot_date = ?", (date_str,)) cursor.executemany("INSERT INTO zup_staff (snapshot_date, fio, fio_clean, department, position) VALUES (?, ?, ?, ?, ?)", data) conn.commit() def load_staff_from_db(date_str: str) -> pd.DataFrame: with get_connection() as conn: df = pd.read_sql_query( "SELECT fio as 'ФИО', fio_clean, department as 'Подразделение', position as 'Должность' FROM zup_staff WHERE snapshot_date = ?", conn, params=(date_str,) ) return df if not df.empty else None def save_absences_to_db(df_absent: pd.DataFrame, date_str: str) -> None: if df_absent is None or df_absent.empty: return data = [ (date_str, r.get('ФИО', r.get('fio_clean', '')), r.get('fio_clean', ''), r.get('Вид_отсутствия', '')) for _, r in df_absent.iterrows() ] with get_connection() as conn: cursor = conn.cursor() cursor.execute("DELETE FROM zup_absences WHERE absence_date = ?", (date_str,)) cursor.executemany("INSERT INTO zup_absences (absence_date, fio, fio_clean, absence_type) VALUES (?, ?, ?, ?)", data) conn.commit() def load_absences_from_db(date_str: str) -> pd.DataFrame: with get_connection() as conn: df = pd.read_sql_query( "SELECT fio as 'ФИО', fio_clean, absence_type as 'Вид_отсутствия' FROM zup_absences WHERE absence_date = ?", conn, params=(date_str,) ) return df if not df.empty else None def save_anomalies_to_db(anomalies_list: List[Dict[str, Any]], date_str: str) -> None: if not anomalies_list: return data = [(date_str, a.get('fio', ''), a.get('type', ''), a.get('details', '')) for a in anomalies_list] with get_connection() as conn: cursor = conn.cursor() cursor.execute("DELETE FROM anomalies_history WHERE anomaly_date = ?", (date_str,)) cursor.executemany("INSERT INTO anomalies_history (anomaly_date, fio, anomaly_type, details) VALUES (?, ?, ?, ?)", data) conn.commit() def get_all_rules_from_db() -> List[str]: with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT rule_text FROM ai_knowledge_base") return [r[0] for r in cursor.fetchall()] def add_rule_to_db(rule_text: str, added_by: str = "Human") -> None: if not rule_text or not rule_text.strip(): return with get_connection() as conn: cursor = conn.cursor() cursor.execute("INSERT OR IGNORE INTO ai_knowledge_base (rule_text, added_by) VALUES (?, ?)", (rule_text.strip(), added_by)) conn.commit() def get_department_synonyms_dict() -> Dict[str, str]: with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT LOWER(short_name), LOWER(full_name) FROM department_synonyms") return {row[0]: row[1] for row in cursor.fetchall()} def add_department_synonym_to_db(short_name: str, full_name: str) -> None: with get_connection() as conn: cursor = conn.cursor() cursor.execute("INSERT OR REPLACE INTO department_synonyms (short_name, full_name) VALUES (?, ?)", (short_name.strip().lower(), full_name.strip().lower())) conn.commit() ``` ## File: `./services/data_loader.py` ```py """ =============================================================================== FILE: services/data_loader.py ROLE: Надежная загрузка штата и отсутствий (MS SQL ЗУП -> Резервный Excel). =============================================================================== """ import os import warnings from datetime import datetime # <-- ДОБАВИТЬ ЭТУ СТРОКУ import pandas as pd from config import ( normalize_fio, clean_scud_fio_light, load_exceptions, DATE_TODAY, DATE_YESTERDAY, find_dated_file, ZUP_1C_DIR, SCUD_DIR, DATA_DIR ) from services.zup_extractor import fetch_zup_absences_from_sql from core.database import ( load_scud_from_db_by_snapshot, load_staff_from_db, load_absences_from_db ) warnings.filterwarnings('ignore', category=UserWarning, module='pandas') def load_staff_data(date_str): """Загружает файл Штатного расписания 1С.""" filepath = find_dated_file("Штат", date_str) if not filepath: fallback_path = os.path.join(ZUP_1C_DIR, "штат.xlsx") if os.path.exists(fallback_path): filepath = fallback_path else: return None try: df_raw = pd.read_excel(filepath, skiprows=8) df_staff = df_raw.iloc[:, [1, 5, 12]].copy() df_staff.columns = ['ФИО', 'Подразделение', 'Должность'] df_staff = df_staff.dropna(subset=['ФИО']).reset_index(drop=True) df_staff = df_staff[~df_staff['ФИО'].astype(str).str.contains('Всего|Организация|Сотрудник|ФИО', case=False, na=False)] df_staff['fio_clean'] = df_staff['ФИО'].apply(normalize_fio) return df_staff except Exception as e: print(f"[❌] Ошибка загрузки штата из {filepath}: {e}") return None def load_absent_from_excel(date_str): """Резервное чтение файла Отсутствия_ДД_ММ_ГГГГ.xlsx.""" filepath = find_dated_file("Отсутствия", date_str) if not filepath: return None try: for skip in [0, 1, 2, 3, 4, 5, 8]: df_try = pd.read_excel(filepath, skiprows=skip) fio_col = None reason_col = None for col in df_try.columns: c_str = str(col).lower() if ('фио' in c_str or 'сотрудник' in c_str) and fio_col is None: fio_col = col if ('вид' in c_str or 'причина' in c_str or 'отсутств' in c_str) and reason_col is None: reason_col = col if fio_col and reason_col: df_res = df_try[[fio_col, reason_col]].copy() df_res.columns = ['ФИО', 'Вид_отсутствия'] df_res = df_res.dropna(subset=['ФИО', 'Вид_отсутствия']) df_res = df_res[~df_res['ФИО'].astype(str).str.contains('Всего|Организация|Сотрудник|ФИО|ЛЕНМОРНИИПРОЕКТ', case=False, na=False)] df_res['fio_clean'] = df_res['ФИО'].apply(normalize_fio) print(f" [✓] Резервный Excel: загружено {len(df_res)} записей отсутствий из {os.path.basename(filepath)}") return df_res[['fio_clean', 'Вид_отсутствия']].dropna(subset=['fio_clean']) except Exception as e: print(f" [⚠️] Ошибка чтения резервного Excel отсутствий {filepath}: {e}") return None def load_absent_data(date_str): """ Загружает отсутствия из MS SQL 1С:ЗУП, а при сбое — из локального Excel. """ df_absent = None try: df_absent = fetch_zup_absences_from_sql(date_str) if df_absent is not None and not df_absent.empty: print(f" [✓] MS SQL ЗУП: получено {len(df_absent)} записей отсутствий за {date_str}") df_absent = df_absent.dropna(subset=['ФИО', 'Вид_отсутствия']) df_absent = df_absent[~df_absent['ФИО'].astype(str).str.contains('АО "ЛЕНМОРНИИПРОЕКТ"|Сотрудник', case=False, na=False)] df_absent['fio_clean'] = df_absent['ФИО'].apply(normalize_fio) df_absent = df_absent[['fio_clean', 'Вид_отсутствия']].dropna(subset=['fio_clean']) except Exception as e: print(f" [⚠️] Ошибка подключения к MS SQL ЗУП за {date_str}: {e}") # Резервный источник: Excel файл из 1С с сетевой шары if df_absent is None or df_absent.empty: df_absent = load_absent_from_excel(date_str) if df_absent is None: df_absent = pd.DataFrame(columns=['fio_clean', 'Вид_отсутствия']) # Обогащение удаленщиками из static_reason_workers.csv с ротацией просроченных записей try: static_path = os.path.join(DATA_DIR, "static_reason_workers.csv") if os.path.exists(static_path): df_static = pd.read_csv(static_path, dtype=str, on_bad_lines='skip').fillna("") for col in ['fio', 'reason', 'department', 'date_from', 'date_to']: if col not in df_static.columns: df_static[col] = "" today_date = datetime.now().date() # Разбор целевой даты отчета clean_target_str = str(date_str).replace('_', '.') try: clean_target_date = datetime.strptime(clean_target_str, "%d.%m.%Y").date() except ValueError: clean_target_date = today_date def parse_date_safe(d_val): if not d_val or str(d_val).lower() in ['nan', 'none', '', 'nat']: return None s = str(d_val).strip().replace('_', '.') for fmt in ("%d.%m.%Y", "%Y-%m-%d"): try: return datetime.strptime(s, fmt).date() except ValueError: pass return None active_for_file = [] new_rows_for_report = [] file_changed = False existing_fios = set(df_absent['fio_clean'].dropna().tolist()) if not df_absent.empty else set() for _, s_row in df_static.iterrows(): fio_raw = str(s_row.get('fio', '')).strip() if not fio_raw: continue fio_c = normalize_fio(fio_raw) reason = str(s_row.get('reason', '')).strip() or "Дистанционная работа" d_from = parse_date_safe(s_row.get('date_from')) d_to = parse_date_safe(s_row.get('date_to')) # 1. Физическая ротация просроченных: только если текущий реальный день (today) строго больше date_to if d_to is not None and today_date > d_to: print(f" [🧹] Удаленка истекла: {fio_raw} (до {d_to.strftime('%d.%m.%Y')}). Удалена из CSV.") file_changed = True continue active_for_file.append(s_row.to_dict()) # 2. Проверка действия удаленки на дату формируемого отчета: # Если date_from не указана — действует всегда до date_to is_after_start = (d_from is None) or (clean_target_date >= d_from) is_before_end = (d_to is None) or (clean_target_date <= d_to) if is_after_start and is_before_end: if fio_c not in existing_fios: new_rows_for_report.append({ 'fio_clean': fio_c, 'Вид_отсутствия': reason }) existing_fios.add(fio_c) # Перезаписываем CSV только если реально были удалены просроченные сотрудники if file_changed: pd.DataFrame(active_for_file).to_csv(static_path, index=False, encoding='utf-8') print(" [✓] Файл static_reason_workers.csv синхронизирован без просроченных записей.") if new_rows_for_report: df_absent = pd.concat([df_absent, pd.DataFrame(new_rows_for_report)], ignore_index=True) print(f" [✓] Реестр удаленщиков: добавлено {len(new_rows_for_report)} чел. в отчет за {date_str}") except Exception as e: print(f" [⚠️] Ошибка обработки static_reason_workers.csv: {e}") return df_absent if not df_absent.empty else None def load_1c_data_smart(date_str, use_db=False): df_staff = None df_absent = None if use_db: df_staff = load_staff_from_db(date_str) df_absent = load_absences_from_db(date_str) if df_staff is None: df_staff = load_staff_data(date_str) if df_absent is None: df_absent = load_absent_data(date_str) return df_staff, df_absent ``` ## File: `./services/scud_export.py` ```py """ =============================================================================== FILE: services/scud_export.py ROLE: Прямой экспорт данных СКУД Орион (MS SQL) в SQLite и чистый Excel (XlsxWriter). Корректная фильтрация транзитных проходов турникетов парковки и двора. =============================================================================== """ import argparse import logging import os import sys import warnings from datetime import datetime, timedelta CURRENT_DIR = os.path.dirname(os.path.abspath(__file__)) ROOT_DIR = os.path.abspath(os.path.join(CURRENT_DIR, "..")) if ROOT_DIR not in sys.path: sys.path.insert(0, ROOT_DIR) import pandas as pd import pyodbc import xlsxwriter from config import SCUD_DIR, clean_scud_fio_light, load_exceptions from core.database import save_scud_to_db, has_scud_logs_for_date, has_yesterday_final_snapshot warnings.filterwarnings("ignore", message="pandas only supports SQLAlchemy connectable") SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) LOG_DIR = os.path.join(SCRIPT_DIR, "..", "logs") os.makedirs(LOG_DIR, exist_ok=True) TODAY_DATE_STR = datetime.now().strftime("%d.%m.%Y") LOG_FILE = os.path.join(LOG_DIR, f"export_{TODAY_DATE_STR}.log") logger = logging.getLogger("scud_export") logger.setLevel(logging.INFO) logger.propagate = False if not logger.handlers: try: _file_handler = logging.FileHandler(LOG_FILE, encoding="utf-8") _formatter = logging.Formatter("[%(asctime)s] [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S") _file_handler.setFormatter(_formatter) logger.addHandler(_file_handler) except (PermissionError, OSError) as e: sys.stderr.write(f"Предупреждение: невозможно создать лог-файл {LOG_FILE}: {e}\n") _console_handler = logging.StreamHandler(sys.stdout) _formatter = logging.Formatter("[%(asctime)s] [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S") _console_handler.setFormatter(_formatter) logger.addHandler(_console_handler) def log(message: str, level: str = "INFO"): level_map = { "INFO": logging.INFO, "ERROR": logging.ERROR, "WARNING": logging.WARNING, "SUCCESS": logging.INFO, } if level == "SUCCESS": logger.info(f"[SUCCESS] {message}") else: logger.log(level_map.get(level, logging.INFO), message) SERVER_NAME = r"172.16.200.147\SQL" DATABASE_NAME = "Orion-14.01.21-1" SQL_USER = "sa" SQL_PASSWORD = "123456" ODBC_DRIVER = "ODBC Driver 18 for SQL Server" SQL_QUERY_TEMPLATE = r""" DECLARE @InputDate DATE = '{target_date}'; DECLARE @TargetDate DATE = @InputDate; DECLARE @StartDate DATETIME = CAST(@TargetDate AS DATETIME); DECLARE @EndDate DATETIME = DATEADD(SECOND, -1, DATEADD(DAY, 1, @StartDate)); WITH DailyLogs AS ( SELECT log.HozOrgan AS EmployeeID, log.TimeVal, log.Event, log.Mode, CASE WHEN log.Mode = 2 OR log.Event IN (2, 27, 29, 33, 55, 65) THEN 'OUT' WHEN log.Mode = 1 OR log.Event IN (1, 21, 26, 28, 32, 54, 64) THEN 'IN' ELSE 'OTHER' END AS Direction, ROW_NUMBER() OVER (PARTITION BY log.HozOrgan ORDER BY log.TimeVal DESC) AS RowNumDesc FROM pLogData log WITH (NOLOCK) WHERE log.TimeVal BETWEEN @StartDate AND @EndDate AND log.HozOrgan IS NOT NULL AND log.HozOrgan > 0 AND log.Event IN (1, 2, 21, 26, 27, 28, 29, 32, 33, 54, 55, 64, 65) ), Passages AS ( SELECT EmployeeID, MIN(TimeVal) AS FirstRawEvent, MAX(TimeVal) AS LastRawEvent, MIN(CASE WHEN Direction = 'IN' THEN TimeVal END) AS FirstIn, -- Реальный выход: должен быть позже входа минимум на 5 минут (отсекает проезд через шлагбаум/парковку) MAX(CASE WHEN Direction = 'OUT' THEN TimeVal END) AS RawLastOut FROM DailyLogs GROUP BY EmployeeID ), EvaluatedPassages AS ( SELECT p.*, CASE WHEN p.RawLastOut IS NOT NULL AND p.FirstIn IS NOT NULL AND p.RawLastOut > DATEADD(MINUTE, 5, p.FirstIn) THEN p.RawLastOut ELSE NULL END AS FilteredLastOut FROM Passages p ) SELECT N'ЛЕНМОРНИИПРОЕКТ' AS [Фирма], ISNULL(CAST(div.Name AS NVARCHAR(255)), N'Без подразделения') AS [Подразделение], LTRIM(RTRIM( ISNULL(CAST(p.Name AS NVARCHAR(255)), N'') + CASE WHEN p.FirstName IS NOT NULL AND CAST(p.FirstName AS NVARCHAR(255)) <> '' THEN N' ' + CAST(p.FirstName AS NVARCHAR(255)) ELSE N'' END + CASE WHEN p.MidName IS NOT NULL AND CAST(p.MidName AS NVARCHAR(255)) <> '' THEN N' ' + CAST(p.MidName AS NVARCHAR(255)) ELSE N'' END )) AS [Сотрудник], ISNULL(CAST(post.Name AS NVARCHAR(255)), N'—') AS [Должность], ISNULL(CAST(p.TabNumber AS NVARCHAR(50)), N'—') AS [Таб_№], CONVERT(VARCHAR(10), @TargetDate, 104) AS [Дата], ISNULL(CAST(CONVERT(VARCHAR(8), pass.FirstIn, 108) AS NVARCHAR(20)), N'Нет входа') AS [Начало_дня], CASE WHEN pass.FirstIn IS NULL AND pass.FirstRawEvent IS NOT NULL THEN CAST(CONVERT(VARCHAR(8), pass.FirstRawEvent, 108) AS NVARCHAR(20)) ELSE N'—' END AS [Первая_активность], CASE WHEN pass.FilteredLastOut IS NOT NULL THEN CAST(CONVERT(VARCHAR(8), pass.FilteredLastOut, 108) AS NVARCHAR(20)) ELSE N'Нет выхода' END AS [Конец_дня], CASE WHEN pass.EmployeeID IS NOT NULL AND (pass.FirstIn IS NOT NULL OR pass.FirstRawEvent IS NOT NULL) THEN RIGHT('0' + CAST(DATEDIFF(MINUTE, ISNULL(pass.FirstIn, pass.FirstRawEvent), CASE WHEN pass.FilteredLastOut IS NOT NULL THEN pass.FilteredLastOut WHEN @TargetDate = CAST(GETDATE() AS DATE) THEN GETDATE() ELSE ISNULL(pass.FirstIn, pass.FirstRawEvent) END) / 60 AS VARCHAR), 2) + ':' + RIGHT('0' + CAST(DATEDIFF(MINUTE, ISNULL(pass.FirstIn, pass.FirstRawEvent), CASE WHEN pass.FilteredLastOut IS NOT NULL THEN pass.FilteredLastOut WHEN @TargetDate = CAST(GETDATE() AS DATE) THEN GETDATE() ELSE ISNULL(pass.FirstIn, pass.FirstRawEvent) END) % 60 AS VARCHAR), 2) ELSE N'00:00' END AS [Находился_в_здании], CASE WHEN pass.EmployeeID IS NOT NULL THEN N'Присутствовал' ELSE N'Отсутствовал (Нет событий)' END AS [Статус] FROM pList p WITH (NOLOCK) LEFT JOIN PDivision div WITH (NOLOCK) ON p.Section = div.ID LEFT JOIN PPost post WITH (NOLOCK) ON p.Post = post.ID LEFT JOIN EvaluatedPassages pass ON p.ID = pass.EmployeeID WHERE ISNULL(p.StatusRecord, 0) = 0 AND p.DateTimeInArchive IS NULL AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT LIKE N'Аренд%' AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT IN (N'Без подразделения', N'') AND p.Name NOT LIKE N'бр.%' AND p.Name NOT LIKE N'Гость%' AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT IN (N'БГИ', N'КНР') AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT LIKE N'Рабоч%' AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT LIKE N'Врем%' AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT LIKE N'Практика%' AND ISNULL(CAST(div.Name AS NVARCHAR(255)), N'') NOT LIKE N'тест%' AND ISNULL(CAST(post.Name AS NVARCHAR(255)), N'') NOT LIKE N'Практикант%' ORDER BY p.Name ASC; """ def save_df_to_clean_excel(df: pd.DataFrame, file_path: str, sheet_name: str = "Отчет"): workbook = xlsxwriter.Workbook(file_path, {'constant_memory': False}) worksheet = workbook.add_worksheet(sheet_name) fmt_header = workbook.add_format({ 'bold': True, 'bg_color': '#D9E1F2', 'border': 1, 'border_color': '#D3D3D3', 'align': 'center', 'valign': 'vcenter', 'font_name': 'Calibri', 'font_size': 11 }) fmt_cell = workbook.add_format({ 'border': 1, 'border_color': '#D3D3D3', 'valign': 'vcenter', 'align': 'left', 'font_name': 'Calibri', 'font_size': 11 }) headers = list(df.columns) col_widths = [len(str(h)) for h in headers] for col_idx, header in enumerate(headers): worksheet.write(0, col_idx, str(header), fmt_header) for row_idx, row_values in enumerate(df.values, start=1): for col_idx, val in enumerate(row_values): if pd.isna(val) or val is None: val_str = "" elif isinstance(val, bool): val_str = "Да" if val else "Нет" else: val_str = str(val) worksheet.write(row_idx, col_idx, val_str, fmt_cell) if len(val_str) > col_widths[col_idx]: col_widths[col_idx] = len(val_str) for col_idx, width in enumerate(col_widths): worksheet.set_column(col_idx, col_idx, min(max(width + 3, 10), 45)) workbook.close() def get_targets(input_date: str | None): targets = [] if input_date: try: parsed = datetime.strptime(input_date, "%d.%m.%Y").date() targets.append({"name": "Указанная дата", "date": parsed}) except ValueError: log(f"ОШИБКА: Неверный формат даты '{input_date}'. Используйте ДД.ММ.ГГГГ", "ERROR") sys.exit(1) else: now = datetime.now() yesterday = (now - timedelta(days=3 if now.weekday() == 0 else 1)).date() today = now.date() targets.append({"name": "Вчера", "date": yesterday}) targets.append({"name": "Сегодня", "date": today}) return targets def run_export(input_date: str | None = None, save_xlsx: bool = True, debug: bool = False): if debug: logger.setLevel(logging.DEBUG) log("=== ВКЛЮЧЕН РЕЖИМ ОТЛАДКИ (DEBUG MODE) ===", "WARNING") log("=== [ЭТАП 0] Выгрузка свежих данных СКУД напрямую из БД Орион ===") os.makedirs(SCUD_DIR, exist_ok=True) targets = get_targets(input_date) conn_str = ( f"DRIVER={{{ODBC_DRIVER}}};" f"SERVER={SERVER_NAME};" f"DATABASE={DATABASE_NAME};" f"UID={SQL_USER};" f"PWD={SQL_PASSWORD};" f"TrustServerCertificate=yes;" f"Encrypt=no;" ) success = True for target in targets: processing_date = target["date"] processing_date_str = processing_date.strftime("%d.%m.%Y") period_label = target["name"] is_yesterday = (period_label == "Вчера") if is_yesterday and has_yesterday_final_snapshot(processing_date_str): log(f"[ℹ️] Вчерашний день ({processing_date_str}) уже зафиксирован финишным снапшотом Y. Пропускаем запрос к MS SQL.") continue if is_yesterday: snapshot_time = f"{processing_date.strftime('%Y-%m-%d')} 23:59:59" else: snapshot_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") log(f"--- Обработка периода: {period_label} ({processing_date_str}) --- [Снапшот: {snapshot_time}]") sql_query = SQL_QUERY_TEMPLATE.format(target_date=processing_date.strftime("%Y-%m-%d")) connection = None try: connection = pyodbc.connect(conn_str, timeout=120) df = pd.read_sql(sql_query, connection) log(f"База вернула {len(df)} строк за {processing_date_str}") if len(df) > 0: df['fio_clean'] = df['Сотрудник'].apply(clean_scud_fio_light) df['anomaly_flag'] = 'NONE' mask_anomaly = (df['Начало_дня'] == 'Нет входа') & (df['Первая_активность'] != '—') df.loc[mask_anomaly, 'anomaly_flag'] = 'ANOMALY_NO_IN_HAS_ACTIVITY' df['Пришел'] = df['Статус'].str.contains('Присутствовал', case=False, na=False) & (~mask_anomaly) save_scud_to_db(df, processing_date_str, snapshot_time=snapshot_time, is_yesterday=is_yesterday) log(f"[✓] Записи за {processing_date_str} успешно сохранены в SQLite!", "SUCCESS") if save_xlsx: file_name = f"Сотрудники_{processing_date_str}.xlsx" file_path = os.path.join(SCUD_DIR, file_name) if os.path.exists(file_path): try: os.remove(file_path) except OSError as e: log(f"ОШИБКА при удалении старого файла {file_name}: {e}", "ERROR") save_df_to_clean_excel(df, file_path, sheet_name="Отчет") log(f"[✓] Успешно экспортирован файл: data/scud/{file_name}", "SUCCESS") else: log(f"Запрос за {processing_date_str} вернул 0 строк.", "WARNING") except Exception as e: log(f"🛑 ОШИБКА выгрузки СКУД за {processing_date_str}: {e}", "ERROR") success = False finally: if connection is not None: connection.close() log("=== Выгрузка СКУД завершена ===\n") return success if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--date", dest="input_date", default=None) parser.add_argument("-d", "--debug", action="store_true") parser.add_argument("--no-xlsx", dest="save_xlsx", action="store_false", default=True) args = parser.parse_args() run_export(args.input_date, save_xlsx=args.save_xlsx, debug=args.debug) ``` ## File: `./services/share_copier.py` ```py """ =============================================================================== FILE: services/share_copier.py ROLE: Синхронизация файлов 1С (Штат и Отсутствия) с сетевой шары в data/1c/. =============================================================================== """ import os import shutil from config import DATE_TODAY, DATE_YESTERDAY, ZUP_1C_DIR, SHARE_1C_DIR def find_file_strictly_on_share(prefix, date_str): """ Ищет файл со строгой привязкой ТОЛЬКО к сетевой шаре SHARE_1C_DIR, не обращаясь к локальным папкам. """ if not os.path.exists(SHARE_1C_DIR): return None date_dots = date_str date_underscores = date_str.replace('.', '_') try: for f in os.listdir(SHARE_1C_DIR): if f.endswith('.xlsx') or f.endswith('.csv'): if f.lower().startswith(prefix.lower()): if date_dots in f or date_underscores in f: return os.path.join(SHARE_1C_DIR, f) except Exception as e: print(f" [⚠️] Ошибка чтения сетевой шары: {e}") return None def copy_1c_files_from_share(): r""" Копирует свежие файлы 1С (Штат и Отсутствия) за Сегодня и Вчера из сетевой шары \\storage\SCUD\Обмен\Штат в локальную папку data/1c/ """ print(f"[0.5/5] Проверка и копирование файлов 1С с шары: {SHARE_1C_DIR}...") if not os.path.exists(SHARE_1C_DIR): print(f"[⚠️] Сетевой шар недоступен или путь не найден: {SHARE_1C_DIR}") print(" Используем ранее сохраненные локальные файлы из data/1c/\n") return False dates_to_copy = [DATE_TODAY, DATE_YESTERDAY] prefixes = ["Штат", "Отсутствия"] copied_count = 0 os.makedirs(ZUP_1C_DIR, exist_ok=True) for d_str in dates_to_copy: for prefix in prefixes: remote_file = find_file_strictly_on_share(prefix, d_str) if remote_file and os.path.exists(remote_file): filename = os.path.basename(remote_file) local_target_path = os.path.join(ZUP_1C_DIR, filename) try: # shutil.copyfile копирует только содержимое потока байтов # без попыток изменить POSIX-права/атрибуты (chmod) на CIFS/SMB шаре shutil.copyfile(remote_file, local_target_path) print(f" [✓] Успешно скопирован с шары: {filename} -> data/1c/") copied_count += 1 except Exception as e: print(f" [⚠️] Ошибка копирования {filename}: {e}") else: d_fmt = d_str.replace('.', '_') print(f" [ℹ️] На сетевой шаре отсутствует {prefix} за {d_str} ({prefix}_{d_fmt}.xlsx)") if copied_count > 0: print(f"[✓] Успешно скопировано файлов с сетевой шары: {copied_count} шт.\n") else: print("[ℹ️] Новых файлов за указанные даты на сетевой шаре не обнаружено.\n") return True ``` ## File: `./services/excel_exporter.py` ```py """ =============================================================================== FILE: services/excel_exporter.py ROLE: Генерация Excel-отчетов (Сводка, Детальный отчет, Сырой СКУД) через XlsxWriter. Корректный расчет часов удаленщиков и исключение лишних списков. =============================================================================== """ import os import math import time import pandas as pd import xlsxwriter from datetime import datetime, timedelta from xlsxwriter.exceptions import FileCreateError from config import REPORTS_DIR MONTHS_RU_GENITIVE = { 1: "января", 2: "февраля", 3: "марта", 4: "апреля", 5: "мая", 6: "июня", 7: "июля", 8: "августа", 9: "сентября", 10: "октября", 11: "ноября", 12: "декабря" } MONTHS_RU_NOMINATIVE = { 1: "январь", 2: "февраль", 3: "март", 4: "апрель", 5: "май", 6: "июнь", 7: "июль", 8: "август", 9: "сентябрь", 10: "октябрь", 11: "ноябрь", 12: "декабрь" } def format_date_ru(date_str): date_clean = str(date_str).replace('_', '.') try: dt = datetime.strptime(date_clean, "%d.%m.%Y") return f"{dt.day} {MONTHS_RU_GENITIVE[dt.month]} {dt.year}" except Exception: return date_str def get_dated_reports_dir(date_str): date_clean = str(date_str).replace('_', '.') try: dt = datetime.strptime(date_clean, "%d.%m.%Y") year_str = str(dt.year) month_name = MONTHS_RU_NOMINATIVE[dt.month] except Exception: now = datetime.now() year_str = str(now.year) month_name = MONTHS_RU_NOMINATIVE[now.month] target_dir = os.path.join(REPORTS_DIR, year_str, month_name) os.makedirs(target_dir, exist_ok=True) return target_dir def safe_close_workbook(wb, output_path, target_dir, filename): try: wb.close() print(f"[✓] Успешно сохранен: {output_path}") return output_path except (FileCreateError, OSError, PermissionError): alt_filename = filename.replace(".xlsx", f"_{int(time.time())}.xlsx") alt_path = os.path.join(target_dir, alt_filename) try: wb.filename = alt_path wb._store_workbook() print(f"[⚠️] Исходный файл открыт в Excel! Сохранено как: {alt_path}") return alt_path except Exception as e: print(f"[❌] Ошибка сохранения даже резервного файла: {e}") return output_path def calculate_autoclose_time(time_in_str: str) -> tuple[str, str, str]: try: parts = time_in_str.strip().split(':') hh = int(parts[0]) mm = int(parts[1]) if len(parts) > 1 else 0 ss = int(parts[2]) if len(parts) > 2 else 0 dt_in = datetime(2000, 1, 1, hh, mm, ss) dt_out = dt_in + timedelta(hours=8, minutes=30) return dt_out.strftime("%H:%M:%S"), "08:30", "0:00" except Exception: return "17:00:00", "08:30", "0:00" def calculate_deviation(time_in_building_str, reason="", norm_hours=8, lunch_minutes=30): """ Расчет отклонения от нормы. Для удаленщиков при наличии физического времени в здании вычисляется реальное отклонение. """ reason_clean = str(reason).strip().lower() if pd.notna(reason) else "" is_remote = "удален" in reason_clean or "дистанцион" in reason_clean has_building_time = isinstance(time_in_building_str, str) and time_in_building_str not in ['00:00', '0', '', 'None', 'nan', 'NaN'] # Если есть уважительная причина (больничный, отпуск, командировка и т.д.) не удаленка if reason_clean != "" and not is_remote: return "0:00" # Если удаленщик работал исключительно из дома (00:00 в здании) if is_remote and not has_building_time: return "0:00" # Если сотрудника не было в здании и нет уважительной причины if not has_building_time: return f"-{norm_hours}:00" try: parts = time_in_building_str.strip().split(':') hh = int(parts[0]) mm = int(parts[1]) if len(parts) > 1 else 0 total_in_building_minutes = hh * 60 + mm if total_in_building_minutes == 0: return "0:00" if is_remote else f"-{norm_hours}:00" work_minutes = max(0, total_in_building_minutes - lunch_minutes) norm_minutes = norm_hours * 60 diff = work_minutes - norm_minutes if diff == 0: return "0:00" sign = "-" if diff < 0 else "" abs_diff = abs(diff) res_hh = abs_diff // 60 res_mm = abs_diff % 60 return f"{sign}{res_hh}:{res_mm:02d}" except Exception: return "0:00" if is_remote else f"-{norm_hours}:00" # ============================================================================= # 1. ЕЖЕДНЕВНАЯ СВОДКА НА СЕГОДНЯ # ============================================================================= def generate_summary_excel(merged_df, date_str="21.08.2026", filename=None): date_clean = str(date_str).replace('_', '.') if not filename: filename = f"{format_date_ru(date_clean)} сводка.xlsx" target_dir = get_dated_reports_dir(date_clean) output_path = os.path.join(target_dir, filename) wb = xlsxwriter.Workbook(output_path) ws = wb.add_worksheet("Лист_1") ws.outline_settings(visible=True, symbols_below=False, symbols_right=False, auto_style=False) def make_fmt(bg_color=None, bold=False, align="left", wrap=False): d = { 'font_name': 'Calibri', 'font_size': 11, 'bold': bold, 'align': align, 'valign': 'vcenter', 'border': 1, 'border_color': '#D3D3D3', 'text_wrap': wrap } if bg_color: d['bg_color'] = bg_color return wb.add_format(d) fmt_hdr_l = make_fmt(bg_color='#D9E1F2', bold=True, align="left") fmt_hdr_r = make_fmt(bg_color='#D9E1F2', bold=True, align="right") fmt_tot_l = make_fmt(bg_color='#F2F2F2', bold=True, align="left") fmt_tot_r = make_fmt(bg_color='#F2F2F2', bold=True, align="right") fmt_empty = make_fmt() ws.set_row(0, 20) ws.write(0, 0, "Сводка на", fmt_hdr_l) ws.write(0, 1, date_clean, fmt_hdr_r) ws.set_row(1, 20) ws.write(1, 0, "", fmt_empty) ws.write(1, 1, "", fmt_empty) ws.set_row(2, 20) ws.write(2, 0, "По списку", fmt_tot_l) ws.write(2, 1, len(merged_df), 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[ (merged_df['Пришел'] == False) & (merged_df['Вид_отсутствия'].isna() | (merged_df['Вид_отсутствия'].astype(str).str.strip() == '')) & (~is_no_pass) & (~is_exc) ] fmt_unexp_hl = make_fmt(bg_color='#FCE4D6', bold=True, align="left") fmt_unexp_hr = make_fmt(bg_color='#FCE4D6', bold=True, align="right") fmt_unexp_rl = make_fmt(bg_color='#FCE4D6', bold=False, align="left") fmt_unexp_rr = make_fmt(bg_color='#FCE4D6', bold=False, align="right") ws.set_row(current_row, 20) ws.write(current_row, 0, "неизвестно", fmt_unexp_hl) ws.write(current_row, 1, len(unexplained), fmt_unexp_hr) current_row += 1 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, 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() fmt_np_hl = make_fmt(bg_color='#E1F5FE', bold=True, align="left") fmt_np_hr = make_fmt(bg_color='#E1F5FE', bold=True, align="right") fmt_np_rl = make_fmt(bg_color='#E1F5FE', bold=False, align="left") fmt_np_rr = make_fmt(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, 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, 1, "", fmt_np_rr) current_row += 1 # 3. Официальные отсутствия (Сотрудники из исключений при наличии документа 1С попадают сюда) reason_clean = merged_df['Вид_отсутствия'].astype(str).str.lower() is_remote_reason = reason_clean.str.contains('удален|дистанцион', regex=True, na=False) absent_only = merged_df[ (merged_df['Пришел'] == False) & (merged_df['Вид_отсутствия'].notna()) & (~merged_df['Вид_отсутствия'].astype(str).str.startswith('Исключение')) & (~is_remote_reason) ] absent_groups = absent_only.groupby('Вид_отсутствия') pastels = ['#FFF2CC', '#E1D5E7', '#E1F5FE', '#FFF0F5', '#FCF3CF'] for idx_cat, (cat_name, group) in enumerate(absent_groups): hex_c = pastels[idx_cat % len(pastels)] fmt_cat_hl = make_fmt(bg_color=hex_c, bold=True, align="left") fmt_cat_hr = make_fmt(bg_color=hex_c, bold=True, align="right") fmt_cat_rl = make_fmt(bg_color=hex_c, bold=False, align="left") fmt_cat_rr = make_fmt(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, 1, len(group), fmt_cat_hr) current_row += 1 for fio in sorted(group['Сотрудник'].dropna().unique()): 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, "", 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)] total_present_count = len(present_scud) + len(exc_without_doc) fmt_pres_hl = make_fmt(bg_color='#E2EFDA', bold=True, align="left") fmt_pres_hr = make_fmt(bg_color='#E2EFDA', bold=True, align="right") ws.set_row(current_row, 20) ws.write(current_row, 0, "Итого на работе", fmt_pres_hl) 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)] fmt_rem_hl = make_fmt(bg_color='#E8F8F5', bold=True, align="left") fmt_rem_hr = make_fmt(bg_color='#E8F8F5', bold=True, align="right") fmt_rem_rl = make_fmt(bg_color='#E8F8F5', bold=False, align="left") fmt_rem_rr = make_fmt(bg_color='#E8F8F5', bold=False, align="right") ws.set_row(current_row, 20) ws.write(current_row, 0, "В том числе на удаленной работе", fmt_rem_hl) ws.write(current_row, 1, len(remote_home), fmt_rem_hr) current_row += 1 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, 1, "", fmt_rem_rr) current_row += 1 # 6. Аномалии СКУД и 1С (Свернуто) anomalies = merged_df[ (~is_exc) & ( ((merged_df['Пришел'] == True) & (merged_df['Вид_отсутствия'].notna()) & (~merged_df['Вид_отсутствия'].astype(str).str.startswith('Исключение')) & (~is_remote_reason) & (~reason_clean.str.contains('командировк', na=False))) | (merged_df.get('anomaly_flag', 'NONE') == 'ANOMALY_NO_IN_HAS_ACTIVITY') ) ] fmt_anom_hl = make_fmt(bg_color='#FCE4D6', bold=True, align="left") fmt_anom_hr = make_fmt(bg_color='#FCE4D6', bold=True, align="right") fmt_anom_rl = make_fmt(bg_color='#FCE4D6', bold=False, align="left") fmt_anom_rr = make_fmt(bg_color='#FCE4D6', bold=False, align="left", wrap=True) ws.set_row(current_row, 20) ws.write(current_row, 0, "Аномалии СКУД и 1С", fmt_anom_hl) ws.write(current_row, 1, len(anomalies), fmt_anom_hr) current_row += 1 chars_per_line_b = 30 if not anomalies.empty: for _, row in anomalies.iterrows(): fio = row.get('Сотрудник', '') reason = row.get('Вид_отсутствия', '') anom_flag = row.get('anomaly_flag', 'NONE') if anom_flag == 'ANOMALY_NO_IN_HAS_ACTIVITY': first_act = row.get('Первая_активность', '—') reason_text = f"🚨 АНОМАЛИЯ СКУД: Нет входа (первая активность: {first_act})" else: reason_text = f"В 1С: {reason}" lines_count = math.ceil(len(reason_text) / chars_per_line_b) if len(reason_text) > chars_per_line_b else 1 row_h = max(lines_count * 18, 20) ws.set_row(current_row, row_h, 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) current_row += 1 ws.set_column(0, 0, 45) ws.set_column(1, 1, 38) safe_close_workbook(wb, output_path, target_dir, filename) # ============================================================================= # 2. ДЕТАЛЬНЫЙ СУТОЧНЫЙ ОТЧЕТ ЗА ВЧЕРА # ============================================================================= def generate_detailed_excel(merged_df, date_str="20.08.2026", filename=None): date_clean = str(date_str).replace('_', '.') if not filename: filename = f"{format_date_ru(date_clean)} отчет.xlsx" if merged_df is not None and not merged_df.empty: df_export = merged_df[merged_df.get('is_excluded', False) == False].copy() else: df_export = pd.DataFrame() target_dir = get_dated_reports_dir(date_clean) output_path = os.path.join(target_dir, filename) wb = xlsxwriter.Workbook(output_path) ws = wb.add_worksheet("Детальный_отчет") def make_fmt(bg_color=None, bold=False, align="left", wrap=False): d = { 'font_name': 'Arial', 'font_size': 10, 'bold': bold, 'align': align, 'valign': 'vcenter', 'border': 1, 'border_color': '#D3D3D3', 'text_wrap': wrap } if bg_color: d['bg_color'] = bg_color return wb.add_format(d) fmt_date_lbl = wb.add_format({'font_name': 'Arial', 'font_size': 10, 'bold': True}) ws.write(1, 1, "Дата:", fmt_date_lbl) ws.write(1, 3, date_clean, fmt_date_lbl) headers = [ "№", "ФИО", "Подразделение", "время входа", "первая активность", "время выхода", "находился в здании", "причина отсутствия", "норма", "отклонение от нормы" ] fmt_hdr = make_fmt(bg_color='#D9E1F2', bold=True, align="center", wrap=True) ws.set_row(3, 26) for col_idx, h_text in enumerate(headers): ws.write(3, col_idx, h_text, fmt_hdr) start_col = 'Начало дня' if 'Начало дня' in df_export.columns else 'Начало_дня' end_col = 'Конец дня' if 'Конец дня' in df_export.columns else 'Конец_дня' hours_col = 'Часы' if 'Часы' in df_export.columns else 'Находился_в_здании' chars_per_line_h = 24 for idx, row in df_export.reset_index(drop=True).iterrows(): row_num = 4 + idx is_present = row.get('Пришел', False) absence_reason = row.get('Вид_отсутствия', '') has_reason = pd.notna(absence_reason) and str(absence_reason).strip() != '' in_val = str(row.get(start_col, 'Нет входа')).strip() out_val = str(row.get(end_col, 'Нет выхода')).strip() in_building_str = str(row.get(hours_col, '00:00')) first_act_val = str(row.get('Первая_активность', '—')).strip() has_first_act = first_act_val not in ['—', '', 'None', 'nan'] if in_val not in ['Нет входа', '—', '', 'nan', 'None'] and out_val in ['Нет выхода', '—', '', 'nan', 'None'] and not has_reason: out_val, in_building_str, deviation_val = calculate_autoclose_time(in_val) else: 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('Подразделение', ''))) row_color = None if is_present and has_reason: row_color = '#E2EFDA' elif not is_present and has_reason: row_color = '#FFF2CC' 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 "" 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), (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), (8, 'center', False), (deviation_val, 'center', False) ] for col_idx, (val, align_type, is_wrap) in enumerate(values): fmt = make_fmt(bg_color=row_color, align=align_type, wrap=is_wrap) ws.write(row_num, col_idx, val, fmt) col_widths = { 0: 4, 1: 33, 2: 13, 3: 11, 4: 11, 5: 11, 6: 12, 7: 24, 8: 6, 9: 11 } for col_idx, width in col_widths.items(): ws.set_column(col_idx, col_idx, width) safe_close_workbook(wb, output_path, target_dir, filename) # ============================================================================= # 3. СЫРОЙ СКУД # ============================================================================= def export_raw_scud(df_scud, filename="СКУД_Сырые_данные.xlsx"): output_path = os.path.join(REPORTS_DIR, filename) target_dir = os.path.dirname(output_path) wb = xlsxwriter.Workbook(output_path) ws = wb.add_worksheet("Сырые_данные") fmt_hdr = wb.add_format({ 'font_name': 'Calibri', 'font_size': 11, 'bold': True, 'bg_color': '#D9E1F2', 'border': 1, 'border_color': '#D3D3D3', 'align': 'center', 'valign': 'vcenter' }) fmt_cell = wb.add_format({ 'font_name': 'Calibri', 'font_size': 11, 'border': 1, 'border_color': '#D3D3D3', 'valign': 'vcenter', 'align': 'left' }) headers = list(df_scud.columns) ws.set_row(3, 28) for col_idx, header in enumerate(headers): ws.write(0, col_idx, str(header), fmt_hdr) col_widths = [len(str(h)) for h in headers] for row_idx, row_values in enumerate(df_scud.values, start=1): ws.set_row(row_idx, 19) for col_idx, val in enumerate(row_values): if pd.isna(val) or val is None: val_str = "" elif isinstance(val, bool): val_str = "Да" if val else "Нет" else: val_str = str(val) ws.write(row_idx, col_idx, val_str, fmt_cell) if len(val_str) > col_widths[col_idx]: col_widths[col_idx] = len(val_str) for col_idx, width in enumerate(col_widths): ws.set_column(col_idx, col_idx, min(max(width + 3, 10), 45)) safe_close_workbook(wb, output_path, target_dir, filename) ``` ## File: `./services/text_reporter.py` ```py import json import difflib import re from services.ai_verifier import ask_ollama from services.knowledge_base import load_knowledge_base def find_ai_identity_suggestions(unexplained_df, raw_staff_df): """ ⭐️ Теневой ИИ-арбитраж: ищет потенциальные опечатки операторов СКУД среди нераспознанных сотрудников и формирует рекомендации для администратора. """ if unexplained_df.empty or raw_staff_df is None or raw_staff_df.empty: return [] staff_fios = raw_staff_df['fio_clean'].dropna().unique().tolist() if 'fio_clean' in raw_staff_df.columns else [] unexp_fios = unexplained_df['fio_clean'].dropna().unique().tolist() if 'fio_clean' in unexplained_df.columns else [] suggestions = [] for scud_fio in unexp_fios: f_parts = scud_fio.split() if not f_parts: continue surname = f_parts[0].lower() for staff_fio in staff_fios: if staff_fio.lower().startswith(surname[:4]): ratio = difflib.SequenceMatcher(None, scud_fio.lower(), staff_fio.lower()).ratio() if 0.75 <= ratio < 1.0: suggestions.append({ "scud_fio": scud_fio, "suggested_zup_fio": staff_fio, "confidence": round(ratio, 2) }) break return suggestions def generate_markdown_report(merged_df, absent_explained, absent_unexplained, scud_present_but_absent_in_1c, anomalies_list=None, raw_scud_df=None, raw_staff_df=None, raw_absent_df=None, date_str="05.08.2026"): kb = load_knowledge_base() custom_rules = kb.get("rules", []) custom_rules_str = "\n".join([f"- {r}" for r in custom_rules]) if custom_rules else "Специфических правил компании пока нет." total_staff = len(merged_df) present_office_cnt = len(merged_df[merged_df['Пришел'] == True]) reason_series = merged_df['Вид_отсутствия'].astype(str).str.lower() is_remote_reason = reason_series.str.contains('удален|дистанцион', regex=True, na=False) remote_home_cnt = len(merged_df[(merged_df['Пришел'] == False) & is_remote_reason]) explained_cnt = len(merged_df[(merged_df['Пришел'] == False) & (merged_df['Вид_отсутствия'].notna()) & (~is_remote_reason) & (~merged_df['Вид_отсутствия'].astype(str).str.startswith('Исключение'))]) unexplained_cnt = len(absent_unexplained) anomalies_cnt = len(anomalies_list) if anomalies_list else 0 # ⭐️ Теневые подсказки ИИ ai_suggestions = find_ai_identity_suggestions(absent_unexplained, raw_staff_df) anomalies_formatted = [] if anomalies_list: for idx, a in enumerate(anomalies_list, 1): anomalies_formatted.append(f"{idx}. {a.get('fio', 'Сотрудник')}: {a.get('description', '')}") anomalies_text_block = "\n".join(anomalies_formatted) if anomalies_formatted else "Аномалий не обнаружено." unexplained_list = [] if not absent_unexplained.empty: for idx, r in absent_unexplained.iterrows(): fio = r.get('Сотрудник', r.get('fio_clean', '')) dept_scud = str(r.get('department_scud', '')).strip() dept_1c = str(r.get('Подразделение', '')).strip() dept = dept_scud if (dept_scud and dept_scud.lower() != 'nan' and dept_scud != '—') else dept_1c if not dept: dept = '—' pos = r.get('Должность', '—') unexplained_list.append(f"{fio} — {dept}, {pos}") prompt = f""" Ты — старший аудитор кадровой безопасности и контроллинга СКУД. Сформируй итоговую сводку кадрового контроля на {date_str}. 📌 БАЗА ЗНАНИЙ И ПРАВИЛА ПРЕДПРИЯТИЯ: {custom_rules_str} 🚨 ВХОДНЫЕ МЕТРИКИ: - Всего сотрудников: {total_staff} - Итого на работе (в офисе): {present_office_cnt} - В том числе на удаленной работе: {remote_home_cnt} - Официально отсутствуют: {explained_cnt} - Неизвестно (истинно неотмеченные): {unexplained_cnt} - Выявлено аномалий/конфликтов реестров: {anomalies_cnt} --- 🚨 ПОДТВЕРЖДЁННЫЕ АНОМАЛИИ ({anomalies_cnt} шт): {anomalies_text_block} --- 💡 ПРЕДЛОЖЕНИЯ ИИ ПО СОПОСТАВЛЕНИЮ ФИО ({len(ai_suggestions)} шт): {json.dumps(ai_suggestions, ensure_ascii=False, indent=2)} --- 📊 СПИСОК НЕИЗВЕСТНЫХ СЛУЧАЕВ ({unexplained_cnt} чел): {json.dumps(unexplained_list, ensure_ascii=False, indent=2)} СТРОГИЕ ИНСТРУКЦИИ: 1. КАТЕГОРИЧЕСКИ ЗАПРЕЩЕНО выводить JSON! В блоке "Выявленные ИИ аномалии" выведи читаемый нумерованный список строк ровно так, как передано в разделе "ПОДТВЕРЖДЁННЫЕ АНОМАЛИИ". 2. В разделе "Подозрения на ошибки сопоставления ФИО" выведи рекомендации по сопоставлению из массива "ПРЕДЛОЖЕНИЯ ИИ". Если массив пуст — напиши "Ошибок сопоставления ФИО не обнаружено." 3. В разделе "Неизвестные случаи" выведи нумерованный список всех {unexplained_cnt} человек. 4. В разделе "Рекомендации" дай 2-3 системные рекомендации. СТРОГИЙ ШАБЛОН ОТВЕТА: **Сводка контроллинга СКУД и 1С:ЗУП на {date_str}** - Всего офисных сотрудников: **{total_staff}** - Итого на работе (в офисе): **{present_office_cnt}** - В том числе на удаленной работе: **{remote_home_cnt}** - Официально отсутствуют: **{explained_cnt}** - Неизвестно (истинно неотмеченные): **{unexplained_cnt}** чел. - Выявлено аномалий/конфликтов реестров: **{anomalies_cnt}** шт. #### 🚨 Выявленные ИИ аномалии и конфликты источников ({anomalies_cnt}): (Список аномалий или 'Аномалий не обнаружено.') #### 💡 Подозрения на ошибки сопоставления ФИО и предложения ИИ: (Список предложений сопоставления или 'Ошибок сопоставления ФИО не обнаружено.') #### Неизвестные случаи: {unexplained_cnt} (Нумерованный список всех {unexplained_cnt} человек) ### Точечные рекомендации: (2-3 системные рекомендации) """ sys_prompt = "Ты — русскоязычный кадровый аудитор. Пиши СТРОГО на русском языке форматированным текстом. Запрещено выводить фигурные скобки JSON." return ask_ollama(prompt, system_prompt=sys_prompt) ``` ## File: `./services/exceptions_repo.py` ```py """ =============================================================================== FILE: services/exceptions_repo.py ROLE: Управление исключениями в SQLite с синхронизацией с exceptions.json. =============================================================================== """ from typing import Dict, List, Any import json import os from config import EXCEPTIONS_PATH, normalize_fio from core.connection import get_connection def init_exceptions_table(): with get_connection() as conn: conn.execute(""" CREATE TABLE IF NOT EXISTS exceptions_registry ( id INTEGER PRIMARY KEY AUTOINCREMENT, category TEXT NOT NULL, value TEXT NOT NULL, comment TEXT DEFAULT '', created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, UNIQUE(category, value) ); """) conn.commit() def get_all_exceptions_from_db() -> Dict[str, List[str]]: init_exceptions_table() cfg = {"departments": [], "positions": [], "fio": [], "position_keywords": [], "include_fio": []} with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT category, value FROM exceptions_registry") rows = cursor.fetchall() if not rows and os.path.exists(EXCEPTIONS_PATH): # Первичная миграция из JSON в SQLite sync_json_to_db() return get_all_exceptions_from_db() for cat, val in rows: if cat in cfg: cfg[cat].append(val) return cfg def add_exception_to_db(category: str, value: str, comment: str = "") -> bool: init_exceptions_table() val_clean = normalize_fio(value) if category in ["fio", "include_fio"] else value.strip() if not val_clean: return False with get_connection() as conn: conn.execute( "INSERT OR REPLACE INTO exceptions_registry (category, value, comment) VALUES (?, ?, ?)", (category, val_clean, comment) ) conn.commit() return True def remove_exception_from_db(category: str, value: str) -> bool: init_exceptions_table() with get_connection() as conn: cursor = conn.cursor() cursor.execute("DELETE FROM exceptions_registry WHERE category = ? AND value = ?", (category, value.strip())) conn.commit() return cursor.rowcount > 0 def sync_json_to_db(): """Переносит данные из exceptions.json в SQLite.""" if not os.path.exists(EXCEPTIONS_PATH): return try: with open(EXCEPTIONS_PATH, "r", encoding="utf-8") as f: data = json.load(f) for cat, items in data.items(): for item in items: add_exception_to_db(cat, item, comment="Импорт из JSON") except Exception as e: print(f"[⚠️] Ошибка синхронизации JSON -> DB: {e}") ``` ## File: `./services/zup_extractor.py` ```py import os import logging from datetime import date, datetime import pyodbc import pandas as pd from config import DATA_DIR, normalize_fio, ZUP_SQL_CONFIG logger = logging.getLogger(__name__) def get_zup_connection_string() -> str: """Формирует строку подключения pyodbc к MS SQL Server из config.py.""" return ( f"DRIVER={ZUP_SQL_CONFIG['driver']};" f"SERVER={ZUP_SQL_CONFIG['server']};" f"DATABASE={ZUP_SQL_CONFIG['database']};" f"UID={ZUP_SQL_CONFIG['user']};" f"PWD={ZUP_SQL_CONFIG['password']};" f"TrustServerCertificate={ZUP_SQL_CONFIG.get('trust_server_certificate', 'yes')};" f"Encrypt={ZUP_SQL_CONFIG.get('encrypt', 'no')};" ) def fetch_zup_absences_from_sql(target_date) -> pd.DataFrame: """ Извлекает оперативные отсутствия и действующие декреты из MS SQL 1С:ЗУП 3.1 на указанную дату (принимает как datetime.date, так и строку 'DD.MM.YYYY' / 'DD_MM_YYYY'). """ if isinstance(target_date, str): clean_date_str = target_date.replace('_', '.') try: target_date = datetime.strptime(clean_date_str, "%d.%m.%Y").date() except ValueError: logger.error(f"[❌] Неверный формат даты для SQL-запроса: {target_date}. Ожидался DD.MM.YYYY") return pd.DataFrame() query = """ DECLARE @TargetDate DATE = ?; -- 1. Оперативные отсутствия (Отпуска, Командировки, Больничные, Отгулы) SELECT LTRIM(RTRIM(ref_emp._Description)) AS [ФИО], CASE state._Fld16925RRef WHEN 0x9C10B2452D414FDF4A90E2B2AB81D3F7 THEN N'Отпуск основной' WHEN 0xBA63FCF94B4AD0664ED369D2E6505D67 THEN N'Командировка' WHEN 0x8C3B61F23954155A40EB0108FC0932DB THEN N'Болезнь' WHEN 0x853001C18D0965EE4B2702405C94054A THEN N'Отпуск неоплачиваемый по разрешению работодателя' WHEN 0xB7335AEFD8708C3E462861FC59489A38 THEN N'Отпуск по беременности и родам' ELSE N'Другое отсутствие' END AS [Вид_отсутствия] FROM dbo._InfoRg16921 state WITH (NOLOCK) INNER JOIN dbo._Reference299 ref_emp WITH (NOLOCK) ON state._Fld16922RRef = ref_emp._IDRRef WHERE @TargetDate BETWEEN CAST(CASE WHEN YEAR(state._Fld16926) > 3000 THEN DATEADD(YEAR, -2000, state._Fld16926) ELSE state._Fld16926 END AS DATE) AND CAST(CASE WHEN YEAR(state._Fld16927) > 3000 THEN DATEADD(YEAR, -2000, state._Fld16927) ELSE state._Fld16927 END AS DATE) UNION ALL -- 2. Динамический выбор ДЕЙСТВУЮЩИХ декретниц по уходу за ребенком SELECT active_state.fio AS [ФИО], N'Отпуск по уходу за ребенком' AS [Вид_отсутствия] FROM ( SELECT LTRIM(RTRIM(ref_emp._Description)) AS fio, all_states._Fld16925RRef AS state_guid, CAST(CASE WHEN YEAR(all_states._Fld16926) > 3000 THEN DATEADD(YEAR, -2000, all_states._Fld16926) ELSE all_states._Fld16926 END AS DATE) AS date_start, ROW_NUMBER() OVER ( PARTITION BY all_states._Fld16922RRef ORDER BY all_states._Fld16926 DESC ) AS rn FROM dbo._InfoRg16921 all_states WITH (NOLOCK) INNER JOIN dbo._Reference299 ref_emp WITH (NOLOCK) ON all_states._Fld16922RRef = ref_emp._IDRRef WHERE CAST(CASE WHEN YEAR(all_states._Fld16926) > 3000 THEN DATEADD(YEAR, -2000, all_states._Fld16926) ELSE all_states._Fld16926 END AS DATE) <= @TargetDate ) active_state WHERE active_state.rn = 1 AND active_state.state_guid = 0xA4FBA038663B3C2A48DA151C262855E1 AND active_state.date_start >= DATEADD(YEAR, -3, @TargetDate) ORDER BY [ФИО] ASC; """ try: conn_str = get_zup_connection_string() with pyodbc.connect(conn_str, timeout=30) as conn: df = pd.read_sql(query, conn, params=[target_date]) if not df.empty: df['fio_clean'] = df['ФИО'].apply(normalize_fio) return df except Exception as e: logger.error(f"[❌] Ошибка SQL-выгрузки отсутствий за {target_date}: {e}") return pd.DataFrame() def fetch_zup_staff_from_sql() -> pd.DataFrame: """Резервная выгрузка штата из MS SQL.""" query = """ SELECT DISTINCT LTRIM(RTRIM(ref_emp._Description)) AS [ФИО], N'Организация' AS [Подразделение], N'Сотрудник' AS [Должность] FROM dbo._Reference299 ref_emp WITH (NOLOCK) WHERE ref_emp._Description <> '' AND ref_emp._Marked = 0x00 ORDER BY [ФИО] ASC; """ try: conn_str = get_zup_connection_string() with pyodbc.connect(conn_str, timeout=5) as conn: df = pd.read_sql(query, conn) if not df.empty: df['fio_clean'] = df['ФИО'].apply(normalize_fio) return df except Exception as e: logger.error(f"[❌] Ошибка выгрузки штата из MS SQL: {e}") return pd.DataFrame() def sync_zup_to_excel(target_date) -> bool: """Создает дамп в Excel при необходимости.""" try: if isinstance(target_date, str): clean_date_str = target_date.replace('_', '.') target_date_obj = datetime.strptime(clean_date_str, "%d.%m.%Y").date() else: target_date_obj = target_date date_str_file = target_date_obj.strftime("%d_%m_%Y") c_1c_dir = os.path.join(DATA_DIR, "1c") os.makedirs(c_1c_dir, exist_ok=True) df_staff = fetch_zup_staff_from_sql() df_absences = fetch_zup_absences_from_sql(target_date_obj) staff_excel_path = os.path.join(c_1c_dir, f"Штат_{date_str_file}.xlsx") absences_excel_path = os.path.join(c_1c_dir, f"Отсутствия_{date_str_file}.xlsx") with pd.ExcelWriter(staff_excel_path, engine='openpyxl') as writer: dummy_headers = pd.DataFrame([[""] * 13] * 8) dummy_headers.to_excel(writer, index=False, header=False) df_staff[['ФИО', 'Подразделение', 'Должность']].to_excel(writer, startrow=8, index=False) with pd.ExcelWriter(absences_excel_path, engine='openpyxl') as writer: dummy_headers = pd.DataFrame([[""] * 3] * 3) dummy_headers.to_excel(writer, index=False, header=False) df_absences[['ФИО', 'Вид_отсутствия']].to_excel(writer, startrow=3, index=False) return True except Exception as e: logger.error(f"[❌] Ошибка прямого импорта из MS SQL 1С: {e}") return False ``` ## File: `./services/ai_verifier.py` ```py import json import re import requests import pandas as pd from config import OLLAMA_URL, OLLAMA_MODEL from core.database import get_all_rules_from_db from core.database import get_department_synonyms_dict, add_department_synonym_to_db def resolve_department_exception_ai(dept_1c, dept_scud, exception_departments): """ Универсальный сопоставитель отделов. Проверяет 1С, СКУД, локальную базу синонимов SQLite и задействует ИИ для сложных случайных аббревиатур. """ if not exception_departments: return False d_1c = str(dept_1c).strip().lower() if dept_1c else "" d_scud = str(dept_scud).strip().lower() if dept_scud else "" exc_list = [d.strip().lower() for d in exception_departments] # 1. Прямая проверка: если точное имя или подстрока уже совпали в 1С или СКУД for exc in exc_list: if exc and (exc in d_1c or exc in d_scud or d_1c in exc or d_scud in exc): return True # 2. Проверка по сохраненной Базе Знаний синонимов из SQLite synonyms = get_department_synonyms_dict() for exc in exc_list: # Если в БД зафиксировано: 'овк' -> 'отдел внутреннего контроля' full_from_db = synonyms.get(exc, "") if full_from_db and (full_from_db in d_1c or full_from_db in d_scud): return True # 3. Умный ИИ-арбитраж (если отдел спорный и еще не сохранен в БД) if d_1c or d_scud: dept_to_check = d_1c if d_1c else d_scud prompt = f""" Ты — кадровый аналитик. Проверь, является ли отдел сотрудника "{dept_to_check}" тем же самым подразделением, что и один из отделов-исключений: {exception_departments}? Примеры: - "Отдел внутреннего контроля" — это "ОВК" (Да) - "Отдел технического обеспечения" — это "ОТО" (Да) Ответь СТРОГО в формате JSON: {{ "is_match": true/false, "matched_exception": "Название из списка исключений", "explanation": "краткое объяснение" }} """ try: raw_res = ask_ollama(prompt, system_prompt="Отвечай только валидным JSON.") match = re.search(r'\{.*\}', raw_res, re.DOTALL) if match: data = json.loads(match.group(0)) if data.get("is_match"): matched_exc = data.get("matched_exception", "").lower() # Запоминаем открытую ИИ связь в SQLite навсегда! add_department_synonym_to_db(matched_exc, dept_to_check) return True except Exception as e: print(f"[⚠️] Ошибка ИИ-арбитража отделов: {e}") return False def ask_ollama(prompt, system_prompt=None): """ Универсальная функция отправки запросов к локальной модели Ollama (Qwen 2.5). """ payload = { "model": OLLAMA_MODEL, "prompt": f"{system_prompt}\n\n{prompt}" if system_prompt else prompt, "stream": False } try: response = requests.post(OLLAMA_URL, json=payload, timeout=120) if response.status_code == 200: data = response.json() if "response" in data: return data["response"].strip() elif "message" in data and "content" in data["message"]: return data["message"]["content"].strip() else: return f"⚠️ Неизвестная структура ответа Ollama: {data}" else: return f"⚠️ Ошибка Ollama (Код {response.status_code}): {response.text}" except requests.exceptions.ConnectionError: return "⚠️ Не удалось подключиться к Ollama. Проверьте, запущен ли сервис (ollama run qwen2.5:14b)." except Exception as e: return f"⚠️ Ошибка при обращении к Ollama: {e}" def get_system_rules_context(): """ Загружает динамические системные правила компании напрямую из Базы Данных SQLite. """ rules = get_all_rules_from_db() if not rules: return "" rules_text = "\n".join([f"- {r}" for r in rules]) return f"\nОБЯЗАТЕЛЬНЫЕ ГЛОБАЛЬНЫЕ ПРАВИЛА И ПРИОРИТЕТЫ КОМПАНИИ (ИЗ SQLITE БД):\n{rules_text}\n" def analyze_scud_mass_failure_ai(df_scud): """ Оценивает процент аномалий 'ANOMALY_NO_IN_HAS_ACTIVITY' в выгрузке. Если процент аномалий превышает 5% от смены или 10 человек, вызывает ИИ для генерации критического алерта. """ if df_scud is None or df_scud.empty: return None total_records = len(df_scud) anomaly_rows = df_scud[df_scud.get('anomaly_flag', 'NONE') == 'ANOMALY_NO_IN_HAS_ACTIVITY'] anomaly_count = len(anomaly_rows) if total_records == 0: return None anomaly_percent = round((anomaly_count / total_records) * 100, 1) if anomaly_percent > 5.0 or anomaly_count >= 10: rules_context = get_system_rules_context() prompt = f""" {rules_context} ВНИМАНИЕ! Проведён анализ смены СКУД: - Всего записей за смену: {total_records} - Выявлено сотрудников без утреннего входа, но с зафиксированной дневной активностью: {anomaly_count} ({anomaly_percent}% от смены) Сформируй понятное предупреждение для Администратора СКУД и Руководителя. Объясни, что это критический массовый сбой турникетов/контроллеров входа на КПП, и порекомендуй действия. """ alert_text = ask_ollama(prompt, system_prompt="Ты — ИИ-аналитик контроллинга СКУД. Отвечай кратко и строго по делу.") return { "is_mass_failure": True, "anomaly_count": anomaly_count, "anomaly_percent": anomaly_percent, "alert_text": alert_text } return { "is_mass_failure": False, "anomaly_count": anomaly_count, "anomaly_percent": anomaly_percent, "alert_text": "Массовых сбоев оборудования не зафиксировано." } def ai_verify_scud_against_staff(unrecognized_scud_fios, staff_fios): if not unrecognized_scud_fios or not staff_fios: return {} # Исключаем точные совпадения staff_fios_set = set(staff_fios) real_unrecognized = [f for f in unrecognized_scud_fios if f not in staff_fios_set] if not real_unrecognized: return {} rules_context = get_system_rules_context() prompt = f""" Ты — кадровый аудитор безопасности СКУД. {rules_context} В СКУД записаны неопознанные ФИО: {json.dumps(real_unrecognized, ensure_ascii=False)} В официальном Штатном расписании 1С записаны ЭТАЛОНЫ: {json.dumps(staff_fios, ensure_ascii=False)} СТРОГИЕ ПРАВИЛА: 1. Запись из Штат 1С — это 100% ПРАВИЛЬНЫЙ эталон. 2. В СКУД допущена опечатка. 3. Любое присутствие сотрудника по СКУД при наличии в 1С документа отсутствия (кроме командировок) является гарантированной аномалией. 4. В поле "warning" опиши обнаруженную опечатку. 5. Запрещено выдумывать опечатки и объединять разных людей/однофамильцев! ОТВЕЧАЙ ТОЛЬКО ИСКЛЮЧИТЕЛЬНО В ФОРМАТЕ ВАЛИДНОГО JSON: {{ "verified_matches": [ {{ "scud_fio": "ФИО из СКУД", "staff_fio": "эталон ФИО из Штат 1С", "warning": "Описание опечатки в СКУД" }} ] }} """ raw_response = ask_ollama( prompt, system_prompt="Ты — строгий JSON API генератор. Отвечай только валидным JSON объектом без пояснительного текста." ) mapping = {} if not raw_response: return mapping # Фаза 1: Попытка прямого разбора с санитарной очисткой try: match = re.search(r'\{.*\}', raw_response, re.DOTALL) if match: json_str = match.group(0) # Убираем висячие запятые: {"a": 1,} -> {"a": 1} json_str = re.sub(r',\s*([\}\]])', r'\1', json_str) # Заменяем одинарные кавычки в ключах/значениях на двойные при необходимости json_str = re.sub(r"(?<=\{|\,)\s*'([^']+)'\s*:", r'"\1":', json_str) data = json.loads(json_str) for item in data.get("verified_matches", []): scud_f = item.get("scud_fio") staff_f = item.get("staff_fio") warn = item.get("warning", "Точное совпадение (без опечаток)") if scud_f and staff_f and staff_f in staff_fios_set: mapping[scud_f] = {"staff_fio": staff_f, "warning": warn} return mapping except Exception: pass # Фаза 2: Резервный Regex-парсер (если JSON синтаксически сломан, но пары ключ-значение есть) try: pattern = r'["\']scud_fio["\']\s*:\s*["\']([^"\']+)["\'].*?["\']staff_fio["\']\s*:\s*["\']([^"\']+)["\']' matches = re.findall(pattern, raw_response, re.DOTALL) for scud_f, staff_f in matches: scud_clean = scud_f.strip() staff_clean = staff_f.strip() if staff_clean in staff_fios_set: mapping[scud_clean] = {"staff_fio": staff_clean, "warning": "Восстановлено парсером опечаток"} except Exception as e: print(f"[!] Ошибка резервного парсинга опечаток: {e}") return mapping def ai_verify_department_exceptions(unexplained_df, exception_departments): """ Локальный ИИ проверяет, не являются ли неотмеченные отделы синонимами отделов-исключений. """ if unexplained_df.empty or not exception_departments: return [] dept_list = unexplained_df['Подразделение'].dropna().unique().tolist() rules_context = get_system_rules_context() prompt = f""" {rules_context} Ты — кадровый аудитор. Список отделов-исключений компании: {json.dumps(exception_departments, ensure_ascii=False)} Список отделов сотрудников, попавших в неизвестные: {json.dumps(dept_list, ensure_ascii=False)} Определи, какие из отделов сотрудников являются ПОЛНЫМИ НАЗВАНИЯМИ или СИНУНИМАМИ отделов-исключений (например: 'Отдел внутреннего контроля' — это 'ОВК'). Выдай ответ строго в формате JSON: {{ "matched_departments": ["Название отдела 1", "Название отдела 2"] }} """ raw_response = ask_ollama(prompt, system_prompt="Выдавай только валидный JSON.") try: match = re.search(r'\{.*\}', raw_response, re.DOTALL) if match: data = json.loads(match.group(0)) return data.get("matched_departments", []) except Exception as e: print(f"[⚠️] Ошибка ИИ-арбитража отделов: {e}") return [] ``` ## File: `./services/knowledge_base.py` ```py import os from core.database import get_all_rules_from_db, add_rule_to_db def load_knowledge_base(): """ Загружает актуальные правила Базы Знаний компании напрямую из базы данных SQLite. """ rules = get_all_rules_from_db() return { "rules": rules if rules else [], "fio_corrections": {} } def add_rule_to_kb(new_rule): """ Добавляет новое принятое человеком правило в SQLite таблицу ai_knowledge_base. """ if not new_rule or not new_rule.strip(): return add_rule_to_db(new_rule.strip(), added_by="Human") print(f"[✓] База знаний SQLite успешно обновлена! Новое правило: {new_rule.strip()}") ``` ## File: `./services/knowledge/service.py` ```py """ =============================================================================== FILE: services/knowledge/service.py PROJECT: SCUD Orion AI (Unified Architecture) MODULE: services / knowledge ROLE: Доменный сервис базы знаний, правил компании и синонимов подразделений. =============================================================================== """ from typing import List, Dict, Any from core.repositories.zup_repo import ( get_all_rules_from_db, add_rule_to_db, get_department_synonyms_dict, add_department_synonym_to_db ) def get_rules() -> List[Dict[str, Any]]: """Получить все правила базы знаний в виде списка словарей.""" raw_rules = get_all_rules_from_db() return [{"id": idx, "rule_text": r} for idx, r in enumerate(raw_rules, 1)] def add_rule(rule_text: str, added_by: str = "Human") -> None: """Добавить новое правило в базу знаний.""" add_rule_to_db(rule_text, added_by=added_by) def get_synonyms() -> Dict[str, str]: """Получить словарь синонимов отделов.""" return get_department_synonyms_dict() def register_department_synonym(short_name: str, full_name: str) -> None: """Сохранить новую пару синонимов подразделения.""" add_department_synonym_to_db(short_name, full_name) ``` ## File: `./services/scud_etl/pipeline.py` ```py """ =============================================================================== FILE: services/scud_etl/pipeline.py =============================================================================== """ import os import logging from typing import Optional, Dict, Any, Tuple import pandas as pd from core.connection import get_connection from core.database import load_scud_from_db_by_snapshot from config import DATA_DIR from services.data_loader import load_staff_data, load_absent_data logger = logging.getLogger("SCUD_PIPELINE") def load_best_snapshot_for_date(date_str: str, prefer_final_y: bool = False) -> Optional[pd.DataFrame]: """ Загружает наилучший срез СКУД за дату. Если prefer_final_y=True — отдает предпочтение финишному Y (23:59:59). """ with get_connection() as conn: cursor = conn.cursor() target_snap_id = None if prefer_final_y: cursor.execute(""" SELECT snapshot_id FROM scud_logs WHERE log_date = ? AND (snapshot_id LIKE 'Y%' OR snapshot_time LIKE '%23:59:59' OR snapshot_time LIKE '%22:00:00') ORDER BY id DESC LIMIT 1 """, (date_str,)) row = cursor.fetchone() if row: target_snap_id = row[0] if not target_snap_id: cursor.execute(""" SELECT snapshot_id FROM scud_logs WHERE log_date = ? ORDER BY id DESC LIMIT 1 """, (date_str,)) row = cursor.fetchone() if row: target_snap_id = row[0] if not target_snap_id: return None return load_scud_from_db_by_snapshot(date_str, snapshot_param=target_snap_id) def load_1c_files_for_date(date_str: str) -> Tuple[Optional[pd.DataFrame], Optional[pd.DataFrame]]: """ Загружает реестры штата и отсутствий 1С на указанную дату через data_loader. """ df_staff = load_staff_data(date_str) df_abs = load_absent_data(date_str) return df_staff, df_abs ``` ## File: `./services/scud_etl/merger.py` ```py """ =============================================================================== 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 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 [] } ``` ## File: `./services/scud_etl/anomaly_detector.py` ```py """ =============================================================================== FILE: services/scud_etl/anomaly_detector.py ROLE: Детектирование аномалий СКУД, дубликатов пропусков и несоответствий с 1С. =============================================================================== """ import logging from typing import List, Dict, Any import pandas as pd from config import normalize_fio logger = logging.getLogger("SCUD_ANOMALY") def detect_registry_anomalies(df_merged: pd.DataFrame, df_raw_scud: pd.DataFrame = None) -> List[Dict[str, Any]]: anomalies = [] if df_merged is None or df_merged.empty: return anomalies # 1. ⭐️ Поиск дубликатов учеток/пропусков в сыром СКУД if df_raw_scud is not None and not df_raw_scud.empty: df_raw = df_raw_scud.copy() if 'fio_clean' not in df_raw.columns: f_col = 'Сотрудник' if 'Сотрудник' in df_raw.columns else 'ФИО' df_raw['fio_clean'] = df_raw[f_col].apply(normalize_fio) counts = df_raw['fio_clean'].value_counts() duplicate_fios = counts[counts > 1].index.tolist() for dup_fio in duplicate_fios: sub = df_raw[df_raw['fio_clean'] == dup_fio] departments = ", ".join(sub['Подразделение'].astype(str).unique()) anomalies.append({ "fio": dup_fio, "type": "DUPLICATE_SCUD_CARD", "description": f"Сотрудник заведен в СКУД {len(sub)} раза (отделы: {departments}). События входа и выхода объединены автоматически." }) # 2. Поиск кадровых аномалий и сбоев оборудования for _, row in df_merged.iterrows(): if row.get("is_excluded", False): continue fio = row.get("fio_clean") or row.get("Сотрудник", "") start_day = str(row.get("Начало_дня", "")).strip() end_day = str(row.get("Конец_дня", "")).strip() reason = str(row.get("причина отсутствия", row.get("Вид_отсутствия", ""))).strip() reason_lower = reason.lower() if "исключен" in reason_lower or "овк" in reason_lower: continue # Приход в офис во время отпуска/больничного (удаленка и командировки разрешены) if start_day not in ["Нет входа", "—", "", "nan", "None"] and reason and reason != "nan": if not ("удален" in reason_lower or "дистанцион" in reason_lower or "командировк" in reason_lower or "поездк" in reason_lower): anomalies.append({ "fio": fio, "type": "PHYSICAL_PRESENCE_DURING_ABSENCE", "description": f"Присутствовал в здании ({start_day}), но в 1С числится документ: '{reason}'." }) # Ошибка считывателя (есть выход без входа) if start_day in ["Нет входа", "—", ""] and end_day not in ["Нет выхода", "—", "", "nan", "None"]: anomalies.append({ "fio": fio, "type": "SCUD_EQUIPMENT_ANOMALY", "description": f"Зафиксирован выход ({end_day}) при отсутствии отметки утреннего входа." }) return anomalies ``` ## File: `./services/snapshots/service.py` ```py """ =============================================================================== FILE: services/snapshots/service.py PROJECT: SCUD Orion AI (Unified Architecture) MODULE: services / snapshots ROLE: Бизнес-логика срезов СКУД (выборка, валидация Y-срезов, удаление). AI-CONTEXT-ANCHORS: - ANCHOR[SNAPSHOT_GET_REGISTRY]: Выборка срезов с разметкой защищенных Y-снапшотов. - ANCHOR[SNAPSHOT_DELETE_SAFE]: Безопасное удаление дневных срезов с защитой итоговых. =============================================================================== """ from typing import Dict, Any, Optional, List from core.connection import get_connection from core.repositories.scud_repo import get_available_snapshots, delete_snapshot_by_id # ANCHOR[SNAPSHOT_GET_REGISTRY] def get_snapshots_registry(date_str: Optional[str] = None) -> Dict[str, Any]: """Возвращает реестр снапшотов за дату или за все доступные дни.""" clean_date = date_str.strip() if date_str else "" rows = get_available_snapshots(date_str=clean_date if clean_date else None) snapshots = [ { "snapshot_id": r[0], "log_date": r[1], "snapshot_time": r[2], "record_count": r[3], "is_final": str(r[0]).startswith("Y") } for r in rows ] return { "query_date": clean_date or "все", "snapshots_count": len(snapshots), "snapshots": snapshots } # ANCHOR[SNAPSHOT_DELETE_SAFE] def delete_snapshots_safely(snapshot_ids: List[str]) -> Dict[str, Any]: """ Удаляет выбранные дневные снапшоты. Итоговые вечерние срезы с префиксом 'Y' гарантированно защищены от удаления. """ if not snapshot_ids: return {"status": "error", "message": "Не указаны ID снапшотов для удаления."} safe_ids = [str(s).strip() for s in snapshot_ids if s and not str(s).strip().startswith("Y")] if not safe_ids: return {"status": "error", "message": "⚠️ Итоговый срез Y защищен от удаления. Выберите дневные снапшоты."} with get_connection() as conn: cursor = conn.cursor() placeholders = ",".join(["?"] * len(safe_ids)) cursor.execute(f"DELETE FROM scud_logs WHERE snapshot_id IN ({placeholders})", safe_ids) deleted_count = cursor.rowcount conn.commit() return { "status": "success", "deleted_count": deleted_count, "deleted_ids": safe_ids, "message": f"Успешно удалено снапшотов: {len(safe_ids)} шт." } ``` ## File: `./services/tasks/repository.py` ```py """ =============================================================================== FILE: services/tasks/repository.py =============================================================================== """ import re from typing import List, Dict, Any, Optional from core.connection import get_connection def normalize_task_id(task_id_input: str) -> str: if not task_id_input: return "" clean_id = str(task_id_input).strip().upper().replace("TASK-", "").replace("TASK", "").replace("#", "") if clean_id.isdigit(): num = int(clean_id) return f"TASK-{num:02d}" if num < 100 else f"TASK-{num:03d}" return f"TASK-{clean_id}" def repo_get_tasks(user_id: int, status: Optional[str] = None) -> List[Dict[str, Any]]: with get_connection(row_factory=True) as conn: cursor = conn.cursor() if status and status.upper() != "ALL": target_status = status.upper() if target_status in ["PROGRESS", "В РАБОТЕ"]: target_status = "IN_PROGRESS" elif target_status in ["DONE", "ГОТОВО"]: target_status = "COMPLETED" elif target_status in ["PLANNED", "ПЛАНЫ"]: target_status = "BACKLOG" cursor.execute(""" SELECT id, task_id, module, title, priority, status, due_date, created_at FROM tasks WHERE user_id = ? AND (status = ? OR (status = 'BACKLOG' AND ? = 'PLANNED')) ORDER BY id DESC """, (user_id, target_status, target_status)) else: cursor.execute(""" SELECT id, task_id, module, title, priority, status, due_date, created_at FROM tasks WHERE user_id = ? ORDER BY id DESC """, (user_id,)) rows = cursor.fetchall() return [dict(r) for r in rows] def repo_add_task( user_id: int, module: str, title: str, priority: str = "MEDIUM", due_date: Optional[str] = None, status: str = "BACKLOG" ) -> Dict[str, Any]: target_status = status.upper() if status else "BACKLOG" if target_status in ["PROGRESS", "В РАБОТЕ"]: target_status = "IN_PROGRESS" elif target_status in ["DONE", "ГОТОВО"]: target_status = "COMPLETED" elif target_status in ["PLANNED", "ПЛАНЫ", "BACKLOG"]: target_status = "BACKLOG" with get_connection() as conn: cursor = conn.cursor() cursor.execute("SELECT MAX(id) FROM tasks") max_id = cursor.fetchone()[0] or 0 new_task_id = f"TASK-{(max_id + 1):02d}" cursor.execute(""" INSERT INTO tasks (task_id, module, title, priority, status, due_date, user_id) VALUES (?, ?, ?, ?, ?, ?, ?) """, (new_task_id, module or "general", title.strip(), priority.upper(), target_status, due_date, user_id)) conn.commit() return {"status": "success", "task_id": new_task_id, "id": max_id + 1} def repo_update_task( user_id: int, task_id: str, title: Optional[str] = None, priority: Optional[str] = None, status: Optional[str] = None, due_date: Optional[str] = None ) -> Dict[str, Any]: clean_num = re.sub(r'\D', '', str(task_id)) formatted_id = normalize_task_id(task_id) updates = [] params = [] if title is not None and title.strip(): updates.append("title = ?") params.append(title.strip()) if priority is not None and priority.strip(): updates.append("priority = ?") params.append(priority.strip().upper()) if status is not None and status.strip(): target_status = status.strip().upper() if target_status in ["PROGRESS", "В РАБОТЕ"]: target_status = "IN_PROGRESS" elif target_status in ["DONE", "ГОТОВО"]: target_status = "COMPLETED" elif target_status in ["PLANNED", "ПЛАНЫ"]: target_status = "BACKLOG" updates.append("status = ?") params.append(target_status) if due_date is not None: updates.append("due_date = ?") params.append(due_date.strip() if due_date.strip() else None) if not updates: return {"status": "success", "message": "Нет данных для обновления"} params.extend([clean_num, formatted_id, f"%{task_id.strip()}", user_id]) sql = f""" UPDATE tasks SET {', '.join(updates)} WHERE (id = ? OR UPPER(task_id) = ? OR task_id LIKE ?) AND user_id = ? """ with get_connection() as conn: cursor = conn.cursor() cursor.execute(sql, params) rows_affected = cursor.rowcount conn.commit() if rows_affected == 0: return {"error": f"Задача {task_id} не найдена или принадлежит другому пользователю"} return {"status": "success", "message": f"Задача #{task_id} успешно обновлена"} def repo_delete_task(user_id: int, task_id: str) -> Dict[str, Any]: clean_num = re.sub(r'\D', '', str(task_id)) formatted_id = normalize_task_id(task_id) with get_connection() as conn: cursor = conn.cursor() cursor.execute(""" DELETE FROM tasks WHERE (id = ? OR UPPER(task_id) = ? OR task_id LIKE ?) AND user_id = ? """, (clean_num, formatted_id, f"%{task_id.strip()}", user_id)) deleted = cursor.rowcount conn.commit() if deleted == 0: return {"error": f"Задача {task_id} не найдена"} return {"status": "success", "message": f"Задача #{task_id} удалена"} ``` ## File: `./services/tasks/service.py` ```py """ =============================================================================== FILE: services/tasks/service.py PROJECT: SCUD Orion AI (Unified Architecture) MODULE: services / tasks ROLE: Единый доменный сервис задач (бизнес-логика и диспетчер операций). AI-CONTEXT-ANCHORS: - ANCHOR[TASK_SERVICE_DISPATCHER]: Маршрутизация действий ADD/UPDATE/DELETE/EXPORT. =============================================================================== """ from typing import Dict, Any, Optional, List from .repository import repo_get_tasks, repo_add_task, repo_update_task, repo_delete_task from .exporter import export_tasks_to_markdown def get_tasks(user_id: int, status: Optional[str] = None) -> List[Dict[str, Any]]: """Получить список задач.""" return repo_get_tasks(user_id, status) def add_task(user_id: int, module: str, title: str, priority: str = "MEDIUM", due_date: Optional[str] = None, status: str = "BACKLOG") -> Dict[str, Any]: """Создать задачу.""" res = repo_add_task(user_id, module, title, priority, due_date, status) return {"status": "success", "task_id": res["task_id"], "message": f"Задача #{res['id']} создана и добавлена в планы"} def update_task_details(user_id: int, task_id: str, title: Optional[str] = None, priority: Optional[str] = None, status: Optional[str] = None, due_date: Optional[str] = None) -> Dict[str, Any]: """Обновить задачу.""" return repo_update_task(user_id, task_id, title, priority, status, due_date) def delete_task(user_id: int, task_id: str) -> Dict[str, Any]: """Удалить задачу.""" return repo_delete_task(user_id, task_id) # ANCHOR[TASK_SERVICE_DISPATCHER] def execute_task_action( user_id: int, action: str, task_id: Optional[str] = None, title: Optional[str] = None, priority: Optional[str] = "MEDIUM", status: Optional[str] = None, module: Optional[str] = "general", due_date: Optional[str] = None, filename: Optional[str] = "ROADMAP.md" ) -> Dict[str, Any]: """Консолидированный диспетчер операций над задачами.""" act = (action or "").strip().upper() if act == "ADD": if not title: return {"status": "error", "message": "Для создания задачи требуется указать title"} return add_task(user_id, module or "general", title, priority or "MEDIUM", due_date, status or "BACKLOG") elif act == "UPDATE": if not task_id: return {"status": "error", "message": "Для обновления требуется указать task_id"} return update_task_details(user_id, str(task_id), title, priority, status, due_date) elif act == "DELETE": if not task_id: return {"status": "error", "message": "Для удаления требуется указать task_id"} return delete_task(user_id, str(task_id)) elif act == "EXPORT": return export_tasks_to_markdown(user_id, filename=filename or "ROADMAP.md", status_filter=status) return {"status": "error", "message": f"Неизвестное действие action='{action}'"} ```