import os import re import docker import logging import datetime from .version import BlenderVersion from .dependency import BlenderDependency from pathlib import Path from docker.errors import BuildError, APIError, ContainerError, ImageNotFound _ANSI_ESCAPE_RE = re.compile(r'\x1b\[[0-9;]*[a-zA-Z]') def _clean_log_line(text: str) -> str: """Strip ANSI escape codes and keep only the last update of a \r-rewritten line.""" text = _ANSI_ESCAPE_RE.sub('', text) return text.split('\r')[-1].strip() class BlenderDocker: def __init__(self, blender: BlenderVersion, severity: int = logging.WARNING): self._blender_version = blender # Todo get a user config self._docker_client = docker.DockerClient(base_url=f'unix:///var/run/docker.sock') self.login() self._docker_image = None self._repository = 'stilobique/blender' self._docker_image_name = self._blender_version.format_to_tag(self._repository) self._docker_file = Path(os.getcwd(), 'Dockerfile') if not self._docker_file.exists(): print(f'Dockerfile not found: {self._docker_file}') exit(1) self._packages = BlenderDependency(blender) self._args = { 'b3d_vs_major': f'{self._blender_version.major}', 'b3d_vs_minor': f'{self._blender_version.minor}.{self._blender_version.hotfix}', 'b3d_dependency': self._packages.__str__() } self._severity = severity def _get_logger(self, name: str) -> logging.Logger: """ Private Method to set a logger for a specific method set (by his name) :param name: Make the logger file name. :return: """ log_dir = Path('logs') log_dir.mkdir(parents=True, exist_ok=True) logger = logging.getLogger(f'blender_docker.{name}') logger.setLevel(self._severity) logger.propagate = False if not logger.handlers: handler = logging.FileHandler(log_dir / f'{name}.log') handler.setFormatter(logging.Formatter( '%(asctime)s | [%(levelname)s] %(message)s' )) logger.addHandler(handler) return logger def login(self): """Login to the Docker Hub""" try: print('Login to Hub Docker.') self._docker_client.login( username=os.environ['DOCKER_HUB_USER'], password=os.environ['DOCKER_HUB_PASSWORD'] ) except APIError as e: print(f'Login failed with error: {e}') exit(1) def build(self): """Build the Blender Docker Image""" time = datetime.datetime.now() logger = self._get_logger(f'build-{time.strftime("%Y-%m-%d-%H-%M")}') collected_logs = [] try: print('Building Blender Docker Image') logger.info('Building Blender Docker Image') logger.info(f'\tBuilds withs args: {self._args}') logger.info(f'\tTag generate: {self._docker_image_name}') # Use the low-level API: images.build() only returns once the whole # build is finished, so it gives no live progress. api.build() with # decode=True streams each step (including cache hits) as it happens. build_logs = self._docker_client.api.build( path=os.getcwd(), # nocache=True, buildargs=self._args, tag=self._docker_image_name, rm=True, forcerm=True, decode=True, ) for chunk in build_logs: collected_logs.append(chunk) if 'stream' in chunk: for raw_line in chunk['stream'].split('\n'): line = _clean_log_line(raw_line) if not line: continue if 'Using cache' in line: line = f'[CACHE] {line}' logger.info(line) elif 'status' in chunk: status = _clean_log_line(chunk['status']) progress = _clean_log_line(chunk.get('progress', '')) line = f"{status} {chunk.get('id', '')} {progress}".strip() if 'Already exists' in status: line = f'[CACHE] {line}' logger.info(line) elif 'error' in chunk: logger.error(chunk['error']) raise BuildError(chunk['error'], collected_logs) self._docker_image = self._docker_client.images.get(self._docker_image_name) except BuildError as e: logger.exception(f'Build docker image failed: {e}') raise @staticmethod def tag(self): print('\tSet Tags version and/or Latest') pass def test(self) -> bool: """ This method request an image (call the method build before) and try to execute the blender inside the container to check if the program can be correctly executed. Based on `docker run --rm stilobique/blender:5.1.2 ./blender --version` :return: True if no issue are returned """ logger = self._get_logger('test') container_name = f"Blender-Docker-Testing-{self._blender_version.tag_name_slugify()}" command = ['./blender', '--version'] try: print(f'Testing your Blender Docker image {self._docker_image_name}') logger.info(f'Testing image "{self._docker_image_name}" in container "{container_name}" ' f'with command: {" ".join(command)}') test = self._docker_client.containers.run( name=container_name, image=self._docker_image_name, command=command, auto_remove=True, remove=True, detach=True, stdout=True, stderr=True, stream=True, ) result = test.wait() exit_code = result.get('StatusCode') container_logs = test.logs(stdout=True, stderr=True).decode('utf-8', errors='replace').strip() for raw_line in container_logs.split('\n'): line = _clean_log_line(raw_line) if line: logger.info(f'[container] {line}') if exit_code == 0: logger.info(f'Test succeeded (exit code {exit_code}) for image "{self._docker_image_name}".') return True logger.error(f'Test failed (exit code {exit_code}) for image "{self._docker_image_name}".') print(f'Testing your Blender Docker image {self._docker_image_name} failed.') return False except ContainerError as e: logger.exception(f'Container error while testing "{self._docker_image_name}": {e}') print(f'Container error: {e}') exit(1) except ImageNotFound as e: logger.exception(f'Image "{self._docker_image_name}" not found: {e}') print(f'Image not found: {e}') exit(1) except APIError as e: logger.exception(f'API error while testing "{self._docker_image_name}": {e}') print(f'API error: {e}') for line in e.build_log: if 'stream' in line: print(line['stream'].strip()) exit(1) def push(self): self._docker_client.images.push( repository=self._repository, tag=self._blender_version.tag_name(), stream=True, )