"""FTS5 full-text search index for Second Brain wiki pages."""
import sqlite3
import logging
from pathlib import Path
logger = logging.getLogger(__name__)
DB_PATH = Path.home() / "second-brain" / "vault.db"
WIKI_DIR = Path.home() / "second-brain" / "wiki"
def get_connection():
"""Get a database connection with FTS5 support."""
conn = sqlite3.connect(str(DB_PATH))
conn.execute("PRAGMA journal_mode=WAL")
conn.row_factory = sqlite3.Row
return conn
def init_db():
"""Create the database and FTS5 virtual table if they don't exist."""
conn = get_connection()
try:
conn.executescript("""
CREATE TABLE IF NOT EXISTS wiki_pages (
slug TEXT PRIMARY KEY,
title TEXT NOT NULL,
tags TEXT,
body TEXT NOT NULL,
created TEXT,
updated TEXT,
mtime REAL NOT NULL,
path TEXT NOT NULL
);
CREATE VIRTUAL TABLE IF NOT EXISTS wiki_fts USING fts5(
title,
tags,
body,
content='wiki_pages',
content_rowid='rowid'
);
""")
except Exception as e:
logger.exception("Failed to initialize FTS5 database: %s", e)
raise
finally:
conn.close()
# Run init on import
init_db()
def _parse_yaml_frontmatter(content):
"""Minimal YAML frontmatter parser for indexing."""
metadata = {}
body = content
if content.startswith("---"):
parts = content.split("---", 2)
if len(parts) >= 3:
yaml_str = parts[1].strip()
body = parts[2].strip()
for line in yaml_str.split("\n"):
if ":" in line:
key, _, value = line.partition(":")
metadata[key.strip()] = value.strip().strip('"').strip("'")
return metadata, body
def index_page(slug, title, tags, body, created, updated, mtime, path):
"""Index or update a single wiki page."""
conn = get_connection()
try:
conn.execute("""
INSERT OR REPLACE INTO wiki_pages
(slug, title, tags, body, created, updated, mtime, path)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (slug, title, tags or "", body, created or "", updated or "", mtime, path))
conn.execute("INSERT INTO wiki_fts(wiki_fts) VALUES('rebuild')")
conn.commit()
except Exception:
logger.exception("Failed to index page: %s", slug)
raise
finally:
conn.close()
def delete_page(slug):
"""Remove a page from the index."""
conn = get_connection()
try:
conn.execute("DELETE FROM wiki_pages WHERE slug = ?", (slug,))
conn.execute("INSERT INTO wiki_fts(wiki_fts) VALUES('rebuild')")
conn.commit()
except Exception:
logger.exception("Failed to delete page from index: %s", slug)
raise
finally:
conn.close()
def rebuild_index():
"""Rebuild the entire index from disk. Call on startup."""
if not WIKI_DIR.exists():
logger.warning("Wiki directory does not exist: %s", WIKI_DIR)
return
conn = get_connection()
try:
# Clear and rebuild
conn.execute("DELETE FROM wiki_pages")
conn.execute("INSERT INTO wiki_fts(wiki_fts) VALUES('rebuild')")
conn.commit()
count = 0
for f in WIKI_DIR.glob("*.md"):
try:
content = f.read_text(encoding="utf-8", errors="ignore")
metadata, body = _parse_yaml_frontmatter(content)
title = metadata.get("title", f.stem.replace("-", " ").title())
tags = metadata.get("tags", "")
created = metadata.get("created", "")
updated = metadata.get("updated", "")
mtime = f.stat().st_mtime
conn.execute("""
INSERT OR REPLACE INTO wiki_pages
(slug, title, tags, body, created, updated, mtime, path)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (f.stem, title, tags, body, created, updated, mtime, str(f)))
count += 1
except Exception:
logger.exception("Failed to index: %s", f.name)
# Final rebuild of FTS index
conn.execute("INSERT INTO wiki_fts(wiki_fts) VALUES('rebuild')")
conn.commit()
logger.info("Rebuilt FTS5 index: %d pages indexed", count)
except Exception:
logger.exception("Failed to rebuild index")
raise
finally:
conn.close()
def search(query, limit=50):
"""Search the wiki using FTS5. Returns list of dicts with highlights."""
if not query or len(query) < 2:
return []
conn = get_connection()
try:
rows = conn.execute("""
SELECT p.slug, p.title, p.tags, p.created, p.updated, p.mtime,
snippet(wiki_fts, 0, '<<b>>', '</b>', '...', 64) as snippet
FROM wiki_fts f
JOIN wiki_pages p ON p.rowid = f.rowid
WHERE wiki_fts MATCH ?
ORDER BY rank
LIMIT ?
""", (query, limit)).fetchall()
results = []
for row in rows:
results.append({
"slug": row["slug"],
"title": row["title"],
"tags": row["tags"],
"created": row["created"],
"updated": row["updated"],
"mtime": row["mtime"],
"snippet": row["snippet"] or "",
"score": 0,
})
return results
except Exception:
logger.exception("FTS5 search failed for query: %s", query)
return []
finally:
conn.close()