480 lines
20 KiB
Python
480 lines
20 KiB
Python
import os
|
||
import sqlite3
|
||
import json
|
||
import pandas as pd
|
||
from datetime import datetime
|
||
from config import DATA_DIR
|
||
|
||
DB_PATH = os.path.join(DATA_DIR, "scud_orion_ai.db")
|
||
|
||
|
||
def get_connection():
|
||
"""Создает подключение к базе данных SQLite с оптимизированными настройками."""
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.execute("PRAGMA foreign_keys = ON;")
|
||
conn.execute("PRAGMA journal_mode = WAL;")
|
||
return conn
|
||
|
||
|
||
def init_db():
|
||
"""Создает структуру таблиц и индексов в базе данных."""
|
||
with get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
|
||
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
|
||
);
|
||
""")
|
||
|
||
cursor.execute("PRAGMA table_info(scud_logs);")
|
||
cols = [col[1] for col in cursor.fetchall()]
|
||
if 'snapshot_id' not in cols:
|
||
cursor.execute("ALTER TABLE scud_logs ADD COLUMN snapshot_id TEXT DEFAULT NULL;")
|
||
|
||
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
|
||
);
|
||
""")
|
||
|
||
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 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_scud_snapshot ON scud_logs(snapshot_time);")
|
||
|
||
conn.commit()
|
||
|
||
sync_knowledge_base_to_db()
|
||
|
||
def has_scud_logs_for_date(date_str):
|
||
"""Проверяет, есть ли в базе данные СКУД за указанную дату."""
|
||
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):
|
||
"""Проверяет, зафиксирован ли уже ИТОГОВЫЙ вчерашний снапшот с индексом Y/22:00."""
|
||
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 '%22:00:00') LIMIT 1",
|
||
(date_str,)
|
||
)
|
||
return cursor.fetchone() is not None
|
||
|
||
|
||
def get_or_create_snapshot_id(snapshot_time, date_str=None, is_yesterday=False):
|
||
"""
|
||
Генерирует датированный составной ID снапшота (YYYYMMDD-NNN или YYYYMMDD-NNN).
|
||
Префикс 'Y' присваивается автоматически, если дата логов (date_str) предшествует дате снятия (snapshot_time).
|
||
"""
|
||
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")
|
||
|
||
# Автоматическая проверка: если дата логов раньше даты выгрузки — проставляем Y
|
||
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()
|
||
|
||
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]
|
||
|
||
cursor.execute("""
|
||
SELECT snapshot_id
|
||
FROM scud_logs
|
||
WHERE snapshot_id LIKE ? OR snapshot_id LIKE ?
|
||
ORDER BY snapshot_id DESC LIMIT 1
|
||
""", (f"{date_prefix}-%", f"Y{date_prefix}-%"))
|
||
|
||
last_row = cursor.fetchone()
|
||
next_seq = 1
|
||
|
||
if last_row and last_row[0]:
|
||
parts = last_row[0].replace("Y", "").split("-")
|
||
if len(parts) > 1 and parts[1].isdigit():
|
||
next_seq = int(parts[1]) + 1
|
||
|
||
return f"{prefix}{date_prefix}-{next_seq:03d}"
|
||
|
||
|
||
def save_scud_to_db(df_scud, date_str, snapshot_time=None, is_yesterday=False):
|
||
"""Сохраняет логи СКУД с автоматическим присвоением составного snapshot_id."""
|
||
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")
|
||
|
||
# Передаем date_str в генератор snapshot_id
|
||
snapshot_id = get_or_create_snapshot_id(snapshot_time, date_str=date_str, is_yesterday=is_yesterday)
|
||
|
||
data_to_insert = []
|
||
for _, r in df_scud.iterrows():
|
||
data_to_insert.append((
|
||
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
|
||
))
|
||
|
||
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=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, snapshot_param=None):
|
||
"""
|
||
Загружает логи СКУД по ID снапшота.
|
||
Если snapshot_param не задан (или искался прошлый день), приоритетно выбирает
|
||
финальный вечерний снапшот, начинающийся с буквы 'Y' (срез на 22:00) за дату date_str.
|
||
"""
|
||
with get_connection() as conn:
|
||
df = pd.DataFrame()
|
||
|
||
# 1. Если передан конкретный ID снапшота (например, '20260810-001')
|
||
if snapshot_param:
|
||
df = pd.read_sql_query(
|
||
"SELECT * FROM scud_logs WHERE snapshot_id = ?",
|
||
conn, params=(str(snapshot_param),)
|
||
)
|
||
|
||
# 2. Если по snapshot_param ничего не найдено ИЛИ snapshot_param=None (запрос за вчерашний день):
|
||
if df.empty and date_str:
|
||
cursor = conn.cursor()
|
||
|
||
# ⭐️ ПРИОРИТЕТ 1: Ищем срез, начальный символ которого 'Y' (вечерний зафиксированный Y-снапшот)
|
||
cursor.execute(
|
||
"SELECT snapshot_id FROM scud_logs WHERE log_date = ? AND snapshot_id LIKE 'Y%' ORDER BY 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 id DESC LIMIT 1",
|
||
(date_str,)
|
||
)
|
||
row = cursor.fetchone()
|
||
|
||
if row and row[0]:
|
||
target_snap_id = row[0]
|
||
df = pd.read_sql_query(
|
||
"SELECT * FROM scud_logs WHERE snapshot_id = ?",
|
||
conn, params=(target_snap_id,)
|
||
)
|
||
|
||
# Приводим названия колонок к стандарту
|
||
if not df.empty:
|
||
rename_map = {
|
||
'department': 'Подразделение',
|
||
'position': 'Должность',
|
||
'fio': 'Сотрудник',
|
||
'time_in': 'Начало_дня',
|
||
'first_activity': 'Первая_активность',
|
||
'time_out': 'Конец_дня',
|
||
'duration': 'Находился_в_здании',
|
||
'is_present': 'Пришел'
|
||
}
|
||
# Переименовываем имеющиеся колонки
|
||
df = df.rename(columns={k: v for k, v in rename_map.items() if k in df.columns})
|
||
|
||
# Заполняем недостающие колонки значениями по умолчанию, если их не было в БД
|
||
required_cols = ['Пришел', 'Начало_дня', 'Первая_активность', 'Конец_дня', 'Находился_в_здании', 'anomaly_flag']
|
||
for col in required_cols:
|
||
if col not in df.columns:
|
||
if col == 'Пришел':
|
||
df[col] = False
|
||
else:
|
||
df[col] = '—'
|
||
|
||
# Приводим тип флага Пришел к bool
|
||
if 'Пришел' in df.columns:
|
||
df['Пришел'] = df['Пришел'].astype(bool)
|
||
|
||
return df
|
||
|
||
def get_available_snapshots(date_str=None):
|
||
"""
|
||
Возвращает список снапшотов.
|
||
Сортировка по времени snapshot_time DESC.
|
||
"""
|
||
with get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
if date_str:
|
||
cursor.execute("""
|
||
SELECT
|
||
snapshot_id,
|
||
log_date,
|
||
snapshot_time,
|
||
COUNT(*) as cnt
|
||
FROM scud_logs
|
||
WHERE log_date = ? AND snapshot_time IS NOT NULL
|
||
GROUP BY snapshot_id, log_date, snapshot_time
|
||
ORDER BY snapshot_time DESC
|
||
""", (date_str,))
|
||
else:
|
||
cursor.execute("""
|
||
SELECT
|
||
snapshot_id,
|
||
log_date,
|
||
snapshot_time,
|
||
COUNT(*) as cnt
|
||
FROM scud_logs
|
||
WHERE snapshot_time IS NOT NULL
|
||
GROUP BY snapshot_id, log_date, snapshot_time
|
||
ORDER BY snapshot_time DESC
|
||
""")
|
||
return cursor.fetchall()
|
||
|
||
|
||
def save_staff_to_db(df_staff, date_str):
|
||
if df_staff is None or df_staff.empty:
|
||
return
|
||
data_to_insert = [
|
||
(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_to_insert)
|
||
conn.commit()
|
||
|
||
|
||
def save_absences_to_db(df_absent, date_str):
|
||
if df_absent is None or df_absent.empty:
|
||
return
|
||
data_to_insert = [
|
||
(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_to_insert)
|
||
conn.commit()
|
||
|
||
|
||
def save_anomalies_to_db(anomalies_list, date_str):
|
||
if not anomalies_list:
|
||
return
|
||
data_to_insert = [
|
||
(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_to_insert)
|
||
conn.commit()
|
||
|
||
|
||
def get_all_rules_from_db():
|
||
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, added_by="Human"):
|
||
if not rule_text or not rule_text.strip():
|
||
return
|
||
with get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
try:
|
||
cursor.execute("INSERT OR IGNORE INTO ai_knowledge_base (rule_text, added_by) VALUES (?, ?)", (rule_text.strip(), added_by))
|
||
conn.commit()
|
||
except Exception as e:
|
||
print(f"[⚠️] Ошибка записи правила в БД: {e}")
|
||
|
||
|
||
def sync_knowledge_base_to_db():
|
||
pass
|
||
|
||
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"[✓] Удален снапшот [{snapshot_id}]. Удалено строк: {deleted_count}")
|
||
return deleted_count
|
||
|
||
def delete_snapshots_by_date(date_str: str):
|
||
"""Удаляет все снапшоты за указанную дату (например, '04.08.2026')."""
|
||
with get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
# Удаляем по log_date или по дате внутри snapshot_id
|
||
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"[✓] Удалены все снапшоты за дату [{date_str}]. Удалено строк: {deleted_count}")
|
||
return deleted_count
|
||
def init_department_synonyms_db():
|
||
"""Создает таблицу синонимов отделов в БД SQLite."""
|
||
with get_connection() as conn:
|
||
cursor = conn.cursor()
|
||
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
|
||
);
|
||
""")
|
||
# По умолчанию добавляем ОВК -> Отдел внутреннего контроля
|
||
cursor.execute("""
|
||
INSERT OR IGNORE INTO department_synonyms (short_name, full_name)
|
||
VALUES ('овк', 'отдел внутреннего контроля')
|
||
""")
|
||
conn.commit()
|
||
|
||
|
||
def get_department_synonyms_dict():
|
||
"""Возвращает словарь всех изученных синонимов {short_name: full_name}."""
|
||
init_department_synonyms_db()
|
||
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, full_name):
|
||
"""Сохраняет новую пару синонимов отдела в базу SQLite."""
|
||
init_department_synonyms_db()
|
||
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()
|
||
print(f"[✓] В базу SQLite добавлен новый синоним отдела: '{short_name}' ⟷ '{full_name}'")
|
||
|
||
def load_staff_from_db(date_str):
|
||
"""Загружает Штатное расписание 1С из базы SQLite за указанную дату."""
|
||
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 load_absences_from_db(date_str):
|
||
"""Загружает документальные отсутствия 1С из базы SQLite за указанную дату."""
|
||
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
|
||
|