code_gallery

Web News Blog Press Pricing

Learn

Training

Profiles

Executive

Templates

VFX

Code

Code Gallery Designer

v5.20

Maya

Check Refs Missing

Jorge Hernandez Ibañez 07/10/2025

Tags
references assembly asset

Check Refs Missing

Jorge Herna…

Action v1

07/10/2025

Maya

Pseudo-Code

Parameters

Talk with TD Support

Import to my Pipeline

Code

import maya.cmds as cmds
import os
import sys
import pipelinepro as pipeline
import requests
import krakenPlugin.maya.assets as krAssets

# PySide imports with version detection
try:
    import dcc_utils
    pyside_version = dcc_utils.get_pySide_version()
except ImportError:
    pyside_version = 2  # Default fallback

if pyside_version == 2:
    from PySide2 import QtWidgets, QtCore
    from PySide2.QtWidgets import QProgressDialog
    from shiboken2 import wrapInstance
else:
    from PySide6 import QtWidgets, QtCore
    from PySide6.QtWidgets import QProgressDialog
    from shiboken6 import wrapInstance

from maya import OpenMayaUI as omui

# Configuration
allowed_dccs = ["Maya", "Zbrush", "Houdini", "Blender"]
allowed_maintasks = ["Rig"]

def get_maya_main_window():
    """Get Maya's main window for proper dialog parenting."""
    maya_main_window_ptr = omui.MQtUtil.mainWindow()
    if maya_main_window_ptr:
        return wrapInstance(int(maya_main_window_ptr), QtWidgets.QWidget)
    return None

def create_progress_dialog(title, maximum_value, parent=None):
    """Create and configure a progress dialog for the loading process.

Args:
        title (str): The title for the progress dialog
        maximum_value (int): Maximum value for the progress bar
        parent (QWidget): Parent widget (Maya main window)

Returns:
        QProgressDialog: Configured progress dialog
    """
    if parent is None:
        parent = get_maya_main_window()

progress_dialog = QProgressDialog(parent)
    progress_dialog.setWindowTitle(title)
    progress_dialog.setLabelText("Initializing...")
    progress_dialog.setMinimum(0)
    progress_dialog.setMaximum(maximum_value)
    progress_dialog.setValue(0)
    progress_dialog.setAutoClose(True)
    progress_dialog.setAutoReset(False)
    progress_dialog.setModal(True)

# Make dialog stay on top and be visible
    progress_dialog.show()
    QtWidgets.QApplication.processEvents()

return progress_dialog

def exec_query(query):
    """Execute a pipeline database query and return the response."""
    if "token:" not in query:
        print("Warning: The token key is not included in the query, it will probably fail.")
    try:
        response = pipeline.query(query)
    except Exception as e:
        print("Error on DB query: ", e)
        return None
    return response

def get_file_asset_key(filename):
    """Get the asset key for a given filename."""
    query = """
    query {
        file(fileName: "{filename}", token: "{token}") {
            asset {
                key
            }
        }
    }
    """
    query = query.replace("{filename}", filename)
    response = exec_query(query)
    if response and "file" in response and response["file"] and "asset" in response["file"] and response["file"]["asset"]:
        return response["file"]["asset"]["key"]
    return None

def get_filename():
    """Return the current Maya file name."""
    try:
        scene_path = cmds.file(query=True, sceneName=True)
        if scene_path:
            return os.path.basename(scene_path)
        else:
            print("No Maya scene file is currently open.")
            return ""
    except Exception as e:
        print(f"Error getting Maya scene filename: {e}")
        return ""

def get_assets_breakdown(filename):
    """Query the pipeline for all assets assigned to the shot based on the filename."""
    query = """
    query {
        file(fileName: "{filename}", token: "{token}") {
            shot {
                assets {
                    name
                    key
                    tecnicalName
                    assetType {
                        name
                        initials
                    }
                    tasks {
                        maintask {
                            name
                            key
                            dcc{
                                name
                            }
                        }
                    }
                }
            }
            asset{
                breakdown{
                    name
                    key
                    tecnicalName
                    assetType {
                        name
                        initials
                    }
                    tasks {
                        maintask {
                            name
                            key
                            dcc{
                                name
                            }
                        }
                    }
                }
            }
            task{
                name
            }
        }
    }
    """
    query = query.replace("{filename}", filename)
    response = exec_query(query)
    if not response:
        return [], None, ""
    file_data = response.get("file")
    if not file_data:
        return [], None, ""
    shot = file_data.get("shot", {})
    asset = file_data.get("asset", {})
    if not shot and not asset:
        return [], None, ""
    if shot:  # Shot breakdown
        assets = shot.get("assets", [])
    else:  # Asset breakdown
        assets = asset.get("breakdown", [])
    return assets if assets else [], "shot" if shot else "asset", file_data.get("task", {}).get("name", "")

