gnommo/gnommo/state.py

"""Persistent per-stage completion tracking.

Each pipeline stage (preprocess, trim, render) records a fingerprint of
its inputs in ``.gnommo_state.json`` when it completes successfully. On the next
run a stage can ask whether its inputs are unchanged (and its output still
present) and skip the work — the same staleness intelligence that ``all``'s
in-memory cascade provides, but persisted so it also applies to stages run on
their own.

Fingerprinting is hybrid:
  - small text manifests (narration.json, videos.json, manuscript.txt,
    project.json, slides.json, audio.json, transcripts) are hashed (sha256) so a
    ``touch`` or a git checkout that only rewrites mtimes doesn't force a
    needless rerun;
  - large media (processed narration segments, source videos and
    images) use mtime+size, which is cheap and good enough to detect real edits.

The state file is purely an optimization: any read/parse/write failure degrades
to "not current" (rerun) and never raises, so a corrupt or missing state file
can't break a build.
"""

from __future__ import annotations

import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable, Optional, Union

STATE_FILENAME = ".gnommo_state.json"
STATE_VERSION = 1

# Fingerprint modes
HASH = "hash"  # sha256 of file contents — for small text manifests
META = "meta"  # mtime_ns + size — for large media

# An input descriptor is a (label, path, mode) triple.
InputSpec = tuple[str, Path, str]


def _state_path(project_path: Path) -> Path:
    return project_path / STATE_FILENAME


def _empty_state() -> dict:
    return {"version": STATE_VERSION, "stages": {}}


def load_state(project_path: Path) -> dict:
    """Load the state file, returning an empty skeleton on any problem."""
    path = _state_path(project_path)
    if not path.exists():
        return _empty_state()
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return _empty_state()
    if not isinstance(data, dict):
        return _empty_state()
    data.setdefault("version", STATE_VERSION)
    if not isinstance(data.get("stages"), dict):
        data["stages"] = {}
    return data


def save_state(project_path: Path, state: dict) -> None:
    """Write the state file. Never raises — state is best-effort."""
    try:
        _state_path(project_path).write_text(
            json.dumps(state, indent=2) + "\n", encoding="utf-8"
        )
    except OSError:
        pass


def fingerprint_path(path: Union[str, Path], mode: str) -> Optional[str]:
    """Return a fingerprint for a single file, or None if it can't be read."""
    p = Path(path)
    try:
        if mode == HASH:
            h = hashlib.sha256()
            with open(p, "rb") as fh:
                for chunk in iter(lambda: fh.read(1 << 20), b""):
                    h.update(chunk)
            return f"sha256:{h.hexdigest()}"
        st = p.stat()
        return f"meta:{st.st_mtime_ns}:{st.st_size}"
    except OSError:
        return None


def compute(inputs: Iterable[InputSpec]) -> dict:
    """Build a {label: fingerprint} map from (label, path, mode) triples.

    A missing file yields a null fingerprint, so a file appearing or disappearing
    counts as a change.
    """
    return {label: fingerprint_path(path, mode) for label, path, mode in inputs}


def get_stage(project_path: Path, stage_key: str) -> dict:
    """Return the recorded record for a stage (``{}`` if none)."""
    return load_state(project_path).get("stages", {}).get(stage_key, {})


def get_items(project_path: Path, stage_key: str) -> dict:
    """Return the per-item fingerprint map recorded for a stage (``{}`` if none)."""
    items = get_stage(project_path, stage_key).get("items")
    return items if isinstance(items, dict) else {}


def is_current(
    project_path: Path,
    stage_key: str,
    inputs: dict,
    outputs: Iterable[Union[str, Path]] = (),
) -> bool:
    """True iff the recorded input fingerprint matches ``inputs`` exactly and
    every output in ``outputs`` exists on disk."""
    for out in outputs:
        if not Path(out).exists():
            return False
    recorded = get_stage(project_path, stage_key).get("inputs")
    return recorded == inputs


def record(project_path: Path, stage_key: str, inputs: dict) -> None:
    """Persist a stage-level input fingerprint, marking the stage complete."""
    state = load_state(project_path)
    state.setdefault("stages", {})[stage_key] = {
        "completed_at": datetime.now(timezone.utc).isoformat(),
        "inputs": inputs,
    }
    save_state(project_path, state)


def record_items(project_path: Path, stage_key: str, items: dict) -> None:
    """Merge per-item fingerprints into a stage's record (for multi-segment
    stages like preprocess/trim). Existing items for other keys are preserved."""
    state = load_state(project_path)
    stage = state.setdefault("stages", {}).setdefault(stage_key, {})
    stage["completed_at"] = datetime.now(timezone.utc).isoformat()
    merged = stage.get("items")
    if not isinstance(merged, dict):
        merged = {}
    merged.update(items)
    stage["items"] = merged
    save_state(project_path, state)