scripts/fetch_docs.py

"""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())