manifest.py

"""
Index manifest: detect when raw source files have changed so the vector
and keyword indexes can be rebuilt together.

The manifest stores a lightweight hash (filenames + mtime + size) of all
raw-data directories.  If the hash changes between runs, load_or_build_database
will trigger a full rebuild even without RAG_FORCE_REBUILD=1.
"""
import os
import json
import hashlib
import logging
from pathlib import Path

logger = logging.getLogger(__name__)

_MANIFEST_FILENAME = "index_manifest.json"


def _manifest_path(chroma_db_path: str) -> Path:
    return Path(chroma_db_path).parent / _MANIFEST_FILENAME


def compute_raw_data_hash(raw_dirs: list[str]) -> str:
    """
    Return an MD5 hex digest computed from the sorted list of
    (filename, mtime, size) tuples across all files in raw_dirs.
    Ignores hidden files (starting with '.').
    """
    h = hashlib.md5()
    for raw_dir in sorted(raw_dirs):
        if not os.path.isdir(raw_dir):
            continue
        for filename in sorted(os.listdir(raw_dir)):
            if filename.startswith("."):
                continue
            filepath = os.path.join(raw_dir, filename)
            if not os.path.isfile(filepath):
                continue
            stat = os.stat(filepath)
            h.update(filename.encode())
            h.update(str(int(stat.st_mtime)).encode())
            h.update(str(stat.st_size).encode())
    # Also hash the product summaries source file so adding curated
    # knowledge triggers a rebuild.
    summary_file = Path(__file__).resolve().parent.parent / "knowledge" / "product_summary.py"
    if summary_file.is_file():
        stat = summary_file.stat()
        h.update(b"product_summary.py")
        h.update(str(int(stat.st_mtime)).encode())
        h.update(str(stat.st_size).encode())
    return h.hexdigest()


def load_manifest(chroma_db_path: str) -> dict:
    path = _manifest_path(chroma_db_path)
    if not path.exists():
        return {}
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception as e:
        logger.warning("Could not read index manifest: %s", e)
        return {}


def save_manifest(chroma_db_path: str, raw_dirs: list[str]) -> None:
    path = _manifest_path(chroma_db_path)
    path.parent.mkdir(parents=True, exist_ok=True)
    data = {"raw_data_hash": compute_raw_data_hash(raw_dirs)}
    try:
        path.write_text(json.dumps(data, indent=2), encoding="utf-8")
    except Exception as e:
        logger.warning("Could not save index manifest: %s", e)


def needs_rebuild(chroma_db_path: str, raw_dirs: list[str]) -> bool:
    """
    Return True when:
    - No manifest file exists yet, OR
    - The current raw-data hash differs from the stored hash.
    """
    manifest = load_manifest(chroma_db_path)
    if not manifest:
        return True
    current_hash = compute_raw_data_hash(raw_dirs)
    stored_hash = manifest.get("raw_data_hash", "")
    changed = current_hash != stored_hash
    if changed:
        logger.info("Raw data changed (hash mismatch) – index rebuild required.")
    return changed