Source code for cgl.plugins.preflight.preflight_check

import logging

from cgl.ui.widgets.dialog import InputDialog
import json
import time
from datetime import datetime




class PreflightCheck(object):
    shared_data = {}

    def __init__(self, parent=None):
        """
        Allows for running check to run on a bg thread, not locking up the GUI.
        Keep thread_safe False if doing things on the main thread, like or running maya cmds or launching QTWidgets,
        use the PreflightCheckThreadSafe class as your base class if you know your preflight check is thread safe
        """
        self.parent = parent
        self.status = False
        self.feedback = ""
        self.items = []
        self.thread_safe = False
        self._manifest = None
        self._start_time = time.time()

    @property
    def manifest_path(self):
        return self.shared_data.get("manifest_path", None)

    @property
    def manifest(self):
        """Lazy-load manifest when first accessed."""
        if self._manifest is None and self.manifest_path:
            self._manifest = self._load_manifest()
        return self._manifest

    def _load_manifest(self):
        if not self.manifest_path:
            return None
        with open(self.manifest_path, "r", encoding="utf-8") as f:
            return json.load(f)



    def save_manifest(self):
        """Validate and write manifest back to disk, then reload it fresh."""
        if self.manifest is None:
            logging.warning("No manifest loaded, nothing to save.")
            return

        try:
            from cgl.msd.validation import validate_ingest_manifest

            validate_ingest_manifest(self.manifest)
        except Exception as e:
            self.fail_check(f"Manifest validation failed: {e}")
            raise

        with open(self.manifest_path, "w", encoding="utf-8") as f:
            json.dump(self.manifest, f, indent=2)

        logging.info("Manifest updated: %s", self.manifest_path)

        # reload fresh copy from disk to avoid stale in-memory state
        self._manifest = self._load_manifest()




    def fail_check(self, feedback):
        logging.info(feedback)
        self.status = False
        self.feedback = feedback




    def pass_check(self, feedback):
        logging.info(feedback)
        self.status = True
        self.feedback = feedback




    def getName(self):
        raise NotImplementedError




    def init(self):
        self.setWindowTitle(self.getName())
        try:
            self.run()
        except Exception as error:
            logging.error(error, exc_info=True)
            self.send_feedback("SOMETHING WENT WRONG:")
            self.parent().checkError(error)




    def run(self):
        pass




    @staticmethod
    def final_check():
        message = "Publish Successful - you can close this window"
        dialog = InputDialog(
            title="Publish Successful", message=message, buttons=["Ok"]
        )
        dialog.exec()




# use this class if you know your preflight check is safe to run on a BG thread


class PreflightCheckThreadSafe(PreflightCheck):
    def __init__(self, parent=None):
        super(PreflightCheckThreadSafe, self).__init__(parent)
        self.thread_safe = True