sources

Source scanning and manifest management for wiki_toolkit.

CLASS	DESCRIPTION
`Batch`	A single batch of files for parallel ingest dispatch.
`BatchPlan`	The full result of a `batch-plan` pass.
`BatchStats`	Aggregate totals across a `batch-plan` run.
`DedupeCandidate`	One file within a duplicate group.
`DedupeGroup`	All `docs/sources/` files sharing a `source` id, where at least one is flagged `duplicate: true`.
`DedupeResult`	The full result of a `source-dedupe` pass.
`Delta`	Result of diffing a source’s current content against its last-known revision on `main`.
`LintViolation`	A single rule violation found in a wiki note or source file.
`LoadError`	A Markdown file whose frontmatter failed to parse.
`SnapshotResult`	Result of writing a new Raw snapshot unit for a source.
`SourceCoverageEntry`	A single `docs/sources/` file’s coverage status.
`SourceCoverageResult`	The full result of a `source-coverage` pass.
`SourceLintResult`	The full result of a `source-lint` pass.
`SourceScanEntry`	A single `docs/sources/` file’s classification result.
`SourceScanResult`	The full result of a `source-scan` pass.

FUNCTION	DESCRIPTION
`apply_source_scan`	Write a `scan_sources` result: stamp source frontmatter, update the manifest.
`compute_source_delta`	Diff a source’s current working-tree content against its last-known revision on `main`.
`diff_content_fields`	Diff two frontmatter metadata dicts, excluding CLI bookkeeping fields (`processed`, `duplicate`, `source`).
`last_known_revision`	Return `rel_path`’s content as of the last commit touching it on `main`, or None if never committed there.
`lint_sources`	Validate every file in `docs_dir/sources/`: required `source` field, `processed`/`duplicate` types.
`plan_batches`	Split the files under `source_dir` into batches of at most `BATCH_BYTE_CAP` bytes or `BATCH_FILE_CAP` files.
`scan_sources`	Classify each file in `docs_dir/sources/` as new, update, or duplicate.
`source_coverage`	Report which `docs_dir/sources/` files are covered by at least one wiki note.
`suggest_dedupe`	Group `docs_dir/sources/` files by shared `source` id and suggest which to keep.
`write_source_snapshot`	Write a new Raw snapshot unit for `source`, for the given mutation type (`comments` or `fields`).

Classes¶

Batch `dataclass` ¶

Batch(id: str, files: list[str] = list(), total_bytes: int = 0)

A single batch of files for parallel ingest dispatch.

BatchPlan `dataclass` ¶

BatchPlan(batches: list[Batch], stats: BatchStats)

The full result of a batch-plan pass.

BatchStats `dataclass` ¶

BatchStats(total_files: int, total_bytes: int, batch_count: int)

Aggregate totals across a batch-plan run.

DedupeCandidate `dataclass` ¶

DedupeCandidate(path: str, mtime: float, similarity: float)

One file within a duplicate group.

DedupeGroup `dataclass` ¶

DedupeGroup(
    source: str,
    keep: str,
    reason: str,
    candidates: list[DedupeCandidate] = list(),
)

All docs/sources/ files sharing a source id, where at least one is flagged duplicate: true.

DedupeResult `dataclass` ¶

DedupeResult(
    groups: list[DedupeGroup] = list(), violations: list[LintViolation] = list()
)

The full result of a source-dedupe pass.

ATTRIBUTE	DESCRIPTION
`needs_attention`	True if any duplicate group was found. TYPE: `bool`

Attributes¶

needs_attention `property` ¶

needs_attention: bool

True if any duplicate group was found.

Delta `dataclass` ¶

Delta(changed_fields: dict[str, tuple[Any, Any]] = dict())

Result of diffing a source’s current content against its last-known revision on main.

LintViolation `dataclass` ¶

LintViolation(path: str, message: str)

A single rule violation found in a wiki note or source file.

LoadError `dataclass` ¶

LoadError(message: str)

A Markdown file whose frontmatter failed to parse.

SnapshotResult `dataclass` ¶

SnapshotResult(source: str, path: str, units: SnapshotUnits, update_sha: str)

Result of writing a new Raw snapshot unit for a source.

