feat: init

app/processor/base.py

@@ -0,0 +1,49 @@
+import logging
+import time
+from abc import ABC, abstractmethod
+from typing import TypeVar
+
+from app.config import Settings, settings
+from app.exceptions import AppError, ImportProcessError
+
+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
+        self.logger = logging.getLogger(f"import.{self.name}")
+
+    def __call__(self, state: T) -> T:
+        """执行节点，统一记录日志、耗时并保留业务异常类型"""
+        start_time = time.perf_counter()
+        self.logger.info("--- %s 开始 ---", self.name)
+        try:
+            result = self.process(state)
+        except AppError:
+            self.logger.exception("%s 执行失败", self.name)
+            raise
+        except Exception as e:
+            self.logger.exception("%s 执行失败", self.name)
+            raise ImportProcessError(str(e), node_name=self.name) from e
+        else:
+            self.logger.info("--- %s 完成 ---", self.name)
+            return result
+        finally:
+            self.logger.info(
+                "%s 耗时 %.3f 秒", self.name, time.perf_counter() - start_time
+            )
+
+    @abstractmethod
+    def process(self, state: T) -> T:
+        """子类实现节点核心逻辑，返回状态更新"""
+
+    def log_step(self, step_name: str, message: str = ""):
+        """记录步骤日志。"""
+        log_msg = f"[{step_name}]"
+        if message:
+            log_msg += f" {message}"
+        self.logger.info(log_msg)