| 1 | """Deterministic, local-only document corpus source. |
| 2 | |
| 3 | The corpus adapter deliberately has no HTTP dependency. It scans explicitly |
| 4 | registered directories, extracts small text documents (and PDFs only when the |
| 5 | local ``pdftotext`` binary is available), and returns normalized ``SourceItem`` |
| 6 | objects for the shared relevance/fusion pipeline. |
| 7 | """ |
| 8 | |
| 9 | from __future__ import annotations |
| 10 | |
| 11 | import hashlib |
| 12 | import json |
| 13 | import os |
| 14 | import subprocess |
| 15 | import threading |
| 16 | from dataclasses import dataclass, field |
| 17 | from datetime import datetime, timezone |
| 18 | from pathlib import Path |
| 19 | from shutil import which |
| 20 | from typing import Any, Iterable |
| 21 | |
| 22 | from . import entity_extract, log, relevance, schema |
| 23 | |
| 24 | SOURCE = "corpus" |
| 25 | SUPPORTED_SUFFIXES = {".md", ".txt", ".pdf"} |
| 26 | IGNORED_DIRECTORIES = {".git", "node_modules"} |
| 27 | MAX_FILES = 500 |
| 28 | MAX_TEXT_CHARS = 1_000_000 |
| 29 | MAX_CACHE_TEXT_CHARS = MAX_TEXT_CHARS |
| 30 | MAX_CACHE_BYTES = 50 * 1024 * 1024 |
| 31 | MAX_CACHE_ENTRIES = 2_000 |
| 32 | CACHE_FILENAME = "corpus-cache.json" |
| 33 | CACHE_SCHEMA_VERSION = "last30days-corpus-cache/v2" |
| 34 | |
| 35 | _CACHE_LOCK = threading.Lock() |
| 36 | |
| 37 | |
| 38 | @dataclass |
| 39 | class CorpusScanResult: |
| 40 | """One bounded scan, including non-fatal extraction notes.""" |
| 41 | |
| 42 | items: list[schema.SourceItem] |
| 43 | notes: list[str] = field(default_factory=list) |
| 44 | files_scanned: int = 0 |
| 45 | cache_hits: int = 0 |
| 46 | |
| 47 | |
| 48 | def resolve_directories( |
| 49 | cli_directories: Iterable[str] | None, |
| 50 | configured: str | Iterable[str] | None, |
| 51 | ) -> list[Path]: |
| 52 | """Merge repeatable CLI paths with ``os.pathsep``-separated config paths.""" |
| 53 | raw: list[str] = [str(value) for value in (cli_directories or []) if str(value).strip()] |
| 54 | if isinstance(configured, str): |
| 55 | raw.extend(value for value in configured.split(os.pathsep) if value.strip()) |
| 56 | elif configured: |
| 57 | raw.extend(str(value) for value in configured if str(value).strip()) |
| 58 | |
| 59 | resolved: list[Path] = [] |
| 60 | seen: set[str] = set() |
| 61 | for value in raw: |
| 62 | path = Path(value.strip()).expanduser().resolve() |
| 63 | key = os.path.normcase(str(path)) |
| 64 | if key in seen: |
| 65 | continue |
| 66 | seen.add(key) |
| 67 | resolved.append(path) |
| 68 | return resolved |
| 69 | |
| 70 | |
| 71 | def _safe_error(exc: BaseException) -> str: |
| 72 | """Describe an error without str(exc), which embeds absolute paths. |
| 73 | |
| 74 | These notes travel into source_status detail and render in coverage |
| 75 | diagnostics outside the private corpus block. |
| 76 | """ |
| 77 | reason = getattr(exc, "strerror", None) |
| 78 | return str(reason) if reason else exc.__class__.__name__ |
| 79 | |
| 80 | |
| 81 | def search( |
| 82 | topic: str, |
| 83 | directories: Iterable[Path | str], |
| 84 | *, |
| 85 | from_date: str, |
| 86 | to_date: str, |
| 87 | all_time: bool = False, |
| 88 | limit: int = 12, |
| 89 | cache_dir: Path | None = None, |
| 90 | ) -> CorpusScanResult: |
| 91 | """Search registered directories without making any network calls.""" |
| 92 | roots = resolve_directories([str(path) for path in directories], None) |
| 93 | notes: list[str] = [] |
| 94 | cache_path = cache_dir / CACHE_FILENAME if cache_dir is not None else None |
| 95 | with _CACHE_LOCK: |
| 96 | cache = _load_cache(cache_path) |
| 97 | cache_entries = cache.setdefault("entries", {}) |
| 98 | cache_entry_sizes = { |
| 99 | path: _cache_entry_fragment_size(path, value) |
| 100 | for path, value in cache_entries.items() |
| 101 | } |
| 102 | |
| 103 | candidates: list[tuple[float, int, schema.SourceItem]] = [] |
| 104 | seen_files: set[str] = set() |
| 105 | files_scanned = 0 |
| 106 | cache_hits = 0 |
| 107 | pdf_available = which("pdftotext") |
| 108 | pdf_unavailable_noted = False |
| 109 | |
| 110 | readable_roots: list[Path] = [] |
| 111 | for root in roots: |
| 112 | if not root.is_dir(): |
| 113 | notes.append(f"Skipped corpus root '{Path(root).name}': not a readable directory") |
| 114 | continue |
| 115 | readable_roots.append(root) |
| 116 | |
| 117 | per_root_limit, extra_slots = divmod(MAX_FILES, len(readable_roots) or 1) |
| 118 | scan_limit_reached = False |
| 119 | for root_index, root in enumerate(readable_roots): |
| 120 | root_limit = per_root_limit + (1 if root_index < extra_slots else 0) |
| 121 | root_files_scanned = 0 |
| 122 | for path in _iter_files(root, notes=notes): |
| 123 | if root_files_scanned >= root_limit: |
| 124 | scan_limit_reached = True |
| 125 | break |
| 126 | key = os.path.normcase(str(path)) |
| 127 | if key in seen_files: |
| 128 | continue |
| 129 | seen_files.add(key) |
| 130 | root_files_scanned += 1 |
| 131 | files_scanned += 1 |
| 132 | |
| 133 | try: |
| 134 | stat = path.stat() |
| 135 | except OSError as exc: |
| 136 | notes.append(f"Skipped {_display_path(path, root)}: {_safe_error(exc)}") |
| 137 | continue |
| 138 | published_at = datetime.fromtimestamp( |
| 139 | stat.st_mtime, tz=timezone.utc |
| 140 | ).date().isoformat() |
| 141 | if not all_time and not (from_date <= published_at <= to_date): |
| 142 | continue |
| 143 | |
| 144 | cached = cache_entries.get(str(path)) |
| 145 | if ( |
| 146 | isinstance(cached, dict) |
| 147 | and cached.get("mtime_ns") == stat.st_mtime_ns |
| 148 | and cached.get("size") == stat.st_size |
| 149 | and isinstance(cached.get("text"), str) |
| 150 | ): |
| 151 | text = cached["text"] |
| 152 | cache_hits += 1 |
| 153 | else: |
| 154 | if path.suffix.lower() == ".pdf" and not pdf_available: |
| 155 | if not pdf_unavailable_noted: |
| 156 | notes.append("Skipped PDF files because pdftotext is not on PATH") |
| 157 | pdf_unavailable_noted = True |
| 158 | continue |
| 159 | try: |
| 160 | text = _extract_text(path, pdftotext=pdf_available) |
| 161 | except (OSError, subprocess.SubprocessError) as exc: |
| 162 | notes.append(f"Skipped {_display_path(path, root)}: {_safe_error(exc)}") |
| 163 | continue |
| 164 | _cache_entry_put(cache_entries, cache_entry_sizes, str(path), { |
| 165 | "mtime_ns": stat.st_mtime_ns, |
| 166 | "size": stat.st_size, |
| 167 | "text": text[:MAX_CACHE_TEXT_CHARS], |
| 168 | }) |
| 169 | |
| 170 | title = _path_title(path) |
| 171 | score = _match_score(topic, f"{title}\n{text}") |
| 172 | if score < 0.15: |
| 173 | continue |
| 174 | relative_path = str(path.relative_to(root)) |
| 175 | path_digest = hashlib.sha256(str(path).encode("utf-8")).hexdigest() |
| 176 | item = schema.SourceItem( |
| 177 | item_id=f"C{path_digest[:12]}", |
| 178 | source=SOURCE, |
| 179 | title=title, |
| 180 | body=text, |
| 181 | url=f"corpus://{path_digest}", |
| 182 | container=str(path.parent), |
| 183 | published_at=published_at, |
| 184 | date_confidence="high", |
| 185 | relevance_hint=score, |
| 186 | why_relevant=f"Matched local file {relative_path}", |
| 187 | # Leave empty so extract_best_snippet derives the matching |
| 188 | # window; a file-prefix snippet is preserved verbatim and can |
| 189 | # show unrelated intro text (and draw entity-miss demotion). |
| 190 | snippet="", |
| 191 | metadata={ |
| 192 | "path": str(path), |
| 193 | "relative_path": relative_path, |
| 194 | "extension": path.suffix.lower(), |
| 195 | "local_only": True, |
| 196 | }, |
| 197 | ) |
| 198 | candidates.append((score, stat.st_mtime_ns, item)) |
| 199 | if scan_limit_reached: |
| 200 | notes.append(f"Stopped after the {MAX_FILES}-file corpus scan limit") |
| 201 | |
| 202 | cache["schema_version"] = CACHE_SCHEMA_VERSION |
| 203 | cache["entries"] = _bounded_entries(cache_entries) |
| 204 | with _CACHE_LOCK: |
| 205 | _write_cache(cache_path, cache, notes) |
| 206 | |
| 207 | candidates.sort(key=lambda row: (-row[0], -row[1], row[2].title.casefold())) |
| 208 | items = [item for _score, _mtime, item in candidates[: max(0, limit)]] |
| 209 | log.source_log( |
| 210 | "Corpus", |
| 211 | f"scanned {files_scanned} file(s), {cache_hits} cache hit(s), {len(items)} match(es)", |
| 212 | tty_only=False, |
| 213 | ) |
| 214 | return CorpusScanResult( |
| 215 | items=items, |
| 216 | notes=notes, |
| 217 | files_scanned=files_scanned, |
| 218 | cache_hits=cache_hits, |
| 219 | ) |
| 220 | |
| 221 | |
| 222 | def _display_path(path: Path | str, root: Path | None = None) -> str: |
| 223 | """Render a note-safe path: never the absolute local path. |
| 224 | |
| 225 | Corpus notes flow into source_status detail and the Partial Coverage |
| 226 | block, which render OUTSIDE the private corpus markers - an absolute |
| 227 | path like /home/user/private/notes/foo.md must not escape there. |
| 228 | """ |
| 229 | candidate = Path(path) |
| 230 | if root is not None: |
| 231 | try: |
| 232 | return str(Path(root).name / candidate.relative_to(root)) |
| 233 | except ValueError: |
| 234 | pass |
| 235 | return candidate.name |
| 236 | |
| 237 | |
| 238 | def _iter_files(root: Path, notes: list[str] | None = None) -> Iterable[Path]: |
| 239 | # Bounded newest-first selection: keep only the newest MAX_FILES paths in a |
| 240 | # heap while walking, so registering a huge tree does not materialize every |
| 241 | # path before the caller's extraction cap applies. |
| 242 | import heapq |
| 243 | |
| 244 | heap: list[tuple[int, str]] = [] |
| 245 | walk_errors = 0 |
| 246 | |
| 247 | def _on_walk_error(error: OSError) -> None: |
| 248 | nonlocal walk_errors |
| 249 | walk_errors += 1 |
| 250 | if notes is not None and walk_errors <= 3: |
| 251 | unreadable = _display_path(error.filename, root) if error.filename else Path(root).name |
| 252 | notes.append(f"corpus: could not read {unreadable}: {error.strerror}") |
| 253 | |
| 254 | for current, directory_names, file_names in os.walk( |
| 255 | root, followlinks=False, onerror=_on_walk_error |
| 256 | ): |
| 257 | directory_names[:] = sorted( |
| 258 | name |
| 259 | for name in directory_names |
| 260 | if name not in IGNORED_DIRECTORIES and not name.startswith(".") |
| 261 | ) |
| 262 | current_path = Path(current) |
| 263 | for name in sorted(file_names): |
| 264 | if name.startswith("."): |
| 265 | continue |
| 266 | path = current_path / name |
| 267 | if path.suffix.lower() in SUPPORTED_SUFFIXES and not path.is_symlink(): |
| 268 | entry = (_safe_mtime_ns(path), str(path)) |
| 269 | if len(heap) < MAX_FILES: |
| 270 | heapq.heappush(heap, entry) |
| 271 | else: |
| 272 | heapq.heappushpop(heap, entry) |
| 273 | if notes is not None and walk_errors > 3: |
| 274 | notes.append(f"corpus: {walk_errors - 3} more unreadable directories suppressed") |
| 275 | ordered = sorted(heap, key=lambda item: (-item[0], item[1].casefold())) |
| 276 | for _mtime, raw_path in ordered: |
| 277 | yield Path(raw_path) |
| 278 | |
| 279 | |
| 280 | def _safe_mtime_ns(path: Path) -> int: |
| 281 | try: |
| 282 | return path.stat().st_mtime_ns |
| 283 | except OSError: |
| 284 | return 0 |
| 285 | |
| 286 | |
| 287 | def _extract_text(path: Path, *, pdftotext: str | None) -> str: |
| 288 | if path.suffix.lower() == ".pdf": |
| 289 | if not pdftotext: |
| 290 | return "" |
| 291 | completed = subprocess.run( |
| 292 | [pdftotext, str(path), "-"], |
| 293 | capture_output=True, |
| 294 | check=True, |
| 295 | text=True, |
| 296 | timeout=20, |
| 297 | ) |
| 298 | return completed.stdout[:MAX_TEXT_CHARS] |
| 299 | with path.open("r", encoding="utf-8", errors="replace") as handle: |
| 300 | return handle.read(MAX_TEXT_CHARS) |
| 301 | |
| 302 | |
| 303 | def _path_title(path: Path) -> str: |
| 304 | title = path.stem.replace("_", " ").replace("-", " ") |
| 305 | return " ".join(title.split()) or path.name |
| 306 | |
| 307 | |
| 308 | def _match_score(topic: str, text: str) -> float: |
| 309 | lexical = relevance.token_overlap_relevance(topic, text) |
| 310 | topic_entities = entity_extract.extract_text_entities(topic) |
| 311 | text_entities = entity_extract.extract_text_entities(text) |
| 312 | entity_score = entity_extract.entity_overlap(topic_entities, text_entities) |
| 313 | return round(max(lexical, entity_score * 0.9), 4) |
| 314 | |
| 315 | |
| 316 | def _load_cache(path: Path | None) -> dict[str, Any]: |
| 317 | if path is None: |
| 318 | return {"schema_version": CACHE_SCHEMA_VERSION, "entries": {}} |
| 319 | try: |
| 320 | if path.stat().st_size > MAX_CACHE_BYTES: |
| 321 | return {"schema_version": CACHE_SCHEMA_VERSION, "entries": {}} |
| 322 | payload = json.loads(path.read_text(encoding="utf-8")) |
| 323 | except (OSError, UnicodeError, json.JSONDecodeError): |
| 324 | return {"schema_version": CACHE_SCHEMA_VERSION, "entries": {}} |
| 325 | if not isinstance(payload, dict) or payload.get("schema_version") != CACHE_SCHEMA_VERSION: |
| 326 | return {"schema_version": CACHE_SCHEMA_VERSION, "entries": {}} |
| 327 | if not isinstance(payload.get("entries"), dict): |
| 328 | payload["entries"] = {} |
| 329 | payload["entries"] = _bounded_entries(payload["entries"]) |
| 330 | return payload |
| 331 | |
| 332 | |
| 333 | def _bounded_entries(entries: Any) -> dict[str, Any]: |
| 334 | if not isinstance(entries, dict): |
| 335 | return {} |
| 336 | ordered = sorted( |
| 337 | ( |
| 338 | (path, value) |
| 339 | for path, value in entries.items() |
| 340 | if ( |
| 341 | isinstance(path, str) |
| 342 | and isinstance(value, dict) |
| 343 | and isinstance(value.get("text"), str) |
| 344 | ) |
| 345 | ), |
| 346 | key=lambda row: int(row[1].get("mtime_ns") or 0), |
| 347 | reverse=True, |
| 348 | ) |
| 349 | base_bytes = len( |
| 350 | json.dumps( |
| 351 | {"schema_version": CACHE_SCHEMA_VERSION, "entries": {}}, |
| 352 | ensure_ascii=False, |
| 353 | ).encode("utf-8") |
| 354 | ) |
| 355 | used_bytes = base_bytes |
| 356 | bounded: dict[str, Any] = {} |
| 357 | for path, value in ordered[:MAX_CACHE_ENTRIES]: |
| 358 | normalized = { |
| 359 | "mtime_ns": value.get("mtime_ns"), |
| 360 | "size": value.get("size"), |
| 361 | "text": value["text"][:MAX_CACHE_TEXT_CHARS], |
| 362 | } |
| 363 | fragment = json.dumps({path: normalized}, ensure_ascii=False).encode("utf-8") |
| 364 | fragment_bytes = len(fragment) - 2 + (2 if bounded else 0) |
| 365 | if used_bytes + fragment_bytes > MAX_CACHE_BYTES: |
| 366 | continue |
| 367 | bounded[path] = normalized |
| 368 | used_bytes += fragment_bytes |
| 369 | return bounded |
| 370 | |
| 371 | |
| 372 | def _cache_entry_fragment_size(path: str, value: dict[str, Any]) -> int: |
| 373 | return len(json.dumps({path: value}, ensure_ascii=False).encode("utf-8")) - 2 |
| 374 | |
| 375 | |
| 376 | def _cache_entry_put( |
| 377 | entries: dict[str, Any], |
| 378 | sizes: dict[str, int], |
| 379 | path: str, |
| 380 | value: dict[str, Any], |
| 381 | ) -> None: |
| 382 | entries[path] = value |
| 383 | sizes[path] = _cache_entry_fragment_size(path, value) |
| 384 | while ( |
| 385 | len(entries) > MAX_CACHE_ENTRIES |
| 386 | or _cache_payload_size(sizes) > MAX_CACHE_BYTES |
| 387 | ): |
| 388 | oldest = min( |
| 389 | entries, |
| 390 | key=lambda candidate: ( |
| 391 | int(entries[candidate].get("mtime_ns") or 0), |
| 392 | candidate, |
| 393 | ), |
| 394 | ) |
| 395 | del entries[oldest] |
| 396 | del sizes[oldest] |
| 397 | |
| 398 | |
| 399 | def _cache_payload_size(sizes: dict[str, int]) -> int: |
| 400 | base_bytes = len( |
| 401 | json.dumps( |
| 402 | {"schema_version": CACHE_SCHEMA_VERSION, "entries": {}}, |
| 403 | ensure_ascii=False, |
| 404 | ).encode("utf-8") |
| 405 | ) |
| 406 | separators = max(0, len(sizes) - 1) * 2 |
| 407 | return base_bytes + sum(sizes.values()) + separators |
| 408 | |
| 409 | |
| 410 | def _write_cache(path: Path | None, payload: dict[str, Any], notes: list[str]) -> None: |
| 411 | if path is None: |
| 412 | return |
| 413 | try: |
| 414 | _ensure_private_directory(path.parent) |
| 415 | payload["entries"] = _bounded_entries(payload.get("entries", {})) |
| 416 | encoded = json.dumps(payload, ensure_ascii=False).encode("utf-8") |
| 417 | temporary = path.with_name(f".{path.name}.{os.getpid()}.tmp") |
| 418 | try: |
| 419 | fd = os.open(temporary, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) |
| 420 | except FileExistsError: |
| 421 | temporary.unlink() |
| 422 | fd = os.open(temporary, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o600) |
| 423 | with os.fdopen(fd, "wb") as handle: |
| 424 | handle.write(encoded) |
| 425 | temporary.replace(path) |
| 426 | path.chmod(0o600) |
| 427 | except OSError as exc: |
| 428 | notes.append(f"Corpus cache unavailable: {_safe_error(exc)}") |
| 429 | |
| 430 | |
| 431 | def _ensure_private_directory(path: Path) -> None: |
| 432 | missing: list[Path] = [] |
| 433 | current = path |
| 434 | while not current.exists(): |
| 435 | missing.append(current) |
| 436 | current = current.parent |
| 437 | path.mkdir(parents=True, exist_ok=True, mode=0o700) |
| 438 | for directory in missing: |
| 439 | directory.chmod(0o700) |
| 440 |