73 lines
1.9 KiB
Python
73 lines
1.9 KiB
Python
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)
|