studio/scripts/library_manifest.py

#!/usr/bin/env python3
"""library_manifest.py — regenerate the ONE canonical asset-library index.

The Karma content pipeline has a single source of truth: a `library.json`
manifest that sits at the root of an org's studio-output subtree (any tenant; pass --root) and
indexes every asset Antje has rendered. The storefront (karma.style) and the
socials/blog queue both read THIS file — there is no second index.

  library root (== S3 prefix  s3://hanzo-studio/orgs/karma/output/
                == studio pod  /app/orgs/karma/output/
                == spark local /home/z/work/hanzo/studio/output/orgs/karma/output/)
    designs/<slug>/<kind>_<role>.png     catalog assets  (ecom|product|lifestyle|hover)
    marketing/<channel>/<name>.png       queue assets    (social|blog|campaign)
    library.json                         <- this file, regenerated here

What this does, and nothing more:
  * walk the tree, classify each image by path -> {design, kind, role}
  * PRESERVE status/caption/tags from the existing library.json (keyed by path)
  * brand-new files enter as status="draft"; deleted files drop out
  * write library.json atomically (tmp + rename) so a reader never sees a
    half-written file, and the S3 mirror sidecar carries it up on its next pass

It writes ONLY library.json. It never touches images and never regenerates
products.json (that file carries human-curated pricing/copy the site owns).

Runs one-shot, or with --watch as the studio pod's `build-manifest` sidecar.
Pure stdlib — no PIL, no deps — so it runs in the studio image or anywhere.
"""
from __future__ import annotations

import argparse
import contextlib
import json
import os
import sys
import tempfile
import time
from datetime import datetime, timezone

try:
    import fcntl
except ImportError:  # non-POSIX (never in the studio image); lock is a no-op
    fcntl = None

LOCK_NAME = ".library.lock"


@contextlib.contextmanager
def library_lock(root: str):
    """Serialize the read-modify-write of library.json across processes — the
    build-manifest sidecar and the studio app both rebuild-then-replace it, so an
    unlocked interleave loses status/caption/tags updates. Same lock file name is
    used by the app writer (studio_home._save_library)."""
    if fcntl is None:
        yield
        return
    lock_path = os.path.join(root, LOCK_NAME)
    fh = open(lock_path, "w")
    try:
        fcntl.flock(fh, fcntl.LOCK_EX)
        yield
    finally:
        with contextlib.suppress(OSError):
            fcntl.flock(fh, fcntl.LOCK_UN)
        fh.close()

IMAGE_EXTS = {".png", ".jpg", ".jpeg", ".webp"}
VIDEO_EXTS = {".mp4", ".webm"}
MODEL3D_EXTS = {".glb"}
# Written marketing content (blog posts, campaign briefs, social ad copy) is
# first-class in the ONE index too — a .md under marketing/ is a queue asset
# whose body is the post/brief and whose caption/hashtags(=tags) are set with
# karma-queue.py. Restricted to marketing/ so catalog stays image-only.
MARKETING_TEXT_EXTS = {".md"}
MANIFEST_NAME = "library.json"

# The kinds a catalog asset may carry (designs/<slug>/<kind>_<role>.png). The
# storefront consumes ecom + product + lifestyle; hover is a card-hover alt.
CATALOG_KINDS = {"ecom", "product", "lifestyle", "hover"}
# Marketing channels double as the kind for marketing/<channel>/* assets.
MARKETING_KINDS = {"social", "blog", "campaign"}


