| """ |
| feedback/pattern_store.py — SQLite CRUD for learned patterns and draft history |
| |
| Uses context manager for safe connection handling. All schema is |
| created on first init. Supports draft history and feedback logging. |
| """ |
|
|
| import sqlite3 |
| from contextlib import contextmanager |
| from datetime import datetime |
| from uuid import uuid4 |
| from pathlib import Path |
|
|
| from loguru import logger |
| from config import PATTERNS_DB_PATH |
|
|
|
|
| SCHEMA = """\ |
| CREATE TABLE IF NOT EXISTS learned_patterns ( |
| id TEXT PRIMARY KEY, |
| pattern TEXT NOT NULL, |
| doc_type TEXT, |
| source_diff TEXT, |
| created_at TEXT NOT NULL, |
| use_count INTEGER DEFAULT 0, |
| active INTEGER DEFAULT 1 |
| ); |
| |
| CREATE TABLE IF NOT EXISTS draft_history ( |
| draft_id TEXT PRIMARY KEY, |
| draft_type TEXT NOT NULL, |
| doc_ids TEXT NOT NULL, |
| draft_text TEXT NOT NULL, |
| validation_json TEXT, |
| created_at TEXT NOT NULL, |
| was_edited INTEGER DEFAULT 0 |
| ); |
| |
| CREATE TABLE IF NOT EXISTS feedback_log ( |
| id TEXT PRIMARY KEY, |
| draft_id TEXT NOT NULL, |
| original_text TEXT NOT NULL, |
| edited_text TEXT NOT NULL, |
| change_ratio REAL, |
| pattern_id TEXT, |
| created_at TEXT NOT NULL |
| ); |
| |
| CREATE INDEX IF NOT EXISTS idx_patterns_active |
| ON learned_patterns (active, doc_type); |
| """ |
|
|
|
|
| class PatternStore: |
| """SQLite-backed store for learned patterns and draft history.""" |
|
|
| def __init__(self, db_path: str | Path = PATTERNS_DB_PATH): |
| self.db_path = str(db_path) |
| Path(self.db_path).parent.mkdir(parents=True, exist_ok=True) |
| self._init_db() |
|
|
| @contextmanager |
| def _conn(self): |
| conn = sqlite3.connect(self.db_path) |
| conn.row_factory = sqlite3.Row |
| try: |
| yield conn |
| conn.commit() |
| except Exception: |
| conn.rollback() |
| raise |
| finally: |
| conn.close() |
|
|
| def _init_db(self): |
| with self._conn() as conn: |
| conn.executescript(SCHEMA) |
| logger.debug(f"PatternStore initialised: {self.db_path}") |
|
|
| |
|
|
| def save_pattern(self, pattern: str, doc_type: str, source_diff: str = "") -> str: |
| pid = str(uuid4()) |
| with self._conn() as conn: |
| conn.execute( |
| "INSERT INTO learned_patterns VALUES (?,?,?,?,?,?,?)", |
| (pid, pattern, doc_type, source_diff, |
| datetime.utcnow().isoformat(), 0, 1), |
| ) |
| logger.info(f"Pattern saved: [{pid[:8]}] {pattern}") |
| return pid |
|
|
| def update_pattern(self, pattern_id: str, new_text: str): |
| with self._conn() as conn: |
| conn.execute( |
| "UPDATE learned_patterns SET pattern=? WHERE id=?", |
| (new_text, pattern_id), |
| ) |
| logger.info(f"Pattern updated: [{pattern_id[:8]}]") |
|
|
| def deactivate_pattern(self, pattern_id: str): |
| with self._conn() as conn: |
| conn.execute( |
| "UPDATE learned_patterns SET active=0 WHERE id=?", |
| (pattern_id,), |
| ) |
|
|
| def delete_pattern(self, pattern_id: str): |
| with self._conn() as conn: |
| conn.execute("DELETE FROM learned_patterns WHERE id=?", (pattern_id,)) |
|
|
| def get_active_patterns( |
| self, doc_type: str | None = None, limit: int = 8 |
| ) -> list[tuple[str, str]]: |
| with self._conn() as conn: |
| if doc_type: |
| rows = conn.execute( |
| "SELECT id, pattern FROM learned_patterns " |
| "WHERE active=1 AND (doc_type=? OR doc_type IS NULL) " |
| "ORDER BY use_count DESC LIMIT ?", |
| (doc_type, limit), |
| ).fetchall() |
| else: |
| rows = conn.execute( |
| "SELECT id, pattern FROM learned_patterns " |
| "WHERE active=1 ORDER BY use_count DESC LIMIT ?", |
| (limit,), |
| ).fetchall() |
| return [(r["id"], r["pattern"]) for r in rows] |
|
|
| def get_all_patterns(self) -> list[dict]: |
| with self._conn() as conn: |
| rows = conn.execute( |
| "SELECT * FROM learned_patterns ORDER BY created_at DESC" |
| ).fetchall() |
| return [dict(r) for r in rows] |
|
|
| def increment_use_count(self, pattern_ids: list[str]): |
| with self._conn() as conn: |
| for pid in pattern_ids: |
| conn.execute( |
| "UPDATE learned_patterns SET use_count = use_count + 1 WHERE id=?", |
| (pid,), |
| ) |
|
|
| def pattern_count(self, active_only: bool = True) -> int: |
| with self._conn() as conn: |
| q = "SELECT COUNT(*) FROM learned_patterns" |
| if active_only: |
| q += " WHERE active=1" |
| return conn.execute(q).fetchone()[0] |
|
|
| |
|
|
| def save_draft(self, draft_id: str, draft_type: str, doc_ids: str, |
| draft_text: str, validation_json: str = ""): |
| with self._conn() as conn: |
| conn.execute( |
| "INSERT OR REPLACE INTO draft_history VALUES (?,?,?,?,?,?,?)", |
| (draft_id, draft_type, doc_ids, draft_text, validation_json, |
| datetime.utcnow().isoformat(), 0), |
| ) |
|
|
| def get_draft(self, draft_id: str) -> dict | None: |
| with self._conn() as conn: |
| row = conn.execute( |
| "SELECT * FROM draft_history WHERE draft_id=?", (draft_id,) |
| ).fetchone() |
| return dict(row) if row else None |
|
|
| def mark_draft_edited(self, draft_id: str): |
| with self._conn() as conn: |
| conn.execute( |
| "UPDATE draft_history SET was_edited=1 WHERE draft_id=?", |
| (draft_id,), |
| ) |
|
|
| |
|
|
| def log_feedback(self, draft_id: str, original: str, edited: str, |
| change_ratio: float, pattern_id: str = ""): |
| fid = str(uuid4()) |
| with self._conn() as conn: |
| conn.execute( |
| "INSERT INTO feedback_log VALUES (?,?,?,?,?,?,?)", |
| (fid, draft_id, original, edited, change_ratio, |
| pattern_id, datetime.utcnow().isoformat()), |
| ) |
|
|
