Diary build pipeline

gutenberg_kg.build_diaries

build_diaries.py - Build .diarykg/ DocKG indices for diary corpora.

Each diary lives under corpus/diaries// with pre-chunked entries in .diary/ and the DocKG index written to .diarykg/. This is a prerequisite for bundle_diaries() in build_corpus.py - it copies the indices but does not build them.

The .diary/ chunk corpus is produced by 'gutenkg chunk-diaries' (see gutenberg_kg.diary.chunk). Chunk strategy and flags match docs/ingestion-pipeline.md: - chunk_strategy = sentence_group (preserves temporal entry structure) - discover_similar = False (chronologically dense entries produce noise) - model = BAAI/bge-small-en-v1.5 (must match EMBED_MODEL in the Docker image)

BuildDiariesOptions(force=False, n_workers=4, dry_run=False, quiet=False) dataclass

Flags controlling a diary index build.

DiaryBuildResult(name, status, elapsed=0.0, nodes=0, edges=0, message='') dataclass

Outcome for one diary index build.

build_diary_index(diary_dir, opts)

Build (or rebuild) the .diarykg/ DiaryKG index for one diary.

Creates a symlink .diarykg/corpus -> ../.diary so DiaryKG finds its pre-chunked corpus, then calls DiaryKG.rebuild_index() which runs Steps 2 (DocKG build), 3 (_inject_topic_edges), and 4 (_enrich_metadata).

Parameters:

Name	Type	Description	Default
diary_dir	Path	Root directory of the diary (e.g. corpus/diaries/Pepys/).	required
opts	BuildDiariesOptions	Build option flags.	required

Returns:

Type	Description
DiaryBuildResult	DiaryBuildResult with timing and graph stats.

Source code in src/gutenberg_kg/build_diaries.py

def build_diary_index(
    diary_dir: Path,
    opts: BuildDiariesOptions,
) -> DiaryBuildResult:
    """Build (or rebuild) the .diarykg/ DiaryKG index for one diary.

    Creates a symlink ``.diarykg/corpus -> ../.diary`` so DiaryKG finds its
    pre-chunked corpus, then calls ``DiaryKG.rebuild_index()`` which runs Steps
    2 (DocKG build), 3 (_inject_topic_edges), and 4 (_enrich_metadata).

    :param diary_dir: Root directory of the diary (e.g. corpus/diaries/Pepys/).
    :param opts: Build option flags.
    :return: DiaryBuildResult with timing and graph stats.
    """
    import shutil

    name = diary_dir.name
    diary_chunks_dir = diary_dir / ".diary"
    diarykg_dir = diary_dir / ".diarykg"
    sqlite_path = diarykg_dir / "graph.sqlite"

    t0 = time.perf_counter()

    if not diary_chunks_dir.exists() or not any(diary_chunks_dir.iterdir()):
        return DiaryBuildResult(
            name=name,
            status="failed",
            elapsed=time.perf_counter() - t0,
            message=f".diary/ not found or empty: {diary_chunks_dir}",
        )

    # A store is "already built" only if it is *complete*. Checking that
    # graph.sqlite merely exists would skip a half-built index -- rebuild_index()
    # writes the graph before the vector store, so an interrupted run leaves a
    # graph with no vectors, which then gets registered with no semantic search.
    # Same check the DocKG path uses; .diarykg has the same on-disk shape.
    defect = dockg_defect(diarykg_dir) if sqlite_path.exists() else None
    already_built = sqlite_path.exists() and defect is None
    if already_built and not opts.force:
        nodes, edges = _sqlite_counts(sqlite_path)
        return DiaryBuildResult(
            name=name,
            status="skipped",
            elapsed=time.perf_counter() - t0,
            nodes=nodes,
            edges=edges,
            message="already built (use --force to rebuild)",
        )
    if defect is not None:
        print(f"  [!] {name}: {defect} — wiping and rebuilding")

    if opts.dry_run:
        chunk_count = sum(1 for _ in diary_chunks_dir.glob("entry_*.md"))
        return DiaryBuildResult(
            name=name,
            status="skipped",
            elapsed=time.perf_counter() - t0,
            message=f"[dry-run] would build from {chunk_count} chunk files in {diary_chunks_dir}",
        )

    # Wipe on an explicit --force, and also on a defective store: building on
    # top of a stale .diarykg would leave whatever the last run wrote behind.
    if diarykg_dir.exists() and (already_built and opts.force or defect is not None):
        shutil.rmtree(diarykg_dir)

    diarykg_dir.mkdir(parents=True, exist_ok=True)

    # Symlink .diarykg/corpus → ../.diary so DiaryKG finds its pre-chunked corpus.
    corpus_link = diarykg_dir / "corpus"
    if not corpus_link.exists():
        corpus_link.symlink_to(diary_chunks_dir.resolve())

    try:
        from diary_kg.kg import DiaryKG
    except ImportError as exc:
        return DiaryBuildResult(
            name=name,
            status="failed",
            elapsed=time.perf_counter() - t0,
            message=f"diary-kg not installed: {exc}",
        )

    try:
        # `with`, not a bare constructor: DiaryKG holds a lazily built DocKG and
        # its SQLite connection, and before diary-kg 0.98.0 there was no way to
        # let it go -- one leaked connection per diary on every corpus build.
        # Closing before _clean_chunk_texts also means that rewrite reopens
        # graph.sqlite on its own rather than racing a live connection.
        with DiaryKG(root=diary_dir, model=DIARY_EMBED_MODEL) as kg:
            kg.rebuild_index()
        _clean_chunk_texts(sqlite_path)
    except (AttributeError, TypeError, NameError):
        # Signature drift against diary_kg. Identical for every diary, so let it
        # abort with a traceback rather than be reported as a per-diary failure.
        # ImportError is handled above, with a friendlier "not installed" message.
        raise
    except Exception as exc:  # noqa: BLE001
        return DiaryBuildResult(
            name=name,
            status="failed",
            elapsed=time.perf_counter() - t0,
            message=f"{exc}\n{textwrap.indent(traceback.format_exc().rstrip(), '      ')}",
        )

    nodes, edges = _sqlite_counts(sqlite_path)
    return DiaryBuildResult(
        name=name,
        status="built",
        elapsed=time.perf_counter() - t0,
        nodes=nodes,
        edges=edges,
    )

