uni-notify/app/db.py

"""SQLite-хранилище событий (с полнотекстовым поиском FTS5) и key-value настроек."""
from __future__ import annotations

import json
import re
import sqlite3
import threading
import time
from dataclasses import dataclass
from pathlib import Path

SCHEMA = """
CREATE TABLE IF NOT EXISTS items (
    id          INTEGER PRIMARY KEY,
    source      TEXT NOT NULL,          -- mail | teams
    kind        TEXT NOT NULL,          -- notification | row
    key         TEXT NOT NULL,
    sender      TEXT,
    title       TEXT,
    body        TEXT,
    created_at  REAL NOT NULL,
    level       INTEGER,                -- 0..3, NULL = ещё не оценено
    summary     TEXT,
    status      TEXT NOT NULL DEFAULT 'new',  -- new | sent | digest | digested | skipped | baseline | dup
    UNIQUE(source, key)
);
CREATE INDEX IF NOT EXISTS items_created ON items(created_at);

CREATE VIRTUAL TABLE IF NOT EXISTS items_fts USING fts5(
    sender, title, body, summary, content='items', content_rowid='id', tokenize='unicode61'
);
CREATE TRIGGER IF NOT EXISTS items_ai AFTER INSERT ON items BEGIN
    INSERT INTO items_fts(rowid, sender, title, body, summary)
    VALUES (new.id, new.sender, new.title, new.body, new.summary);
END;
CREATE TRIGGER IF NOT EXISTS items_au AFTER UPDATE ON items BEGIN
    INSERT INTO items_fts(items_fts, rowid, sender, title, body, summary)
    VALUES ('delete', old.id, old.sender, old.title, old.body, old.summary);
    INSERT INTO items_fts(rowid, sender, title, body, summary)
    VALUES (new.id, new.sender, new.title, new.body, new.summary);
END;

CREATE TABLE IF NOT EXISTS kv (key TEXT PRIMARY KEY, value TEXT NOT NULL);
"""


@dataclass
class Item:
    id: int
    source: str
    kind: str
    key: str
    sender: str
    title: str
    body: str
    created_at: float
    level: int | None
    summary: str | None
    status: str

    def short(self, n: int = 300) -> str:
        parts = [p for p in (self.title, self.body) if p]
        s = " — ".join(parts)
        return s if len(s) <= n else s[: n - 1] + "…"


class DB:
    def __init__(self, path: Path):
        self._lock = threading.Lock()
        self.conn = sqlite3.connect(str(path), check_same_thread=False)
        self.conn.row_factory = sqlite3.Row
        self.conn.execute("PRAGMA journal_mode=WAL")
        self.conn.executescript(SCHEMA)
        self.conn.commit()

    # ---------- items ----------
    def insert(self, source: str, kind: str, key: str, sender: str, title: str, body: str,
               status: str = "new") -> int | None:
        """Возвращает id новой записи или None, если такое событие уже было."""
        with self._lock:
            cur = self.conn.execute(
                "INSERT OR IGNORE INTO items(source, kind, key, sender, title, body, created_at, status)"
                " VALUES (?,?,?,?,?,?,?,?)",
                (source, kind, key, sender, title, body, time.time(), status),
            )
            self.conn.commit()
            return cur.lastrowid if cur.rowcount else None

    def update(self, item_id: int, **fields) -> None:
        if not fields:
            return
        cols = ", ".join(f"{k}=?" for k in fields)
        with self._lock:
            self.conn.execute(f"UPDATE items SET {cols} WHERE id=?", (*fields.values(), item_id))
            self.conn.commit()

    def _rows(self, sql: str, args: tuple = ()) -> list[Item]:
        with self._lock:
            return [Item(**dict(r)) for r in self.conn.execute(sql, args).fetchall()]

    def recent(self, limit: int = 20, min_level: int = 0) -> list[Item]:
        return self._rows(
            "SELECT * FROM items WHERE status NOT IN ('baseline','dup') AND COALESCE(level,0) >= ?"
            " ORDER BY created_at DESC LIMIT ?",
            (min_level, limit),
        )

    def digest_pending(self) -> list[Item]:
        return self._rows("SELECT * FROM items WHERE status='digest' ORDER BY created_at")

    def search(self, query: str, limit: int = 25) -> list[Item]:
        words = [w for w in re.findall(r"\w{3,}", query.lower())][:12]
        if not words:
            return []
        # префиксный поиск по каждому слову, OR — ранжирует bm25
        fts_q = " OR ".join(f'"{w}"*' for w in words)
        try:
            return self._rows(
                "SELECT items.* FROM items_fts JOIN items ON items.id = items_fts.rowid"
                " WHERE items_fts MATCH ? AND items.status NOT IN ('dup')"
                " ORDER BY bm25(items_fts) LIMIT ?",
                (fts_q, limit),
            )
        except sqlite3.OperationalError:
            return []

    def stats(self) -> dict:
        with self._lock:
            day_ago = time.time() - 86400
            r = self.conn.execute(
                "SELECT source, status, COUNT(*) c FROM items WHERE created_at > ? GROUP BY source, status",
                (day_ago,),
            ).fetchall()
            last = self.conn.execute(
                "SELECT source, MAX(created_at) t FROM items GROUP BY source"
            ).fetchall()
        out: dict = {"day": {}, "last": {x["source"]: x["t"] for x in last}}
        for x in r:
            out["day"].setdefault(x["source"], {})[x["status"]] = x["c"]
        return out

    def count_source(self, source: str) -> int:
        with self._lock:
            return self.conn.execute("SELECT COUNT(*) FROM items WHERE source=?", (source,)).fetchone()[0]

    # ---------- kv ----------
    def get(self, key: str, default=None):
        with self._lock:
            r = self.conn.execute("SELECT value FROM kv WHERE key=?", (key,)).fetchone()
        return json.loads(r[0]) if r else default

    def set(self, key: str, value) -> None:
        with self._lock:
            self.conn.execute(
                "INSERT INTO kv(key, value) VALUES(?, ?) ON CONFLICT(key) DO UPDATE SET value=excluded.value",
                (key, json.dumps(value, ensure_ascii=False)),
            )
            self.conn.commit()