| """Project-owned media catalog for the H3 Asset Carousel node. |
| |
| The catalog lives below ComfyUI's input directory so ordinary loader nodes can |
| open every imported file without this pack. A second, content-identical copy |
| is retained below the matching H3 chain output directory for recovery. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import copy |
| import hashlib |
| import json |
| import mimetypes |
| import os |
| import re |
| import shutil |
| import subprocess |
| import uuid |
| from datetime import datetime, timezone |
| from fractions import Fraction |
| from typing import Any |
|
|
| if __package__: |
| from .audio_track_contract import audio_track_bindings |
| else: |
| from audio_track_contract import audio_track_bindings |
|
|
| try: |
| import av |
| except ImportError: |
| av = None |
|
|
| try: |
| from PIL import Image, ImageOps |
| except ImportError: |
| Image = ImageOps = None |
|
|
|
|
| PROJECT_ASSET_FORMAT = "h3_project_assets_v1" |
| PROJECT_ASSET_VERSION = 1 |
| PROJECT_ASSET_ROLES = ( |
| "picture", "semantic_anchor", "video", "motion", |
| "audio_reference", "source_track", |
| ) |
| IMAGE_EXTENSIONS = frozenset(( |
| ".avif", ".bmp", ".gif", ".jpeg", ".jpg", ".png", ".tif", |
| ".tiff", ".webp", |
| )) |
| VIDEO_EXTENSIONS = frozenset(( |
| ".avi", ".m4v", ".mkv", ".mov", ".mp4", ".mpeg", ".mpg", |
| ".webm", |
| )) |
| AUDIO_EXTENSIONS = frozenset(( |
| ".aac", ".flac", ".m4a", ".mp3", ".oga", ".ogg", ".opus", |
| ".wav", |
| )) |
| ALL_MEDIA_EXTENSIONS = IMAGE_EXTENSIONS | VIDEO_EXTENSIONS | AUDIO_EXTENSIONS |
| MAX_CATALOG_ASSETS = 512 |
| MAX_REFERENCE_SLOTS = 512 |
| MAX_ASSET_FOLDERS = 128 |
| MAX_ASSET_LYRICS_CHARACTERS = 100_000 |
| MAX_DERIVED_IMAGE_PIXELS = 268_435_456 |
| MAX_INPUT_RESULTS = 2000 |
| INPUT_BROWSER_EXCLUDED_DIRECTORIES = frozenset(( |
| "clipspace", "h3_projects", |
| )) |
| _TAG_RE = re.compile(r"[A-Za-z][A-Za-z0-9_-]{0,63}") |
|
|
|
|
| def _utc_now() -> str: |
| return datetime.now(timezone.utc).isoformat( |
| timespec="seconds").replace("+00:00", "Z") |
|
|
|
|
| def _safe_name(value: Any, fallback: str = "asset", limit: int = 128) -> str: |
| text = re.sub(r"[^A-Za-z0-9._-]+", "_", str(value or "").strip()) |
| text = text.strip("._-") |
| return (text or fallback)[:limit] |
|
|
|
|
| def _safe_project(value: Any) -> str: |
| project = _safe_name(value, "", 96) |
| if not project: |
| raise ValueError("Run name cannot be blank.") |
| return project |
|
|
|
|
| def _safe_tag(value: Any, fallback: str = "asset") -> str: |
| tag = str(value or "").strip().lstrip("@#") |
| if not tag: |
| tag = _safe_name(fallback, "asset", 64) |
| if not tag[:1].isalpha(): |
| tag = "asset_" + tag |
| if _TAG_RE.fullmatch(tag) is None: |
| tag = re.sub(r"[^A-Za-z0-9_-]+", "_", tag).strip("_-")[:64] |
| if not tag or not tag[0].isalpha(): |
| tag = "asset_" + (tag or uuid.uuid4().hex[:8]) |
| tag = tag[:64] |
| if _TAG_RE.fullmatch(tag) is None: |
| raise ValueError("Asset tag must begin with a letter and contain only letters, numbers, _ or -.") |
| return tag |
|
|
|
|
| def _safe_folder_name(value: Any) -> str: |
| name = re.sub(r"[\x00-\x1f\x7f]+", " ", str(value or "")).strip() |
| name = re.sub(r"\s+", " ", name) |
| if not name: |
| raise ValueError("Asset folder name cannot be blank.") |
| return name[:80] |
|
|
|
|
| def _unique_tag(catalog: dict[str, Any], value: Any, fallback: str, |
| *, exclude_id: str = "") -> str: |
| tag = _safe_tag(value, fallback) |
| used = { |
| str(item.get("tag") or "") for item in catalog.get("assets", []) |
| if str(item.get("id") or "") != exclude_id |
| and bool(item.get("enabled", True)) |
| } |
| if tag not in used: |
| return tag |
| base = tag |
| ordinal = 2 |
| while tag in used: |
| suffix = "_%d" % ordinal |
| tag = base[:64 - len(suffix)] + suffix |
| ordinal += 1 |
| return tag |
|
|
|
|
| def _image_resampling(value: Any): |
| if Image is None: |
| raise RuntimeError("Pillow is required for project image variants.") |
| name = str(value or "lanczos").strip().lower() |
| modes = { |
| "nearest": Image.Resampling.NEAREST, |
| "box": Image.Resampling.BOX, |
| "bilinear": Image.Resampling.BILINEAR, |
| "hamming": Image.Resampling.HAMMING, |
| "bicubic": Image.Resampling.BICUBIC, |
| "lanczos": Image.Resampling.LANCZOS, |
| } |
| if name not in modes: |
| raise ValueError("Image resampling must be nearest, box, bilinear, hamming, bicubic, or lanczos.") |
| return name, modes[name] |
|
|
|
|
| def _image_operation_geometry(image: Any, crop: Any, target: Any) -> tuple[int, int, int, int, int, int]: |
| if not isinstance(crop, dict): |
| raise ValueError("Image crop must be a JSON object.") |
| if not isinstance(target, dict): |
| raise ValueError("Image target size must be a JSON object.") |
| try: |
| x = int(crop.get("x", 0)) |
| y = int(crop.get("y", 0)) |
| width = int(crop.get("width", image.width)) |
| height = int(crop.get("height", image.height)) |
| target_width = int(target.get("width", width)) |
| target_height = int(target.get("height", height)) |
| except (TypeError, ValueError) as exc: |
| raise ValueError("Crop and target dimensions must be whole pixels.") from exc |
| if x < 0 or y < 0 or width < 1 or height < 1: |
| raise ValueError("Crop dimensions must describe a positive area inside the source image.") |
| if x + width > image.width or y + height > image.height: |
| raise ValueError( |
| "Crop %d,%d %dx%d exceeds the oriented source image %dx%d." % |
| (x, y, width, height, image.width, image.height)) |
| if target_width < 1 or target_height < 1: |
| raise ValueError("Target width and height must be positive.") |
| if target_width * target_height > MAX_DERIVED_IMAGE_PIXELS: |
| raise ValueError("Derived image target exceeds %d pixels." % MAX_DERIVED_IMAGE_PIXELS) |
| return x, y, width, height, target_width, target_height |
|
|
|
|
| def _inside(root: str, path: str) -> bool: |
| try: |
| return os.path.commonpath( |
| (os.path.realpath(root), os.path.realpath(path))) == os.path.realpath(root) |
| except ValueError: |
| return False |
|
|
|
|
| def _atomic_json(path: str, value: Any) -> None: |
| os.makedirs(os.path.dirname(path), exist_ok=True) |
| temporary = "%s.%s.tmp" % (path, uuid.uuid4().hex) |
| try: |
| with open(temporary, "w", encoding="utf-8") as handle: |
| json.dump(value, handle, ensure_ascii=False, indent=2, sort_keys=True) |
| handle.write("\n") |
| os.replace(temporary, path) |
| finally: |
| try: |
| os.unlink(temporary) |
| except FileNotFoundError: |
| pass |
|
|
|
|
| def _file_sha256(path: str) -> str: |
| digest = hashlib.sha256() |
| with open(path, "rb") as handle: |
| while True: |
| chunk = handle.read(1024 * 1024) |
| if not chunk: |
| break |
| digest.update(chunk) |
| return digest.hexdigest() |
|
|
|
|
| def _canonical_fingerprint(value: Any) -> str: |
| payload = json.dumps( |
| value, ensure_ascii=False, sort_keys=True, |
| separators=(",", ":")).encode("utf-8") |
| return hashlib.sha256(payload).hexdigest() |
|
|
|
|
| def _media_kind(path: str) -> str: |
| suffix = os.path.splitext(path)[1].lower() |
| if suffix in IMAGE_EXTENSIONS: |
| return "image" |
| if suffix in VIDEO_EXTENSIONS: |
| return "video" |
| if suffix in AUDIO_EXTENSIONS: |
| return "audio" |
| raise ValueError("Unsupported media extension %r." % suffix) |
|
|
|
|
| def _role_for_kind(kind: str, role: Any) -> str: |
| role = str(role or "").strip().lower() |
| defaults = {"image": "picture", "video": "video", "audio": "audio_reference"} |
| role = role or defaults[kind] |
| allowed = { |
| "image": {"picture", "semantic_anchor"}, |
| "video": {"video", "motion", "source_track"}, |
| "audio": {"audio_reference", "source_track"}, |
| } |
| if role not in allowed[kind]: |
| raise ValueError("Role %r is not valid for a %s asset." % (role, kind)) |
| return role |
|
|
|
|
| def _stream_duration(stream: Any, container: Any) -> float: |
| if stream.duration is not None and stream.time_base is not None: |
| return max(0.0, float(stream.duration * stream.time_base)) |
| if container.duration is not None: |
| return max(0.0, float(container.duration) / 1_000_000.0) |
| return 0.0 |
|
|
|
|
| def _fraction_parts(value: Any, fallback: int = 24) -> tuple[int, int]: |
| try: |
| rate = Fraction(value) |
| except (TypeError, ValueError, ZeroDivisionError): |
| rate = Fraction(fallback, 1) |
| if rate <= 0: |
| rate = Fraction(fallback, 1) |
| return int(rate.numerator), int(rate.denominator) |
|
|
|
|
| def _probe_media(path: str, kind: str) -> dict[str, Any]: |
| if kind == "image": |
| if Image is None: |
| return {} |
| with Image.open(path) as image: |
| return { |
| "width": int(image.width), |
| "height": int(image.height), |
| "format": str(image.format or "").lower(), |
| "frames": int(getattr(image, "n_frames", 1)), |
| } |
| if av is None: |
| return {} |
| with av.open(path, mode="r") as container: |
| if kind == "video": |
| videos = list(container.streams.video) |
| if not videos: |
| raise ValueError("Video asset contains no video stream.") |
| video = videos[0] |
| rate = video.average_rate or video.guessed_rate or video.base_rate |
| numerator, denominator = _fraction_parts(rate) |
| duration = _stream_duration(video, container) |
| frames = int(video.frames or 0) |
| if frames <= 0 and duration > 0: |
| frames = int(round(duration * numerator / denominator)) |
| audio_metadata = {} |
| audios = list(container.streams.audio) |
| if audios: |
| audio = audios[0] |
| audio_metadata = { |
| "audio_sample_rate": int( |
| audio.codec_context.sample_rate or audio.rate or 0), |
| "audio_channels": len(audio.codec_context.layout.channels), |
| "audio_start_seconds": float( |
| audio.start_time * audio.time_base) |
| if audio.start_time is not None |
| and audio.time_base is not None else 0.0, |
| } |
| return { |
| "width": int(video.codec_context.width or video.width or 0), |
| "height": int(video.codec_context.height or video.height or 0), |
| "duration_seconds": duration, |
| "frame_count": frames, |
| "source_rate_numerator": numerator, |
| "source_rate_denominator": denominator, |
| "fps": numerator / float(denominator), |
| "codec": str(video.codec_context.name or ""), |
| "video_start_seconds": float( |
| video.start_time * video.time_base) |
| if video.start_time is not None and video.time_base is not None |
| else 0.0, |
| "has_audio": bool(audios), |
| **audio_metadata, |
| } |
| audios = list(container.streams.audio) |
| if not audios: |
| raise ValueError("Audio asset contains no audio stream.") |
| audio = audios[0] |
| sample_rate = int(audio.codec_context.sample_rate or audio.rate or 0) |
| channels = len(audio.codec_context.layout.channels) |
| return { |
| "duration_seconds": _stream_duration(audio, container), |
| "sample_rate": sample_rate, |
| "channels": channels, |
| "codec": str(audio.codec_context.name or ""), |
| } |
|
|
|
|
| def _entry_contract(entry: dict[str, Any]) -> dict[str, Any]: |
| options = entry.get("options") if isinstance(entry.get("options"), dict) else {} |
| return { |
| "id": str(entry.get("id") or ""), |
| "kind": str(entry.get("kind") or ""), |
| "role": str(entry.get("role") or ""), |
| "tag": str(entry.get("tag") or ""), |
| "sha256": str(entry.get("sha256") or ""), |
| "enabled": bool(entry.get("enabled", True)), |
| "options": options, |
| } |
|
|
|
|
| def _slot_contract(slot: dict[str, Any]) -> dict[str, Any]: |
| options = slot.get("options") if isinstance(slot.get("options"), dict) else {} |
| return { |
| "id": str(slot.get("id") or ""), |
| "kind": str(slot.get("kind") or ""), |
| "role": str(slot.get("role") or ""), |
| "tag": str(slot.get("tag") or ""), |
| "content_hash": str(slot.get("content_hash") or ""), |
| "available": bool(slot.get("available", True)), |
| "options": options, |
| } |
|
|
|
|
| class ProjectAssetStore: |
| """Own project media below input/h3_projects and mirror it to a run.""" |
|
|
| def __init__(self, input_root: str, output_root: str): |
| self.input_root = os.path.realpath(os.path.abspath(input_root)) |
| self.output_root = os.path.realpath(os.path.abspath(output_root)) |
| self.projects_root = os.path.join(self.input_root, "h3_projects") |
| self.chains_root = os.path.join(self.output_root, "h3_chains") |
|
|
| def _project_dir(self, project: Any) -> tuple[str, str]: |
| name = _safe_project(project) |
| path = os.path.realpath(os.path.join(self.projects_root, name)) |
| if not _inside(self.input_root, path): |
| raise ValueError("Project asset path escapes the ComfyUI input directory.") |
| return path, name |
|
|
| def _backup_dir(self, project: Any) -> tuple[str, str]: |
| name = _safe_project(project) |
| path = os.path.realpath(os.path.join( |
| self.chains_root, name, "project_assets")) |
| if not _inside(self.output_root, path): |
| raise ValueError("Project backup path escapes the ComfyUI output directory.") |
| return path, name |
|
|
| @staticmethod |
| def _empty_catalog(project: str) -> dict[str, Any]: |
| return { |
| "format": PROJECT_ASSET_FORMAT, |
| "version": PROJECT_ASSET_VERSION, |
| "project": project, |
| "updated_at": _utc_now(), |
| "revision": "", |
| "assets": [], |
| "reference_slots": [], |
| "folders": [], |
| } |
|
|
| def load(self, project: Any, *, create: bool = False) -> dict[str, Any]: |
| directory, name = self._project_dir(project) |
| path = os.path.join(directory, "catalog.json") |
| if not os.path.isfile(path): |
| catalog = self._empty_catalog(name) |
| if create: |
| return self._save_catalog(catalog) |
| return catalog |
| with open(path, "r", encoding="utf-8") as handle: |
| catalog = json.load(handle) |
| if (not isinstance(catalog, dict) |
| or catalog.get("format") != PROJECT_ASSET_FORMAT |
| or int(catalog.get("version", -1)) != PROJECT_ASSET_VERSION |
| or not isinstance(catalog.get("assets"), list)): |
| raise ValueError("Project asset catalog has an unsupported format.") |
| catalog["project"] = name |
| slots = catalog.get("reference_slots") |
| catalog["reference_slots"] = ( |
| slots if isinstance(slots, list) else []) |
| folders = catalog.get("folders") |
| catalog["folders"] = folders if isinstance(folders, list) else [] |
| return catalog |
|
|
| def _save_catalog(self, catalog: dict[str, Any]) -> dict[str, Any]: |
| directory, name = self._project_dir(catalog.get("project")) |
| assets = [dict(item) for item in catalog.get("assets", []) |
| if isinstance(item, dict)] |
| slots = [dict(item) for item in catalog.get("reference_slots", []) |
| if isinstance(item, dict)] |
| folders = [dict(item) for item in catalog.get("folders", []) |
| if isinstance(item, dict)] |
| if len(assets) > MAX_CATALOG_ASSETS: |
| raise ValueError("Project asset catalog supports at most %d assets." % MAX_CATALOG_ASSETS) |
| if len(slots) > MAX_REFERENCE_SLOTS: |
| raise ValueError( |
| "Project asset catalog supports at most %d unresolved " |
| "reference slots." % MAX_REFERENCE_SLOTS) |
| if len(folders) > MAX_ASSET_FOLDERS: |
| raise ValueError( |
| "Project asset catalog supports at most %d folders." % |
| MAX_ASSET_FOLDERS) |
| folder_ids = [str(item.get("id") or "") for item in folders] |
| if (any(not item for item in folder_ids) |
| or len(folder_ids) != len(set(folder_ids))): |
| raise ValueError("Every asset folder must have a unique ID.") |
| known_folders = set(folder_ids) |
| for asset in assets: |
| options = asset.get("options") or {} |
| if options.get("audio_tracks") is not None: |
| if asset.get("kind") != "audio": |
| raise ValueError("Only audio assets can group audio tracks.") |
| asset["options"] = { |
| **options, "audio_tracks": audio_track_bindings( |
| options["audio_tracks"], assets), |
| } |
| folder_id = str(asset.get("folder_id") or "") |
| if folder_id and folder_id not in known_folders: |
| asset["folder_id"] = "" |
| document = { |
| "format": PROJECT_ASSET_FORMAT, |
| "version": PROJECT_ASSET_VERSION, |
| "project": name, |
| "updated_at": _utc_now(), |
| "assets": assets, |
| "reference_slots": slots, |
| "folders": folders, |
| } |
| document["revision"] = _canonical_fingerprint({ |
| "assets": [_entry_contract(item) for item in assets], |
| "reference_slots": [_slot_contract(item) for item in slots], |
| }) |
| for group in ("images", "videos", "audio", "previews", ".uploads"): |
| os.makedirs(os.path.join(directory, group), exist_ok=True) |
| _atomic_json(os.path.join(directory, "catalog.json"), document) |
| backup, _name = self._backup_dir(name) |
| os.makedirs(backup, exist_ok=True) |
| _atomic_json(os.path.join(backup, "catalog.json"), document) |
| return document |
|
|
| def public_catalog(self, project: Any) -> dict[str, Any]: |
| catalog = self.load(project, create=True) |
| return { |
| **catalog, |
| "assets": [dict(item) for item in catalog["assets"]], |
| "reference_slots": [ |
| dict(item) for item in catalog.get("reference_slots", [])], |
| "folders": [dict(item) for item in catalog.get("folders", [])], |
| } |
|
|
| def duplicate_project( |
| self, project: Any, new_project: Any) -> dict[str, Any]: |
| """Clone one complete asset project without copying generated output. |
| |
| Only catalog-referenced media below ``input/h3_projects`` and its |
| ``project_assets`` recovery mirror are copied. The source run's |
| checkpoints, generated clips, assembled videos, and other output |
| files are never traversed. |
| """ |
| source_directory, source_name = self._project_dir(project) |
| target_directory, target_name = self._project_dir(new_project) |
| if source_name == target_name: |
| raise ValueError( |
| "The duplicated asset project needs a different Run name.") |
| source_catalog_path = os.path.join(source_directory, "catalog.json") |
| if not os.path.isfile(source_catalog_path): |
| raise FileNotFoundError( |
| "Asset project %s does not exist." % source_name) |
|
|
| source_catalog = self.load(source_name) |
| media: dict[str, tuple[str, str, int]] = {} |
| for entry in source_catalog.get("assets", []): |
| source_path = self._asset_path(source_name, entry) |
| relative = str(entry.get("relative_path") or "").replace( |
| "/", os.sep) |
| normalized = os.path.normpath(relative) |
| if (not relative or normalized in ("", ".") |
| or os.path.isabs(normalized) |
| or normalized == os.pardir |
| or normalized.startswith(os.pardir + os.sep)): |
| raise ValueError( |
| "Project asset media path escapes its store: %s" % |
| relative) |
| digest = _file_sha256(source_path) |
| expected = str(entry.get("sha256") or "") |
| if expected and digest != expected: |
| raise ValueError( |
| "Project asset %s no longer matches its catalog SHA-256." |
| % entry.get("tag", entry.get("id", "asset"))) |
| existing = media.get(normalized) |
| if existing is not None and existing[1] != digest: |
| raise ValueError( |
| "Project asset catalog maps one media path to different " |
| "content.") |
| media[normalized] = ( |
| source_path, digest, int(os.path.getsize(source_path))) |
|
|
| target_backup, _name = self._backup_dir(target_name) |
| target_run_directory = os.path.dirname(target_backup) |
| if os.path.exists(target_directory): |
| raise FileExistsError( |
| "An asset project named %s already exists. Choose a new Run " |
| "name." % target_name) |
| if os.path.exists(target_run_directory): |
| raise FileExistsError( |
| "An H3 output Run named %s already exists. Choose a new Run " |
| "name so its renders and checkpoints remain separate." % |
| target_name) |
|
|
| os.makedirs(self.projects_root, exist_ok=True) |
| os.makedirs(self.chains_root, exist_ok=True) |
| created_project = False |
| created_run = False |
| try: |
| os.mkdir(target_directory) |
| created_project = True |
| os.mkdir(target_run_directory) |
| created_run = True |
| os.makedirs(target_backup) |
| for group in ("images", "videos", "audio", "previews", ".uploads"): |
| os.makedirs(os.path.join(target_directory, group)) |
| for group in ("images", "videos", "audio"): |
| os.makedirs(os.path.join(target_backup, group)) |
|
|
| for relative, (source_path, digest, _size) in media.items(): |
| for root in (target_directory, target_backup): |
| destination = os.path.realpath(os.path.join(root, relative)) |
| if not _inside(root, destination): |
| raise ValueError( |
| "Project asset media path escapes its duplicate.") |
| os.makedirs(os.path.dirname(destination), exist_ok=True) |
| shutil.copy2(source_path, destination) |
| if _file_sha256(destination) != digest: |
| raise ValueError( |
| "Duplicated project asset failed its SHA-256 check.") |
|
|
| now = _utc_now() |
| cloned_catalog = copy.deepcopy(source_catalog) |
| cloned_catalog["project"] = target_name |
| cloned_catalog["updated_at"] = now |
| for entry in cloned_catalog.get("assets", []): |
| relative = str(entry.get("relative_path") or "") |
| entry["input_path"] = os.path.join( |
| "h3_projects", target_name, relative).replace(os.sep, "/") |
| entry["updated_at"] = now |
| saved = self._save_catalog(cloned_catalog) |
| return { |
| "catalog": saved, |
| "source_project": source_name, |
| "target_project": target_name, |
| "asset_count": len(saved.get("assets", [])), |
| "media_file_count": len(media), |
| "copied_bytes": sum(item[2] for item in media.values()), |
| "render_outputs_copied": False, |
| } |
| except Exception: |
| if created_run: |
| shutil.rmtree(target_run_directory, ignore_errors=True) |
| if created_project: |
| shutil.rmtree(target_directory, ignore_errors=True) |
| raise |
|
|
| def _asset_path(self, project: Any, entry: dict[str, Any]) -> str: |
| directory, _name = self._project_dir(project) |
| relative = str(entry.get("relative_path") or "") |
| path = os.path.realpath(os.path.join(directory, relative)) |
| if not relative or not _inside(directory, path) or not os.path.isfile(path): |
| raise FileNotFoundError("Project asset file is missing: %s" % relative) |
| return path |
|
|
| def asset(self, project: Any, asset_id: Any) -> tuple[dict[str, Any], str]: |
| catalog = self.load(project) |
| wanted = str(asset_id or "") |
| entry = next((item for item in catalog["assets"] |
| if str(item.get("id") or "") == wanted), None) |
| if entry is None: |
| raise FileNotFoundError("Project asset %s was not found." % wanted) |
| return dict(entry), self._asset_path(project, entry) |
|
|
| def upload_path(self, project: Any, filename: Any) -> str: |
| directory, _name = self._project_dir(project) |
| os.makedirs(os.path.join(directory, ".uploads"), exist_ok=True) |
| basename = _safe_name(os.path.basename(str(filename or "asset")), "asset") |
| suffix = os.path.splitext(basename)[1].lower() |
| if suffix not in ALL_MEDIA_EXTENSIONS: |
| raise ValueError("Unsupported upload extension %r." % suffix) |
| return os.path.join(directory, ".uploads", "%s_%s" % ( |
| uuid.uuid4().hex, basename)) |
|
|
| def sync_reference_slots( |
| self, project: Any, templates: Any) -> dict[str, Any]: |
| """Mirror reference metadata without copying or serializing media.""" |
| if not isinstance(templates, list): |
| raise ValueError("Reference templates must be a list.") |
| catalog = self.load(project, create=True) |
| resolved_tags = { |
| str(item.get("tag") or "") for item in catalog["assets"]} |
| current = { |
| str(item.get("tag") or ""): dict(item) |
| for item in catalog.get("reference_slots", []) |
| if isinstance(item, dict) and str(item.get("tag") or "") |
| } |
| incoming_tags = set() |
| slots = [] |
| now = _utc_now() |
| for raw in templates: |
| if not isinstance(raw, dict): |
| raise ValueError("Every reference template must be an object.") |
| kind = str(raw.get("kind") or "").strip().lower() |
| if kind not in ("image", "video", "audio"): |
| raise ValueError( |
| "Reference template kind must be image, video, or audio.") |
| tag = _safe_tag(raw.get("tag"), "%s_reference" % kind) |
| if tag in incoming_tags: |
| raise ValueError("Reference template @%s is duplicated." % tag) |
| incoming_tags.add(tag) |
| if tag in resolved_tags: |
| |
| |
| continue |
| role = _role_for_kind(kind, raw.get("role")) |
| options = raw.get("options") |
| options = dict(options) if isinstance(options, dict) else {} |
| previous = current.get(tag) |
| created_at = str( |
| (previous or {}).get("created_at") or now) |
| slot = { |
| "id": str((previous or {}).get("id") or |
| ("slot_" + uuid.uuid4().hex)), |
| "kind": kind, |
| "role": role, |
| "tag": tag, |
| "content_hash": str(raw.get("content_hash") or ""), |
| "available": True, |
| "source": "tagged_references", |
| "created_at": created_at, |
| "updated_at": now, |
| "options": options, |
| } |
| slots.append(slot) |
| for tag, previous in current.items(): |
| if tag in incoming_tags or tag in resolved_tags: |
| continue |
| stale = dict(previous) |
| stale["available"] = False |
| stale["updated_at"] = now |
| slots.append(stale) |
| old_contract = [ |
| _slot_contract(item) |
| for item in catalog.get("reference_slots", [])] |
| new_contract = [_slot_contract(item) for item in slots] |
| if old_contract == new_contract: |
| return self.public_catalog(project) |
| catalog["reference_slots"] = slots |
| return self._save_catalog(catalog) |
|
|
| def bind_reference_slot( |
| self, project: Any, slot_id: Any, source_path: Any, *, |
| role: Any = "", tag: Any = "", original_name: Any = "", |
| source_kind: str = "path", |
| options: dict[str, Any] | None = None) -> dict[str, Any]: |
| """Bind explicitly selected media to one unresolved reference slot.""" |
| catalog = self.load(project) |
| wanted = str(slot_id or "") |
| slot = next((item for item in catalog.get("reference_slots", []) |
| if str(item.get("id") or "") == wanted), None) |
| if slot is None: |
| raise FileNotFoundError( |
| "Unresolved reference slot %s was not found." % wanted) |
| merged_options = dict(slot.get("options") or {}) |
| if isinstance(options, dict): |
| merged_options.update(options) |
| result = self.import_file( |
| project, source_path, |
| role=role or slot.get("role"), tag=tag or slot.get("tag"), |
| original_name=original_name, source_kind=source_kind, |
| options=merged_options) |
| bound_catalog = result["catalog"] |
| bound_catalog["reference_slots"] = [ |
| item for item in bound_catalog.get("reference_slots", []) |
| if str(item.get("id") or "") != wanted] |
| result["catalog"] = self._save_catalog(bound_catalog) |
| result["bound_slot_id"] = wanted |
| return result |
|
|
| def import_file(self, project: Any, source_path: Any, *, role: Any = "", |
| tag: Any = "", original_name: Any = "", |
| source_kind: str = "path", |
| options: dict[str, Any] | None = None) -> dict[str, Any]: |
| source = os.path.realpath(os.path.abspath(os.path.expanduser( |
| str(source_path or "").strip()))) |
| if not source or not os.path.isfile(source): |
| raise FileNotFoundError("Asset source does not exist: %s" % source) |
| kind = _media_kind(source) |
| role = _role_for_kind(kind, role) |
| basename = os.path.basename(str(original_name or source)) |
| stem = os.path.splitext(basename)[0] |
| tag = _safe_tag(tag, stem) |
| digest = _file_sha256(source) |
| size = int(os.path.getsize(source)) |
| directory, name = self._project_dir(project) |
| catalog = self.load(name, create=True) |
| if role == "source_track" and any( |
| item.get("role") == "source_track" |
| and bool(item.get("enabled", True)) |
| for item in catalog["assets"]): |
| raise ValueError( |
| "This project already has an enabled Source track. Disable " |
| "or reassign it before importing another one.") |
| group = {"image": "images", "video": "videos", "audio": "audio"}[kind] |
| suffix = os.path.splitext(source)[1].lower() |
| filename = "%s_%s%s" % ( |
| digest[:16], _safe_name(stem, kind, 80), suffix) |
| destination = os.path.join(directory, group, filename) |
| os.makedirs(os.path.dirname(destination), exist_ok=True) |
| if not os.path.isfile(destination) or _file_sha256(destination) != digest: |
| temporary = "%s.%s.tmp" % (destination, uuid.uuid4().hex) |
| try: |
| shutil.copy2(source, temporary) |
| if _file_sha256(temporary) != digest: |
| raise ValueError("Imported asset failed its SHA-256 check.") |
| os.replace(temporary, destination) |
| finally: |
| try: |
| os.unlink(temporary) |
| except FileNotFoundError: |
| pass |
| metadata = _probe_media(destination, kind) |
| tag = _unique_tag(catalog, tag, stem) |
| now = _utc_now() |
| entry = { |
| "id": uuid.uuid4().hex, |
| "kind": kind, |
| "role": role, |
| "tag": tag, |
| "enabled": True, |
| "relative_path": os.path.relpath(destination, directory).replace(os.sep, "/"), |
| "input_path": os.path.relpath(destination, self.input_root).replace(os.sep, "/"), |
| "sha256": digest, |
| "size": size, |
| "mime_type": mimetypes.guess_type(destination)[0] or "application/octet-stream", |
| "original_name": basename, |
| "source_kind": str(source_kind or "path")[:40], |
| "created_at": now, |
| "updated_at": now, |
| "metadata": metadata, |
| "options": dict(options or {}), |
| } |
| catalog["assets"].append(entry) |
| catalog = self._save_catalog(catalog) |
| backup, _name = self._backup_dir(name) |
| backup_path = os.path.join(backup, group, filename) |
| os.makedirs(os.path.dirname(backup_path), exist_ok=True) |
| if not os.path.isfile(backup_path) or _file_sha256(backup_path) != digest: |
| temporary = "%s.%s.tmp" % (backup_path, uuid.uuid4().hex) |
| try: |
| shutil.copy2(destination, temporary) |
| os.replace(temporary, backup_path) |
| finally: |
| try: |
| os.unlink(temporary) |
| except FileNotFoundError: |
| pass |
| return {"catalog": catalog, "asset": entry} |
|
|
| def create_folder(self, project: Any, name: Any, *, color: Any = "") -> dict[str, Any]: |
| catalog = self.load(project, create=True) |
| if len(catalog.get("folders", [])) >= MAX_ASSET_FOLDERS: |
| raise ValueError( |
| "Project asset catalog supports at most %d folders." % |
| MAX_ASSET_FOLDERS) |
| folder_name = _safe_folder_name(name) |
| if any(str(item.get("name") or "").casefold() == folder_name.casefold() |
| for item in catalog.get("folders", [])): |
| raise ValueError("An asset folder named %r already exists." % folder_name) |
| now = _utc_now() |
| folder = { |
| "id": "folder_" + uuid.uuid4().hex, |
| "name": folder_name, |
| "color": str(color or "")[:32], |
| "created_at": now, |
| "updated_at": now, |
| } |
| catalog.setdefault("folders", []).append(folder) |
| return {"catalog": self._save_catalog(catalog), "folder": dict(folder)} |
|
|
| def update_folder(self, project: Any, folder_id: Any, |
| changes: Any) -> dict[str, Any]: |
| if not isinstance(changes, dict): |
| raise ValueError("Asset folder changes must be a JSON object.") |
| catalog = self.load(project) |
| wanted = str(folder_id or "") |
| folder = next((item for item in catalog.get("folders", []) |
| if str(item.get("id") or "") == wanted), None) |
| if folder is None: |
| raise FileNotFoundError("Asset folder %s was not found." % wanted) |
| if "name" in changes: |
| name = _safe_folder_name(changes["name"]) |
| if any(str(item.get("name") or "").casefold() == name.casefold() |
| and str(item.get("id") or "") != wanted |
| for item in catalog.get("folders", [])): |
| raise ValueError("An asset folder named %r already exists." % name) |
| folder["name"] = name |
| if "color" in changes: |
| folder["color"] = str(changes["color"] or "")[:32] |
| folder["updated_at"] = _utc_now() |
| return {"catalog": self._save_catalog(catalog), "folder": dict(folder)} |
|
|
| def delete_folder(self, project: Any, folder_id: Any) -> dict[str, Any]: |
| catalog = self.load(project) |
| wanted = str(folder_id or "") |
| folder = next((item for item in catalog.get("folders", []) |
| if str(item.get("id") or "") == wanted), None) |
| if folder is None: |
| raise FileNotFoundError("Asset folder %s was not found." % wanted) |
| catalog["folders"] = [ |
| item for item in catalog.get("folders", []) |
| if str(item.get("id") or "") != wanted] |
| moved = 0 |
| for asset in catalog.get("assets", []): |
| if str(asset.get("folder_id") or "") == wanted: |
| asset["folder_id"] = "" |
| asset["updated_at"] = _utc_now() |
| moved += 1 |
| return { |
| "catalog": self._save_catalog(catalog), |
| "folder": dict(folder), |
| "assets_unfiled": moved, |
| } |
|
|
| def reorder_folders(self, project: Any, folder_ids: Any) -> dict[str, Any]: |
| if not isinstance(folder_ids, list): |
| raise ValueError("Folder order must be a JSON list of folder IDs.") |
| catalog = self.load(project) |
| current = [str(item.get("id") or "") |
| for item in catalog.get("folders", [])] |
| requested = [str(item or "") for item in folder_ids] |
| if (len(requested) != len(set(requested)) |
| or set(requested) != set(current)): |
| raise ValueError( |
| "Folder order must contain every current folder ID exactly once.") |
| by_id = {str(item.get("id") or ""): item |
| for item in catalog.get("folders", [])} |
| catalog["folders"] = [by_id[item] for item in requested] |
| return {"catalog": self._save_catalog(catalog)} |
|
|
| def duplicate(self, project: Any, asset_id: Any, *, tag: Any = "", |
| folder_id: Any = None) -> dict[str, Any]: |
| """Create a second catalog card without copying its media bytes.""" |
| catalog = self.load(project) |
| wanted = str(asset_id or "") |
| source = next((item for item in catalog["assets"] |
| if str(item.get("id") or "") == wanted), None) |
| if source is None: |
| raise FileNotFoundError("Project asset %s was not found." % wanted) |
| target_folder = (str(source.get("folder_id") or "") |
| if folder_id is None else str(folder_id or "")) |
| known_folders = {str(item.get("id") or "") |
| for item in catalog.get("folders", [])} |
| if target_folder and target_folder not in known_folders: |
| raise FileNotFoundError("Asset folder %s was not found." % target_folder) |
| now = _utc_now() |
| clone = dict(source) |
| clone.update({ |
| "id": uuid.uuid4().hex, |
| "tag": _unique_tag( |
| catalog, tag or "%s_copy" % source.get("tag", "asset"), |
| "%s_copy" % source.get("tag", "asset")), |
| "folder_id": target_folder, |
| "parent_asset_id": wanted, |
| "source_kind": "catalog_duplicate", |
| "created_at": now, |
| "updated_at": now, |
| }) |
| if source.get("role") == "source_track": |
| clone["enabled"] = False |
| clone["options"] = dict(source.get("options") or {}) |
| if clone["options"].get("audio_tracks") is not None: |
| clone["options"]["audio_tracks"] = { |
| role: clone["id"] if value == wanted else value |
| for role, value in clone["options"]["audio_tracks"].items()} |
| clone["metadata"] = dict(source.get("metadata") or {}) |
| catalog["assets"].append(clone) |
| return {"catalog": self._save_catalog(catalog), "asset": dict(clone)} |
|
|
| def operation_asset(self, project: Any, operation_id: Any) -> dict[str, Any] | None: |
| wanted = str(operation_id or "")[:128] |
| if not wanted: |
| return None |
| catalog = self.load(project) |
| existing = next((item for item in catalog["assets"] |
| if str((item.get("transform") or {}).get( |
| "operation_id") or "") == wanted), None) |
| if existing is None: |
| return None |
| return {"catalog": catalog, "asset": dict(existing), "reused": True} |
|
|
| def prepare_image_crop(self, project: Any, asset_id: Any, |
| crop: Any, target: Any) -> tuple[dict[str, Any], Any, dict[str, int]]: |
| if Image is None or ImageOps is None: |
| raise RuntimeError("Pillow is required for project image variants.") |
| entry, path = self.asset(project, asset_id) |
| if entry.get("kind") != "image": |
| raise ValueError("Only image assets can create cropped variants.") |
| with Image.open(path) as opened: |
| oriented = ImageOps.exif_transpose(opened).copy() |
| geometry = _image_operation_geometry(oriented, crop, target) |
| x, y, width, height, target_width, target_height = geometry |
| cropped = oriented.crop((x, y, x + width, y + height)) |
| values = { |
| "x": x, "y": y, "width": width, "height": height, |
| "target_width": target_width, "target_height": target_height, |
| "source_width": int(oriented.width), |
| "source_height": int(oriented.height), |
| } |
| return entry, cropped, values |
|
|
| def register_derived_image( |
| self, project: Any, parent_asset_id: Any, rendered_path: Any, *, |
| tag: Any = "", folder_id: Any = None, |
| transform: dict[str, Any] | None = None, |
| operation_id: Any = "") -> dict[str, Any]: |
| parent, _path = self.asset(project, parent_asset_id) |
| if parent.get("kind") != "image": |
| raise ValueError("Only image assets can own image variants.") |
| catalog = self.load(project) |
| operation_id = str(operation_id or "")[:128] |
| if operation_id: |
| existing = next((item for item in catalog["assets"] |
| if str((item.get("transform") or {}).get( |
| "operation_id") or "") == operation_id), None) |
| if existing is not None: |
| return {"catalog": catalog, "asset": dict(existing), "reused": True} |
| target_folder = (str(parent.get("folder_id") or "") |
| if folder_id is None else str(folder_id or "")) |
| result = self.import_file( |
| project, rendered_path, |
| role=parent.get("role"), |
| tag=tag or "%s_variant" % parent.get("tag", "asset"), |
| original_name="%s_variant.png" % parent.get("tag", "asset"), |
| source_kind="derived_image", |
| options=dict(parent.get("options") or {})) |
| derived_id = str(result["asset"]["id"]) |
| catalog = result["catalog"] |
| derived = next(item for item in catalog["assets"] |
| if str(item.get("id") or "") == derived_id) |
| derived["folder_id"] = target_folder |
| derived["parent_asset_id"] = str(parent_asset_id or "") |
| lineage = dict(transform or {}) |
| if operation_id: |
| lineage["operation_id"] = operation_id |
| derived["transform"] = lineage |
| derived["updated_at"] = _utc_now() |
| saved = self._save_catalog(catalog) |
| return {"catalog": saved, "asset": dict(derived), "reused": False} |
|
|
| def derive_image( |
| self, project: Any, asset_id: Any, *, crop: Any, target: Any, |
| resample: Any = "lanczos", tag: Any = "", folder_id: Any = None, |
| operation_id: Any = "") -> dict[str, Any]: |
| existing = self.operation_asset(project, operation_id) |
| if existing is not None: |
| return existing |
| _entry, cropped, geometry = self.prepare_image_crop( |
| project, asset_id, crop, target) |
| resample_name, resample_mode = _image_resampling(resample) |
| if "A" in cropped.getbands() or "transparency" in cropped.info: |
| cropped = cropped.convert("RGBA") |
| output_size = (geometry["target_width"], geometry["target_height"]) |
| if cropped.size != output_size: |
| cropped = cropped.resize(output_size, resample=resample_mode) |
| if cropped.mode not in ("RGB", "RGBA", "L", "LA"): |
| cropped = cropped.convert("RGBA" if "A" in cropped.getbands() else "RGB") |
| temporary = self.upload_path(project, "%s_variant.png" % asset_id) |
| try: |
| cropped.save(temporary, format="PNG") |
| return self.register_derived_image( |
| project, asset_id, temporary, tag=tag, |
| folder_id=folder_id, |
| operation_id=operation_id, |
| transform={ |
| "kind": "crop_resize", |
| "crop": {key: geometry[key] |
| for key in ("x", "y", "width", "height")}, |
| "target": { |
| "width": geometry["target_width"], |
| "height": geometry["target_height"], |
| }, |
| "source": { |
| "width": geometry["source_width"], |
| "height": geometry["source_height"], |
| }, |
| "resample": resample_name, |
| }) |
| finally: |
| try: |
| os.unlink(temporary) |
| except FileNotFoundError: |
| pass |
|
|
| def update(self, project: Any, asset_id: Any, changes: Any) -> dict[str, Any]: |
| if not isinstance(changes, dict): |
| raise ValueError("Asset changes must be a JSON object.") |
| catalog = self.load(project) |
| wanted = str(asset_id or "") |
| entry = next((item for item in catalog["assets"] |
| if str(item.get("id") or "") == wanted), None) |
| if entry is None: |
| raise FileNotFoundError("Project asset %s was not found." % wanted) |
| kind = str(entry["kind"]) |
| if "role" in changes: |
| entry["role"] = _role_for_kind(kind, changes["role"]) |
| if "tag" in changes: |
| tag = _safe_tag(changes["tag"], entry.get("tag") or "asset") |
| if any(str(item.get("tag") or "") == tag |
| and str(item.get("id") or "") != wanted |
| and bool(item.get("enabled", True)) |
| for item in catalog["assets"]): |
| raise ValueError("Another enabled project asset already uses @%s." % tag) |
| entry["tag"] = tag |
| if "enabled" in changes: |
| entry["enabled"] = bool(changes["enabled"]) |
| if "folder_id" in changes: |
| folder_id = str(changes["folder_id"] or "") |
| known_folders = {str(item.get("id") or "") |
| for item in catalog.get("folders", [])} |
| if folder_id and folder_id not in known_folders: |
| raise FileNotFoundError( |
| "Asset folder %s was not found." % folder_id) |
| entry["folder_id"] = folder_id |
| if "lyrics" in changes: |
| if kind != "audio": |
| raise ValueError("Lyrics can only be attached to audio assets.") |
| lyrics = str(changes["lyrics"] or "").replace( |
| "\r\n", "\n").replace("\r", "\n") |
| if len(lyrics) > MAX_ASSET_LYRICS_CHARACTERS: |
| raise ValueError( |
| "Asset lyrics cannot exceed %d characters." % |
| MAX_ASSET_LYRICS_CHARACTERS) |
| if lyrics: |
| entry["lyrics"] = lyrics |
| else: |
| entry.pop("lyrics", None) |
| if "options" in changes: |
| if not isinstance(changes["options"], dict): |
| raise ValueError("Asset options must be a JSON object.") |
| options = dict(entry.get("options") or {}) |
| options.update(changes["options"]) |
| entry["options"] = options |
| if bool(entry.get("enabled", True)): |
| if any(str(item.get("tag") or "") == str(entry.get("tag") or "") |
| and str(item.get("id") or "") != wanted |
| and bool(item.get("enabled", True)) |
| for item in catalog["assets"]): |
| raise ValueError( |
| "Another enabled project asset already uses @%s." % |
| entry.get("tag")) |
| if (entry.get("role") == "source_track" and any( |
| item.get("role") == "source_track" |
| and str(item.get("id") or "") != wanted |
| and bool(item.get("enabled", True)) |
| for item in catalog["assets"])): |
| raise ValueError( |
| "This project already has an enabled Source track.") |
| entry["updated_at"] = _utc_now() |
| catalog = self._save_catalog(catalog) |
| return {"catalog": catalog, "asset": dict(entry)} |
|
|
| def reorder(self, project: Any, asset_ids: Any) -> dict[str, Any]: |
| """Persist one exact permutation of the project's asset cards.""" |
| if not isinstance(asset_ids, list): |
| raise ValueError("Asset order must be a JSON list of asset IDs.") |
| catalog = self.load(project) |
| current_ids = [str(item.get("id") or "") |
| for item in catalog["assets"]] |
| requested = [str(item or "") for item in asset_ids] |
| if (len(requested) != len(set(requested)) |
| or set(requested) != set(current_ids)): |
| raise ValueError( |
| "Asset order must contain every current asset ID exactly once.") |
| by_id = { |
| str(item.get("id") or ""): item for item in catalog["assets"]} |
| catalog["assets"] = [by_id[asset_id] for asset_id in requested] |
| return {"catalog": self._save_catalog(catalog)} |
|
|
| def delete(self, project: Any, asset_id: Any) -> dict[str, Any]: |
| """Remove a catalog asset and its project-owned media copies.""" |
| catalog = self.load(project) |
| wanted = str(asset_id or "") |
| entry = next((item for item in catalog["assets"] |
| if str(item.get("id") or "") == wanted), None) |
| if entry is None: |
| raise FileNotFoundError("Project asset %s was not found." % wanted) |
| removed = dict(entry) |
| for owner in catalog["assets"]: |
| bindings = (owner.get("options") or {}).get("audio_tracks") or {} |
| if str(owner.get("id") or "") != wanted and wanted in bindings.values(): |
| raise ValueError( |
| "This audio is part of %s's track group. Detach it there " |
| "before deleting it." % (owner.get("tag") or owner["id"])) |
| catalog["assets"] = [ |
| item for item in catalog["assets"] |
| if str(item.get("id") or "") != wanted] |
| remaining_paths = { |
| str(item.get("relative_path") or "") |
| for item in catalog["assets"]} |
| saved = self._save_catalog(catalog) |
|
|
| deleted_files = 0 |
| relative = str(removed.get("relative_path") or "") |
| if relative and relative not in remaining_paths: |
| project_dir, name = self._project_dir(project) |
| backup_dir, _name = self._backup_dir(name) |
| for root in (project_dir, backup_dir): |
| path = os.path.realpath(os.path.join(root, relative)) |
| if not _inside(root, path): |
| raise ValueError("Project asset media path escapes its store.") |
| try: |
| os.unlink(path) |
| deleted_files += 1 |
| except FileNotFoundError: |
| pass |
|
|
| project_dir, _name = self._project_dir(project) |
| preview_dir = os.path.join(project_dir, "previews") |
| preview_prefix = wanted + "_" |
| if os.path.isdir(preview_dir): |
| with os.scandir(preview_dir) as previews: |
| for preview in previews: |
| if (preview.is_file(follow_symlinks=False) |
| and preview.name.startswith(preview_prefix)): |
| os.unlink(preview.path) |
| deleted_files += 1 |
| return { |
| "catalog": saved, |
| "asset": removed, |
| "deleted_files": deleted_files, |
| } |
|
|
| def input_media(self, query: Any = "") -> list[dict[str, Any]]: |
| needle = str(query or "").strip().lower() |
| result = [] |
| if not os.path.isdir(self.input_root): |
| return result |
| for root, directories, files in os.walk(self.input_root): |
| |
| |
| |
| directories[:] = [ |
| name for name in directories |
| if name.casefold() not in INPUT_BROWSER_EXCLUDED_DIRECTORIES |
| ] |
| for filename in files: |
| suffix = os.path.splitext(filename)[1].lower() |
| if suffix not in ALL_MEDIA_EXTENSIONS: |
| continue |
| path = os.path.join(root, filename) |
| relative = os.path.relpath(path, self.input_root).replace(os.sep, "/") |
| if needle and needle not in relative.lower(): |
| continue |
| try: |
| stat = os.stat(path) |
| except OSError: |
| continue |
| result.append({ |
| "path": relative, |
| "kind": _media_kind(path), |
| "size": int(stat.st_size), |
| "modified": float(stat.st_mtime), |
| }) |
| if len(result) >= MAX_INPUT_RESULTS: |
| break |
| if len(result) >= MAX_INPUT_RESULTS: |
| break |
| return sorted(result, key=lambda item: item["path"].lower()) |
|
|
| def input_path(self, relative: Any) -> str: |
| value = str(relative or "").strip() |
| path = os.path.realpath(os.path.join(self.input_root, value)) |
| if not value or not _inside(self.input_root, path) or not os.path.isfile(path): |
| raise FileNotFoundError("ComfyUI input asset was not found: %s" % value) |
| return path |
|
|
| def project_catalogs(self, query: Any = "") -> list[dict[str, Any]]: |
| """List existing live Carousel projects, including empty catalogs.""" |
| needle = str(query or "").strip().lower() |
| result = [] |
| if not os.path.isdir(self.projects_root): |
| return result |
| with os.scandir(self.projects_root) as projects: |
| for project in projects: |
| if (not project.is_dir(follow_symlinks=False) |
| or needle and needle not in project.name.lower() |
| or not os.path.isfile(os.path.join( |
| project.path, "catalog.json"))): |
| continue |
| try: |
| catalog = self.load(project.name) |
| except (OSError, TypeError, ValueError, json.JSONDecodeError): |
| continue |
| result.append({ |
| "project": project.name, |
| "asset_count": len(catalog.get("assets", [])), |
| "unassigned_count": len( |
| catalog.get("reference_slots", [])), |
| "folder_count": len(catalog.get("folders", [])), |
| "updated_at": str(catalog.get("updated_at") or ""), |
| "revision": str(catalog.get("revision") or ""), |
| }) |
| result.sort(key=lambda item: item["project"].lower()) |
| result.sort(key=lambda item: item["updated_at"], reverse=True) |
| return result[:MAX_INPUT_RESULTS] |
|
|
| def projects(self, query: Any = "", *, exclude_project: Any = "") -> list[dict[str, Any]]: |
| """List live Carousel catalogs without exposing arbitrary input files.""" |
| needle = str(query or "").strip().lower() |
| excluded = (_safe_project(exclude_project) |
| if str(exclude_project or "").strip() else "") |
| result = [] |
| remaining = MAX_INPUT_RESULTS |
| if not os.path.isdir(self.projects_root): |
| return result |
| with os.scandir(self.projects_root) as projects: |
| for project in projects: |
| if remaining <= 0: |
| break |
| if (not project.is_dir(follow_symlinks=False) |
| or project.name == excluded): |
| continue |
| try: |
| catalog = self.load(project.name) |
| except (OSError, TypeError, ValueError, json.JSONDecodeError): |
| continue |
| assets = [] |
| for raw in catalog.get("assets", []): |
| if not isinstance(raw, dict): |
| continue |
| searchable = " ".join(( |
| project.name, |
| str(raw.get("tag") or ""), |
| str(raw.get("original_name") or ""), |
| str(raw.get("role") or ""), |
| )).lower() |
| if needle and needle not in searchable: |
| continue |
| try: |
| self._asset_path(project.name, raw) |
| except (OSError, TypeError, ValueError): |
| continue |
| assets.append(dict(raw)) |
| remaining -= 1 |
| if remaining <= 0: |
| break |
| if assets: |
| result.append({ |
| "project": project.name, |
| "revision": str(catalog.get("revision") or ""), |
| "assets": assets, |
| }) |
| return sorted(result, key=lambda item: item["project"].lower()) |
|
|
| def import_project_asset( |
| self, project: Any, source_project: Any, asset_id: Any, *, |
| slot_id: Any = "") -> dict[str, Any]: |
| """Copy one asset from another live Carousel into this project.""" |
| target_name = _safe_project(project) |
| source_name = _safe_project(source_project) |
| if target_name == source_name: |
| raise ValueError( |
| "Choose another Run. Use Duplicate for an asset already in " |
| "this Carousel.") |
| source_entry, source_path = self.asset(source_name, asset_id) |
| wanted_slot = str(slot_id or "") |
| options = dict(source_entry.get("options") or {}) |
| bindings = audio_track_bindings( |
| options.pop("audio_tracks", None), self.load(source_name)["assets"]) |
| |
| |
| dependencies = { |
| value: self.asset(source_name, value) |
| for value in (bindings or {}).values() |
| if value and value != str(source_entry["id"])} if not wanted_slot else {} |
| if wanted_slot: |
| result = self.bind_reference_slot( |
| target_name, wanted_slot, source_path, |
| original_name=source_entry.get("original_name", ""), |
| source_kind="project") |
| else: |
| result = self.import_file( |
| target_name, source_path, |
| role=source_entry.get("role", ""), |
| tag=source_entry.get("tag", ""), |
| original_name=source_entry.get("original_name", ""), |
| source_kind="project", |
| options=options) |
| if bindings and not wanted_slot: |
| mapped = {str(source_entry["id"]): result["asset"]["id"]} |
| for value, (entry, path) in dependencies.items(): |
| track_options = dict(entry.get("options") or {}) |
| track_options.pop("audio_tracks", None) |
| imported = self.import_file( |
| target_name, path, role="audio_reference", |
| tag=entry.get("tag", ""), |
| original_name=entry.get("original_name", ""), |
| source_kind="project", options=track_options) |
| mapped[value] = imported["asset"]["id"] |
| self.update(target_name, mapped[value], { |
| "enabled": False, "lyrics": str(entry.get("lyrics") or "")}) |
| result = self.update(target_name, result["asset"]["id"], { |
| "options": {"audio_tracks": { |
| role: mapped[value] if value else "" |
| for role, value in bindings.items()}}}) |
| lyrics = str(source_entry.get("lyrics") or "") |
| if lyrics and result.get("asset", {}).get("kind") == "audio": |
| bound_slot_id = result.get("bound_slot_id") |
| result = self.update( |
| target_name, result["asset"]["id"], {"lyrics": lyrics}) |
| if bound_slot_id: |
| result["bound_slot_id"] = bound_slot_id |
| result["source_project"] = source_name |
| result["source_asset_id"] = str(source_entry.get("id") or "") |
| return result |
|
|
| def backups(self) -> list[dict[str, Any]]: |
| result = [] |
| if not os.path.isdir(self.chains_root): |
| return result |
| with os.scandir(self.chains_root) as runs: |
| for run in runs: |
| if not run.is_dir(follow_symlinks=False): |
| continue |
| path = os.path.join(run.path, "project_assets", "catalog.json") |
| if not os.path.isfile(path): |
| continue |
| try: |
| with open(path, "r", encoding="utf-8") as handle: |
| catalog = json.load(handle) |
| assets = catalog.get("assets") if isinstance(catalog, dict) else None |
| if not isinstance(assets, list): |
| continue |
| result.append({ |
| "run_name": run.name, |
| "revision": str(catalog.get("revision") or ""), |
| "assets": [dict(item) for item in assets if isinstance(item, dict)], |
| }) |
| except (OSError, ValueError, json.JSONDecodeError): |
| continue |
| return sorted(result, key=lambda item: item["run_name"].lower()) |
|
|
| def backup_asset_path(self, run_name: Any, asset_id: Any) -> tuple[dict[str, Any], str]: |
| backup, name = self._backup_dir(run_name) |
| path = os.path.join(backup, "catalog.json") |
| if not os.path.isfile(path): |
| raise FileNotFoundError("H3 chain %s has no project asset backup." % name) |
| with open(path, "r", encoding="utf-8") as handle: |
| catalog = json.load(handle) |
| entry = next((item for item in catalog.get("assets", []) |
| if str(item.get("id") or "") == str(asset_id or "")), None) |
| if entry is None: |
| raise FileNotFoundError("Backup asset %s was not found." % asset_id) |
| relative = str(entry.get("relative_path") or "") |
| source = os.path.realpath(os.path.join(backup, relative)) |
| if not _inside(backup, source) or not os.path.isfile(source): |
| raise FileNotFoundError("Backup media is missing: %s" % relative) |
| return dict(entry), source |
|
|
| def _preview_path(self, project: Any, entry: dict[str, Any], suffix: str) -> str: |
| directory, _name = self._project_dir(project) |
| return os.path.join(directory, "previews", "%s_%s%s" % ( |
| entry["id"], str(entry.get("sha256") or "")[:12], suffix)) |
|
|
| def ensure_poster(self, project: Any, asset_id: Any) -> str: |
| return self._ensure_still_preview( |
| project, asset_id, ".jpg", (960, 540), 86) |
|
|
| def ensure_thumbnail(self, project: Any, asset_id: Any) -> str: |
| """Return a small cached still used by large carousel collections.""" |
| return self._ensure_still_preview( |
| project, asset_id, ".thumb.jpg", (320, 180), 78) |
|
|
| def _ensure_still_preview( |
| self, project: Any, asset_id: Any, suffix: str, |
| bounds: tuple[int, int], quality: int) -> str: |
| entry, source = self.asset(project, asset_id) |
| if entry["kind"] == "audio": |
| raise ValueError("Audio assets do not have poster frames.") |
| target = self._preview_path(project, entry, suffix) |
| os.makedirs(os.path.dirname(target), exist_ok=True) |
| if os.path.isfile(target) and os.path.getsize(target) > 0: |
| return target |
| temporary = "%s.%s.tmp.jpg" % (target, uuid.uuid4().hex) |
| try: |
| if entry["kind"] == "image" and Image is not None: |
| with Image.open(source) as image: |
| frame = image.convert("RGB") |
| frame.thumbnail(bounds) |
| frame.save( |
| temporary, format="JPEG", quality=quality, |
| optimize=True) |
| else: |
| ffmpeg = shutil.which("ffmpeg") |
| if ffmpeg: |
| width, height = bounds |
| command = [ |
| ffmpeg, "-hide_banner", "-loglevel", "error", "-y", |
| "-ss", "0", "-i", source, "-frames:v", "1", |
| "-vf", "scale='min(%d,iw)':'min(%d,ih)':" |
| "force_original_aspect_ratio=decrease" % ( |
| width, height), |
| "-q:v", "3", temporary, |
| ] |
| completed = subprocess.run( |
| command, stdout=subprocess.DEVNULL, |
| stderr=subprocess.PIPE, timeout=60, check=False) |
| if completed.returncode: |
| raise RuntimeError(completed.stderr.decode( |
| "utf-8", errors="replace")[-1200:]) |
| elif av is not None and Image is not None: |
| with av.open(source, mode="r") as container: |
| video = next(iter(container.streams.video), None) |
| frame = next(container.decode(video), None) if video else None |
| if frame is None: |
| raise ValueError("Video contains no decodable frame.") |
| image = frame.to_image().convert("RGB") |
| image.thumbnail(bounds) |
| image.save( |
| temporary, format="JPEG", quality=quality) |
| else: |
| raise RuntimeError("Video poster generation requires ffmpeg or PyAV/Pillow.") |
| if not os.path.isfile(temporary) or os.path.getsize(temporary) <= 0: |
| raise RuntimeError("Preview generator produced an empty poster.") |
| os.replace(temporary, target) |
| finally: |
| try: |
| os.unlink(temporary) |
| except FileNotFoundError: |
| pass |
| return target |
|
|
| @staticmethod |
| def _browser_video(entry: dict[str, Any], path: str) -> bool: |
| suffix = os.path.splitext(path)[1].lower() |
| codec = str((entry.get("metadata") or {}).get("codec") or "").lower() |
| return ((suffix in (".mp4", ".m4v", ".mov") and codec in ( |
| "h264", "av1", "hevc")) or |
| (suffix == ".webm" and codec in ("vp8", "vp9", "av1"))) |
|
|
| def ensure_browser_media(self, project: Any, asset_id: Any) -> str: |
| entry, source = self.asset(project, asset_id) |
| if entry["kind"] != "video" or self._browser_video(entry, source): |
| return source |
| target = self._preview_path(project, entry, ".mp4") |
| os.makedirs(os.path.dirname(target), exist_ok=True) |
| if os.path.isfile(target) and os.path.getsize(target) > 0: |
| return target |
| ffmpeg = shutil.which("ffmpeg") |
| if not ffmpeg: |
| return source |
| temporary = "%s.%s.tmp.mp4" % (target, uuid.uuid4().hex) |
| try: |
| command = [ |
| ffmpeg, "-hide_banner", "-loglevel", "error", "-y", |
| "-i", source, "-vf", "scale='min(1280,iw)':-2", |
| "-c:v", "libx264", "-preset", "veryfast", "-crf", "25", |
| "-pix_fmt", "yuv420p", "-movflags", "+faststart", |
| "-c:a", "aac", "-b:a", "128k", temporary, |
| ] |
| completed = subprocess.run( |
| command, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, |
| timeout=max(120, int(float((entry.get("metadata") or {}).get( |
| "duration_seconds") or 0) * 8)), check=False) |
| if completed.returncode: |
| raise RuntimeError(completed.stderr.decode( |
| "utf-8", errors="replace")[-1600:]) |
| if not os.path.isfile(temporary) or os.path.getsize(temporary) <= 0: |
| raise RuntimeError("Preview proxy generator produced an empty video.") |
| os.replace(temporary, target) |
| finally: |
| try: |
| os.unlink(temporary) |
| except FileNotFoundError: |
| pass |
| return target |
|
|
