From cf2c1fb070de93bca0f57e251cf8dd08362b9e02 Mon Sep 17 00:00:00 2001 From: parker Date: Sun, 5 Oct 2025 16:51:03 +0100 Subject: [PATCH] refactor: separate code into separate files, add docstrings --- src/bb_asset_validation/cli.py | 182 ++---------------------- src/bb_asset_validation/utils.py | 53 +++++++ src/bb_asset_validation/validation.py | 195 ++++++++++++++++++++++++++ 3 files changed, 258 insertions(+), 172 deletions(-) create mode 100644 src/bb_asset_validation/utils.py create mode 100644 src/bb_asset_validation/validation.py diff --git a/src/bb_asset_validation/cli.py b/src/bb_asset_validation/cli.py index b282b12..18e9e87 100644 --- a/src/bb_asset_validation/cli.py +++ b/src/bb_asset_validation/cli.py @@ -1,21 +1,13 @@ import argparse -import os.path -import sys -import re - -# Global Variables. In production code I would not use global variables - -# ANSI color escape codes, will not work on windows. -ANSI_FORE_RED = "\033[31m" -ANSI_FORE_GREEN = "\033[32m" -ANSI_DEFAULT = "\033[0m" +from .utils import AnsiCodes +from .validation import validate_paths_from_file, validate_paths_from_list def parse_args() -> argparse.Namespace: """ Parses cli arguments. Returns: - argparse.Namespace: Parsed argument namespace + argparse.Namespace: Parsed argument namespace. """ parser = argparse.ArgumentParser( @@ -29,176 +21,22 @@ def parse_args() -> argparse.Namespace: return parser.parse_args() -def split_path(path): - PATH_DELIMETER = os.path.sep - - split_path = path.split(PATH_DELIMETER) - - # remove empty values - split_path = [i for i in split_path if i.strip()!=""] - - return split_path - -def get_path_components(asset_path: str): - regex_pattern = r"(?P/shows/(?P\S*?)/\S*?/(?P\S*?)/(?P\S*?)_v(?P\S*?)/(?P\S*?)/\S*?/(?P\S*?)_v(?P\d*)\D*?(?P\d*)\D*?\.(?P\w*)\Z)" - pattern = re.compile(regex_pattern) - match = pattern.match(asset_path) - - if(not match): - raise Exception("Regex pattern matching failed. This is likely due to one of your paths using the incorrect schema.") - - path_components = match.groupdict() - - return path_components - -def assemble_new_path(path_components: dict) -> str: - frame_number = f".{path_components['frame_number']}" if path_components['frame_number'] else "" - file_extension = f".{path_components['file_extension']}" if path_components['file_extension'] else "" - return f"/shows/{path_components['project_name']}/staging/delivery/assets/{path_components['asset_type']}_{path_components['asset_name']}/{path_components['task']}_{path_components['asset_version']}/{path_components['file_name']}{frame_number}{file_extension}" - -def should_skip_line(line): - line = line.strip() - - return ( - len(line)<=1 or # too short - line[0]=="#" # comment - ) - - -def validate_paths_file(parsed_args): - with open(parsed_args.paths_file, "r") as f: - line = f.readline() - while(line): - line = f.readline() - - if(should_skip_line(line)): - continue - # clean line - path = line.strip() - - # validate path - if(not validate_path(parsed_args, path)): - continue - - # all components of the project - all_components = get_path_components(path) - - # map the specific components listed in the brief - stored_components = map_stored_components(all_components) - - # assemble the new path - new_path = assemble_new_path(all_components) - - # print out the resulting values - print_result(path, new_path, stored_components) - -def validate_static_list(parsed_args): - static_list = [ - "/shows/projectX/assets/environment/forest_v001/model/render/forest_beauty_v001.1002.exr", - "/shows/projectX/assets/environment/forest_v001/model/cache/forest_v001.fbx" - ] - - for line in static_list: - if(should_skip_line(line)): - continue - # clean line - path = line.strip() - - # validate path - if(not validate_path(parsed_args, path)): - continue - - # all components of the project - all_components = get_path_components(path) - - # map the specific components listed in the brief - stored_components = map_stored_components(all_components) - - # assemble the new path - new_path = assemble_new_path(all_components) - - # print out the resulting values - print_result(path, new_path, stored_components) - -def map_stored_components(all_components): - stored_components = { - "full_path" : all_components["full_path"], - "project" : all_components["project_name"], - "asset_type" : all_components["asset_type"], - "asset_name" : all_components["asset_name"], - "version" : all_components["asset_version"], - "task" : all_components["task"], - "file_extension" :all_components["file_extension"], - } - return stored_components - -def print_result(original_path, new_path, stored_components): - print(f"{ANSI_FORE_GREEN}\n--------------") - print(f"original path:\n\t{original_path}\n") - print(f"new path:\n\t{new_path}\n") - print(f"stored components:\n\t{stored_components}") - print(f"--------------\n{ANSI_DEFAULT}") - -# def validate_static_list(): - - -def throw_schema_error(path: str, error_message: str) -> None: - print(f"\n{ANSI_FORE_RED}-----\nPath is invalid: {path}\n{error_message}\n-----{ANSI_DEFAULT}", file=sys.stderr) - -def validate_path(parsed_args: argparse.Namespace, asset_path: str) -> bool: - """ - Validates the given path based on the arguments provided. - - Parameters: - parsed_args (argparse.Namespace): CLI arguments - asset_path (str): The path to validate - - Returns: - Whether the provided path is valid - """ - # check path is absolute - if(asset_path[0] != "/"): - throw_schema_error(asset_path, "Expected absolute asset path, recieved path does not start with '/'") - return False - - # check length - EXPECTED_LENGTH=8 - split_asset_path = split_path(asset_path) - if(EXPECTED_LENGTH is not None and len(split_asset_path) != EXPECTED_LENGTH): - throw_schema_error(asset_path, f"Path isn't expected length {EXPECTED_LENGTH}.") - return False - - - # check path is prefixed with shows path - shows_path = os.path.realpath(parsed_args.shows_directory) - valid_show_path = shows_path == asset_path[:len(shows_path)] - if(not valid_show_path): - throw_schema_error(asset_path, f"Path does not start with expected show directory '{parsed_args.shows_directory}'") - return False - - # get components - path_components = get_path_components(asset_path) - - # validate versions - if(path_components["asset_version"] != path_components["file_version"]): - throw_schema_error(asset_path, "Asset version and file version does not match.") - return False - - return True - def main() -> None: - # Parse args parsed_args = parse_args() if(parsed_args.paths_file): - validate_paths_file(parsed_args) + validate_paths_from_file(parsed_args, parsed_args.paths_file) else: print("No file provided.\n") print("Try passing one of the example files with the --paths-file argument.") - print(f"eg. '{ANSI_FORE_GREEN}bb-asset-validation --paths-file example-paths-mixed.txt{ANSI_DEFAULT}'") + print(f"eg. '{AnsiCodes.FORE_GREEN.value}bb-asset-validation --paths-file example-paths-mixed.txt{AnsiCodes.DEFAULT.value}'") print("\nAlternatively, here is the validation of a hardcoded list of paths:") - validate_static_list(parsed_args) + static_list = [ + "/shows/projectX/assets/environment/forest_v001/model/render/forest_beauty_v001.1002.exr", + "/shows/projectX/assets/environment/forest_v001/model/cache/forest_v001.fbx" + ] + validate_paths_from_list(parsed_args, static_list) diff --git a/src/bb_asset_validation/utils.py b/src/bb_asset_validation/utils.py new file mode 100644 index 0000000..801e1d7 --- /dev/null +++ b/src/bb_asset_validation/utils.py @@ -0,0 +1,53 @@ +from enum import Enum +import re +import os + +def split_path(path: str) -> list: + """ + Splits given path into subdirectories. + + Parameters: + path (str): Whole path. + + Returns: + List: List of subdirectories. + """ + + PATH_DELIMETER = os.path.sep + + split_path = path.split(PATH_DELIMETER) + + # remove empty values + split_path = [i for i in split_path if i.strip()!=""] + + return split_path + +def get_path_components(asset_path: str) -> dict: + """ + Identifies important components of the given path using a hardcoded schema. + + Parameters: + asset_path (str): The path to derive components from. + + Returns: + dict: Name mapped components. + """ + regex_pattern = r"(?P/shows/(?P\S*?)/\S*?/(?P\S*?)/(?P\S*?)_v(?P\S*?)/(?P\S*?)/\S*?/(?P\S*?)_v(?P\d*)\D*?(?P\d*)\D*?\.(?P\w*)\Z)" + pattern = re.compile(regex_pattern) + match = pattern.match(asset_path) + + if(match is None): + raise Exception("Regex pattern matching failed. This is likely due to one of your paths using the incorrect schema.") + + path_components = match.groupdict() + + return path_components + +class AnsiCodes(Enum): + """ + Ansi color codes for styling text. + """ + + FORE_RED = "\033[31m" + FORE_GREEN = "\033[32m" + DEFAULT = "\033[0m" diff --git a/src/bb_asset_validation/validation.py b/src/bb_asset_validation/validation.py new file mode 100644 index 0000000..2abccae --- /dev/null +++ b/src/bb_asset_validation/validation.py @@ -0,0 +1,195 @@ +from .utils import AnsiCodes, split_path, get_path_components +import argparse +import os.path +import sys + +def assemble_new_path(path_components: dict) -> str: + """ + Assembles new path based on the brief. + + Parameters: + path_components (dict): Separated components of the original path representing different values of the asset such as type and file extension. + + Returns: + str: New path. + + """ + frame_number = f".{path_components['frame_number']}" if path_components['frame_number'] else "" + file_extension = f".{path_components['file_extension']}" if path_components['file_extension'] else "" + return f"/shows/{path_components['project_name']}/staging/delivery/assets/{path_components['asset_type']}_{path_components['asset_name']}/{path_components['task']}_{path_components['asset_version']}/{path_components['file_name']}{frame_number}{file_extension}" + +def should_skip_line(line: str) -> bool: + """ + Evaluates whether the lines should be skipped from validation. + + Parameters: + line (str): Line of file to check. + + Returns: + bool: Whether the line should be skipped. + """ + + line = line.strip() + + return ( + len(line)<=1 or # ignore lines that are too short + line[0]=="#" # ignore comments + ) + + +def validate_paths_from_file(parsed_args: argparse.Namespace, file: str) -> None: + """ + Validates all paths in the given text file and print results. + + Parameters: + parsed_args (argparse.Namespace): Command line arguments that influence behavior. + file (str): File path containing paths for validation on each line. + """ + + with open(file, "r") as f: + line = f.readline() + while(line): + line = f.readline() + + if(should_skip_line(line)): + continue + # clean line + path = line.strip() + + # validate path + if(not validate_path(parsed_args, path)): + continue + + # all components of the project + all_components = get_path_components(path) + + # map the specific components listed in the brief + stored_components = map_stored_components(all_components) + + # assemble the new path + new_path = assemble_new_path(all_components) + + # print out the resulting values + print_result(path, new_path, stored_components) + +def validate_paths_from_list(parsed_args: argparse.Namespace, path_list:list) -> None: + """ + Validates all paths in the given list and print results. + + Parameters: + parsed_args (argparse.Namespace): Command line arguments that influence behavior. + path_list (list): List containing paths for validation. + """ + + for line in path_list: + if(should_skip_line(line)): + continue + # clean line + path = line.strip() + + # validate path + if(not validate_path(parsed_args, path)): + continue + + # all components of the project + all_components = get_path_components(path) + + # map the specific components listed in the brief + stored_components = map_stored_components(all_components) + + # assemble the new path + new_path = assemble_new_path(all_components) + + # print out the resulting values + print_result(path, new_path, stored_components) + +def map_stored_components(all_components: dict) -> dict: + """ + Maps all path components to the specific components requested by the brief. + + Parameters: + all_components (dict): Unfiltered components. + + Returns: + dict: Filtered components. + + """ + stored_components = { + "full_path" : all_components["full_path"], + "project" : all_components["project_name"], + "asset_type" : all_components["asset_type"], + "asset_name" : all_components["asset_name"], + "version" : all_components["asset_version"], + "task" : all_components["task"], + "file_extension" :all_components["file_extension"], + } + return stored_components + +def print_result(original_path: str, new_path: str, stored_components: dict) -> None: + """ + Prints the test results based on the brief. + + Parameters: + original_path (str): The user inputted path. + new_path (str): Path that's been transformed based on the brief. + stored_components (dict): components of the original path split into a dictionary. + """ + + print(f"{AnsiCodes.FORE_GREEN.value}\n--------------") + print(f"original path:\n\t{original_path}\n") + print(f"new path:\n\t{new_path}\n") + print(f"stored components:\n\t{stored_components}") + print(f"--------------\n{AnsiCodes.DEFAULT.value}") + +def throw_schema_error(path: str, error_message: str) -> None: + """ + Throws an error for incorrect path schemas. + + Parameters: + path (str): Path that failed schema validation. + error_message (str): Message stating the failure. + """ + + print(f"\n{AnsiCodes.FORE_RED.value}-----\nPath is invalid: {path}\n{error_message}\n-----{AnsiCodes.DEFAULT.value}", file=sys.stderr) + +def validate_path(parsed_args: argparse.Namespace, asset_path: str) -> bool: + """ + Validates the given path based on the arguments provided. + + Parameters: + parsed_args (argparse.Namespace): CLI arguments. + asset_path (str): The path to validate. + + Returns: + bool: Whether the provided path is valid. + """ + + # check path is absolute + if(asset_path[0] != "/"): + throw_schema_error(asset_path, "Expected absolute asset path, recieved path does not start with '/'") + return False + + # check length + EXPECTED_LENGTH=8 + split_asset_path = split_path(asset_path) + if(EXPECTED_LENGTH is not None and len(split_asset_path) != EXPECTED_LENGTH): + throw_schema_error(asset_path, f"Path isn't expected length {EXPECTED_LENGTH}.") + return False + + + # check path is prefixed with shows path + shows_path = os.path.realpath(parsed_args.shows_directory) + valid_show_path = shows_path == asset_path[:len(shows_path)] + if(not valid_show_path): + throw_schema_error(asset_path, f"Path does not start with expected show directory '{parsed_args.shows_directory}'") + return False + + # get components + path_components = get_path_components(asset_path) + + # validate versions + if(path_components["asset_version"] != path_components["file_version"]): + throw_schema_error(asset_path, "Asset version and file version does not match.") + return False + + return True