from __future__ import annotations import json import re from collections import defaultdict from typing import Any, Iterable, Mapping _MARKER_PROP_NAMES = ( "hartomat_template_input", "hartomat.template_input", "template_input", "schaeffler_template_input", ) _MARKER_KEY_PROP_NAMES = ( "hartomat_template_input_key", "hartomat.template_input_key", "template_input_key", "schaeffler_template_input_key", ) _MARKER_VALUE_PROP_NAMES = ( "hartomat_template_input_value", "hartomat.template_input_value", "template_input_value", "schaeffler_template_input_value", ) _NAME_PATTERNS = ( re.compile(r"template_input__(?P[^_]+)__(?P[^_]+)", re.IGNORECASE), re.compile(r"template-input:(?P[^=]+)=(?P.+)", re.IGNORECASE), re.compile(r"ti::(?P[^:]+)::(?P.+)", re.IGNORECASE), ) def _normalize_marker_token(value: Any) -> str | None: if value is None: return None if isinstance(value, bool): return "true" if value else "false" text = str(value).strip() return text or None def _parse_marker_text(text: str) -> tuple[str, str] | None: cleaned = text.strip() if not cleaned: return None if cleaned.startswith("{"): try: payload = json.loads(cleaned) except Exception: payload = None if isinstance(payload, dict): key = _normalize_marker_token(payload.get("key")) value = _normalize_marker_token(payload.get("value")) if key and value: return key, value if "=" in cleaned: key, value = cleaned.split("=", 1) key = _normalize_marker_token(key) value = _normalize_marker_token(value) if key and value: return key, value return None def extract_template_input_marker( *, name: str | None = None, props: Mapping[str, Any] | None = None, ) -> tuple[str, str] | None: raw_props = props or {} for prop_name in _MARKER_PROP_NAMES: raw_value = raw_props.get(prop_name) text = _normalize_marker_token(raw_value) if not text: continue marker = _parse_marker_text(text) if marker is not None: return marker key = None value = None for prop_name in _MARKER_KEY_PROP_NAMES: key = _normalize_marker_token(raw_props.get(prop_name)) if key: break for prop_name in _MARKER_VALUE_PROP_NAMES: value = _normalize_marker_token(raw_props.get(prop_name)) if value: break if key and value: return key, value candidate_name = (name or "").strip() if candidate_name: for pattern in _NAME_PATTERNS: match = pattern.search(candidate_name) if not match: continue marker_key = _normalize_marker_token(match.group("key")) marker_value = _normalize_marker_token(match.group("value")) if marker_key and marker_value: return marker_key, marker_value return None def suggest_workflow_input_schema( markers: Iterable[tuple[str, str]], ) -> list[dict[str, Any]]: values_by_key: dict[str, set[str]] = defaultdict(set) for key, value in markers: normalized_key = _normalize_marker_token(key) normalized_value = _normalize_marker_token(value) if not normalized_key or not normalized_value: continue values_by_key[normalized_key].add(normalized_value) schema: list[dict[str, Any]] = [] for key in sorted(values_by_key): options = sorted(values_by_key[key]) if not options: continue label = key.replace("_", " ").strip().title() if len(options) == 2 and set(options) == {"false", "true"}: schema.append( { "key": key, "label": label, "type": "boolean", "section": "Template Inputs", "default": options[0] == "true", } ) continue schema.append( { "key": key, "label": label, "type": "select", "section": "Template Inputs", "default": options[0], "options": [{"value": option, "label": option.replace("_", " ").title()} for option in options], } ) return schema