feat: 完成input_context节点

app/processor/base.py

@@ -1,9 +1,12 @@
 import logging
 import time
 from abc import ABC, abstractmethod
-from typing import TypeVar
+from typing import TYPE_CHECKING, TypeVar
 
-from app.config import Settings, settings
+from pydantic import ValidationError
+
+if TYPE_CHECKING:
+    from app.config import Settings
 from app.exceptions import AppError, ImportProcessError
 
 T = TypeVar("T")
@@ -12,18 +15,26 @@ T = TypeVar("T")
 class BaseNode(ABC):
     name: str = "base_node"
 
-    def __init__(self, config: Settings | None = None):
-        """传入配置时使用传入值，否则使用全局配置"""
-        self.config = config if config is not None else settings
+    def __init__(self, config: "Settings | None" = None):
+        """保存显式配置；全局配置在节点实际使用时才加载。"""
+        self._config = config
         self.logger = logging.getLogger(f"import.{self.name}")
 
+    @property
+    def config(self) -> "Settings":
+        if self._config is None:
+            from app.config import settings
+
+            self._config = settings
+        return self._config
+
     def __call__(self, state: T) -> T:
         """执行节点，统一记录日志、耗时并保留业务异常类型"""
         start_time = time.perf_counter()
         self.logger.info("--- %s 开始 ---", self.name)
         try:
             result = self.process(state)
-        except AppError:
+        except (AppError, ValidationError):
             self.logger.exception("%s 执行失败", self.name)
             raise
         except Exception as e:

app/processor/nodes/input_node.py

@@ -0,0 +1,60 @@
+"""输入解析与校验，可直接运行本文件修改示例进行调试"""
+
+from app.processor.base import BaseNode
+from app.processor.state import QueryState
+from app.schemas.input_context import InputContext
+
+
+class InputNode(BaseNode):
+    """纯输入节点，不依赖模型配置；校验失败保留 ValidationError"""
+
+    name = "input_node"
+
+    def process(self, state: QueryState) -> QueryState:
+        # 深拷贝隔离原始请求，后续修改上下文不会改变提交时的快照
+        context = InputContext.model_validate(state["request"]).model_copy(deep=True)
+        updated_state: QueryState = {**state, "input_context": context}
+        return updated_state
+
+
+if __name__ == "__main__":
+    import json
+
+    # 1. 创建 state：可将 screen 替换成前端的完整大屏 JSON
+    state: QueryState = {
+        "request": {
+            "prompt": "把销售额统计向右移动 50px",
+            "screen": {
+                "canvas": {
+                    "width": 1920,
+                    "height": 1080,
+                    "backgroundColor": "#ffffff10",
+                },
+                "nodes": [
+                    {
+                        "id": "sales-chart",
+                        "type": "bar-chart",
+                        "name": "销售额统计",
+                        "layout": {"x": 26, "y": 34.8, "width": 400, "height": 260},
+                        "props": {"option": {"title": {"text": "销售额统计"}}},
+                    }
+                ],
+                "dataSource": [],
+            },
+            "context": {"selected_node_ids": ["sales-chart"]},
+            "attachments": [],
+        }
+    }
+
+    # 2. 执行节点
+    node = InputNode()
+    final_state = node.process(state)
+
+    # 3. 打印完整状态
+    printable_state = {
+        **final_state,
+        "input_context": final_state["input_context"].model_dump(
+            mode="json", exclude_unset=True
+        ),
+    }
+    print(json.dumps(printable_state, ensure_ascii=False, indent=2))

app/processor/state.py

@@ -0,0 +1,9 @@
+from typing import NotRequired, TypedDict
+
+from app.schemas.base import JsonObject
+from app.schemas.input_context import InputContext
+
+
+class QueryState(TypedDict):
+    request: JsonObject
+    input_context: NotRequired[InputContext]