def iso(ts: float) -> str:
    return datetime.fromtimestamp(ts, tz=timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _sha_of(path: str) -> str | None:
    """Streamed content hash (a 40MB PNG must not balloon memory). None on error."""
    import hashlib
    try:
        h = hashlib.md5()
        with open(path, "rb") as f:
            for chunk in iter(lambda: f.read(1 << 20), b""):
                h.update(chunk)
        return h.hexdigest()
    except OSError:
        return None


def classify(relpath: str, slugs: set[str]) -> dict | None:
    """Map a path relative to the library root -> {design, kind, role}.

    Returns None for anything that is not a manifestable asset.
    """
    parts = relpath.split("/")
    stem = os.path.splitext(parts[-1])[0]

    if parts[0] == "marketing" and len(parts) >= 3:
        channel = parts[1]
        kind = channel if channel in MARKETING_KINDS else "campaign"
        # design = leading slug token if it names a real design, else cross-cut.
        head, _, rest = stem.partition("_")
        if head in slugs and rest:
            design, role = head, rest
        else:
            design, role = None, stem
        return {"design": design, "kind": kind, "role": role}

    if parts[0] == "designs" and len(parts) >= 3:
        # 4k/print masters live alongside but are not catalog/queue items.
        if "4k" in parts[2:-1]:
            return None
        design = parts[1]
        kind, sep, role = stem.partition("_")
        if not sep:  # e.g. bare "front.png" -> treat stem as role, kind unknown
            kind, role = "other", stem
        if kind not in CATALOG_KINDS:
            kind = kind if kind in ("other",) else "other"
        return {"design": design, "kind": kind, "role": role}

    return None


def scan(root: str) -> list[str]:
    """All manifestable image paths (relative to root), sorted for stable diffs."""
    found: list[str] = []
    for base, dirs, files in os.walk(root):
        # Hidden segments are never assets — one rule covering AppleDouble
        # forks (._foo.png) AND cache trees (.thumbs/ thumbnails indexed as
        # library rows sent thumb paths into the fix dispatcher).
        dirs[:] = [d for d in dirs if not d.startswith(".")]
        for f in files:
            if f.startswith("."):
                continue
            ext = os.path.splitext(f)[1].lower()
            full = os.path.join(base, f)
            rel = os.path.relpath(full, root)
            is_image = ext in IMAGE_EXTS
            is_video = ext in VIDEO_EXTS
            is_model3d = ext in MODEL3D_EXTS
            is_mktg_text = (ext in MARKETING_TEXT_EXTS
                            and rel.replace(os.sep, "/").startswith("marketing/"))
            if not (is_image or is_video or is_model3d or is_mktg_text):
                continue
            try:
                if os.path.getsize(full) == 0:  # skip a render caught mid-write
                    continue
            except OSError:
                continue
            found.append(rel)
    return sorted(found)


def load_prior(path: str) -> dict[str, dict]:
    """Existing entries keyed by path, so status/caption/tags survive a rebuild."""
    try:
        with open(path, "r") as fh:
            doc = json.load(fh)
    except (OSError, json.JSONDecodeError):
        return {}
    return {a["path"]: a for a in doc.get("assets", []) if "path" in a}


def build(root: str) -> dict:
    manifest_path = os.path.join(root, MANIFEST_NAME)
    prior = load_prior(manifest_path)

    designs_dir = os.path.join(root, "designs")
    slugs = {d for d in os.listdir(designs_dir)} if os.path.isdir(designs_dir) else set()

    assets: list[dict] = []
    for rel in scan(root):
        norm = rel.replace(os.sep, "/")
        if norm == MANIFEST_NAME:
            continue
        info = classify(norm, slugs)
        prev = prior.get(norm, {})
        if info is None:
            # Taxonomy enriches, never gates: a file outside the naming scheme is
            # still an asset. Dropping it here erased every ingest-indexed render on
            # each sweep cycle. A video keeps kind="video" so it survives every sweep
            # as the motion-judged, video-card kind.
            _e = os.path.splitext(norm)[1].lower()
            kind = prev.get("kind") or (
                "model3d" if _e in MODEL3D_EXTS else "video" if _e in VIDEO_EXTS else "render")
            info = {"design": prev.get("design"), "kind": kind, "role": prev.get("role")}
        abspath = os.path.join(root, rel)
        mtime = os.path.getmtime(abspath)
        # Content hash powers the dedup invariant below. Reuse the prior sha when the
        # file is byte-stable (same mtime), so a sweep hashes only NEW/changed files —
        # O(new), not O(all), even on a large gallery. Only images are hashed (docs are
        # not the byte-duplicate-clutter concern).
        is_img = norm.lower().endswith((".png", ".jpg", ".jpeg", ".webp"))
        sha = prev.get("sha") if (prev.get("mtime") == mtime and prev.get("sha")) \
            else (_sha_of(abspath) if is_img else None)
        entry = {
            "path": norm,
            "design": info["design"],
            "kind": info["kind"],
            "role": info["role"],
            "status": prev.get("status", "draft"),
            "tags": prev.get("tags", []),
            "mtime": mtime,
            "sha": sha,
            "updatedAt": iso(mtime),
        }
        caption = prev.get("caption")
        if caption:
            entry["caption"] = caption
        assets.append(entry)

    # ── Structural invariant: AT MOST ONE active asset per content hash. A re-rendered
    # byte-identical duplicate (a retried / double-dispatched render) is auto-marked
    # deleted here — so the gallery can NEVER accumulate duplicate images, whatever bug
    # fires upstream. Keeper = a curated asset if one exists, else the earliest by
    # mtime; distinct renders (different seed → different bytes → different sha) are
    # untouched. Idempotent (a prior-swept dup stays deleted) and recoverable like any
    # soft-delete. This is the last line of defense that makes duplicate clutter
    # impossible-by-construction, independent of dispatch/retry correctness. ──
    from collections import defaultdict
    _CURATED = {"approved", "published", "queued", "flagged"}
    _by_sha: dict = defaultdict(list)
    for a in assets:
        if a.get("sha") and a["status"] != "deleted":
            _by_sha[a["sha"]].append(a)
    deduped = 0
    for group in _by_sha.values():
        if len(group) < 2:
            continue
        group.sort(key=lambda a: (a["status"] not in _CURATED, a.get("mtime", 0)))
        for dup in group[1:]:
            dup["status"] = "deleted"
            dup["dedup"] = True
            deduped += 1

    by_status: dict[str, int] = {}
    by_kind: dict[str, int] = {}
    for a in assets:
        by_status[a["status"]] = by_status.get(a["status"], 0) + 1
        by_kind[a["kind"]] = by_kind.get(a["kind"], 0) + 1

    return {
        "_meta": {
            "note": "Canonical Karma asset library. ONE index the storefront + "
                    "socials queue read. Regenerated by library_manifest.py; edit "
                    "status/caption/tags with karma-queue.py (they are preserved).",
            "prefix": "orgs/karma/output",
            "generatedAt": iso(time.time()),
            "count": len(assets),
            "byStatus": by_status,
            "byKind": by_kind,
        },
        "assets": assets,
    }


def write_atomic(root: str, doc: dict) -> str:
    # A UNIQUE tmp per writer: a fixed ".library.json.tmp" is shared by every
    # build-manifest sidecar, so two pods overlapping during a roll interleave into
    # one inode -> a torn library.json -> load_prior()/reader gets JSONDecodeError
    # -> {} -> every asset resets to status="draft" (the curation wipe). mkstemp in
    # the SAME dir keeps the rename atomic and same-filesystem.
    manifest_path = os.path.join(root, MANIFEST_NAME)
    fd, tmp = tempfile.mkstemp(prefix=".library.json.", suffix=".tmp", dir=root)
    try:
        with os.fdopen(fd, "w") as fh:
            json.dump(doc, fh, indent=2, ensure_ascii=False)
            fh.write("\n")
        os.replace(tmp, manifest_path)
    except BaseException:
        try:
            os.unlink(tmp)
        except OSError:
            pass
        raise
    return manifest_path


def run_once(root: str, quiet: bool = False, strict: bool = True) -> dict | None:
    if not os.path.isdir(root):
        msg = f"library_manifest: root does not exist yet: {root}"
        if strict:
            sys.exit(msg)
        if not quiet:
            print(msg, file=sys.stderr)
        return None  # watch mode: tolerate a tenant with no renders yet
    # Hold the lock across build (which reads the prior manifest to preserve
    # status/caption/tags) AND the write, so a concurrent app-side status change
    # can't be read-then-clobbered.
    with library_lock(root):
        doc = build(root)
        path = write_atomic(root, doc)
    if not quiet:
        m = doc["_meta"]
        print(f"library_manifest: {m['count']} assets -> {path}  {m['byStatus']}")
    return doc


def main() -> None:
    ap = argparse.ArgumentParser(description="Regenerate the Karma library.json manifest.")
    ap.add_argument("--root", default=os.environ.get(
        "KARMA_LIBRARY_ROOT", "/app/orgs/karma/output"),
        help="Library root (contains designs/ and marketing/).")
    ap.add_argument("--watch", action="store_true", help="Loop forever.")
    ap.add_argument("--interval", type=int, default=30, help="--watch seconds.")
    args = ap.parse_args()

    if not args.watch:
        run_once(args.root)
        return

    print(f"library_manifest: watching {args.root} every {args.interval}s", flush=True)
    while True:
        try:
            run_once(args.root, quiet=True, strict=False)
        except Exception as e:  # a bad cycle must never kill the sidecar
            print(f"library_manifest: cycle error: {e}", file=sys.stderr, flush=True)
        time.sleep(args.interval)


if __name__ == "__main__":
    main()