Corpus discovery

gutenberg_kg.corpus

corpus.py — Library API for GutenbergKG corpus-level operations.

Provides path-parametrised functions for building corpus status reports and snapshot dicts. All functions accept explicit path arguments so they can be called from any context — CLI, tests, or kgrag adapters — without depending on package-level path constants.

Author: Eric G. Suchanek, PhD License: Elastic 2.0

GutenbergSnapshotManager(snapshots_dir, *, registry_path, repo_root, corpus_root)

Bases: SnapshotManager

Corpus-aware snapshot manager backed by kg_utils.snapshots.SnapshotManager.

Snapshots are stored as <key>.json alongside a manifest.json index in snapshots_dir. Use :meth:capture to build a snapshot from the live corpus, then :meth:save_snapshot to persist it.

Since kgmodule-utils 0.19.0 the key is a UTC timestamp, not the git tree hash: the hash was read before git add staged the snapshot, so it named a tree that was never committed. A timestamp is the right key for a corpus anyway — the corpus changes when books are ingested, not when the repo is tagged. Older snapshots on disk keep their hash keys.

Parameters:

Name	Type	Description	Default
snapshots_dir	Path | str	Directory where snapshot JSON files are stored.	required
registry_path	Path	KGRAG registry SQLite path.	required
repo_root	Path	Repository root (for git metadata).	required
corpus_root	Path	Corpus data root (contains authors/ etc.).	required

Initialize the manager and store the paths used to capture snapshots.

Parameters:

Name	Type	Description	Default
snapshots_dir	Path | str	Directory where snapshot JSON files are stored.	required
registry_path	Path	KGRAG registry SQLite path.	required
repo_root	Path	Repository root (for git metadata).	required
corpus_root	Path	Corpus data root (contains authors/ etc.).	required

Source code in src/gutenberg_kg/corpus.py

def __init__(
    self,
    snapshots_dir: Path | str,
    *,
    registry_path: Path,
    repo_root: Path,
    corpus_root: Path,
) -> None:
    """Initialize the manager and store the paths used to capture snapshots.

    :param snapshots_dir: Directory where snapshot JSON files are stored.
    :param registry_path: KGRAG registry SQLite path.
    :param repo_root: Repository root (for git metadata).
    :param corpus_root: Corpus data root (contains ``authors/`` etc.).
    """
    super().__init__(snapshots_dir, package_name="gutenberg-kg")
    self.registry_path = registry_path
    self.repo_root = repo_root
    self.corpus_root = corpus_root

capture(version=None, branch=None, graph_stats_dict=None, tree_hash='', hotspots=None, issues=None, key='', subject='', **extra_metrics)

Build a corpus snapshot from the live registry.

key and subject are named rather than left to extra_metrics: the base takes **extra_metrics too, so an unnamed one would be recorded as a metric instead of reaching the fields it belongs in.

Parameters:

Name	Type	Description	Default
version	str | None	Version string; auto-detected from package if None.	None
branch	str | None	Git branch name; auto-detected if None.	None
graph_stats_dict	dict[str, Any] | None	Ignored; corpus metrics are computed from the registry.	None
tree_hash	str	Git tree hash; recorded as provenance, no longer the key.	''
hotspots	list[dict[str, Any]] | None	Unused; accepted for LSP compatibility.	None
issues	list[str] | None	Unused; accepted for LSP compatibility.	None
key	str	Snapshot identifier; a UTC timestamp when omitted, which is what a corpus wants — pass a release tag only when snapshotting the package rather than the books.	''
subject	str	What was measured; defaults to :data:SNAPSHOT_SUBJECT.	''
extra_metrics	Any	Passed through to the base capture call.	{}

Returns:

Type	Description
Any	New Snapshot instance (not yet persisted).

Source code in src/gutenberg_kg/corpus.py

