Corpus audit

gutenberg_kg.audit

audit.py — Verify Project Gutenberg corpus integrity.

Checks each book under corpus/ for the invariants that download / ingest / build-corpus assume, catching the failure modes that are easy to introduce by hand:

• missing full-text .md or reference.md;
• a diary whose .md does not parse with its declared .diary_format (wrong/missing format);
• a stray .dockg/ inside a diary directory (diaries must use .diarykg/);
• the same Gutenberg ID assigned to more than one book (a mix-up/swap);
• a catalog title override that differs from the directory name of the book already downloaded for that ID (catalog/corpus naming drift);
• a downloaded book that its genre catalog does not list at all;
• a registered KG whose index file no longer exists, or a diary registered to a .dockg index instead of .diarykg.

"Not built" / "not registered" / "not recorded in the catalog" are warnings (the first two are expected on a fresh clone before rebuild-indices); the rest are errors. run_audit returns a non-zero exit code when any error is found, so it is CI-friendly.

Note the two catalog checks run in opposite directions and only together cover membership. The title check walks catalog entries and looks up books; the uncatalogued check walks books and looks up catalog entries. Only the first existed for a long time, which is how 90 of 243 books came to be missing from their catalogs while every book audited clean.

AuditReport(books=list(), registry_orphans=list(), registry_found=True) dataclass

Aggregated audit results plus corpus-wide findings.

n_errors property

Number of books with at least one error (+ registry orphans).

n_warnings property

Number of books with warnings but no errors.

BookAudit(genre, book, is_diary, ebook_id=None, ia_id=None, fmt=None, entries=None, built=False, registered=False, errors=list(), warnings=list()) dataclass

Audit result for a single book directory.

ok property

True when the book has no errors and no warnings.

status property

One-word status: ERROR / WARN / OK.

audit_corpus(genres=None, registry=None)

Audit every book in the selected genres and return an :class:AuditReport.

Parameters:

Name	Type	Description	Default
genres	list[str] | None	Genres to audit (None = all genres present on disk).	None
registry	str | Path | None	Override KGRAG registry path.	None

Returns:

Type	Description
AuditReport	The populated audit report.

Source code in src/gutenberg_kg/audit.py

def audit_corpus(
    genres: list[str] | None = None,
    registry: str | Path | None = None,
) -> AuditReport:
    """Audit every book in the selected genres and return an :class:`AuditReport`.

    :param genres: Genres to audit (``None`` = all genres present on disk).
    :param registry: Override KGRAG registry path.
    :return: The populated audit report.
    """
    genres = list(genres) if genres else list(ALL_GENRES)
    reg_path = _registry_path(registry)
    reg_map = _load_registry(reg_path)
    registry_found = reg_path.exists() and bool(reg_map)

    report = AuditReport(registry_found=registry_found)

    for genre in genres:
        genre_dir = CORPUS_ROOT / genre
        if not genre_dir.is_dir():
            continue
        for book_dir in sorted(genre_dir.iterdir()):
            if not book_dir.is_dir() or book_dir.name.startswith("."):
                continue
            report.books.append(_audit_book(book_dir, genre, reg_map, registry_found))

    # Corpus-wide: duplicate Gutenberg IDs.
    by_id: dict[int, list[BookAudit]] = {}
    for b in report.books:
        if b.ebook_id is not None:
            by_id.setdefault(b.ebook_id, []).append(b)
    for eid, group in by_id.items():
        if len(group) > 1:
            others = ", ".join(f"{g.genre}/{g.book}" for g in group)
            for b in group:
                b.errors.append(f"duplicate Gutenberg ID {eid} (shared by: {others})")

    # The same check for IA items. This had no coverage: the check above keys on
    # ebook_id, which is None for every IA book, so two directories holding one
    # IA item both audited clean. Path-keyed idempotence in ia.download_book
    # made that reachable by simply passing a different --title.
    by_ia_id: dict[str, list[BookAudit]] = {}
    for b in report.books:
        if b.ia_id is not None:
            by_ia_id.setdefault(b.ia_id, []).append(b)
    for ia_id, group in by_ia_id.items():
        if len(group) > 1:
            others = ", ".join(f"{g.genre}/{g.book}" for g in group)
            for b in group:
                b.errors.append(f"duplicate Internet Archive ID {ia_id} (shared by: {others})")

    # Corpus-wide: catalog title override ≠ downloaded directory name. The
    # download path is keyed on the Gutenberg ID, so drift can no longer
    # duplicate books — but the catalog is the source of truth, and an
    # override that misnames an existing directory is still a lie in it.
    def _nfc(s: str) -> str:
        return unicodedata.normalize("NFC", s)

    by_genre_id = {(b.genre, b.ebook_id): b for b in report.books if b.ebook_id is not None}
    for genre in genres:
        if genre in IA_GENRES:
            continue
        catalog = CATALOG_ROOT / f"{genre}.txt"
        if not catalog.exists():
            continue
        for eid, cat_title in parse_catalog(str(catalog)):
            b = by_genre_id.get((genre, eid))
            if b is None or cat_title is None:
                continue  # not downloaded yet, or no override to compare
            if _nfc(cat_title) != _nfc(b.book):
                b.errors.append(
                    f"catalog title '{cat_title}' ≠ directory '{b.book}' "
                    f"(fix scripts/catalogs/{genre}.txt or rename the dir)"
                )

    # Corpus-wide: a downloaded book its catalog does not list. This direction
    # had no check at all — the block above walks catalog entries and looks up
    # books, so a book absent from the catalog is never examined by any loop.
    # That blind spot is why 90 of 243 books could be uncatalogued while the
    # audit reported every book clean.
    #
    # The cost is reproducibility, not correctness: these books are healthy, but
    # `download catalog` replayed over a fresh clone rebuilds only what the
    # catalogs list, so the uncatalogued ones are silently absent. A warning,
    # not an error, for exactly that reason.
    #
    # IA genres are covered too, keyed on the Internet Archive identifier. A
    # genre is either Gutenberg or IA and never both, so which parser reads its
    # catalog is unambiguous.
    for genre in genres:
        catalog = CATALOG_ROOT / f"{genre}.txt"
        is_ia = genre in IA_GENRES
        if not catalog.exists():
            catalogued: set = set()
        elif is_ia:
            catalogued = {ident for ident, _ in parse_ia_catalog(catalog)}
        else:
            catalogued = {eid for eid, _ in parse_catalog(str(catalog))}
        for b in report.books:
            if b.genre != genre:
                continue
            key = b.ia_id if is_ia else b.ebook_id
            if key is not None and key not in catalogued:
                b.warnings.append(f"not recorded in scripts/catalogs/{genre}.txt")

    # Registered KGs (within the audited genres) whose index file is gone.
    if registry_found:
        prefixes = tuple(f"gutenberg-{g}-" for g in genres)
        for name, sp in reg_map.items():
            if name.startswith(prefixes) and (not sp or not Path(sp).exists()):
                report.registry_orphans.append(name)

    return report

