From 8f4727a77263cb452a5dfc3be1d0d43bf6c3c1e0 Mon Sep 17 00:00:00 2001 From: strepsiades Date: Wed, 1 Apr 2026 11:52:57 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96sync=5Fsystem=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E8=B4=A8=E9=87=8F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/architecture.md | 4 +- docs/runtime_config_guide.md | 39 +- docs/state_machine.md | 2 +- docs/sync_flow.md | 6 +- scripts/auto_test_domains/base.py | 10 +- scripts/run_full_sync_pipeline_jsonl_sm.py | 9 +- scripts/run_full_sync_pipeline_simple_sm.py | 9 +- sync_state_machine/common/collection.py | 2 +- sync_state_machine/config/builder.py | 118 ++-- sync_state_machine/config/strategy_config.py | 76 ++- sync_state_machine/datasource/api/client.py | 38 ++ sync_state_machine/datasource/api/handler.py | 19 +- sync_state_machine/datasource/datasource.py | 11 +- .../domain/construction/sync_strategy.py | 3 +- .../contract_settlement/sync_strategy.py | 3 +- .../domain/material/sync_strategy.py | 3 +- .../domain/preparation/sync_strategy.py | 9 +- .../domain/production/sync_strategy.py | 9 +- .../domain/project/api_handler.py | 2 + .../domain/project/sync_node.py | 2 + .../domain/project/sync_strategy.py | 70 ++- .../domain/project_detail/sync_strategy.py | 85 ++- .../domain/supplier/sync_strategy.py | 11 +- sync_state_machine/logging.py | 110 ++++ sync_state_machine/pipeline/factory.py | 123 ++--- .../pipeline/full_sync_pipeline.py | 6 +- sync_state_machine/pipeline/run_logger.py | 56 +- sync_state_machine/sync_system/strategy.py | 411 ++++++-------- .../sync_system/strategy_ops/__init__.py | 14 +- .../sync_system/strategy_ops/bind_ops.py | 6 +- .../strategy_ops/post_check_ops.py | 10 +- .../sync_system/strategy_ops/reset_ops.py | 7 - .../sync_system/strategy_ops/update_ops.py | 505 ++++++++++-------- .../test_full_pipeline_multinode_mock.py | 16 +- ...st_full_pipeline_multinode_persisted_db.py | 13 +- .../test_strategy_business_points_migrated.py | 37 +- tests/unit/test_api_client_rate_limit.py | 66 +++ ...est_pipeline_strategy_continue_on_error.py | 6 +- ...ect_api_handler_dangerous_delete_config.py | 57 -- .../test_project_api_handler_normalization.py | 12 + .../unit/test_project_detail_sync_strategy.py | 68 +++ tests/unit/test_project_sync_strategy.py | 68 +++ tests/unit/test_run_presets_yaml_loader.py | 4 +- tests/unit/test_strategy_runtime_config.py | 61 ++- tests/unit/test_update_ops.py | 110 ++-- ui/runner.py | 15 +- 46 files changed, 1431 insertions(+), 890 deletions(-) create mode 100644 sync_state_machine/logging.py delete mode 100644 tests/unit/test_project_api_handler_dangerous_delete_config.py create mode 100644 tests/unit/test_project_detail_sync_strategy.py create mode 100644 tests/unit/test_project_sync_strategy.py diff --git a/docs/architecture.md b/docs/architecture.md index 2c75d75..3ab01f0 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -7,7 +7,7 @@ - 负责组装并驱动整条流程：加载持久化、执行 bind/create/update、调用 datasource 同步、持久化结果。 - reset 语义： - 加载阶段仅恢复持久化状态（不做状态重置）； - - 加载后通过 `BaseSyncStrategy.run_reset()` 触发 `E01`： + - 加载后通过 `run_phase2_cleanup()` 触发 `E01`： - `create_zombie=true` -> `S15`（随后删除）； - `create_zombie=false` -> `S00`（归并为 `UNCHECKED/NONE/PENDING`，并清空 `error`）。 - 若持久化核心枚举非法（`action/status/binding_status`），节点进入加载异常隔离态 `S17`： @@ -30,7 +30,7 @@ - `create_ops.py`: `run_create()` - `update_ops.py`: `run_update()` - `delete_ops.py`: `run_delete()` - - `reset_ops.py`: `get_phase1_reset_defaults()/run_phase2_cleanup()` + - `reset_ops.py`: `run_phase2_cleanup()` - strategy 负责收集上下文并调用状态机事件函数，不直接硬编码状态值。 ### datasource（执行与回写层） diff --git a/docs/runtime_config_guide.md b/docs/runtime_config_guide.md index b33c14a..c7177d1 100644 --- a/docs/runtime_config_guide.md +++ b/docs/runtime_config_guide.md @@ -70,10 +70,9 @@ strategies: local_orphan_action: create_remote remote_orphan_action: none update_direction: push - field_direction_key: null - field_direction_overrides: {} post_check_ignore_fields: [] compare_ignore_fields: [] + domain_option: {} persist: backend: sqlite db_path: /abs/path/test_results/sync_state_machine/demo.db @@ -180,7 +179,6 @@ persist: - `push`：本地变更推送到远端。 - `pull`：远端变更拉取到本地。 -- `bidirectional`：双向更新（需要明确冲突策略，谨慎使用）。 - `none`：关闭 update 阶段。 ### 4.2 `local_orphan_action` / `remote_orphan_action` @@ -228,31 +226,34 @@ persist: - `remote_datasource.handler_configs.supplier.load_mode: persisted_only` - `persisted_only` 表示只使用已持久化数据，不主动从外部拉新数据（具体以对应 handler 实现为准）。 -### 4.7 `field_direction_key` / `field_direction_overrides` +### 4.7 `domain_option` -- 这两个字段只在少数 node_type 上需要，典型例子是 `project_detail_data`。 -- `field_direction_key` - - 指定从节点 data 里取哪个字段作为“方向查表键”。 - - 例如 `key`，表示用 `data["key"]` 决定这条记录是 push 还是 pull。 -- `field_direction_overrides` - - 指定“字段值 -> 更新方向”的映射。 - - 没命中的值继续使用默认 `update_direction`。 +- `domain_option` 用于承载 domain 自己识别的运行时扩展参数。 +- 通用策略行为仍然放在 `config` 下；只有领域特例才放到 `domain_option`。 -示例： +示例 1：`project` 先从远端拉回两个特殊投产日期字段，再执行默认 push + +```yaml +project: + domain_option: + pull_group_plan_production_date: true + config: + update_direction: push +``` + +示例 2：`project_detail_data` 用单个开关启用固定业务规则，把集团保供/爬坡/挑战产能计划按 pull 处理 ```yaml project_detail_data: + domain_option: + pull_group_production_plans: true config: - field_direction_key: key - field_direction_overrides: - ensure_production: pull - climb_production: pull - challenge_production: pull + update_direction: push ``` 含义： -- `key=ensure_production/climb_production/challenge_production` 时，从远端拉回本地。 -- 其余 key 继续按默认 `update_direction=push` 处理。 +- `project.domain_option.pull_group_plan_production_date=true` 时，`ic_plan_first_production_date` 和 `ic_plan_full_production_date` 会先 remote -> local，再执行剩余字段的常规更新。 +- `project_detail_data.domain_option.pull_group_production_plans=true` 时，固定的 `ensure_production`、`climb_production`、`challenge_production` 会先 remote -> local；其余 key 继续沿用 `config.update_direction`。 ## 5. 阶段行为与配置关系 diff --git a/docs/state_machine.md b/docs/state_machine.md index 396607e..f0ce0f7 100644 --- a/docs/state_machine.md +++ b/docs/state_machine.md @@ -46,7 +46,7 @@ - bind 主入口：`sync_state_machine/sync_system/strategy.py::DefaultSyncStrategy.bind()` -> `sync_state_machine/sync_system/strategy_ops/bind_ops.py::run_bind()` - create 主入口：`sync_state_machine/sync_system/strategy.py::DefaultSyncStrategy.create()` -> `sync_state_machine/sync_system/strategy_ops/create_ops.py::run_create()` - update 主入口：`sync_state_machine/sync_system/strategy.py::DefaultSyncStrategy.update()` -> `sync_state_machine/sync_system/strategy_ops/update_ops.py::run_update()` -- reset 主入口：`sync_state_machine/sync_system/strategy.py::BaseSyncStrategy.run_reset()` -> `sync_state_machine/sync_system/strategy_ops/reset_ops.py::run_phase2_cleanup()` +- reset 主入口：`sync_state_machine/pipeline/full_sync_pipeline.py::_load_persistence_state()` -> `sync_state_machine/sync_system/strategy_ops/reset_ops.py::run_phase2_cleanup()` ## 5. context 字段（以 YAML 为准） diff --git a/docs/sync_flow.md b/docs/sync_flow.md index 2781a18..267e9bb 100644 --- a/docs/sync_flow.md +++ b/docs/sync_flow.md @@ -11,12 +11,12 @@ ## 2. reset 时序 -### 2.1 加载时 reset（phase1 语义） +### 2.1 加载时恢复 - 在 `sync_state_machine/common/collection.py::load_from_persistence()` 执行。 - 语义：仅恢复持久化状态（不在此阶段做状态归并）。 -### 2.2 加载后 cleanup（phase2 语义） -- pipeline 调用 `BaseSyncStrategy.run_reset()`。 +### 2.2 加载后 cleanup +- pipeline 直接调用 `run_phase2_cleanup()`。 - 实际实现：`sync_state_machine/sync_system/strategy_ops/reset_ops.py::run_phase2_cleanup()`。 - 核心事件：`E01`（`S16 -> S00/S15`）。 - `S16` 表示“持久化加载输入态”：本轮从持久化恢复出的节点集合（可包含上轮遗留节点，也可包含上轮已稳定节点）。 diff --git a/scripts/auto_test_domains/base.py b/scripts/auto_test_domains/base.py index fca3c5e..64eaeec 100644 --- a/scripts/auto_test_domains/base.py +++ b/scripts/auto_test_domains/base.py @@ -309,11 +309,11 @@ def build_temp_run_profile_for_project_auto_bind( project_strategy["update_direction"] = project_update_direction or "none" if project_detail_push_keys: - project_detail_config = profile.setdefault("strategies", {}).setdefault("project_detail_data", {}).setdefault("config", {}) - overrides = dict(project_detail_config.get("field_direction_overrides", {})) - for key in project_detail_push_keys: - overrides.pop(key, None) - project_detail_config["field_direction_overrides"] = overrides + fixed_pull_keys = {"ensure_production", "climb_production", "challenge_production"} + if fixed_pull_keys.intersection(project_detail_push_keys): + project_detail_runtime = profile.setdefault("strategies", {}).setdefault("project_detail_data", {}) + domain_option = project_detail_runtime.setdefault("domain_option", {}) + domain_option["pull_group_production_plans"] = False runtime_dir = Path(report_root) / "runtime_configs" runtime_dir.mkdir(parents=True, exist_ok=True) diff --git a/scripts/run_full_sync_pipeline_jsonl_sm.py b/scripts/run_full_sync_pipeline_jsonl_sm.py index 4224479..1ceae34 100644 --- a/scripts/run_full_sync_pipeline_jsonl_sm.py +++ b/scripts/run_full_sync_pipeline_jsonl_sm.py @@ -15,6 +15,7 @@ if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) import sync_state_machine.domain # noqa: F401, E402 +from sync_state_machine.logging import configure_app_logging, get_logger # noqa: E402 from sync_state_machine.pipeline import ( # noqa: E402 build_config, run_pipeline_from_config, @@ -23,15 +24,17 @@ from sync_state_machine.config import load_named_preset_overrides # noqa: E402 PRESET_NAME = "jsonl_sm" +logger = get_logger(__name__) async def main() -> int: try: overrides, preset_file, used_default = load_named_preset_overrides(PROJECT_ROOT, PRESET_NAME) source_label = "default" if used_default else "custom" - print(f"🧩 Loaded preset ({source_label}): {preset_file}") config = build_config(project_root=PROJECT_ROOT, overrides=overrides) + configure_app_logging(config.logging, replace_handlers=True) + logger.info("🧩 Loaded preset (%s): %s", source_label, preset_file) await run_pipeline_from_config( config, title="sync_state_machine JSONL -> JSONL pipeline", @@ -40,10 +43,10 @@ async def main() -> int: return 0 except KeyboardInterrupt: - print("\n⚠️ Interrupted by user") + logger.warning("⚠️ Interrupted by user") return 130 except Exception as exc: - print(f"\n❌ Pipeline failed: {type(exc).__name__}: {exc}") + logger.exception("❌ Pipeline failed: %s: %s", type(exc).__name__, exc) import traceback traceback.print_exc() return 1 diff --git a/scripts/run_full_sync_pipeline_simple_sm.py b/scripts/run_full_sync_pipeline_simple_sm.py index b48f6f0..efbf7af 100644 --- a/scripts/run_full_sync_pipeline_simple_sm.py +++ b/scripts/run_full_sync_pipeline_simple_sm.py @@ -15,6 +15,7 @@ if str(PROJECT_ROOT) not in sys.path: sys.path.insert(0, str(PROJECT_ROOT)) import sync_state_machine.domain # noqa: F401, E402 +from sync_state_machine.logging import configure_app_logging, get_logger # noqa: E402 from sync_state_machine.pipeline import ( # noqa: E402 build_config, run_pipeline_from_config, @@ -23,15 +24,17 @@ from sync_state_machine.config import load_named_preset_overrides # noqa: E402 PRESET_NAME = "simple_sm" +logger = get_logger(__name__) async def main() -> int: try: overrides, preset_file, used_default = load_named_preset_overrides(PROJECT_ROOT, PRESET_NAME) source_label = "default" if used_default else "custom" - print(f"🧩 Loaded preset ({source_label}): {preset_file}") config = build_config(project_root=PROJECT_ROOT, overrides=overrides) + configure_app_logging(config.logging, replace_handlers=True) + logger.info("🧩 Loaded preset (%s): %s", source_label, preset_file) await run_pipeline_from_config( config, title="sync_state_machine JSONL -> API pipeline", @@ -40,10 +43,10 @@ async def main() -> int: return 0 except KeyboardInterrupt: - print("\n⚠️ Interrupted by user") + logger.warning("⚠️ Interrupted by user") return 130 except Exception as exc: - print(f"\n❌ Pipeline failed: {type(exc).__name__}: {exc}") + logger.exception("❌ Pipeline failed: %s: %s", type(exc).__name__, exc) import traceback traceback.print_exc() return 1 diff --git a/sync_state_machine/common/collection.py b/sync_state_machine/common/collection.py index 0f0f426..671b942 100644 --- a/sync_state_machine/common/collection.py +++ b/sync_state_machine/common/collection.py @@ -362,7 +362,7 @@ class DataCollection: - data/origin_data 保留（由 DataSource 覆盖） **初始化归并（E01）**： - - 由 Pipeline 在加载后调用 run_reset() 触发 E01 + - 由 Pipeline 在加载后调用 `run_phase2_cleanup()` 触发 E01 - CREATE 僵尸节点进入 S15 并删除 - 其余节点进入 S00（UNCHECKED/NONE/PENDING） """ diff --git a/sync_state_machine/config/builder.py b/sync_state_machine/config/builder.py index 7a220f1..77db75c 100644 --- a/sync_state_machine/config/builder.py +++ b/sync_state_machine/config/builder.py @@ -9,7 +9,7 @@ from ..common.registry import DomainRegistry from .datasource_config import DataSourceConfig from .pipeline_config import PipelineRunConfig from .run_presets import load_overrides_from_file -from .strategy_config import ConfigPresets, StrategyRuntimeConfig +from .strategy_config import ConfigPresets, StrategyRuntimeConfig, resolve_domain_option_config def _deep_merge(base: Mapping[str, Any], override: Mapping[str, Any]) -> Dict[str, Any]: @@ -28,19 +28,76 @@ def _deep_merge(base: Mapping[str, Any], override: Mapping[str, Any]) -> Dict[st return merged +def _build_default_strategy_runtime_config(node_type: str) -> StrategyRuntimeConfig | None: + strategy_class = DomainRegistry.get_strategy_class(node_type) + if strategy_class is None: + return None + + config = strategy_class.default_config.model_copy(deep=True) + config.domain_option = resolve_domain_option_config( + getattr(strategy_class, "domain_option_model", None), + config.domain_option, + ) + + return StrategyRuntimeConfig( + node_type=node_type, + config=config, + ) + + def _default_strategy_map(node_types: list[str]) -> Dict[str, StrategyRuntimeConfig]: strategy_map: Dict[str, StrategyRuntimeConfig] = {} for node_type in node_types: - strategy_class = DomainRegistry.get_strategy_class(node_type) - if strategy_class is None: + runtime_cfg = _build_default_strategy_runtime_config(node_type) + if runtime_cfg is None: continue - strategy_map[node_type] = StrategyRuntimeConfig( - node_type=node_type, - config=strategy_class.default_config.model_copy(deep=True), - ) + strategy_map[node_type] = runtime_cfg return strategy_map +def _merge_strategy_runtime_config( + runtime_cfg: StrategyRuntimeConfig, + raw_override: Any, +) -> StrategyRuntimeConfig: + override = dict(raw_override or {}) + override.pop("node_type", None) + + deprecated_keys = [key for key in ("force_sync", "load_mode") if key in override] + if deprecated_keys: + joined = ", ".join(deprecated_keys) + raise ValueError( + f"Strategy override for {runtime_cfg.node_type!r} contains unsupported legacy keys: {joined}. " + "Use skip_sync under strategies. or datasource.handler_configs instead." + ) + + preset = override.pop("preset", None) + if preset: + runtime_cfg.config = runtime_cfg.config.model_validate( + _deep_merge( + runtime_cfg.config.model_dump(), + ConfigPresets.get_preset(str(preset)).model_dump(exclude_defaults=True), + ) + ) + + if "config" in override and isinstance(override["config"], dict): + runtime_cfg.config = runtime_cfg.config.model_validate( + _deep_merge(runtime_cfg.config.model_dump(), override.pop("config")) + ) + + if override: + runtime_cfg.config = runtime_cfg.config.model_validate( + _deep_merge(runtime_cfg.config.model_dump(), override) + ) + + strategy_class = DomainRegistry.get_strategy_class(runtime_cfg.node_type) + runtime_cfg.config.domain_option = resolve_domain_option_config( + getattr(strategy_class, "domain_option_model", None) if strategy_class is not None else None, + runtime_cfg.config.domain_option, + ) + + return runtime_cfg + + def _apply_strategy_overrides( node_types: list[str], strategy_overrides: Any, @@ -49,8 +106,16 @@ def _apply_strategy_overrides( if isinstance(strategy_overrides, list): for item in strategy_overrides: - runtime_cfg = StrategyRuntimeConfig.model_validate(item) - strategy_map[runtime_cfg.node_type] = runtime_cfg + raw_item = dict(item or {}) + node_type = str(raw_item.get("node_type") or "") + if not node_type: + raise ValueError("Strategy override list item missing required node_type") + + runtime_cfg = strategy_map.get(node_type) + if runtime_cfg is None: + runtime_cfg = _build_default_strategy_runtime_config(node_type) or StrategyRuntimeConfig(node_type=node_type) + + strategy_map[node_type] = _merge_strategy_runtime_config(runtime_cfg, raw_item) return [strategy_map[n] for n in node_types if n in strategy_map] if not isinstance(strategy_overrides, dict): @@ -58,41 +123,10 @@ def _apply_strategy_overrides( for node_type, raw_override in strategy_overrides.items(): if node_type not in strategy_map: - strategy_map[node_type] = StrategyRuntimeConfig(node_type=node_type) + strategy_map[node_type] = _build_default_strategy_runtime_config(node_type) or StrategyRuntimeConfig(node_type=node_type) runtime_cfg = strategy_map[node_type] - override = dict(raw_override or {}) - - deprecated_keys = [key for key in ("force_sync", "load_mode") if key in override] - if deprecated_keys: - joined = ", ".join(deprecated_keys) - raise ValueError( - f"Strategy override for {node_type!r} contains unsupported legacy keys: {joined}. " - "Use skip_sync under strategies. or datasource.handler_configs instead." - ) - - preset = override.pop("preset", None) - if preset: - runtime_cfg.config = ConfigPresets.get_preset(str(preset)) - - skip_sync = override.pop("skip_sync", None) - skip_post_check = override.pop("skip_post_check", None) - - if skip_sync is not None: - runtime_cfg.skip_sync = bool(skip_sync) - - if skip_post_check is not None: - runtime_cfg.skip_post_check = bool(skip_post_check) - - if "config" in override and isinstance(override["config"], dict): - runtime_cfg.config = runtime_cfg.config.model_validate( - _deep_merge(runtime_cfg.config.model_dump(), override.pop("config")) - ) - - if override: - runtime_cfg.config = runtime_cfg.config.model_validate( - _deep_merge(runtime_cfg.config.model_dump(), override) - ) + strategy_map[node_type] = _merge_strategy_runtime_config(runtime_cfg, raw_override) return [strategy_map[n] for n in node_types if n in strategy_map] diff --git a/sync_state_machine/config/strategy_config.py b/sync_state_machine/config/strategy_config.py index 35cf5d3..555ba49 100644 --- a/sync_state_machine/config/strategy_config.py +++ b/sync_state_machine/config/strategy_config.py @@ -1,11 +1,21 @@ from __future__ import annotations +import logging from enum import Enum -from typing import Dict, List, Optional +from typing import Any, Dict, List, Mapping, Type from pydantic import BaseModel, ConfigDict, Field +def resolve_domain_option_config( + domain_option_model: Type[BaseModel] | None, + domain_option: Mapping[str, Any] | None, +) -> Dict[str, Any]: + if domain_option_model is None: + return dict(domain_option or {}) + return domain_option_model.model_validate(domain_option or {}).model_dump() + + class OrphanAction(str, Enum): CREATE_REMOTE = "create_remote" CREATE_LOCAL = "create_local" @@ -17,7 +27,6 @@ class OrphanAction(str, Enum): class UpdateDirection(str, Enum): PUSH = "push" PULL = "pull" - BIDIRECTIONAL = "bidirectional" NONE = "none" @@ -40,13 +49,6 @@ class StrategyConfig(BaseModel): remote_orphan_action: OrphanAction = OrphanAction.NONE update_direction: UpdateDirection = UpdateDirection.PUSH - # 字段级同步方向覆盖。 - # field_direction_key: 节点 data 中用作查表键的字段名，如 "key"。 - # field_direction_overrides: 字段值 → UpdateDirection 的映射。 - # 未命中映射的节点沿用 update_direction 作为默认方向。 - field_direction_key: Optional[str] = None - field_direction_overrides: Dict[str, UpdateDirection] = Field(default_factory=dict) - # post-check 一致性比较时忽略的顶层字段。 # 只影响最终一致性评估，不影响 create/update 的差异检测。 post_check_ignore_fields: List[str] = Field(default_factory=list) @@ -63,6 +65,60 @@ class StrategyConfig(BaseModel): # 默认关闭，避免在一般场景下引入隐式匹配。 allow_many_to_one_auto_bind: bool = False + # 是否跳过该类型的 create/update 阶段。 + skip_sync: bool = False + + # 是否跳过该类型的 post-check reload 与一致性校验。 + skip_post_check: bool = False + + # 领域策略运行参数，由各 strategy 的 domain_option_model 负责解释与校验。 + domain_option: Dict[str, Any] = Field(default_factory=dict) + + def log_logic_warnings(self, *, node_type: str, logger: logging.Logger) -> None: + direction = self.update_direction + local_action = self.local_orphan_action + remote_action = self.remote_orphan_action + + if direction == UpdateDirection.PUSH: + if local_action == OrphanAction.CREATE_LOCAL: + logger.warning( + f"[{node_type}] Config inconsistency: " + f"update_direction=PUSH but local_orphan_action=CREATE_LOCAL (本地创建)。" + f"应使用 CREATE_REMOTE (推送到远程)" + ) + if remote_action == OrphanAction.CREATE_LOCAL: + logger.warning( + f"[{node_type}] Config inconsistency: " + f"update_direction=PUSH but remote_orphan_action=CREATE_LOCAL (本地创建)。" + f"PUSH 模式应设为 NONE 或 DELETE_REMOTE" + ) + elif direction == UpdateDirection.PULL: + if local_action == OrphanAction.CREATE_REMOTE: + logger.warning( + f"[{node_type}] Config inconsistency: " + f"update_direction=PULL but local_orphan_action=CREATE_REMOTE (推送到远程)。" + f"PULL 模式应设为 NONE 或 DELETE_LOCAL" + ) + if remote_action in {OrphanAction.CREATE_REMOTE, OrphanAction.DELETE_LOCAL}: + logger.warning( + f"[{node_type}] Config inconsistency: " + f"update_direction=PULL but remote_orphan_action={remote_action}。" + f"PULL 模式建议使用 CREATE_LOCAL 或 NONE" + ) + elif direction == UpdateDirection.NONE: + if local_action != OrphanAction.NONE: + logger.warning( + f"[{node_type}] Config inconsistency: " + f"update_direction=NONE but local_orphan_action={local_action}。" + f"应设为 NONE" + ) + if remote_action != OrphanAction.NONE: + logger.warning( + f"[{node_type}] Config inconsistency: " + f"update_direction=NONE but remote_orphan_action={remote_action}。" + f"应设为 NONE" + ) + class StrategyRuntimeConfig(BaseModel): """Pipeline 层策略运行配置（按 node_type）。""" @@ -70,8 +126,6 @@ class StrategyRuntimeConfig(BaseModel): model_config = ConfigDict(validate_assignment=True, extra="forbid") node_type: str - skip_sync: bool = False - skip_post_check: bool = False config: StrategyConfig = Field(default_factory=StrategyConfig) diff --git a/sync_state_machine/datasource/api/client.py b/sync_state_machine/datasource/api/client.py index 0425c0e..d69afda 100644 --- a/sync_state_machine/datasource/api/client.py +++ b/sync_state_machine/datasource/api/client.py @@ -33,6 +33,7 @@ def _build_default_push_steps() -> List[Dict[str, str]]: class ApiClient: """API HTTP 客户端""" _TRACE_TEXT_MAX_LEN = 1000 + _REQUEST_STATS_LOG_PREFIX = "📊 ApiClient request stats" def __init__( self, @@ -72,6 +73,9 @@ class ApiClient: self._rate_limit_window_seconds = float(rate_limit_window_seconds) self._rate_limit_lock = asyncio.Lock() self._rate_limit_timestamps: deque[float] = deque() + self._request_total = 0 + self._request_success = 0 + self._request_failure = 0 def _is_rate_limit_enabled(self) -> bool: return self._rate_limit_max_requests > 0 and self._rate_limit_window_seconds > 0 @@ -105,9 +109,38 @@ class ApiClient: async def close(self): """关闭 HTTP 客户端""" + self.log_request_stats() if self._client: await self._client.aclose() self._client = None + + def get_request_stats(self) -> Dict[str, int]: + """返回当前客户端生命周期内的请求统计。""" + return { + "total": self._request_total, + "success": self._request_success, + "failure": self._request_failure, + } + + def format_request_stats(self) -> str: + """格式化当前客户端生命周期内的请求统计。""" + stats = self.get_request_stats() + return ( + f"{self._REQUEST_STATS_LOG_PREFIX}: " + f"total={stats['total']}, success={stats['success']}, failure={stats['failure']}" + ) + + def log_request_stats(self, level: str = "INFO") -> None: + """输出当前客户端生命周期内的请求统计。""" + self._log(self.format_request_stats(), level=level) + + def _log_request_summary(self, *, method: str, endpoint: str, elapsed_seconds: float) -> None: + if not self._debug: + return + self._log( + f"ApiClient request: method={method.upper()} endpoint={endpoint} elapsed={elapsed_seconds:.3f}s", + level="DEBUG", + ) def _log(self, message: str, level: str = "DEBUG"): """ @@ -382,6 +415,7 @@ class ApiClient: client = self._get_client() url = f"{self._base_url}{endpoint}" await self._acquire_rate_limit_slot() + self._request_total += 1 # 记录请求信息 self._log(f"\n{'='*60}", level="DEBUG") @@ -433,6 +467,7 @@ class ApiClient: # 记录响应时间 elapsed = time.time() - start_time + self._log_request_summary(method=method, endpoint=endpoint, elapsed_seconds=elapsed) self._log(f"Response Time: {elapsed:.3f}s", level="DEBUG") # 检查 HTTP 状态 @@ -453,14 +488,17 @@ class ApiClient: response=result, elapsed_ms=int(elapsed * 1000), ) + self._request_success += 1 return result except Exception as e: import traceback + self._request_failure += 1 # 记录错误 elapsed = time.time() - start_time + self._log_request_summary(method=method, endpoint=endpoint, elapsed_seconds=elapsed) error_msg = "\n========== [ApiClient HTTP Error] ==========\n" error_msg += f"Request: {method} {url}\n" error_msg += f"Params: {params}\n" diff --git a/sync_state_machine/datasource/api/handler.py b/sync_state_machine/datasource/api/handler.py index cbdd394..0551eb1 100644 --- a/sync_state_machine/datasource/api/handler.py +++ b/sync_state_machine/datasource/api/handler.py @@ -20,6 +20,7 @@ from typing import ClassVar, Dict, List, Any, Optional, TYPE_CHECKING, TypeVar from pydantic import BaseModel, ValidationError from ...common.type_safety import get_pydantic_model_fields +from ...logging import get_logger from ..handler import BaseNodeHandler from ..task_result import TaskResult from ..handler import ValidationResult # Import ValidationResult @@ -30,6 +31,7 @@ if TYPE_CHECKING: from .client import ApiClient T = TypeVar("T", bound=BaseModel) +logger = get_logger(__name__) class BaseApiHandler(BaseNodeHandler[T]): @@ -67,6 +69,16 @@ class BaseApiHandler(BaseNodeHandler[T]): """设置 Collection（由 DataSource 调用）""" self._collection = collection + def ensure_api_client(self) -> 'ApiClient': + if self.api_client is None: + raise RuntimeError("API client not set. Call set_api_client() before invoking API handler methods.") + return self.api_client + + def ensure_collection(self) -> 'DataCollection': + if self._collection is None: + raise RuntimeError("Collection not set. Call set_collection() before invoking handler methods.") + return self._collection + @property def node_type(self) -> str: """返回节点类型""" @@ -109,8 +121,7 @@ class BaseApiHandler(BaseNodeHandler[T]): def _create_basic_node(self, data: Dict[str, Any], depend_ids: Optional[List[str]] = None) -> "SyncNode": """创建或复用基础节点（默认使用 data['id'] 或 data['_id']）""" data_id = str(data.get("id") or data.get("_id") or "") - if self._collection is None: - raise RuntimeError("Collection not set. Call set_collection() before creating nodes.") + self.ensure_collection() if data_id: existing_node = self._collection.get_by_data_id(self.node_type, data_id) @@ -329,10 +340,10 @@ class BaseApiHandler(BaseNodeHandler[T]): if diff_fields: node_info = f"[{node_id}]" if node_id else "" - print(f" [DIFF] {self.node_type}{node_info} schema={schema.__name__}") + logger.info(" [DIFF] %s%s schema=%s", self.node_type, node_info, schema.__name__) for k, v in diff_fields.items(): orig_v = origin_dict.get(k) - print(f" • {k}: {orig_v!r} → {v!r}") + logger.info(" • %s: %r → %r", k, orig_v, v) return diff_fields if diff_fields else None diff --git a/sync_state_machine/datasource/datasource.py b/sync_state_machine/datasource/datasource.py index 6bf9369..3642b67 100644 --- a/sync_state_machine/datasource/datasource.py +++ b/sync_state_machine/datasource/datasource.py @@ -37,9 +37,10 @@ from ..engine import ( StateMachineRuntime, e40_sync_execute, ) +from ..logging import get_logger -logger = logging.getLogger(__name__) +logger = get_logger(__name__) class BaseDataSource(ABC): @@ -460,7 +461,7 @@ class BaseDataSource(ABC): # Handler 加载数据并创建节点 nodes = await handler.load() - print(f"Loaded nodes for {node_type}: {len(nodes)}") + logger.info("Loaded nodes for %s: %d", node_type, len(nodes)) # 将节点写入 Collection：优先按 data_id 命中已有节点并更新 await self._upsert_loaded_nodes(handler, node_type, nodes) @@ -468,7 +469,7 @@ class BaseDataSource(ABC): # 统计加载的节点数 self._stats[node_type]["loaded"] += len(nodes) except Exception as exc: - print(f"❌ [{node_type}] 加载失败: {exc}") + logger.error("❌ [%s] 加载失败: %s", node_type, exc) continue # ========== 同步执行 ========== @@ -639,7 +640,7 @@ class BaseDataSource(ABC): f"in_progress={create_in_progress}, failed={create_failed}, skipped={create_skipped}" ) except Exception as e: - print(f"❌ [{handler.node_type}] create_all 异常: {e}") + logger.error("❌ [%s] create_all 异常: %s", handler.node_type, e) for node in create_nodes: ok = self._apply_sync_execute_state( node, @@ -656,7 +657,7 @@ class BaseDataSource(ABC): for result in update_results: await self._handle_task_result(handler, result, async_tasks) except Exception as e: - print(f"❌ [{handler.node_type}] update_all 异常: {e}") + logger.error("❌ [%s] update_all 异常: %s", handler.node_type, e) for node in update_nodes: ok = self._apply_sync_execute_state( node, diff --git a/sync_state_machine/domain/construction/sync_strategy.py b/sync_state_machine/domain/construction/sync_strategy.py index 21217ec..4194ec6 100644 --- a/sync_state_machine/domain/construction/sync_strategy.py +++ b/sync_state_machine/domain/construction/sync_strategy.py @@ -19,7 +19,8 @@ class ConstructionTaskSyncStrategy(DefaultSyncStrategy[ConstructionResponse]): update_direction=UpdateDirection.PUSH, ) - def preprocess_compare_payload(self, payload: dict) -> dict: + def normalize_compare_payload(self, data: dict | None, data_id_map=None) -> dict: + payload = super().normalize_compare_payload(data, data_id_map) plan_nodes = payload.get("plan_node") if not isinstance(plan_nodes, list): return payload diff --git a/sync_state_machine/domain/contract_settlement/sync_strategy.py b/sync_state_machine/domain/contract_settlement/sync_strategy.py index 1d5da17..3e20fab 100644 --- a/sync_state_machine/domain/contract_settlement/sync_strategy.py +++ b/sync_state_machine/domain/contract_settlement/sync_strategy.py @@ -19,7 +19,8 @@ class ContractSettlementSyncStrategy(DefaultSyncStrategy[ContractSettlementRespo update_direction=UpdateDirection.PUSH, ) - def preprocess_compare_payload(self, payload: dict) -> dict: + def normalize_compare_payload(self, data: dict | None, data_id_map=None) -> dict: + payload = super().normalize_compare_payload(data, data_id_map) plan_nodes = payload.get("plan_node") if not isinstance(plan_nodes, list): return payload diff --git a/sync_state_machine/domain/material/sync_strategy.py b/sync_state_machine/domain/material/sync_strategy.py index e5c7967..bd6c0a0 100644 --- a/sync_state_machine/domain/material/sync_strategy.py +++ b/sync_state_machine/domain/material/sync_strategy.py @@ -19,7 +19,8 @@ class MaterialSyncStrategy(DefaultSyncStrategy[MaterialResponse]): update_direction=UpdateDirection.PUSH, ) - def preprocess_compare_payload(self, payload: dict) -> dict: + def normalize_compare_payload(self, data: dict | None, data_id_map=None) -> dict: + payload = super().normalize_compare_payload(data, data_id_map) plan_nodes = payload.get("plan_node") if not isinstance(plan_nodes, list): return payload diff --git a/sync_state_machine/domain/preparation/sync_strategy.py b/sync_state_machine/domain/preparation/sync_strategy.py index ced16f7..ec29247 100644 --- a/sync_state_machine/domain/preparation/sync_strategy.py +++ b/sync_state_machine/domain/preparation/sync_strategy.py @@ -20,16 +20,13 @@ class PreparationSyncStrategy(DefaultSyncStrategy[PreparationDetail]): update_direction=UpdateDirection.PUSH, ) - def _needs_update(self, source_node, target_node, resolved_data=None, data_id_map=None) -> bool: - if resolved_data is None: - raise ValueError(f"[{self.node_type}] resolved_data is required for differential sync.") - - target_data = target_node.get_data() + def should_update_pair(self, source_node, target_node, *, source_data, target_data, data_id_map=None) -> bool: if target_data is None: return False + # 这里只按 PostPreparation 可写字段判定是否需要 update，避免 response-only 字段差异触发空更新。 update_fields = set(PostPreparation.model_fields.keys()) - source_payload = {key: value for key, value in resolved_data.items() if key in update_fields} + source_payload = {key: value for key, value in source_data.items() if key in update_fields} target_payload = {key: value for key, value in target_data.items() if key in update_fields} normalized_source = normalized_data_for_compare(source_payload, data_id_map or {}) diff --git a/sync_state_machine/domain/production/sync_strategy.py b/sync_state_machine/domain/production/sync_strategy.py index 66eeb11..d515da7 100644 --- a/sync_state_machine/domain/production/sync_strategy.py +++ b/sync_state_machine/domain/production/sync_strategy.py @@ -20,16 +20,13 @@ class ProductionSyncStrategy(DefaultSyncStrategy[ProductionDetail]): update_direction=UpdateDirection.PUSH, ) - def _needs_update(self, source_node, target_node, resolved_data=None, data_id_map=None) -> bool: - if resolved_data is None: - raise ValueError(f"[{self.node_type}] resolved_data is required for differential sync.") - - target_data = target_node.get_data() + def should_update_pair(self, source_node, target_node, *, source_data, target_data, data_id_map=None) -> bool: if target_data is None: return False + # 这里只按 PostProduction 可写字段判定是否需要 update，避免 response-only 字段差异触发空更新。 update_fields = set(PostProduction.model_fields.keys()) - source_payload = {key: value for key, value in resolved_data.items() if key in update_fields} + source_payload = {key: value for key, value in source_data.items() if key in update_fields} target_payload = {key: value for key, value in target_data.items() if key in update_fields} normalized_source = normalized_data_for_compare(source_payload, data_id_map or {}) diff --git a/sync_state_machine/domain/project/api_handler.py b/sync_state_machine/domain/project/api_handler.py index 141bc3e..d82dc08 100644 --- a/sync_state_machine/domain/project/api_handler.py +++ b/sync_state_machine/domain/project/api_handler.py @@ -107,6 +107,8 @@ class ProjectApiHandler(BaseApiHandler): project_id: 可选，指定项目ID列表则只加载这些项目（优先级高于构造函数的filter_project_id） """ try: + self.ensure_api_client() + self.ensure_collection() # 清空旧缓存 self.clear_cache() diff --git a/sync_state_machine/domain/project/sync_node.py b/sync_state_machine/domain/project/sync_node.py index f299680..05f2a9a 100644 --- a/sync_state_machine/domain/project/sync_node.py +++ b/sync_state_machine/domain/project/sync_node.py @@ -21,6 +21,8 @@ class ProjectSyncSchema(ProjectResponseBase): completed_investment: float | None = Field(None, description="已完成投资") total_contract_quantity: float | None = Field(None, description="合同总量") completed_settlement_quantity: float | None = Field(None, description="已结算量") + ic_plan_first_production_date: str | None = Field(None, description="集团计划首批投产日期") + ic_plan_full_production_date: str | None = Field(None, description="集团计划全容量投产日期") class ProjectSyncNode(SyncNode[ProjectSyncSchema]): diff --git a/sync_state_machine/domain/project/sync_strategy.py b/sync_state_machine/domain/project/sync_strategy.py index 731592b..385e2a1 100644 --- a/sync_state_machine/domain/project/sync_strategy.py +++ b/sync_state_machine/domain/project/sync_strategy.py @@ -1,8 +1,14 @@ +from pydantic import BaseModel + from ...sync_system.strategy import DefaultSyncStrategy from ...sync_system.config import StrategyConfig, OrphanAction, UpdateDirection from schemas.project.project_base import ProjectResponseBase +class ProjectDomainOption(BaseModel): + pull_group_plan_production_date: bool = False + + class ProjectSyncStrategy(DefaultSyncStrategy[ProjectResponseBase]): """ Project 同步策略 @@ -17,6 +23,12 @@ class ProjectSyncStrategy(DefaultSyncStrategy[ProjectResponseBase]): # 类变量：schema 固定为 ProjectResponseBase schema = ProjectResponseBase + domain_option_model = ProjectDomainOption + + _PULL_ONLY_FIELDS = [ + "ic_plan_first_production_date", + "ic_plan_full_production_date", + ] # 类变量：默认配置 default_config = StrategyConfig( @@ -26,4 +38,60 @@ class ProjectSyncStrategy(DefaultSyncStrategy[ProjectResponseBase]): local_orphan_action=OrphanAction.CREATE_REMOTE, remote_orphan_action=OrphanAction.NONE, update_direction=UpdateDirection.PUSH, - ) \ No newline at end of file + ) + + @property + def domain_option(self) -> ProjectDomainOption: + return ProjectDomainOption.model_validate(self.config.domain_option) + + def get_node_update_payload(self, node): + payload = dict(node.get_data() or {}) + context = getattr(node, "context", None) + if not isinstance(context, dict): + return payload + + all_info = context.get("all_info") + if not isinstance(all_info, dict): + return payload + + for section_name, section_data in all_info.items(): + if not section_name.endswith("_extension") or not isinstance(section_data, dict): + continue + for field_name in self._PULL_ONLY_FIELDS: + if payload.get(field_name) not in (None, ""): + continue + if field_name in section_data: + payload[field_name] = section_data.get(field_name) + + return payload + + async def update_pair(self, local_node, remote_node, data_id_map=None): + updated_by_id = {} + + if self.domain_option.pull_group_plan_production_date: + pull_update = await self.prepare_update_for_direction( + local_node, + remote_node, + UpdateDirection.PULL, + data_id_map, + include_fields=self._PULL_ONLY_FIELDS, + ) + if pull_update is not None: + updated_by_id[pull_update.node_id] = pull_update + + generic_exclude_fields = self._PULL_ONLY_FIELDS + else: + generic_exclude_fields = None + + if self.config.update_direction != UpdateDirection.NONE: + generic_update = await self.prepare_update_for_direction( + local_node, + remote_node, + self.config.update_direction, + data_id_map, + exclude_fields=generic_exclude_fields, + ) + if generic_update is not None: + updated_by_id[generic_update.node_id] = generic_update + + return list(updated_by_id.values()) \ No newline at end of file diff --git a/sync_state_machine/domain/project_detail/sync_strategy.py b/sync_state_machine/domain/project_detail/sync_strategy.py index 2803cd1..d2b579c 100644 --- a/sync_state_machine/domain/project_detail/sync_strategy.py +++ b/sync_state_machine/domain/project_detail/sync_strategy.py @@ -1,8 +1,16 @@ +from pydantic import BaseModel, Field + from ...sync_system.strategy import DefaultSyncStrategy from ...sync_system.config import StrategyConfig, OrphanAction, UpdateDirection +from ...sync_system.strategy_ops.compare_ops import build_data_id_normalization_map +from ...sync_system.strategy_ops.update_ops import emit_schema_diff_report, _build_schema_diff_validator from schemas.project_extensions.project_detail import ProjectDetailKey, ProjectDetailProject +class ProjectDetailDomainOption(BaseModel): + pull_group_production_plans: bool = Field(True, description="固定将集团保供/爬坡/挑战产能计划按 pull 处理") + + class ProjectDetailSyncStrategy(DefaultSyncStrategy[ProjectDetailProject]): """ ProjectDetail 同步策略（项目容量信息）。 @@ -13,6 +21,12 @@ class ProjectDetailSyncStrategy(DefaultSyncStrategy[ProjectDetailProject]): """ schema = ProjectDetailProject + domain_option_model = ProjectDetailDomainOption + _GROUP_PRODUCTION_PLAN_KEYS = { + "ensure_production", + "climb_production", + "challenge_production", + } default_config = StrategyConfig( auto_bind=True, @@ -21,16 +35,54 @@ class ProjectDetailSyncStrategy(DefaultSyncStrategy[ProjectDetailProject]): local_orphan_action=OrphanAction.NONE, remote_orphan_action=OrphanAction.NONE, update_direction=UpdateDirection.PUSH, - field_direction_key="key", - field_direction_overrides={ - "ensure_production": UpdateDirection.PULL, - "climb_production": UpdateDirection.PULL, - "challenge_production": UpdateDirection.PULL, - }, ) - def preprocess_compare_payload(self, payload: dict) -> dict: - normalized = dict(payload or {}) + async def update(self): + self.ensure_runtime() + validator = _build_schema_diff_validator(self) + self._active_schema_diff_validator = validator + data_id_map = await build_data_id_normalization_map( + node_type=self.node_type, + local_collection=self.local_collection, + remote_collection=self.remote_collection, + binding_manager=self.binding_manager, + ) + + pull_pairs = [] + generic_pairs = [] + for pair in await self.collect_update_pairs(): + if self._should_pull_pair(pair.local_node, pair.remote_node): + pull_pairs.append(pair) + else: + generic_pairs.append(pair) + + updated_by_id = {} + try: + for pair in pull_pairs: + updated_node = await self.prepare_update_for_direction( + pair.local_node, + pair.remote_node, + UpdateDirection.PULL, + data_id_map, + ) + if updated_node is not None: + updated_by_id[updated_node.node_id] = updated_node + + for pair in generic_pairs: + for updated_node in await self.update_pair(pair.local_node, pair.remote_node, data_id_map): + updated_by_id[updated_node.node_id] = updated_node + finally: + self._active_schema_diff_validator = None + + emit_schema_diff_report(self, validator) + return list(updated_by_id.values()) + + @property + def domain_option(self) -> ProjectDetailDomainOption: + return ProjectDetailDomainOption.model_validate(self.config.domain_option) + + def normalize_compare_payload(self, data: dict | None, data_id_map=None) -> dict: + normalized = super().normalize_compare_payload(data, data_id_map) key = normalized.get("key") if isinstance(key, ProjectDetailKey): key = key.value @@ -76,3 +128,20 @@ class ProjectDetailSyncStrategy(DefaultSyncStrategy[ProjectDetailProject]): items.append(normalized_item) return items + + @staticmethod + def _extract_pair_key(local_node, remote_node) -> str | None: + local_data = local_node.get_data() or {} + remote_data = remote_node.get_data() or {} + + key = local_data.get("key", remote_data.get("key")) + if isinstance(key, ProjectDetailKey): + return key.value + if isinstance(key, str): + return key + return None + + def _should_pull_pair(self, local_node, remote_node) -> bool: + if not self.domain_option.pull_group_production_plans: + return False + return self._extract_pair_key(local_node, remote_node) in self._GROUP_PRODUCTION_PLAN_KEYS diff --git a/sync_state_machine/domain/supplier/sync_strategy.py b/sync_state_machine/domain/supplier/sync_strategy.py index 807b979..fb4d5cd 100644 --- a/sync_state_machine/domain/supplier/sync_strategy.py +++ b/sync_state_machine/domain/supplier/sync_strategy.py @@ -24,19 +24,16 @@ class SupplierSyncStrategy(DefaultSyncStrategy[SupplierDetailResponse]): update_direction=UpdateDirection.PULL, ) - def _needs_update( + def should_update_pair( self, source_node, target_node, - resolved_data=None, + *, + source_data, + target_data, data_id_map=None, ) -> bool: """Supplier 更新忽略 id 字段差异。""" - if resolved_data is None: - raise ValueError(f"[{self.node_type}] resolved_data is required for differential sync.") - - source_data = resolved_data - target_data = target_node.get_data() if source_data is None or target_data is None: return False diff --git a/sync_state_machine/logging.py b/sync_state_machine/logging.py new file mode 100644 index 0000000..70a0a05 --- /dev/null +++ b/sync_state_machine/logging.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import logging +from pathlib import Path +from typing import Optional + + +APP_LOGGER_NAME = "sync_state_machine" + + +class _NodeLogFilter(logging.Filter): + def filter(self, record: logging.LogRecord) -> bool: + message = record.getMessage() + if "node=" in message: + return False + if "节点 " in message and "状态" in message: + return False + return True + + +def get_app_logger() -> logging.Logger: + return logging.getLogger(APP_LOGGER_NAME) + + +def get_logger(name: Optional[str] = None) -> logging.Logger: + if not name or name == "__main__": + return get_app_logger() + if name == APP_LOGGER_NAME or name.startswith(f"{APP_LOGGER_NAME}."): + return logging.getLogger(name) + return logging.getLogger(f"{APP_LOGGER_NAME}.{name}") + + +def clear_app_logger_handlers(logger: Optional[logging.Logger] = None) -> None: + target = logger or get_app_logger() + handlers = list(target.handlers) + for handler in handlers: + target.removeHandler(handler) + try: + handler.close() + except Exception: + pass + + +def init_app_logger( + *, + log_file_path: Optional[str] = None, + level: int = logging.INFO, + mirror_console: bool = True, + replace_handlers: bool = True, + suppress_node_logs: bool = True, +) -> logging.Logger: + logger = get_app_logger() + logger.setLevel(level) + logger.propagate = False + + if logger.handlers and not replace_handlers: + return logger + + if replace_handlers: + clear_app_logger_handlers(logger) + + formatter = logging.Formatter("%(message)s") + node_filter = _NodeLogFilter() if suppress_node_logs else None + + if mirror_console: + console_handler = logging.StreamHandler() + console_handler.setLevel(level) + console_handler.setFormatter(formatter) + if node_filter is not None: + console_handler.addFilter(node_filter) + logger.addHandler(console_handler) + + if log_file_path: + path = Path(log_file_path) + path.parent.mkdir(parents=True, exist_ok=True) + file_handler = logging.FileHandler(path, mode="w", encoding="utf-8") + file_handler.setLevel(level) + file_handler.setFormatter(formatter) + if node_filter is not None: + file_handler.addFilter(node_filter) + logger.addHandler(file_handler) + + return logger + + +def configure_app_logging(config, *, replace_handlers: bool = False, mirror_console: bool = True) -> logging.Logger: + if not getattr(config, "initialize", True): + return get_app_logger() + + if not getattr(config, "file_path", None): + from .config.logging_config import resolve_log_file_path + + config.file_path = resolve_log_file_path(config) + + return init_app_logger( + log_file_path=config.file_path, + level=config.level, + mirror_console=mirror_console, + replace_handlers=replace_handlers, + suppress_node_logs=config.suppress_node_logs, + ) + + +def ensure_app_logging(config, *, mirror_console: bool = True) -> logging.Logger: + logger = get_app_logger() + if not getattr(config, "initialize", True): + return logger + if logger.handlers: + return logger + return configure_app_logging(config, replace_handlers=False, mirror_console=mirror_console) \ No newline at end of file diff --git a/sync_state_machine/pipeline/factory.py b/sync_state_machine/pipeline/factory.py index 85b27b8..b3aea29 100644 --- a/sync_state_machine/pipeline/factory.py +++ b/sync_state_machine/pipeline/factory.py @@ -45,12 +45,14 @@ from ..config import ( is_multi_project_payload, load_overrides_from_file, ) +from ..config.strategy_config import resolve_domain_option_config from ..sync_system.strategy import BaseSyncStrategy +from ..logging import configure_app_logging, ensure_app_logging, get_logger from .run_logger import init_pipeline_logger from .full_sync_pipeline import FullSyncPipeline -logger = logging.getLogger(__name__) +logger = get_logger(__name__) def _project_root() -> Path: @@ -63,29 +65,6 @@ def _prepare_remote_jsonl_dir(remote_dir: Path, *, project_root: Optional[Path] return prepare_remote_jsonl_dir(remote_dir, project_root=(project_root or _project_root())) -def _ensure_pipeline_logger( - *, - initialize_logger: bool, - logger_file_path: Optional[str], - logger_level: int, - suppress_node_logs: bool, -) -> None: - if not initialize_logger: - return - - app_logger = logging.getLogger("sync_state_machine") - if app_logger.handlers: - return - - init_pipeline_logger( - log_file_path=logger_file_path, - level=logger_level, - mirror_console=True, - replace_handlers=False, - suppress_node_logs=suppress_node_logs, - ) - - def _build_handler_config( node_type: str, ds_config: DataSourceConfig, @@ -198,16 +177,28 @@ def _build_collections_and_bindings( def _resolve_strategy_map(config: PipelineRunConfig, node_types: List[str]) -> Dict[str, StrategyRuntimeConfig]: if config.strategies: - return {item.node_type: item for item in config.strategies} + resolved_from_config = {item.node_type: item for item in config.strategies} + for node_type, runtime_cfg in resolved_from_config.items(): + strategy_class = DomainRegistry.get_strategy_class(node_type) + runtime_cfg.config.domain_option = resolve_domain_option_config( + getattr(strategy_class, "domain_option_model", None) if strategy_class is not None else None, + runtime_cfg.config.domain_option, + ) + return resolved_from_config resolved: Dict[str, StrategyRuntimeConfig] = {} for node_type in node_types: strategy_class = DomainRegistry.get_strategy_class(node_type) if strategy_class is None: continue + strategy_config = strategy_class.default_config.model_copy(deep=True) + strategy_config.domain_option = resolve_domain_option_config( + getattr(strategy_class, "domain_option_model", None), + strategy_config.domain_option, + ) resolved[node_type] = StrategyRuntimeConfig( node_type=node_type, - config=strategy_class.default_config.model_copy(deep=True), + config=strategy_config, ) return resolved @@ -251,20 +242,18 @@ def _build_strategies( raise RuntimeError(f"Missing strategy for node_type: {node_type}") strategy = strategy_class(node_type, local_collection, remote_collection, binding_manager) - if both_jsonl: - strategy.update_config( - local_orphan_action=OrphanAction.CREATE_REMOTE, - remote_orphan_action=OrphanAction.NONE, - update_direction=UpdateDirection.PUSH, - ) - if node_type == "contract": - strategy.update_config(depend_fields={"project_id": "project"}) - runtime_cfg = strategy_map.get(node_type) if runtime_cfg is not None: - strategy.set_config(runtime_cfg.config.model_copy(deep=True)) - strategy.skip_sync = bool(runtime_cfg.skip_sync) - strategy.skip_post_check = bool(runtime_cfg.skip_post_check) + strategy.config = runtime_cfg.config.model_copy(deep=True) + strategy.config.log_logic_warnings(node_type=strategy.node_type, logger=logger) + + if both_jsonl: + strategy.config.local_orphan_action = OrphanAction.CREATE_REMOTE + strategy.config.remote_orphan_action = OrphanAction.NONE + strategy.config.update_direction = UpdateDirection.PUSH + if node_type == "contract": + strategy.config.depend_fields = {"project_id": "project"} + strategy.config.log_logic_warnings(node_type=strategy.node_type, logger=logger) strategies.append(strategy) return strategies @@ -280,15 +269,7 @@ async def create_pipeline_from_config( local_datasource_override=None, remote_datasource_override=None, ) -> FullSyncPipeline: - if config.logging.initialize and not config.logging.file_path: - config.logging.file_path = resolve_log_file_path(config.logging) - - _ensure_pipeline_logger( - initialize_logger=config.logging.initialize, - logger_file_path=config.logging.file_path, - logger_level=config.logging.level, - suppress_node_logs=config.logging.suppress_node_logs, - ) + ensure_app_logging(config.logging) all_types = list(config.node_types) strategy_map = _resolve_strategy_map(config, all_types) @@ -376,15 +357,7 @@ async def run_pipeline_from_config( local_datasource_override=None, remote_datasource_override=None, ) -> Dict[str, Any]: - if config.logging.initialize and not config.logging.file_path: - config.logging.file_path = resolve_log_file_path(config.logging) - - _ensure_pipeline_logger( - initialize_logger=config.logging.initialize, - logger_file_path=config.logging.file_path, - logger_level=config.logging.level, - suppress_node_logs=config.logging.suppress_node_logs, - ) + ensure_app_logging(config.logging) resolved_cfg = config_snapshot(config) resolved_cfg_text = json.dumps(resolved_cfg, ensure_ascii=False, indent=2) @@ -394,28 +367,21 @@ async def run_pipeline_from_config( config.local_datasource.type, config.remote_datasource.type, ) - if config.logging.file_path: - try: - with open(config.logging.file_path, "a", encoding="utf-8") as fp: - fp.write("Resolved Full Config:\n") - fp.write(resolved_cfg_text) - fp.write("\n") - except Exception as exc: - logger.warning("Failed to write resolved config into log file: %s", exc) + logger.info("Resolved Full Config:\n%s", resolved_cfg_text) pipeline: Optional[FullSyncPipeline] = None try: if print_summary: - print("\n" + "=" * 80) - print(f"🚀 Creating {title}") - print(f"📝 Log file: {config.logging.file_path or '-'}") - print( - f"🔧 Resolved Full Config Summary: node_types={len(config.node_types)} " - f"local={config.local_datasource.type} " - f"remote={config.remote_datasource.type}" + 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, ) - print(f"🔧 Resolved Full Config:\n{resolved_cfg_text}") - print("=" * 80) + logger.info("=" * 80) pipeline = await create_pipeline_from_config( config, @@ -428,8 +394,8 @@ async def run_pipeline_from_config( stats = await pipeline.run() if print_summary: - print(f"\n✅ Pipeline completed: stats={stats}") - print(f"🗃️ Persistence DB: {config.persist.db_path}") + logger.info("✅ Pipeline completed: stats=%s", stats) + logger.info("🗃️ Persistence DB: %s", config.persist.db_path) return stats finally: @@ -497,12 +463,13 @@ async def run_profile_from_file( multi_cfg = MultiProjectRunConfig.model_validate(raw) pipeline = MultiProjectPipeline(project_root=project_root, config_path=cfg_path, config=multi_cfg) if print_summary: - print("\n" + "=" * 80) - print(f"🚀 Creating multi-project pipeline from: {cfg_path}") - print("=" * 80) + logger.info("\n%s", "=" * 80) + logger.info("🚀 Creating multi-project pipeline from: %s", cfg_path) + logger.info("=" * 80) return await pipeline.run() config = build_config(project_root=project_root, overrides=raw) + configure_app_logging(config.logging, replace_handlers=False) mode = f"{config.local_datasource.type}->{config.remote_datasource.type}" return await run_pipeline_from_config( config, diff --git a/sync_state_machine/pipeline/full_sync_pipeline.py b/sync_state_machine/pipeline/full_sync_pipeline.py index a25bffa..99a921b 100644 --- a/sync_state_machine/pipeline/full_sync_pipeline.py +++ b/sync_state_machine/pipeline/full_sync_pipeline.py @@ -20,7 +20,7 @@ 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 +from ..sync_system.strategy_ops import finalize_peer_creating_sources, run_phase2_cleanup from ..sync_system.strategy_ops.post_check_ops import evaluate_consistency_for_node_type from .summary_report import print_pipeline_summary @@ -282,7 +282,7 @@ class FullSyncPipeline: # 加载后 reset 清理：清理 CREATE 失败的僵尸绑定 if not self.strategies: raise RuntimeError("no strategies configured: cannot run bootstrap event E01") - total_cleaned = await BaseSyncStrategy.run_reset( + total_cleaned = await run_phase2_cleanup( node_types=self.node_types, local_collection=self.local_collection, remote_collection=self.remote_collection, @@ -397,11 +397,11 @@ class FullSyncPipeline: result = await evaluate_consistency_for_node_type( node_type=node_type, - strategy=strategy, local_collection=self.local_collection, remote_collection=self.remote_collection, binding_manager=self.binding_manager, logger=self._logger, + normalize_compare_payload=strategy.normalize_compare_payload, depend_fields=depend_fields, compare_to_remote=compare_to_remote, ignore_fields=ignore_fields, diff --git a/sync_state_machine/pipeline/run_logger.py b/sync_state_machine/pipeline/run_logger.py index c867b57..dd9a496 100644 --- a/sync_state_machine/pipeline/run_logger.py +++ b/sync_state_machine/pipeline/run_logger.py @@ -1,18 +1,9 @@ from __future__ import annotations import logging -from pathlib import Path from typing import Optional - -class _NodeLogFilter(logging.Filter): - def filter(self, record: logging.LogRecord) -> bool: - message = record.getMessage() - if "node=" in message: - return False - if "节点 " in message and "状态" in message: - return False - return True +from ..logging import clear_app_logger_handlers, init_app_logger def init_pipeline_logger( @@ -23,43 +14,14 @@ def init_pipeline_logger( replace_handlers: bool = True, suppress_node_logs: bool = True, ) -> logging.Logger: - logger = logging.getLogger("sync_state_machine") - logger.setLevel(level) - logger.propagate = False - - if replace_handlers: - clear_pipeline_logger_handlers(logger) - - formatter = logging.Formatter("%(message)s") - node_filter = _NodeLogFilter() if suppress_node_logs else None - - if mirror_console: - console_handler = logging.StreamHandler() - console_handler.setLevel(level) - console_handler.setFormatter(formatter) - if node_filter is not None: - console_handler.addFilter(node_filter) - logger.addHandler(console_handler) - - if log_file_path: - path = Path(log_file_path) - path.parent.mkdir(parents=True, exist_ok=True) - file_handler = logging.FileHandler(path, mode="w", encoding="utf-8") - file_handler.setLevel(level) - file_handler.setFormatter(formatter) - if node_filter is not None: - file_handler.addFilter(node_filter) - logger.addHandler(file_handler) - - return logger + return init_app_logger( + log_file_path=log_file_path, + level=level, + mirror_console=mirror_console, + replace_handlers=replace_handlers, + suppress_node_logs=suppress_node_logs, + ) def clear_pipeline_logger_handlers(logger: Optional[logging.Logger] = None) -> None: - target = logger or logging.getLogger("sync_state_machine") - handlers = list(target.handlers) - for handler in handlers: - target.removeHandler(handler) - try: - handler.close() - except Exception: - pass + clear_app_logger_handlers(logger) diff --git a/sync_state_machine/sync_system/strategy.py b/sync_state_machine/sync_system/strategy.py index 7140498..4a2714d 100644 --- a/sync_state_machine/sync_system/strategy.py +++ b/sync_state_machine/sync_system/strategy.py @@ -1,4 +1,4 @@ -import copy +from abc import ABC, abstractmethod from typing import TypeVar, Generic, Type, Dict, List, Optional, Any from pydantic import BaseModel import logging @@ -7,31 +7,35 @@ from pathlib import Path from ..common.collection import DataCollection from ..common.binding import BindingManager from ..common.sync_node import SyncNode -from .config import StrategyConfig, OrphanAction, UpdateDirection, ConfigPresets +from .config import StrategyConfig, OrphanAction, UpdateDirection +from ..config.strategy_config import resolve_domain_option_config from .strategy_ops import ( - get_phase1_reset_defaults, run_bind, run_create, run_delete, - run_phase2_cleanup, +) +from .strategy_ops.update_ops import ( + collect_bound_node_pairs, + default_get_node_update_payload, + default_needs_update, + default_update_pair, + prepare_directional_update, run_update, ) from ..engine import ( StateMachineConfig, StateMachineRuntime, ) -from ..validation import SchemaDiffValidator from .strategy_ops.compare_ops import normalized_data_for_compare T = TypeVar("T", bound=BaseModel) logger = logging.getLogger(__name__) -class BaseSyncStrategy(Generic[T]): +class BaseSyncStrategy(ABC, Generic[T]): """ - 同步策略基类。 - 纯逻辑层，不负责 IO 执行。 + 同步策略抽象接口。 - 子类可以通过覆盖 default_config 来设置默认配置： + 子类可以通过覆盖以下类变量提供静态配置： ```python class ProjectStrategy(DefaultSyncStrategy): default_config = StrategyConfig( @@ -47,41 +51,122 @@ class BaseSyncStrategy(Generic[T]): 适用于只需要加载数据供其他类型引用的场景（如 supplier） skip_post_check: bool - 如果为 True，则跳过该类型最终 post-check 的 reload 与一致性校验 """ - # 类变量：默认配置（子类可覆盖） - default_config: StrategyConfig = StrategyConfig() - # 类变量：schema 类型（子类必须设置） - schema: Type[T] = None # type: ignore - # 类变量：是否默认跳过同步（子类可覆盖） - default_skip_sync: bool = False - # 类变量：是否默认跳过 post-check（子类可覆盖） - default_skip_post_check: bool = False - + default_config: StrategyConfig + domain_option_model: Type[BaseModel] | None + schema: Type[T] + node_type: str + config: StrategyConfig + sm_runtime: Optional[StateMachineRuntime] + + @abstractmethod def __init__( - self, - node_type: str, - local_collection: DataCollection, - remote_collection: DataCollection, + self, + node_type: str, + local_collection: DataCollection, + remote_collection: DataCollection, + binding_manager: BindingManager, + ): + pass + + @abstractmethod + async def bind(self, **kwargs): + """进行绑定逻辑判定，设置 node.binding_status 和 node.action""" + pass + + @abstractmethod + async def create(self) -> List[SyncNode]: + """准备 CREATE 操作""" + pass + + @abstractmethod + async def update(self) -> List[SyncNode]: + """准备 UPDATE 操作""" + pass + + @abstractmethod + async def delete(self) -> List[SyncNode]: + pass + + @property + @abstractmethod + def skip_sync(self) -> bool: + pass + + @property + @abstractmethod + def skip_post_check(self) -> bool: + pass + + @abstractmethod + def normalize_compare_payload( + self, + data: Optional[Dict[str, Any]], + data_id_map: Optional[Dict[str, str]] = None, + ) -> Dict[str, Any]: + pass + + +class DefaultSyncStrategy(BaseSyncStrategy[T]): + """ + 基于 STATE_DEFINITIONS.md 实现的标准化同步策略。 + 采用四阶段管线：核心状态 -> 依赖检查 -> 自动匹配 -> 动作映射。 + """ + + default_config: StrategyConfig = StrategyConfig() + domain_option_model: Type[BaseModel] | None = None + schema: Type[T] = None # type: ignore + + def __init__( + self, + node_type: str, + local_collection: DataCollection, + remote_collection: DataCollection, binding_manager: BindingManager, ): if self.schema is None: raise ValueError(f"{self.__class__.__name__} must define schema as a class variable") - + self.node_type = node_type self.local_collection = local_collection self.remote_collection = remote_collection self.binding_manager = binding_manager - # 复制类级别的默认配置到实例 self.config = self.default_config.model_copy(deep=True) - # 是否跳过同步（可运行时修改） - self.skip_sync = self.default_skip_sync - # 是否跳过 post-check（可运行时修改） - self.skip_post_check = self.default_skip_post_check - # 统一由 Pipeline 注入 runtime；未注入时在执行阶段显式报错 self.sm_runtime: Optional[StateMachineRuntime] = None - self._schema_diff_validator: Optional[SchemaDiffValidator] = None - - # 配置逻辑校验 - self._validate_config() + self.config.domain_option = resolve_domain_option_config( + self.domain_option_model, + self.config.domain_option, + ) + + self.config.log_logic_warnings(node_type=self.node_type, logger=logger) + + @classmethod + def get_default_domain_option(cls) -> Dict[str, Any]: + return resolve_domain_option_config(cls.domain_option_model, None) + + def _validate_domain_option(self, domain_option: Optional[Dict[str, Any]]) -> BaseModel | Dict[str, Any]: + if self.domain_option_model is None: + return dict(domain_option or {}) + return self.domain_option_model.model_validate(domain_option or {}) + + @property + def domain_option(self) -> BaseModel | Dict[str, Any]: + return self._validate_domain_option(self.config.domain_option) + + @property + def skip_sync(self) -> bool: + return bool(self.config.skip_sync) + + @skip_sync.setter + def skip_sync(self, value: bool) -> None: + self.config.skip_sync = bool(value) + + @property + def skip_post_check(self) -> bool: + return bool(self.config.skip_post_check) + + @skip_post_check.setter + def skip_post_check(self, value: bool) -> None: + self.config.skip_post_check = bool(value) def ensure_runtime(self) -> StateMachineRuntime: runtime = self.sm_runtime @@ -93,128 +178,6 @@ class BaseSyncStrategy(Generic[T]): self.remote_collection.set_state_machine_runtime(runtime) return runtime - def _validate_config(self) -> None: - """ - 校验配置的逻辑一致性。 - - 规则： - - PUSH 模式：local_orphan_action 应为 CREATE_REMOTE，remote_orphan_action 应为 NONE - - PULL 模式：local_orphan_action 应为 NONE，remote_orphan_action 应为 CREATE_LOCAL - - NONE 模式：两者都应为 NONE - """ - direction = self.config.update_direction - local_action = self.config.local_orphan_action - remote_action = self.config.remote_orphan_action - - # PUSH 模式检查 - if direction == UpdateDirection.PUSH: - if local_action == OrphanAction.CREATE_LOCAL: - logger.warning( - f"[{self.node_type}] Config inconsistency: " - f"update_direction=PUSH but local_orphan_action=CREATE_LOCAL (本地创建)。" - f"应使用 CREATE_REMOTE (推送到远程)" - ) - if remote_action == OrphanAction.CREATE_LOCAL: - logger.warning( - f"[{self.node_type}] Config inconsistency: " - f"update_direction=PUSH but remote_orphan_action=CREATE_LOCAL (本地创建)。" - f"PUSH 模式应设为 NONE 或 DELETE_REMOTE" - ) - - # PULL 模式检查 - elif direction == UpdateDirection.PULL: - if local_action == OrphanAction.CREATE_REMOTE: - logger.warning( - f"[{self.node_type}] Config inconsistency: " - f"update_direction=PULL but local_orphan_action=CREATE_REMOTE (推送到远程)。" - f"PULL 模式应设为 NONE 或 DELETE_LOCAL" - ) - if remote_action in {OrphanAction.CREATE_REMOTE, OrphanAction.DELETE_LOCAL}: - logger.warning( - f"[{self.node_type}] Config inconsistency: " - f"update_direction=PULL but remote_orphan_action={remote_action}。" - f"PULL 模式建议使用 CREATE_LOCAL 或 NONE" - ) - - # NONE 模式检查 - elif direction == UpdateDirection.NONE: - if local_action != OrphanAction.NONE: - logger.warning( - f"[{self.node_type}] Config inconsistency: " - f"update_direction=NONE but local_orphan_action={local_action}。" - f"应设为 NONE" - ) - if remote_action != OrphanAction.NONE: - logger.warning( - f"[{self.node_type}] Config inconsistency: " - f"update_direction=NONE but remote_orphan_action={remote_action}。" - f"应设为 NONE" - ) - - def set_config(self, config: StrategyConfig): - """完全替换配置对象""" - self.config = config - self._schema_diff_validator = None - self._validate_config() - - def update_config(self, **kwargs): - """ - 临时修改部分配置项。 - - Example: - strategy.update_config( - local_orphan_action=OrphanAction.NONE, - update_direction=UpdateDirection.PULL - ) - """ - valid_fields = type(self.config).model_fields - for key, value in kwargs.items(): - if key in valid_fields: - setattr(self.config, key, value) - else: - logger.warning(f"[{self.node_type}] Unknown config key: {key}") - self._schema_diff_validator = None - # 更新配置后也要校验 - self._validate_config() - - def apply_preset(self, preset_name: str): - """ - 应用配置预设。 - - Args: - preset_name: 预设名称，可选：first_sync, daily_sync, repair_sync, pull_only, push_only - - Example: - strategy.apply_preset("first_sync") - """ - preset_config = ConfigPresets.get_preset(preset_name) - self.config = preset_config - self._schema_diff_validator = None - self._validate_config() - logger.info(f"[{self.node_type}] Applied preset: {preset_name}") - - def get_schema_diff_validator(self) -> SchemaDiffValidator: - if self._schema_diff_validator is None: - self._schema_diff_validator = SchemaDiffValidator( - schema_name=self.schema.__name__, - ignore_fields=self.config.compare_ignore_fields, - ignore_list_item_fields=self.config.post_check_ignore_list_item_fields, - ) - return self._schema_diff_validator - - def reset_schema_diff_validator(self) -> None: - self.get_schema_diff_validator().reset() - - def emit_schema_diff_report(self) -> None: - validator = self.get_schema_diff_validator() - if not validator.has_records(): - return - for line in validator.format_summary_lines(): - logger.info(f"[{self.node_type}] validate {line}") - - def preprocess_compare_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]: - return payload - def normalize_compare_payload( self, data: Optional[Dict[str, Any]], @@ -226,103 +189,12 @@ class BaseSyncStrategy(Generic[T]): ignore_fields=set(self.config.compare_ignore_fields), ignore_list_item_fields=self.config.post_check_ignore_list_item_fields, ) - return self.preprocess_compare_payload(copy.deepcopy(normalized)) - - async def bind(self, **kwargs): - """进行绑定逻辑判定，设置 node.binding_status 和 node.action""" - raise NotImplementedError - - async def create(self) -> List[SyncNode]: - """准备 CREATE 操作""" - raise NotImplementedError - - async def update(self) -> List[SyncNode]: - """准备 UPDATE 操作""" - raise NotImplementedError - - def _needs_update( - self, - source_node: SyncNode, - target_node: SyncNode, - resolved_data: Optional[Dict[str, Any]] = None, - data_id_map: Optional[Dict[str, str]] = None, - ) -> bool: - from .strategy_ops.update_ops import default_needs_update - - return default_needs_update(self, source_node, target_node, resolved_data, data_id_map) - - @classmethod - def get_phase1_reset_defaults(cls) -> Dict[str, Any]: - return get_phase1_reset_defaults() - - @classmethod - async def run_reset( - cls, - *, - node_types: List[str], - local_collection: DataCollection, - remote_collection: DataCollection, - binding_manager: BindingManager, - runtime: StateMachineRuntime, - ) -> int: - """ - 执行加载后的重置清理（僵尸 CREATE 节点清理）。 - - 说明： - - `DataCollection.load_from_persistence()` 仅恢复持久化状态； - - 这里触发 E01：create_zombie -> S15（删除），其余 -> S00。 - """ - return await run_phase2_cleanup( - node_types=node_types, - local_collection=local_collection, - remote_collection=remote_collection, - binding_manager=binding_manager, - runtime=runtime, - ) - - @classmethod - async def run_phase2_reset( - cls, - *, - node_types: List[str], - local_collection: DataCollection, - remote_collection: DataCollection, - binding_manager: BindingManager, - runtime: StateMachineRuntime, - ) -> int: - """兼容旧命名，等价于 `run_reset`。""" - return await cls.run_reset( - node_types=node_types, - local_collection=local_collection, - remote_collection=remote_collection, - binding_manager=binding_manager, - runtime=runtime, - ) - - -class DefaultSyncStrategy(BaseSyncStrategy[T]): - """ - 基于 STATE_DEFINITIONS.md 实现的标准化同步策略。 - 采用四阶段管线：核心状态 -> 依赖检查 -> 自动匹配 -> 动作映射。 - """ + return normalized async def bind(self, **kwargs): self.ensure_runtime() await run_bind(self, **kwargs) - - - def _get_id_field_hints(self) -> Dict[str, str]: - """ - 获取 ID 字段的 node_type 提示。 - - 默认实现：从 depend_fields 中提取。 - 子类可以重写此方法提供更精确的映射。 - - Returns: - 字段名到 node_type 的映射，例如 {"project_id": "project"} - """ - return dict(self.config.depend_fields) if self.config.depend_fields else {} async def create(self) -> List[SyncNode]: self.ensure_runtime() return await run_create(self) @@ -331,6 +203,51 @@ class DefaultSyncStrategy(BaseSyncStrategy[T]): self.ensure_runtime() return await run_update(self) + def get_node_update_payload(self, node: SyncNode) -> Dict[str, Any]: + return default_get_node_update_payload(node) + + def should_update_pair( + self, + source_node: SyncNode, + target_node: SyncNode, + *, + source_data: Dict[str, Any], + target_data: Dict[str, Any], + data_id_map: Optional[Dict[str, str]] = None, + ) -> bool: + return default_needs_update(self, source_data, target_data, data_id_map) + + async def collect_update_pairs(self): + return await collect_bound_node_pairs(self) + + async def prepare_update_for_direction( + self, + local_node: SyncNode, + remote_node: SyncNode, + direction: UpdateDirection, + data_id_map: Optional[Dict[str, str]] = None, + *, + include_fields: Optional[List[str]] = None, + exclude_fields: Optional[List[str]] = None, + ) -> Optional[SyncNode]: + return await prepare_directional_update( + self, + local_node=local_node, + remote_node=remote_node, + direction=direction, + data_id_map=data_id_map, + include_fields=include_fields, + exclude_fields=exclude_fields, + ) + + async def update_pair( + self, + local_node: SyncNode, + remote_node: SyncNode, + data_id_map: Optional[Dict[str, str]] = None, + ) -> List[SyncNode]: + return await default_update_pair(self, local_node, remote_node, data_id_map) + async def delete(self) -> List[SyncNode]: self.ensure_runtime() return await run_delete(self) diff --git a/sync_state_machine/sync_system/strategy_ops/__init__.py b/sync_state_machine/sync_system/strategy_ops/__init__.py index beaa181..c873868 100644 --- a/sync_state_machine/sync_system/strategy_ops/__init__.py +++ b/sync_state_machine/sync_system/strategy_ops/__init__.py @@ -1,5 +1,11 @@ from .create_ops import add_create_action, run_create, finalize_peer_creating_sources -from .update_ops import add_update_action, run_update +from .update_ops import ( + BoundNodePair, + add_update_action, + collect_bound_node_pairs, + prepare_directional_update, + run_update, +) from .delete_ops import run_delete from .bind_ops import ( run_bind, @@ -10,13 +16,16 @@ from .bind_ops import ( check_dependency_satisfied, collect_dependency_errors, ) -from .reset_ops import get_phase1_reset_defaults, run_phase2_cleanup +from .reset_ops import run_phase2_cleanup __all__ = [ "add_create_action", "run_create", "finalize_peer_creating_sources", + "BoundNodePair", "add_update_action", + "collect_bound_node_pairs", + "prepare_directional_update", "run_update", "run_delete", "run_bind", @@ -26,6 +35,5 @@ __all__ = [ "phase_auto_binding", "check_dependency_satisfied", "collect_dependency_errors", - "get_phase1_reset_defaults", "run_phase2_cleanup", ] diff --git a/sync_state_machine/sync_system/strategy_ops/bind_ops.py b/sync_state_machine/sync_system/strategy_ops/bind_ops.py index f0d2b47..fc61046 100644 --- a/sync_state_machine/sync_system/strategy_ops/bind_ops.py +++ b/sync_state_machine/sync_system/strategy_ops/bind_ops.py @@ -16,6 +16,10 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +def get_id_field_hints(strategy) -> Dict[str, str]: + return dict(strategy.config.depend_fields) if strategy.config.depend_fields else {} + + def _stable_pick_bind_pair(local_candidates: List["SyncNode"], remote_candidates: List["SyncNode"]): local_node = sorted(local_candidates, key=lambda item: ((item.data_id or ""), item.node_id))[0] remote_node = sorted(remote_candidates, key=lambda item: ((item.data_id or ""), item.node_id))[0] @@ -399,7 +403,7 @@ async def phase_auto_binding(strategy, local_nodes: List["SyncNode"], remote_nod resolved_key_dict, resolve_error = await resolve_id_fields_in_key_dict( key_dict, id_resolver, - strategy._get_id_field_hints(), + get_id_field_hints(strategy), ) if not resolved_key_dict: if node.binding_status == BindingStatus.MISSING: diff --git a/sync_state_machine/sync_system/strategy_ops/post_check_ops.py b/sync_state_machine/sync_system/strategy_ops/post_check_ops.py index ba49987..b8fa2f6 100644 --- a/sync_state_machine/sync_system/strategy_ops/post_check_ops.py +++ b/sync_state_machine/sync_system/strategy_ops/post_check_ops.py @@ -1,6 +1,6 @@ from __future__ import annotations -from typing import Any, Dict, List +from typing import Any, Callable, Dict, List from ...sync_system.resolve_ids import IDResolver from ...validation import SchemaDiffValidator @@ -27,11 +27,11 @@ def _append_post_check_error(node, message: str) -> None: async def evaluate_consistency_for_node_type( *, node_type: str, - strategy, local_collection, remote_collection, binding_manager, logger, + normalize_compare_payload: Callable[[Dict[str, Any] | None, Dict[str, str] | None], Dict[str, Any]], depend_fields: Dict[str, str] | None = None, compare_to_remote: bool = True, ignore_fields: List[str] | None = None, @@ -44,6 +44,7 @@ async def evaluate_consistency_for_node_type( schema_name=node_type, ignore_fields=ignore_fields, ) + mismatches: list[str] = [] data_id_map = await build_data_id_normalization_map( node_type=node_type, local_collection=local_collection, @@ -93,8 +94,8 @@ async def evaluate_consistency_for_node_type( if resolved_remote is not None: remote_data = resolved_remote - local_payload = strategy.normalize_compare_payload(local_data, data_id_map) - remote_payload = strategy.normalize_compare_payload(remote_data, data_id_map) + local_payload = normalize_compare_payload(local_data, data_id_map) + remote_payload = normalize_compare_payload(remote_data, data_id_map) diff_map = validator.compare_payloads( local_payload, remote_payload, @@ -129,6 +130,7 @@ async def evaluate_consistency_for_node_type( "fields": report["fields"], "schema_name": report["schema_name"], "ignore_fields": report["ignore_fields"], + "mismatches": mismatches, } return result diff --git a/sync_state_machine/sync_system/strategy_ops/reset_ops.py b/sync_state_machine/sync_system/strategy_ops/reset_ops.py index 5762f59..a59baf3 100644 --- a/sync_state_machine/sync_system/strategy_ops/reset_ops.py +++ b/sync_state_machine/sync_system/strategy_ops/reset_ops.py @@ -14,13 +14,6 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -def get_phase1_reset_defaults() -> Dict[str, Any]: - return { - "binding_status": BindingStatus.UNCHECKED, - "error": None, - } - - async def _resolve_local_id_for_unbind( *, node_type: str, diff --git a/sync_state_machine/sync_system/strategy_ops/update_ops.py b/sync_state_machine/sync_system/strategy_ops/update_ops.py index e5953a8..ecc8857 100644 --- a/sync_state_machine/sync_system/strategy_ops/update_ops.py +++ b/sync_state_machine/sync_system/strategy_ops/update_ops.py @@ -1,17 +1,19 @@ from __future__ import annotations import copy +from dataclasses import dataclass import logging -from typing import TYPE_CHECKING, Callable, List, Optional +from typing import TYPE_CHECKING, Any, Iterable, List, Optional, Sequence from ...common.types import SyncAction from ...sync_system.config import UpdateDirection from ...sync_system.resolve_ids import IDResolver from ...engine import decision_note, e30_update_prepare +from ...validation import SchemaDiffValidator from .compare_ops import ( build_data_id_normalization_map, - normalized_data_for_compare, collect_payload_diffs, + normalized_data_for_compare, ) if TYPE_CHECKING: @@ -21,6 +23,36 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) +def _build_schema_diff_validator(strategy) -> SchemaDiffValidator: + schema = getattr(strategy, "schema", None) + schema_name = getattr(schema, "__name__", None) or str(getattr(strategy, "node_type", "unknown")) + return SchemaDiffValidator( + schema_name=schema_name, + ignore_fields=strategy.config.compare_ignore_fields, + ignore_list_item_fields=strategy.config.post_check_ignore_list_item_fields, + ) + + +def _get_active_schema_diff_validator(strategy) -> SchemaDiffValidator: + validator = getattr(strategy, "_active_schema_diff_validator", None) + if validator is not None: + return validator + return _build_schema_diff_validator(strategy) + + +def emit_schema_diff_report(strategy, validator: SchemaDiffValidator) -> None: + if not validator.has_records(): + return + for line in validator.format_summary_lines(): + logger.info(f"[{strategy.node_type}] validate {line}") + + +@dataclass(frozen=True) +class BoundNodePair: + local_node: "SyncNode" + remote_node: "SyncNode" + + def _extract_snapshot_steps(snapshot) -> List: if not isinstance(snapshot, dict): return [] @@ -41,16 +73,10 @@ def _safe_repr(value, *, max_len: int = 120) -> str: def default_needs_update( strategy, - source_node, - target_node, - resolved_data=None, + source_data, + target_data, data_id_map=None, ) -> bool: - if resolved_data is None: - raise ValueError(f"[{strategy.node_type}] resolved_data is required for differential sync.") - - source_data = resolved_data - target_data = target_node.get_data() if source_data is None or target_data is None: return False @@ -67,10 +93,10 @@ def default_needs_update( ignore_fields=ignore_fields, ignore_list_item_fields=strategy.config.post_check_ignore_list_item_fields, ) - diff_map = strategy.get_schema_diff_validator().compare_payloads( + diff_map = _get_active_schema_diff_validator(strategy).compare_payloads( normalized_source, normalized_target, - sample_id=str(target_node.data_id or target_node.node_id), + sample_id=str(source_data.get("id") or target_data.get("id") or "unknown"), schema_refs=[strategy.schema.__name__], ) is_different = bool(diff_map) @@ -82,8 +108,8 @@ def default_needs_update( return is_different -def resolve_needs_update_check(strategy) -> Callable: - return strategy._needs_update +def default_get_node_update_payload(node: "SyncNode") -> dict[str, Any]: + return node.get_data() or {} def _collect_diff_descriptions( @@ -125,235 +151,268 @@ def _collect_diff_descriptions( return descriptions +async def collect_bound_node_pairs(strategy) -> List[BoundNodePair]: + local_nodes = strategy.local_collection.filter_by_state_ids( + node_type=strategy.node_type, + state_ids=["S01"], + ) + if not local_nodes: + return [] + + pairs: List[BoundNodePair] = [] + for local_node in local_nodes: + remote_node_id = await strategy.binding_manager.get_remote_id(strategy.node_type, local_node.node_id) + if not remote_node_id: + continue + + remote_node = strategy.remote_collection.get_by_node_id(remote_node_id) + if remote_node is None: + continue + + pairs.append(BoundNodePair(local_node=local_node, remote_node=remote_node)) + + return pairs + + +async def default_update_pair( + strategy, + local_node: "SyncNode", + remote_node: "SyncNode", + data_id_map: Optional[dict[str, str]] = None, +) -> List["SyncNode"]: + direction = strategy.config.update_direction + if direction == UpdateDirection.NONE: + return [] + + updated_node = await prepare_directional_update( + strategy, + local_node=local_node, + remote_node=remote_node, + direction=direction, + data_id_map=data_id_map, + ) + return [updated_node] if updated_node is not None else [] + + +def _sync_pending_approval_snapshot(source_node: "SyncNode", target_node: "SyncNode") -> bool: + source_snapshot = source_node.context.get("approval_snapshot") if isinstance(source_node.context, dict) else None + target_snapshot = target_node.context.get("approval_snapshot") if isinstance(target_node.context, dict) else None + if isinstance(source_snapshot, dict): + steps_changed = _steps_payload_changed(source_snapshot, target_snapshot) + pending_snapshot = copy.deepcopy(source_snapshot) + pending_snapshot["steps_changed"] = steps_changed + target_node.context["pending_approval_snapshot"] = pending_snapshot + return steps_changed + + target_node.context.pop("pending_approval_snapshot", None) + return False + + +def _select_fields( + payload: dict[str, Any], + *, + include_fields: Optional[Sequence[str]] = None, + exclude_fields: Optional[Sequence[str]] = None, +) -> dict[str, Any]: + if include_fields: + include_set = set(include_fields) + return {field_name: value for field_name, value in payload.items() if field_name in include_set} + if exclude_fields: + exclude_set = set(exclude_fields) + return {field_name: value for field_name, value in payload.items() if field_name not in exclude_set} + return dict(payload) + + +def build_merged_update_payload( + *, + source_data: dict[str, Any], + target_data: dict[str, Any], + include_fields: Optional[Sequence[str]] = None, + exclude_fields: Optional[Sequence[str]] = None, +) -> dict[str, Any]: + if include_fields: + merged_payload = dict(target_data) + for field_name, value in _select_fields(source_data, include_fields=include_fields).items(): + merged_payload[field_name] = value + return merged_payload + + if exclude_fields: + exclude_set = set(exclude_fields) + merged_payload = dict(source_data) + for field_name, value in target_data.items(): + if field_name in exclude_set: + merged_payload[field_name] = value + return merged_payload + + return dict(source_data) + + +async def _resolve_update_source_payload( + strategy, + *, + source_node: "SyncNode", + source_is_local: bool, +) -> tuple[Optional[dict[str, Any]], Optional[str]]: + source_payload = strategy.get_node_update_payload(source_node) + if not source_payload: + return None, "Node data is missing or invalid" + + id_resolver = IDResolver(strategy.local_collection, strategy.remote_collection, strategy.binding_manager) + return await id_resolver.resolve_and_validate( + data=source_payload, + node_type=strategy.node_type, + depend_fields=dict(strategy.config.depend_fields) if strategy.config.depend_fields else {}, + to_remote=source_is_local, + ) + + +async def prepare_directional_update( + strategy, + *, + local_node: "SyncNode", + remote_node: "SyncNode", + direction: UpdateDirection, + data_id_map: Optional[dict] = None, + include_fields: Optional[Sequence[str]] = None, + exclude_fields: Optional[Sequence[str]] = None, +) -> Optional["SyncNode"]: + if direction == UpdateDirection.NONE: + return None + + source_is_local = direction == UpdateDirection.PUSH + source_node = local_node if source_is_local else remote_node + target_node = remote_node if source_is_local else local_node + + if target_node.action == SyncAction.CREATE: + return None + + target_has_data_id = bool(target_node.data_id) + resolved_source_data = None + resolve_error = None + if target_has_data_id: + resolved_source_data, resolve_error = await _resolve_update_source_payload( + strategy, + source_node=source_node, + source_is_local=source_is_local, + ) + + update_id_resolve_ok = resolved_source_data is not None + target_payload = strategy.get_node_update_payload(target_node) + prepared_payload: dict[str, Any] | None = None + has_diff = False + diff_descriptions: List[str] = [] + + steps_changed = _sync_pending_approval_snapshot(source_node, target_node) + + if update_id_resolve_ok and resolved_source_data is not None: + prepared_payload = build_merged_update_payload( + source_data=resolved_source_data, + target_data=target_payload, + include_fields=include_fields, + exclude_fields=exclude_fields, + ) + has_diff = bool( + strategy.should_update_pair( + source_node, + target_node, + source_data=prepared_payload, + target_data=target_payload, + data_id_map=data_id_map, + ) + ) + if has_diff: + diff_descriptions = _collect_diff_descriptions( + prepared_payload, + target_payload, + data_id_map=data_id_map or {}, + ignore_fields=set(strategy.config.compare_ignore_fields), + ignore_list_item_fields=dict(strategy.config.post_check_ignore_list_item_fields), + ) + + if steps_changed: + has_diff = True + diff_descriptions.append("steps: approval flow changed") + + if target_has_data_id and update_id_resolve_ok and not has_diff: + return None + + decision = e30_update_prepare( + strategy.ensure_runtime(), + source_node=source_node, + target_node=target_node, + update_enabled=True, + target_has_data_id=target_has_data_id, + update_id_resolve_ok=update_id_resolve_ok, + has_diff=has_diff, + id_resolve_detail=resolve_error, + ) + reason = decision_note(decision) + if decision.to_state == "S08": + target_node.error = reason + logger.debug(f"[{strategy.node_type}] 更新准备阻断: target={target_node.node_id}, reason={reason}") + return None + if decision.to_state != "S07": + raise RuntimeError(f"[{strategy.node_type}] unexpected update_prepare state: {decision.to_state}") + if prepared_payload is None: + raise RuntimeError(f"[{strategy.node_type}] E30(S07) requires resolved update data") + + target_node.set_data(prepared_payload) + target_node.error = None + diff_text = "; ".join(diff_descriptions) if diff_descriptions else "(未生成差异明细)" + target_node.append_log(f"检测到差异: 从节点 {source_node.node_id} 更新数据 | fields={diff_text}") + logger.info(f"[{strategy.node_type}] 更新差异字段: target={target_node.node_id}, {diff_text}") + logger.debug(f"[{strategy.node_type}] 更新准备完成: target={target_node.node_id}, reason={reason}") + return target_node + + async def add_update_action( strategy, *, from_collection: "DataCollection", to_collection: "DataCollection", - needs_update_check, source_is_local: bool, data_id_map: Optional[dict] = None, - node_filter: Optional[Callable[["SyncNode"], bool]] = None, + include_fields: Optional[Iterable[str]] = None, + exclude_fields: Optional[Iterable[str]] = None, ) -> List["SyncNode"]: - nodes_to_update = from_collection.filter_by_state_ids( - node_type=strategy.node_type, - state_ids=["S01"], - ) - if not nodes_to_update: - return [] + del from_collection + del to_collection - if node_filter is not None: - nodes_to_update = [n for n in nodes_to_update if node_filter(n)] - if not nodes_to_update: - return [] - - id_resolver = IDResolver(strategy.local_collection, strategy.remote_collection, strategy.binding_manager) updated_nodes: List["SyncNode"] = [] - total_nodes = len(nodes_to_update) - - for idx, src_node in enumerate(nodes_to_update, start=1): - if idx % 100 == 0 or idx == total_nodes: - logger.debug(f"[{strategy.node_type}] 更新检查 {idx}/{total_nodes}") - - src_data = src_node.get_data() - if not src_data: - continue - - if source_is_local: - target_node_id = await strategy.binding_manager.get_remote_id(strategy.node_type, src_node.node_id) - else: - target_node_id = await strategy.binding_manager.get_local_id(strategy.node_type, src_node.node_id) - if not target_node_id: - continue - - target_node = to_collection.get_by_node_id(target_node_id) - if not target_node or target_node.action == SyncAction.CREATE: - continue - - target_has_data_id = bool(target_node.data_id) - data = None - resolve_error = None - if target_has_data_id: - data, resolve_error = await id_resolver.resolve_and_validate( - data=src_data, - node_type=strategy.node_type, - depend_fields=dict(strategy.config.depend_fields) if strategy.config.depend_fields else {}, - to_remote=source_is_local, - ) - - update_id_resolve_ok = data is not None - has_diff = False - diff_descriptions: List[str] = [] - source_snapshot = src_node.context.get("approval_snapshot") if isinstance(src_node.context, dict) else None - target_snapshot = target_node.context.get("approval_snapshot") if isinstance(target_node.context, dict) else None - steps_changed = False - if isinstance(source_snapshot, dict): - steps_changed = _steps_payload_changed(source_snapshot, target_snapshot) - pending_snapshot = copy.deepcopy(source_snapshot) - pending_snapshot["steps_changed"] = steps_changed - target_node.context["pending_approval_snapshot"] = pending_snapshot - else: - target_node.context.pop("pending_approval_snapshot", None) - - if update_id_resolve_ok: - try: - should_update = needs_update_check(src_node, target_node, resolved_data=data, data_id_map=data_id_map) - except TypeError: - try: - should_update = needs_update_check(src_node, target_node) - except TypeError: - should_update = needs_update_check(src_node) - has_diff = bool(should_update) - if has_diff: - diff_descriptions = _collect_diff_descriptions( - data, - target_node.get_data(), - data_id_map=data_id_map or {}, - ignore_fields=set(strategy.config.compare_ignore_fields), - ignore_list_item_fields=dict(strategy.config.post_check_ignore_list_item_fields), - ) - - if steps_changed: - has_diff = True - diff_descriptions.append("steps: approval flow changed") - - if target_has_data_id and update_id_resolve_ok and not has_diff: - continue - - decision = e30_update_prepare( - strategy.ensure_runtime(), - source_node=src_node, - target_node=target_node, - update_enabled=True, - target_has_data_id=target_has_data_id, - update_id_resolve_ok=update_id_resolve_ok, - has_diff=has_diff, - id_resolve_detail=resolve_error, + direction = UpdateDirection.PUSH if source_is_local else UpdateDirection.PULL + for pair in await collect_bound_node_pairs(strategy): + updated_node = await prepare_directional_update( + strategy, + local_node=pair.local_node, + remote_node=pair.remote_node, + direction=direction, + data_id_map=data_id_map, + include_fields=list(include_fields) if include_fields is not None else None, + exclude_fields=list(exclude_fields) if exclude_fields is not None else None, ) - reason = decision_note(decision) - if decision.to_state == "S08": - target_node.error = reason - logger.debug(f"[{strategy.node_type}] 更新准备阻断: target={target_node.node_id}, reason={reason}") - continue - if decision.to_state != "S07": - raise RuntimeError(f"[{strategy.node_type}] unexpected update_prepare state: {decision.to_state}") - if data is None: - raise RuntimeError(f"[{strategy.node_type}] E30(S07) requires resolved update data") - - target_node.set_data(data) - target_node.error = None - diff_text = "; ".join(diff_descriptions) if diff_descriptions else "(未生成差异明细)" - target_node.append_log(f"检测到差异: 从节点 {src_node.node_id} 更新数据 | fields={diff_text}") - logger.info(f"[{strategy.node_type}] 更新差异字段: target={target_node.node_id}, {diff_text}") - logger.debug(f"[{strategy.node_type}] 更新准备完成: target={target_node.node_id}, reason={reason}") - updated_nodes.append(target_node) + if updated_node is not None: + updated_nodes.append(updated_node) return updated_nodes async def run_update(strategy) -> List["SyncNode"]: - updated_nodes: List["SyncNode"] = [] - strategy.reset_schema_diff_validator() - needs_update_check = resolve_needs_update_check(strategy) + updated_by_id: dict[str, "SyncNode"] = {} + validator = _build_schema_diff_validator(strategy) + strategy._active_schema_diff_validator = validator data_id_map = await build_data_id_normalization_map( node_type=strategy.node_type, local_collection=strategy.local_collection, remote_collection=strategy.remote_collection, binding_manager=strategy.binding_manager, ) + try: + for pair in await collect_bound_node_pairs(strategy): + nodes = await strategy.update_pair(pair.local_node, pair.remote_node, data_id_map) + for node in nodes: + updated_by_id[node.node_id] = node + finally: + strategy._active_schema_diff_validator = None - field_key = strategy.config.field_direction_key - overrides: dict = strategy.config.field_direction_overrides - - if field_key and overrides: - # 按方向归组：对每种出现的方向各调一次 add_update_action，传入过滤器。 - # 覆盖表中未命中的节点沿用 update_direction 作为默认方向。 - default_direction = strategy.config.update_direction - active_directions = set(overrides.values()) | {default_direction} - - for direction in (UpdateDirection.PUSH, UpdateDirection.PULL): - if direction not in active_directions: - continue - - captured_direction = direction # 避免闭包捕获循环变量 - captured_overrides = overrides - captured_default = default_direction - captured_field_key = field_key - - def make_filter(d=captured_direction, fk=captured_field_key, ov=captured_overrides, dflt=captured_default): - def _filter(node: "SyncNode") -> bool: - node_data = node.get_data() or {} - key_val = node_data.get(fk) - effective = ov.get(key_val, dflt) - return effective == d - return _filter - - node_filter = make_filter() - - if direction == UpdateDirection.PUSH: - nodes = await add_update_action( - strategy, - from_collection=strategy.local_collection, - to_collection=strategy.remote_collection, - needs_update_check=needs_update_check, - source_is_local=True, - data_id_map=data_id_map, - node_filter=node_filter, - ) - updated_nodes.extend(nodes) - logger.info( - f"[{strategy.node_type}] Updated {len(nodes)} nodes in remote collection" - f" (Push, field_direction_key={field_key})" - ) - - elif direction == UpdateDirection.PULL: - nodes = await add_update_action( - strategy, - from_collection=strategy.remote_collection, - to_collection=strategy.local_collection, - needs_update_check=needs_update_check, - source_is_local=False, - data_id_map=data_id_map, - node_filter=node_filter, - ) - updated_nodes.extend(nodes) - logger.info( - f"[{strategy.node_type}] Updated {len(nodes)} nodes in local collection" - f" (Pull, field_direction_key={field_key})" - ) - - strategy.emit_schema_diff_report() - return updated_nodes - - # ── 无字段级覆盖：原有逻辑不变 ── - if strategy.config.update_direction == UpdateDirection.PUSH: - nodes = await add_update_action( - strategy, - from_collection=strategy.local_collection, - to_collection=strategy.remote_collection, - needs_update_check=needs_update_check, - source_is_local=True, - data_id_map=data_id_map, - ) - updated_nodes.extend(nodes) - logger.info(f"[{strategy.node_type}] Updated {len(nodes)} nodes in remote collection (Push)") - - elif strategy.config.update_direction == UpdateDirection.PULL: - nodes = await add_update_action( - strategy, - from_collection=strategy.remote_collection, - to_collection=strategy.local_collection, - needs_update_check=needs_update_check, - source_is_local=False, - data_id_map=data_id_map, - ) - updated_nodes.extend(nodes) - logger.info(f"[{strategy.node_type}] Updated {len(nodes)} nodes in local collection (Pull)") - - elif strategy.config.update_direction == UpdateDirection.BIDIRECTIONAL: - raise NotImplementedError( - f"[{strategy.node_type}] update_direction=BIDIRECTIONAL requires custom implementation" - ) - - strategy.emit_schema_diff_report() - return updated_nodes + emit_schema_diff_report(strategy, validator) + return list(updated_by_id.values()) diff --git a/tests/integration/test_full_pipeline_multinode_mock.py b/tests/integration/test_full_pipeline_multinode_mock.py index 0b3dd0e..1dca81c 100644 --- a/tests/integration/test_full_pipeline_multinode_mock.py +++ b/tests/integration/test_full_pipeline_multinode_mock.py @@ -1,6 +1,7 @@ from __future__ import annotations import tempfile +import logging from pathlib import Path import copy import json @@ -35,6 +36,12 @@ from tests.fixtures.mock_domain import ( ) +def _configure_strategy(strategy, **kwargs): + for key, value in kwargs.items(): + setattr(strategy.config, key, value) + strategy.config.log_logic_warnings(node_type=strategy.node_type, logger=logging.getLogger(__name__)) + + @pytest_asyncio.fixture async def pipeline_bundle(): register_test_domain() @@ -166,7 +173,8 @@ async def pipeline_bundle(): await binding_manager.persist() project_strategy = TestProjectStrategy("test_project", local_collection, remote_collection, binding_manager) - project_strategy.update_config( + _configure_strategy( + project_strategy, auto_bind=True, auto_bind_fields=["code"], depend_fields={}, @@ -176,7 +184,8 @@ async def pipeline_bundle(): ) contract_strategy = TestContractStrategy("test_contract", local_collection, remote_collection, binding_manager) - contract_strategy.update_config( + _configure_strategy( + contract_strategy, auto_bind=False, depend_fields={"project_ref": "test_project"}, local_orphan_action=OrphanAction.CREATE_REMOTE, @@ -468,7 +477,8 @@ async def test_pipeline_create_then_update_completes_in_single_run_with_reload() remote_ds.add_handler(remote_handler) project_strategy = TestProjectStrategy("test_project", local_collection, remote_collection, binding_manager) - project_strategy.update_config( + _configure_strategy( + project_strategy, auto_bind=True, auto_bind_fields=["code"], depend_fields={}, diff --git a/tests/integration/test_full_pipeline_multinode_persisted_db.py b/tests/integration/test_full_pipeline_multinode_persisted_db.py index 99545f8..e755c0e 100644 --- a/tests/integration/test_full_pipeline_multinode_persisted_db.py +++ b/tests/integration/test_full_pipeline_multinode_persisted_db.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging import sqlite3 from pathlib import Path @@ -26,6 +27,12 @@ from tests.fixtures.mock_domain import ( ) +def _configure_strategy(strategy, **kwargs): + for key, value in kwargs.items(): + setattr(strategy.config, key, value) + strategy.config.log_logic_warnings(node_type=strategy.node_type, logger=logging.getLogger(__name__)) + + @pytest.mark.asyncio async def test_full_pipeline_multinode_mock_persisted_db_snapshot() -> None: register_test_domain() @@ -127,7 +134,8 @@ async def test_full_pipeline_multinode_mock_persisted_db_snapshot() -> None: await binding_manager.persist() project_strategy = TestProjectStrategy("test_project", local_collection, remote_collection, binding_manager) - project_strategy.update_config( + _configure_strategy( + project_strategy, auto_bind=True, auto_bind_fields=["code"], depend_fields={}, @@ -137,7 +145,8 @@ async def test_full_pipeline_multinode_mock_persisted_db_snapshot() -> None: ) contract_strategy = TestContractStrategy("test_contract", local_collection, remote_collection, binding_manager) - contract_strategy.update_config( + _configure_strategy( + contract_strategy, auto_bind=False, depend_fields={"project_ref": "test_project"}, local_orphan_action=OrphanAction.CREATE_REMOTE, diff --git a/tests/integration/test_strategy_business_points_migrated.py b/tests/integration/test_strategy_business_points_migrated.py index a53abda..96f163c 100644 --- a/tests/integration/test_strategy_business_points_migrated.py +++ b/tests/integration/test_strategy_business_points_migrated.py @@ -1,5 +1,6 @@ from __future__ import annotations +import logging import tempfile from pathlib import Path @@ -29,6 +30,12 @@ from tests.fixtures.mock_domain import ( ) +def _configure_strategy(strategy, **kwargs): + for key, value in kwargs.items(): + setattr(strategy.config, key, value) + strategy.config.log_logic_warnings(node_type=strategy.node_type, logger=logging.getLogger(__name__)) + + @pytest_asyncio.fixture async def setup_env(): register_test_domain() @@ -109,7 +116,7 @@ async def test_dependency_resolution_migrated(setup_env): local, remote, bm, _ = setup_env project_strategy = TestProjectStrategy("test_project", local, remote, bm) - project_strategy.update_config(depend_fields={}) + _configure_strategy(project_strategy, depend_fields={}) await local.add(build_project_node("proj_l", "P-1", name="Project A", code="P-A")) await remote.add(build_project_node("proj_r", "P-1-R", name="Project A", code="P-A")) @@ -123,7 +130,8 @@ async def test_dependency_resolution_migrated(setup_env): assert local.get("proj_l").binding_status == BindingStatus.NORMAL contract_strategy = TestContractStrategy("test_contract", local, remote, bm) - contract_strategy.update_config( + _configure_strategy( + contract_strategy, depend_fields={"project_ref": "test_project"}, local_orphan_action=OrphanAction.CREATE_REMOTE, remote_orphan_action=OrphanAction.NONE, @@ -164,7 +172,7 @@ async def test_auto_binding_migrated(setup_env): local, remote, bm, _ = setup_env strategy = TestProjectStrategy("test_project", local, remote, bm) - strategy.update_config(auto_bind=True, auto_bind_fields=["code"], depend_fields={}) + _configure_strategy(strategy, auto_bind=True, auto_bind_fields=["code"], depend_fields={}) await local.add(build_project_node("l1", "P1", name="A", code="AUTO-1")) await remote.add(build_project_node("r1", "RP1", name="A", code="AUTO-1")) @@ -180,7 +188,7 @@ async def test_auto_binding_skip_on_ambiguous_migrated(setup_env): local, remote, bm, _ = setup_env strategy = TestProjectStrategy("test_project", local, remote, bm) - strategy.update_config(auto_bind=True, auto_bind_fields=["code"], depend_fields={}) + _configure_strategy(strategy, auto_bind=True, auto_bind_fields=["code"], depend_fields={}) await local.add(build_project_node("l1", "P1", name="A", code="AMB-1")) await local.add(build_project_node("l2", "P2", name="A", code="AMB-1")) @@ -198,7 +206,8 @@ async def test_auto_binding_many_to_one_can_bind_stable_pick_and_leave_rest_for_ local, remote, bm, _ = setup_env strategy = TestProjectStrategy("test_project", local, remote, bm) - strategy.update_config( + _configure_strategy( + strategy, auto_bind=True, auto_bind_fields=["code"], depend_fields={}, @@ -222,7 +231,8 @@ async def test_pre_bind_data_id_works_when_auto_bind_disabled(setup_env): local, remote, bm, _ = setup_env strategy = TestProjectStrategy("test_project", local, remote, bm) - strategy.update_config( + _configure_strategy( + strategy, auto_bind=False, auto_bind_fields=[], pre_bind_data_id=[{"local_data_id": "PRE-DID-1", "remote_data_id": "PRE-DID-R-1"}], @@ -245,7 +255,7 @@ async def test_dependency_check_data_missing_goes_abnormal(setup_env): local, remote, bm, _ = setup_env strategy = TestContractStrategy("test_contract", local, remote, bm) - strategy.update_config(depend_fields={"project_ref": "test_project"}) + _configure_strategy(strategy, depend_fields={"project_ref": "test_project"}) await local.add( build_contract_node( @@ -270,7 +280,7 @@ async def test_auto_binding_key_missing_migrated(setup_env): local, remote, bm, _ = setup_env strategy = TestProjectStrategy("test_project", local, remote, bm) - strategy.update_config(auto_bind=True, auto_bind_fields=["code", "never_exists"], depend_fields={}) + _configure_strategy(strategy, auto_bind=True, auto_bind_fields=["code", "never_exists"], depend_fields={}) await local.add(build_project_node("km_l", "KM-L", name="KM", code="KM-1")) @@ -286,7 +296,8 @@ async def test_auto_binding_key_id_resolve_fail_migrated(setup_env): local, remote, bm, _ = setup_env strategy = TestProjectStrategy("test_project", local, remote, bm) - strategy.update_config( + _configure_strategy( + strategy, auto_bind=True, auto_bind_fields=["id", "code"], depend_fields={"id": "test_project"}, @@ -308,7 +319,8 @@ async def test_create_prepare_e21_on_id_resolve_fail(setup_env, monkeypatch: pyt local, remote, bm, _ = setup_env strategy = TestProjectStrategy("test_project", local, remote, bm) - strategy.update_config( + _configure_strategy( + strategy, auto_bind=True, auto_bind_fields=["code"], depend_fields={"dummy_id": "test_project"}, @@ -347,7 +359,7 @@ async def test_update_prepare_target_missing_data_id_goes_precondition_failed(se local, remote, bm, _ = setup_env strategy = TestProjectStrategy("test_project", local, remote, bm) - strategy.update_config(update_direction=UpdateDirection.PUSH, depend_fields={}) + _configure_strategy(strategy, update_direction=UpdateDirection.PUSH, depend_fields={}) local_src = TestProjectNode( node_id="upd_src", @@ -382,7 +394,8 @@ async def test_update_prepare_e30b_on_id_resolve_fail(setup_env, monkeypatch: py local, remote, bm, _ = setup_env strategy = TestProjectStrategy("test_project", local, remote, bm) - strategy.update_config( + _configure_strategy( + strategy, depend_fields={"dummy_id": "test_project"}, update_direction=UpdateDirection.PUSH, ) diff --git a/tests/unit/test_api_client_rate_limit.py b/tests/unit/test_api_client_rate_limit.py index 74ee0d7..2f6bdf8 100644 --- a/tests/unit/test_api_client_rate_limit.py +++ b/tests/unit/test_api_client_rate_limit.py @@ -1,9 +1,11 @@ from __future__ import annotations import asyncio +import logging import time import pytest +import httpx from sync_state_machine.datasource.api.client import ApiClient @@ -21,11 +23,28 @@ class _FakeResponse: class _FakeAsyncClient: def __init__(self) -> None: self.call_times: list[float] = [] + self.closed = False async def request(self, **kwargs): self.call_times.append(time.monotonic()) return _FakeResponse() + async def aclose(self) -> None: + self.closed = True + + +class _FailingAsyncClient: + def __init__(self) -> None: + self.closed = False + + async def request(self, **kwargs): + request = httpx.Request(kwargs["method"], kwargs["url"]) + response = httpx.Response(500, request=request, text="boom") + raise httpx.HTTPStatusError("server error", request=request, response=response) + + async def aclose(self) -> None: + self.closed = True + @pytest.mark.asyncio async def test_api_client_global_rate_limit_applies_to_concurrent_requests() -> None: @@ -48,3 +67,50 @@ async def test_api_client_global_rate_limit_applies_to_concurrent_requests() -> assert len(fake_client.call_times) == 3 third_delay = fake_client.call_times[2] - fake_client.call_times[0] assert third_delay >= 0.15 + + +@pytest.mark.asyncio +async def test_api_client_request_stats_track_success_and_failure(caplog: pytest.LogCaptureFixture) -> None: + client = ApiClient( + base_url="http://example.com", + uid="uid", + secret="secret", + ) + logger_name = "sync_state_machine.datasource.api.client" + success_client = _FakeAsyncClient() + client._client = success_client + + with caplog.at_level(logging.INFO, logger=logger_name): + await client.request("GET", "/ping") + + failing_client = _FailingAsyncClient() + client._client = failing_client + with pytest.raises(httpx.HTTPStatusError): + await client.request("GET", "/ping") + + stats = client.get_request_stats() + assert stats == {"total": 2, "success": 1, "failure": 1} + assert client.format_request_stats().endswith("total=2, success=1, failure=1") + + await client.close() + + assert failing_client.closed is True + assert "📊 ApiClient request stats: total=2, success=1, failure=1" in caplog.text + + +@pytest.mark.asyncio +async def test_api_client_logs_one_line_request_summary_in_debug_mode(caplog: pytest.LogCaptureFixture) -> None: + client = ApiClient( + base_url="http://example.com", + uid="uid", + secret="secret", + debug=True, + ) + logger_name = "sync_state_machine.datasource.api.client" + fake_client = _FakeAsyncClient() + client._client = fake_client + + with caplog.at_level(logging.DEBUG, logger=logger_name): + await client.request("GET", "/ping") + + assert "ApiClient request: method=GET endpoint=/ping elapsed=" in caplog.text diff --git a/tests/unit/test_pipeline_strategy_continue_on_error.py b/tests/unit/test_pipeline_strategy_continue_on_error.py index 1c53241..3179f7b 100644 --- a/tests/unit/test_pipeline_strategy_continue_on_error.py +++ b/tests/unit/test_pipeline_strategy_continue_on_error.py @@ -10,14 +10,14 @@ from sync_state_machine.common.persistence import PersistenceBackend from sync_state_machine.common.sqlite_backend import SQLitePersistenceBackend from sync_state_machine.datasource.datasource import BaseDataSource from sync_state_machine.pipeline.full_sync_pipeline import FullSyncPipeline -from sync_state_machine.sync_system.strategy import BaseSyncStrategy +from sync_state_machine.sync_system.strategy import DefaultSyncStrategy class _DS(BaseDataSource): pass -class _FailStrategy(BaseSyncStrategy): +class _FailStrategy(DefaultSyncStrategy): schema = object # type: ignore async def bind(self, **kwargs): @@ -30,7 +30,7 @@ class _FailStrategy(BaseSyncStrategy): return [] -class _OkStrategy(BaseSyncStrategy): +class _OkStrategy(DefaultSyncStrategy): schema = object # type: ignore def __init__(self, *args, **kwargs): diff --git a/tests/unit/test_project_api_handler_dangerous_delete_config.py b/tests/unit/test_project_api_handler_dangerous_delete_config.py deleted file mode 100644 index 9762e10..0000000 --- a/tests/unit/test_project_api_handler_dangerous_delete_config.py +++ /dev/null @@ -1,57 +0,0 @@ -from __future__ import annotations - -import pytest - -from sync_state_machine.common.collection import DataCollection -from sync_state_machine.domain.project.api_handler import ProjectApiHandler - - -class _FakeProjectResponse: - def __init__(self, project_id: str): - self.id = project_id - - def model_dump(self) -> dict[str, str]: - return {"id": self.id, "name": f"project-{self.id}"} - - -@pytest.mark.asyncio -async def test_project_handler_purge_project_data_requires_data_id_filter() -> None: - handler = ProjectApiHandler() - handler.set_handler_config({"purge_project_data": True}) - - with pytest.raises(ValueError, match="requires non-empty data_id_filter"): - await handler.load() - - -@pytest.mark.asyncio -async def test_project_handler_purge_project_data_calls_remote_endpoint(monkeypatch: pytest.MonkeyPatch) -> None: - handler = ProjectApiHandler() - handler.set_handler_config( - { - "purge_project_data": True, - "data_id_filter": ["project-1"], - } - ) - handler.api_client = object() - handler.set_collection(DataCollection("remote")) - - purged: list[str] = [] - - async def fake_purge(client, project_id: str): - purged.append(project_id) - return {"project_id": project_id} - - async def fake_get_project(client, project_id: str): - return _FakeProjectResponse(project_id) - - async def fake_get_all_info(cls, client, project_id: str): - return {} - - monkeypatch.setattr("sync_state_machine.domain.project.api_handler.api_purge_project_data", fake_purge) - monkeypatch.setattr("sync_state_machine.domain.project.api_handler.api_get_project", fake_get_project) - monkeypatch.setattr(ProjectApiHandler, "get_all_info", classmethod(fake_get_all_info)) - - nodes = await handler.load() - - assert purged == ["project-1"] - assert [node.data_id for node in nodes] == ["project-1"] \ No newline at end of file diff --git a/tests/unit/test_project_api_handler_normalization.py b/tests/unit/test_project_api_handler_normalization.py index e46319a..1a41440 100644 --- a/tests/unit/test_project_api_handler_normalization.py +++ b/tests/unit/test_project_api_handler_normalization.py @@ -1,7 +1,10 @@ from __future__ import annotations +import pytest from uuid import uuid4 +from sync_state_machine.common.collection import DataCollection +from sync_state_machine.domain.project.api_handler import ProjectApiHandler from sync_state_machine.domain.project.api_handler import _merge_project_all_info_fields from sync_state_machine.domain.project.sync_node import ProjectSyncNode @@ -82,3 +85,12 @@ def test_project_sync_node_preserves_extension_backed_fields() -> None: assert normalized["implementation_estimate"] == 120.0 assert normalized["static_total_investment"] == 110.0 assert normalized["completed_investment"] == 70.0 + + +@pytest.mark.asyncio +async def test_project_api_handler_load_requires_api_client() -> None: + handler = ProjectApiHandler() + handler.set_collection(DataCollection("remote")) + + with pytest.raises(RuntimeError, match="API client not set"): + await handler.load() diff --git a/tests/unit/test_project_detail_sync_strategy.py b/tests/unit/test_project_detail_sync_strategy.py new file mode 100644 index 0000000..a9f2113 --- /dev/null +++ b/tests/unit/test_project_detail_sync_strategy.py @@ -0,0 +1,68 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from sync_state_machine.domain.project_detail.sync_strategy import ProjectDetailSyncStrategy +from sync_state_machine.sync_system.config import UpdateDirection +from sync_state_machine.sync_system.strategy_ops import BoundNodePair + + +def _fake_node(*, node_id: str, key: str): + data = {"id": node_id, "key": key} + return SimpleNamespace( + node_id=node_id, + data_id=node_id, + action=None, + context={}, + get_data=lambda: data, + ) + + +@pytest.mark.asyncio +async def test_project_detail_update_partitions_group_production_plans_before_generic_logic(): + strategy = ProjectDetailSyncStrategy("project_detail_data", MagicMock(), MagicMock(), MagicMock()) + + pull_pair = BoundNodePair( + local_node=_fake_node(node_id="local-pull", key="ensure_production"), + remote_node=_fake_node(node_id="remote-pull", key="ensure_production"), + ) + generic_pair = BoundNodePair( + local_node=_fake_node(node_id="local-generic", key="production"), + remote_node=_fake_node(node_id="remote-generic", key="production"), + ) + pull_node = SimpleNamespace(node_id="pull-update") + generic_node = SimpleNamespace(node_id="generic-update") + + with patch.object( + strategy, + "collect_update_pairs", + new=AsyncMock(return_value=[pull_pair, generic_pair]), + ), patch( + "sync_state_machine.domain.project_detail.sync_strategy.build_data_id_normalization_map", + new=AsyncMock(return_value={}), + ), patch.object( + strategy, + "prepare_update_for_direction", + new=AsyncMock(return_value=pull_node), + ) as mock_prepare, patch.object( + strategy, + "update_pair", + new=AsyncMock(return_value=[generic_node]), + ) as mock_update_pair: + updated_nodes = await strategy.update() + + assert updated_nodes == [pull_node, generic_node] + mock_prepare.assert_awaited_once_with( + pull_pair.local_node, + pull_pair.remote_node, + UpdateDirection.PULL, + {}, + ) + mock_update_pair.assert_awaited_once_with(generic_pair.local_node, generic_pair.remote_node, {}) + + +def test_project_detail_domain_option_is_typed_schema() -> None: + strategy = ProjectDetailSyncStrategy("project_detail_data", MagicMock(), MagicMock(), MagicMock()) + + assert strategy.domain_option.pull_group_production_plans is True \ No newline at end of file diff --git a/tests/unit/test_project_sync_strategy.py b/tests/unit/test_project_sync_strategy.py new file mode 100644 index 0000000..5706d70 --- /dev/null +++ b/tests/unit/test_project_sync_strategy.py @@ -0,0 +1,68 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from sync_state_machine.domain.project.sync_strategy import ProjectSyncStrategy +from sync_state_machine.sync_system.config import UpdateDirection + + +def _fake_node(*, node_id: str, data_id: str, data: dict, context=None): + return SimpleNamespace( + node_id=node_id, + data_id=data_id, + data=data, + context=context or {}, + action=None, + get_data=lambda: data, + ) + + +def test_project_get_node_update_payload_enriches_group_plan_dates_from_all_info(): + strategy = ProjectSyncStrategy("project", MagicMock(), MagicMock(), MagicMock()) + node = _fake_node( + node_id="node-1", + data_id="project-1", + data={"id": "project-1", "project_name": "demo"}, + context={ + "all_info": { + "wind_extension": { + "ic_plan_first_production_date": "2026-01-01", + "ic_plan_full_production_date": "2026-06-01", + } + } + }, + ) + + payload = strategy.get_node_update_payload(node) + + assert payload["ic_plan_first_production_date"] == "2026-01-01" + assert payload["ic_plan_full_production_date"] == "2026-06-01" + + +@pytest.mark.asyncio +async def test_project_update_pair_runs_special_pull_then_generic_update(): + strategy = ProjectSyncStrategy("project", MagicMock(), MagicMock(), MagicMock()) + strategy.config.domain_option = {"pull_group_plan_production_date": True} + + local_node = _fake_node(node_id="local", data_id="local-1", data={"id": "local-1"}) + remote_node = _fake_node(node_id="remote", data_id="remote-1", data={"id": "remote-1"}) + pull_node = SimpleNamespace(node_id="pull-node") + push_node = SimpleNamespace(node_id="push-node") + + with patch.object( + strategy, + "prepare_update_for_direction", + new=AsyncMock(side_effect=[pull_node, push_node]), + ) as mock_prepare: + updated_nodes = await strategy.update_pair(local_node, remote_node, {"project": "remote-project"}) + + assert updated_nodes == [pull_node, push_node] + + pull_call = mock_prepare.await_args_list[0] + assert pull_call.args == (local_node, remote_node, UpdateDirection.PULL, {"project": "remote-project"}) + assert pull_call.kwargs["include_fields"] == strategy._PULL_ONLY_FIELDS + + generic_call = mock_prepare.await_args_list[1] + assert generic_call.args == (local_node, remote_node, UpdateDirection.PUSH, {"project": "remote-project"}) + assert generic_call.kwargs["exclude_fields"] == strategy._PULL_ONLY_FIELDS \ No newline at end of file diff --git a/tests/unit/test_run_presets_yaml_loader.py b/tests/unit/test_run_presets_yaml_loader.py index f22a07c..ab3401a 100644 --- a/tests/unit/test_run_presets_yaml_loader.py +++ b/tests/unit/test_run_presets_yaml_loader.py @@ -176,7 +176,9 @@ strategies: snapshot = config_snapshot(config) assert snapshot["strategies"][0]["config"]["compare_ignore_fields"] == [] - assert snapshot["strategies"][0]["config"]["field_direction_overrides"] == {} + assert snapshot["strategies"][0]["config"]["domain_option"] == { + "pull_group_plan_production_date": False + } def test_build_config_from_file_rejects_legacy_strategy_keys(tmp_path: Path) -> None: diff --git a/tests/unit/test_strategy_runtime_config.py b/tests/unit/test_strategy_runtime_config.py index 16e7f1c..7a8c567 100644 --- a/tests/unit/test_strategy_runtime_config.py +++ b/tests/unit/test_strategy_runtime_config.py @@ -1,14 +1,21 @@ from __future__ import annotations +from pydantic import BaseModel + from sync_state_machine.config.builder import _apply_strategy_overrides -from sync_state_machine.config.strategy_config import StrategyConfig, StrategyRuntimeConfig +from sync_state_machine.config.strategy_config import ( + StrategyConfig, + StrategyRuntimeConfig, + resolve_domain_option_config, +) def test_strategy_runtime_config_skip_sync_defaults_false() -> None: runtime_config = StrategyRuntimeConfig(node_type="project") - assert runtime_config.skip_sync is False - assert runtime_config.skip_post_check is False + assert runtime_config.config.skip_sync is False + assert runtime_config.config.skip_post_check is False + assert runtime_config.config.domain_option == {} def test_apply_strategy_overrides_allows_skip_sync_override() -> None: @@ -24,8 +31,34 @@ def test_apply_strategy_overrides_allows_skip_sync_override() -> None: assert len(resolved) == 1 assert resolved[0].node_type == "project" - assert resolved[0].skip_sync is True - assert resolved[0].skip_post_check is True + assert resolved[0].config.skip_sync is True + assert resolved[0].config.skip_post_check is True + + +def test_apply_strategy_overrides_allows_domain_option_override() -> None: + resolved = _apply_strategy_overrides( + node_types=["project"], + strategy_overrides={ + "project": { + "domain_option": { + "pull_group_plan_production_date": True, + } + } + }, + ) + + assert resolved[0].config.domain_option == {"pull_group_plan_production_date": True} + + +def test_apply_strategy_overrides_preserves_project_detail_default_domain_option() -> None: + resolved = _apply_strategy_overrides( + node_types=["project_detail_data"], + strategy_overrides={}, + ) + + assert resolved[0].config.domain_option == { + "pull_group_production_plans": True, + } def test_strategy_config_supports_both_post_check_ignore_fields() -> None: @@ -39,4 +72,20 @@ def test_strategy_config_supports_both_post_check_ignore_fields() -> None: assert config.post_check_ignore_fields == ["code"] assert config.post_check_ignore_list_item_fields == {"plan_node": ["is_audit"]} - assert config.allow_many_to_one_auto_bind is True \ No newline at end of file + assert config.allow_many_to_one_auto_bind is True + + +def test_resolve_domain_option_config_fills_schema_defaults() -> None: + class ExampleDomainOption(BaseModel): + enabled: bool = True + mode: str = "strict" + + resolved = resolve_domain_option_config( + ExampleDomainOption, + {"enabled": False}, + ) + + assert resolved == { + "enabled": False, + "mode": "strict", + } \ No newline at end of file diff --git a/tests/unit/test_update_ops.py b/tests/unit/test_update_ops.py index 12ceded..7dbbdaf 100644 --- a/tests/unit/test_update_ops.py +++ b/tests/unit/test_update_ops.py @@ -1,74 +1,78 @@ import pytest +from types import SimpleNamespace from unittest.mock import AsyncMock, patch, MagicMock from sync_state_machine.domain.company.sync_strategy import CompanySyncStrategy -from sync_state_machine.sync_system.config import UpdateDirection -from sync_state_machine.sync_system.strategy_ops.update_ops import run_update +from sync_state_machine.sync_system.strategy_ops.update_ops import BoundNodePair, build_merged_update_payload, run_update from sync_state_machine.domain.construction.sync_strategy import ConstructionTaskSyncStrategy from sync_state_machine.domain.project_detail.sync_strategy import ProjectDetailSyncStrategy from schemas.project_extensions.project_detail import ProjectDetailKey @pytest.mark.asyncio -async def test_run_update_with_direction_overrides(): - class FakeNode: - def __init__(self, data): - self._data = data +async def test_run_update_delegates_to_strategy_pair_planner(): + local_node = SimpleNamespace(node_id="local-1") + remote_node = SimpleNamespace(node_id="remote-1") - def get_data(self): - return self._data - - # Setup mock strategy strategy = MagicMock() strategy.node_type = "TestNode" - strategy.config.field_direction_key = "sync_direction" - strategy.config.field_direction_overrides = { - "local_only": UpdateDirection.PUSH, - "remote_only": UpdateDirection.PULL, - "ignore": UpdateDirection.BIDIRECTIONAL # Just to check filtering - } - strategy.config.update_direction = UpdateDirection.PUSH strategy.local_collection = MagicMock() strategy.remote_collection = MagicMock() + strategy.binding_manager = MagicMock() + strategy.reset_schema_diff_validator = MagicMock() + strategy.emit_schema_diff_report = MagicMock() + updated_node = SimpleNamespace(node_id="target-1") + strategy.update_pair = AsyncMock(return_value=[updated_node]) - with patch('sync_state_machine.sync_system.strategy_ops.update_ops.add_update_action', new_callable=AsyncMock) as mock_add_action: - mock_add_action.return_value = [] - - await run_update(strategy) - - # In the active directions, PUSH and PULL should be handled because of overrides. - # Check PUSH call - assert mock_add_action.call_count == 2 - - push_call = mock_add_action.call_args_list[0] - assert push_call.kwargs['source_is_local'] is True - assert push_call.kwargs['from_collection'] == strategy.local_collection - - pull_call = mock_add_action.call_args_list[1] - assert pull_call.kwargs['source_is_local'] is False - assert pull_call.kwargs['from_collection'] == strategy.remote_collection - - # Test the node filters - push_filter = push_call.kwargs['node_filter'] - pull_filter = pull_call.kwargs['node_filter'] + with patch( + 'sync_state_machine.sync_system.strategy_ops.update_ops.build_data_id_normalization_map', + new=AsyncMock(return_value={"project": "remote-project"}), + ), patch( + 'sync_state_machine.sync_system.strategy_ops.update_ops.collect_bound_node_pairs', + new=AsyncMock(return_value=[BoundNodePair(local_node=local_node, remote_node=remote_node)]), + ): + updated = await run_update(strategy) - # Node that has 'local_only' -> PUSH - node_push = FakeNode({"sync_direction": "local_only"}) - assert push_filter(node_push) is True - assert pull_filter(node_push) is False + assert updated == [updated_node] - # Node that has 'remote_only' -> PULL - node_pull = FakeNode({"sync_direction": "remote_only"}) - assert push_filter(node_pull) is False - assert pull_filter(node_pull) is True + strategy.update_pair.assert_awaited_once_with( + local_node, + remote_node, + {"project": "remote-project"}, + ) - # Node with unknown key -> default direction (PUSH) - node_default = FakeNode({"sync_direction": "unknown"}) - assert push_filter(node_default) is True - assert pull_filter(node_default) is False - # Node with empty data -> default direction (PUSH) - node_empty = FakeNode(None) - assert push_filter(node_empty) is True - assert pull_filter(node_empty) is False +def test_build_merged_update_payload_supports_include_and_exclude_fields(): + source_data = { + "id": "remote-1", + "project_name": "new-name", + "special": "from-source", + } + target_data = { + "id": "remote-1", + "project_name": "old-name", + "special": "from-target", + } + + include_only = build_merged_update_payload( + source_data=source_data, + target_data=target_data, + include_fields=["special"], + ) + exclude_special = build_merged_update_payload( + source_data=source_data, + target_data=target_data, + exclude_fields=["special"], + ) + + assert include_only == { + "id": "remote-1", + "project_name": "old-name", + "special": "from-source", + } + assert exclude_special == { + "id": "remote-1", + "project_name": "new-name", + "special": "from-target", + } def test_construction_task_compare_payload_ignores_plan_node_is_audit(): diff --git a/ui/runner.py b/ui/runner.py index 4f8d9a4..e4d43f7 100644 --- a/ui/runner.py +++ b/ui/runner.py @@ -18,9 +18,8 @@ from sync_state_machine.config import ( JsonlDataSourceConfig, ApiDataSourceConfig, apply_config_overrides, - resolve_log_file_path, ) -from sync_state_machine.pipeline.run_logger import clear_pipeline_logger_handlers, init_pipeline_logger +from sync_state_machine.logging import clear_app_logger_handlers, configure_app_logging from .config_models import PipelineUiConfig, default_ui_config @@ -128,15 +127,11 @@ class PipelineRunnerService: app_logger = None log_cfg = cfg.logging.model_copy(deep=True) - log_path = resolve_log_file_path(log_cfg) - self._status.current_log_file = str(log_path or "") + app_logger = None try: - app_logger = init_pipeline_logger( - log_file_path=log_path, - mirror_console=True, - replace_handlers=True, - ) + app_logger = configure_app_logging(log_cfg, replace_handlers=True) + self._status.current_log_file = str(log_cfg.file_path or "") ui_handler = _UiLogHandler(self._append_log_callback) ui_handler.setFormatter(logging.Formatter("%(message)s")) app_logger.addHandler(ui_handler) @@ -165,7 +160,7 @@ class PipelineRunnerService: ui_handler.close() except Exception: pass - clear_pipeline_logger_handlers(app_logger) + clear_app_logger_handlers(app_logger) self._status.running = False self._status.ended_at = datetime.now().isoformat(timespec="seconds")