File size: 1,608 Bytes
acf77ab | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 | from __future__ import annotations
import hashlib
import json
from collections.abc import Iterable
from datetime import UTC, datetime
from pathlib import Path
from typing import Any
def write_corpus(nodes: Iterable[dict[str, Any]], out_path: Path) -> Path:
"""Write nodes as sorted JSONL."""
out_path.parent.mkdir(parents=True, exist_ok=True)
sorted_nodes = sorted(nodes, key=lambda n: str(n["id"]))
with out_path.open("w", encoding="utf-8") as f:
for node in sorted_nodes:
f.write(json.dumps(node, default=str))
f.write("\n")
return out_path
def compute_corpus_sha256(corpus_path: Path) -> str:
"""SHA256 hex digest of the corpus file."""
return hashlib.sha256(corpus_path.read_bytes()).hexdigest()
def write_manifest(
*,
corpus_path: Path,
sources: list[dict[str, str]],
scraped_files: int,
skipped_files: int,
total_nodes: int,
errors: list[dict[str, str]],
) -> Path:
"""Write a manifest JSON alongside the corpus."""
digest = compute_corpus_sha256(corpus_path)
manifest = {
"generated_at": datetime.now(UTC).isoformat(timespec="seconds"),
"sources": sources,
"scraped_files": scraped_files,
"skipped_files": skipped_files,
"total_nodes": total_nodes,
"errors": errors,
"corpus_sha256": digest,
}
manifest_path = corpus_path.with_suffix(".manifest.json")
manifest_path.parent.mkdir(parents=True, exist_ok=True)
manifest_path.write_text(
json.dumps(manifest, indent=2), encoding="utf-8"
)
return manifest_path
|