Download & ingest

gutenberg_kg.ingest

ingest.py — Build per-book DocKGs for every book in gutenberg_kg, register them in the KGRAG registry, and add them to genre corpora and gutenberg-all.

Usage: python scripts/ingest.py [OPTIONS]

Options: --genre GENRE Process only this genre (repeatable; default: all) --force-build Rebuild even if .dockg already exists --force-register Re-register even if KG name already in registry --push git add + commit + push after each genre completes --dry-run Print actions without executing anything --registry PATH Override registry path

Examples: python scripts/ingest.py python scripts/ingest.py --genre shakespeare --genre ancient-classical python scripts/ingest.py --force-build --genre philosophy python scripts/ingest.py --push python scripts/ingest.py --dry-run

BookResult(name, genre, status, elapsed=0.0, nodes=0, edges=0) dataclass

Timing and outcome for one book.

GenreSummary(genre, results=list(), wall_elapsed=0.0) dataclass

Aggregated results for one genre.

built property

Number of books successfully built in this genre.

edges property

Total DocKG edges across all books in this genre.

elapsed property

Total wall-clock seconds across all books in this genre.

failed property

Number of books that failed to build.

nodes property

Total DocKG nodes across all books in this genre.

skipped property

Number of books skipped (already up-to-date).

total property

Total books processed in this genre.

IngestOptions(force_build=False, force_register=False, dry_run=False, push=False, quiet=False) dataclass

Flags controlling ingest behaviour.

add_to_corpus(corp_reg, corpus_name, kg_entry, dry_run=False)