list_diary_dirs(names=None)

Return sorted diary root directories under corpus/diaries/.

Parameters:

Name	Type	Description	Default
names	list[str] | None	Optional list of diary directory names to filter to.	None

Returns:

Type	Description
list[Path]	Matching (or all) diary directories, sorted alphabetically.

Source code in src/gutenberg_kg/build_diaries.py

def list_diary_dirs(names: list[str] | None = None) -> list[Path]:
    """Return sorted diary root directories under corpus/diaries/.

    :param names: Optional list of diary directory names to filter to.
    :return: Matching (or all) diary directories, sorted alphabetically.
    """
    if not DIARIES_ROOT.exists():
        return []
    dirs = sorted(p for p in DIARIES_ROOT.iterdir() if p.is_dir() and not p.name.startswith("."))
    if names:
        name_set = set(names)
        dirs = [d for d in dirs if d.name in name_set]
    return dirs

run_build_diaries(diary_names, opts)

Build .diarykg/ indices for all (or selected) diaries.

Parameters:

Name	Type	Description	Default
diary_names	list[str]	Diary directory names to process (empty = all).	required
opts	BuildDiariesOptions	Build option flags.	required

Returns:

Type	Description
tuple[int, list[DiaryBuildResult]]	(0 on full success else 1, per-diary build results).

Source code in src/gutenberg_kg/build_diaries.py

def run_build_diaries(
    diary_names: list[str],
    opts: BuildDiariesOptions,
) -> tuple[int, list[DiaryBuildResult]]:
    """Build .diarykg/ indices for all (or selected) diaries.

    :param diary_names: Diary directory names to process (empty = all).
    :param opts: Build option flags.
    :return: (0 on full success else 1, per-diary build results).
    """
    diary_dirs = list_diary_dirs(diary_names if diary_names else None)

    if not diary_dirs:
        print("[!] No diary directories found under corpus/diaries/")
        if diary_names:
            print(f"    Requested: {diary_names}")
        return 1, []

    print("=== gutenkg build-diaries ===")
    print(f"  diaries       : {len(diary_dirs)}")
    print(f"  strategy      : {DIARY_CHUNK_STRATEGY}")
    print(f"  embed model   : {DIARY_EMBED_MODEL}")
    print("  SIMILAR_TO    : disabled")
    print(f"  force rebuild : {opts.force}")
    if opts.dry_run:
        print("  mode          : DRY RUN")
    print()

    results: list[DiaryBuildResult] = []
    t_total = time.perf_counter()

    for diary_dir in diary_dirs:
        name = diary_dir.name
        chunk_count = sum(1 for _ in (diary_dir / ".diary").glob("entry_*.md"))
        print(f"[{name}]  ({chunk_count:,} chunk files)")

        result = build_diary_index(diary_dir, opts)
        results.append(result)

        icon = {"built": "[+]", "skipped": "[=]", "failed": "[x]"}.get(result.status, "[ ]")
        detail = (
            f"nodes={result.nodes:,}  edges={result.edges:,}  {fmt_duration(result.elapsed)}"
            if result.status == "built"
            else result.message
        )
        print(f"  {icon} {result.status}  {detail}")
        print()

    elapsed = time.perf_counter() - t_total
    built = sum(1 for r in results if r.status == "built")
    skipped = sum(1 for r in results if r.status == "skipped")
    failed = sum(1 for r in results if r.status == "failed")
    total_nodes = sum(r.nodes for r in results)
    total_edges = sum(r.edges for r in results)

    print("=== build-diaries complete ===")
    print(f"  built         : {built}")
    print(f"  skipped       : {skipped}")
    print(f"  failed        : {failed}")
    if built or skipped:
        print(f"  total nodes   : {total_nodes:,}")
        print(f"  total edges   : {total_edges:,}")
    print(f"  elapsed       : {fmt_duration(elapsed)}")

    if failed:
        print()
        print("  Failed diaries:")
        for r in results:
            if r.status == "failed":
                print(f"    [x] {r.name}: {r.message}")
    print()

    return (1 if failed else 0), results