import logging
import os
import socket
from cgl.core.config.query import AlchemyConfigManager
from cgl.core.utils.general import cgl_execute
from cgl.plugins.perforce.utils.config import get_perforce_user_name
from cgl.plugins.perforce.utils.depot import get_depot_name
from pathlib import Path
CFG = AlchemyConfigManager()
ALC_PERFORCE_ROOT = CFG.company_config.get("perforce_root", "")
PERFORCE_SERVER = CFG.get_perforce_server()
[docs]
def create_workspace(company, project, location):
"""
Create a workspace for the current perforce user
Args:
depot_name: name of the depot that this workspace is mapped to
workspace_name: Name of the workspace to be created
company:
project:
Return:
The path to the workspace directory on disk
"""
# TODO - need a switch whether there's a perforce server or not.
computer_name = socket.gethostname()
workspace_name = get_workspace_name(company, project, location)
user_name = get_perforce_user_name()
perforce_workspace_path = get_workspace_path(company, project, location)
depot_name = get_depot_name(company, project, location)
if not os.path.exists(perforce_workspace_path):
os.makedirs(perforce_workspace_path)
command = (
"(echo Client: {} & echo Root: {} & echo Owner: {} & echo Host: {} & echo Options: "
"noallwrite clobber nocompress unlocked nomodtime normdir & echo SubmitOptions: "
"submitunchanged & echo LineEnd: local & echo View: //{}/... //{}/...) | p4 client -i".format(
workspace_name,
perforce_workspace_path,
user_name,
computer_name,
depot_name,
workspace_name,
)
)
os.system(command)
switch_to_workspace(company, project, location)
if workspace_exists(company, project, location):
return True
else:
return False
[docs]
def get_workspace_path(company, project, world):
"""
Get the path to the current workspace
Args:
project: Name of the project
"""
# TODO this should be done with path object.
perforce_root = Path(CFG.company_config.get("perforce_root", ""))
workspace_path = perforce_root / company / project / "PERFORCE" / world
return workspace_path
[docs]
def get_workspace_name(company, project, world):
computer_name = socket.gethostname()
return "WS_{}_{}_{}_{}".format(company, project, world, computer_name)
[docs]
def workspace_exists(company, project, world):
"""
Checks if the workspace exists
Args:
workspace_name: Name of the workspace to query
Returns:
True if the workspace exists, False if the workspace doesn't exist
"""
if PERFORCE_SERVER:
workspace_name = get_workspace_name(company, project, world)
workspaces = get_all_workspaces()
if workspace_name in workspaces:
logging.info("FOUND")
return True
else:
logging.info("Workspace doesn't exist")
return False
else:
workspace_name = get_workspace_name(company, project, world)
workspace_path = get_workspace_path(company, project, world)
raise ("workspace_exists()", workspace_path)
[docs]
def delete_workspace(path_object=False, workspace_name=""):
"""
Delete workspace
Args:
path_object: Path object for the current project
workspace_name: Name of workspace to be deleted
Returns:
Output of the p4 delete command
"""
if PERFORCE_SERVER:
if path_object:
perforce_workspace_path = path_object.perforce_workspace
workspace_name = os.path.basename(perforce_workspace_path)
command = "p4 client -d -f {}".format(workspace_name)
output = cgl_execute(command=command, return_output=True, print_output=False)
return output
else:
raise ("No perforce server found delete_workspace()")
[docs]
def switch_to_workspace(company, project, world):
"""
Switches current workspace in perforce
Args:
workspace_name: Name of the workspace to switch to
project:
world:
"""
if PERFORCE_SERVER:
workspace_name = get_workspace_name(company, project, world)
print(f"Switching to {workspace_name}")
command = "p4 set P4CLIENT={}".format(workspace_name)
os.system(command)
cgl_execute(command=command, return_output=True, print_output=False)
print('[switch_to_workspace] needs beefed up so we know if it actually switches or not')
return True
else:
raise ("[switch_to_workspace] No perforce server found.")
[docs]
def get_all_workspaces():
"""
Gets a list of all workspaces registered in perforce
Returns:
List of workspace names
"""
if PERFORCE_SERVER:
command = "p4 clients"
output = cgl_execute(command=command)
workspaces = []
output_lines = output["stdout"].splitlines()
for output_line in output_lines:
workspaces.append(output_line.split(" ")[1])
return workspaces
else:
raise ("No perforce server found get_all_workspaces()")
[docs]
def get_current_workspace():
"""
Gets the current workspace in perforce
Returns:
Name of the current workspace
"""
if PERFORCE_SERVER:
command = 'p4 set | findstr "P4CLIENT"'
output = cgl_execute(command=command, return_output=True, print_output=False)
result_string = output["stdout"].splitlines()[0].split(" ")[0]
current_workspace = result_string.split("=")[1]
logging.info(current_workspace)
return current_workspace
else:
raise ("No perforce server found get_current_workspace()")
if __name__ == "__main__":
# create_workspace("CGL", "arc", "test")
# print(get_workspace_path("CGL", "JLT", "assets"))
# delete_workspace(path_object=None, workspace_name="scienceStation")
print(get_workspace_path("jhcs", "ttas", "assetLibrary"))
print(get_all_workspaces())