feat: use .gitignore
This commit is contained in:
@@ -0,0 +1,8 @@
|
||||
from pathlib import Path
|
||||
|
||||
from .entry import Entry
|
||||
|
||||
|
||||
class EntryDetector:
|
||||
def detect_entry(self, basepath: Path) -> Entry | None:
|
||||
raise NotImplementedError
|
||||
@@ -0,0 +1,50 @@
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import NamedTuple
|
||||
|
||||
from ..common.command import CommandExecutor
|
||||
|
||||
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)}"
|
||||
@@ -0,0 +1,28 @@
|
||||
from pathlib import Path
|
||||
from typing import Iterable
|
||||
|
||||
from pathspec import PathSpec
|
||||
|
||||
from .detector import EntryDetector
|
||||
from .entry import Entry
|
||||
|
||||
|
||||
class EntryResolver:
|
||||
def __init__(self, detector: EntryDetector, ignore: PathSpec, max_depth: int = 4):
|
||||
self._detector = detector
|
||||
self._ignore = ignore
|
||||
self._max_depth = max_depth
|
||||
|
||||
def _resolve(self, basepath: Path, depth: int = 0) -> Iterable[Entry]:
|
||||
if basepath.is_dir():
|
||||
if self._ignore.match_file(basepath):
|
||||
return
|
||||
entry = self._detector.detect_entry(basepath)
|
||||
if entry:
|
||||
yield entry
|
||||
if depth < self._max_depth and basepath.is_dir():
|
||||
for path in basepath.iterdir():
|
||||
yield from self._resolve(path, depth + 1)
|
||||
|
||||
def resolve(self, basepath: Path) -> Iterable[Entry]:
|
||||
return self._resolve(basepath)
|
||||
Reference in New Issue
Block a user