def capture(
    self,
    version: str | None = None,
    branch: str | None = None,
    graph_stats_dict: dict[str, Any] | None = None,
    tree_hash: str = "",
    hotspots: list[dict[str, Any]] | None = None,
    issues: list[str] | None = None,
    key: str = "",
    subject: str = "",
    **extra_metrics: Any,
) -> Any:
    """Build a corpus snapshot from the live registry.

    ``key`` and ``subject`` are named rather than left to ``extra_metrics``:
    the base takes ``**extra_metrics`` too, so an unnamed one would be
    recorded as a metric instead of reaching the fields it belongs in.

    :param version: Version string; auto-detected from package if None.
    :param branch: Git branch name; auto-detected if None.
    :param graph_stats_dict: Ignored; corpus metrics are computed from the registry.
    :param tree_hash: Git tree hash; recorded as provenance, no longer the key.
    :param hotspots: Unused; accepted for LSP compatibility.
    :param issues: Unused; accepted for LSP compatibility.
    :param key: Snapshot identifier; a UTC timestamp when omitted, which is
        what a corpus wants — pass a release tag only when snapshotting the
        package rather than the books.
    :param subject: What was measured; defaults to :data:`SNAPSHOT_SUBJECT`.
    :param extra_metrics: Passed through to the base ``capture`` call.
    :return: New ``Snapshot`` instance (not yet persisted).
    """
    genre_stats = collect_genre_stats(self.registry_path)
    metrics: dict[str, Any] = {
        "total_nodes": sum(g["nodes"] for g in genre_stats),
        "total_edges": sum(g["edges"] for g in genre_stats),
        "total_books": sum(g["books"] for g in genre_stats),
        "total_authors": _count_authors(self.corpus_root),
        "genres": genre_stats,
    }
    return super().capture(
        version=version,
        branch=branch,
        graph_stats_dict=metrics,
        tree_hash=tree_hash,
        key=key,
        subject=subject or SNAPSHOT_SUBJECT,
        **extra_metrics,
    )

save_snapshot(snapshot, *, force=False)

Persist snapshot, guarding against degenerate (zero-book) state.

Parameters:

Name	Type	Description	Default
snapshot	Any	Snapshot to persist.	required
force	bool	If True, always create a new history entry.	False

Raises:

Type	Description
ValueError	If the snapshot has 0 books.

Source code in src/gutenberg_kg/corpus.py

def save_snapshot(self, snapshot: Any, *, force: bool = False) -> Any:  # type: ignore[override]
    """Persist snapshot, guarding against degenerate (zero-book) state.

    :param snapshot: Snapshot to persist.
    :param force: If True, always create a new history entry.
    :raises ValueError: If the snapshot has 0 books.
    """
    m = snapshot.metrics if isinstance(snapshot.metrics, dict) else {}
    if m.get("total_books", 0) == 0:
        raise ValueError(
            "Refusing to save snapshot with 0 books. "
            "Ensure the corpus has ingested books before capturing a snapshot."
        )
    return super().save_snapshot(snapshot, force=force)

collect_genre_stats(registry_path)

Aggregate per-genre stats from the KGRAG registry.

Returns a list of dicts with keys: corpus, label, books, nodes, edges. Only SQLite files that exist on disk are counted.

Parameters:

Name	Type	Description	Default
registry_path	Path	Path to the KGRAG registry SQLite file.	required

Returns:

Type	Description
list[dict[str, Any]]	Per-genre stat dicts in GENRE_LABELS order.

Raises:

Type	Description
OSError	If the registry cannot be opened.

Source code in src/gutenberg_kg/corpus.py

def collect_genre_stats(registry_path: Path) -> list[dict[str, Any]]:
    """Aggregate per-genre stats from the KGRAG registry.

    Returns a list of dicts with keys: ``corpus``, ``label``, ``books``,
    ``nodes``, ``edges``.  Only SQLite files that exist on disk are counted.

    :param registry_path: Path to the KGRAG registry SQLite file.
    :return: Per-genre stat dicts in ``GENRE_LABELS`` order.
    :raises OSError: If the registry cannot be opened.
    """
    results: list[dict[str, Any]] = []
    try:
        reg = sqlite3.connect(str(registry_path))
    except Exception as exc:  # noqa: BLE001
        raise OSError(f"Cannot open registry {registry_path}: {exc}") from exc

    kg_map: dict[str, str | None] = {}
    for row in reg.execute("SELECT id, sqlite_path FROM kg_entries"):
        kg_map[row[0]] = row[1]

    for corpus_key, label in GENRE_LABELS.items():
        row = reg.execute("SELECT kg_ids FROM corpora WHERE name = ?", (corpus_key,)).fetchone()
        if row is None:
            results.append(
                {
                    "corpus": corpus_key,
                    "label": label,
                    "books": 0,
                    "nodes": 0,
                    "edges": 0,
                }
            )
            continue

        kg_ids: list[str] = json.loads(row[0])
        total_nodes = total_edges = live_books = 0
        for kid in kg_ids:
            sqlite_path = kg_map.get(kid)
            if sqlite_path and Path(sqlite_path).exists():
                n, e = _sqlite_counts(sqlite_path)
                total_nodes += n
                total_edges += e
                live_books += 1

        results.append(
            {
                "corpus": corpus_key,
                "label": label,
                "books": live_books,
                "nodes": total_nodes,
                "edges": total_edges,
            }
        )

    reg.close()
    return results

