from __future__ import annotations # pyright: reportMissingImports=false """ WikiStore — SQLite 기반 InsuWiki 위키 데이터 저장소. 기본 DB 경로: /home/jay/projects/insuwiki/data/wiki.db 테스트에서는 :memory: 사용 가능. """ import json import logging import sqlite3 from datetime import datetime, timezone from pathlib import Path from typing import Any logger = logging.getLogger(__name__) _DEFAULT_DB_PATH = "/home/jay/projects/insuwiki/data/wiki.db" # --------------------------------------------------------------------------- # DDL # --------------------------------------------------------------------------- _CREATE_TABLE = """ CREATE TABLE IF NOT EXISTS wiki_entries ( id TEXT PRIMARY KEY, title TEXT NOT NULL, category TEXT NOT NULL, subcategory TEXT, question TEXT, answer TEXT, expert TEXT, source_date TEXT, source_chat TEXT, keywords TEXT, confidence TEXT, raw_thread TEXT, status TEXT DEFAULT 'draft', created_at TEXT, updated_at TEXT ); """ _CREATE_FTS = """ CREATE VIRTUAL TABLE IF NOT EXISTS wiki_fts USING fts5( title, question, answer, keywords, content=wiki_entries, content_rowid=rowid ); """ _CREATE_TRIGGER_AI = """ CREATE TRIGGER IF NOT EXISTS wiki_fts_ai AFTER INSERT ON wiki_entries BEGIN INSERT INTO wiki_fts(rowid, title, question, answer, keywords) VALUES (new.rowid, new.title, new.question, new.answer, new.keywords); END; """ _CREATE_TRIGGER_AD = """ CREATE TRIGGER IF NOT EXISTS wiki_fts_ad AFTER DELETE ON wiki_entries BEGIN INSERT INTO wiki_fts(wiki_fts, rowid, title, question, answer, keywords) VALUES ('delete', old.rowid, old.title, old.question, old.answer, old.keywords); END; """ _CREATE_TRIGGER_AU = """ CREATE TRIGGER IF NOT EXISTS wiki_fts_au AFTER UPDATE ON wiki_entries BEGIN INSERT INTO wiki_fts(wiki_fts, rowid, title, question, answer, keywords) VALUES ('delete', old.rowid, old.title, old.question, old.answer, old.keywords); INSERT INTO wiki_fts(rowid, title, question, answer, keywords) VALUES (new.rowid, new.title, new.question, new.answer, new.keywords); END; """ _COLUMNS = ( "id", "title", "category", "subcategory", "question", "answer", "expert", "source_date", "source_chat", "keywords", "confidence", "raw_thread", "status", "created_at", "updated_at", ) # --------------------------------------------------------------------------- # WikiStore # --------------------------------------------------------------------------- class WikiStore: """SQLite 기반 위키 항목 CRUD 저장소.""" def __init__(self, db_path: str | None = None) -> None: resolved = db_path if db_path is not None else _DEFAULT_DB_PATH if resolved != ":memory:": Path(resolved).parent.mkdir(parents=True, exist_ok=True) self._conn: sqlite3.Connection = sqlite3.connect( resolved, check_same_thread=False ) self._conn.row_factory = sqlite3.Row self.init_db() # ------------------------------------------------------------------ # 초기화 # ------------------------------------------------------------------ def init_db(self) -> None: """테이블, FTS 가상 테이블, 트리거를 생성한다.""" cur = self._conn.cursor() cur.executescript( _CREATE_TABLE + _CREATE_FTS + _CREATE_TRIGGER_AI + _CREATE_TRIGGER_AD + _CREATE_TRIGGER_AU ) self._conn.commit() # ------------------------------------------------------------------ # 헬퍼 # ------------------------------------------------------------------ def _now_iso(self) -> str: return datetime.now(timezone.utc).isoformat() def _row_to_dict(self, row: sqlite3.Row) -> dict[str, Any]: """sqlite3.Row → dict, keywords/raw_thread는 JSON 파싱.""" d = dict(row) for field in ("keywords", "raw_thread"): val = d.get(field) if isinstance(val, str): try: d[field] = json.loads(val) except (json.JSONDecodeError, ValueError): d[field] = [] elif val is None: d[field] = [] return d def _prepare_entry(self, entry: dict[str, Any]) -> dict[str, Any]: """keywords/raw_thread가 list면 json.dumps로 변환한 복사본 반환.""" prepared = dict(entry) for field in ("keywords", "raw_thread"): val = prepared.get(field) if isinstance(val, list): prepared[field] = json.dumps(val, ensure_ascii=False) return prepared # ------------------------------------------------------------------ # CRUD # ------------------------------------------------------------------ def insert_entry(self, entry: dict[str, Any]) -> str: """항목을 삽입하고 id를 반환한다.""" prepared = self._prepare_entry(entry) now = self._now_iso() prepared.setdefault("created_at", now) prepared.setdefault("updated_at", now) prepared.setdefault("status", "draft") cols = [c for c in _COLUMNS if c in prepared] placeholders = ", ".join("?" for _ in cols) values = [prepared[c] for c in cols] sql = f"INSERT INTO wiki_entries ({', '.join(cols)}) VALUES ({placeholders})" cur = self._conn.cursor() cur.execute(sql, values) self._conn.commit() return str(prepared["id"]) def get_entry(self, id: str) -> dict[str, Any] | None: """단건 조회. 없으면 None.""" cur = self._conn.cursor() cur.execute("SELECT * FROM wiki_entries WHERE id = ?", (id,)) row = cur.fetchone() return self._row_to_dict(row) if row else None def update_entry(self, id: str, updates: dict[str, Any]) -> bool: """부분 업데이트. 성공 여부 반환.""" if not updates: return False prepared = self._prepare_entry(updates) prepared["updated_at"] = self._now_iso() # id, created_at은 업데이트 금지 for protected in ("id", "created_at"): prepared.pop(protected, None) set_clause = ", ".join(f"{col} = ?" for col in prepared) values = list(prepared.values()) + [id] sql = f"UPDATE wiki_entries SET {set_clause} WHERE id = ?" cur = self._conn.cursor() cur.execute(sql, values) self._conn.commit() return cur.rowcount > 0 def delete_entry(self, id: str) -> bool: """항목 삭제. 성공 여부 반환.""" cur = self._conn.cursor() cur.execute("DELETE FROM wiki_entries WHERE id = ?", (id,)) self._conn.commit() return cur.rowcount > 0 def list_entries( self, category: str | None = None, status: str | None = None, limit: int = 50, offset: int = 0, ) -> list[dict[str, Any]]: """조건 필터링 목록 조회.""" conditions: list[str] = [] params: list[Any] = [] if category is not None: conditions.append("category = ?") params.append(category) if status is not None: conditions.append("status = ?") params.append(status) where = f"WHERE {' AND '.join(conditions)}" if conditions else "" sql = ( f"SELECT * FROM wiki_entries {where} " f"ORDER BY created_at DESC LIMIT ? OFFSET ?" ) params.extend([limit, offset]) cur = self._conn.cursor() cur.execute(sql, params) return [self._row_to_dict(row) for row in cur.fetchall()] def search_entries(self, query: str, limit: int = 20) -> list[dict[str, Any]]: """FTS5 전문 검색.""" sql = """ SELECT e.* FROM wiki_entries e JOIN wiki_fts f ON e.rowid = f.rowid WHERE wiki_fts MATCH ? ORDER BY rank LIMIT ? """ cur = self._conn.cursor() cur.execute(sql, (query, limit)) return [self._row_to_dict(row) for row in cur.fetchall()] def approve_entry(self, id: str) -> bool: """status를 'approved'로 변경.""" return self.update_entry(id, {"status": "approved"}) def reject_entry(self, id: str) -> bool: """status를 'rejected'로 변경.""" return self.update_entry(id, {"status": "rejected"}) def bulk_import(self, entries: list[dict[str, Any]]) -> int: """일괄 import. 성공 건수 반환.""" success = 0 for entry in entries: try: self.insert_entry(entry) success += 1 except Exception as exc: logger.warning("bulk_import: 항목 삽입 실패 — %s", exc) return success def get_stats(self) -> dict[str, Any]: """카테고리별 / 상태별 통계.""" cur = self._conn.cursor() cur.execute("SELECT COUNT(*) FROM wiki_entries") total_row = cur.fetchone() total: int = total_row[0] if total_row else 0 cur.execute( "SELECT category, COUNT(*) as cnt FROM wiki_entries GROUP BY category" ) by_category: dict[str, int] = { row["category"]: row["cnt"] for row in cur.fetchall() } cur.execute("SELECT status, COUNT(*) as cnt FROM wiki_entries GROUP BY status") by_status: dict[str, int] = { row["status"]: row["cnt"] for row in cur.fetchall() } return { "total": total, "by_category": by_category, "by_status": by_status, } def close(self) -> None: """DB 연결 종료.""" self._conn.close()