SourceCoverageEntry `dataclass` ¶

SourceCoverageEntry(
    source: str,
    path: str,
    title: str,
    covered: bool,
    covered_by: list[str] = list(),
)

A single docs/sources/ file’s coverage status.

SourceCoverageResult `dataclass` ¶

SourceCoverageResult(
    entries: list[SourceCoverageEntry] = list(),
    violations: list[LintViolation] = list(),
)

The full result of a source-coverage pass.

ATTRIBUTE	DESCRIPTION
`covered`	Entries covered by at least one wiki note. TYPE: `list[SourceCoverageEntry]`
`uncovered`	Entries covered by no wiki note. TYPE: `list[SourceCoverageEntry]`

Attributes¶

covered `property` ¶

covered: list[SourceCoverageEntry]

Entries covered by at least one wiki note.

uncovered `property` ¶

uncovered: list[SourceCoverageEntry]

Entries covered by no wiki note.

SourceLintResult `dataclass` ¶

SourceLintResult(
    violations: list[LintViolation] = list(), backlog: list[str] = list()
)

The full result of a source-lint pass.

ATTRIBUTE	DESCRIPTION
`ok`	True if no violations were found. Backlog entries don’t affect this. TYPE: `bool`

Attributes¶

ok `property` ¶

ok: bool

True if no violations were found. Backlog entries don’t affect this.

SourceScanEntry `dataclass` ¶

SourceScanEntry(
    source: str,
    path: str,
    title: str,
    classification: SourceClassification,
    covered: bool = False,
    accepted: bool = True,
)

A single docs/sources/ file’s classification result.

ATTRIBUTE	DESCRIPTION
`needs_accept_covered`	True if this is an `update` to an already-covered source that `--accept-covered` would unblock. TYPE: `bool`

Attributes¶

needs_accept_covered `property` ¶

needs_accept_covered: bool

True if this is an update to an already-covered source that --accept-covered would unblock.

SourceScanResult `dataclass` ¶

SourceScanResult(
    entries: list[SourceScanEntry] = list(),
    skipped: list[str] = list(),
    violations: list[LintViolation] = list(),
)

The full result of a source-scan pass.

ATTRIBUTE	DESCRIPTION
`needs_attention`	True if any entry is a duplicate or an unaccepted covered update. TYPE: `bool`

Attributes¶

needs_attention `property` ¶

needs_attention: bool

True if any entry is a duplicate or an unaccepted covered update.

Functions:¶

apply_source_scan ¶

apply_source_scan(docs_dir: Path, result: SourceScanResult) -> int

Write a scan_sources result: stamp source frontmatter, update the manifest.

Duplicate files are stamped duplicate: true and get no manifest entry. Unaccepted (covered, not --accept-covered) updates are left untouched. Returns the number of manifest entries written.

compute_source_delta ¶

compute_source_delta(docs_dir: Path, source: str) -> Delta

Diff a source’s current working-tree content against its last-known revision on main.

Resolves the source’s path via source-manifest.jsonl. A source with no prior commit on main diffs against a synthetic empty baseline (every field reports as new) rather than erroring.

diff_content_fields ¶

diff_content_fields(old: dict, new: dict) -> dict[str, tuple[Any, Any]]

Diff two frontmatter metadata dicts, excluding CLI bookkeeping fields (processed, duplicate, source).

Returns {field: (old_value, new_value)} for every field that was added, removed, or changed.

last_known_revision ¶

last_known_revision(root: Path, rel_path: str) -> str | None

Return rel_path’s content as of the last commit touching it on main, or None if never committed there.

lint_sources ¶

lint_sources(docs_dir: Path) -> SourceLintResult

Validate every file in docs_dir/sources/: required source field, processed/duplicate types.

Also reports processed sources with no covered_by entry in docs_dir/source-manifest.jsonl as a backlog list, distinct from hard errors.

plan_batches ¶

plan_batches(source_dir: Path) -> BatchPlan

Split the files under source_dir into batches of at most BATCH_BYTE_CAP bytes or BATCH_FILE_CAP files.

Files are visited in sorted order for determinism. A batch closes as soon as adding the next file would cross either cap; a single file larger than BATCH_BYTE_CAP still gets its own batch rather than being split. Reports zero files if source_dir doesn’t exist.

