From f5729d5a18dcc829cccfe4a29abae5d6cdccf8ac Mon Sep 17 00:00:00 2001 From: strepsiades Date: Fri, 3 Apr 2026 09:18:44 +0800 Subject: [PATCH] =?UTF-8?q?=E6=95=B4=E7=90=86=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/quality_cleanup_execution_plan.md | 68 +++++++++ run.py | 53 +------ .../datasource/api/datasource.py | 16 +- sync_state_machine/datasource/api/handler.py | 15 +- sync_state_machine/datasource/datasource.py | 6 + .../datasource/jsonl/handler.py | 18 ++- .../domain/attachment/api_handler.py | 5 +- .../domain/attachment/jsonl_handler.py | 2 +- .../domain/company/api_handler.py | 11 +- .../domain/company/jsonl_handler.py | 2 +- .../domain/construction/api_handler.py | 5 +- .../domain/construction/jsonl_handler.py | 2 +- .../construction_task_detail/api_handler.py | 5 +- .../construction_task_detail/jsonl_handler.py | 2 +- .../domain/contract/api_handler.py | 5 +- .../domain/contract/jsonl_handler.py | 2 +- .../domain/contract_change/api_handler.py | 6 +- .../domain/contract_change/jsonl_handler.py | 2 +- .../domain/contract_settlement/api_handler.py | 5 +- .../contract_settlement/jsonl_handler.py | 2 +- .../contract_settlement_detail/api_handler.py | 5 +- .../jsonl_handler.py | 2 +- sync_state_machine/domain/lar/api_handler.py | 5 +- .../domain/lar/jsonl_handler.py | 2 +- .../domain/lar_change/api_handler.py | 5 +- .../domain/lar_change/jsonl_handler.py | 2 +- .../domain/material/api_handler.py | 5 +- .../domain/material/jsonl_handler.py | 2 +- .../domain/material_detail/api_handler.py | 5 +- .../domain/material_detail/jsonl_handler.py | 2 +- .../domain/personnel/api_handler.py | 5 +- .../domain/personnel/jsonl_handler.py | 2 +- .../domain/personnel_detail/api_handler.py | 5 +- .../domain/personnel_detail/jsonl_handler.py | 2 +- .../domain/preparation/api_handler.py | 11 +- .../domain/preparation/jsonl_handler.py | 2 +- .../domain/production/api_handler.py | 5 +- .../domain/production/jsonl_handler.py | 2 +- .../domain/project/api_handler.py | 5 +- .../domain/project/jsonl_handler.py | 2 +- .../domain/project_detail/api_handler.py | 5 +- .../domain/project_detail/jsonl_handler.py | 2 +- .../domain/supplier/api_handler.py | 11 +- .../domain/supplier/jsonl_handler.py | 2 +- .../domain/units/api_handler.py | 5 +- .../domain/units/jsonl_handler.py | 2 +- sync_state_machine/domain/user/api_handler.py | 11 +- .../domain/user/jsonl_handler.py | 2 +- sync_state_machine/pipeline/factory.py | 112 +++++++++----- .../pipeline/full_sync_pipeline.py | 141 ++++++++++-------- 50 files changed, 305 insertions(+), 296 deletions(-) create mode 100644 docs/quality_cleanup_execution_plan.md diff --git a/docs/quality_cleanup_execution_plan.md b/docs/quality_cleanup_execution_plan.md new file mode 100644 index 0000000..cd0cd48 --- /dev/null +++ b/docs/quality_cleanup_execution_plan.md @@ -0,0 +1,68 @@ +# ECM Sync Quality Cleanup Execution Plan + +## Goal + +This cleanup pass keeps behavior unchanged and prioritizes deleting, merging, and unifying duplicated code on the single-pipeline path. + +Acceptance remains unchanged: + +- `backend/local_tests/sync_system_tests/depm_to_ecm_bootstrap_pull_then_project_push.pipeline.yaml` must still run successfully. +- Local Summary and Remote Summary must keep the same structure and stable output shape. + +## Constraints + +- Do not modify `schemas/`. +- Do not modify `filtered_datasource/`. +- Do not bypass schema validation. +- Do not add fallback logic that changes behavior. +- Prefer helper extraction, deletion, and merge over new abstraction layers. + +## Execution Order + +### Batch 1: Entry + Factory + +Status: completed + +- Turn `run.py` into a thin wrapper around `sync_state_machine.cli.main`. +- Merge duplicated datasource lifecycle and resolved-config logging in `sync_state_machine/pipeline/factory.py`. +- Verify with focused unit tests and the real DEPM to ECM profile. + +### Batch 2: Pipeline Flow + Backend Runner + +Status: completed + +- Reduce repeated phase-completion, strategy-loop, reload-skip, and close-handling code in `sync_state_machine/pipeline/full_sync_pipeline.py`. +- Reduce repeated orchestration code in `backend/app/thirdparty/ecm_sync/scripts/depm_to_ecm/run_pipeline.py` without changing CLI behavior or fail-fast rules. +- Keep log text stable where practical. + +### Batch 3: Datasource + Handler Base Cleanup + +Status: completed + +- Reduce duplicated handler registration and context injection between datasource base classes and subclasses. +- Reduce duplicated node metadata initialization across handler subclasses by pushing safe defaults into handler base classes. +- Delete domain-level constructor boilerplate where subclasses only repeat node type, node class, schema, or no-op `_create_node` wrappers. + +### Batch 4: Domain Repeat Cleanup + Docs + +Status: in progress + +- Inspect domain handlers for repeated load/update helper logic and merge only stable repeated fragments into shared helpers. +- Update docs so they only describe the real maintained execution path. +- Keep backend runner documentation aligned with current script behavior. +- Remove stale references only after code paths are confirmed stable. + +## Validation Plan + +1. Run targeted `ecm_sync_system` unit tests for the touched pipeline code. +2. Run the real backend profile with the backend interpreter. +3. Confirm the output still ends with pipeline completion plus Local Summary and Remote Summary tables. + +## Current Focus + +Current implementation focus is Batch 4. + +- repeated load/update helpers under `sync_state_machine/domain/` +- maintained execution-path docs under `docs/` + +The intent is code reduction only, not behavior redesign. \ No newline at end of file diff --git a/run.py b/run.py index 0fdf97e..a2d4687 100644 --- a/run.py +++ b/run.py @@ -1,8 +1,5 @@ from __future__ import annotations -import argparse -import asyncio -import logging import sys from pathlib import Path @@ -10,55 +7,7 @@ PROJECT_ROOT = Path(__file__).resolve().parent if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) -from sync_state_machine.config import load_overrides_from_file -from sync_state_machine.logging import get_logger, init_app_logger -from sync_state_machine.pipeline import run_profile_from_file - - -logger = get_logger(__name__) - - -def _resolve_config_path(raw_path: str) -> Path: - path = Path(raw_path) - if not path.is_absolute(): - path = PROJECT_ROOT / path - return path.resolve() - - -async def _main() -> int: - init_app_logger(replace_handlers=True) - parser = argparse.ArgumentParser(description="Run sync pipeline from config override file") - parser.add_argument( - "--config_path", - required=True, - help="Path to run profile file (JSON/YAML, supports relative path, e.g. run_profiles/jsonl_to_jsonl.local.json)", - ) - args = parser.parse_args() - - cfg_path = _resolve_config_path(args.config_path) - if not cfg_path.exists() or not cfg_path.is_file(): - raise FileNotFoundError(f"Config file not found: {cfg_path}") - - logger.info("🧩 Loaded config file: %s", cfg_path) - await run_profile_from_file( - project_root=PROJECT_ROOT, - config_path=cfg_path, - print_summary=True, - ) - return 0 - - -def main() -> int: - try: - return asyncio.run(_main()) - except KeyboardInterrupt: - logger.warning("⚠️ Interrupted by user") - return 130 - except Exception as exc: - logger.exception("❌ Pipeline failed: %s: %s", type(exc).__name__, exc) - return 1 - finally: - logging.shutdown() +from sync_state_machine.cli import main if __name__ == "__main__": diff --git a/sync_state_machine/datasource/api/datasource.py b/sync_state_machine/datasource/api/datasource.py index 5e6603f..b68e0cc 100644 --- a/sync_state_machine/datasource/api/datasource.py +++ b/sync_state_machine/datasource/api/datasource.py @@ -65,22 +65,10 @@ class ApiDataSource(BaseDataSource): """关闭数据源（关闭 HTTP 客户端）""" await self.client.close() - def add_handler(self, handler: NodeHandler) -> None: - """ - 添加业务 Handler - - Args: - handler: API Handler 实例 - """ + def _prepare_handler_for_registration(self, handler: NodeHandler) -> None: handler.set_api_client(self.client) handler.set_poll_mode(self.poll_mode) - - # 调用基类添加方法 - super().add_handler(handler) - - # 如果 collection 已设置，确保新 handler 拿到引用 - if self._collection is not None: - handler.set_collection(self._collection) # type: ignore[arg-type] + super()._prepare_handler_for_registration(handler) @staticmethod def _fmt_trace_value(value, max_len: int = 600) -> str: diff --git a/sync_state_machine/datasource/api/handler.py b/sync_state_machine/datasource/api/handler.py index 0551eb1..4ad2059 100644 --- a/sync_state_machine/datasource/api/handler.py +++ b/sync_state_machine/datasource/api/handler.py @@ -43,8 +43,17 @@ class BaseApiHandler(BaseNodeHandler[T]): _schema: Any def __init__(self, **handler_config: Any): - # 子类应该在调用 super().__init__() 之前或之后设置 _node_type, _node_class, _schema - super().__init__() + # 子类可直接传入 node_class；少数特殊场景可显式覆盖 schema/node_type。 + node_class = handler_config.pop("node_class", None) + node_type = handler_config.pop("node_type", None) + schema = handler_config.pop("schema", None) + + if node_class is not None and node_type is None: + node_type = getattr(node_class, "node_type", None) + if node_class is not None and schema is None: + schema = getattr(node_class, "schema", None) + + super().__init__(node_type=node_type, node_class=node_class, schema=schema) self.api_client: 'ApiClient' = None # type: ignore # 由 DataSource 注入 self.poll_mode: str = "async" # async | sync self._collection: 'DataCollection' = None # type: ignore # 由 DataSource 注入 @@ -67,7 +76,7 @@ class BaseApiHandler(BaseNodeHandler[T]): def set_collection(self, collection: 'DataCollection') -> None: """设置 Collection（由 DataSource 调用）""" - self._collection = collection + super().set_collection(collection) def ensure_api_client(self) -> 'ApiClient': if self.api_client is None: diff --git a/sync_state_machine/datasource/datasource.py b/sync_state_machine/datasource/datasource.py index 20a9358..d254870 100644 --- a/sync_state_machine/datasource/datasource.py +++ b/sync_state_machine/datasource/datasource.py @@ -163,6 +163,11 @@ class BaseDataSource(ABC): """Datasource-specific hook after a FAILED result is applied to node.""" return + def _prepare_handler_for_registration(self, handler: "NodeHandler") -> None: + """Allow subclasses to inject datasource-specific dependencies before registration.""" + if self._collection is not None: + handler.set_collection(self._collection) + async def _handle_success_result(self, node: "SyncNode", result: "TaskResult") -> None: previous_data_id = node.data_id ok = self._apply_sync_execute_state( @@ -371,6 +376,7 @@ class BaseDataSource(ABC): 示例： datasource.add_handler(ContractNodeHandler(api_client)) """ + self._prepare_handler_for_registration(handler) self._handlers[handler.node_type] = handler def get_handler(self, node_type: str) -> "NodeHandler": diff --git a/sync_state_machine/datasource/jsonl/handler.py b/sync_state_machine/datasource/jsonl/handler.py index fb91601..0263e5d 100644 --- a/sync_state_machine/datasource/jsonl/handler.py +++ b/sync_state_machine/datasource/jsonl/handler.py @@ -43,11 +43,21 @@ class BaseJsonlHandler(BaseNodeHandler): def __init__( self, datasource: JsonlDataSource, - node_type: str, - node_class: type, - schema: type + node_class_or_node_type, + node_class: type | None = None, + schema: type | None = None, + node_type: str | None = None, ): - super().__init__(node_type, node_class, schema) + if isinstance(node_class_or_node_type, str): + resolved_node_type = node_class_or_node_type + resolved_node_class = node_class + resolved_schema = schema + else: + resolved_node_class = node_class_or_node_type + resolved_node_type = node_type or getattr(resolved_node_class, "node_type", None) + resolved_schema = schema or getattr(resolved_node_class, "schema", None) + + super().__init__(resolved_node_type, resolved_node_class, resolved_schema) self.datasource = datasource async def load(self) -> List['SyncNode']: diff --git a/sync_state_machine/domain/attachment/api_handler.py b/sync_state_machine/domain/attachment/api_handler.py index 8a63b25..7b467ff 100644 --- a/sync_state_machine/domain/attachment/api_handler.py +++ b/sync_state_machine/domain/attachment/api_handler.py @@ -17,10 +17,7 @@ class AttachmentApiHandler(BaseApiHandler): """附件 API Handler""" def __init__(self): - super().__init__() - self._node_type = "attachment" - self._node_class = AttachmentSyncNode - self._schema = AttachmentSchema + super().__init__(node_class=AttachmentSyncNode, schema=AttachmentSchema) async def load(self) -> List[SyncNode]: """加载附件列表（依赖 Project）""" diff --git a/sync_state_machine/domain/attachment/jsonl_handler.py b/sync_state_machine/domain/attachment/jsonl_handler.py index c0f011d..91f8509 100644 --- a/sync_state_machine/domain/attachment/jsonl_handler.py +++ b/sync_state_machine/domain/attachment/jsonl_handler.py @@ -9,4 +9,4 @@ from .sync_node import AttachmentSyncNode class AttachmentJsonlHandler(BaseJsonlHandler): """附件 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "attachment", AttachmentSyncNode, Dict[str, Any]) + super().__init__(datasource, AttachmentSyncNode, schema=Dict[str, Any]) diff --git a/sync_state_machine/domain/company/api_handler.py b/sync_state_machine/domain/company/api_handler.py index af8272f..b6a17ad 100644 --- a/sync_state_machine/domain/company/api_handler.py +++ b/sync_state_machine/domain/company/api_handler.py @@ -6,7 +6,7 @@ Company API Handler (只读) - 不支持创建/更新/删除 """ -from typing import List, Dict, Any +from typing import List from ...datasource.api.handler import BaseApiHandler from ...datasource.task_result import TaskResult from ...common.sync_node import SyncNode @@ -18,10 +18,7 @@ class CompanyApiHandler(BaseApiHandler): """公司 API Handler (只读)""" def __init__(self): - super().__init__() - self._node_type = "company" - self._node_class = CompanySyncNode - self._schema = CompanyResponse + super().__init__(node_class=CompanySyncNode, schema=CompanyResponse) async def load(self) -> List[SyncNode]: """加载公司列表（支持分页）""" @@ -85,10 +82,6 @@ class CompanyApiHandler(BaseApiHandler): """轮询异步任务状态（占位）""" return await super().poll_tasks(task_ids) - def _create_node(self, data: Dict[str, Any]) -> SyncNode: - return self._create_basic_node(data, depend_ids=[]) - - # ========== 静态 API 函数 ========== async def api_get_company_list(client, page: int = 1, page_size: int = 100) -> tuple[List[CompanyResponse], int]: diff --git a/sync_state_machine/domain/company/jsonl_handler.py b/sync_state_machine/domain/company/jsonl_handler.py index f2d5f3c..af483ed 100644 --- a/sync_state_machine/domain/company/jsonl_handler.py +++ b/sync_state_machine/domain/company/jsonl_handler.py @@ -7,5 +7,5 @@ from schemas.company.company import CompanyResponse class CompanyJsonlHandler(BaseJsonlHandler): """公司 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "company", CompanySyncNode, CompanyResponse) + super().__init__(datasource, CompanySyncNode, schema=CompanyResponse) diff --git a/sync_state_machine/domain/construction/api_handler.py b/sync_state_machine/domain/construction/api_handler.py index 2e650d5..836446a 100644 --- a/sync_state_machine/domain/construction/api_handler.py +++ b/sync_state_machine/domain/construction/api_handler.py @@ -28,10 +28,7 @@ class ConstructionTaskApiHandler(BaseApiHandler): """施工任务 API Handler""" def __init__(self): - super().__init__() - self._node_type = "construction_task" - self._node_class = ConstructionTaskSyncNode - self._schema = ConstructionResponse + super().__init__(node_class=ConstructionTaskSyncNode, schema=ConstructionResponse) self.update_schemas = [ConstructionCreate, ConstructionUpdate, ConstructionPlanUpdate] async def load(self) -> List[SyncNode]: diff --git a/sync_state_machine/domain/construction/jsonl_handler.py b/sync_state_machine/domain/construction/jsonl_handler.py index c9c0da6..915ab85 100644 --- a/sync_state_machine/domain/construction/jsonl_handler.py +++ b/sync_state_machine/domain/construction/jsonl_handler.py @@ -7,4 +7,4 @@ from schemas.construction.construction import ConstructionResponse class ConstructionTaskJsonlHandler(BaseJsonlHandler): """施工任务 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "construction_task", ConstructionTaskSyncNode, ConstructionResponse) + super().__init__(datasource, ConstructionTaskSyncNode, schema=ConstructionResponse) diff --git a/sync_state_machine/domain/construction_task_detail/api_handler.py b/sync_state_machine/domain/construction_task_detail/api_handler.py index 9f1cf4f..6a2db44 100644 --- a/sync_state_machine/domain/construction_task_detail/api_handler.py +++ b/sync_state_machine/domain/construction_task_detail/api_handler.py @@ -24,10 +24,7 @@ class ConstructionTaskDetailApiHandler(BaseApiHandler): """施工明细 API Handler""" def __init__(self, **handler_config: Any): - super().__init__(**handler_config) - self._node_type = "construction_task_detail" - self._node_class = ConstructionTaskDetailSyncNode - self._schema = ConstructionDetailResponse + super().__init__(node_class=ConstructionTaskDetailSyncNode, schema=ConstructionDetailResponse, **handler_config) self.update_schemas = [ConstructionDetailCreate, ConstructionDetailUpdate] def _use_aggregate_load(self) -> bool: diff --git a/sync_state_machine/domain/construction_task_detail/jsonl_handler.py b/sync_state_machine/domain/construction_task_detail/jsonl_handler.py index 4506164..566497d 100644 --- a/sync_state_machine/domain/construction_task_detail/jsonl_handler.py +++ b/sync_state_machine/domain/construction_task_detail/jsonl_handler.py @@ -9,4 +9,4 @@ from schemas.construction.construction_detail import ConstructionDetailResponse class ConstructionTaskDetailJsonlHandler(BaseJsonlHandler): """施工任务明细 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "construction_task_detail", ConstructionTaskDetailSyncNode, ConstructionDetailResponse) + super().__init__(datasource, ConstructionTaskDetailSyncNode, schema=ConstructionDetailResponse) diff --git a/sync_state_machine/domain/contract/api_handler.py b/sync_state_machine/domain/contract/api_handler.py index 466b6fa..02facce 100644 --- a/sync_state_machine/domain/contract/api_handler.py +++ b/sync_state_machine/domain/contract/api_handler.py @@ -101,10 +101,7 @@ class ContractApiHandler(BaseApiHandler): """合同 API Handler""" def __init__(self): - super().__init__() - self._node_type = "contract" - self._node_class = ContractSyncNode - self._schema = ContractResponse + super().__init__(node_class=ContractSyncNode, schema=ContractResponse) self.update_schemas = [ContractCreate, ContractUpdate] # ========== Handler 接口实现 ========== diff --git a/sync_state_machine/domain/contract/jsonl_handler.py b/sync_state_machine/domain/contract/jsonl_handler.py index 8235f32..5503e31 100644 --- a/sync_state_machine/domain/contract/jsonl_handler.py +++ b/sync_state_machine/domain/contract/jsonl_handler.py @@ -9,6 +9,6 @@ from schemas.contract.contract import ContractResponse class ContractJsonlHandler(BaseJsonlHandler): """合同 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "contract", ContractSyncNode, ContractResponse) + super().__init__(datasource, ContractSyncNode, schema=ContractResponse) diff --git a/sync_state_machine/domain/contract_change/api_handler.py b/sync_state_machine/domain/contract_change/api_handler.py index 000c279..3109765 100644 --- a/sync_state_machine/domain/contract_change/api_handler.py +++ b/sync_state_machine/domain/contract_change/api_handler.py @@ -24,11 +24,7 @@ class ContractChangeApiHandler(BaseApiHandler): """合同变更 API Handler""" def __init__(self): - super().__init__() - self._node_type = "contract_change" - self._node_class = ContractChangeSyncNode - # TODO: ContractChangeResponse schema in schemas/project_extensions/contract_change.py is missing 'project_id' - self._schema = ContractChangeResponse + super().__init__(node_class=ContractChangeSyncNode, schema=ContractChangeResponse) async def load(self) -> List[SyncNode]: """加载合同变更数据""" diff --git a/sync_state_machine/domain/contract_change/jsonl_handler.py b/sync_state_machine/domain/contract_change/jsonl_handler.py index 94184b1..279fa53 100644 --- a/sync_state_machine/domain/contract_change/jsonl_handler.py +++ b/sync_state_machine/domain/contract_change/jsonl_handler.py @@ -7,4 +7,4 @@ from schemas.project_extensions.contract_change import ContractChangeResponse class ContractChangeJsonlHandler(BaseJsonlHandler): """合同变更 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "contract_change", ContractChangeSyncNode, ContractChangeResponse) + super().__init__(datasource, ContractChangeSyncNode, schema=ContractChangeResponse) diff --git a/sync_state_machine/domain/contract_settlement/api_handler.py b/sync_state_machine/domain/contract_settlement/api_handler.py index 00912fe..355cf00 100644 --- a/sync_state_machine/domain/contract_settlement/api_handler.py +++ b/sync_state_machine/domain/contract_settlement/api_handler.py @@ -28,10 +28,7 @@ class ContractSettlementApiHandler(BaseApiHandler): """合同结算 API Handler""" def __init__(self): - super().__init__() - self._node_type = "contract_settlement" - self._node_class = ContractSettlementSyncNode - self._schema = ContractSettlementResponse + super().__init__(node_class=ContractSettlementSyncNode, schema=ContractSettlementResponse) self.update_schemas = [ContractSettlementCreate, ContractSettlementUpdate, ContractSettlementPlanUpdate] async def load(self) -> List[SyncNode]: diff --git a/sync_state_machine/domain/contract_settlement/jsonl_handler.py b/sync_state_machine/domain/contract_settlement/jsonl_handler.py index 28b64f8..2dece5b 100644 --- a/sync_state_machine/domain/contract_settlement/jsonl_handler.py +++ b/sync_state_machine/domain/contract_settlement/jsonl_handler.py @@ -7,4 +7,4 @@ from schemas.contract_settlement.contract_settlement import ContractSettlementRe class ContractSettlementJsonlHandler(BaseJsonlHandler): """合同结算 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "contract_settlement", ContractSettlementSyncNode, ContractSettlementResponse) + super().__init__(datasource, ContractSettlementSyncNode, schema=ContractSettlementResponse) diff --git a/sync_state_machine/domain/contract_settlement_detail/api_handler.py b/sync_state_machine/domain/contract_settlement_detail/api_handler.py index d5ccd88..b017ef8 100644 --- a/sync_state_machine/domain/contract_settlement_detail/api_handler.py +++ b/sync_state_machine/domain/contract_settlement_detail/api_handler.py @@ -24,10 +24,7 @@ class ContractSettlementDetailApiHandler(BaseApiHandler): """合同结算明细 API Handler""" def __init__(self, **handler_config: Any): - super().__init__(**handler_config) - self._node_type = "contract_settlement_detail" - self._node_class = ContractSettlementDetailSyncNode - self._schema = ContractSettlementDetailResponse + super().__init__(node_class=ContractSettlementDetailSyncNode, schema=ContractSettlementDetailResponse, **handler_config) self.update_schemas = [ContractSettlementDetailCreate, ContractSettlementDetailUpdate] def _use_aggregate_load(self) -> bool: diff --git a/sync_state_machine/domain/contract_settlement_detail/jsonl_handler.py b/sync_state_machine/domain/contract_settlement_detail/jsonl_handler.py index 81e3bcd..9be75dc 100644 --- a/sync_state_machine/domain/contract_settlement_detail/jsonl_handler.py +++ b/sync_state_machine/domain/contract_settlement_detail/jsonl_handler.py @@ -9,4 +9,4 @@ from schemas.contract_settlement.contract_settlement_detail import ContractSettl class ContractSettlementDetailJsonlHandler(BaseJsonlHandler): """合同结算明细 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "contract_settlement_detail", ContractSettlementDetailSyncNode, ContractSettlementDetailResponse) + super().__init__(datasource, ContractSettlementDetailSyncNode, schema=ContractSettlementDetailResponse) diff --git a/sync_state_machine/domain/lar/api_handler.py b/sync_state_machine/domain/lar/api_handler.py index 5812b75..dac9b67 100644 --- a/sync_state_machine/domain/lar/api_handler.py +++ b/sync_state_machine/domain/lar/api_handler.py @@ -31,10 +31,7 @@ class LarApiHandler(BaseApiHandler): """移民征地 API Handler""" def __init__(self): - super().__init__() - self._node_type = "lar" - self._node_class = LarSyncNode - self._schema = LarDetailSchema + super().__init__(node_class=LarSyncNode, schema=LarDetailSchema) async def load(self) -> List[SyncNode]: """ diff --git a/sync_state_machine/domain/lar/jsonl_handler.py b/sync_state_machine/domain/lar/jsonl_handler.py index 52f0682..24ee8f9 100644 --- a/sync_state_machine/domain/lar/jsonl_handler.py +++ b/sync_state_machine/domain/lar/jsonl_handler.py @@ -7,4 +7,4 @@ from schemas.project_extensions.lar import LarDetailSchema class LarJsonlHandler(BaseJsonlHandler): """移民征地 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "lar", LarSyncNode, LarDetailSchema) + super().__init__(datasource, LarSyncNode, schema=LarDetailSchema) diff --git a/sync_state_machine/domain/lar_change/api_handler.py b/sync_state_machine/domain/lar_change/api_handler.py index ff4d175..66cbf2c 100644 --- a/sync_state_machine/domain/lar_change/api_handler.py +++ b/sync_state_machine/domain/lar_change/api_handler.py @@ -28,10 +28,7 @@ class LarChangeApiHandler(BaseApiHandler): """移民变更 API Handler""" def __init__(self): - super().__init__() - self._node_type = "lar_change" - self._node_class = LarChangeSyncNode - self._schema = LarChangeDetailSchema + super().__init__(node_class=LarChangeSyncNode, schema=LarChangeDetailSchema) async def load(self) -> List[SyncNode]: """ diff --git a/sync_state_machine/domain/lar_change/jsonl_handler.py b/sync_state_machine/domain/lar_change/jsonl_handler.py index b9aaf9b..138ffdc 100644 --- a/sync_state_machine/domain/lar_change/jsonl_handler.py +++ b/sync_state_machine/domain/lar_change/jsonl_handler.py @@ -9,4 +9,4 @@ class LarChangeJsonlHandler(BaseJsonlHandler): """LAR Change (移民变更) JSONL 数据处理器""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "lar_change", LarChangeSyncNode, LarChangeDetailSchema) + super().__init__(datasource, LarChangeSyncNode, schema=LarChangeDetailSchema) diff --git a/sync_state_machine/domain/material/api_handler.py b/sync_state_machine/domain/material/api_handler.py index a8d6a0e..cdef176 100644 --- a/sync_state_machine/domain/material/api_handler.py +++ b/sync_state_machine/domain/material/api_handler.py @@ -23,10 +23,7 @@ class MaterialApiHandler(BaseApiHandler): """物资 API Handler""" def __init__(self): - super().__init__() - self._node_type = "material" - self._node_class = MaterialSyncNode - self._schema = MaterialResponse + super().__init__(node_class=MaterialSyncNode, schema=MaterialResponse) self.update_schemas = [MaterialCreate, MaterialUpdate, MaterialPlanUpdate] async def load(self) -> List[SyncNode]: diff --git a/sync_state_machine/domain/material/jsonl_handler.py b/sync_state_machine/domain/material/jsonl_handler.py index f7f4889..ecdeb5d 100644 --- a/sync_state_machine/domain/material/jsonl_handler.py +++ b/sync_state_machine/domain/material/jsonl_handler.py @@ -7,4 +7,4 @@ from schemas.material.material import MaterialResponse class MaterialJsonlHandler(BaseJsonlHandler): """物资 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "material", MaterialSyncNode, MaterialResponse) + super().__init__(datasource, MaterialSyncNode, schema=MaterialResponse) diff --git a/sync_state_machine/domain/material_detail/api_handler.py b/sync_state_machine/domain/material_detail/api_handler.py index 262ae88..f0f6f49 100644 --- a/sync_state_machine/domain/material_detail/api_handler.py +++ b/sync_state_machine/domain/material_detail/api_handler.py @@ -39,10 +39,7 @@ class MaterialDetailApiHandler(BaseApiHandler): """物资明细 API Handler""" def __init__(self, **handler_config: Any): - super().__init__(**handler_config) - self._node_type = "material_detail" - self._node_class = MaterialDetailSyncNode - self._schema = MaterialDetailResponse + super().__init__(node_class=MaterialDetailSyncNode, schema=MaterialDetailResponse, **handler_config) self.update_schemas = [MaterialDetailCreate, MaterialDetailUpdate] def _use_aggregate_load(self) -> bool: diff --git a/sync_state_machine/domain/material_detail/jsonl_handler.py b/sync_state_machine/domain/material_detail/jsonl_handler.py index 2d64979..bc34876 100644 --- a/sync_state_machine/domain/material_detail/jsonl_handler.py +++ b/sync_state_machine/domain/material_detail/jsonl_handler.py @@ -9,4 +9,4 @@ from schemas.material.material_detail import MaterialDetailResponse class MaterialDetailJsonlHandler(BaseJsonlHandler): """物资明细 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "material_detail", MaterialDetailSyncNode, MaterialDetailResponse) + super().__init__(datasource, MaterialDetailSyncNode, schema=MaterialDetailResponse) diff --git a/sync_state_machine/domain/personnel/api_handler.py b/sync_state_machine/domain/personnel/api_handler.py index 790f607..d3c3dd3 100644 --- a/sync_state_machine/domain/personnel/api_handler.py +++ b/sync_state_machine/domain/personnel/api_handler.py @@ -19,10 +19,7 @@ class PersonnelApiHandler(BaseApiHandler): """人员 API Handler""" def __init__(self): - super().__init__() - self._node_type = "personnel" - self._node_class = PersonnelSyncNode - self._schema = PowerResponse + super().__init__(node_class=PersonnelSyncNode, schema=PowerResponse) self.update_schemas = [PowerCreate, PowerUpdate, PowerPlanUpdate] async def load(self) -> List[SyncNode]: diff --git a/sync_state_machine/domain/personnel/jsonl_handler.py b/sync_state_machine/domain/personnel/jsonl_handler.py index 0ec04ca..9ca6529 100644 --- a/sync_state_machine/domain/personnel/jsonl_handler.py +++ b/sync_state_machine/domain/personnel/jsonl_handler.py @@ -7,4 +7,4 @@ from schemas.power.power import PowerResponse class PersonnelJsonlHandler(BaseJsonlHandler): """人员 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "personnel", PersonnelSyncNode, PowerResponse) + super().__init__(datasource, PersonnelSyncNode, schema=PowerResponse) diff --git a/sync_state_machine/domain/personnel_detail/api_handler.py b/sync_state_machine/domain/personnel_detail/api_handler.py index b119241..bb249ac 100644 --- a/sync_state_machine/domain/personnel_detail/api_handler.py +++ b/sync_state_machine/domain/personnel_detail/api_handler.py @@ -20,10 +20,7 @@ class PersonnelDetailApiHandler(BaseApiHandler): """人员明细 API Handler""" def __init__(self, **handler_config: Any): - super().__init__(**handler_config) - self._node_type = "personnel_detail" - self._node_class = PersonnelDetailSyncNode - self._schema = PowerDetail + super().__init__(node_class=PersonnelDetailSyncNode, schema=PowerDetail, **handler_config) self.update_schemas = [PowerDetailCreate, PowerDetailUpdate] def _use_aggregate_load(self) -> bool: diff --git a/sync_state_machine/domain/personnel_detail/jsonl_handler.py b/sync_state_machine/domain/personnel_detail/jsonl_handler.py index 1368a09..2a8eaaf 100644 --- a/sync_state_machine/domain/personnel_detail/jsonl_handler.py +++ b/sync_state_machine/domain/personnel_detail/jsonl_handler.py @@ -9,4 +9,4 @@ from schemas.power.power_detail import PowerDetail class PersonnelDetailJsonlHandler(BaseJsonlHandler): """人员明细 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "personnel_detail", PersonnelDetailSyncNode, PowerDetail) + super().__init__(datasource, PersonnelDetailSyncNode, schema=PowerDetail) diff --git a/sync_state_machine/domain/preparation/api_handler.py b/sync_state_machine/domain/preparation/api_handler.py index 13e7001..95ea328 100644 --- a/sync_state_machine/domain/preparation/api_handler.py +++ b/sync_state_machine/domain/preparation/api_handler.py @@ -10,7 +10,7 @@ PUT: - /preparation - 更新里程碑数据 """ -from typing import List, Dict, Any +from typing import List, Dict from ...datasource.api.handler import BaseApiHandler from ...datasource.task_result import TaskResult from ...common.sync_node import SyncNode @@ -23,10 +23,7 @@ class PreparationApiHandler(BaseApiHandler): """里程碑 API Handler""" def __init__(self): - super().__init__() - self._node_type = "preparation" - self._node_class = PreparationSyncNode - self._schema = PreparationDetail + super().__init__(node_class=PreparationSyncNode, schema=PreparationDetail) self.update_schemas = [PostPreparation] async def load(self) -> List[SyncNode]: @@ -141,10 +138,6 @@ class PreparationApiHandler(BaseApiHandler): for node in nodes ] - def _create_node(self, data: Dict[str, Any]) -> SyncNode: - return self._create_basic_node(data, depend_ids=[]) - - # ========== 静态 API 函数 ========== async def api_update_preparation( diff --git a/sync_state_machine/domain/preparation/jsonl_handler.py b/sync_state_machine/domain/preparation/jsonl_handler.py index 5ee7df4..5b01dd5 100644 --- a/sync_state_machine/domain/preparation/jsonl_handler.py +++ b/sync_state_machine/domain/preparation/jsonl_handler.py @@ -8,4 +8,4 @@ from schemas.project_extensions.preparation import PreparationDetail class PreparationJsonlHandler(BaseJsonlHandler): """里程碑 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "preparation", PreparationSyncNode, PreparationDetail) + super().__init__(datasource, PreparationSyncNode, schema=PreparationDetail) diff --git a/sync_state_machine/domain/production/api_handler.py b/sync_state_machine/domain/production/api_handler.py index 8e03307..9bde722 100644 --- a/sync_state_machine/domain/production/api_handler.py +++ b/sync_state_machine/domain/production/api_handler.py @@ -23,10 +23,7 @@ class ProductionApiHandler(BaseApiHandler): """投产节点 API Handler""" def __init__(self): - super().__init__() - self._node_type = "production" - self._node_class = ProductionSyncNode - self._schema = ProductionDetail + super().__init__(node_class=ProductionSyncNode, schema=ProductionDetail) self.update_schemas = [PostProduction] async def load(self) -> List[SyncNode]: diff --git a/sync_state_machine/domain/production/jsonl_handler.py b/sync_state_machine/domain/production/jsonl_handler.py index b7a3ba9..66dd0a3 100644 --- a/sync_state_machine/domain/production/jsonl_handler.py +++ b/sync_state_machine/domain/production/jsonl_handler.py @@ -7,4 +7,4 @@ from schemas.project_extensions.production import ProductionDetail class ProductionJsonlHandler(BaseJsonlHandler): """投产节点 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "production", ProductionSyncNode, ProductionDetail) + super().__init__(datasource, ProductionSyncNode, schema=ProductionDetail) diff --git a/sync_state_machine/domain/project/api_handler.py b/sync_state_machine/domain/project/api_handler.py index 3055463..b42322c 100644 --- a/sync_state_machine/domain/project/api_handler.py +++ b/sync_state_machine/domain/project/api_handler.py @@ -73,10 +73,7 @@ class ProjectApiHandler(BaseApiHandler): _all_info_cache: Dict[str, Dict[str, Any]] = {} def __init__(self, **handler_config: Any): - super().__init__(**handler_config) - self._node_type = "project" - self._node_class = ProjectSyncNode - self._schema = ProjectResponseBase + super().__init__(node_class=ProjectSyncNode, schema=ProjectResponseBase, **handler_config) @classmethod async def get_all_info(cls, client, project_id: str) -> Dict[str, Any]: diff --git a/sync_state_machine/domain/project/jsonl_handler.py b/sync_state_machine/domain/project/jsonl_handler.py index 89d44ed..c622517 100644 --- a/sync_state_machine/domain/project/jsonl_handler.py +++ b/sync_state_machine/domain/project/jsonl_handler.py @@ -9,4 +9,4 @@ from schemas.project.project_base import ProjectResponseBase class ProjectJsonlHandler(BaseJsonlHandler): """项目 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "project", ProjectSyncNode, ProjectResponseBase) + super().__init__(datasource, ProjectSyncNode, schema=ProjectResponseBase) diff --git a/sync_state_machine/domain/project_detail/api_handler.py b/sync_state_machine/domain/project_detail/api_handler.py index 702db14..8e1ab6b 100644 --- a/sync_state_machine/domain/project_detail/api_handler.py +++ b/sync_state_machine/domain/project_detail/api_handler.py @@ -49,10 +49,7 @@ class ProjectDetailApiHandler(BaseApiHandler): """项目容量信息 API Handler""" def __init__(self): - super().__init__() - self._node_type = "project_detail_data" - self._node_class = ProjectDetailSyncNode - self._schema = ProjectDetailProject + super().__init__(node_class=ProjectDetailSyncNode, schema=ProjectDetailProject) self._pending_push_context: Dict[str, Dict[str, str]] = {} async def load(self) -> List[SyncNode]: diff --git a/sync_state_machine/domain/project_detail/jsonl_handler.py b/sync_state_machine/domain/project_detail/jsonl_handler.py index 81c1559..24e5231 100644 --- a/sync_state_machine/domain/project_detail/jsonl_handler.py +++ b/sync_state_machine/domain/project_detail/jsonl_handler.py @@ -7,4 +7,4 @@ from schemas.project_extensions.project_detail import ProjectDetailProject class ProjectDetailJsonlHandler(BaseJsonlHandler): """项目容量信息 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "project_detail_data", ProjectDetailSyncNode, ProjectDetailProject) + super().__init__(datasource, ProjectDetailSyncNode, schema=ProjectDetailProject) diff --git a/sync_state_machine/domain/supplier/api_handler.py b/sync_state_machine/domain/supplier/api_handler.py index cc6a7cf..4bfb349 100644 --- a/sync_state_machine/domain/supplier/api_handler.py +++ b/sync_state_machine/domain/supplier/api_handler.py @@ -6,7 +6,7 @@ Supplier API Handler (只读) - 不支持创建/更新/删除 """ -from typing import List, Dict, Any +from typing import List, Dict from ...datasource.api.handler import BaseApiHandler from ...datasource.task_result import TaskResult from ...common.sync_node import SyncNode @@ -18,10 +18,7 @@ class SupplierApiHandler(BaseApiHandler): """供应商 API Handler (只读)""" def __init__(self, load_mode: str = "api", max_total: int | None = None): - super().__init__() - self._node_type = "supplier" - self._node_class = SupplierSyncNode - self._schema = SupplierDetailResponse + super().__init__(node_class=SupplierSyncNode, schema=SupplierDetailResponse) self._load_mode = load_mode self._max_total = max_total @@ -99,10 +96,6 @@ class SupplierApiHandler(BaseApiHandler): """轮询异步任务状态（占位）""" return await super().poll_tasks(task_ids) - def _create_node(self, data: Dict[str, Any]) -> SyncNode: - return self._create_basic_node(data, depend_ids=[]) - - # ========== 静态 API 函数 ========== async def api_get_supplier_list(client, page: int = 1, page_size: int = 100) -> tuple[List[SupplierDetailResponse], int]: diff --git a/sync_state_machine/domain/supplier/jsonl_handler.py b/sync_state_machine/domain/supplier/jsonl_handler.py index bcda6d1..adc19c2 100644 --- a/sync_state_machine/domain/supplier/jsonl_handler.py +++ b/sync_state_machine/domain/supplier/jsonl_handler.py @@ -9,4 +9,4 @@ from schemas.supplier.supplier import SupplierDetailResponse class SupplierJsonlHandler(BaseJsonlHandler): """供应商 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "supplier", SupplierSyncNode, SupplierDetailResponse) + super().__init__(datasource, SupplierSyncNode, schema=SupplierDetailResponse) diff --git a/sync_state_machine/domain/units/api_handler.py b/sync_state_machine/domain/units/api_handler.py index 0b63609..6516fad 100644 --- a/sync_state_machine/domain/units/api_handler.py +++ b/sync_state_machine/domain/units/api_handler.py @@ -23,10 +23,7 @@ class UnitsApiHandler(BaseApiHandler): """机组 API Handler""" def __init__(self): - super().__init__() - self._node_type = "units" - self._node_class = UnitsSyncNode - self._schema = GeneratorUnitProject + super().__init__(node_class=UnitsSyncNode, schema=GeneratorUnitProject) async def load(self) -> List[SyncNode]: """加载机组数据（从 project all_info 中提取）""" diff --git a/sync_state_machine/domain/units/jsonl_handler.py b/sync_state_machine/domain/units/jsonl_handler.py index 8ae40fc..251c662 100644 --- a/sync_state_machine/domain/units/jsonl_handler.py +++ b/sync_state_machine/domain/units/jsonl_handler.py @@ -11,7 +11,7 @@ from schemas.project_extensions.generator_unit import GeneratorUnitProject class UnitsJsonlHandler(BaseJsonlHandler): """机组 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "units", UnitsSyncNode, GeneratorUnitProject) + super().__init__(datasource, UnitsSyncNode, schema=GeneratorUnitProject) async def load(self) -> List['SyncNode']: """加载机组数据（支持 generator_unit_N.jsonl 与 units_N.jsonl）""" diff --git a/sync_state_machine/domain/user/api_handler.py b/sync_state_machine/domain/user/api_handler.py index 7355117..14ba01a 100644 --- a/sync_state_machine/domain/user/api_handler.py +++ b/sync_state_machine/domain/user/api_handler.py @@ -10,7 +10,7 @@ GET: - /user/list - 获取用户列表 """ -from typing import List, Dict, Any +from typing import List, Dict from ...datasource.api.handler import BaseApiHandler from ...datasource.task_result import TaskResult from ...common.sync_node import SyncNode @@ -22,10 +22,7 @@ class UserApiHandler(BaseApiHandler): """用户 API Handler (只读)""" def __init__(self): - super().__init__() - self._node_type = "user" - self._node_class = UserSyncNode - self._schema = UserListDetailItem + super().__init__(node_class=UserSyncNode, schema=UserListDetailItem) async def load(self) -> List[SyncNode]: """加载用户列表（支持分页）""" @@ -87,10 +84,6 @@ class UserApiHandler(BaseApiHandler): """轮询异步任务状态（占位）""" return await super().poll_tasks(task_ids) - def _create_node(self, data: Dict[str, Any]) -> SyncNode: - return self._create_basic_node(data, depend_ids=[]) - - # ========== 静态 API 函数 ========== async def api_get_user_list(client, page: int = 1, page_size: int = 100) -> tuple[List[UserListDetailItem], int]: diff --git a/sync_state_machine/domain/user/jsonl_handler.py b/sync_state_machine/domain/user/jsonl_handler.py index 09ad944..89b91c4 100644 --- a/sync_state_machine/domain/user/jsonl_handler.py +++ b/sync_state_machine/domain/user/jsonl_handler.py @@ -7,4 +7,4 @@ from schemas.user.user import UserListDetailItem class UserJsonlHandler(BaseJsonlHandler): """用户 JSONL Handler""" def __init__(self, datasource: JsonlDataSource): - super().__init__(datasource, "user", UserSyncNode, UserListDetailItem) + super().__init__(datasource, UserSyncNode, schema=UserListDetailItem) diff --git a/sync_state_machine/pipeline/factory.py b/sync_state_machine/pipeline/factory.py index 73b5cba..43b6ed2 100644 --- a/sync_state_machine/pipeline/factory.py +++ b/sync_state_machine/pipeline/factory.py @@ -36,11 +36,9 @@ from ..config import ( build_config, StrategyRuntimeConfig, build_config_from_file, - build_multi_project_config_from_file, apply_config_overrides, OrphanAction, UpdateDirection, - resolve_log_file_path, config_snapshot, is_multi_project_payload, load_overrides_from_file, @@ -126,6 +124,72 @@ async def _initialize_datasource(datasource, *, endpoint_name: str) -> None: except Exception as exc: raise RuntimeError(f"Failed to initialize {endpoint_name} datasource: {type(exc).__name__}: {exc}") from exc + +async def _create_and_initialize_datasource( + ds_config: DataSourceConfig, + *, + datasource_override=None, + endpoint_name: str, + project_root: Path | None = None, +): + datasource = await _create_datasource( + ds_config, + datasource_override=datasource_override, + endpoint_name=endpoint_name, + project_root=project_root, + ) + await _initialize_datasource(datasource, endpoint_name=endpoint_name) + return datasource + + +async def _safe_close_resource(name: str, resource) -> None: + if resource is None: + return + try: + await asyncio.wait_for(resource.close(), timeout=3.0) + except Exception as close_exc: + logger.warning("Failed to close %s during create cleanup: %s", name, close_exc) + + +def _resolved_config_summary(config: PipelineRunConfig) -> tuple[str, str]: + summary = ( + "Resolved Full Config Summary: node_types=%d local=%s remote=%s", + len(config.node_types), + config.local_datasource.type, + config.remote_datasource.type, + ) + return summary[0], json.dumps(config_snapshot(config), ensure_ascii=False, indent=2) + + +def _log_resolved_config(config: PipelineRunConfig) -> tuple[str, str]: + summary_template, resolved_cfg_text = _resolved_config_summary(config) + logger.info( + summary_template, + len(config.node_types), + config.local_datasource.type, + config.remote_datasource.type, + ) + logger.info("Resolved Full Config:\n%s", resolved_cfg_text) + return summary_template, resolved_cfg_text + + +def _log_pipeline_start(config: PipelineRunConfig, *, title: str) -> None: + logger.info("\n%s", "=" * 80) + logger.info("🚀 Creating %s", title) + logger.info("📝 Log file: %s", config.logging.file_path or "-") + logger.info( + "🔧 Resolved Full Config Summary: node_types=%d local=%s remote=%s", + len(config.node_types), + config.local_datasource.type, + config.remote_datasource.type, + ) + logger.info("=" * 80) + + +def _log_pipeline_end(config: PipelineRunConfig, *, stats: Dict[str, Any]) -> None: + logger.info("✅ Pipeline completed: stats=%s", stats) + logger.info("🗃️ Persistence DB: %s", config.persist.db_path) + ensure_builtin_datasource_types_registered() @@ -284,20 +348,18 @@ async def create_pipeline_from_config( persistence = create_persistence_backend(config.persist) await persistence.initialize() - local_ds = await _create_datasource( + local_ds = await _create_and_initialize_datasource( config.local_datasource, datasource_override=local_datasource_override, endpoint_name="local", project_root=_project_root(), ) - await _initialize_datasource(local_ds, endpoint_name="local") - remote_ds = await _create_datasource( + remote_ds = await _create_and_initialize_datasource( config.remote_datasource, datasource_override=remote_datasource_override, endpoint_name="remote", project_root=_project_root(), ) - await _initialize_datasource(remote_ds, endpoint_name="remote") _add_handlers(config, local_ds, remote_ds, all_types) local_collection, remote_collection, binding_manager = _build_collections_and_bindings( @@ -332,17 +394,9 @@ async def create_pipeline_from_config( return pipeline except Exception: - async def _safe_close(name: str, resource) -> None: - if resource is None: - return - try: - await asyncio.wait_for(resource.close(), timeout=3.0) - except Exception as close_exc: - logger.warning("Failed to close %s during create cleanup: %s", name, close_exc) - - await _safe_close("remote_datasource", remote_ds) - await _safe_close("local_datasource", local_ds) - await _safe_close("persistence", persistence) + await _safe_close_resource("remote_datasource", remote_ds) + await _safe_close_resource("local_datasource", local_ds) + await _safe_close_resource("persistence", persistence) raise @@ -358,29 +412,12 @@ async def run_pipeline_from_config( ) -> Dict[str, Any]: ensure_app_logging(config.logging) - resolved_cfg = config_snapshot(config) - resolved_cfg_text = json.dumps(resolved_cfg, ensure_ascii=False, indent=2) - logger.info( - "Resolved Full Config Summary: node_types=%d local=%s remote=%s", - len(config.node_types), - config.local_datasource.type, - config.remote_datasource.type, - ) - logger.info("Resolved Full Config:\n%s", resolved_cfg_text) + _log_resolved_config(config) pipeline: Optional[FullSyncPipeline] = None try: if print_summary: - logger.info("\n%s", "=" * 80) - logger.info("🚀 Creating %s", title) - logger.info("📝 Log file: %s", config.logging.file_path or "-") - logger.info( - "🔧 Resolved Full Config Summary: node_types=%d local=%s remote=%s", - len(config.node_types), - config.local_datasource.type, - config.remote_datasource.type, - ) - logger.info("=" * 80) + _log_pipeline_start(config, title=title) pipeline = await create_pipeline_from_config( config, @@ -393,8 +430,7 @@ async def run_pipeline_from_config( stats = await pipeline.run() if print_summary: - logger.info("✅ Pipeline completed: stats=%s", stats) - logger.info("🗃️ Persistence DB: %s", config.persist.db_path) + _log_pipeline_end(config, stats=stats) return stats finally: diff --git a/sync_state_machine/pipeline/full_sync_pipeline.py b/sync_state_machine/pipeline/full_sync_pipeline.py index 16b5374..640166b 100644 --- a/sync_state_machine/pipeline/full_sync_pipeline.py +++ b/sync_state_machine/pipeline/full_sync_pipeline.py @@ -5,10 +5,9 @@ Full Sync Pipeline - Implements document-specified end-to-end workflow. from __future__ import annotations import asyncio -import logging from dataclasses import dataclass -from typing import Any, Dict, List, Optional, TYPE_CHECKING from pathlib import Path +from typing import TYPE_CHECKING, Any, Dict, List, Optional if TYPE_CHECKING: from ..datasource.datasource import BaseDataSource @@ -18,8 +17,8 @@ from ..common.collection import DataCollection from ..common.binding import BindingManager from ..logging import get_logger from ..common.persistence import PersistenceBackend +from ..sync_system.config import UpdateDirection from ..sync_system.strategy import BaseSyncStrategy -from ..sync_system.config import OrphanAction, UpdateDirection from ..engine import StateMachineConfig, StateMachineRuntime from ..engine import e41_post_create_ready from ..sync_system.strategy_ops import finalize_peer_creating_sources, run_phase2_cleanup @@ -130,22 +129,73 @@ class FullSyncPipeline: cfg_path = Path(__file__).resolve().parents[1] / "config" / "node_state_machine.yaml" return StateMachineRuntime(StateMachineConfig.load(cfg_path)) + def _strategy_map(self) -> Dict[str, BaseSyncStrategy]: + return {strategy.node_type: strategy for strategy in self.strategies} + + def _iter_sync_strategies(self): + strategy_map = self._strategy_map() + for node_type in self.sync_order: + strategy = strategy_map.get(node_type) + if strategy is not None: + yield node_type, strategy + + def _get_strategy(self, node_type: str) -> BaseSyncStrategy | None: + return next((item for item in self.strategies if item.node_type == node_type), None) + + def _log_phase_completed(self, phase_name: str) -> None: + self._logger.info("✅ Phase %s completed (%s)", phase_name, self.pipeline_name) + + def _log_strategy_start(self, node_type: str) -> None: + section_width = 96 + self._logger.info("\n" + "-" * section_width) + self._logger.info("🧩 Strategy start: %s", node_type) + self._logger.info("-" * section_width) + + def _log_strategy_result(self, action: str, node_type: str, count: int | None = None) -> None: + if count is None: + self._logger.info("✅ Strategy %s: %s", action, node_type) + return + self._logger.info("✅ Strategy %s: %s (%s=%d)", action, node_type, action, count) + + def _log_strategy_done(self, node_type: str) -> None: + self._logger.info("✅ Strategy done: %s", node_type) + + def _log_strategy_failure(self, stage: str, node_type: str, exc: Exception) -> None: + self._logger.error( + "❌ Strategy %s failed but pipeline continues: %s | %s: %s", + stage, + node_type, + type(exc).__name__, + exc, + ) + + def _log_reload_skipped(self, node_type: str, reason: str) -> None: + self._logger.info("⏭️ Reload skipped for JSONL pipeline: node_type=%s reason=%s", node_type, reason) + + async def _safe_close(self, name: str, close_coro, timeout: float = 3.0) -> None: + try: + await asyncio.wait_for(close_coro, timeout=timeout) + except asyncio.TimeoutError: + self._logger.warning("⚠️ close timeout: %s (>%ss)", name, timeout) + except Exception as exc: + self._logger.warning("⚠️ close failed: %s (%s)", name, exc) + async def run(self) -> Dict[str, Any]: persistence_started = False try: await self.phase_bootstrap() - self._logger.info("✅ Phase bootstrap completed (%s)", self.pipeline_name) + self._log_phase_completed("bootstrap") await self.phase_sync_by_node_type() - self._logger.info("✅ Phase sync-by-node-type completed (%s)", self.pipeline_name) + self._log_phase_completed("sync-by-node-type") post_ok = await self.phase_post_check() self.stats["post_check_passed"] = post_ok - self._logger.info("✅ Phase post-check completed (%s)", self.pipeline_name) + self._log_phase_completed("post-check") persistence_started = True await self.phase_persistence() - self._logger.info("✅ Phase persistence completed (%s)", self.pipeline_name) + self._log_phase_completed("persistence") return self.stats except Exception as exc: if not persistence_started: @@ -180,29 +230,22 @@ class FullSyncPipeline: async def phase_sync_by_node_type(self) -> None: self._prepare_datasources() - strategy_map = {s.node_type: s for s in self.strategies} - for node_type in self.sync_order: - strategy = strategy_map.get(node_type) - if strategy is None: - continue + for node_type, strategy in self._iter_sync_strategies(): try: - section_width = 96 - self._logger.info("\n" + "-" * section_width) - self._logger.info(f"🧩 Strategy start: {node_type}") - self._logger.info("-" * section_width) + self._log_strategy_start(node_type) await self._load_node_type(node_type, reason="pre-strategy refresh") # Load failure is a hard stop: create/update cannot safely treat missing data as empty data. await strategy.bind() - self._logger.info(f"✅ Strategy bind: {node_type}") + self._log_strategy_result("bind", node_type) if strategy.skip_sync: - self._logger.info(f"⏭️ Skipping create/update for {node_type} (skip_sync=True)") + self._logger.info("⏭️ Skipping create/update for %s (skip_sync=True)", node_type) continue created = await strategy.create() - self._logger.info(f"✅ Strategy create: {node_type} (created={len(created)})") + self._log_strategy_result("create", node_type, len(created)) created_local_node_ids = { node.node_id for node in created if self.local_collection.get_by_node_id(node.node_id) is not None } @@ -228,7 +271,7 @@ class FullSyncPipeline: ) updated = await strategy.update() - self._logger.info(f"✅ Strategy update: {node_type} (updated={len(updated)})") + self._log_strategy_result("update", node_type, len(updated)) update_outcome = await self.commit_updates(node_type) if update_outcome.any_written: await self._reload_node_type_after_write( @@ -238,12 +281,12 @@ class FullSyncPipeline: reload_remote=update_outcome.remote_written, ) - self._logger.info(f"✅ Strategy done: {node_type}") + self._log_strategy_done(node_type) except NodeTypeLoadError: raise except Exception as exc: self.stats["failed"] += 1 - self._logger.error(f"❌ Strategy failed but pipeline continues: {node_type} | {type(exc).__name__}: {exc}") + self._log_strategy_failure("sync", node_type, exc) continue async def phase_bootstrap(self) -> None: @@ -266,31 +309,23 @@ class FullSyncPipeline: async def phase_bind(self) -> None: self._prepare_datasources() - strategy_map = {s.node_type: s for s in self.strategies} - for node_type in self.sync_order: - strategy = strategy_map.get(node_type) - if strategy is None: - continue + for node_type, strategy in self._iter_sync_strategies(): try: await strategy.bind() except Exception as exc: self.stats["failed"] += 1 - self._logger.error(f"❌ Strategy bind failed but pipeline continues: {node_type} | {type(exc).__name__}: {exc}") + self._log_strategy_failure("bind", node_type, exc) continue async def phase_create(self) -> None: self._prepare_datasources() - strategy_map = {s.node_type: s for s in self.strategies} - for node_type in self.sync_order: - strategy = strategy_map.get(node_type) - if strategy is None: - continue + for node_type, strategy in self._iter_sync_strategies(): if strategy.skip_sync: - self._logger.info(f"⏭️ Skipping create for {node_type} (skip_sync=True)") + self._logger.info("⏭️ Skipping create for %s (skip_sync=True)", node_type) continue try: created = await strategy.create() - self._logger.info(f"✅ Strategy create: {node_type} (created={len(created)})") + self._log_strategy_result("create", node_type, len(created)) created_local_node_ids = { node.node_id for node in created if self.local_collection.get_by_node_id(node.node_id) is not None } @@ -316,22 +351,18 @@ class FullSyncPipeline: ) except Exception as exc: self.stats["failed"] += 1 - self._logger.error(f"❌ Strategy create failed but pipeline continues: {node_type} | {type(exc).__name__}: {exc}") + self._log_strategy_failure("create", node_type, exc) continue async def phase_update(self) -> None: self._prepare_datasources() - strategy_map = {s.node_type: s for s in self.strategies} - for node_type in self.sync_order: - strategy = strategy_map.get(node_type) - if strategy is None: - continue + for node_type, strategy in self._iter_sync_strategies(): if strategy.skip_sync: - self._logger.info(f"⏭️ Skipping update for {node_type} (skip_sync=True)") + self._logger.info("⏭️ Skipping update for %s (skip_sync=True)", node_type) continue try: updated = await strategy.update() - self._logger.info(f"✅ Strategy update: {node_type} (updated={len(updated)})") + self._log_strategy_result("update", node_type, len(updated)) update_outcome = await self.commit_updates(node_type) if update_outcome.any_written: await self._reload_node_type_after_write( @@ -343,7 +374,7 @@ class FullSyncPipeline: await self._evaluate_consistency_for_node_type(node_type) except Exception as exc: self.stats["failed"] += 1 - self._logger.error(f"❌ Strategy update failed but pipeline continues: {node_type} | {type(exc).__name__}: {exc}") + self._log_strategy_failure("update", node_type, exc) continue async def close(self) -> None: @@ -352,17 +383,9 @@ class FullSyncPipeline: return # 已经关闭过，避免重复关闭 self._closed = True - async def _safe_close(name: str, close_coro, timeout: float = 3.0) -> None: - try: - await asyncio.wait_for(close_coro, timeout=timeout) - except asyncio.TimeoutError: - self._logger.warning("⚠️ close timeout: %s (>%ss)", name, timeout) - except Exception as exc: - self._logger.warning("⚠️ close failed: %s (%s)", name, exc) - - await _safe_close("remote_datasource", self.remote_datasource.close()) - await _safe_close("local_datasource", self.local_datasource.close()) - await _safe_close("persistence", self.persistence.close()) + await self._safe_close("remote_datasource", self.remote_datasource.close()) + await self._safe_close("local_datasource", self.local_datasource.close()) + await self._safe_close("persistence", self.persistence.close()) async def _load_persistence_state(self) -> None: """加载持久化数据到内存，等待 bootstrap datasource refresh 与 E01 cleanup。""" @@ -544,7 +567,7 @@ class FullSyncPipeline: async def _reload_node_type(self, node_type: str, *, reason: str) -> None: if self._should_skip_reload(): - self._logger.info("⏭️ Reload skipped for JSONL pipeline: node_type=%s reason=%s", node_type, reason) + self._log_reload_skipped(node_type, reason) return await self._load_node_type(node_type, reason=reason) @@ -560,7 +583,7 @@ class FullSyncPipeline: return if self._should_skip_reload(): - self._logger.info("⏭️ Reload skipped for JSONL pipeline: node_type=%s reason=%s", node_type, reason) + self._log_reload_skipped(node_type, reason) return self._logger.info( @@ -718,13 +741,13 @@ class FullSyncPipeline: else: self._logger.info("🔄 Post-check: reloading all node types for final consistency evaluation...") for node_type in self.node_types: - strategy = next((item for item in self.strategies if item.node_type == node_type), None) + strategy = self._get_strategy(node_type) if strategy is not None and strategy.skip_post_check: self._logger.info("⏭️ Post-check reload skipped: node_type=%s", node_type) continue await self._reload_node_type(node_type, reason="post-check final reload") for node_type in self.node_types: - strategy = next((item for item in self.strategies if item.node_type == node_type), None) + strategy = self._get_strategy(node_type) if strategy is not None and strategy.skip_post_check: self._logger.info("⏭️ Post-check skipped: node_type=%s", node_type) self._consistency_by_type[node_type] = {