""" DataSource V2 - 状态管理 + Handler 编排层 DataSource 职责： 1. 注册和管理 Handler 2. 按依赖顺序加载数据 3. 执行同步（调用 Handler） 4. 管理节点状态流转（PENDING → IN_PROGRESS → SUCCESS/FAILED） 5. 处理异步任务轮询 6. 统一异常处理 DataSource 不负责： - ❌ 后端 API 调用（由 Handler 负责） - ❌ ID 映射（由 Strategy 负责） - ❌ 业务逻辑（由 Strategy 负责） """ from __future__ import annotations import asyncio import logging from typing import Any, List, Dict, Optional, TYPE_CHECKING from abc import ABC from pathlib import Path if TYPE_CHECKING: from ..common.collection import DataCollection from .handler import NodeHandler from ..common.sync_node import SyncNode from ..common.types import SyncAction from .task_result import TaskStatus from ..common.types import SyncStatus from ..common.type_safety import get_pydantic_model_fields from .task_result import TaskResult from ..engine import ( StateMachineConfig, StateMachineRuntime, e40_sync_execute, ) from ..logging import get_logger logger = get_logger(__name__) class BaseDataSource(ABC): """ 数据源基类 V2（状态管理 + 编排层） 核心流程： 1. 注册 Handler：每个节点类型对应一个 Handler 2. 加载数据：按依赖顺序调用 Handler.load() 3. 执行同步：根据 node.action 调用 Handler 的 create/update/delete 4. 状态管理：根据 TaskResult 更新 node.status 5. 异步轮询：处理 IN_PROGRESS 节点 示例： # 1. 初始化 DataSource datasource = MyDataSource() # 2. 添加 Handler datasource.add_handler(project_handler) datasource.add_handler(contract_handler) # 3. 设置 Collection datasource.set_collection(collection) # 4. 加载数据（按顺序） await datasource.load_all(order=["project", "contract"]) # 5. 执行同步（按顺序） await datasource.sync_all(order=["project", "contract"]) """ def __init__(self, poll_max_retries: int = 1, poll_interval: float = 0.5): self._handlers: Dict[str, "NodeHandler"] = {} self._collection: Optional["DataCollection"] = None # 按节点类型统计信息 # {"company": {"loaded": 10, "synced": 10, "success": 10, "failed": 0}, ...} self._stats: Dict[str, Dict[str, int]] = {} # 记录本次 sync 的动作统计（CREATE/UPDATE/DELETE/NONE） self._action_summary: Dict[str, Dict[str, Dict[str, int]]] = {} # 记录本次 sync 中的节点动作（用于异步完成后的统计） self._action_by_node_id: Dict[str, "SyncAction"] = {} self._poll_max_retries = max(1, poll_max_retries) self._poll_interval = max(0, poll_interval) self._sm_runtime: Optional[StateMachineRuntime] = None def _ensure_sm_runtime(self) -> StateMachineRuntime: if self._sm_runtime is None: cfg_path = Path(__file__).resolve().parents[1] / "config" / "node_state_machine.yaml" self._sm_runtime = StateMachineRuntime(StateMachineConfig.load(cfg_path)) return self._sm_runtime def set_state_machine_runtime(self, runtime: StateMachineRuntime) -> None: """注入统一状态机 runtime（通常由 pipeline 负责）。""" self._sm_runtime = runtime def _apply_sync_execute_state( self, node: "SyncNode", *, handler_status: "TaskStatus", poll_timeout: bool = False, detail: str = "", action_override: Optional["SyncAction"] = None, result_data_id: Optional[str] = None, ) -> bool: action_for_event = action_override or node.action decision = e40_sync_execute( self._ensure_sm_runtime(), node=node, handler_result=handler_status.name, action=action_for_event.value, poll_timeout=poll_timeout, result_data_id=result_data_id, ) if decision is None: if detail: node.append_log( f"执行状态未变更: action={action_for_event.value}, result={handler_status.name}, detail={detail}" ) return False if detail: node.append_log( f"执行状态写回: action={action_for_event.value}, result={handler_status.name}, detail={detail}" ) return True def _compose_node_error(self, node: "SyncNode", *, source: str, detail: str) -> str: return ( f"{source} | node={node.node_id} | type={node.node_type} | " f"action={node.action.value} | status={node.status.value} | detail={detail}" ) def _on_node_loaded(self, handler: "NodeHandler", node: "SyncNode") -> None: """Datasource-specific hook when a loaded node is upserted into collection.""" return async def _intercept_success_result( self, handler: "NodeHandler", node: "SyncNode", result: "TaskResult", async_tasks: Dict[str, str], ) -> bool: """Datasource-specific hook to intercept SUCCESS result handling. Return True if handled/consumed by subclass and no further default SUCCESS flow is needed. """ return False def _on_in_progress_result(self, handler: "NodeHandler", node: "SyncNode", result: "TaskResult") -> None: """Datasource-specific hook after an IN_PROGRESS task is queued.""" return def _on_poll_result(self, handler: "NodeHandler", node: "SyncNode", task_id: str, result: "TaskResult") -> None: """Datasource-specific hook after polling result is applied to node.""" return def _on_failed_result(self, handler: "NodeHandler", node: "SyncNode", result: "TaskResult") -> None: """Datasource-specific hook after a FAILED result is applied to node.""" return async def _handle_success_result(self, node: "SyncNode", result: "TaskResult") -> None: previous_data_id = node.data_id ok = self._apply_sync_execute_state( node, handler_status=result.status, detail="远程操作成功完成", result_data_id=result.data_id, ) if not ok: raise RuntimeError(f"[{node.node_type}] SUCCESS result cannot map to state machine: node={node.node_id}") self._record_action_result(node, result.status) await self._apply_success_result(node, result.data_id, previous_data_id=previous_data_id) node.error = None async def _handle_failed_result(self, node: "SyncNode", result: "TaskResult") -> None: detail = result.error or "Unknown error" ok = self._apply_sync_execute_state( node, handler_status=result.status, detail=f"远程操作失败: {detail}", ) if not ok: raise RuntimeError(f"[{node.node_type}] FAILED result cannot map to state machine: node={node.node_id}") self._record_action_result(node, result.status) node.error = self._compose_node_error(node, source="handler_failed", detail=detail) async def _handle_skipped_result(self, node: "SyncNode", result: "TaskResult") -> None: from .task_result import TaskStatus as _TaskStatus original_action = self._action_by_node_id.get(node.node_id) if original_action is None or original_action.value != "UPDATE": raise RuntimeError( f"[{node.node_type}] SKIPPED is only allowed for UPDATE action: node={node.node_id}, action={original_action.value if original_action is not None else None}" ) ok = self._apply_sync_execute_state( node, handler_status=_TaskStatus.SKIPPED, detail="远程操作被跳过", action_override=original_action, ) if not ok: raise RuntimeError(f"[{node.node_type}] SKIPPED result cannot map to state machine: node={node.node_id}") self._record_action_result(node, _TaskStatus.SKIPPED) if result.sync_log: node.append_log(f"同步跳过: {result.sync_log}") node.error = None async def _handle_in_progress_result( self, node: "SyncNode", result: "TaskResult", async_tasks: Dict[str, str], ) -> None: if not result.task_id: node.error = self._compose_node_error( node, source="handler_in_progress_invalid", detail="missing task_id", ) node.append_log("异步任务返回无 task_id，已忽略") return if not result.node_id: node.error = self._compose_node_error( node, source="handler_in_progress_invalid", detail="missing node_id", ) node.append_log("异步任务返回无 node_id，已忽略") return async_tasks[result.node_id] = result.task_id ok = self._apply_sync_execute_state( node, handler_status=result.status, detail=f"异步任务进行中: {result.task_id}", ) if not ok: node.append_log(f"异步任务进行中（无状态变更）: {result.task_id}") def set_polling_config(self, max_retries: int, interval: float = 0.5) -> None: """设置异步轮询参数（默认仅轮询一次）""" self._poll_max_retries = max(1, max_retries) self._poll_interval = max(0, interval) def reset_stats(self) -> None: """重置统计信息""" self._stats = {} self._action_summary = {} self._action_by_node_id = {} def get_stats(self) -> Dict[str, Dict[str, int]]: """ 获取按节点类型的统计信息 Returns: dict: {"company": {"loaded": 10, "synced": 10, "success": 10, "failed": 0}, ...} """ return self._stats.copy() def get_action_summary(self) -> Dict[str, Dict[str, Dict[str, int]]]: """ 获取本次 sync 的动作统计（按节点类型）。 Returns: dict: { "company": { "create": {"total": 10, "success": 8, "failed": 2}, "update": {"total": 2, "success": 2, "failed": 0}, "delete": {"total": 0, "success": 0, "failed": 0}, "none": {"total": 5, "success": 0, "failed": 0}, "total": {"total": 17, "success": 10, "failed": 2}, }, ... } """ return {k: {kk: vv.copy() for kk, vv in v.items()} for k, v in self._action_summary.items()} def _init_node_stats(self, node_type: str) -> None: """初始化节点类型的统计信息""" if node_type not in self._stats: self._stats[node_type] = { "loaded": 0, "synced": 0, "success": 0, "failed": 0, } if node_type not in self._action_summary: self._action_summary[node_type] = { "create": {"total": 0, "success": 0, "failed": 0, "skipped": 0}, "update": {"total": 0, "success": 0, "failed": 0, "skipped": 0}, "delete": {"total": 0, "success": 0, "failed": 0, "skipped": 0}, "none": {"total": 0, "success": 0, "failed": 0, "skipped": 0}, "total": {"total": 0, "success": 0, "failed": 0, "skipped": 0}, } def _set_action_totals(self, node_type: str, create: int, update: int, delete: int, none: int) -> None: summary = self._action_summary[node_type] summary["create"]["total"] = create summary["update"]["total"] = update summary["delete"]["total"] = delete summary["none"]["total"] = none summary["total"]["total"] = create + update + delete + none def _record_action_result(self, node: "SyncNode", result_status: "TaskStatus") -> None: from ..common.types import SyncAction from .task_result import TaskStatus as _TaskStatus # 情况 A: 原始动作就是 NONE # 情况 B: Handler 运行时发现没差异，把 action 降级为了 NONE action = node.action # 获取最初锁定的动作（用于修正统计列） original_action = self._action_by_node_id.pop(node.node_id, SyncAction.NONE) summary = self._action_summary.get(node.node_type) if summary is None: return # 如果当前 action 是 NONE，说明这次“同步”被取消了 if action == SyncAction.NONE: # 如果原始动作不是 NONE，说明发生了降级动作，我们需要把 total 从对应列挪到 none 列 if original_action != SyncAction.NONE: orig_key = original_action.value.lower() if orig_key in summary and summary[orig_key]["total"] > 0: summary[orig_key]["total"] -= 1 summary["none"]["total"] += 1 return # 正常记录成功/失败/跳过 key = action.value.lower() if result_status == _TaskStatus.SUCCESS: summary[key]["success"] += 1 summary["total"]["success"] += 1 elif result_status == _TaskStatus.FAILED: summary[key]["failed"] += 1 summary["total"]["failed"] += 1 elif result_status == _TaskStatus.SKIPPED: summary[key]["skipped"] += 1 summary["total"]["skipped"] += 1 def set_collection(self, collection: "DataCollection") -> None: """ 设置 Collection（在 load_all/sync_all 前调用） Args: collection: 数据集合 """ self._collection = collection collection.set_state_machine_runtime(self._ensure_sm_runtime()) # 同时设置到所有已注册的 handler for handler in self._handlers.values(): handler.set_collection(collection) # ========== Handler 管理 ========== def add_handler(self, handler: "NodeHandler") -> None: """ 添加 Handler 到当前 datasource Args: handler: 节点处理器实例 示例： datasource.add_handler(ContractNodeHandler(api_client)) """ self._handlers[handler.node_type] = handler def get_handler(self, node_type: str) -> "NodeHandler": """获取 Handler""" if node_type not in self._handlers: raise KeyError(f"Handler for '{node_type}' not registered") return self._handlers[node_type] async def _upsert_loaded_nodes( self, handler: "NodeHandler", node_type: str, nodes: List["SyncNode"] ) -> None: """将加载的节点写入 Collection（按 data_id 优先命中已有节点）""" if self._collection is None: raise RuntimeError("Collection not set. Call set_collection() first.") for node in nodes: existing_node = None if node.data_id: existing_node = self._collection.get_by_data_id(node_type, node.data_id) load_source = f"{self.__class__.__name__}.load" if existing_node is not None: data_payload = node.get_data() existing_node.depend_ids = list(node.depend_ids or []) if node.context: existing_node.context.update(node.context) existing_node.set_data(data_payload) existing_node.set_origin_data(data_payload) if existing_node.context.pop("_loaded_from_persistence", False): existing_node.append_log(f"加载来源: persistence -> {load_source} 覆盖刷新") else: existing_node.append_log(f"加载来源: {load_source} 覆盖刷新") await self._collection.update(existing_node) self._on_node_loaded(handler, existing_node) else: node.append_log(f"加载来源: {load_source} 新增节点") await self._collection.add(node) self._on_node_loaded(handler, node) # ========== 数据加载 ========== async def load_all( self, order: Optional[List[str]] = None, data_type: Optional[str] = None ) -> None: """ 按依赖顺序加载所有数据 注意：调用前需先 set_collection() Args: order: 节点类型顺序（如 ["project", "contract", "construction"]) data_type: 仅加载单一类型（与 order 互斥） 工作流程： 1. 按顺序遍历每个节点类型 2. 调用 Handler.load(collection) 3. Handler 可以从 collection 查询依赖节点（如 Contract 查询 Project） 4. 使用 Handler.create_node() 创建 SyncNode 5. 添加到 Collection 示例： datasource.set_collection(collection) await datasource.load_all(order=["project", "contract"]) """ if self._collection is None: raise RuntimeError("Collection not set. Call set_collection() first.") if data_type is not None: order_to_load = [data_type] elif order is not None: order_to_load = order else: order_to_load = list(self._handlers.keys()) for node_type in order_to_load: handler = self.get_handler(node_type) # 初始化统计 self._init_node_stats(node_type) try: # Handler 加载数据并创建节点 nodes = await handler.load() logger.info("Loaded nodes for %s: %d", node_type, len(nodes)) # 将节点写入 Collection：优先按 data_id 命中已有节点并更新 await self._upsert_loaded_nodes(handler, node_type, nodes) # 统计加载的节点数 self._stats[node_type]["loaded"] += len(nodes) except Exception as exc: logger.error("❌ [%s] 加载失败: %s", node_type, exc) continue # ========== 同步执行 ========== async def sync_all( self, order: Optional[List[str]] = None, data_type: Optional[str] = None, poll_async_tasks: bool = True ) -> Dict[str, Dict[str, str]]: """ 按依赖顺序执行同步 注意：调用前需先 set_collection() Args: order: 节点类型顺序 data_type: 仅同步单一类型（与 order 互斥） 工作流程： 1. 按顺序遍历每个节点类型 2. 筛选需要同步的节点（status=PENDING, action!=NONE） 3. 调用 Handler 的 create/update/delete 4. 根据 TaskResult 更新节点状态 5. 处理异步任务轮询 状态流转： - PENDING + action=CREATE/UPDATE/DELETE → IN_PROGRESS → SUCCESS/FAILED - PENDING + action=NONE → 保持 PENDING - DEPENDENCY_ERROR / ABNORMAL / WARNING → SKIPPED 示例： datasource.set_collection(collection) await datasource.sync_all(order=["project", "contract"]) """ if self._collection is None: raise RuntimeError("Collection not set. Call set_collection() first.") if data_type is not None: order_to_sync = [data_type] elif order is not None: order_to_sync = order else: order_to_sync = list(self._handlers.keys()) async_tasks_by_type: Dict[str, Dict[str, str]] = {} for node_type in order_to_sync: handler = self.get_handler(node_type) # 批量同步该类型的所有节点 async_tasks = await self._sync_nodes(handler, poll_async_tasks=poll_async_tasks) if async_tasks: async_tasks_by_type[node_type] = async_tasks return async_tasks_by_type async def _sync_nodes( self, handler: "NodeHandler", poll_async_tasks: bool = True ) -> Dict[str, str]: """ 同步一批节点（批量接口） Args: handler: 节点处理器 流程： 1. DataSource 筛选需要处理的节点 2. 调用 handler.create_all(nodes) 3. 调用 handler.update_all(nodes) 4. 调用 handler.delete_all(nodes) 5. 根据 List[TaskResult] 更新节点状态 6. 收集并轮询异步任务 """ from ..common.types import SyncAction, SyncStatus from .task_result import TaskStatus as _TaskStatus if self._collection is None: raise RuntimeError("Collection not set. Call set_collection() first.") collection = self._collection # 初始化统计 self._init_node_stats(handler.node_type) async_tasks: Dict[str, str] = {} # {node_id: task_id} # DataSource 负责筛选需要创建的节点（排除已失败的节点） create_nodes = [ n for n in collection.filter_by_state_ids( node_type=handler.node_type, state_ids=["S06"], ) if n.action == SyncAction.CREATE ] # DataSource 负责筛选需要更新的节点（排除已失败的节点） update_nodes = [ n for n in collection.filter_by_state_ids( node_type=handler.node_type, state_ids=["S07"], ) if n.action == SyncAction.UPDATE ] # DataSource 负责筛选需要删除的节点（排除已失败的节点） delete_nodes = [ n for n in collection.filter_by_state_ids( node_type=handler.node_type, state_ids=["S09"], ) if n.action == SyncAction.DELETE ] if delete_nodes: raise NotImplementedError( f"[{handler.node_type}] delete path is not implemented yet, " f"but found {len(delete_nodes)} node(s) in S09/DELETE." ) # 记录本次 sync 的动作统计（在 action 被重置前） for node in create_nodes: self._action_by_node_id[node.node_id] = SyncAction.CREATE for node in update_nodes: self._action_by_node_id[node.node_id] = SyncAction.UPDATE for node in delete_nodes: self._action_by_node_id[node.node_id] = SyncAction.DELETE none_count = len( self._collection.filter( node_type=handler.node_type, node_filter={"action": SyncAction.NONE} ) ) self._set_action_totals( handler.node_type, create=len(create_nodes), update=len(update_nodes), delete=len(delete_nodes), none=none_count, ) # 批量创建（传入节点列表） if create_nodes: logger.info(f"[{handler.node_type}] create提交开始: nodes={len(create_nodes)}") try: create_results = await handler.create_all(create_nodes) create_success = 0 create_in_progress = 0 create_failed = 0 create_skipped = 0 for result in create_results: if result.status == _TaskStatus.SUCCESS: create_success += 1 elif result.status == _TaskStatus.IN_PROGRESS: create_in_progress += 1 elif result.status == _TaskStatus.FAILED: create_failed += 1 elif result.status == _TaskStatus.SKIPPED: create_skipped += 1 await self._handle_task_result(handler, result, async_tasks) logger.info( f"[{handler.node_type}] create提交完成: success={create_success}, " f"in_progress={create_in_progress}, failed={create_failed}, skipped={create_skipped}" ) except Exception as e: logger.error("❌ [%s] create_all 异常: %s", handler.node_type, e) for node in create_nodes: ok = self._apply_sync_execute_state( node, handler_status=_TaskStatus.FAILED, detail=str(e), ) if not ok: raise RuntimeError(f"[{handler.node_type}] create_all exception cannot map to state machine") # 批量更新（传入节点列表） if update_nodes: try: update_results = await handler.update_all(update_nodes) for result in update_results: await self._handle_task_result(handler, result, async_tasks) except Exception as e: logger.error("❌ [%s] update_all 异常: %s", handler.node_type, e) for node in update_nodes: ok = self._apply_sync_execute_state( node, handler_status=_TaskStatus.FAILED, detail=str(e), ) if not ok: raise RuntimeError(f"[{handler.node_type}] update_all exception cannot map to state machine") # 批量删除（delete stage 当前未实现，检测到 S09 会在上方直接 fail-fast） # 轮询异步任务 if async_tasks and poll_async_tasks: logger.info( f"[{handler.node_type}] create已提交，push_id检查即将开始: pending={len(async_tasks)}, " f"first_wait={self._poll_interval:.3f}s" ) await self._poll_async_tasks(handler, async_tasks) # 统计同步结果 all_nodes = self._collection.filter(node_type=handler.node_type) synced_count = len([n for n in all_nodes if n.action != SyncAction.NONE]) success_count = len([n for n in all_nodes if n.status == SyncStatus.SUCCESS]) failed_count = len([n for n in all_nodes if n.status == SyncStatus.FAILED]) self._stats[handler.node_type]["synced"] = synced_count self._stats[handler.node_type]["success"] = success_count self._stats[handler.node_type]["failed"] = failed_count return async_tasks async def poll_async_tasks(self, async_tasks_by_type: Dict[str, Dict[str, str]]) -> None: """由 Pipeline 调度的异步任务轮询""" for node_type, async_tasks in async_tasks_by_type.items(): if not async_tasks: continue handler = self.get_handler(node_type) await self._poll_async_tasks(handler, async_tasks) async def _handle_task_result( self, handler: "NodeHandler", result: "TaskResult", async_tasks: Dict[str, str] ) -> None: """ 处理 Handler 返回的 TaskResult Args: collection: Collection 上下文 result: Handler 返回的结果（包含 node_id） async_tasks: 异步任务字典 """ from .task_result import TaskStatus as _TaskStatus if not result.node_id: # 没有 node_id，无法定位节点 return if self._collection is None: # Collection 未设置，无法处理结果 return # 从 self._collection 中查找节点 node = self._collection.get(result.node_id) if not node: return if result.status == _TaskStatus.SUCCESS: if await self._intercept_success_result(handler, node, result, async_tasks): return if result.status == _TaskStatus.SUCCESS: await self._handle_success_result(node, result) return if result.status == _TaskStatus.FAILED: self._on_failed_result(handler, node, result) await self._handle_failed_result(node, result) return if result.status == _TaskStatus.SKIPPED: await self._handle_skipped_result(node, result) return if result.status == _TaskStatus.IN_PROGRESS: self._on_in_progress_result(handler, node, result) await self._handle_in_progress_result(node, result, async_tasks) return def _write_back_data_id_to_data(self, node, data_id: str) -> None: data = node.get_data() if data is None: return schema = node.get_schema() candidate_keys = ["id", "_id"] model_fields = get_pydantic_model_fields(schema) for key in candidate_keys: if key in data: data[key] = data_id node.set_data(data) return if key in model_fields: data[key] = data_id node.set_data(data) return async def _apply_success_result(self, node, data_id: Optional[str], previous_data_id: str) -> None: from ..common.types import SyncAction if data_id and node.action == SyncAction.CREATE: if node.data_id != data_id: raise RuntimeError( f"[{node.node_type}] data_id must be updated in event apply path: expected={data_id}, actual={node.data_id}" ) self._write_back_data_id_to_data(node, data_id) if self._collection is not None and data_id != previous_data_id: await self._collection.update(node) if node.action == SyncAction.UPDATE: current_data = node.get_data() if current_data is not None: node.set_origin_data(current_data) if node.action == SyncAction.DELETE: if self._collection is None: return await self._collection.delete(node.node_id) return # 成功后 action 是否重置由上层策略决定 async def _poll_async_tasks( self, handler: "NodeHandler", async_tasks: Dict[str, str] ) -> None: """ 轮询异步任务直到全部完成 Args: handler: 节点处理器 collection: Collection 上下文 async_tasks: {node_id: task_id} 映射 流程： 1. 批量调用 Handler.poll_tasks() 2. 更新节点状态 3. 如果还有未完成任务，等待后重试 """ from .task_result import TaskStatus as _TaskStatus max_retries = self._poll_max_retries retry_count = 0 poll_interval = self._poll_interval # 轮询间隔（秒） if self._collection is None: # Collection 未设置，无法轮询任务 return if async_tasks and poll_interval > 0: logger.info( f"[{handler.node_type}] push_id首次检查前等待 {poll_interval:.3f}s, pending={len(async_tasks)}" ) await asyncio.sleep(poll_interval) while async_tasks and retry_count < max_retries: logger.info( f"[{handler.node_type}] push_id检查 {retry_count + 1}/{max_retries}, pending={len(async_tasks)}" ) # 批量查询 task_ids = list(async_tasks.values()) results = await handler.poll_tasks(task_ids) # 处理结果 completed_nodes = [] for node_id, task_id in list(async_tasks.items()): if task_id not in results: continue result = results[task_id] node = self._collection.get(node_id) if not node: completed_nodes.append(node_id) continue if result.status == _TaskStatus.SUCCESS: self._on_poll_result(handler, node, task_id, result) previous_data_id = node.data_id ok = self._apply_sync_execute_state( node, handler_status=result.status, detail=f"异步任务成功完成: {task_id}", result_data_id=result.data_id, ) if not ok: raise RuntimeError(f"[{node.node_type}] async SUCCESS cannot map to state machine: node={node.node_id}") self._record_action_result(node, result.status) await self._apply_success_result(node, result.data_id, previous_data_id=previous_data_id) node.error = None logger.info( f"[{handler.node_type}] polling成功: node={node.node_id}, push_id={task_id}, data_id={result.data_id or node.data_id}" ) completed_nodes.append(node_id) elif result.status == _TaskStatus.FAILED: self._on_poll_result(handler, node, task_id, result) ok = self._apply_sync_execute_state( node, handler_status=result.status, detail=f"异步任务失败: {result.error}", ) if not ok: raise RuntimeError(f"[{node.node_type}] async FAILED cannot map to state machine: node={node.node_id}") self._record_action_result(node, result.status) node.error = result.error completed_nodes.append(node_id) # IN_PROGRESS 继续等待 # 移除已完成的任务 for node_id in completed_nodes: async_tasks.pop(node_id, None) # 还有未完成任务，等待后重试 if async_tasks: retry_count += 1 # 超时未完成的任务标记为失败 if async_tasks: for node_id in async_tasks: node = self._collection.get(node_id) if node: timeout_msg = f"异步任务超时 (等待 {max_retries * poll_interval} 秒后仍未完成)" ok = self._apply_sync_execute_state( node, handler_status=_TaskStatus.FAILED, poll_timeout=True, detail=timeout_msg, ) if not ok: raise RuntimeError(f"[{node.node_type}] poll timeout cannot map to state machine: node={node.node_id}") node.error = f"Async task timeout after {max_retries * poll_interval} seconds" self._record_action_result(node, _TaskStatus.FAILED) # ========== 统计信息 ========== def get_sync_summary(self, collection: "DataCollection") -> Dict[str, Any]: """ 获取同步统计信息 Returns: { "total": 总节点数, "success": 成功数, "failed": 失败数, "in_progress": 进行中, "skipped": 跳过数, "by_type": {...} # 按类型统计 } """ nodes = list(collection._nodes.values()) summary = { "total": len(nodes), "success": sum(1 for n in nodes if n.status == SyncStatus.SUCCESS), "failed": sum(1 for n in nodes if n.status == SyncStatus.FAILED), "in_progress": sum(1 for n in nodes if n.status == SyncStatus.IN_PROGRESS), "skipped": sum(1 for n in nodes if n.status == SyncStatus.SKIPPED), "by_type": {} } # 按类型统计 for node_type in self._handlers.keys(): type_nodes = collection.filter(node_type=node_type) summary["by_type"][node_type] = { "total": len(type_nodes), "success": sum(1 for n in type_nodes if n.status == SyncStatus.SUCCESS), "failed": sum(1 for n in type_nodes if n.status == SyncStatus.FAILED), } return summary async def initialize(self) -> None: """初始化数据源资源（子类可覆盖）""" return None async def close(self) -> None: """关闭数据源资源（子类可覆盖以实现特定的清理逻辑）""" return None