def get_asset_files(asset_key, maintask_key):
    """Query the pipeline for published files of a given asset and maintask."""
    query_asset_files = """
        query{
          task(token:"{token}", assetKey: "{asset_key}", maintask:"{maintask_key}"){
            publishFolder{
              files{
                fileName
                fileLocal
                link
                pathToFile
                folder {
                    name
                    type
                    dependence {
                        name
                        type
                    }
                }
                task {
                    name
                    maintask {
                        tecnicalName
                        name
                    }
                }
                asset {
                    tecnicalName
                    assetType {
                        initials
                    }
                }
                shot{
                    name
                    tecnicalName
                }
              }
            }
            lastReviewFolder{
              files{
                fileName
                fileLocal
                link
                pathToFile
                folder {
                    name
                    type
                    dependence {
                        name
                        type
                    }
                }
                task {
                    name
                    maintask {
                        tecnicalName
                        name
                    }
                }
                asset {
                    tecnicalName
                    assetType {
                        initials
                    }
                }
                shot{
                    name
                    tecnicalName
                }
              }
            }
          }
        }
    """
    query_asset_files = query_asset_files.replace(
        "{maintask_key}", maintask_key
    ).replace("{asset_key}", asset_key)
    response = exec_query(query_asset_files)
    if (
        response
        and "task" in response
        and response["task"] is not None
    ):
        if "publishFolder" in response["task"]:
            folder_active = response["task"].get("publishFolder", None)
            if folder_active and "files" in folder_active and folder_active["files"]:
                return folder_active["files"]
    return []

def download_file(download_link, file_local_path):
    """Download the Maya file from the cloud and save it to the correct path.
    Args:
        download_link (str): The link to download the file.
        file_local_path (str): The local path to save the file.
    """
    print(f"Downloading file to: {file_local_path}")
    file_local_path = os.path.normpath(file_local_path)
    file_local_dir = os.path.dirname(file_local_path)
    try:
        print("Starting download request ----")
        response = requests.get(download_link)
        response.raise_for_status()
        print("Download request ended ----")
        os.makedirs(file_local_dir, exist_ok=True)
        with open(file_local_path, "wb") as f:
            f.write(response.content)
        print(f"File downloaded successfully in: {file_local_path}")
    except requests.exceptions.RequestException as e:
        print(f"Error on file download: {e}")

def load_asset(asset, maintask_key, asset_type, parent=None, target_file=None):
    """Load and reference the asset into the Maya scene."""
    asset_key = asset.get("key")
    files = get_asset_files(asset_key, maintask_key)
    asset_name = asset.get("tecnicalName", "").replace(" ", "_")

if not files:
        print(f'Asset {asset_name} has no files')
        return False

for file in files:
        print(f'Processing: {file["fileName"]}')
        if not file.get("fileName") or not (file.get("fileName").endswith(".ma") or file.get("fileName").endswith(".mb")):
            continue

filename = file["fileName"]
        asset_path = file.get("fileLocal", None)

# If fileLocal does not exist -> download
        if not os.path.isfile(asset_path):
            file_link = file["link"]
            download_file(file_link, asset_path)

# Import the asset file using the improved import function
        result = krAssets.import_asset_file(file, asset_key, parent=parent, target_file=target_file)
        if result:
            print(f"Asset '{asset_name}' imported successfully.")
        else:
            print(f"Failed to import asset '{asset_name}'.")
    return True

def get_current_scene_asset_keys():
    """
    Get all asset keys currently referenced in the Maya scene.

Returns:
        set: Set of asset keys found in the scene
    """
    try:
        # Get all reference nodes
        ref_nodes = cmds.ls(type='reference')
        asset_keys = set()

for ref_node in ref_nodes:
            # Skip the sharedReferenceNode
            if ref_node == 'sharedReferenceNode':
                continue

try:
                # Get the reference file path
                ref_file = cmds.referenceQuery(ref_node, filename=True, withoutCopyNumber=True)

if ref_file and os.path.isabs(ref_file):
                    # Extract filename from the reference path
                    ref_filename = os.path.basename(ref_file)

# Query the database to get the asset key for this file
                    asset_key = get_file_asset_key(ref_filename)

if asset_key:
                        asset_keys.add(asset_key)
                        print(f"Found asset key in scene: {asset_key} (from file: {ref_filename})")
            except Exception as e:
                print(f"Error querying reference {ref_node}: {e}")
                continue

return asset_keys
    except Exception as e:
        print(f"Error getting scene asset keys: {e}")
        return set()

def get_missing_assets():
    """
    Compare expected assets from breakdown with currently loaded assets.

Returns:
        tuple: (missing_assets: list, entity_type: str, task_name: str)
    """
    try:
        # Get current filename
        filename = get_filename()
        if not filename:
            return [], None, ""

# Get expected assets from breakdown
        expected_assets, entity_type, task_name = get_assets_breakdown(filename)

if not expected_assets:
            print(f"No assets assigned to this shot/asset - file name: {filename}")
            return [], entity_type, task_name

# Get currently loaded asset keys in the scene
        loaded_asset_keys = get_current_scene_asset_keys()

# Find missing assets
        missing_assets = []
        for asset in expected_assets:
            asset_key = asset.get("key")
            asset_tasks = asset.get("tasks", [])

