from __future__ import annotations import json import importlib from pathlib import Path from typing import Any, Dict, List, Tuple PRESET_FILES: Dict[str, str] = { "jsonl_sm": "jsonl_to_jsonl", "simple_sm": "jsonl_to_api", } def _run_profiles_dir() -> Path: return Path(__file__).resolve().parents[2] / "run_profiles" def _preset_templates_dir() -> Path: return _run_profiles_dir() / "preset" def _replace_placeholders(value: Any, project_root: Path) -> Any: if isinstance(value, str): return value.replace("${PROJECT_ROOT}", str(project_root)) if isinstance(value, dict): return {k: _replace_placeholders(v, project_root) for k, v in value.items()} if isinstance(value, list): return [_replace_placeholders(v, project_root) for v in value] return value def _import_from_path(import_path: str) -> Any: module_path, sep, attr_path = import_path.partition(":") if not module_path: raise ValueError(f"Invalid import path: {import_path}") module = importlib.import_module(module_path) if not sep: return module obj: Any = module for attr_name in attr_path.split("."): obj = getattr(obj, attr_name) return obj def _apply_datasource_bootstraps(payload: Dict[str, Any]) -> Dict[str, Any]: extensions = payload.pop("extensions", None) if extensions is None: return payload if not isinstance(extensions, dict): raise ValueError("extensions must be an object mapping") bootstrap_paths = extensions.get("datasource_bootstraps") or [] if not isinstance(bootstrap_paths, list): raise ValueError("extensions.datasource_bootstraps must be a list") for item in bootstrap_paths: if not isinstance(item, str) or not item.strip(): raise ValueError("extensions.datasource_bootstraps entries must be non-empty strings") loaded = _import_from_path(item) if not callable(loaded): raise TypeError(f"datasource bootstrap must resolve to a callable: {item}") loaded() return payload def list_preset_names() -> List[str]: return sorted(PRESET_FILES.keys()) def get_preset_file_candidates(preset_name: str) -> List[Path]: if preset_name not in PRESET_FILES: raise ValueError(f"Unknown preset: {preset_name}. Available: {list_preset_names()}") base = PRESET_FILES[preset_name] run_profiles_root = _run_profiles_dir() preset_templates_root = _preset_templates_dir() return [ run_profiles_root / f"{base}.local.yaml", run_profiles_root / f"{base}.local.yml", run_profiles_root / f"{base}.local.json", preset_templates_root / f"{base}.default.yaml", preset_templates_root / f"{base}.default.yml", preset_templates_root / f"{base}.default.json", ] def resolve_preset_file(preset_name: str) -> Tuple[Path, bool]: candidates = get_preset_file_candidates(preset_name) for idx, path in enumerate(candidates): if path.exists() and path.is_file(): return path, idx > 0 raise FileNotFoundError( f"No preset file found for {preset_name}. Tried: {[str(p) for p in candidates]}" ) def load_overrides_from_file(file_path: Path, project_root: Path) -> Dict[str, Any]: suffix = file_path.suffix.lower() text = file_path.read_text(encoding="utf-8") if suffix in {".yaml", ".yml"}: try: import yaml # type: ignore except Exception as exc: raise RuntimeError( "YAML config requires PyYAML. Install with: pip install pyyaml" ) from exc payload = yaml.safe_load(text) else: payload = json.loads(text) resolved = _replace_placeholders(payload, project_root) if not isinstance(resolved, dict): raise ValueError(f"Preset file must contain object mapping: {file_path}") return _apply_datasource_bootstraps(resolved) def load_named_preset_overrides(project_root: Path, preset_name: str) -> Tuple[Dict[str, Any], Path, bool]: file_path, used_default = resolve_preset_file(preset_name) overrides = load_overrides_from_file(file_path, project_root) return overrides, file_path, used_default