import logging from pathlib import Path from typing import Dict, Iterable, List, Optional from .command import CommandExecutor __all__ = ( "EntryTaskNotImplementedError", "Entry", "EntryDetector", "EntryResolver", ) logger = logging.getLogger("entry") class EntryTaskNotImplementedError(NotImplementedError): pass class Entry: _logger = logger def __init__( self, tasks: Dict[str, List[str]], basepath: Path, ): self._tasks = tasks self._basepath = basepath def show(self): self._logger.info(f"{self}") for task_name, commands in self._tasks.items(): self._logger.info(f" {task_name}") for command in commands: self._logger.debug(f" {command}") def run_task(self, task_name: str): if task_name == "show": self.show() return if task_name not in self._tasks: raise EntryTaskNotImplementedError(task_name) self._logger.info(f"{task_name} {self}") for command in self._tasks[task_name]: CommandExecutor(command).run(self._basepath) def __repr__(self): return f"{str(self._basepath)}" class EntryDetector: def detect_entry(self, basepath: Path) -> Optional["Entry"]: raise NotImplemented class EntryResolver: def __init__(self, detector: EntryDetector, ignore: List[str]): self._detector = detector self._ignore = ignore def resolve(self, basepath: Path, depth: int = 128) -> Iterable[Entry]: if basepath.is_dir(): if basepath.name in self._ignore: return entry = self._detector.detect_entry(basepath) if entry: yield entry if depth > 0 and basepath.is_dir(): for path in basepath.iterdir(): yield from self.resolve(path, depth - 1)