480 lines
17 KiB
Python
480 lines
17 KiB
Python
"""Хранилище (SQLite): сертификаты и шаблоны."""
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import sqlite3
|
||
import threading
|
||
from datetime import datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
|
||
from . import builtin, config
|
||
from .config import DB_PATH, TEMPLATES_DIR, ensure_dirs
|
||
|
||
_LOCK = threading.Lock()
|
||
|
||
SCHEMA = """
|
||
CREATE TABLE IF NOT EXISTS certificates (
|
||
seq INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
number TEXT NOT NULL,
|
||
template TEXT NOT NULL,
|
||
fio TEXT,
|
||
title TEXT,
|
||
body TEXT,
|
||
date_text TEXT,
|
||
fields TEXT,
|
||
png_path TEXT,
|
||
pdf_path TEXT,
|
||
recipient_email TEXT,
|
||
created_by INTEGER,
|
||
public_token TEXT,
|
||
created_at TEXT NOT NULL
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_cert_number ON certificates(number);
|
||
|
||
CREATE TABLE IF NOT EXISTS templates (
|
||
slug TEXT PRIMARY KEY,
|
||
title TEXT NOT NULL,
|
||
kind TEXT NOT NULL,
|
||
background TEXT,
|
||
layout TEXT,
|
||
html TEXT,
|
||
page TEXT,
|
||
defaults TEXT,
|
||
builtin INTEGER NOT NULL DEFAULT 0,
|
||
created_at TEXT NOT NULL
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS users (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
email TEXT NOT NULL UNIQUE COLLATE NOCASE,
|
||
password_hash TEXT NOT NULL,
|
||
name TEXT,
|
||
role TEXT NOT NULL DEFAULT 'user',
|
||
is_active INTEGER NOT NULL DEFAULT 1,
|
||
created_at TEXT NOT NULL
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS sessions (
|
||
token TEXT PRIMARY KEY,
|
||
user_id INTEGER NOT NULL,
|
||
created_at TEXT NOT NULL,
|
||
expires_at TEXT NOT NULL
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_sessions_user ON sessions(user_id);
|
||
|
||
CREATE TABLE IF NOT EXISTS meta (
|
||
key TEXT PRIMARY KEY,
|
||
value TEXT
|
||
);
|
||
"""
|
||
|
||
|
||
def _migrate(conn: sqlite3.Connection) -> None:
|
||
"""Добавить недостающие колонки в старые базы."""
|
||
cols = {r["name"] for r in conn.execute("PRAGMA table_info(certificates)").fetchall()}
|
||
for name, ddl in (("recipient_email", "TEXT"), ("created_by", "INTEGER"),
|
||
("public_token", "TEXT")):
|
||
if name not in cols:
|
||
conn.execute(f"ALTER TABLE certificates ADD COLUMN {name} {ddl}")
|
||
conn.execute("CREATE INDEX IF NOT EXISTS idx_cert_recipient ON certificates(recipient_email)")
|
||
conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS idx_cert_token ON certificates(public_token)")
|
||
|
||
|
||
def _connect() -> sqlite3.Connection:
|
||
conn = sqlite3.connect(DB_PATH, timeout=30)
|
||
conn.row_factory = sqlite3.Row
|
||
conn.execute("PRAGMA journal_mode=WAL")
|
||
return conn
|
||
|
||
|
||
def _now() -> str:
|
||
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
||
|
||
|
||
def init_db() -> None:
|
||
ensure_dirs()
|
||
with _LOCK, _connect() as conn:
|
||
conn.executescript(SCHEMA)
|
||
_migrate(conn)
|
||
builtin.ensure_assets()
|
||
ensure_builtin_templates()
|
||
|
||
|
||
# ---------------------------------------------------------------- templates
|
||
|
||
def _row_to_template(row: sqlite3.Row) -> dict:
|
||
return {
|
||
"slug": row["slug"],
|
||
"title": row["title"],
|
||
"kind": row["kind"],
|
||
"background": row["background"],
|
||
"layout": json.loads(row["layout"]) if row["layout"] else [],
|
||
"html": row["html"],
|
||
"page": json.loads(row["page"]) if row["page"] else None,
|
||
"defaults": json.loads(row["defaults"]) if row["defaults"] else {},
|
||
"builtin": bool(row["builtin"]),
|
||
"created_at": row["created_at"],
|
||
}
|
||
|
||
|
||
def ensure_builtin_templates() -> None:
|
||
with _LOCK, _connect() as conn:
|
||
row = conn.execute("SELECT value FROM meta WHERE key='builtin_version'").fetchone()
|
||
current = row["value"] if row else None
|
||
if current == builtin.BUILTIN_VERSION:
|
||
return
|
||
for tpl in builtin.builtin_templates():
|
||
upsert_template(
|
||
slug=tpl["slug"],
|
||
title=tpl["title"],
|
||
kind=tpl["kind"],
|
||
background=tpl.get("background"),
|
||
layout=tpl.get("layout"),
|
||
html=tpl.get("html"),
|
||
page=tpl.get("page"),
|
||
defaults=tpl.get("defaults"),
|
||
is_builtin=True,
|
||
)
|
||
with _LOCK, _connect() as conn:
|
||
conn.execute(
|
||
"INSERT INTO meta(key,value) VALUES('builtin_version',?) "
|
||
"ON CONFLICT(key) DO UPDATE SET value=excluded.value",
|
||
(builtin.BUILTIN_VERSION,),
|
||
)
|
||
|
||
|
||
def upsert_template(*, slug: str, title: str, kind: str, background=None,
|
||
layout=None, html=None, page=None, defaults=None,
|
||
is_builtin: bool = False) -> dict:
|
||
with _LOCK, _connect() as conn:
|
||
conn.execute(
|
||
"""INSERT INTO templates
|
||
(slug, title, kind, background, layout, html, page, defaults, builtin, created_at)
|
||
VALUES (?,?,?,?,?,?,?,?,?,?)
|
||
ON CONFLICT(slug) DO UPDATE SET
|
||
title=excluded.title, kind=excluded.kind, background=excluded.background,
|
||
layout=excluded.layout, html=excluded.html, page=excluded.page,
|
||
defaults=excluded.defaults, builtin=excluded.builtin""",
|
||
(
|
||
slug, title, kind, background,
|
||
json.dumps(layout, ensure_ascii=False) if layout is not None else None,
|
||
html,
|
||
json.dumps(page, ensure_ascii=False) if page is not None else None,
|
||
json.dumps(defaults or {}, ensure_ascii=False),
|
||
1 if is_builtin else 0, _now(),
|
||
),
|
||
)
|
||
return get_template(slug)
|
||
|
||
|
||
def get_template(slug: str) -> dict | None:
|
||
with _LOCK, _connect() as conn:
|
||
row = conn.execute("SELECT * FROM templates WHERE slug=?", (slug,)).fetchone()
|
||
return _row_to_template(row) if row else None
|
||
|
||
|
||
def list_templates() -> list[dict]:
|
||
with _LOCK, _connect() as conn:
|
||
rows = conn.execute("SELECT * FROM templates ORDER BY builtin DESC, slug").fetchall()
|
||
out = []
|
||
for row in rows:
|
||
t = _row_to_template(row)
|
||
t.pop("html", None) # не тянем тяжёлое поле в список
|
||
t.pop("layout", None)
|
||
out.append(t)
|
||
return out
|
||
|
||
|
||
def delete_template(slug: str) -> bool:
|
||
with _LOCK, _connect() as conn:
|
||
cur = conn.execute("DELETE FROM templates WHERE slug=? AND builtin=0", (slug,))
|
||
return cur.rowcount > 0
|
||
|
||
|
||
# ------------------------------------------------------------- certificates
|
||
|
||
def _format_number(seq: int, prefix: str | None) -> str:
|
||
base = f"{seq:0{config.NUMBER_PAD}d}"
|
||
prefix = (prefix or "").strip()
|
||
return f"{prefix}-{base}" if prefix else base
|
||
|
||
|
||
def reserve_certificate(*, template: str, values: dict, prefix: str | None = None,
|
||
number: int | str | None = None,
|
||
recipient_email: str | None = None,
|
||
created_by: int | None = None) -> dict:
|
||
"""Зарезервировать номер: вставить строку и вернуть её (файлы пока пустые).
|
||
|
||
По умолчанию номер берётся из общей нумерации (начиная с config.START_SEQ).
|
||
Если передан `number`, используется явный номер из зарезервированного
|
||
диапазона 1 .. START_SEQ-1 (для особых случаев).
|
||
"""
|
||
with _LOCK, _connect() as conn:
|
||
conn.execute("BEGIN IMMEDIATE")
|
||
if number is not None and str(number).strip():
|
||
try:
|
||
seq = int(str(number).strip())
|
||
except (TypeError, ValueError):
|
||
raise ValueError("Номер должен быть целым числом") from None
|
||
if seq < 1:
|
||
raise ValueError("Номер должен быть положительным")
|
||
if seq >= config.START_SEQ:
|
||
raise ValueError(
|
||
f"Явный номер допускается только из резерва 1..{config.START_SEQ - 1}"
|
||
)
|
||
if conn.execute("SELECT 1 FROM certificates WHERE seq=?", (seq,)).fetchone():
|
||
raise ValueError(f"Номер {seq} уже занят")
|
||
else:
|
||
row = conn.execute("SELECT MAX(seq) AS m FROM certificates").fetchone()
|
||
# обычная нумерация никогда не заходит в зарезервированный диапазон
|
||
seq = max(int(row["m"] or 0), config.START_SEQ - 1) + 1
|
||
number = _format_number(seq, prefix)
|
||
conn.execute(
|
||
"""INSERT INTO certificates
|
||
(seq, number, template, fio, title, body, date_text, fields,
|
||
png_path, pdf_path, recipient_email, created_by, created_at)
|
||
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)""",
|
||
(
|
||
seq, number, template,
|
||
values.get("fio"), values.get("title"), values.get("body"),
|
||
values.get("date"),
|
||
json.dumps(values.get("fields") or {}, ensure_ascii=False),
|
||
None, None,
|
||
(recipient_email or "").strip().lower() or None,
|
||
created_by, _now(),
|
||
),
|
||
)
|
||
return get_certificate(seq)
|
||
|
||
|
||
def update_certificate_files(seq: int, png_path: str, pdf_path: str) -> None:
|
||
with _LOCK, _connect() as conn:
|
||
conn.execute(
|
||
"UPDATE certificates SET png_path=?, pdf_path=? WHERE seq=?",
|
||
(png_path, pdf_path, seq),
|
||
)
|
||
|
||
|
||
def _row_to_cert(row: sqlite3.Row) -> dict:
|
||
return {
|
||
"seq": row["seq"],
|
||
"number": row["number"],
|
||
"template": row["template"],
|
||
"fio": row["fio"],
|
||
"title": row["title"],
|
||
"body": row["body"],
|
||
"date": row["date_text"],
|
||
"fields": json.loads(row["fields"]) if row["fields"] else {},
|
||
"png_path": row["png_path"],
|
||
"pdf_path": row["pdf_path"],
|
||
"recipient_email": row["recipient_email"],
|
||
"created_by": row["created_by"],
|
||
"public_token": row["public_token"],
|
||
"created_at": row["created_at"],
|
||
}
|
||
|
||
|
||
def get_certificate(ref: int | str) -> dict | None:
|
||
with _LOCK, _connect() as conn:
|
||
if isinstance(ref, int) or str(ref).isdigit():
|
||
row = conn.execute("SELECT * FROM certificates WHERE seq=?", (int(ref),)).fetchone()
|
||
else:
|
||
row = conn.execute("SELECT * FROM certificates WHERE number=?", (str(ref),)).fetchone()
|
||
return _row_to_cert(row) if row else None
|
||
|
||
|
||
def list_certificates(limit: int = 50, offset: int = 0,
|
||
recipient_email: str | None = None) -> list[dict]:
|
||
"""Если recipient_email задан — вернуть только документы этого получателя."""
|
||
with _LOCK, _connect() as conn:
|
||
if recipient_email:
|
||
rows = conn.execute(
|
||
"SELECT * FROM certificates WHERE lower(recipient_email)=lower(?) "
|
||
"ORDER BY seq DESC LIMIT ? OFFSET ?",
|
||
(recipient_email.strip(), limit, offset),
|
||
).fetchall()
|
||
else:
|
||
rows = conn.execute(
|
||
"SELECT * FROM certificates ORDER BY seq DESC LIMIT ? OFFSET ?",
|
||
(limit, offset),
|
||
).fetchall()
|
||
return [_row_to_cert(r) for r in rows]
|
||
|
||
|
||
def count_certificates(recipient_email: str | None = None) -> int:
|
||
with _LOCK, _connect() as conn:
|
||
if recipient_email:
|
||
row = conn.execute(
|
||
"SELECT COUNT(*) AS n FROM certificates WHERE lower(recipient_email)=lower(?)",
|
||
(recipient_email.strip(),),
|
||
).fetchone()
|
||
else:
|
||
row = conn.execute("SELECT COUNT(*) AS n FROM certificates").fetchone()
|
||
return int(row["n"])
|
||
|
||
|
||
def set_public_token(ref: int | str, token: str | None) -> dict | None:
|
||
cert = get_certificate(ref)
|
||
if not cert:
|
||
return None
|
||
with _LOCK, _connect() as conn:
|
||
conn.execute("UPDATE certificates SET public_token=? WHERE seq=?", (token, cert["seq"]))
|
||
return get_certificate(cert["seq"])
|
||
|
||
|
||
def get_certificate_by_token(token: str) -> dict | None:
|
||
if not token:
|
||
return None
|
||
with _LOCK, _connect() as conn:
|
||
row = conn.execute("SELECT * FROM certificates WHERE public_token=?", (token,)).fetchone()
|
||
return _row_to_cert(row) if row else None
|
||
|
||
|
||
def delete_certificate(ref: int | str) -> bool:
|
||
cert = get_certificate(ref)
|
||
if not cert:
|
||
return False
|
||
with _LOCK, _connect() as conn:
|
||
conn.execute("DELETE FROM certificates WHERE seq=?", (cert["seq"],))
|
||
for key in ("png_path", "pdf_path"):
|
||
p = cert.get(key)
|
||
if p:
|
||
Path(p).unlink(missing_ok=True)
|
||
return True
|
||
|
||
|
||
# -------------------------------------------------------------------- users
|
||
|
||
def _row_to_user(row: sqlite3.Row) -> dict:
|
||
return {
|
||
"id": row["id"],
|
||
"email": row["email"],
|
||
"name": row["name"],
|
||
"role": row["role"],
|
||
"is_active": bool(row["is_active"]),
|
||
"created_at": row["created_at"],
|
||
}
|
||
|
||
|
||
def create_user(email: str, password_hash: str, name: str | None = None,
|
||
role: str = "user") -> dict:
|
||
email = (email or "").strip().lower()
|
||
with _LOCK, _connect() as conn:
|
||
try:
|
||
cur = conn.execute(
|
||
"INSERT INTO users (email, password_hash, name, role, is_active, created_at) "
|
||
"VALUES (?,?,?,?,1,?)",
|
||
(email, password_hash, name, role, _now()),
|
||
)
|
||
except sqlite3.IntegrityError:
|
||
raise ValueError("Пользователь с таким email уже существует") from None
|
||
uid = cur.lastrowid
|
||
return get_user(uid)
|
||
|
||
|
||
def get_user(user_id: int) -> dict | None:
|
||
with _LOCK, _connect() as conn:
|
||
row = conn.execute("SELECT * FROM users WHERE id=?", (user_id,)).fetchone()
|
||
return _row_to_user(row) if row else None
|
||
|
||
|
||
def get_user_by_email(email: str) -> dict | None:
|
||
with _LOCK, _connect() as conn:
|
||
row = conn.execute("SELECT * FROM users WHERE lower(email)=lower(?)",
|
||
((email or "").strip(),)).fetchone()
|
||
return _row_to_user(row) if row else None
|
||
|
||
|
||
def get_password_hash(user_id: int) -> str | None:
|
||
with _LOCK, _connect() as conn:
|
||
row = conn.execute("SELECT password_hash FROM users WHERE id=?", (user_id,)).fetchone()
|
||
return row["password_hash"] if row else None
|
||
|
||
|
||
def list_users() -> list[dict]:
|
||
with _LOCK, _connect() as conn:
|
||
rows = conn.execute(
|
||
"SELECT * FROM users ORDER BY (role='admin') DESC, email"
|
||
).fetchall()
|
||
return [_row_to_user(r) for r in rows]
|
||
|
||
|
||
def count_users(role: str | None = None) -> int:
|
||
with _LOCK, _connect() as conn:
|
||
if role:
|
||
row = conn.execute("SELECT COUNT(*) AS n FROM users WHERE role=?", (role,)).fetchone()
|
||
else:
|
||
row = conn.execute("SELECT COUNT(*) AS n FROM users").fetchone()
|
||
return int(row["n"])
|
||
|
||
|
||
def update_user(user_id: int, *, role: str | None = None, is_active: bool | None = None,
|
||
name: str | None = None, password_hash: str | None = None) -> bool:
|
||
sets, params = [], []
|
||
if role is not None:
|
||
sets.append("role=?"); params.append(role)
|
||
if is_active is not None:
|
||
sets.append("is_active=?"); params.append(1 if is_active else 0)
|
||
if name is not None:
|
||
sets.append("name=?"); params.append(name)
|
||
if password_hash is not None:
|
||
sets.append("password_hash=?"); params.append(password_hash)
|
||
if not sets:
|
||
return False
|
||
params.append(user_id)
|
||
with _LOCK, _connect() as conn:
|
||
cur = conn.execute(f"UPDATE users SET {', '.join(sets)} WHERE id=?", params)
|
||
return cur.rowcount > 0
|
||
|
||
|
||
def delete_user(user_id: int) -> bool:
|
||
with _LOCK, _connect() as conn:
|
||
conn.execute("DELETE FROM sessions WHERE user_id=?", (user_id,))
|
||
cur = conn.execute("DELETE FROM users WHERE id=?", (user_id,))
|
||
return cur.rowcount > 0
|
||
|
||
|
||
# ----------------------------------------------------------------- sessions
|
||
|
||
def create_session(token: str, user_id: int, ttl_days: int = 30) -> None:
|
||
now = datetime.now(timezone.utc)
|
||
expires = now + timedelta(days=ttl_days)
|
||
with _LOCK, _connect() as conn:
|
||
conn.execute(
|
||
"INSERT INTO sessions (token, user_id, created_at, expires_at) VALUES (?,?,?,?)",
|
||
(token, user_id, now.isoformat(timespec="seconds"),
|
||
expires.isoformat(timespec="seconds")),
|
||
)
|
||
|
||
|
||
def get_session_user(token: str) -> dict | None:
|
||
if not token:
|
||
return None
|
||
with _LOCK, _connect() as conn:
|
||
row = conn.execute("SELECT * FROM sessions WHERE token=?", (token,)).fetchone()
|
||
if not row:
|
||
return None
|
||
if row["expires_at"] <= datetime.now(timezone.utc).isoformat(timespec="seconds"):
|
||
conn.execute("DELETE FROM sessions WHERE token=?", (token,))
|
||
return None
|
||
urow = conn.execute("SELECT * FROM users WHERE id=?", (row["user_id"],)).fetchone()
|
||
if not urow or not urow["is_active"]:
|
||
return None
|
||
return _row_to_user(urow)
|
||
|
||
|
||
def delete_session(token: str) -> None:
|
||
with _LOCK, _connect() as conn:
|
||
conn.execute("DELETE FROM sessions WHERE token=?", (token,))
|
||
|
||
|
||
def delete_user_sessions(user_id: int) -> None:
|
||
with _LOCK, _connect() as conn:
|
||
conn.execute("DELETE FROM sessions WHERE user_id=?", (user_id,))
|
||
|
||
|
||
def purge_expired_sessions() -> None:
|
||
with _LOCK, _connect() as conn:
|
||
conn.execute("DELETE FROM sessions WHERE expires_at <= ?",
|
||
(datetime.now(timezone.utc).isoformat(timespec="seconds"),))
|