def parse_yaml(source: str | Path) -> PipelineDefinition:
    """Parse a YAML pipeline file or string."""

def _load_yaml(source):
    path = Path(source)
    if path.exists() and path.is_file():
        # 作为文件路径处理
        with open(path, "r", encoding="utf-8") as f:
            data = yaml.safe_load(f)
    else:
        # 作为 YAML 字符串处理
        data = yaml.safe_load(source)

def validate_pipeline(pipeline: PipelineDefinition) -> list[str]:
    """Validate a parsed pipeline. Returns warnings."""

_VALID_STEP_ID = re.compile(r"^[a-z0-9_-]+$")

if not _VALID_STEP_ID.match(step.id):
    raise PipelineValidationError(
        f"Step id '{step.id}' is invalid. "
        f"step_id must match [a-z0-9_-] (no dots allowed)."
    )

if step.id in seen_ids:
    raise PipelineValidationError(f"Duplicate step id '{step.id}'.")

if not registry.has(step.use):
    raise PipelineValidationError(
        f"Step '{step.id}' uses '{step.use}' which is not registered."
    )

def _validate_value_refs(step_id, key, value, available_outputs, variables):
    if isinstance(value, dict):
        for k, v in value.items():
            _validate_value_refs(step_id, f"{key}.{k}", v, ...)
    if isinstance(value, list):
        for i, v in enumerate(value):
            _validate_value_refs(step_id, f"{key}[{i}]", v, ...)
    # 检查步骤引用和变量引用

if step.on_error not in ("fail", "skip", "retry"):
    raise PipelineValidationError(...)

for key, ref in pipeline.outputs.items():
    if isinstance(ref, str) and ref.startswith("$"):
        step_id = ref[1:].split(".")[0]
        if step_id not in seen_ids:
            raise PipelineValidationError(...)

def resolve_step_params(pipeline, step_index, context) -> dict:
    step = pipeline.steps[step_index]
    return context.resolve_params(step.params)

class PipelineContext:
    def __init__(self, variables=None):
        self.variables = dict(variables or {})
        self._step_outputs = {}  # step_id → StepResult

def _resolve_step_ref(self, ref):
    ref_body = ref[1:]               # 去掉 $
    step_id, attr = ref_body.split(".", 1)
    result = self._step_outputs[step_id]
    return getattr(result, attr)      # 使用 __getattr__

def _substitute_variables(self, text):
    # 完整变量引用 → 保持原始类型
    match = re.fullmatch(r"\$\{(\w+)\}", text)
    if match:
        return self.variables[match.group(1)]

    # 嵌入式替换 → 字符串
    return re.sub(r"\$\{(\w+)\}", replace_func, text)

class StepContext:
    def __init__(self, params, backend=None, pipeline_context=None):
        self._params = params
        self.backend = backend
        self.pipeline_context = pipeline_context

def execute_pipeline(pipeline: PipelineDefinition) -> dict:
    context = PipelineContext(variables=pipeline.variables)
    registry = StepRegistry()
    backend_manager = BackendManager()
    step_reports = []

    for step_def in pipeline.steps:
        # 1. 条件执行检查
        # 2. 执行步骤（含重试逻辑）
        # 3. 记录结果
        ...

    return build_report(...)

if step_def.when:
    if not _evaluate_condition(step_def.when, context):
        step_report["status"] = "skipped"
        context.set_output(step_def.id, StepResult())  # 空结果
        continue

def _evaluate_condition(condition, context):
    # 1. 替换 ${var} 和 $step.attr
    resolved = condition
    resolved = re.sub(r"\$\{(\w+)\}", replace_var, resolved)
    resolved = re.sub(r"\$(\w+)\.(\w+)", replace_ref, resolved)

    # 2. AST 验证
    tree = ast.parse(resolved, mode="eval")
    _SAFE_NODES = (
        ast.Expression, ast.Compare, ast.BoolOp, ast.UnaryOp,
        ast.BinOp, ast.Constant, ast.Name, ast.Load,
        ast.Eq, ast.NotEq, ast.Lt, ast.LtE, ast.Gt, ast.GtE,
        ast.And, ast.Or, ast.Not, ...
    )
    for node in ast.walk(tree):
        if not isinstance(node, _SAFE_NODES):
            return False  # 不安全的表达式返回 False

    # 3. 安全 eval
    return bool(eval(
        compile(tree, "<when>", "eval"),
        {"__builtins__": {}}, {},
    ))

_MAX_RETRIES = 3

def _execute_step_with_retry(step_def, ...):
    max_attempts = _MAX_RETRIES if step_def.on_error == "retry" else 1

    for attempt in range(1, max_attempts + 1):
        try:
            # 执行步骤
            ...
            return  # 成功
        except Exception as e:
            if attempt < max_attempts:
                time.sleep(0.5 * attempt)  # 退避
            else:
                raise

# 1. 解析参数
resolved_params = resolve_step_params(pipeline, step_index, context)

# 2. 查找步骤信息
step_info = registry.get(step_def.use)

# 3. 选择后端（IO 步骤不需要后端）
backend = None
if step_info.category not in ("io",):
    backend = backend_manager.get(step_def.backend)

# 4. 构建 StepContext
step_ctx = StepContext(
    params=resolved_params,
    backend=backend,
    pipeline_context=context,
)

# 5. 执行步骤函数
result = step_info.func(step_ctx)

# 6. 存储结果
context.set_output(step_def.id, result)

def _suggest_fix(step_use, error):
    msg = str(error).lower()
    if "crs" in msg and ("mismatch" in msg or "degree" in msg):
        return "Add a vector.reproject step before this step."
    if "file not found" in msg or "no such file" in msg:
        return "Check that the input file path is correct."
    if "permission" in msg:
        return "Check file permissions for the input/output paths."
    return None

def build_report(pipeline_name, status, duration, step_reports, outputs):
    return {
        "pipeline": pipeline_name,
        "status": status,              # "success" 或 "error"
        "duration_seconds": duration,   # 总执行时间
        "steps": step_reports,          # 各步骤报告
        "outputs": outputs,             # 流水线输出
    }