# Filter tasks by allowed DCCs and maintasks
            if allowed_dccs:
                asset_tasks = [task for task in asset_tasks if task.get("maintask", {}).get("dcc", {}).get("name") in allowed_dccs]
            if allowed_maintasks:
                asset_tasks = [task for task in asset_tasks if task.get("maintask", {}).get("name") in allowed_maintasks]

# Skip assets with no valid tasks
            if not asset_tasks:
                continue

# Check if asset is loaded in scene
            if asset_key not in loaded_asset_keys:
                missing_assets.append(asset)
                print(f"Missing asset: {asset.get('name')} (key: {asset_key})")

return missing_assets, entity_type, task_name

except Exception as e:
        print(f"Error getting missing assets: {e}")
        return [], None, ""

def check_process():
    """
    Check if all expected assets are referenced in the Maya scene.

Returns:
        tuple: (check_result: bool, check_error: str or False)
    """
    try:
        missing_assets, entity_type, task_name = get_missing_assets()

if not missing_assets:
            # All expected assets are present
            return True, False
        else:
            # Found missing assets
            asset_names = [asset.get('name', 'Unknown') for asset in missing_assets]
            error_msg = f"Found {len(missing_assets)} missing asset reference(s): {', '.join(asset_names)}"
            return False, error_msg

except Exception as e:
        return False, f"Error checking missing references: {str(e)}"

def select_process():
    """
    Select operation for missing references.
    Since the references are missing, there's nothing to select.

Returns:
        tuple: (select_result: bool, select_error: str or False)
    """
    try:
        # Clear selection since we can't select missing references
        cmds.select(clear=True)
        return True, False

except Exception as e:
        return False, f"Error in select operation: {str(e)}"

def fix_process():
    """
    Load missing asset references into the Maya scene.

Returns:
        bool: True if fix was successful, False otherwise
    """
    try:
        missing_assets, entity_type, task_name = get_missing_assets()

if not missing_assets:
            print("No missing assets found")
            return True

# Get current file data for reference
        file_data = krAssets.get_current_file_data()

# Create progress dialog
        progress_dialog = None
        try:
            total_assets = len(missing_assets)
            progress_dialog = create_progress_dialog("Loading Missing References", total_assets)
            progress_dialog.setLabelText(f"Preparing to load {total_assets} missing asset(s)...")
            QtWidgets.QApplication.processEvents()

loaded_count = 0

# Process each missing asset
            for i, asset in enumerate(missing_assets):
                # Check if user cancelled the operation
                if progress_dialog.wasCanceled():
                    print("Loading cancelled by user.")
                    break

# Update progress dialog
                asset_name = asset.get("name", "Unknown Asset")
                progress_dialog.setLabelText(f"Loading asset {i + 1} of {total_assets}: {asset_name}")
                progress_dialog.setValue(i)
                QtWidgets.QApplication.processEvents()

print(f"Loading missing asset: {asset_name} - key: {asset.get('key')}")

asset_tasks = asset.get("tasks", [])

# Filter tasks by allowed DCCs and maintasks
                if allowed_dccs:
                    asset_tasks = [task for task in asset_tasks if task.get("maintask", {}).get("dcc", {}).get("name") in allowed_dccs]
                if allowed_maintasks:
                    asset_tasks = [task for task in asset_tasks if task.get("maintask", {}).get("name") in allowed_maintasks]

if not asset_tasks:
                    print(f"The asset {asset_name} does not have valid tasks.")
                    continue

# Try to load asset with the first valid task
                for task in reversed(asset_tasks):
                    maintask = task.get("maintask", {})
                    dcc_name = maintask.get("dcc", {}).get("name", "Unknown DCC")
                    print(f"Processing task: {maintask.get('name')} for DCC: {dcc_name}")

if entity_type == "shot":
                        success = load_asset(asset, maintask.get("key"), asset.get("assetType", {}), target_file=file_data)
                    else:
                        success = load_asset(asset, maintask.get("key"), asset.get("assetType", {}), parent=task_name, target_file=file_data)

if success:
                        loaded_count += 1
                        break

# Complete the progress
            if not progress_dialog.wasCanceled():
                progress_dialog.setValue(total_assets)
                progress_dialog.setLabelText(f"Loaded {loaded_count} missing asset(s) successfully!")
                QtWidgets.QApplication.processEvents()
                print(f"Successfully loaded {loaded_count} missing asset(s)")

return loaded_count > 0

finally:
            # Clean up progress dialog
            if progress_dialog:
                progress_dialog.close()
                QtWidgets.QApplication.processEvents()

except Exception as e:
        print(f"Error fixing missing references: {str(e)}")
        return False

#---------------------------------------------------------------------------------------------
# UI Functions - do not change
def check():
    """Check for missing asset references."""
    check_result, check_error = check_process()

if check_result:
        return True
    else:
        return check_error

def select():
    """Select missing references (clears selection since they don't exist)."""
    select_result, select_error = select_process()

if select_result:
        return True
    else:
        return select_error

def fix():
    """Load missing asset references into the scene."""
    fix_result = fix_process()
    check_result = check()
    return fix_result, check_result

Import Action

1. Copy Key

Copy

2. Go to your Action pannel

Go