106 lines
2.9 KiB
Python
106 lines
2.9 KiB
Python
import logging
|
|
from pathlib import Path
|
|
from typing import NamedTuple, Self
|
|
|
|
from ..common.command import CommandExecutor
|
|
|
|
logger = logging.getLogger("entry")
|
|
|
|
|
|
class TaskKey(NamedTuple):
|
|
name: str
|
|
tag: str
|
|
sub_tag: str
|
|
|
|
def __repr__(self) -> str:
|
|
return self.to_str()
|
|
|
|
def to_str(self, parts: int | None = None) -> str:
|
|
values = list(self)
|
|
if parts is not None:
|
|
values = values[:parts]
|
|
return ":".join(filter(bool, values))
|
|
|
|
|
|
class PartialTaskKey(TaskKey):
|
|
tag: str | None
|
|
sub_tag: str | None
|
|
|
|
@classmethod
|
|
def from_str(cls, value: str) -> Self:
|
|
parts = value.split(":")
|
|
parts = parts[:3] + [None for _ in range(3 - len(parts))]
|
|
return cls(*parts)
|
|
|
|
def __eq__(self, value: object) -> bool:
|
|
if isinstance(value, TaskKey):
|
|
if self.name != value.name:
|
|
return False
|
|
if self.tag and self.tag != value.tag:
|
|
return False
|
|
if self.sub_tag and self.sub_tag != value.sub_tag:
|
|
return False
|
|
return True
|
|
return super().__eq__(value)
|
|
|
|
|
|
class Task(NamedTuple):
|
|
key: TaskKey
|
|
commands: list[str]
|
|
priority: int
|
|
cwd: str | None = None
|
|
|
|
|
|
class Entry:
|
|
_logger = logger
|
|
|
|
def __init__(
|
|
self,
|
|
tasks: list[Task],
|
|
basepath: Path,
|
|
rootpath: Path,
|
|
):
|
|
self._tasks = tasks
|
|
self._basepath = basepath
|
|
self._roopath = rootpath
|
|
|
|
def show(self):
|
|
self._logger.info("🔵 %s", self)
|
|
level = logging.root.level
|
|
logged = set()
|
|
for task in self._tasks:
|
|
if level == logging.INFO:
|
|
if task.key.name not in logged:
|
|
self._logger.info(" %s", task.key.name)
|
|
logged.add(task.key.name)
|
|
elif level == logging.DEBUG:
|
|
task_title = task.key.to_str(2)
|
|
if task_title not in logged:
|
|
self._logger.info(" %s", task_title)
|
|
logged.add(task_title)
|
|
elif level == logging.TRACE:
|
|
self._logger.info(" %s", task.key)
|
|
if task.cwd:
|
|
self._logger.debug(" > %s", task.cwd)
|
|
for command in task.commands:
|
|
self._logger.debug(" %s", command)
|
|
|
|
def run_task(self, key: PartialTaskKey) -> int:
|
|
if key.name == "show":
|
|
self.show()
|
|
return True
|
|
c = 0
|
|
for task in self._tasks:
|
|
if key == task.key:
|
|
self._logger.info("🔵 %s %s", task.key, self)
|
|
for command in task.commands:
|
|
CommandExecutor(command).run(
|
|
(task.cwd and Path(task.cwd)) or self._basepath
|
|
)
|
|
c += 1
|
|
return c
|
|
|
|
def __repr__(self):
|
|
path = self._basepath.relative_to(self._roopath.parent)
|
|
return f"[{str(path)}]"
|