from __future__ import annotations from typing import Any from ... import windows_automation from ..context import WorkflowContext from ..registry import control_ports, field_def, register_node def _number(value: Any, default: float = 0) -> float: try: return float(value) except (TypeError, ValueError): return default def click_node(node: dict[str, Any], inputs: dict[str, Any], context: WorkflowContext) -> dict[str, Any]: params = node.get("params", {}) result = windows_automation.mouse_action( "click", x=int(_number(inputs.get("x", params.get("x")))), y=int(_number(inputs.get("y", params.get("y")))), duration=_number(params.get("duration"), 0), button=params.get("button") or "left", clicks=int(_number(params.get("clicks"), 1)), ) return {"x": result.get("x"), "y": result.get("y"), "action": result.get("action")} def double_click_node(node: dict[str, Any], inputs: dict[str, Any], context: WorkflowContext) -> dict[str, Any]: params = node.get("params", {}) result = windows_automation.mouse_action( "double_click", x=int(_number(inputs.get("x", params.get("x")))), y=int(_number(inputs.get("y", params.get("y")))), button=params.get("button") or "left", ) return {"x": result.get("x"), "y": result.get("y"), "action": result.get("action")} def right_click_node(node: dict[str, Any], inputs: dict[str, Any], context: WorkflowContext) -> dict[str, Any]: params = node.get("params", {}) result = windows_automation.mouse_action( "right_click", x=int(_number(inputs.get("x", params.get("x")))), y=int(_number(inputs.get("y", params.get("y")))), ) return {"x": result.get("x"), "y": result.get("y"), "action": result.get("action")} def move_node(node: dict[str, Any], inputs: dict[str, Any], context: WorkflowContext) -> dict[str, Any]: params = node.get("params", {}) result = windows_automation.mouse_action( "move_to", x=int(_number(inputs.get("x", params.get("x")))), y=int(_number(inputs.get("y", params.get("y")))), duration=_number(params.get("duration"), 0), ) return {"x": result.get("x"), "y": result.get("y"), "action": result.get("action")} def scroll_node(node: dict[str, Any], inputs: dict[str, Any], context: WorkflowContext) -> dict[str, Any]: amount = int(_number(inputs.get("amount", node.get("params", {}).get("amount")), 0)) result = windows_automation.mouse_action("scroll", amount=amount) return {"x": result.get("x"), "y": result.get("y"), "amount": amount, "action": result.get("action")} COORD_INPUTS = { "x": field_def("number", "X 坐标", required=True), "y": field_def("number", "Y 坐标", required=True), } MOUSE_OUTPUTS = { "x": {"type": "number", "label": "鼠标 X 坐标"}, "y": {"type": "number", "label": "鼠标 Y 坐标"}, "action": {"type": "string", "label": "动作名称"}, } for node_type, label, executor, params in [ ( "mouse.click", "鼠标点击", click_node, { "button": field_def("select", "按键", "left", options=["left", "middle", "right"]), "clicks": field_def("number", "点击次数", 1, minimum=1, maximum=20), "duration": field_def("number", "移动耗时", 0, minimum=0, maximum=60), }, ), ( "mouse.double_click", "鼠标双击", double_click_node, {"button": field_def("select", "按键", "left", options=["left", "middle", "right"])}, ), ("mouse.right_click", "鼠标右键", right_click_node, {}), ("mouse.move", "鼠标移动", move_node, {"duration": field_def("number", "移动耗时", 0, minimum=0, maximum=60)}), ]: register_node( { "type": node_type, "category": "mouse", "label": label, "params": params, "inputs": COORD_INPUTS, "outputs": MOUSE_OUTPUTS, "control_ports": control_ports(), }, executor, ) register_node( { "type": "mouse.scroll", "category": "mouse", "label": "鼠标滚动", "params": {"amount": field_def("number", "滚动量", 0)}, "inputs": {"amount": field_def("number", "滚动量")}, "outputs": MOUSE_OUTPUTS | {"amount": {"type": "number", "label": "滚动量"}}, "control_ports": control_ports(), }, scroll_node, )