import logging from pathlib import Path from typing import Iterable, NamedTuple from .common.command import CommandExecutor __all__ = ( "TaskKey", "Entry", "EntryDetector", "EntryResolver", ) logger = logging.getLogger("entry") class TaskKey(NamedTuple): name: str tag: str def __repr__(self) -> str: return f"{self.name}:{self.tag}" class Entry: _logger = logger def __init__( self, tasks: dict[TaskKey, list[str]], basepath: Path, ): self._tasks = tasks self._basepath = basepath def show(self): self._logger.info("%s", self) for task_name, commands in self._tasks.items(): self._logger.info(" %s", task_name) for command in commands: self._logger.debug(" %s", command) def run_task(self, task_name: str, tag: str | None = None) -> bool: if task_name == "show": self.show() return True runned = False for task_key, commands in self._tasks.items(): if task_key.name == task_name and (tag is None or tag == task_key.tag): self._logger.info("%s %s", task_key, self) runned = True for command in commands: CommandExecutor(command).run(self._basepath) return runned def __repr__(self): return f"{str(self._basepath)}" class EntryDetector: def detect_entry(self, basepath: Path) -> Entry | None: raise NotImplementedError 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)