corpus_status(registry_path, repo_root, corpus_root)

Return a live corpus-wide status dict.

Parameters:

Name	Type	Description	Default
registry_path	Path	KGRAG registry SQLite path.	required
repo_root	Path	Root of the gutenberg_kg repository (for git info).	required
corpus_root	Path	Corpus data directory (contains authors/ etc.).	required

Returns:

Type	Description
dict[str, Any]	Dict with kind, timestamp, version, branch, commit, host, platform, totals, and genres.

Source code in src/gutenberg_kg/corpus.py

def corpus_status(
    registry_path: Path,
    repo_root: Path,
    corpus_root: Path,
) -> dict[str, Any]:
    """Return a live corpus-wide status dict.

    :param registry_path: KGRAG registry SQLite path.
    :param repo_root: Root of the gutenberg_kg repository (for git info).
    :param corpus_root: Corpus data directory (contains ``authors/`` etc.).
    :return: Dict with ``kind``, ``timestamp``, ``version``, ``branch``,
        ``commit``, ``host``, ``platform``, ``totals``, and ``genres``.
    """
    import platform
    import socket

    genre_stats = collect_genre_stats(registry_path)
    git = _git_info(repo_root)

    try:
        version = importlib.metadata.version("gutenberg-kg")
    except Exception:  # noqa: BLE001
        version = "unknown"

    return {
        "kind": "corpus_status",
        "timestamp": datetime.now(UTC).isoformat(timespec="seconds"),
        "version": version,
        "branch": git["branch"],
        "commit": git["commit"],
        "host": socket.gethostname(),
        "platform": platform.platform(),
        "totals": {
            "books": sum(g["books"] for g in genre_stats),
            "authors": _count_authors(corpus_root),
            "nodes": sum(g["nodes"] for g in genre_stats),
            "edges": sum(g["edges"] for g in genre_stats),
        },
        "genres": genre_stats,
    }

snapshot_list(snapshots_dir)

Load and return all snapshots, oldest first.

Reads what :class:GutenbergSnapshotManager writes: a manifest.json listing one <key>.json per snapshot. Each record carries key, branch, timestamp, version and a metrics dict.

This used to glob snapshot-*.json — the filename the retired :func:snapshot_save produced — which silently returned nothing once gutenkg snapshot save moved to the manager, taking gutenkg viz-timeline with it. The manifest is authoritative and already ordered; the glob is a fallback for a directory whose manifest is missing.

Parameters:

Name	Type	Description	Default
snapshots_dir	Path	Directory containing manifest.json and the per-snapshot files, e.g. corpus/.snapshots/.	required

Returns:

Type	Description
list[dict[str, Any]]	Snapshot dicts oldest first; empty if the directory is absent or holds nothing readable.

Source code in src/gutenberg_kg/corpus.py

def snapshot_list(snapshots_dir: Path) -> list[dict[str, Any]]:
    """Load and return all snapshots, oldest first.

    Reads what :class:`GutenbergSnapshotManager` writes: a ``manifest.json``
    listing one ``<key>.json`` per snapshot.  Each record carries ``key``,
    ``branch``, ``timestamp``, ``version`` and a ``metrics`` dict.

    This used to glob ``snapshot-*.json`` — the filename the retired
    :func:`snapshot_save` produced — which silently returned nothing once
    ``gutenkg snapshot save`` moved to the manager, taking ``gutenkg
    viz-timeline`` with it.  The manifest is authoritative and already ordered;
    the glob is a fallback for a directory whose manifest is missing.

    :param snapshots_dir: Directory containing ``manifest.json`` and the
        per-snapshot files, e.g. ``corpus/.snapshots/``.
    :return: Snapshot dicts oldest first; empty if the directory is absent or
        holds nothing readable.
    """
    if not snapshots_dir.is_dir():
        return []

    manifest = snapshots_dir / "manifest.json"
    records: list[dict[str, Any]] = []
    if manifest.is_file():
        try:
            records = json.loads(manifest.read_text(encoding="utf-8")).get("snapshots", [])
        except Exception:  # noqa: BLE001
            records = []

    if not records:
        # No manifest, or an unreadable one: read the snapshot files directly.
        for p in sorted(snapshots_dir.glob("*.json")):
            if p.name == "manifest.json":
                continue
            try:
                records.append(json.loads(p.read_text(encoding="utf-8")))
            except Exception:  # noqa: BLE001
                pass

    # Oldest first, by the field the caller plots against rather than by
    # filename — older files are named for a tree hash, which does not sort.
    return sorted(records, key=lambda s: s.get("timestamp", ""))