feat: add task key sub_tag
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
import logging
|
||||
from pathlib import Path
|
||||
from typing import NamedTuple
|
||||
from typing import NamedTuple, Self
|
||||
|
||||
from ..common.command import CommandExecutor
|
||||
|
||||
@@ -10,9 +10,45 @@ logger = logging.getLogger("entry")
|
||||
class TaskKey(NamedTuple):
|
||||
name: str
|
||||
tag: str
|
||||
sub_tag: str
|
||||
|
||||
def __repr__(self) -> str:
|
||||
return f"{self.name}:{self.tag}"
|
||||
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:
|
||||
@@ -20,31 +56,50 @@ class Entry:
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
tasks: dict[TaskKey, list[str]],
|
||||
tasks: list[Task],
|
||||
basepath: Path,
|
||||
rootpath: Path,
|
||||
):
|
||||
self._tasks = tasks
|
||||
self._basepath = basepath
|
||||
self._roopath = rootpath
|
||||
|
||||
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)
|
||||
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, task_name: str, tag: str | None = None) -> bool:
|
||||
if task_name == "show":
|
||||
def run_task(self, key: PartialTaskKey) -> bool:
|
||||
if key.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)
|
||||
for task in self._tasks:
|
||||
if key == task.key:
|
||||
self._logger.info("🔵 %s %s", task.key, self)
|
||||
runned = True
|
||||
for command in commands:
|
||||
CommandExecutor(command).run(self._basepath)
|
||||
for command in task.commands:
|
||||
CommandExecutor(command).run(
|
||||
(task.cwd and Path(task.cwd)) or self._basepath
|
||||
)
|
||||
return runned
|
||||
|
||||
def __repr__(self):
|
||||
return f"{str(self._basepath)}"
|
||||
path = self._basepath.relative_to(self._roopath.parent)
|
||||
return f"[{str(path)}]"
|
||||
|
||||
Reference in New Issue
Block a user