#!/usr/bin/env python3 """Build and query a compact official-source retrieval database.""" from __future__ import annotations import argparse import hashlib import html import importlib import math import os from contextlib import contextmanager from html.parser import HTMLParser import json import re import sqlite3 import sys from dataclasses import asdict, dataclass from pathlib import Path from types import SimpleNamespace from typing import Any from urllib.parse import urljoin, urlparse from urllib.request import Request, urlopen DEFAULT_SOURCES_PATH = Path("benchmarks/sources.json") DEFAULT_NOTES_PATH = Path("retrieval/source_passage_notes.jsonl") DEFAULT_DB_PATH = Path("outputs/source_retrieval/official_sources.harrier.sqlite3") DEFAULT_DOC_CACHE_DIR = Path("outputs/source_retrieval/document_cache") DEFAULT_NOLEGRAPH_SRC = Path("/Users/hasse_h/nolegraph/src") DEFAULT_EMBEDDING_MODEL = "microsoft/harrier-oss-v1-0.6b" QUERY_INSTRUCTION = "Represent this query for retrieving relevant passages: " MAX_SEARCH_TERMS = 64 MAX_CRAWL_LINKS = 500 MIN_PARAGRAPH_CHARS = 35 EMBED_BATCH_SIZE = 8 EMBED_TEXT_MAX_CHARS = 1800 USER_AGENT = "Synderesis-SourceIndexer/0.1 (+local research cache)" WORD_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9']+") EMBED_TOKEN_RE = re.compile(r"[\w-]+", re.UNICODE) OFFICIAL_LOCATION_RE = re.compile(r"^\s*(?:no\.\s*)?(\d{1,4})(?:[\.\s:,-]|$)", re.IGNORECASE) STOPWORDS = { "about", "after", "also", "and", "are", "background", "based", "because", "been", "being", "but", "can", "could", "does", "document", "documents", "for", "from", "go", "goes", "has", "have", "how", "into", "its", "like", "make", "means", "not", "one", "other", "out", "post", "reach", "reflects", "regardless", "says", "should", "that", "the", "their", "there", "this", "through", "together", "togheter", "what", "when", "where", "while", "with", "word", } SYNONYMS = { "catholic": ["catholic", "universal", "church"], "church": ["church", "catholic", "mission"], "convert": ["convert", "conversion", "proselytism", "evangelization", "mission", "witness"], "conversion": ["convert", "conversion", "proselytism", "evangelization", "mission", "witness"], "evangelize": ["evangelization", "mission", "witness", "proclaim"], "jew": ["jew", "jews", "jewish", "judaism", "israel"], "jews": ["jew", "jews", "jewish", "judaism", "israel"], "jewish": ["jew", "jews", "jewish", "judaism", "israel"], "mission": ["mission", "evangelization", "witness", "proclaim", "universal"], "universal": ["universal", "catholic", "whole", "human", "mission"], "vatican": ["vatican", "council", "conciliar"], } class SourceRetrievalError(Exception): """Raised when the source retrieval database cannot be built or queried.""" @dataclass(frozen=True) class SourceChunk: """One compact searchable source passage note.""" source_id: str chunk_kind: str title: str url: str publisher: str source_type: str location: str paragraph_index: int topics: list[str] summary: str text: str keywords: list[str] @dataclass(frozen=True) class SearchResult: """One retrieved source passage.""" source_id: str chunk_kind: str title: str url: str publisher: str source_type: str location: str paragraph_index: int topics: list[str] summary: str text: str keywords: list[str] score: float def to_dict(self) -> dict[str, Any]: """Return an API-safe dictionary representation.""" return asdict(self) def embed_tokenize(text: str) -> list[str]: """Tokenize text for the dependency-free hash embedder.""" return [token.casefold() for token in EMBED_TOKEN_RE.findall(text) if len(token) > 1] def l2_normalize(vector: list[float]) -> list[float]: """Return an L2-normalized vector.""" norm = math.sqrt(sum(value * value for value in vector)) if norm == 0.0: return vector return [value / norm for value in vector] class HashEmbedder: """Deterministic, dependency-free feature-hashing embedder.""" def __init__(self, dim: int = 256) -> None: if dim <= 0: raise ValueError("dim must be positive") self._dim = dim @property def dim(self) -> int: return self._dim @property def signature(self) -> str: return f"hash-{self._dim}" def encode(self, texts: list[str], *, is_query: bool = False) -> list[list[float]]: """Encode text using stable token hashing.""" vectors: list[list[float]] = [] for text in texts: source = (QUERY_INSTRUCTION + text) if is_query else text vector = [0.0] * self._dim for token in embed_tokenize(source): digest = hashlib.sha256(token.encode("utf-8")).digest() bucket = int.from_bytes(digest[:8], "big") % self._dim vector[bucket] += 1.0 vectors.append(l2_normalize(vector)) return vectors class HarrierEmbedder: """sentence-transformers-backed Harrier embedder compatible with nolegraph.""" def __init__(self, model_name: str = DEFAULT_EMBEDDING_MODEL) -> None: module = importlib.import_module("sentence_transformers") transformer_cls: Any = module.SentenceTransformer self._model_name = model_name try: self._model: Any = transformer_cls(model_name, local_files_only=True) except Exception: self._model = transformer_cls(model_name) get_dim = getattr(self._model, "get_embedding_dimension", self._model.get_sentence_embedding_dimension) dimension = get_dim() self._dim = int(dimension) if dimension is not None else 1024 @property def dim(self) -> int: return self._dim @property def signature(self) -> str: return f"st:{self._model_name}:{self._dim}" def encode(self, texts: list[str], *, is_query: bool = False) -> list[list[float]]: """Encode text using Harrier and normalized embeddings.""" prepared = [QUERY_INSTRUCTION + text for text in texts] if is_query else list(texts) raw = self._model.encode(prepared, normalize_embeddings=True) if hasattr(raw, "tolist"): raw = raw.tolist() return [[float(value) for value in row] for row in raw] def load_default_embedder() -> Any | None: """Load the default Harrier embedder only when explicitly enabled.""" flag = os.environ.get("NOLEGRAPH_AUTOLOAD_EMBEDDER", "").strip().lower() if flag not in {"1", "true", "yes", "on"}: return None model_name = os.environ.get("NOLEGRAPH_EMBEDDING_MODEL", DEFAULT_EMBEDDING_MODEL) try: return HarrierEmbedder(model_name) except Exception as error: print(f"source_retrieval: embedder {model_name!r} failed to load, falling back to lexical search: {error}", file=sys.stderr) return None def load_sources(path: Path) -> dict[str, dict[str, Any]]: """Load the official source registry keyed by source id.""" try: payload = json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError as exc: raise SourceRetrievalError(f"source registry not found: {path}") from exc except json.JSONDecodeError as exc: raise SourceRetrievalError(f"source registry is invalid JSON: {path}") from exc sources = payload.get("sources") if not isinstance(sources, list): raise SourceRetrievalError("source registry must contain a sources list") indexed: dict[str, dict[str, Any]] = {} for source in sources: if not isinstance(source, dict) or not isinstance(source.get("id"), str): raise SourceRetrievalError("each source registry entry must include an id") indexed[source["id"]] = source return indexed def load_notes(path: Path, sources: dict[str, dict[str, Any]]) -> dict[tuple[str, str], dict[str, Any]]: """Load curated passage notes keyed by source id and location.""" notes: dict[tuple[str, str], dict[str, Any]] = {} if not path.exists(): return notes for line_number, raw_line in enumerate(path.read_text(encoding="utf-8").splitlines(), start=1): line = raw_line.strip() if not line: continue try: note = json.loads(line) except json.JSONDecodeError as exc: raise SourceRetrievalError(f"invalid JSONL in {path}:{line_number}") from exc source_id = note.get("source_id") location = note.get("location") summary = note.get("summary") if not isinstance(source_id, str) or source_id not in sources: raise SourceRetrievalError(f"unknown source_id in {path}:{line_number}") if not isinstance(location, str) or not location.strip(): raise SourceRetrievalError(f"missing location in {path}:{line_number}") if not isinstance(summary, str) or not summary.strip(): raise SourceRetrievalError(f"missing summary in {path}:{line_number}") keywords = note.get("keywords", []) if isinstance(keywords, str): keywords = [keywords] if not isinstance(keywords, list) or not all(isinstance(item, str) for item in keywords): raise SourceRetrievalError(f"keywords must be strings in {path}:{line_number}") notes[(source_id, location)] = {"summary": summary.strip(), "keywords": [item.strip() for item in keywords if item.strip()]} return notes def registry_summary(source: dict[str, Any], location: str) -> str: """Create a fallback searchable summary from registry metadata.""" topics = ", ".join(str(topic) for topic in source.get("topics", [])) return f"{source.get('title', source['id'])} {location}. Official {source.get('type', 'source')} covering {topics}." def registry_keywords(source: dict[str, Any], location: str) -> list[str]: """Create fallback retrieval keywords from registry metadata.""" values = [source.get("title", ""), source.get("official_id", ""), source.get("type", ""), location] values.extend(str(topic) for topic in source.get("topics", [])) return [value for value in values if value] def clean_text(value: str) -> str: """Normalize HTML-derived whitespace.""" return re.sub(r"\s+", " ", html.unescape(value)).strip() class VaticanHtmlParser(HTMLParser): """Extract links and block-level text from official source HTML.""" block_tags = {"p", "li", "h1", "h2", "h3", "h4", "h5", "h6", "blockquote"} skip_tags = {"script", "style", "noscript"} def __init__(self) -> None: super().__init__(convert_charrefs=True) self.links: list[str] = [] self.paragraphs: list[str] = [] self._skip_depth = 0 self._block_depth = 0 self._buffer: list[str] = [] def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: tag = tag.lower() if tag in self.skip_tags: self._skip_depth += 1 return if tag == "a": for key, value in attrs: if key.lower() == "href" and value: self.links.append(value) if tag in self.block_tags: if self._block_depth == 0: self._buffer = [] self._block_depth += 1 def handle_endtag(self, tag: str) -> None: tag = tag.lower() if tag in self.skip_tags and self._skip_depth: self._skip_depth -= 1 return if tag in self.block_tags and self._block_depth: self._block_depth -= 1 if self._block_depth == 0: text = clean_text(" ".join(self._buffer)) if len(text) >= MIN_PARAGRAPH_CHARS: self.paragraphs.append(text) self._buffer = [] def handle_data(self, data: str) -> None: if self._skip_depth or not self._block_depth: return value = clean_text(data) if value: self._buffer.append(value) def noisy_paragraph(text: str) -> bool: """Filter obvious navigation and language-switcher fragments.""" normalized = clean_text(text) if len(normalized) < MIN_PARAGRAPH_CHARS: return True if normalized.startswith("[") and normalized.endswith("]") and len(normalized) < 220: return True lowered = normalized.lower() noisy_exact = { "the holy see", "copyright", "vatican.va", } if lowered in noisy_exact: return True if lowered.count(" - ") >= 5 and len(normalized) < 260: return True return False def parse_with_bs4(html_text: str) -> tuple[list[str], list[str]] | None: """Extract links and paragraphs with BeautifulSoup when available.""" try: from bs4 import BeautifulSoup except Exception: return None soup = BeautifulSoup(html_text, "html.parser") for tag in soup(["script", "style", "noscript", "nav", "header", "footer"]): tag.decompose() links = [str(tag.get("href")) for tag in soup.find_all("a") if tag.get("href")] paragraphs: list[str] = [] seen: set[str] = set() for tag in soup.find_all(["h1", "h2", "h3", "h4", "h5", "h6", "p", "li", "blockquote"]): text = clean_text(tag.get_text(" ", strip=True)) if noisy_paragraph(text) or text in seen: continue seen.add(text) paragraphs.append(text) return links, paragraphs def cache_path_for_url(cache_dir: Path, source_id: str, url: str) -> Path: """Return the local HTML cache path for a source URL.""" digest = hashlib.sha256(url.encode("utf-8")).hexdigest()[:16] return cache_dir / source_id / f"{digest}.html" def fetch_url(url: str, cache_dir: Path, source_id: str, force_fetch: bool = False) -> str: """Fetch URL with a generated local cache under outputs.""" cache_path = cache_path_for_url(cache_dir, source_id, url) if cache_path.exists() and not force_fetch: return cache_path.read_text(encoding="utf-8", errors="replace") cache_path.parent.mkdir(parents=True, exist_ok=True) request = Request(url, headers={"User-Agent": USER_AGENT}) with urlopen(request, timeout=30) as response: raw = response.read() charset = response.headers.get_content_charset() or "utf-8" text = raw.decode(charset, errors="replace") cache_path.write_text(text, encoding="utf-8") return text def parse_html_document(html_text: str) -> VaticanHtmlParser: """Parse one HTML document into paragraphs and links.""" bs4_result = parse_with_bs4(html_text) if bs4_result is not None: links, paragraphs = bs4_result parser = VaticanHtmlParser() parser.links = links parser.paragraphs = paragraphs return parser parser = VaticanHtmlParser() parser.feed(html_text) parser.close() parser.paragraphs = [paragraph for paragraph in parser.paragraphs if not noisy_paragraph(paragraph)] return parser def same_document_link(base_url: str, candidate_url: str) -> bool: """Return whether a link should be crawled as part of the same source.""" base = urlparse(base_url) candidate = urlparse(candidate_url) if candidate.scheme not in {"http", "https"}: return False if candidate.netloc != base.netloc: return False base_dir = base.path.rsplit("/", 1)[0] + "/" if not candidate.path.startswith(base_dir): return False return candidate.path.lower().endswith((".htm", ".html")) def document_urls_for_source(source: dict[str, Any], cache_dir: Path, force_fetch: bool) -> list[str]: """Return URLs to fetch for one source, crawling Catechism-style index pages.""" source_id = str(source["id"]) base_url = str(source.get("url", "")) if not base_url: return [] urls = [base_url] if base_url.upper().endswith("_INDEX.HTM"): parser = parse_html_document(fetch_url(base_url, cache_dir, source_id, force_fetch)) discovered: list[str] = [] for href in parser.links: absolute = urljoin(base_url, href.split("#", 1)[0]) if same_document_link(base_url, absolute) and absolute not in urls and absolute not in discovered: discovered.append(absolute) if len(discovered) >= MAX_CRAWL_LINKS: break urls.extend(discovered) return urls def official_location(text: str, paragraph_index: int, used: set[str]) -> str: """Infer an official paragraph number when present, otherwise use a local paragraph id.""" match = OFFICIAL_LOCATION_RE.match(text) if match: candidate = match.group(1) if candidate not in used: used.add(candidate) return candidate fallback = f"p{paragraph_index:04d}" used.add(fallback) return fallback def paragraph_summary(text: str, max_chars: int = 700) -> str: """Return a bounded paragraph summary/excerpt.""" if len(text) <= max_chars: return text clipped = text[:max_chars].rsplit(" ", 1)[0].rstrip() return f"{clipped}..." def document_chunks( sources: dict[str, dict[str, Any]], cache_dir: Path, force_fetch: bool = False, ) -> list[SourceChunk]: """Fetch official documents and return paragraph-level chunks.""" chunks: list[SourceChunk] = [] for source_id, source in sources.items(): topics = [str(topic) for topic in source.get("topics", [])] used_locations: set[str] = set() paragraph_index = 0 for url in document_urls_for_source(source, cache_dir, force_fetch): parser = parse_html_document(fetch_url(url, cache_dir, source_id, force_fetch)) for paragraph in parser.paragraphs: text = clean_text(paragraph) if len(text) < MIN_PARAGRAPH_CHARS: continue paragraph_index += 1 location = official_location(text, paragraph_index, used_locations) chunks.append( SourceChunk( source_id=source_id, chunk_kind="document", title=str(source.get("title", source_id)), url=url, publisher=str(source.get("publisher", "")), source_type=str(source.get("type", "")), location=location, paragraph_index=paragraph_index, topics=topics, summary=paragraph_summary(text), text=text, keywords=registry_keywords(source, location), ) ) return chunks def load_chunks( sources_path: Path, notes_path: Path, *, include_documents: bool = False, cache_dir: Path = DEFAULT_DOC_CACHE_DIR, force_fetch: bool = False, ) -> list[SourceChunk]: """Build source chunks from registry key refs plus curated notes.""" sources = load_sources(sources_path) notes = load_notes(notes_path, sources) chunks: dict[tuple[str, str, str, int], SourceChunk] = {} for source_id, source in sources.items(): key_refs = source.get("key_refs", []) if not isinstance(key_refs, list): raise SourceRetrievalError(f"{source_id} key_refs must be a list") topics = [str(topic) for topic in source.get("topics", [])] for raw_location in key_refs: location = str(raw_location) note = notes.get((source_id, location), {}) summary = str(note.get("summary") or registry_summary(source, location)) chunk_kind = "note" if note else "registry" chunks[(source_id, location, chunk_kind, 0)] = SourceChunk( source_id=source_id, chunk_kind=chunk_kind, title=str(source.get("title", source_id)), url=str(source.get("url", "")), publisher=str(source.get("publisher", "")), source_type=str(source.get("type", "")), location=location, paragraph_index=0, topics=topics, summary=summary, text=summary, keywords=list(note.get("keywords") or registry_keywords(source, location)), ) for (source_id, location), note in notes.items(): if (source_id, location, "note", 0) in chunks: continue source = sources[source_id] topics = [str(topic) for topic in source.get("topics", [])] chunks[(source_id, location, "note", 0)] = SourceChunk( source_id=source_id, chunk_kind="note", title=str(source.get("title", source_id)), url=str(source.get("url", "")), publisher=str(source.get("publisher", "")), source_type=str(source.get("type", "")), location=location, paragraph_index=0, topics=topics, summary=str(note["summary"]), text=str(note["summary"]), keywords=list(note.get("keywords", [])), ) if include_documents: for chunk in document_chunks(sources, cache_dir, force_fetch): chunks[(chunk.source_id, chunk.location, chunk.chunk_kind, chunk.paragraph_index)] = chunk return sorted(chunks.values(), key=lambda chunk: (chunk.source_id, chunk.chunk_kind, chunk.paragraph_index, chunk.location)) def connect(db_path: Path) -> sqlite3.Connection: """Open a SQLite connection for source retrieval.""" connection = sqlite3.connect(db_path) connection.row_factory = sqlite3.Row return connection @contextmanager def db_connection(db_path: Path) -> Any: """Open a source-retrieval SQLite connection and always close it.""" connection = connect(db_path) try: yield connection connection.commit() finally: connection.close() def init_source_db(connection: sqlite3.Connection) -> None: """Create source retrieval tables.""" connection.executescript( """ CREATE TABLE IF NOT EXISTS source_chunks ( id INTEGER PRIMARY KEY AUTOINCREMENT, source_id TEXT NOT NULL, chunk_kind TEXT NOT NULL, title TEXT NOT NULL, url TEXT NOT NULL, publisher TEXT NOT NULL, source_type TEXT NOT NULL, location TEXT NOT NULL, paragraph_index INTEGER NOT NULL, topics TEXT NOT NULL, summary TEXT NOT NULL, text TEXT NOT NULL, keywords TEXT NOT NULL, UNIQUE(source_id, location, chunk_kind, paragraph_index) ); CREATE TABLE IF NOT EXISTS source_chunk_embeddings ( chunk_id INTEGER PRIMARY KEY, vector TEXT NOT NULL, content_hash TEXT NOT NULL, embedder_signature TEXT NOT NULL, FOREIGN KEY(chunk_id) REFERENCES source_chunks(id) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS meta ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); CREATE VIRTUAL TABLE IF NOT EXISTS source_chunks_fts USING fts5( source_id, chunk_kind, title, location, topics, summary, text, keywords, content='source_chunks', content_rowid='id', tokenize='unicode61 remove_diacritics 2' ); """ ) def insert_chunk(connection: sqlite3.Connection, chunk: SourceChunk) -> None: """Insert one source chunk and its FTS row.""" cursor = connection.execute( """ INSERT INTO source_chunks ( source_id, chunk_kind, title, url, publisher, source_type, location, paragraph_index, topics, summary, text, keywords ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( chunk.source_id, chunk.chunk_kind, chunk.title, chunk.url, chunk.publisher, chunk.source_type, chunk.location, chunk.paragraph_index, json.dumps(chunk.topics, ensure_ascii=False), chunk.summary, chunk.text, json.dumps(chunk.keywords, ensure_ascii=False), ), ) rowid = cursor.lastrowid connection.execute( """ INSERT INTO source_chunks_fts ( rowid, source_id, chunk_kind, title, location, topics, summary, text, keywords ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( rowid, chunk.source_id, chunk.chunk_kind, chunk.title, chunk.location, " ".join(chunk.topics), chunk.summary, chunk.text, " ".join(chunk.keywords), ), ) def import_nolegraph_embed() -> Any: """Import nolegraph.embed from an installed package or the local repo checkout.""" if str(DEFAULT_NOLEGRAPH_SRC) not in sys.path and DEFAULT_NOLEGRAPH_SRC.exists(): sys.path.insert(0, str(DEFAULT_NOLEGRAPH_SRC)) try: from nolegraph import embed as nolegraph_embed except Exception as exc: if os.environ.get("SYNDERESIS_REQUIRE_NOLEGRAPH_EMBED", "").strip().lower() in {"1", "true", "yes", "on"}: raise SourceRetrievalError("could not import nolegraph.embed") from exc return SimpleNamespace( HashEmbedder=HashEmbedder, HarrierEmbedder=HarrierEmbedder, load_default_embedder=load_default_embedder, ) return nolegraph_embed def resolve_embedder(mode: str) -> Any | None: """Resolve an optional embedder using nolegraph's embedding interfaces.""" mode = mode.strip().lower() if mode in {"", "none", "off"}: return None nolegraph_embed = import_nolegraph_embed() if mode == "hash": return nolegraph_embed.HashEmbedder(dim=512) if mode == "harrier": try: return nolegraph_embed.HarrierEmbedder() except Exception as exc: raise SourceRetrievalError( "Harrier embedder failed to load; install sentence-transformers in this Python environment " "or run from the nolegraph environment that has it" ) from exc if mode == "auto": return nolegraph_embed.load_default_embedder() raise SourceRetrievalError("embedder must be one of: none, hash, harrier, auto") def content_hash(text: str) -> str: """Return a stable content hash.""" return hashlib.sha256(text.encode("utf-8")).hexdigest() def embedding_text(row: sqlite3.Row) -> str: """Return bounded text for the embedding model while preserving full DB text.""" value = f"{row['title']} {row['source_id']}:{row['location']} {row['chunk_kind']}\n{row['summary']}\n{row['text']}" if len(value) <= EMBED_TEXT_MAX_CHARS: return value return value[:EMBED_TEXT_MAX_CHARS].rsplit(" ", 1)[0].rstrip() def write_embeddings(connection: sqlite3.Connection, embedder: Any | None) -> None: """Encode and persist source chunk vectors when an embedder is configured.""" if embedder is None: return rows = connection.execute( "SELECT id, source_id, location, chunk_kind, title, summary, text FROM source_chunks ORDER BY id" ).fetchall() if not rows: return connection.execute("DELETE FROM source_chunk_embeddings") signature = str(embedder.signature) for offset in range(0, len(rows), EMBED_BATCH_SIZE): batch = rows[offset : offset + EMBED_BATCH_SIZE] texts = [embedding_text(row) for row in batch] vectors = embedder.encode(texts, is_query=False) for row, vector in zip(batch, vectors, strict=True): text = f"{row['summary']}\n{row['text']}" connection.execute( """ INSERT INTO source_chunk_embeddings(chunk_id, vector, content_hash, embedder_signature) VALUES (?, ?, ?, ?) """, (row["id"], json.dumps(vector), content_hash(text), signature), ) connection.execute( "INSERT OR REPLACE INTO meta(key, value) VALUES ('embedder_signature', ?)", (signature,), ) def build_source_db( db_path: Path, sources_path: Path = DEFAULT_SOURCES_PATH, notes_path: Path = DEFAULT_NOTES_PATH, *, include_documents: bool = False, cache_dir: Path = DEFAULT_DOC_CACHE_DIR, force_fetch: bool = False, embedder: str = "none", ) -> int: """Rebuild the source retrieval SQLite database and return the chunk count.""" chunks = load_chunks( sources_path, notes_path, include_documents=include_documents, cache_dir=cache_dir, force_fetch=force_fetch, ) resolved_embedder = resolve_embedder(embedder) db_path.parent.mkdir(parents=True, exist_ok=True) if db_path.exists(): db_path.unlink() with db_connection(db_path) as connection: connection.execute("PRAGMA foreign_keys=ON") init_source_db(connection) for chunk in chunks: insert_chunk(connection, chunk) write_embeddings(connection, resolved_embedder) return len(chunks) def query_terms(query: str) -> list[str]: """Extract and expand FTS-safe search terms.""" terms: list[str] = [] seen: set[str] = set() for raw_term in WORD_RE.findall(query.lower()): term = raw_term.strip("'") if len(term) < 2 or term in STOPWORDS: continue expanded = SYNONYMS.get(term, [term]) for candidate in expanded: normalized = re.sub(r"[^a-z0-9]", "", candidate.lower()) if len(normalized) < 2 or normalized in STOPWORDS or normalized in seen: continue terms.append(normalized) seen.add(normalized) if len(terms) >= MAX_SEARCH_TERMS: return terms return terms def fts_query(terms: list[str]) -> str: """Build a safe FTS5 OR query from normalized terms.""" return " OR ".join(f"{term}*" for term in terms) def parse_json_list(value: str) -> list[str]: """Parse a JSON list from the database.""" try: parsed = json.loads(value) except json.JSONDecodeError: return [] if isinstance(parsed, list): return [str(item) for item in parsed] return [] def vector_dot(left: list[float], right: list[float]) -> float: """Dot product for L2-normalized embedding vectors.""" return sum(a * b for a, b in zip(left, right, strict=False)) def stored_embedder_signature(db_path: Path) -> str | None: """Return the embedder signature stored in the source DB, if present.""" if not db_path.exists(): return None with db_connection(db_path) as connection: try: row = connection.execute("SELECT value FROM meta WHERE key = 'embedder_signature'").fetchone() except sqlite3.OperationalError: return None return None if row is None else str(row["value"]) def resolve_search_embedder(db_path: Path) -> Any | None: """Resolve an embedder for query vectors when the runtime allows it.""" signature = stored_embedder_signature(db_path) if signature is None: return None requested = os.environ.get("SYNDERESIS_SOURCE_EMBEDDER", "").strip().lower() if not requested: requested = "auto" if os.environ.get("NOLEGRAPH_AUTOLOAD_EMBEDDER", "").strip().lower() in {"1", "true", "yes", "on"} else "none" if requested in {"", "none", "off"}: return None cache_key = ( signature, requested, os.environ.get("NOLEGRAPH_EMBEDDING_MODEL", DEFAULT_EMBEDDING_MODEL), os.environ.get("NOLEGRAPH_AUTOLOAD_EMBEDDER", ""), ) cache = getattr(resolve_search_embedder, "_cache", {}) if cache_key in cache: return cache[cache_key] try: embedder = resolve_embedder(requested) except SourceRetrievalError as error: print(f"source_retrieval: search embedder failed, falling back to lexical search: {error}", file=sys.stderr) cache[cache_key] = None setattr(resolve_search_embedder, "_cache", cache) return None if embedder is None or str(embedder.signature) != signature: cache[cache_key] = None setattr(resolve_search_embedder, "_cache", cache) return None cache[cache_key] = embedder setattr(resolve_search_embedder, "_cache", cache) return embedder def semantic_search(db_path: Path, query: str, limit: int) -> list[tuple[int, float]]: """Return chunk ids and semantic scores using stored embeddings.""" embedder = resolve_search_embedder(db_path) if embedder is None: return [] query_vector = embedder.encode([query], is_query=True)[0] with db_connection(db_path) as connection: try: rows = connection.execute("SELECT chunk_id, vector FROM source_chunk_embeddings").fetchall() except sqlite3.OperationalError: return [] scored: list[tuple[int, float]] = [] for row in rows: try: vector = [float(value) for value in json.loads(row["vector"])] except (TypeError, ValueError, json.JSONDecodeError): continue if len(vector) != len(query_vector): continue scored.append((int(row["chunk_id"]), vector_dot(query_vector, vector))) scored.sort(key=lambda item: item[1], reverse=True) return scored[: max(limit * 4, 24)] def token_set(*values: str) -> set[str]: """Return lowercase tokens from one or more strings.""" return {token.strip("'") for value in values for token in WORD_RE.findall(value.lower())} def overlap_score(terms: list[str], row: sqlite3.Row) -> float: """Compute a small rerank score to prefer semantically dense matches.""" topics = " ".join(parse_json_list(row["topics"])) keywords = " ".join(parse_json_list(row["keywords"])) haystack = token_set(row["source_id"], row["title"], row["location"], topics, row["summary"], keywords) keyword_tokens = token_set(keywords) score = 0.0 for term in terms: matched = any(token.startswith(term) or term.startswith(token) for token in haystack) if matched: score += 1.0 keyword_matched = any(token.startswith(term) or term.startswith(token) for token in keyword_tokens) if keyword_matched: score += 0.75 return score def normalize_search_filters(filters: dict[str, Any] | None) -> dict[str, set[str]]: """Normalize optional API search filters.""" if not filters: return {} normalized: dict[str, set[str]] = {} for key in ("source_ids", "source_types", "publishers", "chunk_kinds", "topics"): raw_values = filters.get(key, []) if isinstance(raw_values, str): raw_values = [raw_values] if not isinstance(raw_values, list): continue values = {str(value).strip().casefold() for value in raw_values if str(value).strip()} if values: normalized[key] = values return normalized def row_matches_search_filters(row: sqlite3.Row, filters: dict[str, set[str]]) -> bool: """Return whether a source row satisfies normalized filters.""" if not filters: return True if "source_ids" in filters and str(row["source_id"]).casefold() not in filters["source_ids"]: return False if "source_types" in filters and str(row["source_type"]).casefold() not in filters["source_types"]: return False if "publishers" in filters and str(row["publisher"]).casefold() not in filters["publishers"]: return False if "chunk_kinds" in filters and str(row["chunk_kind"]).casefold() not in filters["chunk_kinds"]: return False if "topics" in filters: row_topics = {topic.casefold() for topic in parse_json_list(row["topics"])} if row_topics.isdisjoint(filters["topics"]): return False return True def result_from_row(row: sqlite3.Row, score: float) -> SearchResult: """Build a SearchResult from a source_chunks row.""" return SearchResult( source_id=row["source_id"], chunk_kind=row["chunk_kind"], title=row["title"], url=row["url"], publisher=row["publisher"], source_type=row["source_type"], location=row["location"], paragraph_index=int(row["paragraph_index"]), topics=parse_json_list(row["topics"]), summary=row["summary"], text=row["text"], keywords=parse_json_list(row["keywords"]), score=round(score, 6), ) def search_source_chunks(db_path: Path, query: str, limit: int = 6, filters: dict[str, Any] | None = None) -> list[SearchResult]: """Search source chunks by query content.""" if limit <= 0: return [] if not db_path.exists(): raise SourceRetrievalError(f"source retrieval database not found: {db_path}") terms = query_terms(query) if not terms: return [] expression = fts_query(terms) normalized_filters = normalize_search_filters(filters) candidate_limit = max(limit * 40, 200) if normalized_filters else max(limit * 8, 24) semantic_scores = dict(semantic_search(db_path, query, candidate_limit if normalized_filters else limit)) with db_connection(db_path) as connection: rows = connection.execute( """ SELECT c.id, c.source_id, c.chunk_kind, c.title, c.url, c.publisher, c.source_type, c.location, c.paragraph_index, c.topics, c.summary, c.text, c.keywords, bm25(source_chunks_fts) AS bm25_rank FROM source_chunks_fts JOIN source_chunks c ON c.id = source_chunks_fts.rowid WHERE source_chunks_fts MATCH ? ORDER BY bm25_rank LIMIT ? """, (expression, candidate_limit), ).fetchall() semantic_rows: list[sqlite3.Row] = [] if semantic_scores: placeholders = ",".join("?" for _ in semantic_scores) semantic_rows = connection.execute( f""" SELECT id, source_id, chunk_kind, title, url, publisher, source_type, location, paragraph_index, topics, summary, text, keywords FROM source_chunks WHERE id IN ({placeholders}) """, tuple(semantic_scores), ).fetchall() results: list[SearchResult] = [] seen_ids: set[int] = set() for row in rows: if not row_matches_search_filters(row, normalized_filters): continue chunk_id = int(row["id"]) seen_ids.add(chunk_id) semantic_score = overlap_score(terms, row) bm25_rank = float(row["bm25_rank"]) score = semantic_score + max(0.0, -bm25_rank) + semantic_scores.get(chunk_id, 0.0) * 20.0 results.append(result_from_row(row, score)) for row in semantic_rows: if not row_matches_search_filters(row, normalized_filters): continue chunk_id = int(row["id"]) if chunk_id in seen_ids: continue score = semantic_scores.get(chunk_id, 0.0) * 20.0 results.append(result_from_row(row, score)) results.sort(key=lambda item: (-item.score, item.source_id, item.location)) max_per_source = max(3, limit // 2) selected: list[SearchResult] = [] selected_refs: set[tuple[str, str]] = set() source_counts: dict[str, int] = {} overflow: list[SearchResult] = [] for result in results: ref_key = (result.source_id, result.location) if ref_key in selected_refs: continue if source_counts.get(result.source_id, 0) >= max_per_source: overflow.append(result) continue selected.append(result) selected_refs.add(ref_key) source_counts[result.source_id] = source_counts.get(result.source_id, 0) + 1 if len(selected) >= limit: return selected for result in overflow: ref_key = (result.source_id, result.location) if ref_key in selected_refs: continue selected.append(result) selected_refs.add(ref_key) if len(selected) >= limit: break return selected def parse_args() -> argparse.Namespace: """Parse CLI arguments.""" parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--db-path", default=str(DEFAULT_DB_PATH)) parser.add_argument("--sources-path", default=str(DEFAULT_SOURCES_PATH)) parser.add_argument("--notes-path", default=str(DEFAULT_NOTES_PATH)) parser.add_argument("--cache-dir", default=str(DEFAULT_DOC_CACHE_DIR)) parser.add_argument("--build", action="store_true", help="Build or rebuild the SQLite retrieval database") parser.add_argument("--include-documents", action="store_true", help="Fetch official source pages and add paragraph-level chunks") parser.add_argument("--force-fetch", action="store_true", help="Refresh cached HTML documents") parser.add_argument("--embedder", choices=["none", "hash", "harrier", "auto"], default="harrier", help="Optional vector embedder for chunk embeddings") parser.add_argument("--search", default="", help="Search query to run after opening the database") parser.add_argument("--limit", type=int, default=6) parser.add_argument("--json", action="store_true", help="Print search results as JSON") return parser.parse_args() def main() -> int: """Build or search the source retrieval database.""" args = parse_args() db_path = Path(args.db_path) if args.build: count = build_source_db( db_path, Path(args.sources_path), Path(args.notes_path), include_documents=args.include_documents, cache_dir=Path(args.cache_dir), force_fetch=args.force_fetch, embedder=args.embedder, ) print(json.dumps({"db_path": str(db_path), "chunk_count": count}, indent=2)) if args.search: results = search_source_chunks(db_path, args.search, args.limit) if args.json: print(json.dumps([result.to_dict() for result in results], indent=2, ensure_ascii=False)) else: for result in results: print(f"{result.source_id}:{result.location}\t{result.score:.3f}\t{result.title}\t{result.summary}") if not args.build and not args.search: raise SystemExit("provide --build and/or --search") return 0 if __name__ == "__main__": raise SystemExit(main())