"""Import A9-format directories into the Brain of Reese knowledge base. Examples:: uv run python -m scripts.import_docs # BOR_GIT_SOURCES, else ~/Homelab + ~/Deployments uv run python -m scripts.import_docs --source ~/OtherDocs # explicit dir(s); always wins uv run python -m scripts.import_docs --prune # also drop deleted/out-of-scope files uv run python -m scripts.import_docs --limit 5 # debug: first 5 files only Source resolution (phase 28, extended in phases 35 and 38), in precedence order: 1. ``--source PATH`` — explicit manual directories (repeatable) always win; git and local sources are ignored when this flag is used. 2. The effective sources — the admin-managed ``git_sources`` table rows (both kinds), else the ``BOR_GIT_SOURCES`` (comma-separated) git-only fallback (:func:`app.rag.git_sources.effective_sources`, the same shared resolver the in-app Sync button uses). Git rows are cloned (first run, shallow ``--depth 1``) or fast-forwarded (``git pull --ff-only``) into ``BOR_SOURCES_DIR//`` (default ``~/bor-sources``); local rows are the existing directories themselves, walked directly. A failing clone/pull — or a local directory that is missing at run time — aborts the whole run *before* anything is imported. 3. Fallback — the legacy ``DEFAULT_SOURCES`` (``~/Homelab`` + ``~/Deployments``), kept for backwards compatibility (reached only while both the table and ``BOR_GIT_SOURCES`` are empty). Imported formats (PLAN anchor A9, revised): ``md, markdown, txt, yaml, yml, json, py`` (case-insensitive; narrow with ``BOR_IMPORT_EXTENSIONS``). Any path with a dot-prefixed component (hidden files/dirs — vendored caches) is skipped, along with non-content dirs (``.venv``, ``node_modules``, ``.git``, ``__pycache``, ``.pytest_cache``, ``dist``, ``build``). Re-runs are cheap: files are diffed by sha256 and unchanged ones are not re-embedded; ``--prune`` also drops documents whose files no longer match the format filter. After a run that **changed** the knowledge base (at least one document added or updated — or no outline stored yet), the single ``kb_overview`` row is regenerated with the ``lite`` model (phase 31): the plain-text outline of the KB's basic categories that every chat turn injects into the system prompt as ````. The regeneration is **best-effort and change-gated** — unchanged re-imports and ``--limit`` debug runs never burn a ``lite`` call, and a ``lite`` failure only reports ``overview=failed`` on the summary line: the import's exit code is about files, and the previous outline stays (an old outline is better than none). """ from __future__ import annotations import argparse import asyncio import logging import re import sys from pathlib import Path from app.config import Settings, get_settings from app.core.debugging import configure_debugging from app.core.logging import configure_logging from app.db import SessionLocal from app.models import KbOverview from app.rag.git_sources import effective_sources from app.rag.importer import ImportSummary, import_sources from app.rag.llm import LLMClient from app.rag.overview import regenerate_overview from scripts.git_sync import GitSyncError, clone_or_pull logger = logging.getLogger("scripts.import_docs") DEFAULT_SOURCES: list[Path] = [Path("~/Homelab"), Path("~/Deployments")] def build_parser() -> argparse.ArgumentParser: p = argparse.ArgumentParser( prog="python -m scripts.import_docs", description="Import A9-format files (md/txt/yaml/json/py) into the knowledge base.", ) p.add_argument( "--source", action="append", type=Path, metavar="PATH", help=( "directory to import (repeatable; always wins over the git_sources " "DB rows and BOR_GIT_SOURCES; default when neither is given: " "~/Homelab ~/Deployments)" ), ) p.add_argument( "--prune", action="store_true", help="also delete documents whose files no longer exist or match the format filter", ) p.add_argument( "--limit", type=int, default=None, metavar="N", help="only process the first N files (debug; disables --prune)", ) return p def repo_name(url: str) -> str: """Local directory name for a git URL (phase 28). Strips a trailing ``.git`` and takes the basename after the last ``/`` (``:`` for scp-style ``git@host:repo.git`` URLs); falls back to a slug of the whole URL when no usable basename remains. """ name = url.strip() if name.endswith(".git"): name = name[: -len(".git")] base = name.rsplit("/", 1)[-1].rsplit(":", 1)[-1].strip() if base: return base slug = re.sub(r"[^A-Za-z0-9]+", "-", name).strip("-") return slug or "repo" def _resolve_sources(cli_sources: list[Path] | None, settings: Settings) -> list[Path]: """Resolve the directories to import (phase 28, extended in phases 35 and 38). Precedence: ``--source`` (explicit manual paths — always wins) > the effective sources — the ``git_sources`` DB rows (git + local), else the ``BOR_GIT_SOURCES`` git-only fallback (:func:`app.rag.git_sources.effective_sources`; the import needs the database anyway, so resolution opens a short session and there is no DB-down branch) — git rows cloned/pulled via :func:`scripts.git_sync.clone_or_pull` into ``BOR_SOURCES_DIR//``, local rows walked directly (the stored directory, re-verified ``.is_dir()`` at run time) > the legacy ``DEFAULT_SOURCES``. A :class:`GitSyncError` from a failing clone/pull — or a missing local directory (``local source missing: ``) — propagates to :func:`main`, which aborts the run before importing anything. """ if cli_sources: return [path.expanduser() for path in cli_sources] db = SessionLocal() try: rows, origin = effective_sources(db) finally: db.close() if rows: git_count = sum(1 for row in rows if row.kind == "git") logger.info( "sources: %d repo(s) git=%d local=%d origin=%s", len(rows), git_count, len(rows) - git_count, origin, ) sources_root = Path(settings.sources_dir).expanduser() sources: list[Path] = [] for row in rows: if row.kind == "git": sources.append(clone_or_pull(row.url, sources_root / repo_name(row.url))) else: # kind=local — the stored expanded path (phase 38 also # mirrors it in the NOT-NULL ``url`` location column, the # ``or`` keeps the type checker honest); a missing # directory aborts before importing, the same pre-import # fail-loud as a failing git clone. path = Path(row.path or row.url).expanduser() if not path.is_dir(): raise GitSyncError(f"local source missing: {path}") sources.append(path) return sources return [path.expanduser() for path in DEFAULT_SOURCES] def _overview_row_exists() -> bool: """Whether the single ``kb_overview`` row (id = 1) is present. One indexed PK lookup (phase 31, task 04): a missing outline after an unchanged re-import — e.g. the first run after migration 0005 — still gets a fresh outline, while a present one is left untouched until the KB actually changes. """ with SessionLocal() as session: return session.get(KbOverview, 1) is not None def main(argv: list[str] | None = None) -> int: args = build_parser().parse_args(argv) settings = get_settings() configure_logging(settings.log_level) configure_debugging() # Git sources resolve (and clone/pull) *before* any import: a failing # repo aborts the run with a non-zero exit, naming the failure — a bad # URL must never silently import partial junk. try: sources = _resolve_sources(args.source, settings) except GitSyncError as e: print(f"import_docs: source sync failed: {e}", file=sys.stderr) return 1 logger.info( "import_docs: importing %d source dir(s): %s", len(sources), ", ".join(str(s) for s in sources), ) missing = [s for s in sources if not s.is_dir()] for s in missing: print(f"import_docs: source dir not found: {s}", file=sys.stderr) if all(not s.is_dir() for s in sources): print("import_docs: no source directories found — nothing to do.", file=sys.stderr) return 1 llm = LLMClient() async def _run() -> tuple[ImportSummary, str]: """Import, then (change-gated) refresh the stored KB overview. One event loop, one ``LLMClient`` (phase 31, task 04): the outline that every chat prompt injects as ```` is regenerated only when this run added/updated at least one document — or when no outline exists yet after a walk that actually saw files (e.g. the first run after migration 0005). ``--limit`` debug runs (an incomplete walk must not rewrite the outline — mirrors the ``--prune``-with-``--limit`` guard) and unchanged re-imports never burn a ``lite`` call, and a ``lite`` failure only flips the status token (``failed``) — the import's exit code is unchanged. """ summary = await import_sources(sources, llm, prune=args.prune, limit=args.limit) if args.limit is not None: logger.info("overview: skipped (--limit)") return summary, "skipped" if summary.added + summary.updated == 0: if summary.files == 0: logger.info("overview: skipped (nothing imported)") return summary, "skipped" if _overview_row_exists(): logger.info("overview: skipped (KB unchanged)") return summary, "skipped" # No outline yet after an unchanged re-import (e.g. the first # run after migration 0005) — fall through and generate one. ok = await regenerate_overview(llm) return summary, "updated" if ok else "failed" summary, overview_status = asyncio.run(_run()) print( f"import_docs: files={summary.files} added={summary.added} " f"updated={summary.updated} unchanged={summary.unchanged} " f"pruned={summary.pruned} errors={summary.errors} chunks={summary.chunks} " f"embed_batches={summary.embed_batches} summaries={summary.summaries} " f"summary_errors={summary.summary_errors} formats={summary.format_counts()} " f"overview={overview_status}" ) # Non-zero if any file failed, so cron/CI notice — the rest of the KB # was imported and the failed files are retried on the next run. The # overview is best-effort: a failed outline never changes the exit code. return 1 if summary.errors else 0 if __name__ == "__main__": sys.exit(main())