lib

def check_libs():
    """Check if `OpenHarmony`_ is available.

    AYON expects either path in `LIB_OPENHARMONY_PATH` or `openHarmony.js`
    present in `TOONBOOM_GLOBAL_SCRIPT_LOCATION`.

    Throws:
        RuntimeError: If openHarmony is not found.

    .. _OpenHarmony:
        https://github.com/cfourney/OpenHarmony

    """
    if os.getenv("LIB_OPENHARMONY_PATH"):
        return

    script_location = os.getenv("TOONBOOM_GLOBAL_SCRIPT_LOCATION")
    if not script_location:
        log.error(
            "Cannot find OpenHarmony library."
            " Please set path to it in LIB_OPENHARMONY_PATH"
            " environment variable."
        )
        raise RuntimeError("Missing OpenHarmony library.")

    script_path = os.path.join(script_location, "openHarmony.js")
    if os.path.exists(script_path):
        os.environ["LIB_OPENHARMONY_PATH"] = script_location

def delete_node(node):
    """ Physically delete node from scene. """
    send(
        {
            "function": "AyonHarmonyAPI.deleteNode",
            "args": node
        }
    )

def find_backdrop_by_name(name: str) -> Optional[dict]:
    """Find backdrop by its name.

    Args:
        name (str): Name of the backdrop.

    Returns:
        dict: Backdrop.
    """
    backdrops = send(
        {"function": "Backdrop.backdrops", "args": ["Top"]}
    )["result"]
    for backdrop in backdrops:
        if backdrop["title"]["text"] == name:
            return backdrop

    return None

def find_node_by_name(name, node_type):
    """Find node by its name.

    Args:
        name (str): Name of the Node. (without part before '/')
        node_type (str): Type of the Node.
            'READ' - for loaded data with Loaders (background)
            'GROUP' - for loaded data with Loaders (templates)
            'WRITE' - render nodes

    Returns:
        str: FQ Node name.

    """
    nodes = send(
        {"function": "node.getNodes", "args": [[node_type]]}
    )["result"]
    for node in nodes:
        node_name = node.split("/")[-1]
        if name == node_name:
            return node

    return None

def get_all_top_names() -> set:
    """Get all top node and backdrop names in the scene.

    Returns:
        set: Set of top node names.

    """
    nodes = send({"function": "node.subNodes", "args": ["Top"]})["result"]
    backdrops = {
        backdrop["title"]["text"]
        for backdrop in send(
            {"function": "Backdrop.backdrops", "args": ["Top"]}
        )["result"]
    }
    return set(nodes) | backdrops

def get_layers_info() -> list[dict[str, str]]:
    """Returns list of dicts with info about timeline layers

    'position' goes from 0 at the top and increases to bottom on timeline
    """
    layers_info = send(
        {
            "function": "AyonHarmony.getLayerInfos",
            "args": []
        }
    )["result"]
    layers_info = [layer for layer in layers_info if layer["enabled"]]
    return sorted(
        layers_info,
        key=lambda layer: layer["position"],
        reverse=True
    )

def get_local_harmony_path(filepath):
    """From the provided path get the equivalent local Harmony path."""
    basename = os.path.splitext(os.path.basename(filepath))[0]
    harmony_path = os.path.join(os.path.expanduser("~"), ".ayon", "harmony")
    return os.path.join(harmony_path, basename)

def get_palettes_paths() -> set:
    """Get all palettes paths in the scene.

    Returns:
        set: Set of palettes paths.
    """
    return {pal["_path"] for pal in send(
        {"function": "AyonHarmony.getAllPalettesPaths"}
    )["result"]}

def imprint(node_id, data, remove=False):
    """Write `data` to the `node` as json.

    Arguments:
        node_id (str): Path to node or id of object.
        data (dict): Dictionary of key/value pairs.
        remove (bool): Removes the data from the scene.

    Example:
        >>> from ayon_harmony.api import lib
        >>> node = "Top/Display"
        >>> data = {"str": "something", "int": 1, "float": 0.32, "bool": True}
        >>> lib.imprint(layer, data)
    """
    scene_data = get_scene_data()

    if remove and (node_id in scene_data):
        scene_data.pop(node_id, None)
    else:
        if node_id in scene_data:
            scene_data[node_id].update(data)
        else:
            scene_data[node_id] = data

    set_scene_data(scene_data)

