CoRE.Vision/backend/app/db.py

"""SQLite-индекс архива (aiosqlite)."""
from __future__ import annotations

import os
import time

import aiosqlite

from .models import Recording


def _like_prefix(prefix: str) -> str:
    """LIKE-шаблон для путей внутри хранилища: '<path>/%' с экранированием %, _, \\."""
    p = prefix.rstrip("/") + "/"
    p = p.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
    return p + "%"


SCHEMA = """
CREATE TABLE IF NOT EXISTS recordings (
  id          INTEGER PRIMARY KEY AUTOINCREMENT,
  camera_id   TEXT NOT NULL,
  path        TEXT NOT NULL UNIQUE,
  started_at  INTEGER NOT NULL,
  ended_at    INTEGER,
  duration_s  REAL,
  size_bytes  INTEGER,
  created_at  INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_rec_cam_time ON recordings(camera_id, started_at);

CREATE TABLE IF NOT EXISTS events (
  id         INTEGER PRIMARY KEY AUTOINCREMENT,
  camera_id  TEXT NOT NULL,
  type       TEXT NOT NULL,
  ts         INTEGER NOT NULL,
  meta       TEXT
);
"""


class Database:
    def __init__(self, path: str | None = None):
        self.path = path or os.environ.get("NVR_DB", "/data/nvr.db")
        self._conn: aiosqlite.Connection | None = None

    async def connect(self) -> None:
        os.makedirs(os.path.dirname(self.path) or ".", exist_ok=True)
        self._conn = await aiosqlite.connect(self.path)
        self._conn.row_factory = aiosqlite.Row
        await self._conn.executescript(SCHEMA)
        await self._conn.commit()

    async def close(self) -> None:
        if self._conn:
            await self._conn.close()
            self._conn = None

    @property
    def conn(self) -> aiosqlite.Connection:
        if self._conn is None:
            raise RuntimeError("Database not connected")
        return self._conn

    # ── recordings ─────────────────────────────────────────────
    async def has_path(self, path: str) -> bool:
        cur = await self.conn.execute("SELECT 1 FROM recordings WHERE path = ?", (path,))
        return await cur.fetchone() is not None

    async def add_recording(
        self, camera_id: str, path: str, started_at: int,
        duration_s: float | None, size_bytes: int | None,
    ) -> None:
        ended_at = int(started_at + duration_s) if duration_s else None
        await self.conn.execute(
            """INSERT OR IGNORE INTO recordings
               (camera_id, path, started_at, ended_at, duration_s, size_bytes, created_at)
               VALUES (?, ?, ?, ?, ?, ?, ?)""",
            (camera_id, path, started_at, ended_at, duration_s, size_bytes, int(time.time())),
        )
        await self.conn.commit()

    async def list_recordings(
        self, camera_id: str | None, ts_from: int | None, ts_to: int | None,
        limit: int = 2000,
    ) -> list[Recording]:
        sql = "SELECT * FROM recordings WHERE 1=1"
        args: list = []
        if camera_id:
            sql += " AND camera_id = ?"
            args.append(camera_id)
        if ts_from is not None:
            sql += " AND started_at >= ?"
            args.append(ts_from)
        if ts_to is not None:
            sql += " AND started_at <= ?"
            args.append(ts_to)
        sql += " ORDER BY started_at ASC LIMIT ?"
        args.append(limit)
        cur = await self.conn.execute(sql, args)
        rows = await cur.fetchall()
        return [Recording(**dict(r)) for r in rows]

    async def get_recording(self, rec_id: int) -> Recording | None:
        cur = await self.conn.execute("SELECT * FROM recordings WHERE id = ?", (rec_id,))
        row = await cur.fetchone()
        return Recording(**dict(row)) if row else None

    async def oldest_recordings(self, limit: int = 50) -> list[Recording]:
        cur = await self.conn.execute(
            "SELECT * FROM recordings ORDER BY started_at ASC LIMIT ?", (limit,)
        )
        rows = await cur.fetchall()
        return [Recording(**dict(r)) for r in rows]

    # ── per-storage (по префиксу пути) — для квоты на каждое хранилище ──
    async def total_size_under(self, prefix: str) -> int:
        cur = await self.conn.execute(
            "SELECT COALESCE(SUM(size_bytes), 0) FROM recordings WHERE path LIKE ? ESCAPE '\\'",
            (_like_prefix(prefix),),
        )
        row = await cur.fetchone()
        return int(row[0]) if row else 0

    async def oldest_recordings_under(self, prefix: str, limit: int = 50) -> list[Recording]:
        cur = await self.conn.execute(
            "SELECT * FROM recordings WHERE path LIKE ? ESCAPE '\\' "
            "ORDER BY started_at ASC LIMIT ?",
            (_like_prefix(prefix), limit),
        )
        rows = await cur.fetchall()
        return [Recording(**dict(r)) for r in rows]

    async def recordings_before(self, ts: int) -> list[Recording]:
        cur = await self.conn.execute(
            "SELECT * FROM recordings WHERE started_at < ? ORDER BY started_at ASC", (ts,)
        )
        rows = await cur.fetchall()
        return [Recording(**dict(r)) for r in rows]

    async def delete_recording(self, rec_id: int) -> None:
        await self.conn.execute("DELETE FROM recordings WHERE id = ?", (rec_id,))
        await self.conn.commit()

    async def delete_missing(self, path: str) -> None:
        await self.conn.execute("DELETE FROM recordings WHERE path = ?", (path,))
        await self.conn.commit()

    async def total_size(self) -> int:
        cur = await self.conn.execute("SELECT COALESCE(SUM(size_bytes), 0) FROM recordings")
        row = await cur.fetchone()
        return int(row[0]) if row else 0

    async def known_paths(self) -> set[str]:
        cur = await self.conn.execute("SELECT path FROM recordings")
        rows = await cur.fetchall()
        return {r[0] for r in rows}