55 lines
1.5 KiB
Python
55 lines
1.5 KiB
Python
import logging
|
|
from collections import Counter
|
|
from enum import StrEnum, auto
|
|
from pathlib import Path
|
|
from typing import Iterable
|
|
|
|
from .common.command import CommandError
|
|
from .common.ignore import create_ignore
|
|
from .entry.entry import Entry, PartialTaskKey
|
|
from .entry.resolver import EntryResolver
|
|
from .preset.detector import PresetEntryDetector
|
|
|
|
logger = logging.getLogger("project")
|
|
|
|
|
|
class ResultType(StrEnum):
|
|
SUCCESS = auto()
|
|
FAILED = auto()
|
|
|
|
|
|
RESULT_ICONS = {
|
|
ResultType.SUCCESS: "🟢",
|
|
ResultType.FAILED: "🔴",
|
|
}
|
|
|
|
|
|
class ProjectManager:
|
|
_logger = logger
|
|
|
|
def __init__(self, basepath: Path):
|
|
self._basepaht = basepath
|
|
detector = PresetEntryDetector()
|
|
ignore = create_ignore(basepath)
|
|
self._resolver = EntryResolver(detector, ignore)
|
|
|
|
def _iter_entries(self) -> Iterable[Entry]:
|
|
return self._resolver.resolve(self._basepaht)
|
|
|
|
def run_task(self, key: PartialTaskKey) -> dict[str, int]:
|
|
counter = Counter()
|
|
for entry in self._iter_entries():
|
|
try:
|
|
if count := entry.run_task(key):
|
|
counter[ResultType.SUCCESS] += count
|
|
except CommandError as error:
|
|
self._logger.error("%s\n%s", entry, error.output)
|
|
counter[ResultType.FAILED] += 1
|
|
except:
|
|
self._logger.exception(entry)
|
|
counter[ResultType.FAILED] += 1
|
|
self._logger.info(
|
|
"%s", "\n".join(f"{RESULT_ICONS[k]} {v}" for k, v in counter.items())
|
|
)
|
|
return counter
|