refactor: separate code into separate files, add docstrings
This commit is contained in:
@@ -1,21 +1,13 @@
|
|||||||
import argparse
|
import argparse
|
||||||
import os.path
|
from .utils import AnsiCodes
|
||||||
import sys
|
from .validation import validate_paths_from_file, validate_paths_from_list
|
||||||
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"
|
|
||||||
|
|
||||||
def parse_args() -> argparse.Namespace:
|
def parse_args() -> argparse.Namespace:
|
||||||
"""
|
"""
|
||||||
Parses cli arguments.
|
Parses cli arguments.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
argparse.Namespace: Parsed argument namespace
|
argparse.Namespace: Parsed argument namespace.
|
||||||
"""
|
"""
|
||||||
|
|
||||||
parser = argparse.ArgumentParser(
|
parser = argparse.ArgumentParser(
|
||||||
@@ -29,176 +21,22 @@ def parse_args() -> argparse.Namespace:
|
|||||||
|
|
||||||
return parser.parse_args()
|
return parser.parse_args()
|
||||||
|
|
||||||
def split_path(path):
|
def main() -> None:
|
||||||
PATH_DELIMETER = os.path.sep
|
parsed_args = parse_args()
|
||||||
|
|
||||||
split_path = path.split(PATH_DELIMETER)
|
if(parsed_args.paths_file):
|
||||||
|
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. '{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:")
|
||||||
|
|
||||||
# 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<full_path>/shows/(?P<project_name>\S*?)/\S*?/(?P<asset_type>\S*?)/(?P<asset_name>\S*?)_v(?P<asset_version>\S*?)/(?P<task>\S*?)/\S*?/(?P<file_name>\S*?)_v(?P<file_version>\d*)\D*?(?P<frame_number>\d*)\D*?\.(?P<file_extension>\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 = [
|
static_list = [
|
||||||
"/shows/projectX/assets/environment/forest_v001/model/render/forest_beauty_v001.1002.exr",
|
"/shows/projectX/assets/environment/forest_v001/model/render/forest_beauty_v001.1002.exr",
|
||||||
"/shows/projectX/assets/environment/forest_v001/model/cache/forest_v001.fbx"
|
"/shows/projectX/assets/environment/forest_v001/model/cache/forest_v001.fbx"
|
||||||
]
|
]
|
||||||
|
validate_paths_from_list(parsed_args, static_list)
|
||||||
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)
|
|
||||||
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("\nAlternatively, here is the validation of a hardcoded list of paths:")
|
|
||||||
|
|
||||||
validate_static_list(parsed_args)
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
53
src/bb_asset_validation/utils.py
Normal file
53
src/bb_asset_validation/utils.py
Normal file
@@ -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<full_path>/shows/(?P<project_name>\S*?)/\S*?/(?P<asset_type>\S*?)/(?P<asset_name>\S*?)_v(?P<asset_version>\S*?)/(?P<task>\S*?)/\S*?/(?P<file_name>\S*?)_v(?P<file_version>\d*)\D*?(?P<frame_number>\d*)\D*?\.(?P<file_extension>\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"
|
||||||
195
src/bb_asset_validation/validation.py
Normal file
195
src/bb_asset_validation/validation.py
Normal file
@@ -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
|
||||||
Reference in New Issue
Block a user