win_monitor/backend/app/automation/nodes/flow.py

from __future__ import annotations

from typing import Any

from ..context import WorkflowContext
from ..registry import control_ports, register_node


def start_node(node: dict[str, Any], inputs: dict[str, Any], context: WorkflowContext) -> dict[str, Any]:
    return {"message": "workflow started"}


def end_node(node: dict[str, Any], inputs: dict[str, Any], context: WorkflowContext) -> dict[str, Any]:
    return {"message": "workflow ended"}


def condition_node(node: dict[str, Any], inputs: dict[str, Any], context: WorkflowContext) -> dict[str, Any]:
    left = inputs.get("left", node.get("params", {}).get("left"))
    right = inputs.get("right", node.get("params", {}).get("right"))
    operator = node.get("params", {}).get("operator") or "equals"
    matched = {
        "equals": left == right,
        "not_equals": left != right,
        "truthy": bool(left),
        "falsy": not bool(left),
        "contains": str(right) in str(left),
    }.get(operator, False)
    return {"matched": matched, "next_port": "true" if matched else "false"}


register_node(
    {
        "type": "flow.start",
        "category": "flow",
        "label": "开始",
        "params": {},
        "inputs": {},
        "outputs": {},
        "control_ports": {"inputs": [], "outputs": ["next"]},
    },
    start_node,
)

register_node(
    {
        "type": "flow.end",
        "category": "flow",
        "label": "结束",
        "params": {},
        "inputs": {},
        "outputs": {},
        "control_ports": {"inputs": ["run"], "outputs": []},
    },
    end_node,
)

register_node(
    {
        "type": "flow.condition",
        "category": "flow",
        "label": "条件判断",
        "params": {
            "operator": {
                "type": "select",
                "label": "判断方式",
                "default": "equals",
                "options": ["equals", "not_equals", "truthy", "falsy", "contains"],
            },
            "left": {"type": "text", "label": "左值"},
            "right": {"type": "text", "label": "右值"},
        },
        "inputs": {
            "left": {"type": "any", "label": "左值"},
            "right": {"type": "any", "label": "右值"},
        },
        "outputs": {"matched": {"type": "boolean", "label": "是否匹配"}},
        "control_ports": control_ports(["true", "false", "failure"]),
    },
    condition_node,
)