SniperGold_ML/engine/manifest.py

"""Dataset manifest builder / verifier (Layer 10, spec 16, MS_V1).

Non-self-referential hashing: ``manifest_hash`` excludes both hash fields;
``dataset_hash`` is computed over the canonical ``relpath|content_id`` list
(never over the manifest that contains it). Files are sorted by relpath.
"""

import os
import json

from . import versions
from .util import atomic_write_json, canonical_json, canonical_json_sha256, sha256_bytes


def dataset_hash_from_files(files):
    """SHA-256 over sorted ``relpath|content_id\\n`` concatenation."""
    lines = []
    for relpath in sorted(files):
        cid = files[relpath].get("content_id") or ""
        lines.append("%s|%s\n" % (relpath, cid))
    return sha256_bytes("".join(lines))


def build_manifest(*, dataset_id, dataset_version, engine_versions, source_identity,
                   timeframe=None, row_counts=None, files, config_snapshot,
                   config_sha, extra=None):
    """Assemble and sign an MS_V1 manifest."""
    payload = {
        "schema_version": versions.MANIFEST_SCHEMA_VERSION,
        "dataset_id": dataset_id,
        "dataset_version": dataset_version,
        "engine_version": versions.ENGINE_VERSION,
        "parser_version": versions.PARSER_VERSION,
        "algorithm_version": versions.ALGORITHM_VERSION,
        "source_identity": source_identity,
        "row_counts": row_counts or {},
        "files": dict(sorted(files.items())),
        "config_snapshot": config_snapshot or {},
        "config_sha256": config_sha or "",
        "creation_ts_utc": _utc_now_iso(),
        "manifest_hash": None,
        "dataset_hash": None,
    }
    if timeframe:
        payload["timeframe"] = timeframe
    if extra:
        for k, v in extra.items():
            if k in payload:
                raise ValueError("extra manifest field collides: %s" % k)
            payload[k] = v
    payload["dataset_hash"] = dataset_hash_from_files(payload["files"])
    body = {k: v for k, v in payload.items() if k not in ("manifest_hash", "dataset_hash")}
    payload["manifest_hash"] = canonical_json_sha256(body)
    return payload


def write_manifest(path, manifest):
    os.makedirs(os.path.dirname(os.path.abspath(path)), exist_ok=True)
    atomic_write_json(path, manifest)


def load_manifest(path):
    with open(path, "r", encoding="utf-8") as fh:
        return json.load(fh)


def verify_manifest(manifest, out_root):
    """Recompute manifest_hash and dataset_hash from content; verify every
    listed file's physical sha. Returns (ok, diffs)."""
    diffs = []
    body = {k: v for k, v in manifest.items()
            if k not in ("manifest_hash", "dataset_hash")}
    if canonical_json_sha256(body) != manifest.get("manifest_hash"):
        diffs.append("manifest_hash mismatch")
    if dataset_hash_from_files(manifest.get("files", {})) != manifest.get("dataset_hash"):
        diffs.append("dataset_hash mismatch")
    for relpath, rec in manifest.get("files", {}).items():
        full = os.path.join(out_root, relpath)
        if not os.path.exists(full):
            diffs.append("missing:" + relpath)
    return (not diffs), diffs


def _utc_now_iso():
    import datetime
    return datetime.datetime.now(datetime.timezone.utc).isoformat().replace("+00:00", "Z")