78 lines
3.4 KiB
Python
78 lines
3.4 KiB
Python
import json
|
|
import operator
|
|
from os import environ
|
|
from pathlib import Path
|
|
|
|
from ..common.template import PathTemplate
|
|
from ..entry.detector import ENTRY_NONE, ENTRY_SKIP, EntryDetector
|
|
from ..entry.entry import Entry, Task, TaskKey
|
|
from .loader import PresetLoader, PresetLoaderConfig
|
|
from .model import Variables
|
|
|
|
|
|
class PresetEntryDetector(EntryDetector):
|
|
|
|
def __init__(self, preset_loader_config: PresetLoaderConfig | None = None):
|
|
self._preset_loader_config = preset_loader_config
|
|
|
|
# pylint:disable=too-many-locals,too-many-nested-blocks,too-many-branches
|
|
def detect_entry(self, basepath: Path, rootpath: Path) -> Entry:
|
|
template = PathTemplate(basepath)
|
|
preset = PresetLoader(basepath, self._preset_loader_config).load()
|
|
if preset.modules:
|
|
modules = [Path(item).resolve() for item in preset.modules]
|
|
if basepath not in modules:
|
|
return ENTRY_SKIP
|
|
if preset.skip:
|
|
return ENTRY_SKIP
|
|
tasks = []
|
|
global_variables = {f"env_{k.lower()}": v for k, v in environ.items()}
|
|
presets = sorted(
|
|
preset.presets.items(), key=lambda item: item[1].priority, reverse=True
|
|
)
|
|
for tag, preset in presets:
|
|
preset_variables: Variables = {
|
|
**global_variables,
|
|
"entry_path": str(basepath),
|
|
}
|
|
if preset.variables:
|
|
if tag == "*":
|
|
template.update_variables(preset.variables, global_variables)
|
|
template.update_variables(preset.variables, preset_variables)
|
|
if preset.conditions and not template.check_conditions(
|
|
preset.conditions, preset_variables
|
|
):
|
|
continue
|
|
for task_name, task_pack in preset.tasks.items():
|
|
for sub_tag, task_preset in task_pack.items():
|
|
sub_tags = template.fill_template(sub_tag, preset_variables)
|
|
try:
|
|
sub_tags = json.loads(sub_tags)
|
|
except ValueError:
|
|
sub_tags = {sub_tags: None}
|
|
for i_sub_tag, i_sub_tag_value in sub_tags.items():
|
|
if i_sub_tag_value:
|
|
preset_variables["item"] = i_sub_tag_value
|
|
if not task_preset.conditions or template.check_conditions(
|
|
task_preset.conditions, preset_variables
|
|
):
|
|
variables = preset_variables.copy()
|
|
if task_preset.variables:
|
|
template.update_variables(
|
|
task_preset.variables, variables
|
|
)
|
|
task = Task(
|
|
key=TaskKey(task_name, tag, i_sub_tag),
|
|
commands=[
|
|
template.fill_template(command, variables)
|
|
for command in task_preset.commands
|
|
],
|
|
priority=task_preset.priority,
|
|
cwd=task_preset.cwd,
|
|
)
|
|
tasks.append(task)
|
|
if tasks:
|
|
tasks = sorted(tasks, key=operator.attrgetter("priority"), reverse=True)
|
|
return Entry(tasks, basepath, rootpath)
|
|
return ENTRY_NONE
|