94 lines
3.3 KiB
Python
94 lines
3.3 KiB
Python
"""
|
||
===============================================================================
|
||
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": [],
|
||
"turnstile_fio": [],
|
||
"turnstile_departments": [],
|
||
"fligel_fio": [], # ⭐️ Сотрудники с доступом через Флигель (DoorIndex = 23)
|
||
"fligel_departments": [] # ⭐️ Отделы с доступом через Флигель
|
||
}
|
||
|
||
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):
|
||
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", "turnstile_fio", "fligel_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}") |