82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
import logging
|
|
import inspect
|
|
import time
|
|
from abc import ABC, abstractmethod
|
|
from typing import TYPE_CHECKING, TypeVar
|
|
|
|
from pydantic import ValidationError
|
|
|
|
if TYPE_CHECKING:
|
|
from app.config import 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
|
|
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:
|
|
"""执行节点，统一记录日志、耗时并保留业务异常类型"""
|
|
if inspect.iscoroutinefunction(self.process):
|
|
return self._acall(state)
|
|
start_time = time.perf_counter()
|
|
self.logger.info("--- %s 开始 ---", self.name)
|
|
try:
|
|
result = self.process(state)
|
|
except (AppError, ValidationError):
|
|
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:
|
|
"""子类实现节点核心逻辑，返回状态更新"""
|
|
|
|
async def _acall(self, state: T) -> T:
|
|
"""异步节点等待执行完成后再记录结果和耗时"""
|
|
start_time = time.perf_counter()
|
|
self.logger.info("--- %s 开始 ---", self.name)
|
|
try:
|
|
result = await self.process(state)
|
|
except (AppError, ValidationError):
|
|
self.logger.exception("%s 执行失败", self.name)
|
|
raise
|
|
except Exception as error:
|
|
self.logger.exception("%s 执行失败", self.name)
|
|
raise ImportProcessError(str(error), node_name=self.name) from error
|
|
else:
|
|
self.logger.info("--- %s 完成 ---", self.name)
|
|
return result
|
|
finally:
|
|
self.logger.info("%s 耗时 %.3f 秒", self.name, time.perf_counter() - start_time)
|
|
|
|
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)
|