47 lines
1.5 KiB
Python
47 lines
1.5 KiB
Python
import logging
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
from typing import Dict, Iterable
|
|
|
|
from .command import CommandError
|
|
from .entry import Entry, EntryResolver, EntryTaskNotImplementedError
|
|
from .preset.entry import PresetEntryDetector
|
|
|
|
__all__ = ("ProjectManager",)
|
|
|
|
logger = logging.getLogger("project")
|
|
|
|
|
|
class ProjectManager:
|
|
DEFAULT_PRESET = Path(__file__).parent / "default_preset.toml"
|
|
DEFAULT_IGNORE = ["node_modules", ".venv", ".git"]
|
|
_logger = logger
|
|
|
|
def __init__(self, basepath: Path):
|
|
self._basepaht = basepath
|
|
self._resolver = EntryResolver(
|
|
PresetEntryDetector(self.DEFAULT_PRESET),
|
|
self.DEFAULT_IGNORE,
|
|
)
|
|
|
|
def _iter_entries(self) -> Iterable[Entry]:
|
|
return self._resolver.resolve(self._basepaht)
|
|
|
|
def run_task(self, task_name: str) -> Dict[str, int]:
|
|
counter = Counter()
|
|
for entry in self._iter_entries():
|
|
try:
|
|
entry.run_task(task_name)
|
|
counter["success"] += 1
|
|
except EntryTaskNotImplementedError:
|
|
self._logger.warn(f"{task_name} {entry}")
|
|
counter["skipped"] += 1
|
|
except CommandError as error:
|
|
self._logger.error(f"{entry}\n{error.output}")
|
|
counter["failed"] += 1
|
|
except Exception:
|
|
self._logger.exception(entry)
|
|
counter["failed"] += 1
|
|
self._logger.info(f"result {dict(counter)}")
|
|
return counter
|