128 lines
5.8 KiB
Python
128 lines
5.8 KiB
Python
import sys
|
|
import os
|
|
import sqlite3
|
|
import pandas as pd
|
|
|
|
# Добавляем корневую директорию проекта в sys.path
|
|
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
PROJECT_ROOT = os.path.dirname(SCRIPT_DIR)
|
|
sys.path.append(PROJECT_ROOT)
|
|
|
|
from config import DATA_DIR
|
|
|
|
DB_PATH = os.path.join(DATA_DIR, "scud_orion_ai.db")
|
|
|
|
|
|
def query_db(sql, params=()):
|
|
"""Выполняет SQL-запрос к локальной БД и возвращает результат в виде DataFrame."""
|
|
if not os.path.exists(DB_PATH):
|
|
print(f"❌ База данных не найдена: {DB_PATH}")
|
|
return None
|
|
try:
|
|
conn = sqlite3.connect(DB_PATH)
|
|
df = pd.read_sql_query(sql, conn, params=params)
|
|
conn.close()
|
|
return df
|
|
except Exception as e:
|
|
print(f"❌ Ошибка выполнения запроса: {e}")
|
|
return None
|
|
|
|
|
|
def print_db_stats():
|
|
"""Выводит статистику по количеству записей во всех таблицах базы данных."""
|
|
if not os.path.exists(DB_PATH):
|
|
print(f"❌ База данных не найдена: {DB_PATH}")
|
|
return
|
|
|
|
conn = sqlite3.connect(DB_PATH)
|
|
cursor = conn.cursor()
|
|
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
|
|
tables = [t[0] for t in cursor.fetchall() if t[0] != 'sqlite_sequence']
|
|
|
|
print("\n" + "=" * 55)
|
|
print(f"📊 СТАТИСТИКА ЛОКАЛЬНОЙ БАЗЫ ДАННЫХ: {DB_PATH}")
|
|
print("=" * 55)
|
|
|
|
for t in tables:
|
|
cursor.execute(f"SELECT COUNT(*) FROM {t}")
|
|
count = cursor.fetchone()[0]
|
|
print(f" • {t:<22} : {count} строк")
|
|
|
|
conn.close()
|
|
print("=" * 55 + "\n")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) < 2:
|
|
print("\nИспользование CLI-модуля базы данных:")
|
|
print(" python scripts/db_cli.py stats -- Посмотреть статистику таблиц БД")
|
|
print(" python scripts/db_cli.py anomalies -- Просмотреть все обнаруженные аномалии")
|
|
print(" python scripts/db_cli.py rules -- Просмотреть все правила Базы Знаний ИИ")
|
|
print(" python scripts/db_cli.py scud 03.08.2026 -- Посмотреть логи СКУД за конкретную дату")
|
|
print(" python scripts/db_cli.py staff 03.08.2026 -- Посмотреть Штатное расписание за дату")
|
|
print(" python scripts/db_cli.py absences 03.08.2026 -- Посмотреть отсутствия из 1С за дату")
|
|
print(" python scripts/db_cli.py dump [filename.xlsx] -- Дамп всей базы SQLite в Excel-файл")
|
|
sys.exit(0)
|
|
|
|
cmd = sys.argv[1].lower()
|
|
|
|
if cmd == "stats":
|
|
print_db_stats()
|
|
|
|
elif cmd == "anomalies":
|
|
df = query_db("SELECT * FROM anomalies_history ORDER BY id DESC")
|
|
if df is not None:
|
|
print("\n🚨 РЕЕСТР НАЙДЕННЫХ АНОМАЛИЙ СКУД ⟷ 1С:")
|
|
print(df.to_string(index=False) if not df.empty else "Аномалий в базе пока нет.")
|
|
|
|
elif cmd == "rules":
|
|
df = query_db("SELECT id, added_by, created_at, rule_text FROM ai_knowledge_base ORDER BY id ASC")
|
|
if df is not None and not df.empty:
|
|
print("\n🧠 ПРАВИЛА БАЗЫ ЗНАНИЙ ИИ (ОБУЧЕНИЕ):")
|
|
print("=" * 80)
|
|
for _, row in df.iterrows():
|
|
print(f"📌 [ID: {row['id']}] | Добавил: {row['added_by']} | Дата: {row['created_at']}")
|
|
print(f"{row['rule_text']}")
|
|
print("-" * 80)
|
|
else:
|
|
print("\n🧠 База знаний пуста.")
|
|
|
|
elif cmd == "scud" and len(sys.argv) > 2:
|
|
date_str = sys.argv[2]
|
|
df = query_db("SELECT * FROM scud_logs WHERE log_date = ?", (date_str,))
|
|
if df is not None:
|
|
print(f"\n🚪 ЛОГИ СКУД ЗА {date_str} ({len(df)} записей):")
|
|
print(df.head(20).to_string(index=False))
|
|
|
|
elif cmd == "staff" and len(sys.argv) > 2:
|
|
date_str = sys.argv[2]
|
|
df = query_db("SELECT * FROM zup_staff WHERE snapshot_date = ?", (date_str,))
|
|
if df is not None:
|
|
print(f"\n📋 ШТАТНОЕ РАСПИСАНИЕ 1С ЗА {date_str} ({len(df)} сотрудников):")
|
|
print(df.head(20).to_string(index=False))
|
|
|
|
elif cmd == "absences" and len(sys.argv) > 2:
|
|
date_str = sys.argv[2]
|
|
df = query_db("SELECT * FROM zup_absences WHERE absence_date = ?", (date_str,))
|
|
if df is not None:
|
|
print(f"\n🏥 ДОКУМЕНТЫ ОТСУТСТВИЙ 1С ЗА {date_str} ({len(df)} записей):")
|
|
print(df.to_string(index=False) if not df.empty else f"Отсутствий за {date_str} не зарегистрировано.")
|
|
|
|
elif cmd == "dump":
|
|
out_path = sys.argv[2] if len(sys.argv) > 2 else "db_dump.xlsx"
|
|
if not os.path.exists(DB_PATH):
|
|
print(f"❌ Файл базы {DB_PATH} не найден.")
|
|
sys.exit(1)
|
|
|
|
conn = sqlite3.connect(DB_PATH)
|
|
tables = ["scud_logs", "zup_staff", "zup_absences", "anomalies_history", "ai_knowledge_base"]
|
|
|
|
with pd.ExcelWriter(out_path, engine='openpyxl') as writer:
|
|
for table in tables:
|
|
df = pd.read_sql_query(f"SELECT * FROM {table}", conn)
|
|
df.to_excel(writer, sheet_name=table, index=False)
|
|
conn.close()
|
|
print(f"✅ Вся база данных успешно экспортирована в Excel-файл: {out_path}")
|
|
|
|
else:
|
|
print(f"❌ Неизвестная команда: {cmd}") |