| """Startup example acquisition/localization and attribution rendering.""" |
| from __future__ import annotations |
|
|
| import shutil |
| from pathlib import Path |
|
|
| from huggingface_hub import hf_hub_download |
|
|
| from space_config import EXAMPLE_SPECS |
| from . import runtime_utils |
| from .app_runtime_support import APP_LOGGER, APP_ROOT, HF_TOKEN, RUNTIME_EXAMPLE_DIR, sha256_file |
|
|
|
|
| def prepare_examples() -> tuple[list[list[str | None]], dict]: |
| rows = [] |
| items = [] |
| hub_cache = {} |
| localized_cache = {} |
| shutil.rmtree(RUNTIME_EXAMPLE_DIR, ignore_errors=True) |
| RUNTIME_EXAMPLE_DIR.mkdir(parents=True, exist_ok=True) |
|
|
| def localize(source_path: str, source_name: str) -> tuple[str, str]: |
| source_path = str(source_path) |
| if source_path in localized_cache: |
| return localized_cache[source_path] |
| digest = sha256_file(source_path) |
| suffix = Path(source_name).suffix.lower() or Path(source_path).suffix.lower() or ".bin" |
| target = RUNTIME_EXAMPLE_DIR / f"{digest[:16]}{suffix}" |
| if not target.exists(): |
| shutil.copy2(source_path, target) |
| result = (str(target), digest) |
| localized_cache[source_path] = result |
| return result |
|
|
| for raw in EXAMPLE_SPECS or []: |
| if not isinstance(raw, dict): |
| continue |
| label = str(raw.get("label") or "Example").strip() |
| repo_id = str(raw.get("repo_id") or "").strip() |
| repo_type = str(raw.get("repo_type") or "model").strip().lower() |
| revision = str(raw.get("revision") or "main").strip() or "main" |
| start_name = str(raw.get("start_image") or "").strip() or None |
| end_name = str(raw.get("end_image") or "").strip() or None |
| prompt = str(raw.get("prompt") or "").strip() |
| if not repo_id or not start_name or not prompt or repo_type not in {"model", "dataset", "space"}: |
| APP_LOGGER.info("[EXAMPLES] skipping invalid spec label=%r", label) |
| continue |
| try: |
| def get_file(filename: str | None): |
| if not filename: |
| return None |
| key = (repo_id, repo_type, revision, filename) |
| if key not in hub_cache: |
| hub_cache[key] = hf_hub_download( |
| repo_id=repo_id, |
| repo_type=None if repo_type == "model" else repo_type, |
| revision=revision, |
| filename=filename, |
| token=HF_TOKEN, |
| ) |
| return hub_cache[key] |
|
|
| start_source = get_file(start_name) |
| end_source = get_file(end_name) |
| resolved = runtime_utils.resolved_revision_from_hub_path(start_source) or revision |
| start_path, start_sha = localize(start_source, start_name) |
| if end_source: |
| end_path, end_sha = localize(end_source, end_name) |
| else: |
| end_path, end_sha = None, None |
| rows.append([start_path, end_path, prompt]) |
| items.append({ |
| "label": label, |
| "repo_id": repo_id, |
| "repo_type": repo_type, |
| "requested_revision": revision, |
| "resolved_revision": resolved, |
| "start_image": start_name, |
| "start_sha256": start_sha, |
| "end_image": end_name, |
| "end_sha256": end_sha, |
| "served_from": str(RUNTIME_EXAMPLE_DIR.relative_to(APP_ROOT)), |
| "license": str(raw.get("license") or "").strip() or None, |
| "attribution": str(raw.get("attribution") or "").strip() or None, |
| }) |
| APP_LOGGER.info( |
| "[EXAMPLES] ready label=%r source=%s:%s@%s served_from=%s", |
| label, repo_type, repo_id, resolved, RUNTIME_EXAMPLE_DIR, |
| ) |
| except Exception as exc: |
| APP_LOGGER.info("[EXAMPLES] FAILED label=%r: %s: %s", label, type(exc).__name__, exc) |
| return rows, {"status": "ready", "count": len(rows), "items": items} |
|
|
|
|
| def attribution_markdown(state: dict) -> str: |
| items = state.get("items") or [] |
| if not items: |
| return "Examples are unavailable in this startup. Generation is unaffected." |
| groups = [] |
| seen = set() |
| for item in items: |
| key = (item.get("repo_type"), item.get("repo_id"), item.get("license"), item.get("attribution")) |
| if key in seen: |
| continue |
| seen.add(key) |
| repo_type, repo_id, license_id, attribution = key |
| text = f"`{repo_type}:{repo_id}`" |
| if attribution: |
| text += f" · {attribution}" |
| if license_id: |
| text += f" · {license_id}" |
| groups.append(text) |
| return "Startup-fetched example assets: " + " | ".join(groups) |
|
|