def launch(application_path, *args):
    """Set Harmony for launch.

    Launches Harmony and the server, then starts listening on the main thread
    for callbacks from the server. This is to have Qt applications run in the
    main thread.

    Args:
        application_path (str): Path to Harmony.

    """
    from ayon_core.pipeline import install_host
    from ayon_harmony.api import HarmonyHost

    install_host(HarmonyHost())

    ProcessContext.port = random.randrange(49152, 65535)
    os.environ["AYON_HARMONY_PORT"] = str(ProcessContext.port)
    ProcessContext.application_path = application_path

    # Launch Harmony.
    setup_startup_scripts()
    check_libs()

    if len(args) > 0 and (scene_path := Path(args[-1])).suffix == ".zip":
        launch_zip_file(scene_path)

    open_workfile_app = env_value_to_bool("AYON_HARMONY_WORKFILES_ON_LAUNCH")
    workfile_already_open = ProcessContext.workfile_path
    if open_workfile_app or not workfile_already_open:
        ProcessContext.workfile_tool = host_tools.get_tool_by_name(
            "workfiles"
        )
        host_tools.show_workfiles(save=False)
        ProcessContext.execute_in_main_thread(check_workfiles_tool)

def launch_zip_file(filepath):
    """Launch a Harmony application instance with the provided zip file.

    Args:
        filepath (str): Path to file.
    """
    # Close existing scene.
    if ProcessContext.pid:
        os.kill(ProcessContext.pid, signal.SIGTERM)

    # Stop server.
    if ProcessContext.server:
        ProcessContext.server.stop()

    # Launch AYON server.
    ProcessContext.server = Server(ProcessContext.port)
    ProcessContext.server.start()
    # thread = threading.Thread(target=self.server.start)
    # thread.daemon = True
    # thread.start()

    # Save workfile path for later.
    ProcessContext.workfile_path = filepath

    # Unzip the scene file and get the .xstage path
    try:
        scene_path = unzip_scene_file(filepath)
    except Exception as e:
        print(f"Error unzipping scene file: {e}")
        ProcessContext.server.stop()
        return

    print("Launching {}".format(scene_path))
    # QUESTION Could we use 'run_detached_process' from 'ayon_core.lib'?
    kwargs = {}
    if (
        platform.system().lower() == "windows"
        and not is_using_ayon_console()
    ):
        kwargs.update({
            "creationflags": subprocess.CREATE_NO_WINDOW,
            "stdout": subprocess.DEVNULL,
            "stderr": subprocess.DEVNULL
        })

    process = subprocess.Popen(
        [ProcessContext.application_path, scene_path],
        **kwargs
    )
    ProcessContext.pid = process.pid
    ProcessContext.process = process
    ProcessContext.stdout_broker.host_connected()

@contextlib.contextmanager
def maintained_nodes_state(nodes):
    """Maintain nodes states during context."""
    # Collect current state.
    states = send(
        {
            "function": "AyonHarmonyAPI.areEnabled", "args": nodes
        })["result"]

    # Disable all nodes.
    send(
        {
            "function": "AyonHarmonyAPI.disableNodes", "args": nodes
        })

    try:
        yield
    finally:
        send(
            {
                "function": "AyonHarmonyAPI.setState",
                "args": [nodes, states]
            })

def on_file_changed(path, threaded=True):
    """Threaded zipping and move of the project directory.

    This method is called when the `.xstage` file is changed.
    """
    log.debug("File changed: " + path)

    if ProcessContext.workfile_path is None:
        return

    if threaded:
        thread = threading.Thread(
            target=zip_and_move,
            args=(os.path.dirname(path), ProcessContext.workfile_path)
        )
        thread.start()
    else:
        zip_and_move(os.path.dirname(path), ProcessContext.workfile_path)

def read(node_id):
    """Read object metadata in to a dictionary.

    Args:
        node_id (str): Path to node or id of object.

    Returns:
        dict
    """
    scene_data = get_scene_data()
    if node_id in scene_data:
        return scene_data[node_id]

    return {}

