"""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()