import logging
import os
import re
from importlib import reload
import cgl.core.msd as msd
from cgl.core.path.object import PathObject
from cgl.core.path.support import (
edit_project_msd,
)
from cgl.core.utils.general import (
cgl_copy,
Logger,
)
reload(msd)
# FOLDER AND FILE CREATION FUNCTIONS
# VERSIONING FUNCTIONS
[docs]
def remap_msd(source_object, dest_object):
"""
Remaps the msd for the pathObject - ensures all paths are going to the correct sync_root/company/project
this is for use after a copy operation when you have to repath the msd file.
Args:
path_object (PathObject): A PathObject instance that represents a path in the file system
Returns:
None
"""
from cgl.core.utils.general import save_json
msd_path = dest_object.get_msd_path()
if os.path.exists(msd_path):
print("Remapping the msd for: {}".format(dest_object.get_msd_path()))
dest_company = dest_object.company
dest_project = dest_object.project
root_string = f"{dest_company}/{dest_project}/PUBLISHES"
dest_dict = dest_object.get_msd_dict()
dirty = False
if dest_dict["render_files"]:
for ext in dest_dict["render_files"]:
file_list = dest_dict["render_files"][ext]
if file_list:
dest_dict["render_files"][ext] = process_msd_file_list(
file_list, root_string
)
dirty = True
if dest_dict["source_files"]:
for ext in dest_dict["source_files"]:
file_list = dest_dict["source_files"][ext]
if file_list:
dest_dict["source_files"][ext] = process_msd_file_list(
file_list, root_string
)
dirty = True
if dest_dict["dependencies"]:
for ext in dest_dict["dependencies"]:
file_list = dest_dict["dependencies"][ext]
if file_list:
dest_dict["dependencies"][ext] = process_msd_file_list(
file_list, root_string
)
dirty = True
if dirty:
save_json(dest_object.get_msd_path(), dest_dict)
[docs]
def process_msd_file_list(file_list, new_root):
"""
processes a list of files, splits each string at "PUBLISHES" with a regex, and then joins the new root with the
second half of the split string.
Args:
file_list:
new_root:
Returns:
"""
new_list = []
for file in file_list:
if "PUBLISHES" not in file:
print(
"Need to do some sophisticated remapping here - and fix this at the publish stage."
)
split = re.split(r"PUBLISHES", file)
new_string = f"{new_root}{split[1]}"
new_list.append(new_string)
return new_list
[docs]
def update_msd(path_object, dcc):
"""Updates the msd for the pathObject"""
create_msd(path_object, dcc=dcc)
edit_project_msd(path_object)
[docs]
def push(source_path, dest_path_object, publish=False):
"""
push a file or folder that is not in magic_browser to a destination pathObject
Args:
source_path (str): source file or folder to push to the destination path object.
dest_path_object (PathObject): path object where you want to push the files. Must be a user pathObject()
publish (bool): publish the resulting pathObject()
Returns (PathObject):
publish pathObject()
"""
# figure out if source_path is a file or folder
render_dir = dest_path_object.get_render_path(dirname=True)
source_dir = dest_path_object.get_source_path(dirname=True)
if os.path.isfile(source_path):
source_folder, source_file = os.path.split(source_path)
# copy the file to the destination
dest_path = os.path.join(render_dir, source_file)
print(source_path, dest_path)
cgl_copy(source_path, dest_path)
if publish:
# TODO - make sure this publishes ALL resolutions
dest_path_object.publish()
else:
source_folder = source_path
print(source_folder, render_dir)
cgl_copy(source_folder, render_dir)
if publish:
# TODO - make sure this publishes ALL resolutions
dest_path_object.publish()
if not os.path.exists(source_dir):
os.makedirs(source_dir)
return dest_path_object
# def pull(path_object):
# """
# pulls the files and dependencies for the path_object from the cloud to your local machine.
# Args:
# path_object:
#
# Returns:
#
# """
# # get the dependencies from the task config
# deps = []
# cfg = ProjectConfig.getLastConfig()
# long_name = cfg.tasks_config["default"]["short_to_long"][path_object.entity_type][
# path_object.task
# ]
# dependencies = cfg.tasks_config["dependencies"][path_object.sequence][long_name]
# for dep in dependencies:
# # get the short name for the dependency
# short_name = cfg.tasks_config["default"]["long_to_short"][
# path_object.entity_type
# ][dep]
# # figure out the latest version of that dependency
# dep_po = path_object.copy(task=short_name, user="publish", latest=True)
# source_path = dep_po.get_source_path()
# render_path = dep_po.get_render_path()
# if os.path.exists(source_path):
# deps.append(source_path)
# deps.append(render_path)
# else:
# logging.error(f"DEPENDENCY NOT FOUND: {source_path} does not exist")
# # TODO - clean up the "tasks" stuff from the shotgrid config now that we have a task config
# # TODO - clean up the dependency stuff from the bid config now that we have a task config.
# return deps
#
# def remote_publish(path_object: PathObject):
# from cgl.core.cloud.auth import Cognito
# from cgl.core.cloud.constants import cloud_constants
# from cgl.core.cloud.support import generate_unique_hash
#
# cc = ProjectConfig.getLastConfig().cloud_config
# if "remote_publish" not in cc:
# logging.error(
# "No cloud publish is configured"
# ) # remote publish is not enabled.
# return
#
# # print(Cognito().is_authenticated())
# bucket = Cognito.get_s3_bucket(cloud_constants["remote_publish_s3"])
# q = Cognito.get_sqs_client()
# source_path = path_object.get_path()
# render_path = source_path.replace(
# "source", "render"
# ) # tried doing a copy(tree='render), didnt work
# prefix = generate_unique_hash()
# project_path = path_object.get_project_path()
#
# for path in [source_path, render_path]:
# if os.path.isfile(path):
# # convertin gpath to directory
# path = os.path.dirname(path)
#
# logging.info(f"Uploading {prefix}:{path}!")
# print(f"Uploading {prefix}:{path}!")
# bucket.upload_folder(path, prefix)
# logging.info(f"Uploaded")
# print(f"Uploaded")
# msg = f"REMOTEPUBLISH|{prefix}|{path}|{project_path}"
# logging.info(f"mailing head office | {msg} |!")
# print(f"mailing head office | {msg} |!")
#
# q.publish_message(cloud_constants["remote_publish_sqs"], msg, prefix)
#
# logging.info(f"sent!")
# print(f"sent!")
def _test():
# TODO - This should be moved to a test file, maybe?
from PySide6.QtWidgets import QApplication
import sys
from cgl.core.utils.general import load_style_sheet
app = QApplication(sys.argv)
app.log = Logger(log_level=logging.INFO)
load_style_sheet()
path_ = r"D:\alchemy\premise\render\JFATEST\shots\53\0000\amc\default\\"
path_object = PathObject().from_path_string(path_)
print(path_object)
remote_publish(path_object)
# no = version_up(path_object, version_type='minor', default_file=True)
[docs]
def push_test():
res_source = r"Y:\prod\premise\source\ASJ\assets\char\maleLargeTest\rig\data\stage"
source_file = ""
# create the destination object based off the version i want to go to.
dest_string = (
r"Y:\prod\premise\source\ASJ\assets\char\jesus\rig\default\tom.m\000.006\high"
)
dest_dict = {
"project": "ASJ",
"entity_type": "assets",
"sequence": "char",
"shot": "maleLargeTest",
"task": "rig",
"variant": "default",
"user": "tom.m",
"resolution": "high",
}
dest_obj = PathObject().from_path_string(dest_string)
for file in os.listdir(res_source):
file_, ext = os.path.splitext(file)
try:
source_file = os.path.join(res_source, file)
# figure out what the resolution is
asset, variant, resolution, other = file_.split("_")
# create a path object with the correct resolution
res_do = dest_obj.copy(resolution=resolution)
# use the "push" function to push the file to the correct resolution
# for this test i'm not going to publish the file, just put it in the user area.
push(source_file, dest_path_object=res_do)
except ValueError:
# i didn't know what to do with some of the patterns.
print(file_, "does not have the right pattern")
print("publishing")
dest_obj.publish()
# dest_path_object = PathObject().from_path_string(r'')
[docs]
def pull_test():
path_ = (
r"Y:\prod\premise\source\ASJ\assets\char\jesus\rig\default\tom.m\000.006\high"
)
path_object = PathObject().from_path_string(path_)
pull(path_object)