def remove(node_id):
    """
        Remove node data from scene metadata.

        Args:
            node_id (str): full name (eg. 'Top/renderAnimation')
    """
    data = get_scene_data()
    del data[node_id]
    set_scene_data(data)

def rename_node(node_name, new_name):
    """ Rename node name """
    send(
        {
            "function": "AyonHarmony.renameNode",
            "args": [node_name, new_name]
        }
    )

def save_scene(zip_and_move=True):
    """Save the Harmony scene safely.

    The built-in (to AYON) background zip and moving of the Harmony scene
    folder, interferes with server/client communication by sending two
    requests at the same time. This only happens when sending
    "scene.saveAll()". This method prevents this double request and safely
    saves the scene.

    """
    # Need to turn off the background watcher else the communication with
    # the server gets spammed with two requests at the same time.
    scene_path = send(
        {"function": "AyonHarmonyAPI.saveScene"})["result"]

    # # Manually update the remote file.
    if zip_and_move:
        on_file_changed(scene_path, threaded=False)

    # Re-enable the background watcher.
    send({"function": "AyonHarmonyAPI.enableFileWather"})

def save_scene_as(filepath):
    """Save Harmony scene as `filepath`."""
    scene_dir = os.path.dirname(filepath)
    destination = os.path.join(
        os.path.dirname(ProcessContext.workfile_path),
        os.path.splitext(os.path.basename(filepath))[0] + ".zip"
    )

    if os.path.exists(scene_dir):
        try:
            shutil.rmtree(scene_dir)
        except Exception as e:
            log.error(f"Cannot remove {scene_dir}")
            raise Exception(f"Cannot remove {scene_dir}") from e

    send(
        {"function": "scene.saveAs", "args": [scene_dir]}
    )["result"]

    zip_and_move(scene_dir, destination)

    ProcessContext.workfile_path = destination

    send(
        {"function": "AyonHarmonyAPI.addPathToWatcher", "args": filepath}
    )

def select_nodes(nodes):
    """ Selects nodes in Node View """
    _ = send(
        {
            "function": "AyonHarmonyAPI.selectNodes",
            "args": nodes
        }
    )

def send(request):
    """Public method for sending requests to Harmony."""
    return ProcessContext.server.send(request)

def set_scene_data(data):
    """Write scene data to metadata.

    Args:
        data (dict): Data to write.

    """
    # Write scene data.
    send(
        {
            "function": "AyonHarmonyAPI.setSceneData",
            "args": data
        })

def setup_startup_scripts():
    """Manages installation of ayon's TB_sceneOpened.js for Harmony launch.

    If a studio already has defined "TOONBOOM_GLOBAL_SCRIPT_LOCATION", copies
    the TB_sceneOpened.js to that location if the file is different.
    Otherwise, will set the env var to point to the ayon/harmony folder.

    Admins should be aware that this will overwrite TB_sceneOpened in the
    "TOONBOOM_GLOBAL_SCRIPT_LOCATION", and that if they want to have additional
    logic, they will need to one of the following:
        * Create a Harmony package to manage startup logic
        * Use TB_sceneOpenedUI.js instead to manage startup logic
        * Add their startup logic to ayon/harmony/TB_sceneOpened.js
    """
    ayon_host_dir = os.path.join(HARMONY_ADDON_ROOT, "api")
    startup_js = "TB_sceneOpened.js"

    env_location = os.getenv("TOONBOOM_GLOBAL_SCRIPT_LOCATION")
    if not env_location:
        os.environ["TOONBOOM_GLOBAL_SCRIPT_LOCATION"] = ayon_host_dir
        return

    ayon_harmony_startup = os.path.join(ayon_host_dir, startup_js)
    env_harmony_startup = os.path.join(env_location, startup_js)

    # Check if destination file exists or if files are the same
    if (
        os.path.exists(env_harmony_startup)
        and filecmp.cmp(ayon_harmony_startup, env_harmony_startup)
    ):
        return

    try:
        shutil.copy(ayon_harmony_startup, env_harmony_startup)
    except Exception:
        log.warning(
            f"Failed to copy {ayon_harmony_startup} to {env_harmony_startup}!"
            " Defaulting to AYON's TOONBOOM_GLOBAL_SCRIPT_LOCATION.",
            exc_info=True
        )

        os.environ["TOONBOOM_GLOBAL_SCRIPT_LOCATION"] = ayon_host_dir

