"""Explicit registry of project-owned runnable commands. Only commands declared in ``config/commands.json`` are public. Runtime files are never made executable merely because they contain a ``__main__`` guard or use a launcher suffix. """ from __future__ import annotations import json import re from dataclasses import dataclass from pathlib import Path, PurePosixPath from types import MappingProxyType from typing import Iterable, Literal, Mapping from gyxx_flow.notification_routing import NOTIFICATION_CAPABILITIES CommandKind = Literal["python", "batch", "powershell"] _COMMAND_ID_PATTERN = re.compile( r"^[a-z][a-z0-9]*(?:[._-][a-z0-9]+(?:[a-z0-9_-]*[a-z0-9])?)*$" ) _KIND_SUFFIXES: Mapping[CommandKind, frozenset[str]] = MappingProxyType( { "python": frozenset({".py", ".pyw"}), "batch": frozenset({".bat", ".cmd"}), "powershell": frozenset({".ps1"}), } ) _DEFAULT_MODULES = ( "content_marketing", "product_commerce", "shop_intelligence", "supply_chain", ) class ScriptCatalogError(LookupError): """Raised when the command registry is invalid or unavailable.""" @dataclass(frozen=True, slots=True) class ScriptEntry: """One explicitly registered command and its legacy path-based alias.""" command_id: str module: str entry: str kind: CommandKind path: Path default_args: tuple[str, ...] = () notification_workflow_id: str | None = None @property def legacy_script_id(self) -> str: """Return the path-based identifier used before the explicit registry.""" return f"{self.module}:{self.entry}" @property def script_id(self) -> str: """Compatibility name for callers that still use the legacy identifier.""" return self.legacy_script_id class ScriptCatalog: """Immutable command registry with canonical IDs and legacy aliases.""" def __init__(self, scripts: Iterable[ScriptEntry]) -> None: commands: dict[str, ScriptEntry] = {} aliases: dict[str, ScriptEntry] = {} for script in scripts: if script.command_id in commands: raise ScriptCatalogError(f"duplicate command id: {script.command_id}") existing = aliases.get(script.command_id) if existing is not None: raise ScriptCatalogError( f"command id conflicts with legacy alias: {script.command_id}" ) alias = script.legacy_script_id existing = commands.get(alias) or aliases.get(alias) if existing is not None and existing != script: raise ScriptCatalogError(f"duplicate legacy script id: {alias}") commands[script.command_id] = script aliases[alias] = script self._commands = MappingProxyType(dict(sorted(commands.items()))) self._aliases = MappingProxyType(dict(sorted(aliases.items()))) @classmethod def load( cls, config_path: Path | str, *, roots: Mapping[str, Path | str], ) -> "ScriptCatalog": """Load an explicit command manifest and resolve every entry under a root.""" path = Path(config_path).expanduser().resolve() try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise ScriptCatalogError(f"cannot load command registry: {path}") from exc if not isinstance(payload, dict) or payload.get("schema_version") != 1: raise ScriptCatalogError("unsupported command registry schema") records = payload.get("commands") if not isinstance(records, list): raise ScriptCatalogError("command registry commands must be a list") resolved_roots: dict[str, Path] = {} for module, configured_root in roots.items(): try: root = Path(configured_root).expanduser().resolve(strict=True) except OSError as exc: raise ScriptCatalogError( f"cannot resolve runtime root for module {module}: {configured_root}" ) from exc if not root.is_dir(): raise ScriptCatalogError(f"runtime root is not a directory: {root}") resolved_roots[module] = root scripts: list[ScriptEntry] = [] for index, record in enumerate(records): scripts.append( _load_entry(record, index=index, roots=resolved_roots) ) return cls(scripts) @classmethod def load_default(cls) -> "ScriptCatalog": """Load the project command manifest without scanning runtime directories.""" project_root = Path(__file__).resolve().parents[2] module_root = project_root / "src" / "gyxx_flow" / "modules" roots = {module: module_root / module for module in _DEFAULT_MODULES} catalog = cls.load(project_root / "config" / "commands.json", roots=roots) if not all(script.path.is_relative_to(project_root) for script in catalog.scripts): raise ScriptCatalogError("default commands must stay inside the project") required = _load_workflow_script_ids(project_root / "config" / "workflows.json") missing = required.difference(catalog.script_ids) if missing: names = ", ".join(sorted(missing)) raise ScriptCatalogError(f"workflow commands are not registered: {names}") return catalog @classmethod def discover_default(cls) -> "ScriptCatalog": """Compatibility alias for :meth:`load_default`. The method name is retained for callers, but no filesystem discovery takes place. """ return cls.load_default() @property def scripts(self) -> tuple[ScriptEntry, ...]: return tuple(self._commands.values()) @property def command_ids(self) -> tuple[str, ...]: return tuple(self._commands) @property def script_ids(self) -> tuple[str, ...]: """Return legacy path-based IDs for compatibility integrations.""" return tuple(sorted(self._aliases)) def get(self, identifier: str) -> ScriptEntry: """Resolve a canonical command ID or a legacy ``module:entry`` alias.""" if not isinstance(identifier, str) or not identifier.strip(): raise ScriptCatalogError(f"invalid command id: {identifier!r}") command = self._commands.get(identifier) or self._aliases.get(identifier) if command is None: raise ScriptCatalogError(f"unknown command: {identifier}") return command def _load_entry( record: object, *, index: int, roots: Mapping[str, Path], ) -> ScriptEntry: if not isinstance(record, dict): raise ScriptCatalogError(f"command at index {index} must be an object") command_id = record.get("id") module = record.get("module") entry = record.get("entry") kind = record.get("kind") default_args = record.get("default_args", []) notification_workflow_id = record.get("notification_workflow_id") if not isinstance(command_id, str) or not _COMMAND_ID_PATTERN.fullmatch(command_id): raise ScriptCatalogError(f"invalid command id at index {index}: {command_id!r}") if not isinstance(module, str) or module not in roots: raise ScriptCatalogError(f"unknown command module at index {index}: {module!r}") if not isinstance(entry, str) or not _is_safe_entry(entry): raise ScriptCatalogError(f"unsafe command entry at index {index}: {entry!r}") if kind not in _KIND_SUFFIXES: raise ScriptCatalogError(f"invalid command kind at index {index}: {kind!r}") if not isinstance(default_args, list) or not all( isinstance(argument, str) for argument in default_args ): raise ScriptCatalogError( f"invalid command default_args at index {index}: {default_args!r}" ) if ( notification_workflow_id is not None and notification_workflow_id not in NOTIFICATION_CAPABILITIES ): raise ScriptCatalogError( "invalid command notification_workflow_id at index " f"{index}: {notification_workflow_id!r}" ) suffixes = _KIND_SUFFIXES[kind] if PurePosixPath(entry).suffix.casefold() not in suffixes: raise ScriptCatalogError( f"command kind does not match entry suffix at index {index}: {entry}" ) root = roots[module] try: target = root.joinpath(*PurePosixPath(entry).parts).resolve(strict=True) except OSError as exc: raise ScriptCatalogError(f"command entry is unavailable: {module}:{entry}") from exc if not target.is_file() or not target.is_relative_to(root): raise ScriptCatalogError(f"command entry leaves its module root: {entry}") return ScriptEntry( command_id, module, entry, kind, target, tuple(default_args), notification_workflow_id, ) def _is_safe_entry(entry: str) -> bool: if "\\" in entry: return False path = PurePosixPath(entry) return bool( entry and not path.is_absolute() and path.parts and all(part not in {"", ".", ".."} for part in path.parts) ) def _load_workflow_script_ids(path: Path) -> frozenset[str]: try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise ScriptCatalogError(f"cannot validate workflow command coverage: {path}") from exc workflows = payload.get("workflows") if isinstance(payload, dict) else None if not isinstance(workflows, list): raise ScriptCatalogError("workflow catalog workflows must be a list") required: set[str] = set() for workflow in workflows: if not isinstance(workflow, dict) or workflow.get("trigger") == "unavailable": continue module = workflow.get("module") execution = workflow.get("execution") if not isinstance(module, str) or not isinstance(execution, dict): raise ScriptCatalogError("cannot validate malformed workflow command entry") steps = execution.get("steps") if steps is None: entries = (execution.get("entry"),) elif isinstance(steps, list): entries = tuple( step.get("entry") if isinstance(step, dict) else None for step in steps ) else: raise ScriptCatalogError("workflow execution steps must be a list") if not entries or not all(isinstance(entry, str) for entry in entries): raise ScriptCatalogError("workflow command entry must be a string") required.update(f"{module}:{entry}" for entry in entries) return frozenset(required) __all__ = ["CommandKind", "ScriptCatalog", "ScriptCatalogError", "ScriptEntry"]