scan_sources ¶

scan_sources(
    docs_dir: Path, *, accept_covered: bool = False
) -> SourceScanResult

Classify each file in docs_dir/sources/ as new, update, or duplicate.

Skips version-controlled sources (frontmatter kind: version_controlled). Within a scan pass, the first-seen file for a source id is canonical; later files with the same id are classified duplicate. A file already stamped duplicate: true from a prior scan stays excluded regardless of scan order, until a human resolves it via source-dedupe. A file with malformed frontmatter is reported as a violation instead of raising.

source_coverage ¶

source_coverage(docs_dir: Path) -> SourceCoverageResult

Report which docs_dir/sources/ files are covered by at least one wiki note.

Cross-references source-manifest.jsonl’s covered_by field against catalog.jsonl’s sources lists. Duplicates are excluded using the same rule as scan_sources: a file stamped duplicate: true, or a later file sharing a source id already seen in this pass. A file with malformed frontmatter is reported as a violation instead of raising.

suggest_dedupe ¶

suggest_dedupe(docs_dir: Path) -> DedupeResult

Group docs_dir/sources/ files by shared source id and suggest which to keep.

A group is included only for source ids with at least one duplicate: true file and more than one file sharing the id (a lone duplicate: true file with no sibling has nothing to compare against). The suggested keeper is the file with the latest mtime; content-similarity scores (difflib ratio against the keeper) are reported per candidate so a human can confirm the call. Never modifies or deletes files — suggestion only. A file with malformed frontmatter is reported as a violation instead of raising.

write_source_snapshot ¶

write_source_snapshot(
    docs_dir: Path, source: str, units: str
) -> SnapshotResult

Write a new Raw snapshot unit for source, for the given mutation type (comments or fields).

Resolves the source’s path via source-manifest.jsonl, resets processed: false on the file — the existing reprocessing signal (see apply_source_scan) — and records the current on-disk content’s hash and timestamp as the manifest’s new update_sha/updated, giving an explicit, versioned record of this mutation (the “computed SHA hash of files for mutable sources” case the manifest’s update_sha field already covers). Works entirely against content already materialized on disk; v1 has no live adapter fetch to populate units: comments from, so units is a bookkeeping distinction for downstream PR framing, not a different write.

sources

Classes¶

Batch dataclass ¶

BatchPlan dataclass ¶

BatchStats dataclass ¶

DedupeCandidate dataclass ¶

DedupeGroup dataclass ¶

DedupeResult dataclass ¶

Attributes¶

needs_attention property ¶

Delta dataclass ¶

LintViolation dataclass ¶

LoadError dataclass ¶

SnapshotResult dataclass ¶

SourceCoverageEntry dataclass ¶

SourceCoverageResult dataclass ¶

Attributes¶

covered property ¶

uncovered property ¶

SourceLintResult dataclass ¶

Attributes¶

ok property ¶

SourceScanEntry dataclass ¶

Attributes¶

needs_accept_covered property ¶

SourceScanResult dataclass ¶

Attributes¶

needs_attention property ¶

Functions:¶

apply_source_scan ¶

compute_source_delta ¶

diff_content_fields ¶

last_known_revision ¶

lint_sources ¶

plan_batches ¶

scan_sources ¶

source_coverage ¶

suggest_dedupe ¶

write_source_snapshot ¶

Batch `dataclass` ¶

BatchPlan `dataclass` ¶

BatchStats `dataclass` ¶

DedupeCandidate `dataclass` ¶

DedupeGroup `dataclass` ¶

DedupeResult `dataclass` ¶

needs_attention `property` ¶

Delta `dataclass` ¶

LintViolation `dataclass` ¶

LoadError `dataclass` ¶

SnapshotResult `dataclass` ¶

SourceCoverageEntry `dataclass` ¶

SourceCoverageResult `dataclass` ¶

covered `property` ¶

uncovered `property` ¶

SourceLintResult `dataclass` ¶

ok `property` ¶

SourceScanEntry `dataclass` ¶

needs_accept_covered `property` ¶

SourceScanResult `dataclass` ¶

needs_attention `property` ¶