209 lines
9.0 KiB
Python
209 lines
9.0 KiB
Python
"""Read-only проверки файловых SQLite для подсказок LLM (без произвольного SQL из чата)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import sqlite3
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from flask import Flask
|
||
|
||
from app.services.admin_dashboard_service import sqlite_bind_paths
|
||
|
||
|
||
def _table_exists(conn: sqlite3.Connection, table: str) -> bool:
|
||
cur = conn.execute(
|
||
"SELECT 1 FROM sqlite_master WHERE type='table' AND name=? LIMIT 1",
|
||
(table,),
|
||
)
|
||
return cur.fetchone() is not None
|
||
|
||
|
||
def _open_sqlite_readonly(path: Path) -> sqlite3.Connection:
|
||
uri = path.as_uri() + "?mode=ro"
|
||
return sqlite3.connect(uri, uri=True, timeout=12.0)
|
||
|
||
|
||
def _pragma_integrity(conn: sqlite3.Connection) -> Dict[str, Any]:
|
||
rows = conn.execute("PRAGMA integrity_check").fetchall()
|
||
msgs = [str(r[0]) for r in rows if r and r[0] is not None]
|
||
ok = len(msgs) == 1 and str(msgs[0]).lower() == "ok"
|
||
return {"ok": ok, "messages": msgs[:25] if not ok else []}
|
||
|
||
|
||
def _pragma_foreign_keys(conn: sqlite3.Connection) -> Dict[str, Any]:
|
||
try:
|
||
rows = conn.execute("PRAGMA foreign_key_check").fetchall()
|
||
except sqlite3.OperationalError:
|
||
return {"count": 0, "samples": [], "note": "foreign_key_check недоступен"}
|
||
samples: List[str] = []
|
||
for r in rows[:8]:
|
||
samples.append("|".join(str(x) for x in r))
|
||
return {"count": len(rows), "samples": samples}
|
||
|
||
|
||
def _domain_recipes_bind(conn: sqlite3.Connection) -> Dict[str, Any]:
|
||
"""Типичные логические проблемы в основной БД (recipes)."""
|
||
out: Dict[str, Any] = {}
|
||
if _table_exists(conn, "loading_report"):
|
||
try:
|
||
n = conn.execute(
|
||
"SELECT COUNT(*) FROM loading_report WHERE "
|
||
"recipe_id IS NULL OR length(recipe_id) != 36"
|
||
).fetchone()[0]
|
||
out["loading_report_bad_recipe_id_count"] = int(n)
|
||
except sqlite3.Error as e:
|
||
out["loading_report_check_error"] = str(e)[:200]
|
||
if _table_exists(conn, "unloading_report"):
|
||
try:
|
||
n = conn.execute(
|
||
"SELECT COUNT(*) FROM unloading_report WHERE "
|
||
"recipe_id IS NULL OR length(recipe_id) != 36"
|
||
).fetchone()[0]
|
||
out["unloading_report_bad_recipe_id_count"] = int(n)
|
||
except sqlite3.Error as e:
|
||
out["unloading_report_check_error"] = str(e)[:200]
|
||
if _table_exists(conn, "sync_queue"):
|
||
try:
|
||
n = conn.execute(
|
||
"SELECT COUNT(*) FROM sync_queue WHERE "
|
||
"(COALESCE(is_deleted, 0) = 0) AND status = 'failed'"
|
||
).fetchone()[0]
|
||
out["sync_queue_failed_count"] = int(n)
|
||
except sqlite3.Error as e:
|
||
out["sync_queue_check_error"] = str(e)[:200]
|
||
return out
|
||
|
||
|
||
def _one_bind_health(bind_name: str, path: Path) -> Dict[str, Any]:
|
||
entry: Dict[str, Any] = {"bind": bind_name, "path": str(path)}
|
||
if not path.is_file():
|
||
entry["error"] = "файл не найден"
|
||
return entry
|
||
try:
|
||
entry["size_bytes"] = path.stat().st_size
|
||
except OSError as e:
|
||
entry["error"] = str(e)
|
||
return entry
|
||
try:
|
||
conn = _open_sqlite_readonly(path)
|
||
except sqlite3.Error as e:
|
||
entry["error"] = f"не удалось открыть read-only: {e}"
|
||
return entry
|
||
try:
|
||
ver = conn.execute("SELECT sqlite_version()").fetchone()
|
||
entry["sqlite_version"] = str(ver[0]) if ver else "?"
|
||
entry["integrity"] = _pragma_integrity(conn)
|
||
entry["foreign_keys"] = _pragma_foreign_keys(conn)
|
||
try:
|
||
ntbl = conn.execute(
|
||
"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'"
|
||
).fetchone()[0]
|
||
entry["user_table_count"] = int(ntbl)
|
||
except sqlite3.Error:
|
||
pass
|
||
if bind_name == "recipes":
|
||
entry["domain_checks"] = _domain_recipes_bind(conn)
|
||
finally:
|
||
conn.close()
|
||
return entry
|
||
|
||
|
||
def build_llm_db_health_payload(app: Flask) -> Dict[str, Any]:
|
||
"""Структура для API и для текста LLM."""
|
||
paths = sqlite_bind_paths(app)
|
||
if not paths:
|
||
return {
|
||
"binds": {},
|
||
"note": "Нет файловых баз SQLite в конфигурации (или только :memory:) — PRAGMA-проверки не запускались.",
|
||
}
|
||
binds_out: Dict[str, Any] = {}
|
||
for name, pth in sorted(paths.items(), key=lambda x: x[0]):
|
||
binds_out[name] = _one_bind_health(name, pth)
|
||
return {"binds": binds_out}
|
||
|
||
|
||
def build_llm_db_health_block(app: Flask) -> str:
|
||
"""Текст на русском для system: что проверено и что похоже на ошибку."""
|
||
payload = build_llm_db_health_payload(app)
|
||
if payload.get("note"):
|
||
return (
|
||
"Проверка баз данных SQLite\n"
|
||
+ str(payload["note"])
|
||
)
|
||
lines: List[str] = [
|
||
"Проверка баз данных SQLite (только чтение: PRAGMA integrity_check, foreign_key_check, несколько эвристик).",
|
||
"Ниже не полный аудит — ориентир для диагностики; исправления делает администратор (SQL/админка/миграции).",
|
||
"",
|
||
]
|
||
for bind_name in sorted(payload.get("binds", {}).keys()):
|
||
b = payload["binds"][bind_name]
|
||
title = f"База «{bind_name}»"
|
||
if b.get("error"):
|
||
lines.append(f"- {title}: ошибка — {b['error']}")
|
||
lines.append("")
|
||
continue
|
||
lines.append(f"- {title}: файл {b.get('path', '—')}")
|
||
if b.get("size_bytes") is not None:
|
||
sz = int(b["size_bytes"])
|
||
lines.append(f" размер файла: {sz} байт")
|
||
if b.get("sqlite_version"):
|
||
lines.append(f" SQLite: {b['sqlite_version']}")
|
||
if b.get("user_table_count") is not None:
|
||
lines.append(f" пользовательских таблиц: {b['user_table_count']}")
|
||
integ = b.get("integrity") or {}
|
||
if integ.get("ok") is True:
|
||
lines.append(" PRAGMA integrity_check: ok")
|
||
else:
|
||
lines.append(" PRAGMA integrity_check: ПРОБЛЕМА")
|
||
for m in integ.get("messages") or ["(нет деталей)"]:
|
||
lines.append(f" · {m}")
|
||
fk = b.get("foreign_keys") or {}
|
||
fc = int(fk.get("count") or 0)
|
||
if fc == 0:
|
||
lines.append(" PRAGMA foreign_key_check: нарушений не найдено (0)")
|
||
else:
|
||
lines.append(f" PRAGMA foreign_key_check: найдено нарушений: {fc} (проверьте ссылки между таблицами)")
|
||
for s in fk.get("samples") or []:
|
||
lines.append(f" · {s}")
|
||
dom = b.get("domain_checks") or {}
|
||
if dom:
|
||
if "loading_report_bad_recipe_id_count" in dom:
|
||
n = dom["loading_report_bad_recipe_id_count"]
|
||
if n > 0:
|
||
lines.append(
|
||
f" Эвристика: в loading_report строк с recipe_id не UUID (NULL или длина≠36): {n}"
|
||
)
|
||
else:
|
||
lines.append(
|
||
" Эвристика: loading_report — подозрительных recipe_id (NULL / не UUID) не найдено"
|
||
)
|
||
if "unloading_report_bad_recipe_id_count" in dom:
|
||
n = dom["unloading_report_bad_recipe_id_count"]
|
||
if n > 0:
|
||
lines.append(
|
||
f" Эвристика: в unloading_report строк с recipe_id не UUID: {n}"
|
||
)
|
||
else:
|
||
lines.append(
|
||
" Эвристика: unloading_report — подозрительных recipe_id не найдено"
|
||
)
|
||
if "sync_queue_failed_count" in dom:
|
||
n = dom["sync_queue_failed_count"]
|
||
if n > 0:
|
||
lines.append(
|
||
f" Эвристика: sync_queue с status=failed (не удалённые): {n}"
|
||
)
|
||
else:
|
||
lines.append(" Эвристика: sync_queue — задач со status=failed не найдено")
|
||
lines.append("")
|
||
lines.append(
|
||
"Как отвечать: если пользователь спрашивает про ошибки в базе — опирайся на этот блок; "
|
||
"явно различай повреждение файла (integrity), битые FK (foreign_key_check) и логические эвристики (UUID, очередь sync)."
|
||
)
|
||
text = "\n".join(lines).strip()
|
||
max_len = 5500
|
||
if len(text) > max_len:
|
||
return text[: max_len - 1] + "…"
|
||
return text
|