416 lines
14 KiB
Python
416 lines
14 KiB
Python
"""Просмотр и точечное редактирование файловых SQLite для суперпользователя (админка)."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import sqlite3
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Tuple
|
||
|
||
from flask import Flask
|
||
|
||
from app.services.admin_dashboard_service import sqlite_bind_paths
|
||
|
||
_IDENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
|
||
|
||
|
||
def is_safe_sql_identifier(name: str) -> bool:
|
||
return bool(_IDENT.fullmatch(name or ""))
|
||
|
||
|
||
def _qi(ident: str) -> str:
|
||
if not _IDENT.fullmatch(ident or ""):
|
||
raise ValueError(f"Недопустимый идентификатор: {ident!r}")
|
||
return '"' + ident.replace('"', '""') + '"'
|
||
|
||
|
||
def _connect(path: Path) -> sqlite3.Connection:
|
||
conn = sqlite3.connect(str(path), timeout=30.0)
|
||
conn.row_factory = sqlite3.Row
|
||
conn.execute("PRAGMA foreign_keys = ON")
|
||
return conn
|
||
|
||
|
||
def sqlite_connection_for_bind(app: Flask, bind: str) -> Tuple[sqlite3.Connection, Path]:
|
||
paths = sqlite_bind_paths(app)
|
||
key = (bind or "").strip().lower()
|
||
if key not in paths:
|
||
raise FileNotFoundError(f"Неизвестный bind: {bind}")
|
||
p = paths[key]
|
||
if not p.is_file():
|
||
raise FileNotFoundError(f"Файл БД не найден: {p}")
|
||
return _connect(p), p
|
||
|
||
|
||
def table_has_rowid(conn: sqlite3.Connection, table: str) -> bool:
|
||
cur = conn.execute(
|
||
"SELECT sql FROM sqlite_master WHERE type='table' AND name=?",
|
||
(table,),
|
||
)
|
||
row = cur.fetchone()
|
||
if not row or not row[0]:
|
||
return True
|
||
return "WITHOUT ROWID" not in str(row[0]).upper()
|
||
|
||
|
||
def list_user_tables(conn: sqlite3.Connection) -> List[str]:
|
||
cur = conn.execute(
|
||
"""
|
||
SELECT name FROM sqlite_master
|
||
WHERE type='table' AND name NOT LIKE 'sqlite_%'
|
||
ORDER BY name COLLATE NOCASE
|
||
"""
|
||
)
|
||
return [str(r[0]) for r in cur.fetchall()]
|
||
|
||
|
||
def pragma_columns(conn: sqlite3.Connection, table: str) -> List[Dict[str, Any]]:
|
||
cur = conn.execute(f"PRAGMA table_info({_qi(table)})")
|
||
out: List[Dict[str, Any]] = []
|
||
for cid, name, col_type, notnull, default, pk in cur.fetchall():
|
||
ctype = (col_type or "").upper()
|
||
editable = "BLOB" not in ctype
|
||
out.append(
|
||
{
|
||
"cid": int(cid),
|
||
"name": str(name),
|
||
"type": str(col_type or ""),
|
||
"notnull": bool(notnull),
|
||
"default": default,
|
||
"pk": int(pk or 0),
|
||
"editable": editable,
|
||
}
|
||
)
|
||
return out
|
||
|
||
|
||
def fetch_table_page(
|
||
conn: sqlite3.Connection,
|
||
table: str,
|
||
*,
|
||
limit: int,
|
||
offset: int,
|
||
) -> Dict[str, Any]:
|
||
cols_meta = pragma_columns(conn, table)
|
||
col_names = [c["name"] for c in cols_meta]
|
||
has_rowid = table_has_rowid(conn, table)
|
||
if has_rowid:
|
||
q = f"SELECT rowid AS _wesp_rowid, * FROM {_qi(table)} LIMIT ? OFFSET ?"
|
||
else:
|
||
q = f"SELECT * FROM {_qi(table)} LIMIT ? OFFSET ?"
|
||
cur = conn.execute(q, (limit, offset))
|
||
rows_out: List[Dict[str, Any]] = []
|
||
for row in cur.fetchall():
|
||
d = dict(zip(row.keys(), row))
|
||
rows_out.append(d)
|
||
cur_total = conn.execute(f"SELECT COUNT(*) FROM {_qi(table)}")
|
||
total = int(cur_total.fetchone()[0])
|
||
pk_names = [c["name"] for c in cols_meta if c["pk"]]
|
||
return {
|
||
"columns": cols_meta,
|
||
"column_names": col_names,
|
||
"rows": rows_out,
|
||
"total": total,
|
||
"has_rowid": has_rowid,
|
||
"pk_names": pk_names,
|
||
}
|
||
|
||
|
||
def _normalize_incoming_value(val: Any) -> Any:
|
||
if val is None:
|
||
return None
|
||
if isinstance(val, (int, float, str, bool)):
|
||
return val
|
||
return str(val)
|
||
|
||
|
||
def update_row_cells(
|
||
conn: sqlite3.Connection,
|
||
table: str,
|
||
row_key: Dict[str, Any],
|
||
changes: Dict[str, Any],
|
||
) -> int:
|
||
if not changes:
|
||
return 0
|
||
cols_meta = pragma_columns(conn, table)
|
||
col_by_name = {c["name"]: c for c in cols_meta}
|
||
has_rowid = table_has_rowid(conn, table)
|
||
|
||
set_parts: List[str] = []
|
||
set_vals: List[Any] = []
|
||
for col_name, raw in changes.items():
|
||
if not _IDENT.fullmatch(col_name or ""):
|
||
raise ValueError(f"Недопустимое имя столбца: {col_name!r}")
|
||
meta = col_by_name.get(col_name)
|
||
if meta is None:
|
||
raise ValueError(f"Нет столбца {col_name!r} в таблице {table!r}")
|
||
if not meta.get("editable", True):
|
||
raise ValueError(f"Столбец {col_name!r} нельзя править через админку (BLOB)")
|
||
set_parts.append(f"{_qi(col_name)} = ?")
|
||
set_vals.append(_normalize_incoming_value(raw))
|
||
|
||
where_sql: str
|
||
where_args: List[Any]
|
||
if "_wesp_rowid" in row_key and has_rowid:
|
||
where_sql = "rowid = ?"
|
||
where_args = [int(row_key["_wesp_rowid"])]
|
||
else:
|
||
pk_names = [c["name"] for c in cols_meta if c["pk"]]
|
||
if not pk_names:
|
||
raise ValueError(
|
||
"Для таблицы без rowid нужен первичный ключ; укажите все PK-поля в row_key"
|
||
)
|
||
parts: List[str] = []
|
||
where_args = []
|
||
for pk in pk_names:
|
||
if pk not in row_key:
|
||
raise ValueError(f"В row_key нет PK-поля {pk!r}")
|
||
pv = row_key[pk]
|
||
if pv is None:
|
||
parts.append(f"{_qi(pk)} IS NULL")
|
||
else:
|
||
parts.append(f"{_qi(pk)} = ?")
|
||
where_args.append(_normalize_incoming_value(pv))
|
||
where_sql = " AND ".join(parts)
|
||
|
||
sql = f"UPDATE {_qi(table)} SET {', '.join(set_parts)} WHERE {where_sql}"
|
||
args = [*set_vals, *where_args]
|
||
cur = conn.execute(sql, args)
|
||
return int(cur.rowcount)
|
||
|
||
|
||
def _where_clause_for_row_key(
|
||
conn: sqlite3.Connection,
|
||
table: str,
|
||
row_key: Dict[str, Any],
|
||
cols_meta: List[Dict[str, Any]],
|
||
) -> Tuple[str, List[Any]]:
|
||
has_rowid = table_has_rowid(conn, table)
|
||
if "_wesp_rowid" in row_key and has_rowid:
|
||
return "rowid = ?", [int(row_key["_wesp_rowid"])]
|
||
pk_names = [c["name"] for c in cols_meta if c["pk"]]
|
||
if not pk_names:
|
||
raise ValueError(
|
||
"Для таблицы без rowid нужен первичный ключ в row_key"
|
||
)
|
||
parts: List[str] = []
|
||
args: List[Any] = []
|
||
for pk in pk_names:
|
||
if pk not in row_key:
|
||
raise ValueError(f"В row_key нет PK-поля {pk!r}")
|
||
pv = row_key[pk]
|
||
if pv is None:
|
||
parts.append(f"{_qi(pk)} IS NULL")
|
||
else:
|
||
parts.append(f"{_qi(pk)} = ?")
|
||
args.append(_normalize_incoming_value(pv))
|
||
return " AND ".join(parts), args
|
||
|
||
|
||
def fetch_row_by_key(
|
||
conn: sqlite3.Connection,
|
||
table: str,
|
||
row_key: Dict[str, Any],
|
||
) -> Dict[str, Any]:
|
||
cols_meta = pragma_columns(conn, table)
|
||
wh, args = _where_clause_for_row_key(conn, table, row_key, cols_meta)
|
||
has_rowid = table_has_rowid(conn, table)
|
||
if has_rowid:
|
||
q = f"SELECT rowid AS _wesp_rowid, * FROM {_qi(table)} WHERE {wh}"
|
||
else:
|
||
q = f"SELECT * FROM {_qi(table)} WHERE {wh}"
|
||
cur = conn.execute(q, args)
|
||
row = cur.fetchone()
|
||
if not row:
|
||
raise ValueError("Строка не найдена")
|
||
return dict(zip(row.keys(), row))
|
||
|
||
|
||
def row_key_from_row(
|
||
conn: sqlite3.Connection,
|
||
table: str,
|
||
row: Dict[str, Any],
|
||
) -> Dict[str, Any]:
|
||
cols_meta = pragma_columns(conn, table)
|
||
has_rowid = table_has_rowid(conn, table)
|
||
rk: Dict[str, Any] = {}
|
||
if has_rowid and "_wesp_rowid" in row:
|
||
rk["_wesp_rowid"] = row["_wesp_rowid"]
|
||
for c in cols_meta:
|
||
if c["pk"] and c["name"] in row:
|
||
rk[c["name"]] = row[c["name"]]
|
||
return rk
|
||
|
||
|
||
def _incoming_fk_groups(conn: sqlite3.Connection, parent_table: str) -> List[Dict[str, Any]]:
|
||
"""Все FK в дочерних таблицах, ссылающиеся на parent_table (с учётом составных ключей)."""
|
||
out: List[Dict[str, Any]] = []
|
||
for child in list_user_tables(conn):
|
||
cur = conn.execute(f"PRAGMA foreign_key_list({_qi(child)})")
|
||
groups: Dict[int, Dict[str, Any]] = {}
|
||
for r in cur.fetchall():
|
||
ref_parent = str(r[2])
|
||
if ref_parent != parent_table:
|
||
continue
|
||
rid = int(r[0])
|
||
on_delete = str(r[6]) if len(r) > 6 else "NO ACTION"
|
||
if rid not in groups:
|
||
groups[rid] = {
|
||
"child_table": child,
|
||
"pairs": [], # (child_col, parent_col)
|
||
"on_delete": on_delete,
|
||
}
|
||
groups[rid]["pairs"].append((str(r[3]), str(r[4])))
|
||
for g in groups.values():
|
||
g["pairs"].sort(key=lambda p: p[0])
|
||
out.append(g)
|
||
return out
|
||
|
||
|
||
def _select_child_rows(
|
||
conn: sqlite3.Connection,
|
||
child_table: str,
|
||
pairs: List[Tuple[str, str]],
|
||
parent_row: Dict[str, Any],
|
||
) -> List[Dict[str, Any]]:
|
||
wheres: List[str] = []
|
||
args: List[Any] = []
|
||
for child_col, parent_col in pairs:
|
||
pv = parent_row.get(parent_col)
|
||
if pv is None:
|
||
wheres.append(f"{_qi(child_col)} IS NULL")
|
||
else:
|
||
wheres.append(f"{_qi(child_col)} = ?")
|
||
args.append(_normalize_incoming_value(pv))
|
||
has_rowid = table_has_rowid(conn, child_table)
|
||
if has_rowid:
|
||
q = f"SELECT rowid AS _wesp_rowid, * FROM {_qi(child_table)} WHERE {' AND '.join(wheres)}"
|
||
else:
|
||
q = f"SELECT * FROM {_qi(child_table)} WHERE {' AND '.join(wheres)}"
|
||
cur = conn.execute(q, args)
|
||
return [dict(zip(r.keys(), r)) for r in cur.fetchall()]
|
||
|
||
|
||
def build_cascade_delete_plan(
|
||
conn: sqlite3.Connection,
|
||
root_table: str,
|
||
root_row_key: Dict[str, Any],
|
||
) -> List[Dict[str, Any]]:
|
||
"""
|
||
Порядок: сначала потомки (транзитивно), в конце корневая строка.
|
||
Каждый элемент: {"table", "row_key", "on_delete_hints": [...]}.
|
||
"""
|
||
incoming_cache: Dict[str, List[Dict[str, Any]]] = {}
|
||
|
||
def incoming(pt: str) -> List[Dict[str, Any]]:
|
||
if pt not in incoming_cache:
|
||
incoming_cache[pt] = _incoming_fk_groups(conn, pt)
|
||
return incoming_cache[pt]
|
||
|
||
order: List[Dict[str, Any]] = []
|
||
visited: set[str] = set()
|
||
|
||
def freeze(table: str, rk: Dict[str, Any]) -> str:
|
||
return f"{table}\0{json.dumps(rk, sort_keys=True, default=str, ensure_ascii=False)}"
|
||
|
||
def dfs(table: str, row_key: Dict[str, Any]) -> None:
|
||
sig = freeze(table, row_key)
|
||
if sig in visited:
|
||
return
|
||
try:
|
||
parent_row = fetch_row_by_key(conn, table, row_key)
|
||
except ValueError:
|
||
return
|
||
visited.add(sig)
|
||
|
||
hints: List[str] = []
|
||
for g in incoming(table):
|
||
child = g["child_table"]
|
||
pairs = g["pairs"]
|
||
on_del = (g.get("on_delete") or "NO ACTION").upper()
|
||
children = _select_child_rows(conn, child, pairs, parent_row)
|
||
if children and on_del not in ("NO ACTION", ""):
|
||
hints.append(f"{child}: ON DELETE {on_del}")
|
||
for crow in children:
|
||
crk = row_key_from_row(conn, child, crow)
|
||
dfs(child, crk)
|
||
|
||
order.append(
|
||
{
|
||
"table": table,
|
||
"row_key": row_key,
|
||
"on_delete_hints": hints,
|
||
}
|
||
)
|
||
|
||
dfs(root_table, root_row_key)
|
||
return order
|
||
|
||
|
||
def summarize_delete_plan(plan: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||
by_table: Dict[str, int] = {}
|
||
for item in plan:
|
||
t = item["table"]
|
||
by_table[t] = by_table.get(t, 0) + 1
|
||
lines: List[str] = []
|
||
for t, n in sorted(by_table.items()):
|
||
lines.append(f"• {t}: {n} строк")
|
||
items: List[Dict[str, Any]] = []
|
||
for item in plan[:200]:
|
||
rk = item["row_key"]
|
||
if "_wesp_rowid" in rk:
|
||
label = f"rowid={rk['_wesp_rowid']}"
|
||
else:
|
||
label = ", ".join(f"{k}={v!r}" for k, v in sorted(rk.items()) if k != "_wesp_rowid")
|
||
items.append({"table": item["table"], "label": label})
|
||
return {
|
||
"total": len(plan),
|
||
"by_table": by_table,
|
||
"lines": lines,
|
||
"items": items,
|
||
"truncated": len(plan) > 200,
|
||
}
|
||
|
||
|
||
def delete_row_by_key(
|
||
conn: sqlite3.Connection,
|
||
table: str,
|
||
row_key: Dict[str, Any],
|
||
) -> int:
|
||
cols_meta = pragma_columns(conn, table)
|
||
wh, args = _where_clause_for_row_key(conn, table, row_key, cols_meta)
|
||
sql = f"DELETE FROM {_qi(table)} WHERE {wh}"
|
||
cur = conn.execute(sql, args)
|
||
return int(cur.rowcount)
|
||
|
||
|
||
def execute_cascade_delete(
|
||
conn: sqlite3.Connection,
|
||
root_table: str,
|
||
root_row_key: Dict[str, Any],
|
||
) -> Dict[str, Any]:
|
||
plan = build_cascade_delete_plan(conn, root_table, root_row_key)
|
||
deleted = 0
|
||
for item in plan:
|
||
n = delete_row_by_key(conn, item["table"], item["row_key"])
|
||
deleted += n
|
||
return {"deleted": deleted, "plan_len": len(plan)}
|
||
|
||
|
||
def build_tree_payload(app: Flask) -> Dict[str, Any]:
|
||
paths = sqlite_bind_paths(app)
|
||
binds_out: List[Dict[str, Any]] = []
|
||
for bind_name in sorted(paths.keys()):
|
||
p = paths[bind_name]
|
||
if not p.is_file():
|
||
tables: List[str] = []
|
||
else:
|
||
conn = _connect(p)
|
||
try:
|
||
tables = list_user_tables(conn)
|
||
finally:
|
||
conn.close()
|
||
binds_out.append({"bind": bind_name, "tables": [{"name": t} for t in tables]})
|
||
return {"binds": binds_out}
|