ROS 2 docs RAG index and tooling

scripts/fetch_docs.py

@@ -0,0 +1,165 @@
+"""Fetch the documentation source for the configured project.
+
+Reads the RST rather than crawling the rendered site. Rendered pages carry
+navigation, version switchers and generated API listings, all of which land in
+chunks and compete with prose during retrieval; the RST is the authoritative
+text and is versioned and diffable.
+
+ROS 2 documents each distribution on its own branch, so --ref selects the
+distribution: rolling (default), jazzy, humble.
+
+    python scripts/fetch_docs.py
+    python scripts/fetch_docs.py --ref jazzy
+"""
+
+from __future__ import annotations
+
+import argparse
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+
+# The default Windows console is cp1252 and raises UnicodeEncodeError on any
+# non-Latin-1 character. These scripts print document titles and paths straight
+# from the Zephyr tree, which is full of them.
+if hasattr(sys.stdout, "reconfigure"):
+    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
+    sys.stderr.reconfigure(encoding="utf-8", errors="replace")
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+import source_config as cfg  # noqa: E402
+
+REPO = cfg.REPO
+ROOT = Path(__file__).resolve().parent.parent
+DEFAULT_CLONE = ROOT / "data" / "doc-src"
+DEFAULT_OUT = ROOT / "data" / "raw_docs"
+
+SKIP_DIRS = cfg.SKIP_DIRS
+
+
+def run(cmd: list[str], cwd: Path | None = None, check: bool = True) -> int:
+    result = subprocess.run(cmd, cwd=cwd, text=True)
+    if check and result.returncode != 0:
+        raise SystemExit(f"command failed ({result.returncode}): {' '.join(cmd)}")
+    return result.returncode
+
+
+def sparse_clone(clone_dir: Path, ref: str) -> None:
+    """Shallow + blobless + sparse: only doc/, only one commit."""
+    # An empty leftover directory is not a checkout. Testing `exists()` sent the
+    # reuse path at a directory with no .git in it, which then failed on fetch
+    # and again on removal, reporting a file lock that was never the problem.
+    if (clone_dir / ".git").exists():
+        # A --branch clone sets a narrow fetch refspec, so a bare `git fetch
+        # origin main` fails with "couldn't find remote ref". Ask for the ref
+        # explicitly, and treat the cache as disposable if anything goes wrong:
+        # it is a checkout of someone else's repository, not state worth
+        # rescuing.
+        print(f"Reusing existing checkout at {clone_dir}")
+        refspec = f"+refs/heads/{ref}:refs/remotes/origin/{ref}"
+        ok = run(["git", "fetch", "--depth", "1", "origin", refspec], cwd=clone_dir, check=False)
+        if ok == 0:
+            ok = run(["git", "checkout", "-f", "FETCH_HEAD"], cwd=clone_dir, check=False)
+        if ok == 0:
+            return
+        print("  cached checkout is unusable - re-cloning")
+        shutil.rmtree(clone_dir, ignore_errors=True)
+        if clone_dir.exists():
+            raise SystemExit(
+                f"could not remove {clone_dir} (a file may be locked). Delete it and retry."
+            )
+
+    clone_dir.parent.mkdir(parents=True, exist_ok=True)
+    print(f"Cloning {REPO} ({ref}, {cfg.DOC_ROOT}/ only) -> {clone_dir}")
+    run(
+        [
+            "git",
+            "clone",
+            "--depth",
+            "1",
+            "--filter=blob:none",
+            "--sparse",
+            "--branch",
+            ref,
+            REPO,
+            str(clone_dir),
+        ]
+    )
+    run(["git", "sparse-checkout", "set", cfg.DOC_ROOT], cwd=clone_dir)
+
+
+def collect(clone_dir: Path, out_dir: Path) -> tuple[int, int]:
+    """Copy documentation sources out of the checkout, flattened by path."""
+    doc_root = clone_dir / cfg.DOC_ROOT
+    if not doc_root.is_dir():
+        raise SystemExit(f"no {cfg.DOC_ROOT}/ directory in {clone_dir}")
+
+    if out_dir.exists():
+        shutil.rmtree(out_dir)
+    out_dir.mkdir(parents=True)
+
+    copied = 0
+    total_bytes = 0
+    for path in sorted(doc_root.rglob("*")):
+        if not path.is_file() or path.suffix.lower() not in {".rst", ".md", ".txt"}:
+            continue
+        relative = path.relative_to(doc_root)
+        if any(part in SKIP_DIRS for part in relative.parts):
+            continue
+
+        # Flatten so the source path survives as the filename. Retrieval cites
+        # the file, and "kernel/services/threads.rst" is a far more useful
+        # citation than "threads.rst" repeated across a dozen subsystems.
+        flat = str(relative).replace("\\", "/").replace("/", "__")
+        destination = out_dir / flat
+        destination.write_bytes(path.read_bytes())
+        copied += 1
+        total_bytes += destination.stat().st_size
+
+    return copied, total_bytes
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--ref",
+        default=cfg.DEFAULT_REF,
+        help=f"branch or tag to fetch (default: {cfg.DEFAULT_REF})",
+    )
+    parser.add_argument("--clone", type=Path, default=DEFAULT_CLONE)
+    parser.add_argument("--out", type=Path, default=DEFAULT_OUT)
+    parser.add_argument(
+        "--keep-clone",
+        action="store_true",
+        help="keep the sparse checkout so a later run can update instead of re-cloning",
+    )
+    args = parser.parse_args()
+
+    sparse_clone(args.clone, args.ref)
+    copied, total_bytes = collect(args.clone, args.out)
+
+    revision = subprocess.run(
+        ["git", "rev-parse", "--short", "HEAD"],
+        cwd=args.clone,
+        text=True,
+        capture_output=True,
+    ).stdout.strip()
+
+    # Provenance travels with the corpus. An index built from an unknown commit
+    # cannot be reproduced or explained later.
+    (args.out / "_SOURCE.txt").write_text(
+        f"repository: {REPO}\nref: {args.ref}\ncommit: {revision}\nfiles: {copied}\n",
+        encoding="utf-8",
+    )
+
+    if not args.keep_clone:
+        shutil.rmtree(args.clone, ignore_errors=True)
+
+    print(f"\n{copied} documents ({total_bytes / 1_048_576:.1f} MB) -> {args.out}")
+    print(f"{cfg.PROJECT} {args.ref} @ {revision}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())