| 1 | """Offline FTS search across the saved research library and store sightings.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | import hashlib |
| 6 | import os |
| 7 | import re |
| 8 | import sqlite3 |
| 9 | from dataclasses import dataclass, replace |
| 10 | from datetime import date |
| 11 | from pathlib import Path |
| 12 | |
| 13 | from . import library |
| 14 | |
| 15 | |
| 16 | DEFAULT_LIBRARY_DB = library.DEFAULT_BRIEFS_DIR.parent / "library.db" |
| 17 | DEFAULT_STORE_DB = library.DEFAULT_BRIEFS_DIR.parent / "research.db" |
| 18 | INDEX_FINGERPRINT_VERSION = "last30days-library-index/v2" |
| 19 | LIBRARY_CONTEXT_START = "<!-- last30days:library-context:start -->" |
| 20 | LIBRARY_CONTEXT_END = "<!-- last30days:library-context:end -->" |
| 21 | _TOKEN = re.compile(r"[^\W_]+", re.UNICODE) |
| 22 | _MARKED_LIBRARY_CONTEXT = re.compile( |
| 23 | rf"^{re.escape(LIBRARY_CONTEXT_START)}\s*$.*?" |
| 24 | rf"^{re.escape(LIBRARY_CONTEXT_END)}\s*$\n?", |
| 25 | re.MULTILINE | re.DOTALL, |
| 26 | ) |
| 27 | _LEGACY_LIBRARY_CONTEXT = re.compile( |
| 28 | r"^## From your library\s*$.*?(?=^##\s|\Z)", |
| 29 | re.MULTILINE | re.DOTALL, |
| 30 | ) |
| 31 | _PRIVATE_CORPUS_BLOCK = re.compile( |
| 32 | r"<!-- LAST30DAYS_PRIVATE_CORPUS_START -->.*?" |
| 33 | r"<!-- LAST30DAYS_PRIVATE_CORPUS_END -->\s*", |
| 34 | re.DOTALL, |
| 35 | ) |
| 36 | |
| 37 | |
| 38 | class LibrarySearchUnavailable(RuntimeError): |
| 39 | """Raised when this Python SQLite build cannot provide FTS5.""" |
| 40 | |
| 41 | |
| 42 | @dataclass(frozen=True, slots=True) |
| 43 | class LibrarySearchMatch: |
| 44 | topic: str |
| 45 | published_date: date |
| 46 | headline: str |
| 47 | snippet: str |
| 48 | source_kind: str |
| 49 | rank: float |
| 50 | source_path: str = "" |
| 51 | url: str = "" |
| 52 | engagement: float | None = None |
| 53 | |
| 54 | @property |
| 55 | def run_key(self) -> tuple[str, date]: |
| 56 | return self.topic, self.published_date |
| 57 | |
| 58 | |
| 59 | @dataclass(frozen=True, slots=True) |
| 60 | class SyncResult: |
| 61 | indexed: int = 0 |
| 62 | removed: int = 0 |
| 63 | unchanged: int = 0 |
| 64 | notes: tuple[str, ...] = () |
| 65 | rebuilt: bool = False |
| 66 | |
| 67 | |
| 68 | _SCHEMA = """ |
| 69 | CREATE TABLE IF NOT EXISTS library_documents ( |
| 70 | entry_id TEXT PRIMARY KEY, |
| 71 | source_path TEXT UNIQUE NOT NULL, |
| 72 | source_mtime_ns INTEGER NOT NULL, |
| 73 | source_size INTEGER NOT NULL, |
| 74 | content_hash TEXT NOT NULL, |
| 75 | topic TEXT NOT NULL, |
| 76 | published_date TEXT NOT NULL, |
| 77 | headline TEXT NOT NULL, |
| 78 | summary TEXT NOT NULL, |
| 79 | source_format TEXT NOT NULL |
| 80 | ); |
| 81 | CREATE VIRTUAL TABLE IF NOT EXISTS library_fts USING fts5( |
| 82 | entry_id UNINDEXED, |
| 83 | topic, |
| 84 | headline, |
| 85 | summary, |
| 86 | content, |
| 87 | tokenize='porter unicode61' |
| 88 | ); |
| 89 | """ |
| 90 | |
| 91 | |
| 92 | def fts5_available() -> bool: |
| 93 | try: |
| 94 | with sqlite3.connect(":memory:") as conn: |
| 95 | conn.execute("CREATE VIRTUAL TABLE probe USING fts5(value)") |
| 96 | except sqlite3.DatabaseError: |
| 97 | return False |
| 98 | return True |
| 99 | |
| 100 | |
| 101 | def sync_library( |
| 102 | memory_dir: Path | str = library.DEFAULT_MEMORY_DIR, |
| 103 | briefs_dir: Path | str = library.DEFAULT_BRIEFS_DIR, |
| 104 | *, |
| 105 | db_path: Path | str = DEFAULT_LIBRARY_DB, |
| 106 | ) -> SyncResult: |
| 107 | """Incrementally index the shared ``scan_library`` view of saved research.""" |
| 108 | if not fts5_available(): |
| 109 | raise LibrarySearchUnavailable( |
| 110 | "library search requires a Python SQLite build with FTS5 support" |
| 111 | ) |
| 112 | target = Path(db_path).expanduser() |
| 113 | try: |
| 114 | return _sync_library(memory_dir, briefs_dir, target) |
| 115 | except sqlite3.DatabaseError as exc: |
| 116 | if "fts5" in str(exc).lower() and "malformed" not in str(exc).lower(): |
| 117 | raise LibrarySearchUnavailable( |
| 118 | "library search requires a Python SQLite build with FTS5 support" |
| 119 | ) from exc |
| 120 | if not _is_confirmed_corruption(exc): |
| 121 | raise |
| 122 | _remove_database(target) |
| 123 | return replace(_sync_library(memory_dir, briefs_dir, target), rebuilt=True) |
| 124 | |
| 125 | |
| 126 | def index_brief( |
| 127 | path: Path | str, |
| 128 | *, |
| 129 | db_path: Path | str = DEFAULT_LIBRARY_DB, |
| 130 | ) -> bool: |
| 131 | """Index one saved artifact, parsing it through ``scan_library``.""" |
| 132 | source = Path(path).expanduser().resolve() |
| 133 | if source.suffix.lower() == ".json": |
| 134 | entries, _ = library.scan_library(source.parent / ".missing", source.parent) |
| 135 | else: |
| 136 | entries, _ = library.scan_library(source.parent, source.parent / ".missing") |
| 137 | entry = next((item for item in entries if item.source_path.resolve() == source), None) |
| 138 | if entry is None: |
| 139 | return False |
| 140 | target = Path(db_path).expanduser() |
| 141 | _ensure_private_directory(target.parent) |
| 142 | with _connect(target) as conn: |
| 143 | _upsert_entry(conn, entry) |
| 144 | conn.commit() |
| 145 | return True |
| 146 | |
| 147 | |
| 148 | def search( |
| 149 | query: str, |
| 150 | *, |
| 151 | limit: int = 20, |
| 152 | db_path: Path | str = DEFAULT_LIBRARY_DB, |
| 153 | store_db_path: Path | str = DEFAULT_STORE_DB, |
| 154 | ) -> list[LibrarySearchMatch]: |
| 155 | """Search indexed briefs plus dated per-run findings from the research store.""" |
| 156 | expression = _fts_expression(query) |
| 157 | if not expression or limit <= 0: |
| 158 | return [] |
| 159 | target = Path(db_path).expanduser() |
| 160 | brief_matches: list[LibrarySearchMatch] = [] |
| 161 | if target.is_file(): |
| 162 | try: |
| 163 | with _connect(target) as conn: |
| 164 | rows = conn.execute( |
| 165 | """SELECT d.topic, d.published_date, d.headline, |
| 166 | snippet(library_fts, 4, '', '', ' … ', 36) AS snippet, |
| 167 | d.source_path, bm25(library_fts) AS rank |
| 168 | FROM library_fts |
| 169 | JOIN library_documents d ON d.entry_id = library_fts.entry_id |
| 170 | WHERE library_fts MATCH ? |
| 171 | ORDER BY rank, d.published_date DESC |
| 172 | LIMIT ?""", |
| 173 | (expression, limit), |
| 174 | ).fetchall() |
| 175 | except sqlite3.DatabaseError: |
| 176 | rows = [] |
| 177 | brief_matches = [ |
| 178 | LibrarySearchMatch( |
| 179 | topic=str(row["topic"]), |
| 180 | published_date=date.fromisoformat(str(row["published_date"])), |
| 181 | headline=str(row["headline"]), |
| 182 | snippet=_clean_snippet(row["snippet"]), |
| 183 | source_kind="brief", |
| 184 | rank=float(row["rank"]), |
| 185 | source_path=str(row["source_path"]), |
| 186 | ) |
| 187 | for row in rows |
| 188 | ] |
| 189 | store_matches = _search_store_sightings( |
| 190 | expression, Path(store_db_path).expanduser(), limit |
| 191 | ) |
| 192 | return _merge_ranked_matches([brief_matches, store_matches], limit=limit) |
| 193 | |
| 194 | |
| 195 | def sync_and_search( |
| 196 | query: str, |
| 197 | *, |
| 198 | memory_dir: Path | str = library.DEFAULT_MEMORY_DIR, |
| 199 | briefs_dir: Path | str = library.DEFAULT_BRIEFS_DIR, |
| 200 | db_path: Path | str = DEFAULT_LIBRARY_DB, |
| 201 | store_db_path: Path | str = DEFAULT_STORE_DB, |
| 202 | limit: int = 20, |
| 203 | ) -> tuple[list[LibrarySearchMatch], SyncResult]: |
| 204 | synced = sync_library(memory_dir, briefs_dir, db_path=db_path) |
| 205 | return search( |
| 206 | query, |
| 207 | limit=limit, |
| 208 | db_path=db_path, |
| 209 | store_db_path=store_db_path, |
| 210 | ), synced |
| 211 | |
| 212 | |
| 213 | def _sync_library( |
| 214 | memory_dir: Path | str, |
| 215 | briefs_dir: Path | str, |
| 216 | db_path: Path, |
| 217 | ) -> SyncResult: |
| 218 | entries, notes = library.scan_library(memory_dir, briefs_dir) |
| 219 | _ensure_private_directory(db_path.parent) |
| 220 | indexed = unchanged = 0 |
| 221 | with _connect(db_path) as conn: |
| 222 | existing = { |
| 223 | row["entry_id"]: (row["source_mtime_ns"], row["source_size"], row["content_hash"]) |
| 224 | for row in conn.execute( |
| 225 | "SELECT entry_id, source_mtime_ns, source_size, content_hash FROM library_documents" |
| 226 | ) |
| 227 | } |
| 228 | current_ids: set[str] = set() |
| 229 | # If the FTS table was lost or recreated empty while library_documents |
| 230 | # survived, the fingerprint check alone would mark everything unchanged |
| 231 | # and searches would silently return nothing. Verify row counts agree |
| 232 | # before trusting fingerprints. |
| 233 | fts_rows = conn.execute("SELECT count(*) FROM library_fts").fetchone()[0] |
| 234 | fts_trustworthy = fts_rows >= len(existing) if existing else True |
| 235 | for entry in entries: |
| 236 | current_ids.add(entry.entry_id) |
| 237 | stat = entry.source_path.stat() |
| 238 | fingerprint = _fingerprint(_indexable_content(entry.content)) |
| 239 | if fts_trustworthy and existing.get(entry.entry_id) == ( |
| 240 | stat.st_mtime_ns, stat.st_size, fingerprint |
| 241 | ): |
| 242 | unchanged += 1 |
| 243 | continue |
| 244 | _upsert_entry(conn, entry, fingerprint=fingerprint) |
| 245 | indexed += 1 |
| 246 | stale_ids = set(existing) - current_ids |
| 247 | for entry_id in stale_ids: |
| 248 | conn.execute("DELETE FROM library_fts WHERE entry_id = ?", (entry_id,)) |
| 249 | conn.execute("DELETE FROM library_documents WHERE entry_id = ?", (entry_id,)) |
| 250 | conn.commit() |
| 251 | return SyncResult( |
| 252 | indexed=indexed, |
| 253 | removed=len(stale_ids), |
| 254 | unchanged=unchanged, |
| 255 | notes=tuple(notes), |
| 256 | ) |
| 257 | |
| 258 | |
| 259 | def _connect(path: Path) -> sqlite3.Connection: |
| 260 | _ensure_private_directory(path.parent) |
| 261 | if not path.exists(): |
| 262 | try: |
| 263 | fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) |
| 264 | except FileExistsError: |
| 265 | pass |
| 266 | else: |
| 267 | os.close(fd) |
| 268 | path.chmod(0o600) |
| 269 | conn = sqlite3.connect(str(path)) |
| 270 | try: |
| 271 | conn.row_factory = sqlite3.Row |
| 272 | conn.execute("PRAGMA busy_timeout=5000") |
| 273 | conn.executescript(_SCHEMA) |
| 274 | except Exception: |
| 275 | conn.close() |
| 276 | raise |
| 277 | return conn |
| 278 | |
| 279 | |
| 280 | def _upsert_entry( |
| 281 | conn: sqlite3.Connection, |
| 282 | entry: library.LibraryEntry, |
| 283 | *, |
| 284 | fingerprint: str | None = None, |
| 285 | ) -> None: |
| 286 | stat = entry.source_path.stat() |
| 287 | private_free_content = _PRIVATE_CORPUS_BLOCK.sub("", entry.content) |
| 288 | indexed_content = _indexable_content(private_free_content) |
| 289 | headline = entry.headline |
| 290 | summary = entry.summary |
| 291 | if private_free_content != entry.content and entry.source_format == "markdown": |
| 292 | headline = library._markdown_headline(private_free_content) or entry.topic |
| 293 | summary = library._markdown_summary(private_free_content) or headline |
| 294 | content_hash = fingerprint or _fingerprint(indexed_content) |
| 295 | source_path = str(entry.source_path.resolve()) |
| 296 | replaced = conn.execute( |
| 297 | "SELECT entry_id FROM library_documents WHERE source_path = ? AND entry_id != ?", |
| 298 | (source_path, entry.entry_id), |
| 299 | ).fetchall() |
| 300 | for row in replaced: |
| 301 | conn.execute("DELETE FROM library_fts WHERE entry_id = ?", (row["entry_id"],)) |
| 302 | conn.execute("DELETE FROM library_documents WHERE entry_id = ?", (row["entry_id"],)) |
| 303 | conn.execute("DELETE FROM library_fts WHERE entry_id = ?", (entry.entry_id,)) |
| 304 | conn.execute( |
| 305 | """INSERT INTO library_documents |
| 306 | (entry_id, source_path, source_mtime_ns, source_size, content_hash, |
| 307 | topic, published_date, headline, summary, source_format) |
| 308 | VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) |
| 309 | ON CONFLICT(entry_id) DO UPDATE SET |
| 310 | source_path=excluded.source_path, |
| 311 | source_mtime_ns=excluded.source_mtime_ns, |
| 312 | source_size=excluded.source_size, |
| 313 | content_hash=excluded.content_hash, |
| 314 | topic=excluded.topic, |
| 315 | published_date=excluded.published_date, |
| 316 | headline=excluded.headline, |
| 317 | summary=excluded.summary, |
| 318 | source_format=excluded.source_format""", |
| 319 | ( |
| 320 | entry.entry_id, |
| 321 | source_path, |
| 322 | stat.st_mtime_ns, |
| 323 | stat.st_size, |
| 324 | content_hash, |
| 325 | entry.topic, |
| 326 | entry.published_date.isoformat(), |
| 327 | headline, |
| 328 | summary, |
| 329 | entry.source_format, |
| 330 | ), |
| 331 | ) |
| 332 | conn.execute( |
| 333 | "INSERT INTO library_fts(entry_id, topic, headline, summary, content) VALUES (?, ?, ?, ?, ?)", |
| 334 | (entry.entry_id, entry.topic, headline, summary, indexed_content), |
| 335 | ) |
| 336 | |
| 337 | |
| 338 | def _search_store_sightings( |
| 339 | expression: str, |
| 340 | store_db_path: Path, |
| 341 | limit: int, |
| 342 | ) -> list[LibrarySearchMatch]: |
| 343 | if not store_db_path.is_file(): |
| 344 | return [] |
| 345 | try: |
| 346 | with sqlite3.connect(str(store_db_path)) as conn: |
| 347 | conn.row_factory = sqlite3.Row |
| 348 | rows = conn.execute( |
| 349 | """SELECT t.name AS topic, rr.run_date, |
| 350 | COALESCE(fs.source_title, f.source_title, f.summary) AS headline, |
| 351 | snippet(findings_fts, 0, '', '', ' … ', 30) AS snippet, |
| 352 | fs.source_url, fs.engagement_score, bm25(findings_fts) AS rank |
| 353 | FROM findings_fts |
| 354 | JOIN findings f ON f.id = findings_fts.rowid |
| 355 | JOIN finding_sightings fs ON fs.finding_id = f.id |
| 356 | JOIN research_runs rr ON rr.id = fs.run_id |
| 357 | JOIN topics t ON t.id = fs.topic_id |
| 358 | WHERE findings_fts MATCH ? AND rr.status = 'completed' |
| 359 | AND fs.source != 'corpus' |
| 360 | ORDER BY rank, rr.run_date DESC |
| 361 | LIMIT ?""", |
| 362 | (expression, limit), |
| 363 | ).fetchall() |
| 364 | except (sqlite3.DatabaseError, OSError): |
| 365 | return [] |
| 366 | matches: list[LibrarySearchMatch] = [] |
| 367 | for row in rows: |
| 368 | try: |
| 369 | published = date.fromisoformat(str(row["run_date"])[:10]) |
| 370 | except ValueError: |
| 371 | continue |
| 372 | matches.append( |
| 373 | LibrarySearchMatch( |
| 374 | topic=str(row["topic"]), |
| 375 | published_date=published, |
| 376 | headline=str(row["headline"] or "Saved finding"), |
| 377 | snippet=_clean_snippet(row["snippet"]), |
| 378 | source_kind="store", |
| 379 | rank=float(row["rank"]), |
| 380 | url=str(row["source_url"] or ""), |
| 381 | engagement=( |
| 382 | float(row["engagement_score"]) |
| 383 | if row["engagement_score"] is not None |
| 384 | else None |
| 385 | ), |
| 386 | ) |
| 387 | ) |
| 388 | return matches |
| 389 | |
| 390 | |
| 391 | def _fts_expression(query: str) -> str: |
| 392 | tokens = _TOKEN.findall(query) |
| 393 | return " AND ".join(f'"{token.replace(chr(34), chr(34) * 2)}"' for token in tokens) |
| 394 | |
| 395 | |
| 396 | def _fingerprint(content: str) -> str: |
| 397 | payload = f"{INDEX_FINGERPRINT_VERSION}\0{content}" |
| 398 | return hashlib.sha256(payload.encode("utf-8")).hexdigest() |
| 399 | |
| 400 | |
| 401 | def _clean_snippet(value: object) -> str: |
| 402 | return re.sub(r"\s+", " ", str(value or "")).strip()[:500] |
| 403 | |
| 404 | |
| 405 | def _indexable_content(content: str) -> str: |
| 406 | without_private = _PRIVATE_CORPUS_BLOCK.sub("", content) |
| 407 | without_marked = _MARKED_LIBRARY_CONTEXT.sub("", without_private) |
| 408 | return _LEGACY_LIBRARY_CONTEXT.sub("", without_marked) |
| 409 | |
| 410 | |
| 411 | def _ensure_private_directory(path: Path) -> None: |
| 412 | missing: list[Path] = [] |
| 413 | current = path |
| 414 | while not current.exists(): |
| 415 | missing.append(current) |
| 416 | current = current.parent |
| 417 | path.mkdir(parents=True, exist_ok=True, mode=0o700) |
| 418 | for directory in missing: |
| 419 | directory.chmod(0o700) |
| 420 | |
| 421 | |
| 422 | def _is_confirmed_corruption(exc: sqlite3.DatabaseError) -> bool: |
| 423 | message = str(exc).casefold() |
| 424 | return any( |
| 425 | marker in message |
| 426 | for marker in ( |
| 427 | "file is not a database", |
| 428 | "database disk image is malformed", |
| 429 | "database schema is corrupt", |
| 430 | "malformed database schema", |
| 431 | ) |
| 432 | ) |
| 433 | |
| 434 | |
| 435 | def _merge_ranked_matches( |
| 436 | corpora: list[list[LibrarySearchMatch]], |
| 437 | *, |
| 438 | limit: int, |
| 439 | ) -> list[LibrarySearchMatch]: |
| 440 | normalized: list[LibrarySearchMatch] = [] |
| 441 | for matches in corpora: |
| 442 | for position, match in enumerate(matches, start=1): |
| 443 | normalized.append(replace(match, rank=-(1.0 / (60 + position)))) |
| 444 | combined = _dedupe_matches(normalized) |
| 445 | return sorted( |
| 446 | combined, |
| 447 | key=lambda match: ( |
| 448 | match.rank, |
| 449 | -match.published_date.toordinal(), |
| 450 | match.topic.casefold(), |
| 451 | match.headline.casefold(), |
| 452 | ), |
| 453 | )[:limit] |
| 454 | |
| 455 | |
| 456 | def _dedupe_matches(matches: list[LibrarySearchMatch]) -> list[LibrarySearchMatch]: |
| 457 | seen: set[tuple[str, date, str, str]] = set() |
| 458 | kept: list[LibrarySearchMatch] = [] |
| 459 | for match in matches: |
| 460 | key = ( |
| 461 | match.topic.casefold(), |
| 462 | match.published_date, |
| 463 | match.headline.casefold(), |
| 464 | match.source_kind, |
| 465 | ) |
| 466 | if key not in seen: |
| 467 | seen.add(key) |
| 468 | kept.append(match) |
| 469 | return kept |
| 470 | |
| 471 | |
| 472 | def _remove_database(path: Path) -> None: |
| 473 | for candidate in (path, Path(f"{path}-wal"), Path(f"{path}-shm")): |
| 474 | try: |
| 475 | candidate.unlink() |
| 476 | except FileNotFoundError: |
| 477 | pass |
| 478 |