scripts/fingerprint.py

"""
Fingerprint the artifacts and source files the process actually loaded.

A deployed container can diverge from the repository in two independent ways:
the code can be stale (an old commit, or a cached layer), or the artifacts can
be stale (a previous build's download that `ensure_artifacts` then skips,
because the files already exist on disk). Reading the app's output cannot tell
these apart — both produce "wrong answers" — and neither can comparing the Hub
against a local checkout, since that says nothing about what the container has.

Hashing from inside the process resolves it. Run this locally and read the same
lines from the container's startup log: if the source hashes differ, the code
path is wrong; if the artifact hashes differ, the artifacts are stale; if
everything matches and behaviour still differs, the cause is neither.

AI Attribution: Fingerprint design assisted by Claude (Anthropic, https://claude.ai).
"""

import hashlib
from pathlib import Path


# Ordered so the source files — the usual suspect after a push — come first.
TRACKED_PATHS = [
    "scripts/recommender.py",
    "scripts/artifacts.py",
    "main.py",
    "data/processed/finetuned.index",
    "data/processed/base.index",
    "data/processed/finetuned_embeddings.npy",
    "data/processed/base_embeddings.npy",
    "data/processed/paper_metadata.pkl",
    "models/fine_tuned/model.safetensors",
    "models/base/model.safetensors",
]

CHUNK_BYTES = 1 << 20  # 1 MiB


def sha256_file(path, digest_chars=16):
    """Hash a file with SHA-256, streaming so large files stay off the heap.

    Args:
        path: Path to the file.
        digest_chars: How many leading hex characters to return. Sixteen is
            far more than needed to distinguish two builds and stays readable
            in a log line.

    Returns:
        Truncated hex digest, or a marker string if the file is absent.
    """
    p = Path(path)
    if not p.exists():
        return "MISSING"

    h = hashlib.sha256()
    with open(p, "rb") as f:
        for chunk in iter(lambda: f.read(CHUNK_BYTES), b""):
            h.update(chunk)
    return h.hexdigest()[:digest_chars]


def fingerprint(paths=None):
    """Hash every tracked path.

    Args:
        paths: Optional list of paths to hash. Defaults to TRACKED_PATHS.

    Returns:
        Dict mapping path to truncated digest, size in bytes, and existence.
    """
    result = {}
    for path in paths or TRACKED_PATHS:
        p = Path(path)
        result[path] = {
            "sha256": sha256_file(p),
            "bytes": p.stat().st_size if p.exists() else 0,
        }
    return result


def print_fingerprint(paths=None, header="FINGERPRINT"):
    """Print the fingerprint table.

    Called at app startup so the lines land in the container log, and runnable
    locally so the two can be diffed line by line.

    Args:
        paths: Optional list of paths to hash.
        header: Label for the table.
    """
    fp = fingerprint(paths)
    width = max(len(p) for p in fp)

    print("\n" + "=" * (width + 34))
    print(f"  {header}")
    print("=" * (width + 34))
    for path, info in fp.items():
        size = f"{info['bytes'] / 1e6:.1f}MB" if info["bytes"] else "-"
        print(f"  {path:<{width}}  {info['sha256']}  {size:>9}")
    print("=" * (width + 34) + "\n")

    return fp


def main():
    """Print the fingerprint of the current working tree."""
    print_fingerprint(header="LOCAL FINGERPRINT")


if __name__ == "__main__":
    main()