123 lines
4.2 KiB
Python
123 lines
4.2 KiB
Python
import json
|
|
import re
|
|
import tomllib
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
|
|
import dotenv
|
|
from jinja2 import Environment
|
|
|
|
from ..entry import Entry, EntryDetector, TaskKey
|
|
from .loader import PresetLoader, PresetLoaderConfig
|
|
from .model import Variables
|
|
|
|
|
|
def _build_env(basepath: Path) -> Environment:
|
|
env = Environment()
|
|
|
|
def do_file(filename: str) -> Path | None:
|
|
result = Path(filename)
|
|
if not result.is_absolute():
|
|
result = basepath / result
|
|
if result.exists():
|
|
return result
|
|
return None
|
|
|
|
def do_read(filename: str) -> dict | None:
|
|
path = do_file(filename)
|
|
if path:
|
|
if path.suffix == ".toml":
|
|
return tomllib.loads(path.read_text())
|
|
if path.suffix == ".json":
|
|
return json.loads(path.read_text())
|
|
if path.suffix == ".env" or path.name == ".env":
|
|
return dotenv.dotenv_values(path)
|
|
return None
|
|
|
|
def do_get(data: dict | None, keypath: str) -> str | None:
|
|
if not data:
|
|
return None
|
|
path = keypath.split(".")
|
|
result = data.get(path[0])
|
|
for key in path[1:]:
|
|
if result:
|
|
result = result.get(key)
|
|
return result
|
|
|
|
def do_search(data: str | None, pattern: str, position: int = 1) -> str | None:
|
|
if not data:
|
|
return None
|
|
matched = re.compile(pattern).search(data)
|
|
if matched:
|
|
return matched.group(position)
|
|
return None
|
|
|
|
def do_key(data: dict | None, keyindex: int = 0) -> str | None:
|
|
if not data:
|
|
return None
|
|
keys = list(data.keys())
|
|
return keys[keyindex]
|
|
|
|
env.filters["file"] = do_file
|
|
env.filters["read"] = do_read
|
|
env.filters["get"] = do_get
|
|
env.filters["search"] = do_search
|
|
env.filters["key"] = do_key
|
|
return env
|
|
|
|
|
|
class PathProcessor:
|
|
def __init__(self, basepath: Path):
|
|
self._basepath = basepath
|
|
self._env = _build_env(basepath)
|
|
|
|
def update_variables(self, templates: Variables, variables: Variables):
|
|
for k, v in templates.items():
|
|
variables[k] = self._env.from_string(v).render(**variables)
|
|
|
|
def fill_template(self, template: str, variables: Variables) -> str:
|
|
return self._env.from_string(template).render(**variables)
|
|
|
|
def check_conditions(self, templates: list[str], variables: Variables) -> bool:
|
|
for template in templates:
|
|
value = self.fill_template(template, variables)
|
|
if value.lower() in ["false", "null", "none", "0", ""]:
|
|
return False
|
|
return True
|
|
|
|
|
|
class PresetEntryDetector(EntryDetector):
|
|
|
|
def __init__(self, preset_loader_config: PresetLoaderConfig | None = None):
|
|
self._preset_loader_config = preset_loader_config
|
|
|
|
def detect_entry(self, basepath: Path) -> Entry | None:
|
|
processor = PathProcessor(basepath)
|
|
preset = PresetLoader(basepath, self._preset_loader_config).load()
|
|
if preset.skip:
|
|
return None
|
|
tasks = defaultdict(list)
|
|
for tag, preset in preset.presets.items():
|
|
preset_variables: Variables = {"entry_path": basepath}
|
|
if preset.variables:
|
|
processor.update_variables(preset.variables, preset_variables)
|
|
if preset.conditions and not processor.check_conditions(
|
|
preset.conditions, preset_variables
|
|
):
|
|
continue
|
|
for task_name, task_pack in preset.tasks.items():
|
|
for task in task_pack.values():
|
|
if not task.conditions or processor.check_conditions(
|
|
task.conditions, preset_variables
|
|
):
|
|
variables = preset_variables.copy()
|
|
if task.variables:
|
|
processor.update_variables(task.variables, variables)
|
|
tasks[TaskKey(task_name, tag)] += [
|
|
processor.fill_template(command, variables)
|
|
for command in task.commands
|
|
]
|
|
if tasks:
|
|
return Entry(tasks, basepath)
|
|
return None
|