import logging
import os
from textwrap import dedent
from cgl.core.utils.general import cgl_execute
from cgl.plugins.perforce.utils.git_operations import push, commit
from cgl.core.config.query import AlchemyConfigManager
from typing import List
import re
from zipfile import ZipFile
CFG = AlchemyConfigManager()
PERFORCE_SERVER = CFG.get_perforce_server()
[docs]
def get_depot_name(company, project, world):
"""
Args:
company:
project:
world:
Returns:
"""
depot_name = f"DP_{company}_{project}_{world}"
return depot_name
[docs]
def create_depot(company, project, world="world"):
"""
Creates a new depot in perforce
Args:
project: Name of the project
world: Name of the world
depot_name: Name of the new depot
Returns:
True if depot is created, False if depot is not created
"""
depot_name = get_depot_name(company, project, world)
command = (
"(echo Depot: {} & echo Owner: {} & echo Type: local & echo Map: {}/...) | p4 depot -i".format(
depot_name, "tmakota", depot_name
)
)
os.system(command)
depot_name = depot_exists(company, project, world)
if depot_name:
return depot_name
else:
logging.error(f"[create_depot] Depot not created for {world}")
return None
[docs]
def depot_exists(company, project, world):
"""
Checks to see if a specific depot exists in perforce name
Args:
depot_name: Name of depot to query
Args:
True if depot exists, False if depot does not exist
"""
depot_name = get_depot_name(company, project, world)
for depot in get_depots():
if depot_name == depot:
return depot
return False
[docs]
def delete_depot(company, project, world):
"""
Deletes a depot in perforce
Args:
depot_name: Name of depot to be deleted
project:
world:
Return:
False if depot does not exist, returns the output of the p4 delete command if the depot does exist
"""
depot_name = get_depot_name(company=company, project=project, world=world)
# depot_name = "AlchemyTestB_world"
command = "p4 obliterate -y //{}/...".format(depot_name)
output = cgl_execute(command=command)
print("[delete_depot]", output)
command = "p4 depot -d -f {}".format(depot_name)
output = cgl_execute(command=command)
print("[delete_depot]",output)
return output
[docs]
def get_depots():
"""
Gets a list of all depot names registered in Perforce.
Returns:
list[str]: all depot names (e.g. ["DP_jhcs_ttas_test", "DP_dev_xyz"])
"""
if not PERFORCE_SERVER:
print("No Perforce Server Configured")
return []
command = ["p4", "depots"]
result = cgl_execute(command=command, return_output=True, print_output=False)
stdout = result.get("stdout", "")
print(stdout)
depots = []
# Example line:
# Depot DP_jhcs_ttas_test 2025/09/18 local DP_jhcs_ttas_test/... ''
pattern = re.compile(r"^Depot\s+(\S+)\s")
for line in stdout.splitlines():
m = pattern.match(line)
if m:
depots.append(m.group(1))
logging.info(f"Found depots: {depots}")
return depots
[docs]
def get_depots_for_project(company: str, project: str) -> List[str]:
"""
Return list of depot names matching pattern:
Depot <depotname> ... where depotname follows DP_{company}_{project}_{location}
"""
if not project:
return []
depot_line_pattern = re.compile(r"^Depot\s+(\S+)\s+\S+\s+(?:local|stream|remote|archive|spec)\s+(\S+)\s+.*$")
name_pattern = re.compile(rf"^DP_{re.escape(company)}_{re.escape(project)}_(.+)$")
depots = get_depots() # assuming this returns list of depot names (just names, not full lines)
print("depots", depots)
found = []
for depot_name in depots:
print(depot_name)
# Alternatively if you somehow have the full depot spec line, you could filter there first
m = name_pattern.match(depot_name)
if m:
found.append(depot_name)
return found
[docs]
def get_world_from_depot_name(project, depot_name):
world_name = depot_name.split(f"{project}_")[-1]
return world_name
[docs]
def get_shotgrid_locations(project="ASJ"):
import cgl.plugins.shotgrid.query as sg_query
assets = sg_query.assets(project)
sg_assets = []
for ass in assets:
if ass["sg_asset_type"] == "Environment":
location = ass["code"]
# if depot does not exist add to sg_assets
sg_assets.append(location)
return sg_assets
[docs]
def initialize_depot(company, project, location):
"""
Create a workspace locally, then create a depot and sync the local workspace to the depot
"""
from cgl.plugins.perforce.utils.workspace import (
workspace_exists,
create_workspace,
get_workspace_path,
switch_to_workspace,
)
workspace_path = get_workspace_path(company, project, location)
if PERFORCE_SERVER:
if not depot_exists(company, project, location):
create_depot(company, project, location)
else:
print(f"Depot DP_{company}_{project}_{location} Already Exists")
return
if not workspace_exists(company, project, location):
create_workspace(company, project, location)
# at this stage we are certain that the depot and the workspace exist.
initialize_new_location(company, project, location)
return workspace_path
[docs]
def initialize_new_location(company, project, location):
"""
Args:
company:
project:
location:
Returns:
"""
from cgl.plugins.perforce.utils.workspace import (
workspace_exists,
get_workspace_path,
switch_to_workspace,
)
from cgl.plugins.ue.create import create_alchemy_location
workspace_path = get_workspace_path(company, project, location)
if not workspace_exists(company, project, location):
logging.error(f"[initialize_new_location] Workspace does not exist to create new location {location}")
return
switch_to_workspace(company, project, location)
if os.path.exists(workspace_path):
contents = os.listdir(workspace_path)
if contents:
logging.error(f"[initialize_new_location] Workspace contents already exist, no need to create new location: {location}")
create_alchemy_location(location)
commit(company, project, location)
push()