Add kg_entry to corpus_name (idempotent via add_kg's dedup).

Source code in src/gutenberg_kg/ingest.py

def add_to_corpus(
    corp_reg: CorpusRegistry,
    corpus_name: str,
    kg_entry: KGEntry,
    dry_run: bool = False,
) -> bool:
    """Add kg_entry to corpus_name (idempotent via add_kg's dedup)."""
    if dry_run:
        print(f"    [dry] corpus add {corpus_name!r} {kg_entry.name!r}")
        return True
    result = corp_reg.add_kg(corpus_name, kg_entry.id)
    if result is None:
        print(f"    [!] corpus not found: {corpus_name!r}")
        return False
    return True

build_dockg(book_dir, dry_run=False, embedder=None, quiet=False)

Build DocKG for book_dir in-process, reusing a shared embedder if provided.

Mirrors the three-step pipeline used by dockg build: corpus → SQLite, SQLite → JSONL embedding cache, cache → sqlite-vec.

Programming errors (a bad import, a signature drift against doc_kg) are re-raised rather than reported per book: they fail identically for every book, so surfacing them once aborts the run instead of printing the same message 249 times. Per-book data failures are caught and reported.

Source code in src/gutenberg_kg/ingest.py

def build_dockg(
    book_dir: Path,
    dry_run: bool = False,
    embedder=None,
    quiet: bool = False,
) -> bool:
    """Build DocKG for book_dir in-process, reusing a shared embedder if provided.

    Mirrors the three-step pipeline used by ``dockg build``:
    corpus → SQLite, SQLite → JSONL embedding cache, cache → sqlite-vec.

    Programming errors (a bad import, a signature drift against doc_kg) are
    re-raised rather than reported per book: they fail identically for every
    book, so surfacing them once aborts the run instead of printing the same
    message 249 times. Per-book data failures are caught and reported.
    """
    if dry_run:
        print(f"    [dry] dockg build --repo {book_dir}")
        return True
    from doc_kg.kg import DocKG

    # Bound before the try so the finally can clean up whatever got as far as
    # being created, including when the constructor itself raises.
    kg = None
    cache_path = None
    try:
        kg = DocKG(book_dir, embedder=embedder)
        db_path = kg.db_path
        # .jsonl, not .json: the suffix picks the code path. The JSONL branch
        # streams and reuses the embedder passed in above, while the .json
        # branch builds its own CorpusEmbedder from the shared embedder's
        # model_name -- a second model load per book, which on MPS is the
        # double-load SIGBUS doc_kg added the streaming path to avoid.
        cache_path = db_path.parent / "embeddings.jsonl"
        kg.build_graph(wipe=True, quiet=quiet)
        kg.build_embeddings(out=cache_path, n_workers=4, quiet=quiet)
        # SIMILAR_TO stays on here, unlike `gutenkg build-corpus`, which ships a
        # served bundle whose handler is semantic-first and never traverses edges.
        # Per-book KGs are reached through KGRAG federation, which calls
        # DocKG.query() -- and DEFAULT_RELS expands over SIMILAR_TO, so these
        # edges are the only thing that path gains from. similar_k=5 with
        # similar_max_degree=8 is the exact configuration the cap evaluation
        # measured; see analysis/SIMILAR_TO_CAP_RECOMMENDATION.md.
        kg.build_index_from_cache(
            cache_path,
            wipe=True,
            discover_similar=True,
            similar_k=5,
            similar_max_degree=8,
            quiet=quiet,
        )
    except (ImportError, AttributeError, TypeError, NameError):
        # Signature drift against doc_kg, or a missing dependency. Identical for
        # every book, so let it abort the run with a traceback instead of being
        # laundered into a per-book "build failed".
        raise
    except Exception as exc:  # noqa: BLE001
        print(f"    [x] dockg build failed: {exc}")
        if not quiet:
            print(textwrap.indent(traceback.format_exc().rstrip(), "        "))
        return False
    finally:
        if kg is not None:
            with contextlib.suppress(Exception):
                kg.close()
        if cache_path is not None:
            cache_path.unlink(missing_ok=True)

    # Stamp the book's publication year as the shared temporal contract, so
    # a federated time-scoped query can reach this book at all. Books with
    # no usable date -- common for Gutenberg texts, which carry no IA
    # metadata sheet -- are simply left undated rather than guessed at.
    year = publication_year(book_dir)
    if year:
        stamped = stamp_publication_year(db_path, year)
        if stamped and not quiet:
            print(f"    [+] dated {stamped} nodes to {year}")
    return True

dockg_defect(store_dir)

Return why a .dockg store is unusable, or None if it is complete.

:meth:build_dockg writes graph.sqlite before it builds the vector store, so a run that dies in between leaves a valid graph with no vectors. Checking the graph alone would call that shape "already built": the book would be skipped on the next run and then registered with no semantic search, which fails silently at query time rather than at ingest time.

The vector row count comes from the vec_nodes_rowids shadow table rather than vec_nodes itself, because the latter is a vec0 virtual table and cannot be opened without the sqlite-vec extension loaded.

Parameters:

Name	Type	Description	Default
store_dir	Path	The book's .dockg directory.	required

Returns:

Type	Description
str | None	A short reason string, or None when the store is complete.

Source code in src/gutenberg_kg/ingest.py

def dockg_defect(store_dir: Path) -> str | None:
    """Return why a ``.dockg`` store is unusable, or ``None`` if it is complete.

    :meth:`build_dockg` writes ``graph.sqlite`` before it builds the vector
    store, so a run that dies in between leaves a *valid* graph with no vectors.
    Checking the graph alone would call that shape "already built": the book
    would be skipped on the next run and then registered with no semantic
    search, which fails silently at query time rather than at ingest time.

    The vector row count comes from the ``vec_nodes_rowids`` shadow table rather
    than ``vec_nodes`` itself, because the latter is a ``vec0`` virtual table
    and cannot be opened without the sqlite-vec extension loaded.

    :param store_dir: The book's ``.dockg`` directory.
    :return: A short reason string, or ``None`` when the store is complete.
    """
    if not is_sqlite_valid(store_dir / "graph.sqlite"):
        return "corrupt/empty graph.sqlite"

    vectors = resolve_vectors_path(store_dir)
    if vectors is None:
        if (store_dir / "lancedb").exists():
            return "only a LanceDB store, which is no longer read"
        return "no vector store"

    import sqlite3

    try:
        with sqlite3.connect(f"file:{vectors}?mode=ro", uri=True) as con:
            if not con.execute(
                "SELECT COUNT(*) FROM sqlite_master WHERE name='vec_nodes'"
            ).fetchone()[0]:
                return "vector store has no vec_nodes table"
            (rows,) = con.execute("SELECT COUNT(*) FROM vec_nodes_rowids").fetchone()
    except Exception as exc:  # noqa: BLE001
        return f"unreadable vector store ({exc})"
    if not rows:
        return "empty vector store"
    return None

ensure_corpus(corp_reg, name, description='', dry_run=False)

Create corpus if it doesn't exist (idempotent).

Source code in src/gutenberg_kg/ingest.py

def ensure_corpus(
    corp_reg: CorpusRegistry,
    name: str,
    description: str = "",
    dry_run: bool = False,
) -> None:
    """Create corpus if it doesn't exist (idempotent)."""
    from kg_rag.primitives import CorpusEntry

    if corp_reg.get(name) is not None:
        return
    if dry_run:
        print(f"  [dry] corpus create {name!r}")
        return
    corp_reg.create(CorpusEntry(name=name, description=description))
    print(f"  [+] corpus: {name}")

fmt_duration(seconds)

Human-readable duration string.

Source code in src/gutenberg_kg/ingest.py

def fmt_duration(seconds: float) -> str:
    """Human-readable duration string."""
    if seconds < 60:
        return f"{seconds:.1f}s"
    m, s = divmod(int(seconds), 60)
    if m < 60:
        return f"{m}m {s:02d}s"
    h, m = divmod(m, 60)
    return f"{h}h {m:02d}m {s:02d}s"

git_commit_push_genre(genre_dir, genre, dry_run=False)

Stage genre_dir, commit, and push. Skips silently if nothing to commit.

Source code in src/gutenberg_kg/ingest.py

def git_commit_push_genre(genre_dir: Path, genre: str, dry_run: bool = False) -> None:
    """Stage genre_dir, commit, and push. Skips silently if nothing to commit."""
    if dry_run:
        print(f"  [dry] git add {genre_dir}/ && git commit && git push")
        return

    subprocess.run(["git", "add", str(genre_dir) + "/"], check=True, cwd=REPO_ROOT)

    # Check if there's anything staged
    result = subprocess.run(
        ["git", "diff", "--cached", "--quiet"],
        check=False,
        cwd=REPO_ROOT,
    )
    if result.returncode == 0:
        print(f"  [=] {genre}: nothing to commit, skipping push")
        return

    n_files = (
        int(
            subprocess.check_output(
                ["git", "diff", "--cached", "--name-only"],
                cwd=REPO_ROOT,
                text=True,
            )
            .strip()
            .count("\n")
        )
        + 1
    )

    msg = f"chore(dockg): rebuild {genre} DocKG indices ({n_files} files)"
    subprocess.run(["git", "commit", "-m", msg], check=True, cwd=REPO_ROOT)
    subprocess.run(["git", "push"], check=True, cwd=REPO_ROOT)
    print(f"  [↑] {genre}: pushed {n_files} file(s)")

ingest_diaries(registry_path, opts)

Run the DiaryKG pipeline for the diaries genre and register the indices.

Diaries are not built through the standard DocKG path — they use a separate pipeline: chunk-diaries (.md → .diary/) then build-diaries (.diary/ → .diarykg/). This routes them so a single gutenkg ingest / rebuild-indices rebuilds prose and diaries. Runs before the Live progress display to avoid nesting Rich renderers.

Parameters:

Name	Type	Description	Default
registry_path	Path	Resolved KGRAG registry database path.	required
opts	IngestOptions	Ingest option flags (force_build re-chunks + rebuilds).	required

Returns:

Type	Description
tuple[int, GenreSummary | None]	(0 on success else 1, a GenreSummary for the "diaries" genre so it rolls into the same Totals/report as the standard DocKG genres — or None if the build step failed before any results were collected).

Source code in src/gutenberg_kg/ingest.py

def ingest_diaries(
    registry_path: Path,
    opts: IngestOptions,
) -> tuple[int, GenreSummary | None]:
    """Run the DiaryKG pipeline for the ``diaries`` genre and register the indices.

    Diaries are *not* built through the standard DocKG path — they use a separate
    pipeline: ``chunk-diaries`` (``.md`` → ``.diary/``) then ``build-diaries``
    (``.diary/`` → ``.diarykg/``).  This routes them so a single ``gutenkg ingest``
    / ``rebuild-indices`` rebuilds prose *and* diaries.  Runs before the Live
    progress display to avoid nesting Rich renderers.

    :param registry_path: Resolved KGRAG registry database path.
    :param opts: Ingest option flags (``force_build`` re-chunks + rebuilds).
    :return: (0 on success else 1, a GenreSummary for the "diaries" genre so it
        rolls into the same Totals/report as the standard DocKG genres — or
        ``None`` if the build step failed before any results were collected).
    """
    from kg_rag.corpus_registry import CorpusRegistry
    from kg_rag.registry import KGRegistry

    from gutenberg_kg.build_diaries import (
        BuildDiariesOptions,
        list_diary_dirs,
        run_build_diaries,
    )
    from gutenberg_kg.diary.chunk import ChunkDiariesOptions, run_chunk_diaries

    print("=== diaries (DiaryKG pipeline) ===")

    rc_chunk = run_chunk_diaries(
        [], ChunkDiariesOptions(force=opts.force_build, dry_run=opts.dry_run)
    )
    if rc_chunk != 0:
        return rc_chunk, None
    rc_build, build_results = run_build_diaries(
        [], BuildDiariesOptions(force=opts.force_build, dry_run=opts.dry_run)
    )
    genre_summary = GenreSummary(
        genre="diaries",
        results=[
            BookResult(
                name=r.name,
                genre="diaries",
                status=r.status,
                elapsed=r.elapsed,
                nodes=r.nodes,
                edges=r.edges,
            )
            for r in build_results
        ],
        wall_elapsed=sum(r.elapsed for r in build_results),
    )
    if rc_build != 0:
        return rc_build, genre_summary

    # --- Register each .diarykg with KGRAG + corpus membership ---
    diary_corpus = "gutenberg-diaries"
    with (
        KGRegistry(db_path=registry_path) as kg_reg,
        CorpusRegistry(db_path=registry_path) as corp_reg,
    ):
        ensure_corpus(
            corp_reg, diary_corpus, description="Project Gutenberg — diaries", dry_run=opts.dry_run
        )
        ensure_corpus(corp_reg, TOP_CORPUS, dry_run=opts.dry_run)
        for diary_dir in list_diary_dirs():
            name = f"gutenberg-diaries-{slugify(diary_dir.name)}-doc"
            existing = kg_reg.get(name)
            if existing is not None and not opts.force_register:
                print(f"  [{diary_dir.name}] already registered: {name}")
                entry = existing
            else:
                verb = "re-registering" if existing else "registering"
                print(f"  [{diary_dir.name}] {verb}: {name}")
                entry = register_diary_book(kg_reg, name, diary_dir, dry_run=opts.dry_run)
            if entry is not None:
                add_to_corpus(corp_reg, diary_corpus, entry, dry_run=opts.dry_run)
                add_to_corpus(corp_reg, TOP_CORPUS, entry, dry_run=opts.dry_run)
    print()
    return 0, genre_summary

is_sqlite_valid(path)

Return True if path is a readable, non-corrupt SQLite database.

Source code in src/gutenberg_kg/ingest.py

def is_sqlite_valid(path: Path) -> bool:
    """Return True if path is a readable, non-corrupt SQLite database."""
    import sqlite3

    if not path.exists() or path.stat().st_size < 100:
        return False
    try:
        with sqlite3.connect(path) as con:
            con.execute("SELECT COUNT(*) FROM nodes")
        return True
    except Exception:  # noqa: BLE001
        return False

print_summary(genre_summaries, opts, registry_path, wall_start, wall_elapsed, embed_model='', diary_rc=0)

Print a rich job summary to stdout.

Parameters:

Name	Type	Description	Default
diary_rc	int	Non-zero if the diary stage failed. Diaries build through a separate pipeline and contribute no GenreSummary when they fail, so without this the report would call a failed run a success.	0

Source code in src/gutenberg_kg/ingest.py

def print_summary(
    genre_summaries: list[GenreSummary],
    opts: IngestOptions,
    registry_path: Path,
    wall_start: datetime,
    wall_elapsed: float,
    embed_model: str = "",
    diary_rc: int = 0,
) -> None:
    """Print a rich job summary to stdout.

    :param diary_rc: Non-zero if the diary stage failed. Diaries build through a
        separate pipeline and contribute no GenreSummary when they fail, so
        without this the report would call a failed run a success.
    """
    thick = "═" * W
    thin = "─" * W

    total_built = sum(g.built for g in genre_summaries)
    total_skipped = sum(g.skipped for g in genre_summaries)
    total_failed = sum(g.failed for g in genre_summaries)
    total_books = sum(g.total for g in genre_summaries)
    total_nodes = sum(g.nodes for g in genre_summaries)
    total_edges = sum(g.edges for g in genre_summaries)

    if opts.dry_run:
        status_icon, status_text = "[~]", "DRY RUN — no changes made"
    elif total_failed == 0 and diary_rc == 0:
        status_icon, status_text = "[ok]", "SUCCESS — all books ingested"
    else:
        status_icon, status_text = "[!]", f"PARTIAL — {_failure_text(total_failed, diary_rc)}"

    flags = (
        " ".join(
            f
            for f in [
                "--force-build" if opts.force_build else "",
                "--force-register" if opts.force_register else "",
                "--push" if opts.push else "",
                "--dry-run" if opts.dry_run else "",
            ]
            if f
        )
        or "(none)"
    )

    print()
    print("╔" + thick + "╗")
    print(f"║  {'Gutenberg KG Ingest  —  Job Summary':<{W - 2}}║")
    print("╠" + thick + "╣")

    print(f"║{'':{W}}║")
    print(f"║{_row('Started', wall_start.strftime('%Y-%m-%d  %H:%M:%S UTC')):<{W}}║")
    print(f"║{_row('Elapsed', fmt_duration(wall_elapsed)):<{W}}║")
    print(f"║{_row('Host', socket.gethostname()):<{W}}║")
    print(
        f"║{_row('Platform', f'{platform.system()} {platform.release()}  /  {platform.machine()}'):<{W}}║"
    )
    print(f"║{_row('Python', sys.version.split()[0]):<{W}}║")
    print(f"║{_row('Registry', str(registry_path)):<{W}}║")
    if embed_model:
        print(f"║{_row('Embedder', embed_model):<{W}}║")
    print(f"║{_row('Flags', flags):<{W}}║")
    print(f"║{'':{W}}║")

    print("╠" + thin + "╣")
    print(f"║  {'Totals':<{W - 2}}║")
    print("╠" + thin + "╣")
    print(f"║{'':{W}}║")
    print(f"║{_row('Genres processed', str(len(genre_summaries))):<{W}}║")
    print(f"║{_row('Books total', str(total_books)):<{W}}║")
    print(f"║{_row('Built / rebuilt', str(total_built)):<{W}}║")
    print(f"║{_row('Skipped (up-to-date)', str(total_skipped)):<{W}}║")
    print(f"║{_row('Failed', str(total_failed)):<{W}}║")
    print(f"║{_row('Total nodes', f'{total_nodes:,}'):<{W}}║")
    print(f"║{_row('Total edges', f'{total_edges:,}'):<{W}}║")
    print(f"║{_row('Status', f'{status_icon}  {status_text}'):<{W}}║")
    print(f"║{'':{W}}║")

    print("╠" + thin + "╣")
    print(f"║  {'Per-Genre Breakdown':<{W - 2}}║")
    print("╠" + thin + "╣")
    print(
        f"║  {'Genre':<22}{'Books':>6}{'Built':>7}{'Skip':>6}{'Fail':>6}{'Nodes':>10}{'Edges':>10}{'Time':>8}  ║"
    )
    print(f"║  {'─' * 76}  ║")
    for g in genre_summaries:
        fail_flag = " !" if g.failed else ""
        print(
            f"║  {g.genre:<22}{g.total:>6}{g.built:>7}{g.skipped:>6}{g.failed:>6}"
            f"{g.nodes:>10,}{g.edges:>10,}{fmt_duration(g.wall_elapsed):>8}{fail_flag:<3}║"
        )
    print(f"║{'':{W}}║")

    all_failed = [r for g in genre_summaries for r in g.results if r.status == "failed"]
    if all_failed:
        print("╠" + thin + "╣")
        print(f"║  {'Failed Books':<{W - 2}}║")
        print("╠" + thin + "╣")
        print(f"║{'':{W}}║")
        for r in all_failed:
            print(f"║  ✗  {r.genre}/{r.name:<{W - 8}}║")
        print(f"║{'':{W}}║")
        print(f"║  {'Tip: re-run with --force-build to retry failed books.':<{W - 2}}║")
        print(f"║{'':{W}}║")

    print("╚" + thick + "╝")
    print()

process_book(book_dir, genre, kg_reg, corp_reg, opts, embedder=None)

Process one book directory. Returns a BookResult with timing and graph stats.

Source code in src/gutenberg_kg/ingest.py

def process_book(
    book_dir: Path,
    genre: str,
    kg_reg: KGRegistry,
    corp_reg: CorpusRegistry,
    opts: IngestOptions,
    embedder=None,
) -> BookResult:
    """Process one book directory. Returns a BookResult with timing and graph stats."""
    book_name = book_dir.name
    slug = slugify(book_name)
    kg_name = f"gutenberg-{genre}-{slug}-doc"
    genre_corpus = f"gutenberg-{genre}"
    dockg_sqlite = book_dir / ".dockg" / "graph.sqlite"

    print(f"  [{book_name}]")
    t0 = time.perf_counter()

    # --- Build ---
    dockg_dir = book_dir / ".dockg"
    already_built = dockg_sqlite.exists()

    # Auto-wipe if the store is corrupt, or half-built from an interrupted run
    defect = dockg_defect(dockg_dir) if already_built else None
    if defect:
        print(f"    [!] {defect} — wiping and rebuilding")
        if not opts.dry_run:
            shutil.rmtree(dockg_dir)
        already_built = False

    if already_built and not opts.force_build:
        print("    [=] already built, skipping dockg build")
        status = "skipped"
    else:
        if already_built and opts.force_build:
            if not opts.dry_run:
                shutil.rmtree(dockg_dir)
                print("    [~] wiped .dockg")
            else:
                print("    [dry] rm -rf .dockg")
        label = "rebuilding" if already_built else "building"
        print(f"    [.] {label} DocKG...")
        if not build_dockg(book_dir, dry_run=opts.dry_run, embedder=embedder, quiet=opts.quiet):
            elapsed = time.perf_counter() - t0
            return BookResult(name=book_name, genre=genre, status="failed", elapsed=elapsed)
        status = "built"

    # --- Register ---
    existing = kg_reg.get(kg_name)
    if existing is not None and not opts.force_register:
        print(f"    [=] already registered: {kg_name}")
        entry = existing
    else:
        verb = "re-registering" if existing else "registering"
        print(f"    [.] {verb}: {kg_name}")
        entry = register_book(kg_reg, kg_name, book_dir, dry_run=opts.dry_run)
        if entry is None and not opts.dry_run:
            elapsed = time.perf_counter() - t0
            return BookResult(name=book_name, genre=genre, status="failed", elapsed=elapsed)

    if entry is not None:
        add_to_corpus(corp_reg, genre_corpus, entry, dry_run=opts.dry_run)
        add_to_corpus(corp_reg, TOP_CORPUS, entry, dry_run=opts.dry_run)

    elapsed = time.perf_counter() - t0
    nodes, edges = _sqlite_counts(book_dir)
    print(f"    [✓] {fmt_duration(elapsed)}  nodes={nodes:,}  edges={edges:,}")
    return BookResult(
        name=book_name, genre=genre, status=status, elapsed=elapsed, nodes=nodes, edges=edges
    )

register_book(kg_reg, name, book_dir, dry_run=False)

Register a book DocKG. Returns the KGEntry (new or existing). Returns None on dry-run or failure.

Source code in src/gutenberg_kg/ingest.py

def register_book(
    kg_reg: KGRegistry,
    name: str,
    book_dir: Path,
    dry_run: bool = False,
) -> KGEntry | None:
    """
    Register a book DocKG. Returns the KGEntry (new or existing).
    Returns None on dry-run or failure.
    """
    from kg_rag.primitives import KGEntry, KGKind

    sqlite = book_dir / ".dockg" / "graph.sqlite"
    vectors = resolve_vectors_path(book_dir / ".dockg")
    entry = KGEntry(
        name=name,
        kind=KGKind.GUTENBERG,
        repo_path=book_dir,
        venv_path=book_dir / ".venv",
        sqlite_path=sqlite if sqlite.exists() else None,
        vectors_path=vectors,
        tags=[date.today().isoformat()],
    )
    if dry_run:
        print(f"    [dry] register {name!r} -> {book_dir}")
        return None
    kg_reg.register(entry)
    return entry

register_diary_book(kg_reg, name, diary_dir, dry_run=False)

Register a diary's .diarykg/ index (DiaryKG, not standard .dockg/).

Parameters:

Name	Type	Description	Default
kg_reg	KGRegistry	Open KGRAG registry.	required
name	str	KG name, e.g. gutenberg-diaries-<slug>-doc.	required
diary_dir	Path	Diary book directory under corpus/diaries/.	required
dry_run	bool	Print the action without writing to the registry.	False

Returns:

Type	Description
KGEntry | None	The registered :class:KGEntry, or None on dry-run/failure.

Source code in src/gutenberg_kg/ingest.py

def register_diary_book(
    kg_reg: KGRegistry,
    name: str,
    diary_dir: Path,
    dry_run: bool = False,
) -> KGEntry | None:
    """Register a diary's ``.diarykg/`` index (DiaryKG, not standard ``.dockg/``).

    :param kg_reg: Open KGRAG registry.
    :param name: KG name, e.g. ``gutenberg-diaries-<slug>-doc``.
    :param diary_dir: Diary book directory under ``corpus/diaries/``.
    :param dry_run: Print the action without writing to the registry.
    :return: The registered :class:`KGEntry`, or ``None`` on dry-run/failure.
    """
    from kg_rag.primitives import KGEntry, KGKind

    sqlite = diary_dir / ".diarykg" / "graph.sqlite"
    vectors = resolve_vectors_path(diary_dir / ".diarykg")
    entry = KGEntry(
        name=name,
        kind=KGKind.GUTENBERG,
        repo_path=diary_dir,
        venv_path=diary_dir / ".venv",
        sqlite_path=sqlite if sqlite.exists() else None,
        vectors_path=vectors,
        tags=[date.today().isoformat()],
    )
    if dry_run:
        print(f"    [dry] register diary {name!r} -> {diary_dir}")
        return None
    kg_reg.register(entry)
    return entry

run_ingest(genres, opts, registry=None)

Orchestrate DocKG builds, KGRAG registration, and corpus membership.

Parameters:

Name	Type	Description	Default
genres	list[str]	Genre names to process (already validated against ALL_GENRES).	required
opts	IngestOptions	Ingest option flags.	required
registry	str | Path | None	Override path to the KGRAG registry database; None uses the default location returned by default_registry_path().	None

Returns:

Type	Description
int	0 on full success, 1 if any book failed.

Source code in src/gutenberg_kg/ingest.py

def run_ingest(
    genres: list[str],
    opts: IngestOptions,
    registry: str | Path | None = None,
) -> int:
    """Orchestrate DocKG builds, KGRAG registration, and corpus membership.

    :param genres: Genre names to process (already validated against ALL_GENRES).
    :param opts: Ingest option flags.
    :param registry: Override path to the KGRAG registry database; ``None``
        uses the default location returned by ``default_registry_path()``.
    :return: 0 on full success, 1 if any book failed.
    """
    from kg_rag.corpus_registry import CorpusRegistry
    from kg_rag.registry import (
        KGRegistry,
        default_registry_path,
    )

    registry_path = Path(registry).resolve() if registry else default_registry_path()

    # Diaries use the DiaryKG pipeline, not the standard DocKG path. Handle them
    # first (before the Live display) and drop them from the prose genre list.
    diary_rc = 0
    diary_summary: GenreSummary | None = None
    if "diaries" in genres:
        diary_rc, diary_summary = ingest_diaries(registry_path, opts)
        genres = [g for g in genres if g != "diaries"]

    # Pre-count total books so the progress bar has an accurate total.
    def _book_dirs(genre: str) -> list[Path]:
        """:return: Sorted non-hidden book subdirectories under ``CORPUS_ROOT/genre``."""
        d = CORPUS_ROOT / genre
        if not d.is_dir():
            return []
        return sorted(p for p in d.iterdir() if p.is_dir() and not p.name.startswith("."))

    total_books = sum(len(_book_dirs(g)) for g in genres)

    genre_summaries: list[GenreSummary] = []
    if diary_summary is not None:
        genre_summaries.append(diary_summary)
    wall_start = datetime.now(UTC)
    wall_t0 = time.perf_counter()
    embed_model_name: str = ""

    log = _LogCapture(maxlines=24)
    progress = _make_progress()
    layout = _build_layout(progress, log)
    stderr_console = RichConsole(stderr=True)
    live = Live(layout, console=stderr_console, refresh_per_second=4)

    log.start()
    live.start()
    try:
        task = progress.add_task("starting…", total=total_books)

        with (
            KGRegistry(db_path=registry_path) as kg_reg,
            CorpusRegistry(db_path=registry_path) as corp_reg,
        ):
            print("--- Ensuring corpora ---")
            for genre in genres:
                ensure_corpus(
                    corp_reg,
                    f"gutenberg-{genre}",
                    description=f"Project Gutenberg — {genre}",
                    dry_run=opts.dry_run,
                )
            ensure_corpus(
                corp_reg,
                TOP_CORPUS,
                description="Project Gutenberg — complete library",
                dry_run=opts.dry_run,
            )
            print()

            for genre in genres:
                genre_dir = CORPUS_ROOT / genre
                book_dirs = _book_dirs(genre)
                if not genre_dir.is_dir():
                    print(f"[!] Genre directory not found: {genre_dir} — skipping\n")
                    continue
                if not book_dirs:
                    print(f"[!] No book directories in {genre_dir} — skipping\n")
                    continue

                print(f"=== {genre} ({len(book_dirs)} books) ===")
                genre_summary = GenreSummary(genre=genre)
                genre_t0 = time.perf_counter()

                from doc_kg.index import (
                    SentenceTransformerEmbedder,  # type: ignore[import-untyped]
                )

                shared_embedder = SentenceTransformerEmbedder()
                if not embed_model_name:
                    embed_model_name = shared_embedder.model_name
                print(f"  [embedder] {shared_embedder!r}")

                for book_dir in book_dirs:
                    progress.update(
                        task,
                        description=f"[bold]{genre}[/bold] › {book_dir.name[:45]}",
                    )
                    result = process_book(
                        book_dir=book_dir,
                        genre=genre,
                        kg_reg=kg_reg,
                        corp_reg=corp_reg,
                        opts=opts,
                        embedder=shared_embedder,
                    )
                    genre_summary.results.append(result)
                    progress.advance(task)

                genre_summary.wall_elapsed = time.perf_counter() - genre_t0
                genre_summaries.append(genre_summary)

                if opts.push:
                    git_commit_push_genre(genre_dir, genre, dry_run=opts.dry_run)
                print()

    finally:
        live.stop()
        log.stop()

    if not genre_summaries:
        return 1 if diary_rc != 0 else 0

    wall_elapsed = time.perf_counter() - wall_t0
    print_summary(
        genre_summaries,
        opts,
        registry_path,
        wall_start,
        wall_elapsed,
        embed_model_name,
        diary_rc=diary_rc,
    )

    if opts.dry_run:
        # A dry run changes nothing, so it should not leave a report behind
        # claiming it did — that is the one thing that fills reports/ with noise.
        print("  [dry] no report written\n")
    else:
        report_path = save_summary(
            genre_summaries,
            opts,
            registry_path,
            wall_start,
            wall_elapsed,
            embed_model_name,
            diary_rc=diary_rc,
        )
        print(f"  Report saved: {report_path}\n")

    return 1 if (diary_rc != 0 or any(g.failed for g in genre_summaries)) else 0

run_reregister(genres, registry=None, dry_run=False)

Re-register all built books from disk with the current KGKind without rebuilding DocKGs.

Walks each genre directory, finds book dirs that have a built .dockg/graph.sqlite, and calls register_book for each one. Idempotent — safe to run on any machine to fix stale kind values or populate a fresh registry after cloning.

Parameters:

Name	Type	Description	Default
genres	list[str]	Genre names to process.	required
registry	str | Path | None	Override path to the KGRAG registry database.	None
dry_run	bool	Print actions without writing to the registry.	False

Returns:

Type	Description
int	0 on success.

Source code in src/gutenberg_kg/ingest.py

def run_reregister(
    genres: list[str],
    registry: str | Path | None = None,
    dry_run: bool = False,
) -> int:
    """Re-register all built books from disk with the current KGKind without rebuilding DocKGs.

    Walks each genre directory, finds book dirs that have a built .dockg/graph.sqlite,
    and calls register_book for each one. Idempotent — safe to run on any machine to
    fix stale kind values or populate a fresh registry after cloning.

    :param genres: Genre names to process.
    :param registry: Override path to the KGRAG registry database.
    :param dry_run: Print actions without writing to the registry.
    :return: 0 on success.
    """
    from kg_rag.corpus_registry import CorpusRegistry
    from kg_rag.registry import (
        KGRegistry,
        default_registry_path,
    )

    registry_path = Path(registry).resolve() if registry else default_registry_path()
    total = updated = skipped = 0

    print(f"Registry: {registry_path}")
    print()

    with (
        KGRegistry(db_path=registry_path) as kg_reg,
        CorpusRegistry(db_path=registry_path) as corp_reg,
    ):
        for genre in genres:
            genre_dir = CORPUS_ROOT / genre
            if not genre_dir.is_dir():
                print(f"[!] {genre}: directory not found — skipping")
                continue

            book_dirs = sorted(
                p for p in genre_dir.iterdir() if p.is_dir() and not p.name.startswith(".")
            )
            genre_corpus = f"gutenberg-{genre}"
            ensure_corpus(corp_reg, genre_corpus, dry_run=dry_run)
            ensure_corpus(corp_reg, TOP_CORPUS, dry_run=dry_run)

            # Diaries register their .diarykg/ index, not a standard .dockg/.
            if genre == "diaries":
                print(f"=== {genre} ({len(book_dirs)} diaries) ===")
                for diary_dir in book_dirs:
                    if not (diary_dir / ".diarykg" / "graph.sqlite").exists():
                        print(f"  [{diary_dir.name}] no .diarykg — skipping")
                        skipped += 1
                        continue
                    name = f"gutenberg-diaries-{slugify(diary_dir.name)}-doc"
                    total += 1
                    print(f"  [{diary_dir.name}] re-registering diary")
                    entry = register_diary_book(kg_reg, name, diary_dir, dry_run=dry_run)
                    if entry:
                        add_to_corpus(corp_reg, genre_corpus, entry, dry_run=dry_run)
                        add_to_corpus(corp_reg, TOP_CORPUS, entry, dry_run=dry_run)
                    updated += 1
                continue

            print(f"=== {genre} ({len(book_dirs)} books) ===")
            for book_dir in book_dirs:
                sqlite = book_dir / ".dockg" / "graph.sqlite"
                if not sqlite.exists():
                    print(f"  [{book_dir.name}] no .dockg — skipping")
                    skipped += 1
                    continue

                slug = slugify(book_dir.name)
                kg_name = f"gutenberg-{genre}-{slug}-doc"
                total += 1

                from kg_rag.primitives import KGKind

                existing = kg_reg.get(kg_name)
                if existing is not None and existing.kind == KGKind.GUTENBERG:
                    print(f"  [{book_dir.name}] already gutenberg — ok")
                    continue

                verb = "re-registering" if existing else "registering"
                print(f"  [{book_dir.name}] {verb} as gutenberg")
                entry = register_book(kg_reg, kg_name, book_dir, dry_run=dry_run)
                if entry:
                    add_to_corpus(corp_reg, genre_corpus, entry, dry_run=dry_run)
                    add_to_corpus(corp_reg, TOP_CORPUS, entry, dry_run=dry_run)
                updated += 1

            print()

    print(f"Done — {updated} registered/updated, {skipped} skipped (no .dockg), {total} total")
    return 0

save_summary(genre_summaries, opts, registry_path, wall_start, wall_elapsed, embed_model='', diary_rc=0)

Write a Markdown ingest report to reports/ and return its path.

Parameters:

Name	Type	Description	Default
diary_rc	int	Non-zero if the diary stage failed. See :func:print_summary.	0

Source code in src/gutenberg_kg/ingest.py

def save_summary(
    genre_summaries: list[GenreSummary],
    opts: IngestOptions,
    registry_path: Path,
    wall_start: datetime,
    wall_elapsed: float,
    embed_model: str = "",
    diary_rc: int = 0,
) -> Path:
    """Write a Markdown ingest report to reports/ and return its path.

    :param diary_rc: Non-zero if the diary stage failed. See :func:`print_summary`.
    """
    reports_dir = REPO_ROOT / "reports"
    reports_dir.mkdir(exist_ok=True)

    ts = wall_start.strftime("%Y-%m-%d_%H%M%S")
    report_path = reports_dir / f"ingest_{ts}.md"

    total_built = sum(g.built for g in genre_summaries)
    total_skipped = sum(g.skipped for g in genre_summaries)
    total_failed = sum(g.failed for g in genre_summaries)
    total_books = sum(g.total for g in genre_summaries)
    total_nodes = sum(g.nodes for g in genre_summaries)
    total_edges = sum(g.edges for g in genre_summaries)

    if opts.dry_run:
        status = "DRY RUN — no changes made"
    elif total_failed == 0 and diary_rc == 0:
        status = "SUCCESS — all books ingested"
    else:
        status = f"PARTIAL — {_failure_text(total_failed, diary_rc)}"

    flags = (
        " ".join(
            f
            for f in [
                "--force-build" if opts.force_build else "",
                "--force-register" if opts.force_register else "",
                "--push" if opts.push else "",
                "--dry-run" if opts.dry_run else "",
            ]
            if f
        )
        or "(none)"
    )

    lines = [
        "# Gutenberg KG Ingest Report",
        "",
        f"**Date:** {wall_start.strftime('%Y-%m-%d %H:%M:%S UTC')}  ",
        f"**Elapsed:** {fmt_duration(wall_elapsed)}  ",
        f"**Host:** {socket.gethostname()} / {platform.system()} {platform.release()} / {platform.machine()}  ",
        f"**Python:** {sys.version.split()[0]}  ",
        f"**Registry:** `{registry_path}`  ",
        f"**Embedder:** `{embed_model}`  " if embed_model else "",
        f"**Flags:** `{flags}`  ",
        f"**Status:** {status}",
        "",
        "## Totals",
        "",
        "| Metric | Value |",
        "|--------|-------|",
        f"| Genres processed | {len(genre_summaries)} |",
        f"| Books total | {total_books} |",
        f"| Built / rebuilt | {total_built} |",
        f"| Skipped (up-to-date) | {total_skipped} |",
        f"| Failed | {total_failed} |",
        f"| Total nodes | {total_nodes:,} |",
        f"| Total edges | {total_edges:,} |",
        "",
        "## Per-Genre Breakdown",
        "",
        "| Genre | Books | Built | Skipped | Failed | Nodes | Edges | Time |",
        "|-------|------:|------:|--------:|-------:|------:|------:|------|",
    ]

    for g in genre_summaries:
        lines.append(
            f"| {g.genre} | {g.total} | {g.built} | {g.skipped} | {g.failed} "
            f"| {g.nodes:,} | {g.edges:,} | {fmt_duration(g.wall_elapsed)} |"
        )

    # Per-genre book detail
    lines += ["", "## Book Detail", ""]
    for g in genre_summaries:
        lines += [
            f"### {g.genre}",
            "",
            "| Book | Status | Nodes | Edges | Time |",
            "|------|--------|------:|------:|------|",
        ]
        for r in g.results:
            icon = "✓" if r.status == "built" else ("=" if r.status == "skipped" else "✗")
            lines.append(
                f"| {r.name} | {icon} {r.status} | {r.nodes:,} | {r.edges:,} | {fmt_duration(r.elapsed)} |"
            )
        lines.append("")

    all_failed = [r for g in genre_summaries for r in g.results if r.status == "failed"]
    if all_failed:
        lines += ["## Failed Books", ""]
        for r in all_failed:
            lines.append(f"- `{r.genre}/{r.name}`")
        lines += ["", "_Re-run with `--force-build` to retry._", ""]

    report_path.write_text("\n".join(lines), encoding="utf-8")
    return report_path

slugify(s)

Lowercase and replace non-alphanumeric runs with hyphens.

Source code in src/gutenberg_kg/ingest.py

def slugify(s: str) -> str:
    """Lowercase and replace non-alphanumeric runs with hyphens."""
    return re.sub(r"-+", "-", re.sub(r"[^a-z0-9]", "-", s.lower())).strip("-")