run_audit(genres=None, registry=None, as_json=False)

Run the corpus audit and print a report; return a process exit code.

Parameters:

Name	Type	Description	Default
genres	list[str] | None	Genres to audit (empty/None = all).	None
registry	str | Path | None	Override KGRAG registry path.	None
as_json	bool	Emit machine-readable JSON instead of a Rich table.	False

Returns:

Type	Description
int	1 if any errors were found, else 0.

Source code in src/gutenberg_kg/audit.py

def run_audit(
    genres: list[str] | None = None,
    registry: str | Path | None = None,
    as_json: bool = False,
) -> int:
    """Run the corpus audit and print a report; return a process exit code.

    :param genres: Genres to audit (empty/None = all).
    :param registry: Override KGRAG registry path.
    :param as_json: Emit machine-readable JSON instead of a Rich table.
    :return: 1 if any errors were found, else 0.
    """
    report = audit_corpus(genres, registry)

    if as_json:
        import json

        payload = {
            "registry_found": report.registry_found,
            "errors": report.n_errors,
            "warnings": report.n_warnings,
            "registry_orphans": report.registry_orphans,
            "books": [
                {
                    "genre": b.genre,
                    "book": b.book,
                    "is_diary": b.is_diary,
                    "ebook_id": b.ebook_id,
                    "format": b.fmt,
                    "entries": b.entries,
                    "built": b.built,
                    "registered": b.registered,
                    "status": b.status,
                    "errors": b.errors,
                    "warnings": b.warnings,
                }
                for b in report.books
            ],
        }
        print(json.dumps(payload, indent=2, ensure_ascii=False))
        return 1 if report.n_errors else 0

    from rich.console import Console
    from rich.table import Table

    console = Console()
    table = Table(title="GutenbergKG corpus audit", show_lines=False)
    table.add_column("status", no_wrap=True)
    table.add_column("genre", no_wrap=True)
    table.add_column("book", overflow="fold")
    table.add_column("id", justify="right", no_wrap=True)
    table.add_column("detail", overflow="fold")

    style = {"OK": "green", "WARN": "yellow", "ERROR": "bold red"}
    for b in report.books:
        if b.ok:
            continue  # keep the table to problems only; counts cover the rest
        detail = "; ".join(b.errors + b.warnings)
        table.add_row(
            f"[{style[b.status]}]{b.status}[/]",
            b.genre,
            b.book,
            str(b.ebook_id) if b.ebook_id is not None else "—",
            detail,
        )

    n_ok = sum(1 for b in report.books if b.ok)
    if table.row_count:
        console.print(table)
    else:
        console.print("[green]All audited books are clean.[/]")

    if not report.registry_found:
        console.print("[yellow]note:[/] KGRAG registry not found — registration checks skipped.")
    for name in report.registry_orphans:
        console.print(f"[bold red]ERROR[/] registry orphan: {name} → index file missing")

    console.print(
        f"\n  audited {len(report.books)} books  ·  "
        f"[green]{n_ok} ok[/]  ·  "
        f"[yellow]{report.n_warnings} warn[/]  ·  "
        f"[bold red]{report.n_errors} error[/]"
    )
    return 1 if report.n_errors else 0