def show(tool_name):
    """Call show on "module_name".

    This allows to make a QApplication ahead of time and always "exec_" to
    prevent crashing.

    Args:
        module_name (str): Name of module to call "show" on.

    """
    # Requests often get doubled up when showing tools, so we wait a second
    #   for requests to be received properly.
    time.sleep(1)

    kwargs = {}
    if tool_name == "loader":
        kwargs["use_context"] = True
    elif tool_name == "publisher":
        kwargs["tab"] = "publish"
    elif tool_name == "creator":
        tool_name = "publisher"
        kwargs["tab"] = "create"

    ProcessContext.execute_in_main_thread(
        lambda: host_tools.show_tool_by_name(tool_name, **kwargs)
    )

    # Required return statement.
    return "nothing"

def signature(postfix="func") -> str:
    """Return random ECMA6 compatible function name.

    Args:
        postfix (str): name to append to random string.
    Returns:
        str: random function name.

    """
    return "f{}_{}".format(str(uuid4()).replace("-", "_"), postfix)

def unzip_scene_file(filepath: str) -> str:
    """Unzip a Harmony scene file and return the path to the .xstage file.

    Args:
        filepath (str): Path to the zip file.

    Returns:
        str: Path to the .xstage file.

    Raises:
        Exception: If no .xstage file is found or if the working
            folder cannot be deleted.

    """
    print(f"Localizing {filepath}")

    local_scene_dir_path = get_local_harmony_path(filepath)
    scene_name = os.path.basename(local_scene_dir_path)
    if os.path.exists(os.path.join(local_scene_dir_path, scene_name)):
        # unzipped with duplicated scene_name
        local_scene_dir_path = os.path.join(local_scene_dir_path, scene_name)

    scene_path = os.path.join(
        local_scene_dir_path, f"{scene_name}.xstage"
    )

    unzip = True
    if os.path.exists(scene_path):
        unzip = False
        # Check remote scene is newer than local.
        if os.path.getmtime(scene_path) < os.path.getmtime(filepath):
            try:
                shutil.rmtree(local_scene_dir_path)
            except Exception as e:
                log.error(e)
                raise Exception("Cannot delete working folder") from e
            unzip = True

    if unzip:
        with _ZipFile(filepath, "r") as zip_ref:
            zip_ref.extractall(local_scene_dir_path)

        if os.path.exists(os.path.join(local_scene_dir_path, scene_name)):
            # unzipped with duplicated scene_name
            local_scene_dir_path = os.path.join(
                local_scene_dir_path, scene_name
            )

    # find any xstage files is directory, prefer the one with the same name
    # as directory (plus extension)
    xstage_files = []
    for root, _, files in os.walk(local_scene_dir_path):
        for file in files:
            if os.path.splitext(file)[1] == ".xstage":
                full_path = os.path.join(root, file)
                relative_path = os.path.relpath(
                    full_path, local_scene_dir_path
                )
                xstage_files.append(relative_path)

    if not os.path.basename("temp.zip"):
        if not xstage_files:
            raise Exception("No xstage file was found.")

    # prefer the one named as zip file
    zip_based_name = "{}.xstage".format(
        os.path.splitext(os.path.basename(filepath))[0])

    xstage_files.reverse()  # prefer 0 found xstage
    for relative_path_xstage in xstage_files:
        scene_path = os.path.join(
            local_scene_dir_path, relative_path_xstage
        )
        if zip_based_name in relative_path_xstage:
            break

    if not os.path.exists(scene_path):
        raise Exception(
            f"Expected '{scene_path}' not found in '{local_scene_dir_path}'."
        )

    return scene_path

def zip_and_move(source, destination):
    """Zip a directory and move to `destination`.

    Args:
        source (str): Directory to zip and move to destination.
        destination (str): Destination file path to zip file.

    """
    os.chdir(os.path.dirname(source))
    shutil.make_archive(os.path.basename(source), "zip", source)
    with _ZipFile(os.path.basename(source) + ".zip") as zr:
        if zr.testzip() is not None:
            raise Exception("File archive is corrupted.")
    shutil.move(os.path.basename(source) + ".zip", destination)
    log.debug(f"Saved '{source}' to '{destination}'")